Observability & SRE
Structured logging, metrics, tracing, health checks, and SLO-friendly instrumentation.
Calculate Error Rate from Log Stream in Python
Parses a mock log stream to count errors and compute the error percentage using a rolling window of recent entries.
import re
from collections import deque
def error_rate_from_log_stream(message):
log_pattern = r'^\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\] (ERROR|INFO|DEBUG): (.*)$'
recent_entries = deque(maxlen=100)
error_count = 0
total_count = 0
for line in message.strip().split('\n'):
match = re.mat…Check if a Timestamp Falls in a Daily Maintenance Window in Python
A small Python function that returns True when a datetime falls inside a daily maintenance window, and a demo printing yes/no for sample timestamps.
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
def in_maintenance_window(now: datetime, start_hour: int = 2, duration_hours: int = 4) -> bool:
"""Return True if 'now' falls inside the daily maintenance window."""
day_start = now.replace(hour=start_hour, minute=0, second=0, microsecond…Generate Mock CPU and Memory Metrics in Python
Build a mock_host_metrics() generator that outputs realistic CPU and memory usage percentages for monitoring demos and tests.
import time
import random
def mock_host_metrics():
"""Generate mock CPU and memory metrics for a host."""
cpu_percent = round(random.uniform(10.0, 95.0), 1)
memory_percent = round(random.uniform(20.0, 90.0), 1)
memory_used_mb = round(random.uniform(512, 8192), 1)
return {
"timestamp": in…Generate Synthetic CPU Utilization Metrics in Python
Creates realistic time-series CPU utilization samples with timestamps, noise, and output as structured JSON for observability demos and testing.
from datetime import datetime, timedelta
import random
import json
def generate_metric_samples(base_value, noise, count=60, interval_minutes=1):
"""Generate realistic CPU utilization samples for a given time window."""
timestamps = []
values = []
now = datetime.utcnow()
start_time = now - timede…Generate Synthetic SRE Metrics and Calculate Availability in Python
Create realistic service metrics with random latency, error rate, and request counts, then compute availability and summarize the stream for SLO checks.
from datetime import datetime, timedelta
import random
def generate_service_metrics(service_name: str, minutes: int = 30) -> list[dict]:
"""Generate synthetic SRE metrics for a service across recent minutes."""
metrics = []
now = datetime.now()
for i in range(minutes):
timestamp = now - t…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 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 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 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…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.