AWS CloudWatch Monitoring Best Practices: Complete 2024 Guide with Code Examples
Master AWS CloudWatch monitoring with comprehensive best practices, 8+ code examples, and production-ready configurations for metrics, logs, alarms, dashboards, and advanced monitoring strategies.
CloudWatch is the cornerstone of AWS observability, providing comprehensive monitoring, logging, and alerting capabilities for your cloud infrastructure. Whether you're managing a single application or a complex multi-tier enterprise system, implementing CloudWatch best practices ensures visibility, faster troubleshooting, and proactive issue resolution. This guide covers fundamental concepts, practical implementations, and advanced strategies with production-ready code examples.
Understanding AWS CloudWatch Fundamentals
What is AWS CloudWatch?
AWS CloudWatch is a comprehensive monitoring and logging service that collects and tracks metrics, monitors log files, and sets alarms. It provides visibility into your AWS resources and applications through a unified dashboard experience. CloudWatch collects data from multiple sources including EC2 instances, RDS databases, Lambda functions, and custom applications.
CloudWatch Core Components
AWS CloudWatch consists of five essential components that work together:
1. Metrics: Quantitative measurements of your infrastructure and applications (CPU utilization, request count, latency) 2. Logs: Full text event logs from your applications and AWS services 3. Alarms: Automatic actions triggered when metrics cross defined thresholds 4. Dashboards: Visual representations of your metrics and logs 5. Events: Real-time system events and scheduled actions
Metric Types and Namespaces
CloudWatch organizes metrics into namespaces for logical grouping. Understanding namespace organization is critical for effective monitoring:
AWS/EC2 - EC2 instance metrics (CPUUtilization, NetworkIn, NetworkOut)
AWS/RDS - RDS database metrics (DatabaseConnections, CPUUtilization)
AWS/Lambda - Lambda function metrics (Duration, Errors, Throttles)
AWS/DynamoDB - DynamoDB table metrics (ConsumedReadCapacity, UserErrors)
AWS/ApplicationELB - Load balancer metrics (TargetResponseTime, HTTPCode_Target_5XX)
Custom/MyApplication - Custom application metrics
Implementing CloudWatch Metrics Collection
Publishing Custom Metrics with AWS CLI
Publishing custom metrics allows you to track application-specific data. Here's how to send custom metrics using the AWS CLI:
#!/bin/bash
# Publish custom application metrics to CloudWatch
NAMESPACE="CustomApp/Production"
INSTANCE_ID=$(curl -s http://169.254.169.254/latest/meta-data/instance-id)
TIMESTAMP=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
# Get system memory usage
MEMORY_USED=$(free | grep Mem | awk '{print ($3/($3+$4)) * 100.0}')
# Publish memory metric
aws cloudwatch put-metric-data \
--namespace "$NAMESPACE" \
--metric-name "MemoryUtilization" \
--value $MEMORY_USED \
--unit "Percent" \
--timestamp "$TIMESTAMP" \
--dimensions \
Name=InstanceId,Value=$INSTANCE_ID \
Name=Environment,Value=production \
Name=Service,Value=api-server
# Get disk usage
DISK_USED=$(df / | tail -1 | awk '{print $5}' | sed 's/%//')
aws cloudwatch put-metric-data \
--namespace "$NAMESPACE" \
--metric-name "DiskUtilization" \
--value $DISK_USED \
--unit "Percent" \
--dimensions \
Name=InstanceId,Value=$INSTANCE_ID \
Name=MountPoint,Value="/"
Custom Metrics with Python Boto3
For applications running on AWS, use Python with boto3 to publish metrics directly:
import boto3
import time
from datetime import datetime
import psutil
cloudwatch = boto3.client('cloudwatch')
def publish_application_metrics():
"""
Publish custom application metrics to CloudWatch.
Run this periodically from your application.
"""
# Get system metrics
cpu_percent = psutil.cpu_percent(interval=1)
memory = psutil.virtual_memory()
memory_percent = memory.percent
# Application-specific metrics
active_connections = get_active_connections() # Your function
request_latency_ms = measure_request_latency() # Your function
timestamp = datetime.utcnow()
# Publish metrics as a batch
cloudwatch.put_metric_data(
Namespace='CustomApp/Production',
MetricData=[
{
'MetricName': 'CPUUtilization',
'Value': cpu_percent,
'Unit': 'Percent',
'Timestamp': timestamp,
'Dimensions': [
{'Name': 'Environment', 'Value': 'production'},
{'Name': 'Service', 'Value': 'api-server'},
{'Name': 'Instance', 'Value': 'i-1234567890abcdef0'}
]
},
{
'MetricName': 'MemoryUtilization',
'Value': memory_percent,
'Unit': 'Percent',
'Timestamp': timestamp,
'Dimensions': [
{'Name': 'Environment', 'Value': 'production'},
{'Name': 'Service', 'Value': 'api-server'}
]
},
{
'MetricName': 'ActiveConnections',
'Value': active_connections,
'Unit': 'Count',
'Timestamp': timestamp,
'Dimensions': [
{'Name': 'Environment', 'Value': 'production'}
]
},
{
'MetricName': 'RequestLatency',
'Value': request_latency_ms,
'Unit': 'Milliseconds',
'Timestamp': timestamp,
'StatisticValues': {
'SampleCount': 100,
'Sum': request_latency_ms * 100,
'Minimum': request_latency_ms * 0.8,
'Maximum': request_latency_ms * 1.2
}
}
]
)
print("Metrics published successfully")
# Schedule this to run every minute
if __name__ == "__main__":
publish_application_metrics()
Embedded Metrics Format for Lambda
AWS Lambda functions can emit metrics using the Embedded Metrics Format (EMF) for automatic metric parsing:
import json
from aws_embedded_metrics import metric_scope
from aws_embedded_metrics.logger.metrics_logger import MetricsLogger
@metric_scope
def lambda_handler(event, context, metrics: MetricsLogger):
"""
Lambda function that automatically emits metrics to CloudWatch.
EMF automatically creates CloudWatch logs and extracts metrics.
"""
# Set namespace and environment
metrics.set_namespace("CustomApp/Lambda")
metrics.set_property("Environment", "production")
metrics.set_property("FunctionName", context.function_name)
metrics.set_property("RequestId", context.request_id)
# Process records and track metrics
records_processed = 0
errors_encountered = 0
try:
# Simulate processing
for record in event.get('Records', []):
try:
process_record(record)
records_processed += 1
except Exception as e:
errors_encountered += 1
metrics.set_property("Error", str(e))
# Emit metrics
metrics.put_metric("RecordsProcessed", records_processed, "Count")
metrics.put_metric("ErrorCount", errors_encountered, "Count")
metrics.put_metric("SuccessRate",
((records_processed - errors_encountered) / records_processed * 100)
if records_processed > 0 else 0,
"Percent"
)
return {
'statusCode': 200,
'body': json.dumps({
'processed': records_processed,
'errors': errors_encountered
})
}
except Exception as e:
metrics.put_metric("FunctionError", 1, "Count")
raise
def process_record(record):
"""Process individual record logic"""
pass
Configuring CloudWatch Alarms and Notifications
Creating Effective Alarms with CloudFormation
Proper alarm configuration prevents alert fatigue while ensuring critical issues are detected:
AWSTemplateFormatVersion: '2010-09-09'
Description: 'Production CloudWatch Alarms Configuration'
Resources:
# EC2 CPU Utilization Alarm
EC2HighCPUAlarm:
Type: AWS::CloudWatch::Alarm
Properties:
AlarmName: !Sub '${Environment}-ec2-high-cpu'
AlarmDescription: Alert when EC2 CPU exceeds 80% for 10 minutes
MetricName: CPUUtilization
Namespace: AWS/EC2
Statistic: Average
Period: 300 # 5 minutes
EvaluationPeriods: 2 # Must breach twice (10 minutes total)
Threshold: 80
ComparisonOperator: GreaterThanThreshold
TreatMissingData: notBreaching
AlarmActions:
- !Ref SNSAlertTopic
Dimensions:
- Name: InstanceId
Value: !Ref EC2Instance
# RDS Database CPU Alarm
RDSHighCPUAlarm:
Type: AWS::CloudWatch::Alarm
Properties:
AlarmName: !Sub '${Environment}-rds-cpu-high'
AlarmDescription: Alert when RDS CPU exceeds 75%
MetricName: CPUUtilization
Namespace: AWS/RDS
Statistic: Average
Period: 300
EvaluationPeriods: 2
Threshold: 75
ComparisonOperator: GreaterThanThreshold
TreatMissingData: notBreaching
AlarmActions:
- !Ref SNSAlertTopic
Dimensions:
- Name: DBInstanceIdentifier
Value: !Ref RDSDatabase
# Lambda Error Rate Alarm
LambdaErrorRateAlarm:
Type: AWS::CloudWatch::Alarm
Properties:
AlarmName: !Sub '${Environment}-lambda-errors'
AlarmDescription: Alert when Lambda error rate exceeds 5%
Metrics:
- Id: m1
ReturnData: true
MetricStat:
Metric:
Namespace: AWS/Lambda
MetricName: Errors
Dimensions:
- Name: FunctionName
Value: !Ref LambdaFunction
Period: 300
Stat: Sum
- Id: m2
ReturnData: true
MetricStat:
Metric:
Namespace: AWS/Lambda
MetricName: Invocations
Dimensions:
- Name: FunctionName
Value: !Ref LambdaFunction
Period: 300
Stat: Sum
- Id: e1
Expression: "(m1 / m2) * 100"
EvaluationPeriods: 1
Threshold: 5
ComparisonOperator: GreaterThanThreshold
TreatMissingData: notBreaching
AlarmActions:
- !Ref SNSAlertTopic
# Composite Alarm for Overall Application Health
ApplicationHealthCompositeAlarm:
Type: AWS::CloudWatch::CompositeAlarm
Properties:
AlarmName: !Sub '${Environment}-application-health'
AlarmDescription: Overall application health status
AlarmRule: !Sub |
ALARM(${EC2HighCPUAlarm.AlarmName}) OR
ALARM(${RDSHighCPUAlarm.AlarmName}) OR
ALARM(${LambdaErrorRateAlarm.AlarmName})
AlarmActions:
- !Ref CriticalSNSTopic
# SNS Topic for Alarm Notifications
SNSAlertTopic:
Type: AWS::SNS::Topic
Properties:
TopicName: !Sub '${Environment}-cloudwatch-alerts'
DisplayName: CloudWatch Alarm Notifications
KmsMasterKeyId: alias/aws/sns
# SNS Topic Subscription (Email)
EmailSubscription:
Type: AWS::SNS::Subscription
Properties:
Protocol: email
TopicArn: !Ref SNSAlertTopic
Endpoint: ops-team@example.com
Parameters:
Environment:
Type: String
Default: production
AllowedValues: [development, staging, production]
EC2Instance:
Type: String
Description: EC2 Instance ID to monitor
RDSDatabase:
Type: String
Description: RDS Database Instance Identifier
LambdaFunction:
Type: String
Description: Lambda Function Name to monitor
Outputs:
CompositeAlarmArn:
Value: !GetAtt ApplicationHealthCompositeAlarm.AlarmArn
Export:
Name: !Sub '${AWS::StackName}-CompositeAlarmArn'
CloudWatch Logs: Organization, Filtering, and Analysis
Organizing Log Groups and Streams
Proper log organization makes troubleshooting faster and reduces costs:
#!/bin/bash
# Create organized log group structure
# Application logs
aws logs create-log-group --log-group-name /app/production/api --tags Environment=production Service=api
aws logs create-log-group --log-group-name /app/production/workers --tags Environment=production Service=workers
aws logs create-log-group --log-group-name /app/production/scheduler --tags Environment=production Service=scheduler
# AWS service logs
aws logs create-log-group --log-group-name /aws/lambda/data-processor --tags Service=lambda
aws logs create-log-group --log-group-name /aws/rds/production/error --tags Service=rds
aws logs create-log-group --log-group-name /aws/ecs/production/api --tags Service=ecs
# Set retention policies to optimize costs
aws logs put-retention-policy --log-group-name /app/production/api --retention-in-days 30
aws logs put-retention-policy --log-group-name /app/production/workers --retention-in-days 30
aws logs put-retention-policy --log-group-name /aws/lambda/data-processor --retention-in-days 14
aws logs put-retention-policy --log-group-name /aws/rds/production/error --retention-in-days 90
Metric Filters for Derived Metrics
Extract critical metrics from unstructured logs:
import boto3
logs = boto3.client('logs')
def create_metric_filters():
"""
Create metric filters to extract metrics from CloudWatch Logs.
These derived metrics are useful for alarming on specific patterns.
"""
# Filter for API errors
logs.put_metric_filter(
logGroupName='/app/production/api',
filterName='ErrorRequests',
filterPattern='[timestamp, request_id, level="ERROR", ...]',
metricTransformations=[
{
'metricName': 'ErrorCount',
'metricNamespace': 'CustomApp/API',
'metricValue': '1',
'defaultValue': 0
}
]
)
# Filter for slow requests (latency > 1000ms)
logs.put_metric_filter(
logGroupName='/app/production/api',
filterName='SlowRequests',
filterPattern='[timestamp, request_id, level, latency > 1000, ...]',
metricTransformations=[
{
'metricName': 'SlowRequestCount',
'metricNamespace': 'CustomApp/API',
'metricValue': '1',
'defaultValue': 0
}
]
)
# Filter for specific error codes
logs.put_metric_filter(
logGroupName='/app/production/api',
filterName='DatabaseErrors',
filterPattern='[..., error_code="DB_TIMEOUT", ...]',
metricTransformations=[
{
'metricName': 'DatabaseErrors',
'metricNamespace': 'CustomApp/Errors',
'metricValue': '1',
'defaultValue': 0
}
]
)
print("Metric filters created successfully")
if __name__ == "__main__":
create_metric_filters()
CloudWatch Logs Insights Queries
CloudWatch Logs Insights allows you to run SQL-like queries on your logs:
-- Query 1: Find top 10 slowest requests
fields @timestamp, @duration, @message, requestId
| filter @duration > 100
| stats max(@duration) as maxDuration, pct(@duration, 95) as p95Duration by requestId
| sort maxDuration desc
| limit 10
-- Query 2: Error rate per service
fields @timestamp, service, @message
| filter @message like /ERROR/
| stats count() as errorCount by service
| sort errorCount desc
-- Query 3: Trace requests across microservices
fields @timestamp, @message, traceId, service, @duration
| filter traceId = "abc-123-def-456"
| stats sum(@duration) as totalDuration by service
| sort @timestamp
-- Query 4: Identify database connection issues
fields @timestamp, @message, dbPool, @message
| filter @message like /connection timeout|pool exhausted/i
| stats count() as occurrences, max(@duration) as maxWaitTime by dbPool
| sort occurrences desc
-- Query 5: Request distribution by response time buckets
fields @duration
| filter @duration > 0
| stats count() as requests, pct(@duration, 50) as median, pct(@duration, 95) as p95, pct(@duration, 99) as p99
Creating and Managing Dashboards
CloudWatch Dashboard with Terraform
Infrastructure as Code approach for dashboard management:
# main.tf - Create comprehensive CloudWatch Dashboard
variable "environment" {
default = "production"
}
variable "application_name" {
default = "MyApplication"
}
resource "aws_cloudwatch_dashboard" "main" {
dashboard_name = "${var.environment}-${var.application_name}"
dashboard_body = jsonencode({
widgets = [
# Top row: Key metrics overview
{
type = "metric"
properties = {
metrics = [
["AWS/EC2", "CPUUtilization", { stat = "Average", label = "Avg CPU" }],
[".", "NetworkIn", { stat = "Sum", label = "Network In" }],
[".", "NetworkOut", { stat = "Sum", label = "Network Out" }]
]
period = 300
stat = "Average"
region = data.aws_region.current.name
title = "EC2 Instance Metrics"
yAxis = {
left = {
min = 0
max = 100
}
}
}
},
# Application performance
{
type = "metric"
properties = {
metrics = [
["CustomApp/API", "RequestLatency", { stat = "Average" }],
["...", { stat = "p95" }],
["...", { stat = "p99" }]
]
period = 60
stat = "Average"
region = data.aws_region.current.name
title = "API Latency (ms)"
yAxis = {
left = {
min = 0
}
}
}
},
# Error rate
{
type = "metric"
properties = {
metrics = [
["CustomApp/API", "ErrorCount", { stat = "Sum" }],
["AWS/Lambda", "Errors", { stat = "Sum" }]
]
period = 300
stat = "Sum"
region = data.aws_region.current.name
title = "Error Count"
yAxis = {
left = {
min = 0
}
}
}
},
# Database performance
{
type = "metric"
properties = {
metrics = [
["AWS/RDS", "DatabaseConnections", { stat = "Average" }],
[".", "CPUUtilization", { stat = "Average" }],
[".", "ReadLatency", { stat = "Average" }],
[".", "WriteLatency", { stat = "Average" }]
]
period = 300
stat = "Average"
region = data.aws_region.current.name
title = "RDS Performance"
}
},
# Application logs
{
type = "log"
properties = {
query = "fields @timestamp, @message | filter @message like /ERROR/ | stats count() as error_count"
region = data.aws_region.current.name
title = "Recent Errors (Last 1 hour)"
}
}
]
})
}
data "aws_region" "current" {}
output "dashboard_url" {
value = "https://console.aws.amazon.com/cloudwatch/home?region=${data.aws_region.current.name}#dashboards:name=${aws_cloudwatch_dashboard.main.dashboard_name}"
description = "URL to CloudWatch Dashboard"
}
Advanced CloudWatch Features
Anomaly Detection
CloudWatch can automatically detect anomalous behavior using machine learning:
import boto3
cloudwatch = boto3.client('cloudwatch')
def create_anomaly_alarm():
"""
Create an alarm based on anomaly detection.
CloudWatch learns normal patterns and alerts on deviations.
"""
# Create metric math alarm for anomaly detection
cloudwatch.put_metric_alarm(
AlarmName='API-Latency-Anomaly',
ComparisonOperator='LessThanLowerOrGreaterThanUpperThreshold',
EvaluationPeriods=2,
Metrics=[
{
'Id': 'm1',
'ReturnData': True,
'MetricStat': {
'Metric': {
'Namespace': 'CustomApp/API',
'MetricName': 'RequestLatency',
'Dimensions': [
{'Name': 'Environment', 'Value': 'production'}
]
},
'Period': 300,
'Stat': 'Average'
}
},
{
'Id': 'ad1',
'Expression': 'ANOMALY_DETECTION_BAND(m1, 2)',
'ReturnData': True
}
],
ThresholdMetricId='ad1',
TreatMissingData='notBreaching',
AlarmActions=['arn:aws:sns:us-east-1:123456789012:ops-alerts']
)
print("Anomaly detection alarm created")
if __name__ == "__main__":
create_anomaly_alarm()
Application Performance Monitoring (APM) with CloudWatch
Implement application-level monitoring for better insights:
import time
from functools import wraps
from datetime import datetime
import boto3
cloudwatch = boto3.client('cloudwatch')
class PerformanceMonitor:
"""
Monitor application performance and publish metrics to CloudWatch.
"""
def __init__(self, namespace='CustomApp/APM', environment='production'):
self.namespace = namespace
self.environment = environment
self.cloudwatch = cloudwatch
def track_function_performance(self, function_name, business_unit='general'):
"""
Decorator to track function execution metrics.
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
error_occurred = False
try:
result = func(*args, **kwargs)
return result
except Exception as e:
error_occurred = True
raise
finally:
duration_ms = (time.time() - start_time) * 1000
# Publish metrics
metrics = [
{
'MetricName': 'FunctionDuration',
'Value': duration_ms,
'Unit': 'Milliseconds',
'Dimensions': [
{'Name': 'FunctionName', 'Value': function_name},
{'Name': 'Environment', 'Value': self.environment},
{'Name': 'BusinessUnit', 'Value': business_unit}
]
}
]
if error_occurred:
metrics.append({
'MetricName': 'FunctionErrors',
'Value': 1,
'Unit': 'Count',
'Dimensions': [
{'Name': 'FunctionName', 'Value': function_name},
{'Name': 'Environment', 'Value': self.environment}
]
})
self.cloudwatch.put_metric_data(
Namespace=self.namespace,
MetricData=metrics
)
return wrapper
return decorator
# Usage example
monitor = PerformanceMonitor()
@monitor.track_function_performance('process_order', 'commerce')
def process_order(order_id):
# Your order processing logic
time.sleep(0.1) # Simulate work
return {'order_id': order_id, 'status': 'completed'}
# Call function - metrics are automatically published
process_order('ORD-12345')
Container Insights for ECS Monitoring
Monitor containerized applications with Container Insights:
# CloudFormation template for ECS with Container Insights
AWSTemplateFormatVersion: '2010-09-09'
Description: 'ECS Cluster with Container Insights enabled'
Resources:
ECSCluster:
Type: AWS::ECS::Cluster
Properties:
ClusterName: !Sub '${Environment}-cluster'
ClusterSettings:
- Name: containerInsights
Value: enabled
# CloudWatch Log Group for Container Insights
ContainerInsightsLogGroup:
Type: AWS::Logs::LogGroup
Properties:
LogGroupName: !Sub '/ecs/containerinsights/${Environment}-cluster/performance'
RetentionInDays: 7
# ECS Task Definition with CloudWatch logging
TaskDefinition:
Type: AWS::ECS::TaskDefinition
Properties:
Family: !Sub '${Environment}-task'
NetworkMode: awsvpc
RequiresCompatibilities:
- FARGATE
Cpu: '256'
Memory: '512'
ExecutionRoleArn: !GetAtt TaskExecutionRole.Arn
ContainerDefinitions:
- Name: app-container
Image: !Sub '${AWS::AccountId}.dkr.ecr.${AWS::Region}.amazonaws.com/my-app:latest'
Essential: true
PortMappings:
- ContainerPort: 8080
Protocol: tcp
LogConfiguration:
LogDriver: awslogs
Options:
awslogs-group: !Ref TaskLogGroup
awslogs-region: !Ref AWS::Region
awslogs-stream-prefix: ecs
Environment:
- Name: LOG_LEVEL
Value: INFO
- Name: ENABLE_METRICS
Value: 'true'
TaskLogGroup:
Type: AWS::Logs::LogGroup
Properties:
LogGroupName: !Sub '/ecs/${Environment}-task'
RetentionInDays: 30
TaskExecutionRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service: ecs-tasks.amazonaws.com
Action: sts:AssumeRole
ManagedPolicyArns:
- arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy
Policies:
- PolicyName: CloudWatchLogs
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- logs:CreateLogStream
- logs:PutLogEvents
Resource: !GetAtt TaskLogGroup.Arn
# ECS Service
ECSService:
Type: AWS::ECS::Service
Properties:
ServiceName: !Sub '${Environment}-service'
Cluster: !Ref ECSCluster
TaskDefinition: !Ref TaskDefinition
DesiredCount: 2
LaunchType: FARGATE
NetworkConfiguration:
AwsvpcConfiguration:
Subnets:
- !Ref PrivateSubnet1
- !Ref PrivateSubnet2
SecurityGroups:
- !Ref ContainerSecurityGroup
Parameters:
Environment:
Type: String
Default: production
AllowedValues: [development, staging, production]
PrivateSubnet1:
Type: AWS::EC2::Subnet::Id
PrivateSubnet2:
Type: AWS::EC2::Subnet::Id
ContainerSecurityGroup:
Type: AWS::EC2::SecurityGroup::Id
Lambda Monitoring Best Practices
Comprehensive monitoring for serverless applications:
import json
import time
from functools import wraps
import boto3
cloudwatch = boto3.client('cloudwatch')
def lambda_monitoring_decorator(lambda_handler):
"""
Decorator for Lambda functions to automatically emit monitoring metrics.
"""
@wraps(lambda_handler)
def wrapper(event, context):
start_time = time.time()
# Emit custom metrics
try:
result = lambda_handler(event, context)
duration_ms = (time.time() - start_time) * 1000
cloudwatch.put_metric_data(
Namespace='CustomApp/Lambda',
MetricData=[
{
'MetricName': 'ExecutionDuration',
'Value': duration_ms,
'Unit': 'Milliseconds',
'Dimensions': [
{'Name': 'FunctionName', 'Value': context.function_name},
{'Name': 'Environment', 'Value': 'production'}
]
},
{
'MetricName': 'SuccessCount',
'Value': 1,
'Unit': 'Count',
'Dimensions': [
{'Name': 'FunctionName', 'Value': context.function_name}
]
}
]
)
return result
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
cloudwatch.put_metric_data(
Namespace='CustomApp/Lambda',
MetricData=[
{
'MetricName': 'ExecutionDuration',
'Value': duration_ms,
'Unit': 'Milliseconds',
'Dimensions': [
{'Name': 'FunctionName', 'Value': context.function_name}
]
},
{
'MetricName': 'ErrorCount',
'Value': 1,
'Unit': 'Count',
'Dimensions': [
{'Name': 'FunctionName', 'Value': context.function_name},
{'Name': 'ErrorType', 'Value': type(e).__name__}
]
}
]
)
raise
return wrapper
@lambda_monitoring_decorator
def handler(event, context):
"""Example Lambda function with automatic monitoring"""
print(json.dumps({
'level': 'INFO',
'message': 'Processing event',
'event_id': event.get('id'),
'timestamp': time.time()
}))
# Your function logic here
return {
'statusCode': 200,
'body': json.dumps({'message': 'Success'})
}
Cost Optimization for CloudWatch
Implementing Cost-Effective Monitoring
Reduce CloudWatch costs without sacrificing visibility:
import boto3
logs = boto3.client('logs')
def optimize_log_retention():
"""
Optimize log retention to reduce storage costs.
Development/staging: 7-14 days
Production errors: 90 days
Production access logs: 30 days
"""
retention_policies = {
'/app/development': 7,
'/app/staging': 14,
'/app/production/errors': 90,
'/app/production/access': 30,
'/aws/lambda/batch-processor': 14,
'/aws/rds/production': 90
}
for log_group, days in retention_policies.items():
try:
logs.put_retention_policy(
logGroupName=log_group,
retentionInDays=days
)
print(f"Set {log_group} retention to {days} days")
except logs.exceptions.ResourceNotFoundException:
print(f"Log group {log_group} not found")
def disable_unnecessary_metrics():
"""
Disable detailed monitoring on non-critical resources.
Use basic monitoring (5-minute intervals) instead of detailed (1-minute).
"""
ec2 = boto3.client('ec2')
# List all instances
response = ec2.describe_instances()
development_instances = []
for reservation in response['Reservations']:
for instance in reservation['Instances']:
if instance.get('Environment') == 'development':
development_instances.append(instance['InstanceId'])
# Disable detailed monitoring for development
if development_instances:
ec2.monitor_instances(InstanceIds=development_instances)
print(f"Disabled detailed monitoring for {len(development_instances)} development instances")
if __name__ == "__main__":
optimize_log_retention()
disable_unnecessary_metrics()
Integrating CloudWatch with Other Services
CloudWatch Integration with SNS for Notifications
Connect CloudWatch alarms to SNS for flexible notification routing:
import boto3
import json
sns = boto3.client('sns')
cloudwatch = boto3.client('cloudwatch')
def create_notification_system():
"""
Create SNS topics for different severity levels and integrate with CloudWatch.
"""
severity_levels = {
'critical': 'ops-critical-alerts',
'warning': 'ops-warning-alerts',
'info': 'ops-info-notifications'
}
topics = {}
# Create SNS topics
for severity, topic_name in severity_levels.items():
response = sns.create_topic(TopicName=topic_name)
topics[severity] = response['TopicArn']
# Add subscriptions
if severity == 'critical':
sns.subscribe(
TopicArn=topics[severity],
Protocol='sms',
Endpoint='+1234567890' # On-call phone number
)
sns.subscribe(
TopicArn=topics[severity],
Protocol='email',
Endpoint='ops-critical@example.com'
)
else:
sns.subscribe(
TopicArn=topics[severity],
Protocol='email',
Endpoint='ops-team@example.com'
)
print(f"Created topic: {topic_name}")
return topics
def create_tiered_alarms(topics):
"""
Create alarms at different severity levels with appropriate actions.
"""
# Critical alarm: immediate notification
cloudwatch.put_metric_alarm(
AlarmName='DatabaseConnectionPoolExhausted',
MetricName='AvailableConnections',
Namespace='AWS/RDS',
Statistic='Average',
Period=60,
EvaluationPeriods=1,
Threshold=5,
ComparisonOperator='LessThanThreshold',
AlarmActions=[topics['critical']],
OKActions=[topics['info']]
)
print("Tiered alarms created")
if __name__ == "__main__":
topics = create_notification_system()
create_tiered_alarms(topics)
Working with Warqline
We are a cloud engineering consultancy and an official AWS and Google Cloud partner. If you are running this in production and want a second pair of eyes, we scope work in a free 45-minute technical call: you describe what you are running and what worries you, and we tell you what we would look at first.
Conclusion: Building Comprehensive Cloud Observability
Effective CloudWatch monitoring is foundational to operating reliable, performant, and cost-efficient AWS environments. By implementing the practices covered in this guide:
- Start with fundamentals: Understand metrics, logs, and alarms architecture
- Implement systematically: Use Infrastructure as Code for reproducible configurations
- Monitor comprehensively: Track infrastructure, applications, and business metrics
- Optimize continuously: Use data to reduce costs and improve performance
- Integrate strategically: Connect CloudWatch with SNS, Lambda, and other services
- Measure impact: Track monitoring ROI through faster issue resolution and reduced downtime
CloudWatch, combined with a review by our engineers, enables organizations to achieve operational excellence and maintain visibility across their entire AWS infrastructure. Start by assessing your current monitoring setup and progressively implement these best practices for sustainable, scalable observability.