Amazon CloudWatch — Alarms & Dashboards¶

# app/alarms.py (excerpt)
def dlq_depth(queue: str, topic: str) -> dict:
    """Any visible message in a dead-letter queue is an incident."""
    return dict(
        AlarmName=f"sqs-{queue}-dlq-not-empty", Namespace="AWS/SQS", MetricName="ApproximateNumberOfMessagesVisible",
        Dimensions=[{"Name": "QueueName", "Value": queue}], Statistic="Maximum", Period=60, EvaluationPeriods=1,
        Threshold=0, ComparisonOperator="GreaterThanThreshold",
        TreatMissingData="notBreaching",                  # an idle queue reports no data; that is not an incident
        AlarmActions=[topic], OKActions=[topic],
    )


def bedrock_latency_p99(model_id: str, topic: str, ms: int = 20_000) -> dict:
    return dict(
        AlarmName=f"bedrock-{model_id}-latency-p99", Namespace="AWS/Bedrock", MetricName="InvocationLatency",
        Dimensions=[{"Name": "ModelId", "Value": model_id}], ExtendedStatistic="p99", Period=300,
        EvaluationPeriods=3, DatapointsToAlarm=2, Threshold=ms, ComparisonOperator="GreaterThanThreshold",
        EvaluateLowSampleCountPercentile="ignore",        # do not alarm on a p99 computed from a handful of requests
        TreatMissingData="notBreaching", AlarmActions=[topic],
    )


def error_rate(namespace: str, errors: str, requests: str, topic: str, percent: float = 5) -> dict:
    """Metric math alarm: 100 * errors / requests over one-minute sums."""
    stat = lambda i, name: {"Id": i, "MetricStat": {"Metric": {"Namespace": namespace, "MetricName": name},
                                                    "Period": 60, "Stat": "Sum"}, "ReturnData": False}
    return dict(
        AlarmName=f"{namespace.replace('/', '-')}-error-rate", EvaluationPeriods=5, DatapointsToAlarm=3,
        Threshold=percent, ComparisonOperator="GreaterThanThreshold", TreatMissingData="notBreaching",
        Metrics=[stat("e", errors), stat("r", requests),
                 {"Id": "rate", "Expression": "100 * e / r", "Label": "error %", "ReturnData": True}],
        AlarmActions=[topic],
    )


cw.put_metric_alarm(**dlq_depth("orders-dlq", TOPIC))

# Composite: fire only when the queue is backing up AND the workers are saturated; page on this one, not on its children
cw.put_composite_alarm(
    AlarmName="orders-pipeline-unhealthy",
    AlarmRule='ALARM("sqs-orders-oldest-message-age") AND ALARM("ecs-prod-orders-worker-cpu-high")',
    AlarmActions=[TOPIC], AlarmDescription="queue is backing up and workers are saturated",
)

# Anomaly detection: train a model on the metric, then alarm when it drops below the band
cw.put_anomaly_detector(SingleMetricAnomalyDetector={
    "Namespace": "QA/Orders", "MetricName": "OrdersCreated", "Stat": "Sum",
    "Dimensions": [{"Name": "Service", "Value": "orders-api"}]})
cw.put_metric_alarm(
    AlarmName="orders-created-anomaly", ComparisonOperator="LessThanLowerThreshold",
    EvaluationPeriods=3, DatapointsToAlarm=3, ThresholdMetricId="band", TreatMissingData="notBreaching",
    Metrics=[{"Id": "m1", "ReturnData": True, "MetricStat": {"Period": 300, "Stat": "Sum", "Metric": {
                 "Namespace": "QA/Orders", "MetricName": "OrdersCreated",
                 "Dimensions": [{"Name": "Service", "Value": "orders-api"}]}}},
             {"Id": "band", "Expression": "ANOMALY_DETECTION_BAND(m1, 2)", "Label": "expected", "ReturnData": True}],
    AlarmActions=[TOPIC])

alarms = [a for page in cw.get_paginator("describe_alarms").paginate(AlarmNamePrefix="sqs-", StateValue="ALARM")
          for a in page["MetricAlarms"]]                                  # composite alarms are in page["CompositeAlarms"]

cw.set_alarm_state(AlarmName="sqs-orders-dlq-dlq-not-empty", StateValue="ALARM", StateReason="drill")
cw.disable_alarm_actions(AlarmNames=["sqs-orders-dlq-dlq-not-empty"])      # keep evaluating, stop notifying

# app/alarm_history.py
def wait_for_state_update(cw, alarm_name, new_state, *, timeout=60, interval=2, sleep=time.sleep, clock=time.monotonic):
    deadline = clock() + timeout
    while True:
        response = cw.describe_alarm_history(AlarmName=alarm_name, HistoryItemType="StateUpdate",
                                             ScanBy="TimestampDescending", MaxRecords=20)
        for item in response["AlarmHistoryItems"]:
            if json.loads(item["HistoryData"])["newState"]["stateValue"] == new_state:
                return item
        if clock() >= deadline:
            raise AssertionError(f"{alarm_name} never entered {new_state} within {timeout}s")
        sleep(interval)

{"source": ["aws.cloudwatch"], "detail-type": ["CloudWatch Alarm State Change"],
 "detail": {"alarmName": ["sqs-orders-dlq-dlq-not-empty"], "previousState": {"value": ["OK"]}, "state": {"value": ["ALARM"]}}}

# app/dashboards.py (excerpt)
def metric_widget(title, x, y, metrics, *, stat="Average", period=60, region="us-east-1", width=12, height=6):
    return {"type": "metric", "x": x, "y": y, "width": width, "height": height,
            "properties": {"title": title, "region": region, "stat": stat, "period": period,
                           "metrics": metrics, "view": "timeSeries"}}

body = orders_dashboard("us-east-1", "123456789012", "orders")        # text, 2 metric widgets, alarm status, log query
assert problems(body) == []                                           # structural checks, no AWS call
result = cw.put_dashboard(DashboardName="orders", DashboardBody=dumps(body))   # the body is a JSON string
assert result["DashboardValidationMessages"] == []                    # warnings: some widgets may not render; errors: not saved