Monitoring › Implementation

# uv pip install prometheus-client opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp
# uv pip install opentelemetry-instrumentation-fastapi fastapi uvicorn
# uv pip install numpy matplotlib

import numpy as np


# Simulate the prometheus_client API for demonstration
class Counter:
    def __init__(self, name, description, labelnames=()):
        self.name = name
        self._value = 0
        self._labels = {}

    def labels(self, **kwargs):
        key = tuple(sorted(kwargs.items()))
        if key not in self._labels:
            self._labels[key] = {"value": 0}
        return _LabeledCounter(self._labels[key])

    def inc(self, amount=1):
        self._value += amount


class _LabeledCounter:
    def __init__(self, store):
        self._store = store

    def inc(self, amount=1):
        self._store["value"] += amount


class Histogram:
    def __init__(self, name, description, labelnames=(), buckets=None):
        self.name = name
        self._observations = []

    def labels(self, **kwargs):
        return self

    def observe(self, value):
        self._observations.append(value)

    @property
    def mean(self):
        return np.mean(self._observations) if self._observations else 0

    @property
    def p50(self):
        return (
            np.percentile(self._observations, 50) if self._observations else 0
        )

    @property
    def p95(self):
        return (
            np.percentile(self._observations, 95) if self._observations else 0
        )

    @property
    def p99(self):
        return (
            np.percentile(self._observations, 99) if self._observations else 0
        )


class Gauge:
    def __init__(self, name, description, labelnames=()):
        self.name = name
        self._value = 0

    def set(self, value):
        self._value = value

    def inc(self, amount=1):
        self._value += amount

    def dec(self, amount=1):
        self._value -= amount


# Define the metrics registry
REQUEST_COUNTER = Counter(
    "llm_requests_total",
    "Total number of LLM requests",
    labelnames=["model", "status"],
)

REQUEST_LATENCY = Histogram(
    "llm_request_duration_seconds",
    "LLM request latency in seconds",
    labelnames=["model"],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0],
)

TTFT_HISTOGRAM = Histogram(
    "llm_time_to_first_token_seconds",
    "Time to first token in seconds",
    labelnames=["model"],
    buckets=[0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0],
)

TOKEN_COUNTER = Counter(
    "llm_tokens_total", "Total tokens processed", labelnames=["model", "type"]
)

QUALITY_SCORE = Histogram(
    "llm_response_quality_score",
    "Automated quality score for LLM responses (0-1)",
    labelnames=["model"],
    buckets=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0],
)

GPU_MEMORY_GAUGE = Gauge("llm_gpu_memory_bytes", "GPU memory currently in use")

import datetime
import random
from dataclasses import dataclass


@dataclass
class RequestMetadata:
    request_id: str
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    ttft_ms: float
    quality_score: float
    status: str
    timestamp: str


def simulate_request(
    model: str, quality_degraded: bool = False
) -> RequestMetadata:
    """Simulate a single LLM API request with realistic characteristics."""
    request_id = f"req_{random.randint(100000, 999999)}"

    # Input token distribution: most requests are short, some are long
    input_tokens = int(np.random.lognormal(mean=5.5, sigma=1.0))
    input_tokens = min(input_tokens, 8192)

    # Output tokens proportional to input with some noise
    output_tokens = int(input_tokens * random.uniform(0.3, 2.0))
    output_tokens = min(output_tokens, 4096)

    # Latency scales with total token count
    total_tokens = input_tokens + output_tokens
    base_latency = 0.5 + (total_tokens / 1000) * 2.0
    latency_ms = base_latency * 1000 * np.random.lognormal(0, 0.3)

    # Time to first token is faster (prefill phase)
    ttft_ms = latency_ms * random.uniform(0.1, 0.3)

    # Quality score: high normally, degraded if drift is simulated
    if quality_degraded:
        quality_score = random.betavariate(2, 5)  # Right-skewed toward 0
    else:
        quality_score = random.betavariate(8, 2)  # Right-skewed toward 1

    # 2% error rate
    status = "error" if random.random() < 0.02 else "success"

    timestamp = datetime.datetime.utcnow().isoformat() + "Z"

    return RequestMetadata(
        request_id=request_id,
        model=model,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        latency_ms=latency_ms,
        ttft_ms=ttft_ms,
        quality_score=quality_score,
        status=status,
        timestamp=timestamp,
    )


def record_request_metrics(req: RequestMetadata):
    """Push request metrics to the Prometheus registry."""
    REQUEST_COUNTER.labels(model=req.model, status=req.status).inc()
    REQUEST_LATENCY.labels(model=req.model).observe(req.latency_ms / 1000)
    TTFT_HISTOGRAM.labels(model=req.model).observe(req.ttft_ms / 1000)
    TOKEN_COUNTER.labels(model=req.model, type="input").inc(req.input_tokens)
    TOKEN_COUNTER.labels(model=req.model, type="output").inc(req.output_tokens)

    if req.status == "success":
        QUALITY_SCORE.labels(model=req.model).observe(req.quality_score)


# Generate a week of simulated traffic
np.random.seed(42)
n_requests = 5000
requests_normal = [
    simulate_request("gpt-4o", quality_degraded=False)
    for _ in range(int(n_requests * 0.7))
]
requests_degraded = [
    simulate_request("gpt-4o", quality_degraded=True)
    for _ in range(int(n_requests * 0.3))
]
# Preserve arrival order for the drift dashboard, then shuffle a copy for
# aggregate distributions where temporal order is irrelevant.
timeline_requests = requests_normal + requests_degraded
all_requests = timeline_requests.copy()
random.shuffle(all_requests)

# Record all requests to metrics
for req in all_requests:
    record_request_metrics(req)

# Simulate GPU memory varying over time
gpu_samples = np.random.normal(18.5, 2.0, 1000)  # GB
gpu_samples = np.clip(gpu_samples, 0, 24)

# Alert rule definitions (YAML-style config representation)
ALERT_RULES = [
    {
        "name": "HighErrorRate",
        "expr": "rate(llm_requests_total{status='error'}[5m]) / rate(llm_requests_total[5m]) > 0.05",
        "for": "2m",
        "severity": "critical",
        "summary": "Error rate exceeded 5% over the last 5 minutes",
        "runbook": "Check model backend health, review recent deployments",
    },
    {
        "name": "HighP99Latency",
        "expr": "histogram_quantile(0.99, rate(llm_request_duration_seconds_bucket[5m])) > 30",
        "for": "5m",
        "severity": "warning",
        "summary": "p99 latency exceeded 30 seconds",
        "runbook": "Check for requests with very long contexts, review GPU utilization",
    },
    {
        "name": "LowQualityScore",
        "expr": "histogram_quantile(0.5, rate(llm_response_quality_score_bucket[30m])) < 0.6",
        "for": "10m",
        "severity": "warning",
        "summary": "Median quality score dropped below 0.6 over the last 30 minutes",
        "runbook": "Review recent model outputs, check for prompt injection or adversarial inputs",
    },
    {
        "name": "HighGPUMemoryPressure",
        "expr": "llm_gpu_memory_bytes / (24 * 1e9) > 0.90",
        "for": "3m",
        "severity": "warning",
        "summary": "GPU memory utilization exceeded 90%",
        "runbook": "Review context length distribution, consider enabling memory-efficient attention",
    },
    {
        "name": "ModelBackendDown",
        "expr": "up{job='llm-inference'} == 0",
        "for": "1m",
        "severity": "critical",
        "summary": "LLM inference backend is unreachable",
        "runbook": "Check inference server process, review system logs",
    },
]

for rule in ALERT_RULES:
    print(f"Alert: {rule['name']}")
    print(f"  Severity: {rule['severity']}")
    print(f"  Condition: {rule['expr'][:60]}...")
    print(f"  Summary: {rule['summary']}")
    print()

def evaluate_quality_alert(requests_window, threshold=0.6):
    """Evaluate whether the quality alert condition is met."""
    successful = [r for r in requests_window if r.status == "success"]
    if not successful:
        return False, None

    scores = [r.quality_score for r in successful]
    median_score = float(np.median(scores))

    return median_score < threshold, median_score


# Evaluate alert on normal traffic window
alert_fired_normal, score_normal = evaluate_quality_alert(requests_normal[:500])

# Evaluate alert on degraded traffic window
alert_fired_degraded, score_degraded = evaluate_quality_alert(
    requests_degraded[:500]
)

class LLMRequestLogger:
    """Emits structured JSON logs for LLM requests."""

    def __init__(self, service_name: str, sample_rate: float = 0.05):
        self.service_name = service_name
        self.sample_rate = sample_rate
        self._log_buffer = []

    def log_request(self, req: RequestMetadata, include_text: bool = False):
        """Log a completed request."""
        # Always log metadata; only log text for sampled fraction
        should_log_text = random.random() < self.sample_rate

        entry = {
            "service": self.service_name,
            "request_id": req.request_id,
            "timestamp": req.timestamp,
            "model": req.model,
            "status": req.status,
            "latency_ms": round(req.latency_ms, 2),
            "ttft_ms": round(req.ttft_ms, 2),
            "input_tokens": req.input_tokens,
            "output_tokens": req.output_tokens,
            "quality_score": round(req.quality_score, 4),
            "text_sampled": should_log_text,
        }

        self._log_buffer.append(entry)
        return entry


logger = LLMRequestLogger(service_name="language-api", sample_rate=0.05)

# Log a sample of requests
for req in all_requests[:20]:
    logger.log_request(req)

# Show a few representative log entries
text_sampled_count = sum(1 for e in logger._log_buffer if e["text_sampled"])