Amazon CloudWatch が OpenTelemetry メトリクスをネイティブサポート（GA）— OTLP 直接送信と PromQL を試してみた

"""Generate dummy metrics data: cpu_usage_percent (Gauge) and network_bytes_sent_total (Counter)."""

import json
import math
import random
import time
from datetime import datetime, timezone

INTERVAL_SEC = 60
POINTS = 1440  # 24h * 60min
SEED = 42

def generate_cpu_usage(now_ts: float) -> list[dict]:
    """Generate cpu_usage_percent: sine wave with day/night pattern + noon spike + noise."""
    random.seed(SEED)
    start_ts = now_ts - (POINTS * INTERVAL_SEC)
    data = []
    for i in range(POINTS):
        ts = start_ts + i * INTERVAL_SEC
        dt = datetime.fromtimestamp(ts, tz=timezone.utc)
        hour = dt.hour + dt.minute / 60.0

        # Base: sine wave peaking at 13:30
        base = math.sin((hour - 7) / 24 * 2 * math.pi)
        if 9 <= hour <= 18:
            value = 40 + 40 * max(0, base)  # 40-80%
        else:
            value = 10 + 20 * max(0, base + 0.5)  # 10-30%

        # Noon spike
        if 11.5 <= hour <= 12.5:
            value = 90 + random.uniform(0, 10)

        # Random noise
        value += random.uniform(-3, 3)
        value = max(0, min(100, value))

        data.append({"timestamp_ns": int(ts * 1e9), "value": round(value, 2)})
    return data

def generate_network_bytes(now_ts: float) -> list[dict]:
    """Generate network_bytes_sent_total: monotonically increasing counter."""
    random.seed(SEED + 1)
    start_ts = now_ts - (POINTS * INTERVAL_SEC)
    cumulative = 1_000_000
    data = []
    for i in range(POINTS):
        ts = start_ts + i * INTERVAL_SEC
        dt = datetime.fromtimestamp(ts, tz=timezone.utc)
        hour = dt.hour + dt.minute / 60.0

        if 9 <= hour <= 18:
            rate = random.uniform(5000, 15000)
        else:
            rate = random.uniform(500, 2000)

        if 11.5 <= hour <= 12.5:
            rate = random.uniform(30000, 50000)

        cumulative += rate
        data.append({"timestamp_ns": int(ts * 1e9), "value": round(cumulative, 2)})
    return data

def main():
    now_ts = time.time()
    cpu_data = generate_cpu_usage(now_ts)
    net_data = generate_network_bytes(now_ts)

    output = {
        "generated_at": datetime.fromtimestamp(now_ts, tz=timezone.utc).isoformat(),
        "interval_sec": INTERVAL_SEC,
        "points": POINTS,
        "metrics": {
            "cpu_usage_percent": {
                "type": "gauge",
                "unit": "percent",
                "data": cpu_data,
            },
            "network_bytes_sent_total": {
                "type": "sum",
                "unit": "bytes",
                "is_monotonic": True,
                "aggregation_temporality": "CUMULATIVE",
                "data": net_data,
            },
        },
    }

    with open("results/generated_data.json", "w") as f:
        json.dump(output, f)
    print(f"Generated {POINTS} points for 2 metrics")

if __name__ == "__main__":
    main()

"""Send metrics to CloudWatch OTLP endpoint with SigV4 signing."""

import json
import time
from datetime import datetime, timezone

import requests
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.session import Session
from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import (
    ExportMetricsServiceRequest,
)
from opentelemetry.proto.common.v1.common_pb2 import AnyValue, KeyValue
from opentelemetry.proto.metrics.v1.metrics_pb2 import (
    AggregationTemporality,
    Gauge,
    Metric,
    NumberDataPoint,
    ResourceMetrics,
    ScopeMetrics,
    Sum,
)
from opentelemetry.proto.resource.v1.resource_pb2 import Resource

REGION = "ap-northeast-1"
ENDPOINT = f"https://monitoring.{REGION}.amazonaws.com/v1/metrics"
SERVICE_NAME = "monitoring"
BATCH_SIZE = 1000

def get_credentials():
    session = Session()
    return session.get_credentials().get_frozen_credentials()

def build_resource() -> Resource:
    return Resource(
        attributes=[
            KeyValue(key="service.name", value=AnyValue(string_value="otel-metrics-demo")),
            KeyValue(key="service.version", value=AnyValue(string_value="1.0.0")),
            KeyValue(key="deployment.environment", value=AnyValue(string_value="test")),
        ]
    )

def build_gauge_request(data_points: list[dict]) -> ExportMetricsServiceRequest:
    """Build OTLP request for cpu_usage_percent (Gauge)."""
    start_time_ns = data_points[0]["timestamp_ns"]
    ndps = [
        NumberDataPoint(
            time_unix_nano=dp["timestamp_ns"],
            start_time_unix_nano=start_time_ns,
            as_double=dp["value"],
        )
        for dp in data_points
    ]
    metric = Metric(
        name="cpu_usage_percent",
        unit="percent",
        gauge=Gauge(data_points=ndps),
    )
    return _wrap_metric(metric)

def build_sum_request(data_points: list[dict]) -> ExportMetricsServiceRequest:
    """Build OTLP request for network_bytes_sent_total (Sum/Counter)."""
    start_time_ns = data_points[0]["timestamp_ns"]
    ndps = [
        NumberDataPoint(
            time_unix_nano=dp["timestamp_ns"],
            start_time_unix_nano=start_time_ns,
            as_double=dp["value"],
        )
        for dp in data_points
    ]
    metric = Metric(
        name="network_bytes_sent_total",
        unit="bytes",
        sum=Sum(
            data_points=ndps,
            aggregation_temporality=AggregationTemporality.AGGREGATION_TEMPORALITY_CUMULATIVE,
            is_monotonic=True,
        ),
    )
    return _wrap_metric(metric)

def _wrap_metric(metric: Metric) -> ExportMetricsServiceRequest:
    return ExportMetricsServiceRequest(
        resource_metrics=[
            ResourceMetrics(
                resource=build_resource(),
                scope_metrics=[ScopeMetrics(metrics=[metric])],
            )
        ]
    )

def send_request(payload: bytes, credentials) -> dict:
    """Send protobuf payload with SigV4 signature."""
    headers = {"Content-Type": "application/x-protobuf"}
    aws_request = AWSRequest(method="POST", url=ENDPOINT, data=payload, headers=headers)
    SigV4Auth(credentials, SERVICE_NAME, REGION).add_auth(aws_request)

    response = requests.post(
        ENDPOINT,
        data=payload,
        headers=dict(aws_request.headers),
        timeout=30,
    )
    return {"status_code": response.status_code, "body": response.text}

def send_metric_batched(data_points: list[dict], build_fn, credentials, metric_name: str):
    """Send metric data in batches of 1000 datapoints."""
    for i in range(0, len(data_points), BATCH_SIZE):
        batch = data_points[i : i + BATCH_SIZE]
        request = build_fn(batch)
        payload = request.SerializeToString()
        result = send_request(payload, credentials)
        print(f"  {metric_name} batch {i // BATCH_SIZE + 1}: {len(batch)} points, "
              f"{len(payload)} bytes -> {result['status_code']}")
        time.sleep(0.5)

def main():
    with open("results/generated_data.json") as f:
        data = json.load(f)

    credentials = get_credentials()

    print("=== Sending cpu_usage_percent (Gauge) ===")
    send_metric_batched(
        data["metrics"]["cpu_usage_percent"]["data"],
        build_gauge_request, credentials, "cpu_usage_percent",
    )

    print("\n=== Sending network_bytes_sent_total (Sum/Counter) ===")
    send_metric_batched(
        data["metrics"]["network_bytes_sent_total"]["data"],
        build_sum_request, credentials, "network_bytes_sent_total",
    )

if __name__ == "__main__":
    main()

"""Query CloudWatch PromQL API and record results."""

import json
import time
from datetime import datetime, timezone

import requests
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.session import Session

REGION = "ap-northeast-1"
BASE_URL = f"https://monitoring.{REGION}.amazonaws.com"
SERVICE_NAME = "monitoring"

def get_credentials():
    session = Session()
    return session.get_credentials().get_frozen_credentials()

def promql_query(query: str, credentials, time_param: str = None) -> dict:
    """Run PromQL instant query (/api/v1/query)."""
    url = f"{BASE_URL}/api/v1/query"
    params = {"query": query}
    if time_param:
        params["time"] = time_param

    aws_req = AWSRequest(
        method="POST", url=url, data=params,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    SigV4Auth(credentials, SERVICE_NAME, REGION).add_auth(aws_req)
    resp = requests.post(url, data=params, headers=dict(aws_req.headers), timeout=30)
    return resp.json()

def promql_query_range(query: str, start: str, end: str, step: str, credentials) -> dict:
    """Run PromQL range query (/api/v1/query_range)."""
    url = f"{BASE_URL}/api/v1/query_range"
    params = {"query": query, "start": start, "end": end, "step": step}

    aws_req = AWSRequest(
        method="POST", url=url, data=params,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    SigV4Auth(credentials, SERVICE_NAME, REGION).add_auth(aws_req)
    resp = requests.post(url, data=params, headers=dict(aws_req.headers), timeout=30)
    return resp.json()

def main():
    credentials = get_credentials()
    now = time.time()

    # Gauge queries
    print("avg_over_time(cpu_usage_percent[1h]):")
    r = promql_query("avg_over_time(cpu_usage_percent[1h])", credentials)
    print(f"  -> {r['data']['result'][0]['value'][1]}")

    print("max_over_time(cpu_usage_percent[24h]):")
    r = promql_query("max_over_time(cpu_usage_percent[24h])", credentials)
    print(f"  -> {r['data']['result'][0]['value'][1]}")

    print("quantile_over_time(0.95, cpu_usage_percent[24h]):")
    r = promql_query("quantile_over_time(0.95, cpu_usage_percent[24h])", credentials)
    print(f"  -> {r['data']['result'][0]['value'][1]}")

    # Counter queries
    print("rate(network_bytes_sent_total[15m]):")
    r = promql_query("rate(network_bytes_sent_total[15m])", credentials)
    print(f"  -> {r['data']['result'][0]['value'][1]}")

    print("increase(network_bytes_sent_total[1h]):")
    r = promql_query("increase(network_bytes_sent_total[1h])", credentials)
    print(f"  -> {r['data']['result'][0]['value'][1]}")

    # Range query: threshold
    start_24h = str(int(now - 86400))
    end = str(int(now))
    print("cpu_usage_percent > 70 (range query, step=60s):")
    r = promql_query_range("cpu_usage_percent > 70", start_24h, end, "60", credentials)
    points = len(r["data"]["result"][0]["values"])
    print(f"  -> {points} evaluation steps above threshold")

if __name__ == "__main__":
    main()

"""Send same metrics to CloudWatch Classic via PutMetricData."""

import json
import time
from datetime import datetime, timezone

import boto3

REGION = "ap-northeast-1"
NAMESPACE = "OTelComparison/Classic"
BATCH_SIZE = 1000

def send_metric_data(client, metric_name: str, unit: str, data_points: list[dict]):
    """Send metric data in batches of 1000."""
    for i in range(0, len(data_points), BATCH_SIZE):
        batch = data_points[i : i + BATCH_SIZE]
        metric_data = [
            {
                "MetricName": metric_name,
                "Timestamp": datetime.fromtimestamp(dp["timestamp_ns"] / 1e9, tz=timezone.utc),
                "Value": dp["value"],
                "Unit": unit,
            }
            for dp in batch
        ]
        response = client.put_metric_data(Namespace=NAMESPACE, MetricData=metric_data)
        status = response["ResponseMetadata"]["HTTPStatusCode"]
        print(f"  {metric_name} batch {i // BATCH_SIZE + 1}: {len(batch)} points -> {status}")
        time.sleep(0.5)

def main():
    with open("results/generated_data.json") as f:
        data = json.load(f)

    client = boto3.client("cloudwatch", region_name=REGION)

    print("=== Sending cpu_usage_percent (Classic) ===")
    send_metric_data(client, "cpu_usage_percent", "Percent",
                     data["metrics"]["cpu_usage_percent"]["data"])

    print("\n=== Sending network_bytes_sent_total (Classic) ===")
    send_metric_data(client, "network_bytes_sent_total", "Bytes",
                     data["metrics"]["network_bytes_sent_total"]["data"])

if __name__ == "__main__":
    main()

"""Query CloudWatch Classic (GetMetricData) and record results."""

import json
from datetime import datetime, timezone, timedelta

import boto3

REGION = "ap-northeast-1"
NAMESPACE = "OTelComparison/Classic"

def get_metric_data(client, queries: list[dict], start: datetime, end: datetime) -> dict:
    """Run GetMetricData."""
    response = client.get_metric_data(
        MetricDataQueries=queries, StartTime=start, EndTime=end,
    )
    return response["MetricDataResults"]

def main():
    client = boto3.client("cloudwatch", region_name=REGION)
    now = datetime.now(timezone.utc)
    start_24h = now - timedelta(hours=24)
    start_1h = now - timedelta(hours=1)

    # 1h Average
    results = get_metric_data(client, [{
        "Id": "m1",
        "MetricStat": {
            "Metric": {"Namespace": NAMESPACE, "MetricName": "cpu_usage_percent"},
            "Period": 3600, "Stat": "Average",
        },
    }], start_1h, now)
    print(f"1h Average: {results[0]['Values']}")

    # RATE
    results = get_metric_data(client, [
        {
            "Id": "m1",
            "MetricStat": {
                "Metric": {"Namespace": NAMESPACE, "MetricName": "network_bytes_sent_total"},
                "Period": 300, "Stat": "Average",
            },
            "ReturnData": False,
        },
        {"Id": "rate_m1", "Expression": "RATE(m1)", "ReturnData": True},
    ], start_24h, now)
    print(f"RATE: {results[0]['Values'][:5]}...")

    # IF threshold
    results = get_metric_data(client, [
        {
            "Id": "m1",
            "MetricStat": {
                "Metric": {"Namespace": NAMESPACE, "MetricName": "cpu_usage_percent"},
                "Period": 60, "Stat": "Average",
            },
            "ReturnData": False,
        },
        {"Id": "threshold", "Expression": "IF(m1 > 70, m1)", "ReturnData": True},
    ], start_24h, now)
    print(f"Threshold > 70: {len(results[0]['Values'])} points")

if __name__ == "__main__":
    main()