Infrastructure Monitoring on AWS

import boto3
from datetime import datetime

cloudwatch = boto3.client("cloudwatch")

def put_metric(namespace: str, metric_name: str, value: float,
               unit: str = "Count", dimensions: dict = None) -> None:
    cloudwatch.put_metric_data(
        Namespace=namespace,
        MetricData=[{
            "MetricName": metric_name,
            "Value": value,
            "Unit": unit,
            "Timestamp": datetime.utcnow(),
            "Dimensions": [
                {"Name": k, "Value": v}
                for k, v in (dimensions or {}).items()
            ],
        }],
    )

# Usage in application code
put_metric("MyApp/Orders", "OrderProcessingTime", 142.5,
           unit="Milliseconds", dimensions={"Environment": "prod"})
put_metric("MyApp/Orders", "OrdersPlaced", 1,
           dimensions={"Region": "eu-west-1"})

# Embedded Metrics Format (EMF) — structured logs that auto-create metrics
import json

def log_emf(metric_name: str, value: float, unit: str = "Count") -> None:
    print(json.dumps({
        "_aws": {
            "Timestamp": int(datetime.utcnow().timestamp() * 1000),
            "CloudWatchMetrics": [{
                "Namespace": "MyApp",
                "Dimensions": *"Environment"*,
                "Metrics": [{"Name": metric_name, "Unit": unit}],
            }],
        },
        "Environment": "prod",
        metric_name: value,
    }))

# Works in Lambda — zero SDK overhead, metrics created from log ingestion
log_emf("PaymentProcessed", 1)
log_emf("ProcessingLatencyMs", 95.3, unit="Milliseconds")

from aws_cdk import aws_cloudwatch as cw

dashboard = cw.Dashboard(self, "AppDashboard", dashboard_name="OrderService")

# Error rate widget
error_rate = cw.GraphWidget(
    title="Error Rate",
    left=[
        cw.Metric(
            namespace="AWS/ApiGateway",
            metric_name="5XXError",
            dimensions_map={"ApiName": "OrderAPI"},
            statistic="Sum",
            period=Duration.minutes(1),
            color=cw.Color.RED,
        )
    ],
    width=12, height=6,
)

# P99 latency
latency_p99 = cw.GraphWidget(
    title="P99 Latency",
    left=[
        cw.Metric(
            namespace="MyApp",
            metric_name="ProcessingLatencyMs",
            statistic="p99",
            period=Duration.minutes(5),
        )
    ],
    width=12, height=6,
)

dashboard.add_widgets(
    cw.Row(error_rate, latency_p99),
)

# Standard threshold alarm
cw.Alarm(self, "HighErrorRate",
    metric=cw.Metric(
        namespace="AWS/Lambda",
        metric_name="Errors",
        dimensions_map={"FunctionName": "order-processor"},
        statistic="Sum",
        period=Duration.minutes(5),
    ),
    threshold=10,
    evaluation_periods=2,
    comparison_operator=cw.ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD,
    alarm_description="Lambda error rate high — check DLQ",
    treat_missing_data=cw.TreatMissingData.NOT_BREACHING,
)

# Anomaly detection alarm (learns from historical pattern)
anomaly_detector = cw.CfnAnomalyDetector(self, "LatencyAnomalyDetector",
    namespace="MyApp",
    metric_name="ProcessingLatencyMs",
    stat="p99",
    dimensions=[cw.CfnAnomalyDetector.DimensionProperty(
        name="Environment", value="prod",
    )],
    configuration=cw.CfnAnomalyDetector.ConfigurationProperty(
        excluded_time_ranges=[],
        metric_time_zone="UTC",
    ),
)

# Lambda + FastAPI with X-Ray auto-instrumentation
from aws_xray_sdk.core import xray_recorder, patch_all
from aws_xray_sdk.core.sampling.sampler import DefaultSampler

patch_all()  # patches boto3, requests, httpx, SQLAlchemy automatically

xray_recorder.configure(
    sampling=True,
    context_missing="LOG_ERROR",   # don't crash on missing context
    plugins=("EC2Plugin", "ECSPlugin"),
)

# Manual subsegment
@xray_recorder.capture("validate_order")
def validate_order(order: dict) -> bool:
    with xray_recorder.in_subsegment("schema_check") as subsegment:
        subsegment.put_annotation("order_id", order["id"])
        subsegment.put_metadata("order_payload", order)
        return "items" in order and len(order["items"]) > 0

# Lambda function — X-Ray active tracing
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  OrderProcessor:
    Type: AWS::Serverless::Function
    Properties:
      Tracing: Active   # PassThrough = only sample if upstream sampled; Active = always sample

// canary.js — smoke-tests your production endpoints every 5 minutes
const synthetics = require("Synthetics");
const syntheticsLogger = require("SyntheticsLogger");

const apiCanaryBlueprint = async function () {
    const requestOptions = {
        hostname: "api.myapp.com",
        method: "GET",
        path: "/health/ready",
        port: 443,
        protocol: "https:",
    };

    const stepConfig = {
        continueOnStepFailure: false,
        includeRequestHeaders: true,
        includeResponseHeaders: true,
        includeResponseBody: true,
    };

    await synthetics.executeHttpStep(
        "VerifyAPIHealth",
        requestOptions,
        validateHealthResponse,
        stepConfig,
    );
};

const validateHealthResponse = async function (res) {
    if (res.statusCode !== 200) {
        throw new Error(`Expected 200, got ${res.statusCode}`);
    }
    const body = JSON.parse(res.body);
    if (body.status !== "healthy") {
        throw new Error(`Unhealthy: ${JSON.stringify(body)}`);
    }
    syntheticsLogger.info("Health check passed");
};

exports.handler = async () => {
    return await apiCanaryBlueprint();
};

# CDK: create and schedule the canary
from aws_cdk import aws_synthetics as synthetics

canary = synthetics.Canary(self, "ApiHealthCanary",
    canary_name="api-health-check",
    schedule=synthetics.Schedule.rate(Duration.minutes(5)),
    test=synthetics.Test.custom(
        code=synthetics.Code.from_asset("canaries/"),
        handler="canary.handler",
    ),
    runtime=synthetics.Runtime.SYNTHETICS_NODEJS_PUPPETEER_6_2,
    environment_variables={"API_HOST": "api.myapp.com"},
)

# Alarm on canary failure
cw.Alarm(self, "CanaryAlarm",
    metric=canary.metric_failed(),
    threshold=1,
    evaluation_periods=1,
)

# Calculate and emit error budget metrics
import boto3

def calculate_and_emit_slo_metrics(service: str, window_minutes: int = 60) -> dict:
    cw = boto3.client("cloudwatch", region_name="eu-west-1")

    now = datetime.utcnow()
    start = now - timedelta(minutes=window_minutes)

    def get_sum(metric_name: str) -> float:
        response = cw.get_metric_statistics(
            Namespace="MyApp",
            MetricName=metric_name,
            Dimensions=[{"Name": "Service", "Value": service}],
            StartTime=start, EndTime=now,
            Period=window_minutes * 60,
            Statistics=["Sum"],
        )
        datapoints = response["Datapoints"]
        return datapoints[0]["Sum"] if datapoints else 0.0

    total_requests = get_sum("RequestsTotal")
    error_requests = get_sum("RequestErrors")

    if total_requests == 0:
        return {"availability": 1.0, "error_budget_remaining": 1.0}

    availability = 1 - (error_requests / total_requests)
    slo_target = 0.999  # 99.9% SLO
    error_budget_remaining = (availability - slo_target) / (1 - slo_target)

    # Emit back as custom metrics
    put_metric("MyApp/SLO", "Availability", availability * 100, unit="Percent",
               dimensions={"Service": service})
    put_metric("MyApp/SLO", "ErrorBudgetRemaining", error_budget_remaining * 100,
               unit="Percent", dimensions={"Service": service})

    return {"availability": availability, "error_budget_remaining": error_budget_remaining}