prometheus_demo

prometheus_demo.py

#!/usr/bin/env python3
"""Prometheus Demo - All 4 metric types: Counter, Gauge, Histogram, Summary"""

# before >>> pip install prometheus-client>=0.19.0
# document >>> https://opentelemetry-python.readthedocs.io/en/latest/api/metrics.html

import time
import random
from prometheus_client import Counter, Gauge, Histogram, Summary, start_http_server

# Metrics
http_requests = Counter('http_requests_total', 'Total requests', ['method', 'status'])
errors = Counter('errors_total', 'Total errors', ['type'])
connections = Gauge('active_connections', 'Active connections')
memory = Gauge('memory_bytes', 'Memory usage')
request_duration = Histogram('request_duration_seconds', 'Request duration')
response_size = Histogram('response_size_bytes', 'Response size', buckets=[100, 1000, 10000, 100000])
latency = Summary('latency_seconds', 'Request latency')
db_duration = Summary('db_duration_seconds', 'DB query duration', ['type'])


def simulate():
    """Simulate requests and metrics"""
    method = random.choice(['GET', 'POST', 'PUT'])
    status = random.choices(['200', '400', '500'], weights=[80, 15, 5])[0]
    
    # Counter
    http_requests.labels(method=method, status=status).inc()
    if status != '200':
        errors.labels(type='client' if status == '400' else 'server').inc()
    
    # Gauge
    connections.set(max(0, connections._value._value + random.choice([-1, 0, 1])))
    if random.random() < 0.3:
        memory.set(500_000_000 + random.randint(-50_000_000, 100_000_000))
    
    # Histogram
    request_duration.observe(random.uniform(0.01, 2.0))
    response_size.observe(random.randint(100, 100000))
    
    # Summary
    latency.observe(random.uniform(0.01, 1.5))
    db_duration.labels(type=random.choice(['SELECT', 'INSERT'])).observe(random.uniform(0.01, 0.5))


def main():
    print("Prometheus metrics: http://localhost:8000/metrics\n")
    start_http_server(8000)
    
    connections.set(10)
    memory.set(500_000_000)
    
    try:
        while True:
            for _ in range(random.randint(1, 3)):
                simulate()
            time.sleep(0.2)
    except KeyboardInterrupt:
        print("\nStopped")


if __name__ == '__main__':
    main()