Observability & SRE
Structured logging, metrics, tracing, health checks, and SLO-friendly instrumentation.
How to Add Metadata Attributes to a Span in Python
Create a lightweight dataclass-based Span mock that stores key-value metadata attributes for tracing or event logging.
from dataclasses import dataclass, field
from typing import Dict, Any
@dataclass
class Span:
name: str
attributes: Dict[str, Any] = field(default_factory=dict)
def set_attribute(self, key: str, value: Any) -> None:
self.attributes[key] = value
def get_attribute(self, key: str) -> Any…How to Build a Consumer Lag Gauge in Python
Simulate Kafka consumer lag with a Python class that tracks lag over time and reports health and averages.
import time
import random
from collections import deque
class ConsumerLagGauge:
"""Mock consumer lag gauge measuring how far behind a consumer is."""
def __init__(self, producer_rate=10, consumer_rate=7, initial_lag=0):
self.producer_rate = producer_rate
self.consumer_rate = consumer_rate
…How to Build a Metrics Counter with Increment and Snapshot in Python
A simple dict-backed MetricsCounter class that increments named counters and returns a snapshot of the current values.
class MetricsCounter:
def __init__(self):
self._metrics = {}
def increment(self, key, delta=1):
self._metrics[key] = self._metrics.get(key, 0) + delta
def snapshot(self):
return dict(self._metrics)
if __name__ == "__main__":
counter = MetricsCounter()
counter.increment("…How to Calculate Apdex Score from Latency Data in Python
Generate simulated latency samples and compute the Apdex score to gauge user satisfaction with an application's performance.
import random
import statistics
def generate_latencies(count=100, base=100, stddev=30):
return [max(0, random.gauss(base, stddev)) for _ in range(count)]
def apdex(latencies, threshold=200):
satisfied = sum(1 for lat in latencies if lat < threshold)
tolerating = sum(1 for lat in latencies if lat >= thres…How to Calculate Percentile Latency in Python
Generate mock latency samples with occasional spikes and compute 50th, 90th, 95th, and 99th percentile values in milliseconds.
import random
import statistics
def generate_latency_samples(n=1000):
"""Generate realistic mock latency data (ms) with occasional spikes."""
samples = []
for _ in range(n):
# Normal case: ~50ms with jitter
base = random.gauss(50, 5)
# 2% spike chance: slow downstream or GC pause
…How to Calculate SLO Error Budget in Python
Simulate an SLO error budget by computing allowed downtime from a target availability percentage and mocking monthly incidents.
```python
import random
def calculate_error_budget(total_seconds: int, target_availability: float) -> float:
return (1.0 - target_availability) * total_seconds
def simulate_monthly_availability(seconds_in_month: int, budget_seconds: float) -> float:
# Mock: randomly consume a fraction of the error budget i…How to Check Service Readiness Dependencies in Python
This code simulates a readiness check for external dependencies (database, cache, queue) with mock availability data and reports readiness status.
import sys
from datetime import datetime
def check_dependencies(config):
results = []
for dep, required in config.items():
available = mock_availability(dep)
status = "READY" if available >= required else "NOT READY"
results.append((dep, available, required, status))
return result…How to Compute SRE Metrics Like Error Rate and Availability in Python
Tracks log events in a sliding time window and calculates error rate per second and availability percentage using an easy-to-follow class.
from collections import deque
from datetime import datetime, timedelta
from typing import Dict, Deque
class LogMetrics:
"""Simple observability helper to track log events and calculate SRE metrics."""
def __init__(self, window_seconds: int = 60):
self.window_seconds = window_seconds
self.eve…How to Create a Deep Health Check Database in Python
Setup a SQLite-backed health check database, insert mock data with response times and statuses, and generate a report ordered by most recent check.
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path
DB_PATH = Path("deep_health_check.db")
def setup_database():
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS health_checks (
id INTEGER PRIMARY KEY AU…How to Create a Deployment Environment Tag in Python
Generate a standardized deployment tag string by combining service and environment names with an f-string.
def mock_env_tag(service, environment):
return f"{service}-{environment}"
if __name__ == "__main__":
service = "api-gateway"
environment = "production"
tag = mock_env_tag(service, environment)
print(f"Deployment tag: {tag}")How to Do Structured JSON Line Logging in Python
Create a simple JSON-lines logger that writes one JSON object per line to stdout with timestamp, level, message, and custom context fields.
import json
import sys
from datetime import datetime
class JsonLineLogger:
def __init__(self, stream=sys.stdout):
self.stream = stream
def log(self, level, message, **context):
record = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"level": level,
"me…How to Do Structured JSON Logging in Python
Create a custom logging formatter that outputs each log entry as a single JSON line with timestamp, level, logger name, and message.
import json
import logging
from datetime import datetime
class JsonFormatter(logging.Formatter):
def format(self, record):
log_entry = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"level": record.levelname,
"logger": record.name,
"message": record.ge…How to Flush Metrics on Graceful Shutdown in Python
Register an atexit handler to automatically flush collected metrics when a Python process exits gracefully.
import atexit
import time
import random
class MetricsCollector:
def __init__(self):
self._metrics = []
atexit.register(self.flush)
def record(self, name, value):
self._metrics.append((name, value, time.time()))
def flush(self):
print(f"Flushing {len(self._metrics)} metri…How to Generate and Propagate W3C Trace Context Headers in Python
Generate and propagate W3C traceparent and tracestate headers for distributed tracing in Python, with mock service headers.
import uuid
def generate_w3c_traceparent(trace_id=None, parent_id=None, flags="01"):
if trace_id is None:
trace_id = uuid.uuid4().hex[:32]
if parent_id is None:
parent_id = uuid.uuid4().hex[:16]
return f"00-{trace_id}-{parent_id}-{flags}"
def create_mock_headers(service_name, trace_id=N…How to Implement Tail Sampling in Python
Sample the slowest subset of calls (tail) for latency analysis using a deque with a random ratio gate.
import random
import time
from collections import deque
class TailSampler:
def __init__(self, tail_ratio=0.1, max_samples=100):
self.tail_ratio = tail_ratio
self.max_samples = max_samples
self.samples = deque(maxlen=max_samples)
self.total_calls = 0
def record(self, latency_ms…How to Link Parent and Child Span Elements in Python
This code defines a lightweight mock element class and a function that links child elements to a parent when their ranges are nested within the parent's range.
class MockElement:
def __init__(self, name, start, end, children=None):
self.name = name
self.start = start
self.end = end
self.children = children or []
def __repr__(self):
return f"MockElement({self.name}, {self.start}-{self.end})"
def link_parent_child(parent, chil…How to Mock Database Query Duration in Python
Simulate realistic database query durations with random jitter for testing dashboards, alerts, and SLO calculations.
import random
import time
def mock_query_duration(db_name, avg_ms, jitter_ms=5, runs=3):
"""Simulate database query durations with realistic variation."""
durations = []
for _ in range(runs):
# Base duration plus random jitter (can be negative)
duration = avg_ms + random.uniform(-jitter_m…How to Mock HTTP Client Latency in Python
Simulate outbound HTTP request latency with configurable ranges to test timeouts, retries, and SLO monitoring without external services.
import time
import random
def mock_latency(host: str, min_ms: int = 100, max_ms: int = 500) -> dict:
"""Simulate an outbound HTTP request with mock latency."""
latency_ms = random.randint(min_ms, max_ms)
start = time.perf_counter()
time.sleep(latency_ms / 1000)
elapsed_ms = (time.perf_counter() - …How to Mock Service Resource Attributes in Python
Temporarily override service name, version, and other resource attributes with a context manager, then restore them automatically.
from contextlib import contextmanager
import random
_SERVICE_ATTRIBUTES = {
"service.name": "payment-api",
"service.version": "1.4.2",
"service.instance.id": str(random.randint(10000, 99999)),
"service.namespace": "production",
}
@contextmanager
def mock_service_attributes(**overrides):
"""Tempor…How to Mock a Baggage Context (Key-Value Store) in Python
This code implements an in-memory key-value mock of a baggage context, letting you set, get, check, and delete keys for tracing-style metadata.
class BaggageContext:
def __init__(self):
self._store = {}
def set(self, key, value):
self._store[key] = value
return value
def get(self, key, default=None):
return self._store.get(key, default)
def has(self, key):
return key in self._store
def delete(sel…How to Mock an OTLP HTTP Endpoint in Python
This code implements a lightweight HTTP server that accepts OTLP/HTTP trace exports, stores spans by trace ID, and exposes them via a simple GET endpoint for debugging.
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from collections import defaultdict
class TraceHandler(BaseHTTPRequestHandler):
traces = defaultdict(list)
def do_POST(self):
if self.path == "/v1/traces":
length = int(self.headers.get("Content-Length", 0))
…How to Model Span Events in Python
Define a Span class with timestamped milestone events and a completion marker to track operation lifecycle.
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import List
class SpanStatus(Enum):
STARTED = "started"
COMPLETED = "completed"
@dataclass
class SpanEvent:
name: str
timestamp: float = field(default_factory=time.time)
attributes: dict = field(default_facto…How to Parse Log Lines with Regex in Python
Extracts timestamp, log level, service name, and message from a log line using compiled regex named groups.
import re
LOG_PATTERN = re.compile(
r'^(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) '
r'\[(?P<level>\w+)\] '
r'\((?P<service>[^)]+)\) '
r'(?P<message>.*)$'
)
def parse_log_line(line: str) -> dict:
match = LOG_PATTERN.match(line)
if not match:
return {"error": "invalid log format…How to Process System Metrics (RSS, CPU) in Python
Simulate and aggregate RSS and CPU system metrics to compute averages and maximums for monitoring dashboards.
import random
import time
from collections import namedtuple
Metric = namedtuple("Metric", ["name", "value", "unit"])
def generate_metrics(num_metrics: int = 5) -> list:
"""Simulate a batch of system metrics."""
metrics = []
for i in range(num_metrics):
rss = random.randint(50, 500) # MB
…Browse by section
Each section groups closely related Python snippets.
Observability & SRE — Python code examples
What you will find here
This page collects observability & sre snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.
Samples vs tutorials and challenges
Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.