Observability & SRE
Structured logging, metrics, tracing, health checks, and SLO-friendly instrumentation.
How to Create a TCP DNS Mock Server in Python
This code creates a mock TCP DNS server that listens on a specified port, accepts probe connections, and returns a fixed DNS response header to simulate a live DNS service for testing and observability.
import socket
import threading
def handle_client(client_socket, address):
print(f"[+] Connection from {address}")
try:
while True:
data = client_socket.recv(1024)
if not data:
break
print(f"[*] Received {len(data)} bytes (TCP DNS probe)")
…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 Group Alerts by Time Window in Python
Group alert occurrences that fall within a sliding time window per alert key, reducing noise and summarizing bursts into single events.
from collections import defaultdict
from datetime import datetime, timedelta
def group_alerts(alerts, window_minutes=10):
"""Group alerts that occur within the same time window."""
alerts_by_key = defaultdict(list)
for alert in alerts:
key = alert["key"]
timestamp = alert["timestamp"]…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 Implement an Error Budget Policy in Python
This code implements a mock error budget policy that decides whether to freeze deployments based on a simulated error rate and monthly freeze limits.
from datetime import datetime, timedelta
class ErrorBudgetPolicy:
FREEZE_WINDOW_HOURS = 24
MAX_FREEZES_PER_MONTH = 3
def __init__(self, budget_percentage=99.9):
self.budget_percentage = budget_percentage
self.freeze_count = 0
self.last_freeze_start = None
self.freeze_enabl…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
…How to Redact Secrets from Log Messages in Python
Build a lightweight RedactingFormatter class that replaces sensitive tokens like passwords and API keys with [REDACTED] before log messages are printed.
class RedactingFormatter:
def __init__(self, secrets):
self.secrets = secrets
def redact(self, message):
for secret in self.secrets:
message = message.replace(secret, "[REDACTED]")
return message
def format(self, record):
message = record["message"]
ret…How to Route Alerts by Severity in Python
Map alert severity levels to routing targets and simulate dispatching alerts to on-call pages, email, Slack, or logs.
def main():
# Severity levels with corresponding alert routing targets
routing_map = {
"critical": "call_page",
"high": "call_page",
"medium": "email_team",
"low": "slack_channel",
"info": "log_only"
}
# Simulated alerts with severity
alerts = [
{"na…How to Ship Logs to an Aggregator Endpoint in Python
Ship batched log entries to a mock HTTP aggregator endpoint with proper error handling and response status.
import json
import requests
from datetime import datetime, timezone
LOG_ENTRIES = [
{"timestamp": "2024-01-15T10:00:00Z", "level": "INFO", "message": "Server started"},
{"timestamp": "2024-01-15T10:00:05Z", "level": "WARN", "message": "High memory usage"},
{"timestamp": "2024-01-15T10:00:10Z", "level": "E…How to Simulate Trace Sampling Head in Python
Simulate head-based probabilistic trace sampling on mock trace data with a configurable sample rate and optional seed for reproducibility.
import random
def trace_sampling_head(mock_traces, sample_rate=0.5, seed=None):
"""Simulate probabilistic trace sampling (head-based) on mock data.
Args:
mock_traces: list of trace dictionaries with a unique 'trace_id'
sample_rate: float 0.0-1.0, probability of keeping a trace
see…How to Simulate a Queue Depth Gauge in Python
Simulate a queue depth over time using a random enqueue/dequeue process, returning depth values that can be used for monitoring or testing dashboards.
import collections
import random
import time
def simulate_queue_depth(max_depth=10, steps=20):
queue = collections.deque()
depth_history = []
for _ in range(steps):
# Randomly enqueue or dequeue
if random.random() < 0.6 and len(queue) < max_depth:
queue.append("task")
…How to Track Cache Hit Ratio in Python
Simulate an LRU cache with hit/miss tracking and compute a real-time hit ratio from random access patterns.
import random
import time
from collections import OrderedDict
class LRUCache:
def __init__(self, capacity: int):
self.cache = OrderedDict()
self.capacity = capacity
self.hits = 0
self.misses = 0
def get(self, key):
if key in self.cache:
self.hits += 1
…How to Use Log Levels DEBUG INFO WARNING ERROR in Python
Demonstrates Python's logging levels (DEBUG, INFO, WARNING, ERROR) with basicConfig and a logger, showing how severity filtering controls output.
import logging
# Configure a mock logger to demonstrate log levels
logging.basicConfig(level=logging.DEBUG, format="%(levelname)s: %(message)s")
logger = logging.getLogger("mock_logger")
# Simulate events at each severity level
logger.debug("Detailed diagnostic info")
logger.info("General system operation")
logger.w…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.