Cloud Monitoring and Observability Guide — Metrics, Logs, Traces, and Alerting

# cloudwatch_demo.py
# Simulate CloudWatch metrics and alarm evaluation
import time
import random
from datetime import datetime, timedelta
from collections import deque

class CloudWatchMetric:
    def __init__(self, namespace, metric_name, unit="Count"):
        self.namespace = namespace
        self.metric_name = metric_name
        self.unit = unit
        self.datapoints = deque(maxlen=100)

    def put_data(self, value, timestamp=None):
        if timestamp is None:
            timestamp = datetime.now()
        self.datapoints.append({"timestamp": timestamp, "value": value})

    def get_statistics(self, statistic, period_minutes=5):
        cutoff = datetime.now() - timedelta(minutes=period_minutes)
        recent = [d for d in self.datapoints if d["timestamp"] >= cutoff]
        values = [d["value"] for d in recent]
        if not values:
            return 0
        if statistic == "Average":
            return sum(values) / len(values)
        elif statistic == "Sum":
            return sum(values)
        elif statistic == "Maximum":
            return max(values)
        elif statistic == "Minimum":
            return min(values)
        return values[-1]

class CloudWatchAlarm:
    def __init__(self, name, metric, operator, threshold, period_minutes=5, alarm_actions=None):
        self.name = name
        self.metric = metric
        self.operator = operator
        self.threshold = threshold
        self.period = period_minutes
        self.alarm_actions = alarm_actions or []
        self.state = "OK"
        self.evaluation_count = 0

    def evaluate(self):
        self.evaluation_count += 1
        value = self.metric.get_statistics("Average", self.period)

        if self.operator == "GreaterThanThreshold" and value > self.threshold:
            new_state = "ALARM"
        elif self.operator == "LessThanThreshold" and value < self.threshold:
            new_state = "ALARM"
        else:
            new_state = "OK"

        if new_state != self.state:
            print(f"  [STATE CHANGE] {self.name}: {self.state} → {new_state} (value: {value:.1f})")
            if new_state == "ALARM":
                for action in self.alarm_actions:
                    print(f"  [ACTION] {action}")
            self.state = new_state

        return self.state

# Simulate monitoring
cpu_metric = CloudWatchMetric("AWS/EC2", "CPUUtilization", "Percent")
cpu_alarm = CloudWatchAlarm(
    name="High-CPU-Alarm",
    metric=cpu_metric,
    operator="GreaterThanThreshold",
    threshold=80,
    period_minutes=5,
    alarm_actions=["SNS: notify-ops@dodatech.com", "AutoScaling: add-instance"],
)

print("=== CloudWatch Monitoring Simulation ===\n")
for i in range(20):
    cpu = random.uniform(30, 95)
    cpu_metric.put_data(cpu)
    print(f"  Minute {i+1:>2}: CPU = {cpu:>5.1f}%  Alarm state: {cpu_alarm.evaluate()}")
    time.sleep(0.1)

print(f"\nFinal alarm state: {cpu_alarm.state}")
print(f"Total evaluations: {cpu_alarm.evaluation_count}")

# azure_monitor_demo.py
# Simulate Azure Monitor metrics and log queries
from datetime import datetime
import json

class AzureMonitor:
    def __init__(self, resource_id):
        self.resource_id = resource_id
        self.metrics = {}
        self.logs = []

    def record_metric(self, name, value, unit="Count"):
        if name not in self.metrics:
            self.metrics[name] = []
        self.metrics[name].append({"timestamp": datetime.now().isoformat(), "value": value, "unit": unit})

    def log_event(self, level, message, properties=None):
        self.logs.append({
            "timestamp": datetime.now().isoformat(),
            "resource": self.resource_id,
            "level": level,
            "message": message,
            "properties": properties or {},
        })

    def query(self, kql_query):
        """Simulate Kusto Query Language (KQL) on logs."""
        results = []
        for log in self.logs:
            if log["level"] in kql_query.get("levels", ["ERROR", "WARN"]):
                results.append(log)
        return results

    def create_alert(self, metric_name, condition, threshold, action_group):
        """Simulate creating an alert rule."""
        current_value = self.metrics.get(metric_name, [{}])[-1].get("value", 0)
        triggered = False
        if condition == "greater_than" and current_value > threshold:
            triggered = True
        elif condition == "less_than" and current_value < threshold:
            triggered = True

        if triggered:
            print(f"[ALERT] {metric_name} triggered {condition} {threshold} (current: {current_value})")
            print(f"[ACTION] Notifying {action_group}")
        return triggered

monitor = AzureMonitor("/subscriptions/demo/resourceGroups/prod/providers/Microsoft.Compute/virtualMachines/web-01")

print("=== Azure Monitor Demo ===\n")

# Record metrics
for i in range(10):
    monitor.record_metric("Percentage CPU", 50 + i * 5, "Percent")
    monitor.record_metric("Available Memory MB", 2048 - i * 100, "Bytes")

# Log events
monitor.log_event("INFO", "Application started", {"version": "1.0.0"})
monitor.log_event("WARN", "Memory usage above 75%", {"memory_mb": 1536})
monitor.log_event("ERROR", "Connection pool exhausted", {"pool_size": 10, "active": 10})
monitor.log_event("INFO", "Request completed", {"path": "/api/users", "duration_ms": 245})

# Query logs
errors = monitor.query({"levels": ["ERROR", "WARN"]})
print("Recent Errors/Warnings:")
for log in errors:
    print(f"  [{log['level']}] {log['message']}")

# Create alert
print("\nCreating alerts...")
monitor.create_alert("Percentage CPU", "greater_than", 80, "ops-team-email")
monitor.create_alert("Available Memory MB", "less_than", 500, "ops-team-sms")

# Create a CloudWatch alarm for high CPU on an EC2 instance
aws cloudwatch put-metric-alarm \
  --alarm-name "web-prod-high-cpu" \
  --alarm-description "CPU > 80% for 5 minutes" \
  --metric-name CPUUtilization \
  --namespace AWS/EC2 \
  --statistic Average \
  --period 300 \
  --evaluation-periods 2 \
  --threshold 80 \
  --comparison-operator GreaterThanThreshold \
  --dimensions Name=InstanceId,Value=i-1234567890abcdef0 \
  --alarm-actions arn:aws:sns:us-east-1:123456789012:ops-alerts \
  --ok-actions arn:aws:sns:us-east-1:123456789012:ops-alerts

# monitoring_dashboard.py
# Build a simple monitoring dashboard simulation
from datetime import datetime
import random

class Dashboard:
    def __init__(self, name):
        self.name = name
        self.widgets = []

    def add_widget(self, title, metric_func, unit=""):
        self.widgets.append({"title": title, "metric": metric_func, "unit": unit})

    def render(self):
        print(f"\n{'='*60}")
        print(f"  {self.name}")
        print(f"  Last updated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"{'='*60}\n")
        for widget in self.widgets:
            value = widget["metric"]()
            print(f"  📊 {widget['title']:<25} {value:>10.1f} {widget['unit']}")

dash = Dashboard("DodaTech Production Overview")
dash.add_widget("Web Servers (req/s)", lambda: random.uniform(150, 350), "req/s")
dash.add_widget("API Latency (p99)", lambda: random.uniform(45, 200), "ms")
dash.add_widget("Error Rate", lambda: random.uniform(0.1, 2.5), "%")
dash.add_widget("Database Connections", lambda: random.randint(10, 50), "conn")
dash.add_widget("Daily Active Users", lambda: random.randint(8000, 12000), "users")
dash.add_widget("S3 Storage (TB)", lambda: 12.4 + random.uniform(-0.1, 0.1), "TB")
dash.add_widget("Lambda Invocations/min", lambda: random.randint(200, 800), "inv")
dash.add_widget("Estimated Monthly Cost", lambda: random.uniform(4500, 5200), "USD")
dash.render()

# observability_dashboard.py
# Build a multi-service observability dashboard
import random
from datetime import datetime, timedelta

class Service:
    def __init__(self, name):
        self.name = name
        self.metrics = {"request_rate": 0, "error_rate": 0, "latency_p99": 0, "health": "healthy"}

    def collect(self):
        self.metrics["request_rate"] = random.randint(50, 500)
        self.metrics["error_rate"] = random.uniform(0, 3)
        self.metrics["latency_p99"] = random.uniform(20, 500)
        self.metrics["health"] = "degraded" if self.metrics["error_rate"] > 2 or self.metrics["latency_p99"] > 400 else "healthy"

class ObservabilityDashboard:
    def __init__(self):
        self.services = []

    def add_service(self, service):
        self.services.append(service)

    def refresh(self):
        print(f"\n=== Observability Dashboard [{datetime.now().strftime('%H:%M:%S')}] ===")
        print(f"{'Service':<20} {'Req/s':<10} {'Errors':<10} {'p99 (ms)':<12} {'Status'}")
        print("-" * 65)
        for svc in self.services:
            svc.collect()
            m = svc.metrics
            icon = "✓" if m["health"] == "healthy" else "⚠"
            print(f"{icon} {svc.name:<18} {m['request_rate']:<10} {m['error_rate']:<10.1f}% {m['latency_p99']:<12.0f}{m['health']}")

dash = ObservabilityDashboard()
for name in ["api-gateway", "user-service", "order-service", "payment-service", "notification-service", "analytics-service", "ml-inference"]:
    dash.add_service(Service(name))

dash.refresh()
dash.refresh()