"""
Tools for managing OCI Monitoring and Observability resources.
"""
import logging
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
import oci
logger = logging.getLogger(__name__)
def list_alarms(monitoring_client: oci.monitoring.MonitoringClient, compartment_id: str) -> List[Dict[str, Any]]:
"""
List all alarms in a compartment.
Args:
monitoring_client: OCI Monitoring client
compartment_id: OCID of the compartment
Returns:
List of alarms with their details
"""
try:
alarms_response = oci.pagination.list_call_get_all_results(
monitoring_client.list_alarms,
compartment_id
)
alarms = []
for alarm in alarms_response.data:
alarms.append({
"id": alarm.id,
"display_name": alarm.display_name,
"compartment_id": alarm.compartment_id,
"metric_compartment_id": alarm.metric_compartment_id,
"namespace": alarm.namespace,
"query": alarm.query,
"severity": alarm.severity,
"lifecycle_state": alarm.lifecycle_state,
"is_enabled": alarm.is_enabled,
"destinations": alarm.destinations,
"time_created": str(alarm.time_created),
"time_updated": str(alarm.time_updated),
})
logger.info(f"Found {len(alarms)} alarms in compartment {compartment_id}")
return alarms
except Exception as e:
logger.exception(f"Error listing alarms: {e}")
raise
def get_alarm(monitoring_client: oci.monitoring.MonitoringClient, alarm_id: str) -> Dict[str, Any]:
"""
Get details of a specific alarm.
Args:
monitoring_client: OCI Monitoring client
alarm_id: OCID of the alarm
Returns:
Details of the alarm
"""
try:
alarm = monitoring_client.get_alarm(alarm_id).data
alarm_details = {
"id": alarm.id,
"display_name": alarm.display_name,
"compartment_id": alarm.compartment_id,
"metric_compartment_id": alarm.metric_compartment_id,
"metric_compartment_id_in_subtree": alarm.metric_compartment_id_in_subtree,
"namespace": alarm.namespace,
"resource_group": alarm.resource_group,
"query": alarm.query,
"resolution": alarm.resolution,
"pending_duration": alarm.pending_duration,
"severity": alarm.severity,
"body": alarm.body,
"is_enabled": alarm.is_enabled,
"lifecycle_state": alarm.lifecycle_state,
"suppress": {
"time_suppress_from": str(alarm.suppression.time_suppress_from) if alarm.suppression else None,
"time_suppress_until": str(alarm.suppression.time_suppress_until) if alarm.suppression else None,
"description": alarm.suppression.description if alarm.suppression else None,
} if alarm.suppression else None,
"destinations": alarm.destinations,
"repeat_notification_duration": alarm.repeat_notification_duration,
"time_created": str(alarm.time_created),
"time_updated": str(alarm.time_updated),
}
logger.info(f"Retrieved details for alarm {alarm_id}")
return alarm_details
except Exception as e:
logger.exception(f"Error getting alarm details: {e}")
raise
def get_alarm_history(monitoring_client: oci.monitoring.MonitoringClient,
alarm_id: str,
alarm_historytype: str = "STATE_TRANSITION_HISTORY") -> List[Dict[str, Any]]:
"""
Get alarm state history.
Args:
monitoring_client: OCI Monitoring client
alarm_id: OCID of the alarm
alarm_historytype: Type of history (STATE_TRANSITION_HISTORY, STATE_HISTORY, RULE_HISTORY)
Returns:
List of alarm history entries
"""
try:
history_response = oci.pagination.list_call_get_all_results(
monitoring_client.get_alarm_history,
alarm_id,
alarm_historytype=alarm_historytype
)
history = []
for entry in history_response.data:
history.append({
"summary": entry.summary,
"timestamp": str(entry.timestamp),
"timestamp_triggered": str(entry.timestamp_triggered) if hasattr(entry, 'timestamp_triggered') and entry.timestamp_triggered else None,
})
logger.info(f"Retrieved {len(history)} history entries for alarm {alarm_id}")
return history
except Exception as e:
logger.exception(f"Error getting alarm history: {e}")
raise
def list_metrics(monitoring_client: oci.monitoring.MonitoringClient,
compartment_id: str,
namespace: Optional[str] = None) -> List[Dict[str, Any]]:
"""
List available metrics in a compartment.
Args:
monitoring_client: OCI Monitoring client
compartment_id: OCID of the compartment
namespace: Optional namespace to filter metrics
Returns:
List of available metrics
"""
try:
list_metrics_details = oci.monitoring.models.ListMetricsDetails(
compartment_id=compartment_id,
namespace=namespace
)
metrics_response = oci.pagination.list_call_get_all_results(
monitoring_client.list_metrics,
compartment_id,
list_metrics_details
)
metrics = []
for metric in metrics_response.data:
metrics.append({
"name": metric.name,
"namespace": metric.namespace,
"resource_group": metric.resource_group,
"compartment_id": metric.compartment_id,
"dimensions": metric.dimensions,
})
logger.info(f"Found {len(metrics)} metrics in compartment {compartment_id}")
return metrics
except Exception as e:
logger.exception(f"Error listing metrics: {e}")
raise
def query_metric_data(monitoring_client: oci.monitoring.MonitoringClient,
compartment_id: str,
query: str,
start_time: str,
end_time: str,
resolution: str = "1m") -> List[Dict[str, Any]]:
"""
Query metric data for a time range.
Args:
monitoring_client: OCI Monitoring client
compartment_id: OCID of the compartment
query: Metric query in MQL format
start_time: Start time in ISO format (YYYY-MM-DDTHH:MM:SSZ)
end_time: End time in ISO format (YYYY-MM-DDTHH:MM:SSZ)
resolution: Data resolution (1m, 5m, 1h)
Returns:
List of metric data points
"""
try:
summarize_metrics_data_details = oci.monitoring.models.SummarizeMetricsDataDetails(
namespace="oci_computeagent", # This will be overridden by the query
query=query,
start_time=datetime.fromisoformat(start_time.replace('Z', '+00:00')),
end_time=datetime.fromisoformat(end_time.replace('Z', '+00:00')),
resolution=resolution
)
metrics_data_response = monitoring_client.summarize_metrics_data(
compartment_id,
summarize_metrics_data_details
)
data_points = []
for metric_data in metrics_data_response.data:
data_points.append({
"namespace": metric_data.namespace,
"resource_group": metric_data.resource_group,
"compartment_id": metric_data.compartment_id,
"name": metric_data.name,
"dimensions": metric_data.dimensions,
"aggregated_datapoints": [
{
"timestamp": str(dp.timestamp),
"value": dp.value
}
for dp in metric_data.aggregated_datapoints
] if metric_data.aggregated_datapoints else [],
"resolution": metric_data.resolution,
})
logger.info(f"Retrieved metric data with {len(data_points)} series")
return data_points
except Exception as e:
logger.exception(f"Error querying metric data: {e}")
raise
def search_logs(logging_search_client: oci.loggingsearch.LogSearchClient,
search_logs_details: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
Search logs using the Logging Search API.
Args:
logging_search_client: OCI LoggingSearch client
search_logs_details: Search parameters including query, time range, and limit
Returns:
List of log entries matching the search criteria
"""
try:
search_details = oci.loggingsearch.models.SearchLogsDetails(
time_start=datetime.fromisoformat(search_logs_details['time_start'].replace('Z', '+00:00')),
time_end=datetime.fromisoformat(search_logs_details['time_end'].replace('Z', '+00:00')),
search_query=search_logs_details['search_query'],
is_return_field_info=search_logs_details.get('is_return_field_info', False)
)
logs_response = logging_search_client.search_logs(search_details)
log_entries = []
for result in logs_response.data.results:
log_entries.append({
"time": str(result.time),
"log_content": result.data,
})
logger.info(f"Found {len(log_entries)} log entries")
return log_entries
except Exception as e:
logger.exception(f"Error searching logs: {e}")
raise
def list_log_groups(logging_client: oci.logging.LoggingManagementClient,
compartment_id: str) -> List[Dict[str, Any]]:
"""
List all log groups in a compartment.
Args:
logging_client: OCI Logging Management client
compartment_id: OCID of the compartment
Returns:
List of log groups with their details
"""
try:
log_groups_response = oci.pagination.list_call_get_all_results(
logging_client.list_log_groups,
compartment_id
)
log_groups = []
for log_group in log_groups_response.data:
log_groups.append({
"id": log_group.id,
"display_name": log_group.display_name,
"description": log_group.description,
"compartment_id": log_group.compartment_id,
"time_created": str(log_group.time_created),
"time_last_modified": str(log_group.time_last_modified),
"lifecycle_state": log_group.lifecycle_state,
})
logger.info(f"Found {len(log_groups)} log groups in compartment {compartment_id}")
return log_groups
except Exception as e:
logger.exception(f"Error listing log groups: {e}")
raise
def list_logs(logging_client: oci.logging.LoggingManagementClient,
log_group_id: str) -> List[Dict[str, Any]]:
"""
List all logs in a log group.
Args:
logging_client: OCI Logging Management client
log_group_id: OCID of the log group
Returns:
List of logs with their details
"""
try:
logs_response = oci.pagination.list_call_get_all_results(
logging_client.list_logs,
log_group_id
)
logs = []
for log in logs_response.data:
logs.append({
"id": log.id,
"log_group_id": log.log_group_id,
"display_name": log.display_name,
"log_type": log.log_type,
"lifecycle_state": log.lifecycle_state,
"is_enabled": log.is_enabled,
"retention_duration": log.retention_duration,
"compartment_id": log.compartment_id,
"time_created": str(log.time_created),
"time_last_modified": str(log.time_last_modified),
})
logger.info(f"Found {len(logs)} logs in log group {log_group_id}")
return logs
except Exception as e:
logger.exception(f"Error listing logs: {e}")
raise
