Observability: Metrics, Tracing & Logging at Scale

# Prometheus metrics for a web service
from prometheus_client import Counter, Histogram, start_http_server
import time, random

REQUESTS = Counter("http_requests_total", "Total HTTP requests", ["method", "endpoint"])
ERRORS = Counter("http_errors_total", "Total HTTP errors", ["status_code"])
LATENCY = Histogram("http_request_duration_seconds", "Request latency", ["method"])

def handle_request(method: str, endpoint: str):
    start = time.time()
    REQUESTS.labels(method=method, endpoint=endpoint).inc()
    try:
        time.sleep(random.uniform(0.01, 0.3))
        if random.random() < 0.05:  # 5% error rate
            raise Exception("Internal error")
        LATENCY.labels(method=method).observe(time.time() - start)
    except Exception:
        ERRORS.labels(status_code=500).inc()

print("Metrics exposed on :8000")
print("Example output at curl localhost:8000")
print("http_requests_total{method='GET',endpoint='/api/users'} 87")
print("http_errors_total{status_code='500'} 4")
print("http_request_duration_seconds_count{method='GET'} 87")

# OpenTelemetry tracing example
from opentelemetry import trace
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

provider = TracerProvider()
jaeger = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)
provider.add_span_processor(BatchSpanProcessor(jaeger))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span("process_order") as span:
    span.set_attribute("order_id", "ORD-12345")
    with tracer.start_as_current_span("validate_payment") as child:
        child.set_attribute("amount", 29.99)
        child.add_event("payment_processed")
    with tracer.start_as_current_span("update_inventory"):
        pass  # Simulated work

print("Trace sent to Jaeger. View at http://localhost:16686")

import json, logging

class StructuredLogger:
    def __init__(self, service: str):
        self.service = service
        self.logger = logging.getLogger(service)

    def log(self, level: str, message: str, **kwargs):
        entry = {
            "timestamp": logging.Formatter().formatTime(logging.LogRecord(
                "", 0, "", 0, "", (), None
            )),
            "level": level,
            "service": self.service,
            "message": message,
            **kwargs
        }
        print(json.dumps(entry))

logger = StructuredLogger("order-service")
logger.log("INFO", "Order created", order_id="ORD-123", user_id=42, amount=29.99)
logger.log("ERROR", "Payment failed", order_id="ORD-123", error_code="INSUFFICIENT_FUNDS")

{"timestamp": "2026-06-20 10:30:00,000", "level": "INFO", "service": "order-service", "message": "Order created", "order_id": "ORD-123", "user_id": 42, "amount": 29.99}
{"timestamp": "2026-06-20 10:30:00,500", "level": "ERROR", "service": "order-service", "message": "Payment failed", "order_id": "ORD-123", "error_code": "INSUFFICIENT_FUNDS"}

# alerting-rules.yml
groups:
  - name: service-health
    rules:
      - alert: HighErrorRate
        expr: rate(http_errors_total[5m]) / rate(http_requests_total[5m]) > 0.05
        for: 3m
        labels:
          severity: critical
        annotations:
          summary: "Error rate above 5% for service {{ $labels.service }}"

      - alert: HighLatency
        expr: histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m])) > 1.0
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "p99 latency above 1s"

# SLO compliance tracker
class SLO:
    def __init__(self, name: str, target: float, window_seconds: int = 604800):
        self.name = name
        self.target = target
        self.window_seconds = window_seconds
        self.good_events = 0
        self.total_events = 0

    def record(self, good: bool):
        self.total_events += 1
        if good:
            self.good_events += 1

    @property
    def compliance(self) -> float:
        if self.total_events == 0:
            return 1.0
        return self.good_events / self.total_events

    @property
    def burn_rate(self) -> float:
        error_budget = 1 - self.target
        actual_errors = 1 - self.compliance
        return actual_errors / error_budget if error_budget > 0 else 0

slo = SLO("p99-latency", target=0.999)
for _ in range(10000):
    # Simulate 99.9% good requests
    slo.record(good=random.random() < 0.999)
print(f"SLO compliance: {slo.compliance:.4%}")
print(f"Burn rate: {slo.burn_rate:.2f}")

import random, time, json, threading
from collections import defaultdict

class ObservabilityDemo:
    def __init__(self):
        self.metrics = defaultdict(int)
        self.traces = {}
        self.logs = []

    def simulate_request(self, service: str, endpoint: str):
        trace_id = f"trace-{random.randint(1000, 9999)}"
        start = time.time()
        self.metrics[f"requests:{service}:{endpoint}"] += 1
        success = random.random() > 0.05
        duration = time.time() - start
        self.traces[trace_id] = {
            "service": service,
            "duration_ms": round(duration * 1000, 2),
            "success": success
        }
        self.logs.append(json.dumps({
            "trace_id": trace_id,
            "service": service,
            "level": "INFO" if success else "ERROR",
            "message": f"Request to {endpoint} {'succeeded' if success else 'failed'}",
            "duration_ms": round(duration * 1000, 2)
        }))

    def report(self):
        print("=== Metrics ===")
        for k, v in sorted(self.metrics.items()):
            print(f"{k}: {v}")
        print("\n=== Recent Trace ===")
        tid = max(self.traces.keys())
        print(json.dumps(self.traces[tid], indent=2))
        print("\n=== Recent Log ===")
        print(self.logs[-1])

demo = ObservabilityDemo()
for _ in range(100):
    demo.simulate_request("order-svc", "/api/orders")
demo.report()