Lambda Monitoring with CloudWatch — Observability and Debugging
In this tutorial, you will learn about Lambda Monitoring with CloudWatch. We cover key concepts, practical examples, and best practices to help you master this topic.
AWS CloudWatch provides monitoring for Lambda functions including metrics for invocations, errors, duration, throttles, and concurrent executions with customizable alarms and dashboards.
What You'll Learn
By the end of this lesson you will understand Lambda monitoring with CloudWatch metrics, structured logging best practices, CloudWatch alarms, X-Ray tracing, and building observability dashboards.
Why It Matters
Serverless functions are ephemeral -- you cannot SSH into a running function to debug. Monitoring and logging are your only visibility into function behavior. Without proper observability, you cannot diagnose failures, optimize performance, or detect issues.
Real-World Use
DodaZIP's serverless conversion pipeline uses CloudWatch dashboards showing invocation rate, error rate, average duration, and cold start frequency. CloudWatch alarms trigger Slack notifications when error rates exceed 1 percent.
flowchart LR
L[AWS Lambda] --> CW[CloudWatch]
CW --> M[Metrics]
CW --> L1[Logs]
CW --> T[X-Ray Traces]
M --> AL[Alarms]
AL --> N[SNS Notification]
L1 --> I[Logs Insights]
I --> Q[Query Analysis]
style CW fill:#f90,color:#fff
Key Lambda Metrics
Lambda publishes metrics to CloudWatch automatically: Invocations, Errors, Duration, Throttles, ConcurrentExecutions, and IteratorAge for stream sources.
# lambda_metrics.py
# Understanding Lambda metrics
def describe_metrics():
metrics = [
("Invocations", "Number of times the function is invoked", "Count"),
("Errors", "Number of invocations that failed", "Count"),
("Duration", "Execution time in milliseconds", "ms"),
("Throttles", "Invocations rate-limited by concurrency", "Count"),
("ConcurrentExecutions", "Active invocations at a point in time", "Count"),
("IteratorAge", "Age of oldest record for stream sources", "ms"),
("ProvisionedConcurrencyInvocations", "Invocations served by provisioned concurrency", "Count"),
("ColdStarts", "Invocations that required a new environment", "Count"),
]
print(f"{'Metric':35s} {'Description':55s} {'Unit':10s}")
print("-" * 100)
for name, desc, unit in metrics:
print(f"{name:35s} {desc:55s} {unit:10s}")
describe_metrics()
Expected output:
Metric Description Unit
----------------------------------------------------------------------------------------------------
Invocations Number of times the function is invoked Count
Errors Number of invocations that failed Count
...
Structured Logging
Log to stdout using structured JSON for querying with CloudWatch Logs Insights.
# structured_logging.py
# Structured JSON logging for CloudWatch
import json
import time
class StructuredLogger:
def __init__(self, service_name, version="1.0"):
self.service = service_name
self.version = version
def info(self, message, **extra):
self._log("INFO", message, **extra)
def error(self, message, **extra):
self._log("ERROR", message, **extra)
def warn(self, message, **extra):
self._log("WARN", message, **extra)
def _log(self, level, message, **extra):
entry = {
"timestamp": time.time(),
"level": level,
"service": self.service,
"version": self.version,
"message": message,
**extra
}
print(json.dumps(entry))
logger = StructuredLogger("order-service")
logger.info("Order created", order_id="ORD-001", amount=49.99, currency="USD")
logger.error("Payment failed", order_id="ORD-001", error_code="card_declined", attempt=2)
logger.warn("High latency detected", duration_ms=8500, threshold=5000)
Expected output:
{"timestamp": ..., "level": "INFO", "service": "order-service", "message": "Order created", "order_id": "ORD-001", "amount": 49.99, "currency": "USD"}
{"timestamp": ..., "level": "ERROR", "service": "order-service", "message": "Payment failed", "order_id": "ORD-001", "error_code": "card_declined", "attempt": 2}
{"timestamp": ..., "level": "WARN", "service": "order-service", "message": "High latency detected", "duration_ms": 8500, "threshold": 5000}
CloudWatch Alarms
Set alarms on key metrics to detect anomalies and trigger notifications.
# cloudwatch_alarms.py
# Configuring CloudWatch alarms
class CloudWatchAlarm:
def __init__(self, name, metric, threshold, period, evaluation_periods):
self.name = name
self.metric = metric
self.threshold = threshold
self.period = period
self.evaluations = evaluation_periods
def evaluate(self, value):
alarm_state = "ALARM" if value > self.threshold else "OK"
print(f"[{alarm_state}] {self.name}: {self.metric}={value} (threshold={self.threshold})")
return alarm_state
alarms = [
CloudWatchAlarm("High Error Rate", "Errors", 5, 300, 2),
CloudWatchAlarm("High Duration", "Duration", 5000, 300, 3),
CloudWatchAlarm("Throttle Rate", "Throttles", 10, 60, 1),
]
# Simulate metric evaluation
alarms[0].evaluate(3) # OK
alarms[0].evaluate(7) # ALARM - trigger notification
alarms[2].evaluate(15) # ALARM - concurrency limit near
X-Ray Tracing
Enable X-Ray tracing on Lambda to see end-to-end request flow and identify bottlenecks.
# xray_tracing.py
# Understanding X-Ray tracing for Lambda
def simulate_xray_trace():
trace = {
"trace_id": "1-abc123-def456",
"segments": [
{
"name": "API Gateway",
"duration_ms": 45,
"annotations": {"method": "GET", "path": "/users"}
},
{
"name": "Lambda Handler",
"duration_ms": 320,
"subsegments": [
{"name": "Auth Check", "duration_ms": 15},
{"name": "DynamoDB Query", "duration_ms": 120},
{"name": "External API Call", "duration_ms": 150}
]
},
{
"name": "DynamoDB",
"duration_ms": 110,
"annotations": {"table": "users", "operation": "Query"}
}
]
}
total = sum(s["duration_ms"] for s in trace["segments"])
print(f"Trace: {trace['trace_id']}")
print(f"Total duration: {total}ms")
print(f"\nSegments:")
for seg in trace["segments"]:
pct = seg["duration_ms"] / total * 100
print(f" {seg['name']:20s} {seg['duration_ms']:6}ms ({pct:.0f}%)")
for sub in seg.get("subsegments", []):
print(f" {sub['name']:18s} {sub['duration_ms']:6}ms")
simulate_xray_trace()
Common Mistakes
Not using structured JSON logs: Plain text logs cannot be queried with Logs Insights. Use JSON format for all log output.
Ignoring error metrics: Without monitoring Error and Throttle metrics, failures go undetected until users report them.
Not setting duration alarms: Functions that approach the timeout limit indicate performance issues. Set alarms at 80 percent of timeout.
Logging sensitive information: Never log passwords, tokens, or personal data. Use structured logging with field redaction.
Not using X-Ray for distributed tracing: Without tracing, debugging latency across multiple services requires manual correlation of timestamps.
Practice Questions
What metrics does Lambda publish to CloudWatch automatically? Invocations, Errors, Duration, Throttles, ConcurrentExecutions, IteratorAge, and DeadLetterErrors.
Why use structured JSON logging? CloudWatch Logs Insights can query JSON fields, enabling analysis like "find all errors in the last hour grouped by error code."
What is X-Ray tracing used for? Tracing end-to-end request flow across Lambda, API Gateway, DynamoDB, and other services to identify bottlenecks.
How do you set up error notifications? Create a CloudWatch alarm on the Errors metric with SNS topic as the notification target.
Challenge: Design a monitoring dashboard for a serverless application showing invocation volume, error rate, p95 duration, cold start rate, and concurrent executions.
FAQ
Mini Project
Create a monitoring setup for a serverless application that logs structured JSON, creates CloudWatch alarms on error rate and duration, and sets up a dashboard.
import json
import random
class MonitoringSimulation:
def __init__(self, function_name):
self.function = function_name
self.logs = []
def invoke(self, success=True, duration_ms=200):
if success:
level = "INFO"
else:
level = "ERROR"
log = {"level": level, "function": self.function, "duration": duration_ms, "success": success}
self.logs.append(log)
print(json.dumps(log))
def analyze(self):
total = len(self.logs)
errors = sum(1 for l in self.logs if not l["success"])
durations = [l["duration"] for l in self.logs]
p95 = sorted(durations)[int(len(durations) * 0.95)] if durations else 0
print(f"\n=== Analysis for {self.function} ===")
print(f"Total invocations: {total}")
print(f"Error rate: {errors/total*100:.1f}%")
print(f"Average duration: {sum(durations)/len(durations):.0f}ms")
print(f"P95 duration: {p95}ms")
mon = MonitoringSimulation("order-processor")
for _ in range(100):
ok = random.random() > 0.05
dur = random.randint(100, 3000)
mon.invoke(success=ok, duration_ms=dur)
mon.analyze()
What's Next
Next: Serverless Framework for tooling and deployment.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro