Oraclecloud Query Transform

MetricName[interval]{dimensionKey = "value"}.groupingFunction.statistic

import oci
from datetime import datetime, timedelta

config = oci.config.from_file("~/.oci/config")
monitoring = oci.monitoring.MonitoringClient(config)

# CPU utilization across all instances (last 1 hour, 5-minute intervals)
response = monitoring.summarize_metrics_data(
    compartment_id=config["tenancy"],
    summarize_metrics_data_details=oci.monitoring.models.SummarizeMetricsDataDetails(
        namespace="oci_computeagent",
        query='CpuUtilization[5m].mean()',
        start_time=datetime.utcnow() - timedelta(hours=1),
        end_time=datetime.utcnow(),
    ),
)

for metric in response.data:
    resource = metric.dimensions.get("resourceDisplayName", "unknown")
    for dp in metric.aggregated_datapoints:
        print(f"{resource} | {dp.timestamp} | CPU: {dp.value:.1f}%")

# Memory utilization (requires OCI monitoring agent on instance)
mem_query = 'MemoryUtilization[5m].mean()'

# Network bytes in/out
net_in_query = 'NetworkBytesIn[5m].sum()'
net_out_query = 'NetworkBytesOut[5m].sum()'

# Disk I/O
disk_read_query = 'DiskBytesRead[5m].sum()'
disk_write_query = 'DiskBytesWritten[5m].sum()'

# Query helper function
def query_metric(query, namespace="oci_computeagent", hours=1):
    """Query a single metric and return results."""
    response = monitoring.summarize_metrics_data(
        compartment_id=config["tenancy"],
        summarize_metrics_data_details=oci.monitoring.models.SummarizeMetricsDataDetails(
            namespace=namespace,
            query=query,
            start_time=datetime.utcnow() - timedelta(hours=hours),
            end_time=datetime.utcnow(),
        ),
    )
    return response.data

# Example: get all core metrics for the last hour
for name, query in [
    ("CPU", "CpuUtilization[5m].mean()"),
    ("Memory", "MemoryUtilization[5m].mean()"),
    ("Net In", "NetworkBytesIn[5m].sum()"),
    ("Net Out", "NetworkBytesOut[5m].sum()"),
    ("Disk Read", "DiskBytesRead[5m].sum()"),
    ("Disk Write", "DiskBytesWritten[5m].sum()"),
]:
    results = query_metric(query)
    if results:
        latest = results[0].aggregated_datapoints[-1]
        print(f"{name}: {latest.value:.2f} at {latest.timestamp}")
    else:
        print(f"{name}: no data (check monitoring agent)")

# Query a specific instance by OCID
instance_id = "ocid1.instance.oc1..."
filtered_query = f'CpuUtilization[5m]{{resourceId = "{instance_id}"}}.max()'

response = monitoring.summarize_metrics_data(
    compartment_id=config["tenancy"],
    summarize_metrics_data_details=oci.monitoring.models.SummarizeMetricsDataDetails(
        namespace="oci_computeagent",
        query=filtered_query,
        start_time=datetime.utcnow() - timedelta(hours=6),
        end_time=datetime.utcnow(),
    ),
)

for metric in response.data:
    peak = max(metric.aggregated_datapoints, key=lambda dp: dp.value)
    print(f"Peak CPU in last 6h: {peak.value:.1f}% at {peak.timestamp}")

metrics = monitoring.list_metrics(
    compartment_id=config["tenancy"],
    list_metrics_details=oci.monitoring.models.ListMetricsDetails(
        namespace="oci_computeagent",
    ),
).data

unique_metrics = set()
for m in metrics:
    unique_metrics.add(m.name)

print("Available metrics:")
for name in sorted(unique_metrics):
    print(f"  {name}")

# First, create a notification topic
notifications = oci.ons.NotificationDataPlaneClient(config)
control_plane = oci.ons.NotificationControlPlaneClient(config)

topic = control_plane.create_topic(
    oci.ons.models.CreateTopicDetails(
        compartment_id=config["tenancy"],
        name="high-cpu-alerts",
        description="Alerts for high CPU utilization",
    )
).data

# Create a subscription (email)
notifications.create_subscription(
    oci.ons.models.CreateSubscriptionDetails(
        compartment_id=config["tenancy"],
        topic_id=topic.topic_id,
        protocol="EMAIL",
        endpoint="ops-team@example.com",
    )
)

# Create the alarm
monitoring.create_alarm(
    oci.monitoring.models.CreateAlarmDetails(
        compartment_id=config["tenancy"],
        display_name="High CPU Alert",
        namespace="oci_computeagent",
        query="CpuUtilization[5m].mean() > 80",
        severity="CRITICAL",
        destinations=[topic.topic_id],
        is_enabled=True,
        body="CPU utilization exceeded 80% for 5 minutes.",
        pending_duration="PT5M",  # ISO 8601 — must be high for 5 minutes
        repeat_notification_duration="PT15M",  # Re-alert every 15 minutes
    )
)
print("Alarm created — email confirmation sent to subscriber")