Monitoring & Observability

# Create an alarm: alert when EC2 CPU > 80% for 5 minutes
aws cloudwatch put-metric-alarm \
  --alarm-name "HighCPU-web-server" \
  --alarm-description "CPU above 80% for 5 min" \
  --metric-name CPUUtilization \
  --namespace AWS/EC2 \
  --statistic Average \
  --period 300 \
  --threshold 80 \
  --comparison-operator GreaterThanOrEqualToThreshold \
  --evaluation-periods 1 \
  --dimensions Name=InstanceId,Value=i-0123456789abcdef0 \
  --alarm-actions arn:aws:sns:us-east-1:123456789012:ops-alerts

# List alarms currently in ALARM state
aws cloudwatch describe-alarms --state-value ALARM

import boto3

cw = boto3.client("cloudwatch")

def record_order_latency(duration_ms: float, service: str) -> None:
    cw.put_metric_data(
        Namespace="MyApp/Orders",
        MetricData=[{
            "MetricName": "ProcessingLatency",
            "Dimensions": [{"Name": "Service", "Value": service}],
            "Value": duration_ms,
            "Unit": "Milliseconds",
        }]
    )

import time
start = time.time()
process_order(order_id)
record_order_latency((time.time() - start) * 1000, "OrderService")

# Install the agent on Amazon Linux 2:
# sudo yum install -y amazon-cloudwatch-agent

# Config: /opt/aws/amazon-cloudwatch-agent/etc/amazon-cloudwatch-agent.json
{
  "logs": {
    "logs_collected": {
      "files": {
        "collect_list": [
          {
            "file_path": "/var/log/myapp/app.log",
            "log_group_name": "/myapp/production",
            "log_stream_name": "{instance_id}/app",
            "timestamp_format": "%Y-%m-%dT%H:%M:%S"
          },
          {
            "file_path": "/var/log/nginx/access.log",
            "log_group_name": "/myapp/nginx",
            "log_stream_name": "{instance_id}/nginx"
          }
        ]
      }
    }
  },
  "metrics": {
    "metrics_collected": {
      "mem": { "measurement": ["mem_used_percent"] },
      "disk": { "measurement": ["disk_used_percent"] }
    }
  }
}

import json
import time
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def handler(event, context):
    start = time.time()
    result = process_order(event["orderId"])

    # Emit structured JSON - every field becomes queryable in Logs Insights
    logger.info(json.dumps({
        "event": "order_processed",
        "orderId": event["orderId"],
        "userId": event["userId"],
        "amount": result["amount"],
        "duration_ms": round((time.time() - start) * 1000),
        "status": "success",
    }))
    return result

# In Logs Insights you can now query individual fields:
# fields @timestamp, orderId, duration_ms, status
# | filter status = "success"
# | stats avg(duration_ms) as avg_ms by bin(5m)

import boto3

logs = boto3.client("logs")

# Route ALL logs to S3 via Kinesis Firehose (compliance archival)
# roleArn is required for Firehose destinations: CloudWatch Logs assumes this
# role to call firehose:PutRecordBatch. Lambda destinations use resource policies
# instead, so no roleArn is needed for the Lambda filter below.
logs.put_subscription_filter(
    logGroupName="/myapp/production",
    filterName="all-logs-to-s3",
    filterPattern="",   # empty pattern matches every log event
    destinationArn="arn:aws:firehose:us-east-1:123456789012:deliverystream/logs-to-s3",
    roleArn="arn:aws:iam::123456789012:role/CloudWatchLogsToFirehoseRole",
)

# Route only ERROR logs to a Lambda for real-time alerting
logs.put_subscription_filter(
    logGroupName="/myapp/production",
    filterName="errors-to-lambda",
    filterPattern='{ $.status = "error" }',
    destinationArn="arn:aws:lambda:us-east-1:123456789012:function:error-alerter",
)

# Set a 90-day retention policy on the log group
logs.put_retention_policy(
    logGroupName="/myapp/production",
    retentionInDays=90,
)

# 1. Top 10 slowest Lambda invocations
fields @timestamp, @duration, @requestId
| sort @duration desc
| limit 10

# 2. Count errors by error code (structured JSON logs)
fields event, errorCode
| filter status = "error"
| stats count(*) as total by errorCode
| sort total desc

# 3. p99 and average latency per endpoint
fields path, duration_ms
| filter ispresent(duration_ms)
| stats pct(duration_ms, 99) as p99_ms,
        avg(duration_ms) as avg_ms,
        count(*) as requests by path
| sort p99_ms desc

# 4. Lambda cold start frequency per hour
filter ispresent(@initDuration)
| stats count(*) as cold_starts,
        avg(@initDuration) as avg_cold_start_ms
        by bin(1h)

# 5. Compare exception counts across multiple services
#    (select multiple log groups in the console before running)
fields @logStream, @message
| filter @message like /Exception/
| stats count(*) as exceptions by @logStream
| sort exceptions desc

import boto3
from aws_xray_sdk.core import xray_recorder, patch_all

patch_all()  # auto-instruments boto3, requests, psycopg2, etc.

@xray_recorder.capture("process_order")
def process_order(order_id: str):
    dynamodb = boto3.resource("dynamodb")
    # X-Ray automatically traces this DynamoDB call as a sub-segment
    return dynamodb.Table("Orders").get_item(Key={"orderId": order_id})

import boto3

logs = boto3.client("logs")

# Create an anomaly detector - AWS builds an ML baseline from your log patterns
response = logs.create_log_anomaly_detector(
    logGroupArnList=[
        "arn:aws:logs:us-east-1:123456789012:log-group:/aws/lambda/payment-service"
    ],
    detectorName="payment-service-anomalies",
    evaluationFrequency="FIVE_MIN",
    anomalyVisibilityTime=14,   # retain detected anomalies for 14 days
)

detector_arn = response["anomalyDetectorArn"]

# List active anomalies (after the detector has collected enough data)
result = logs.list_anomalies(
    anomalyDetectorArn=detector_arn,
    suppressionState="UNSUPPRESSED",
)
for anomaly in result["anomalies"]:
    print(f"Pattern: {anomaly['patternString']}")
    print(f"Score:   {anomaly['anomalyScore']}")

# Suppress a known noisy pattern permanently
# (useful for scheduled health checks or routine deployment logs)
if result["anomalies"]:
    logs.update_anomaly(
        anomalyDetectorArn=detector_arn,
        anomalyId=result["anomalies"][0]["anomalyId"],
        suppressionType="INFINITE",
    )

import json
import boto3
from datetime import datetime, timedelta, timezone

cw = boto3.client("cloudwatch")

# Rule: find the top IP addresses generating 5xx HTTP errors
# Requires ALB access logs delivered to CloudWatch Logs in JSON format
rule_body = json.dumps({
    "Schema": {"Name": "CloudWatchLogRule", "Version": 1},
    "AggregateOn": "Count",
    "Contribution": {
        "Filters": [
            {"Match": "$.status", "GreaterThan": 499}
        ],
        "Keys": ["$.clientIp"],
    },
    "LogFormat": "JSON",
    "LogGroupNames": ["/alb/access-logs"],
})

cw.put_insight_rule(
    RuleName="top-5xx-source-ips",
    RuleState="ENABLED",
    RuleDefinition=rule_body,
)

# Query the top 10 contributors over the past hour
now = datetime.now(timezone.utc)
report = cw.get_insight_rule_report(
    RuleName="top-5xx-source-ips",
    StartTime=now - timedelta(hours=1),
    EndTime=now,
    Period=3600,
    MaxContributorCount=10,
)

for contributor in report["Contributors"]:
    ip = contributor["Keys"][0]
    error_count = contributor["ApproximateAggregateValue"]
    print(f"{ip}: {error_count:.0f} errors")