Observability Telemetry

Observability & Telemetry

Comprehensive observability practices for production systems: OpenTelemetry instrumentation (Python + TypeScript), Prometheus metrics and alerting, Grafana dashboards, structured logging, distributed tracing, RAG/LLM-specific monitoring, event stream consumers, error tracking with Sentry, chaos engineering, and infrastructure-as-code for observability.

When to Use

• Instrumenting microservices, APIs, or ML systems for production observability
• Setting up metrics, traces, and logs with OpenTelemetry exporters
• Building Grafana dashboards for application, API, and ML workload monitoring
• Implementing structured logging with correlation IDs and cost controls
• Tracing distributed requests across service boundaries with span baggage
• Monitoring RAG/LLM systems: embedding latency, retrieval hit-rate, hallucination rates
• Tracking Kafka/Redpanda consumer health, dead-letter queues, and replay strategies
• Integrating Sentry for error tracking, alert routing, and deduplication
• Running chaos drills and maintaining operational runbooks
• Provisioning observability infrastructure with Terraform and dashboards as code

Core Concepts

1. OpenTelemetry SDK Setup

Python SDK:

# otel_setup.py
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.resources import Resource
import os

def setup_otel(service_name: str, service_version: str):
    """Initialize OpenTelemetry for Python services."""
    resource = Resource.create({
        "service.name": service_name,
        "service.version": service_version,
        "deployment.environment": os.environ.get("ENVIRONMENT", "production"),
    })

    # Tracing
    tracer_provider = TracerProvider(resource=resource)
    otlp_exporter = OTLPSpanExporter(
        endpoint=os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "otel-collector:4317"),
        insecure=True,  # Use TLS in production
    )
    tracer_provider.add_span_processor(BatchSpanProcessor(otlp_exporter))
    trace.set_tracer_provider(tracer_provider)

    # Metrics
    metric_reader = PeriodicExportingMetricReader(
        OTLPMetricExporter(
            endpoint=os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "otel-collector:4317"),
            insecure=True,
        ),
        export_interval_millis=10000,  # Export every 10s
    )
    meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
    metrics.set_meter_provider(meter_provider)

    return trace.get_tracer(service_name), metrics.get_meter(service_name)

tracer, meter = setup_otel("rag-api", "2.3.0")

TypeScript SDK:

// otel-setup.ts
import { NodeSDK } from "@opentelemetry/sdk-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-grpc";
import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-grpc";
import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
import { Resource } from "@opentelemetry/resources";
import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions";
import { getNodeAutoInstrumentations } from "@opentelemetry/auto-instrumentations-node";

const sdk = new NodeSDK({
  resource: new Resource({
    [SemanticResourceAttributes.SERVICE_NAME]: "rag-frontend-api",
    [SemanticResourceAttributes.SERVICE_VERSION]: "1.0.0",
    "deployment.environment": process.env.NODE_ENV || "production",
  }),
  traceExporter: new OTLPTraceExporter({
    url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT || "http://otel-collector:4317",
  }),
  metricReader: new PeriodicExportingMetricReader({
    exporter: new OTLPMetricExporter({
      url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT || "http://otel-collector:4317",
    }),
    exportIntervalMillis: 10000,
  }),
  instrumentations: [getNodeAutoInstrumentations()],
});

sdk.start();

OpenTelemetry Collector Config:

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318
  prometheus:
    config:
      scrape_configs:
        - job_name: "otel-collector"
          scrape_interval: 10s
          static_configs:
            - targets: ["localhost:8888"]

processors:
  batch:
    timeout: 5s
    send_batch_max_size: 1000
  memory_limiter:
    limit_mib: 512
    spike_limit_mib: 128
    check_interval: 5s
  probabilistic_sampler:
    sampling_percentage: 25  # Sample 25% of spans in production

exporters:
  otlp/jaeger:
    endpoint: "jaeger:4317"
    tls:
      insecure: true
  prometheus:
    endpoint: "0.0.0.0:8889"
  logging:
    loglevel: debug

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch, probabilistic_sampler]
      exporters: [otlp/jaeger]
    metrics:
      receivers: [otlp, prometheus]
      processors: [memory_limiter, batch]
      exporters: [prometheus]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [logging]

2. Prometheus Metrics & Alerting

Custom Application Metrics:

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

# Request metrics
REQUEST_COUNT = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status_code"],
)
REQUEST_DURATION = Histogram(
    "http_request_duration_seconds",
    "HTTP request duration",
    ["method", "endpoint"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0],
)

# RAG-specific metrics
EMBEDDING_LATENCY = Histogram(
    "rag_embedding_latency_ms",
    "Embedding generation latency in milliseconds",
    ["model"],
    buckets=[10, 50, 100, 200, 500, 1000, 2000],
)
RETRIEVAL_HIT_RATE = Gauge(
    "rag_retrieval_hit_rate",
    "Fraction of queries where top-N contains relevant document",
    ["top_k"],
)
CACHE_HIT_RATE = Gauge("rag_cache_hit_rate", "Semantic cache hit rate")
HALLUCINATION_RATE = Gauge("rag_hallucination_rate", "Fraction of responses flagged as hallucinated")
ACTIVE_CONVERSATIONS = Gauge("rag_active_conversations", "Number of active chat sessions")
TOKEN_USAGE = Counter("rag_token_usage_total", "Total tokens consumed", ["model", "type"])  # type: prompt/completion

# Middleware for automatic request tracking
from starlette.middleware.base import BaseHTTPMiddleware

class MetricsMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start

        REQUEST_COUNT.labels(
            method=request.method,
            endpoint=request.url.path,
            status_code=response.status_code,
        ).inc()
        REQUEST_DURATION.labels(
            method=request.method, endpoint=request.url.path
        ).observe(duration)

        return response

Prometheus Alert Rules:

# prometheus-alerts.yml
groups:
  - name: rag-api-alerts
    rules:
      # SLO: 99% of requests succeed
      - alert: HighErrorRate
        expr: |
          sum(rate(http_requests_total{status_code=~"5.."}[5m]))
          / sum(rate(http_requests_total[5m])) > 0.01
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "RAG API error rate > 1% for 5 minutes"
          runbook_url: "https://wiki.internal/runbooks/high-error-rate"

      # Latency SLO: P95  3
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "P95 latency > 3s for 10 minutes"

      # Embedding latency regression
      - alert: EmbeddingLatencyDegradation
        expr: |
          histogram_quantile(0.95, sum(rate(rag_embedding_latency_ms_bucket[5m])) by (le)) > 1000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "P95 embedding latency > 1s"

      # Cache hit rate dropped
      - alert: LowCacheHitRate
        expr: rag_cache_hit_rate  14.4 * 0.001  # 14.4x burn rate for 99.9% SLO
        for: 3m
        labels:
          severity: critical
          page: "true"
        annotations:
          summary: "Budget burn rate too high — 1h window"

      - alert: HighBurnRate6h
        expr: |
          sum(rate(http_requests_total{status_code=~"5.."}[6h]))
          / sum(rate(http_requests_total[6h])) > 6 * 0.001
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "Budget burn rate elevated — 6h window"

3. Grafana Dashboards

Dashboard-as-Code (Jsonnet):

// dashboards/rag-api.jsonnet
local grafana = import 'github.com/grafana/grafonnet-lib/grafonnet/grafana.libsonnet';
local template = grafana.template;
local panel = grafana.graphPanel;

{
  dashboard: grafana.dashboard.new(
    'RAG API Dashboard',
    tags=['rag', 'production'],
    refresh='30s',
    time={from: 'now-6h', to: 'now'},
  )
  .addTemplate(template.new('datasource', 'Datasource', 'prometheus').hide())
  .addTemplate(template.new('service', 'Service', 'label_values(http_requests_total, service)').hide())
  .addRow(
    grafana.row.new('Overview')
    .addPanel(
      panel.new('Request Rate', datasource='$datasource')
      .addTarget(grafana.target.new('sum(rate(http_requests_total{service="$service"}[5m]))'))
      .setYAxisLabel('req/s'),
      {gridPos: {x: 0, y: 0, w: 8, h: 8}}
    )
    .addPanel(
      panel.new('Error Rate %', datasource='$datasource')
      .addTarget(grafana.target.new(
        'sum(rate(http_requests_total{service="$service", status_code=~"5.."}[5m])) / sum(rate(http_requests_total{service="$service"}[5m])) * 100'
      ))
      .setYAxisLabel('%'),
      {gridPos: {x: 8, y: 0, w: 8, h: 8}}
    )
    .addPanel(
      panel.new('P95 Latency', datasource='$datasource')
      .addTarget(grafana.target.new(
        'histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket{service="$service"}[5m])) by (le))'
      ))
      .setYAxisLabel('seconds'),
      {gridPos: {x: 16, y: 0, w: 8, h: 8}}
    )
  )
  .addRow(
    grafana.row.new('RAG Metrics')
    .addPanel(
      panel.new('Embedding Latency P95', datasource='$datasource')
      .addTarget(grafana.target.new(
        'histogram_quantile(0.95, sum(rate(rag_embedding_latency_ms_bucket[5m])) by (le))'
      )),
      {gridPos: {x: 0, y: 8, w: 8, h: 8}}
    )
    .addPanel(
      panel.new('Retrieval Hit Rate', datasource='$datasource')
      .addTarget(grafana.target.new('rag_retrieval_hit_rate{top_k="5"}')),
      {gridPos: {x: 8, y: 8, w: 8, h: 8}}
    )
    .addPanel(
      panel.new('Cache Hit Rate', datasource='$datasource')
      .addTarget(grafana.target.new('rag_cache_hit_rate')),
      {gridPos: {x: 16, y: 8, w: 8, h: 8}}
    )
  )
}

Drill-Down Pattern — top-level dashboard → service dashboard → trace view:

┌─────────────────────────────────────────────────┐
│           Overview Dashboard                    │
│  [Req/s] [Error%] [P95 Latency] [SLO Budget]   │
└────────────────────┬────────────────────────────┘
                     │ Click service
┌────────────────────▼────────────────────────────┐
│         Service Dashboard (RAG API)             │
│  [Endpoints] [Embeddings] [Retrieval] [Cache]  │
└────────────────────┬────────────────────────────┘
                     │ Click spike
┌────────────────────▼────────────────────────────┐
│           Trace View (Jaeger/Tempo)             │
│  Span timeline: → retriever → embedder → LLM   │
└─────────────────────────────────────────────────┘

4. Structured Logging

Python with Loguru:

from loguru import logger
import sys
import json

# Configure structured JSON logging for production
def serialize_record(record):
    """Format log record as JSON for ingestion."""
    return json.dumps({
        "timestamp": record["time"].isoformat(),
        "level": record["level"].name,
        "message": record["message"],
        "service": record["extra"].get("service", "unknown"),
        "trace_id": record["extra"].get("trace_id"),
        "span_id": record["extra"].get("span_id"),
        "user_id": record["extra"].get("user_id"),
        "request_id": record["extra"].get("request_id"),
        "file": record["file"].name,
        "line": record["line"],
    }) + "\n"

# Remove default handler
logger.remove()

# Add structured JSON handler for production
if os.environ.get("ENVIRONMENT") == "production":
    logger.add(sys.stdout, format=serialize_record, level="INFO", serialize=True)
else:
    logger.add(sys.stdout, level="DEBUG", colorize=True)

# Add rotation to control log volume/cost
logger.add(
    "logs/app-{time:YYYY-MM-DD}.log",
    rotation="100 MB",
    retention="7 days",
    compression="gzip",
    level="INFO",
)

# Usage with correlation ID
def process_request(request_id: str, trace_id: str, user_id: str):
    ctx_logger = logger.bind(
        request_id=request_id,
        trace_id=trace_id,
        user_id=user_id,
        service="rag-api",
    )
    ctx_logger.info("Processing RAG request", query_length=len(request_id))

TypeScript with Winston:

import winston from "winston";

const logger = winston.createLogger({
  level: process.env.LOG_LEVEL || "info",
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.errors({ stack: true }),
    winston.format.json(),
  ),
  defaultMeta: {
    service: "rag-frontend-api",
    environment: process.env.NODE_ENV,
  },
  transports: [
    new winston.transports.Console(),
    new winston.transports.File({
      filename: "logs/error.log",
      level: "error",
      maxsize: 100 * 1024 * 1024, // 100MB
      maxFiles: 7,
    }),
  ],
});

// Middleware: inject correlation ID
import { v4 as uuidv4 } from "uuid";
import { AsyncLocalStorage } from "async_hooks";

const als = new AsyncLocalStorage>();

export function correlationMiddleware(req, res, next) {
  const correlationId = req.headers["x-correlation-id"] || uuidv4();
  const store = new Map([
    ["correlationId", correlationId],
    ["traceId", req.headers["x-b3-traceid"] || ""],
  ]);
  res.setHeader("x-correlation-id", correlationId);
  als.run(store, next);
}

// Usage
function handleRequest(query: string) {
  const store = als.getStore();
  const correlationId = store?.get("correlationId");
  logger.info("Processing query", { correlationId, queryLength: query.length });
}

5. Distributed Tracing

Span Naming Conventions:

from opentelemetry import trace
from opentelemetry.trace import SpanKind, Status, StatusCode

tracer = trace.get_tracer("rag-api")

def process_rag_query(query: str, top_k: int = 5):
    """Process a full RAG pipeline with tracing."""
    with tracer.start_as_current_span(
        "rag.process_query",
        kind=SpanKind.SERVER,
        attributes={
            "rag.query_length": len(query),
            "rag.top_k": top_k,
        },
    ) as root_span:
        # Retrieve context
        contexts = retrieve_context(query, top_k)
        root_span.set_attribute("rag.context_count", len(contexts))

        # Generate response
        response = generate_response(query, contexts)
        root_span.set_attribute("rag.response_length", len(response))

        root_span.set_status(Status(StatusCode.OK))
        return response

def retrieve_context(query: str, top_k: int) -> list[str]:
    with tracer.start_as_current_span(
        "rag.retrieve",
        kind=SpanKind.INTERNAL,
        attributes={"rag.top_k": top_k},
    ) as span:
        embeddings = generate_embeddings([query])
        span.set_attribute("rag.embedd

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [noah-sheldon](https://github.com/noah-sheldon)
- **Source:** [noah-sheldon/ai-dev-kit](https://github.com/noah-sheldon/ai-dev-kit)
- **License:** MIT
- **Homepage:** https://noahsheldon.dev

Install and usage instructions live in the source repository linked above.