Observability &amp; SRE

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…

Observability & SRE medium

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.

burn-rate alerts time-windows

Observability & SRE medium

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

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.

histogram latency metrics

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…

Observability & SRE medium

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.

http.server histogram performance

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…

observability sre metrics

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…

sampling latency observability

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…

latency mocking http-client

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 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.

logging requests json

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…

queue simulation monitoring

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

sli availability monitoring

How to mock SLI availability success ratio in Python

Simulate request outcomes with deterministic randomness and compute the SLI availability success ratio to check if a target is met.

import random
from collections import defaultdict

def mock_availability(num_requests=1000, target_ratio=0.995):
    """
    Simulate request outcomes and compute the SLI availability success ratio.
    
    Args:
        num_requests: Total number of requests to simulate
        target_ratio: Target availability rati…

Track Success Rates and Latency in Python: SRE Metrics Helper

A beginner-friendly Python class to record request outcomes and latencies, then report success rate, average latency, and p99.

sre metrics latency

import random
import time
from collections import defaultdict


class MetricsTracker:
    """Simple helper to track success rates and latencies for SRE beginners."""

    def __init__(self):
        self.successes = 0
        self.failures = 0
        self.latencies = []

    def record(self, success, latency_ms):
   …