Observability &amp; SRE

import logging
import uuid
from contextlib import contextmanager

logging.basicConfig(level=logging.INFO, format='%(levelname)s | %(correlation_id)s | %(message)s')


@contextmanager
def correlation_id_context(correlation_id):
    """Temporarily inject a correlation_id into the logging context."""
    extra = {'correl…

16 0 Open

otlp metrics observability

Export Metrics with OTLP Mock in Python

Simulates system metric collection and exports them as an OTLP-like JSON payload using only Python's standard library.

from dataclasses import dataclass, asdict
import json
import random
import time


@dataclass
class Metric:
    name: str
    value: float
    timestamp: int
    unit: str = "1"


def collect_system_metrics() -> list[Metric]:
    """Mock metric collection for OTLP export simulation."""
    now = int(time.time())
    re…

14 0 Open

burn-rate alerts time-windows

How to Build a Burn Rate Alert with Multiple Time Windows in Python

Track token consumption and trigger alerts when the burn rate exceeds a threshold across multiple time windows using deque and time-based sliding windows.

import time
from collections import deque

class BurnRateAlert:
    def __init__(self, windows_seconds=(60, 300, 900), threshold_rate=0.8):
        self.windows = {w: deque() for w in windows_seconds}
        self.threshold_rate = threshold_rate
        self.previous_tokens = None

    def record_sample(self, current_…

16 0 Open

histogram latency metrics

How to Build a Python Latency Histogram with Mock Buckets

This code implements a mock latency histogram that records request durations into configurable buckets and outputs counts, total, and average latency.

import time
import random
from collections import Counter


class LatencyHistogram:
    def __init__(self, buckets):
        self.buckets = sorted(buckets)
        self.counts = Counter()
        self.total = 0
        self.sum_latency = 0

    def record(self, latency_ms):
        for i, boundary in enumerate(self.bu…

http.server histogram performance

How to Build an HTTP Server Request Duration Histogram in Python

Create a small HTTP server that times each GET request, buckets the duration, and prints a histogram on shutdown.

import time
import random
from collections import Counter
from http.server import HTTPServer, BaseHTTPRequestHandler


class HistogramHandler(BaseHTTPRequestHandler):
    response_times = Counter()

    def do_GET(self):
        start = time.perf_counter()
        time.sleep(random.uniform(0.001, 0.1))
        duratio…

opentelemetry tracing testing

How to Create a Mock OpenTelemetry Trace in Python

Create a mock OpenTelemetry trace in memory to test span creation, attributes, and parent-child relationships without exporting to a backend.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter


def create_mock_trace():
    tracer_provider = TracerProvider()
    span_exporter =…

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.

socket dns tcp

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)")
          …

16 0 Open

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.

cache lru hit-ratio

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
     …