Observability &amp; SRE

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…

15 0 Open

prometheus metrics observability

Generate Prometheus Text Exposition Format in Python

Mock a Prometheus metrics endpoint by formatting metrics into the text exposition format with HELP, TYPE, and sample lines.

import time
from random import randint

# Mock a Prometheus metrics endpoint output
metrics = {
    "http_requests_total": {
        "help": "Total number of HTTP requests",
        "type": "counter",
        "samples": [
            {"labels": {"method": "get", "code": "200"}, "value": randint(1000, 9999)},
         …

13 0 Open

observability metrics time-series

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…

sre synthetic-data metrics

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…

metrics counter observability

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

13 0 Open

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…

atexit metrics graceful-shutdown

How to Flush Metrics on Graceful Shutdown in Python

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…

observability mock metrics

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 Process System Metrics (RSS, CPU) in Python

Simulate and aggregate RSS and CPU system metrics to compute averages and maximums for monitoring dashboards.

metrics rss cpu

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
      …

12 0 Open

Mocking a Metrics Gauge's set_value Method in Python

Demonstrates using unittest.mock.Mock with wraps to intercept a gauge's set_value call while verifying arguments and preserving real behavior.

unittest mocking metrics

from unittest.mock import Mock

class MetricsGauge:
    def __init__(self, name):
        self.name = name
        self.value = 0.0

    def set_value(self, new_value):
        self.value = float(new_value)
        return self.value

# Usage demonstration with a mock
gauge = MetricsGauge("cpu_usage")
gauge_mock = Mock…

13 0 Open

observability logging metrics

Python Observability Data Helper for Beginners

A beginner-friendly Python helper to log events, record metrics, summarize observability data, and export it as JSON.

import json
from datetime import datetime
from collections import defaultdict


class ObservabilityDataHelper:
    """Helper for exploring basic observability data patterns."""

    def __init__(self):
        self.events = []
        self.metrics = defaultdict(list)

    def log_event(self, service, level, message):
…

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