Lesson 16.39: AI Observability, Tracing & Reliability Monitoring

3.1 Logs — “What exactly happened?”

Logs are discrete timestamped event records. They capture exact stack traces, provider responses, and error conditions for a distinct request:

{
  "timestamp": "2026-10-16T10:14:22.104Z",
  "level": "ERROR",
  "event": "ai_provider_timeout",
  "request_id": "req-98b7c4-fa21",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7",
  "route": "quality",
  "provider": "anthropic",
  "model_alias": "claude-3-5-sonnet",
  "retry_count": 1,
  "timeout_seconds": 15.0
}

Core Benefit: Complete contextual depth of exceptions. In production, correlating request_id, trace_id, and span_id enables 1-click jumps between logs and traces.

3.2 Metrics — “How frequently does it occur and what is the trend?”

Metrics are numerical data points aggregated across time-series windows. They do not record granular request text, but reveal aggregate macro health:

• • Request Rate: Inbound AI requests per second/minute (RPS / RPM).
• • Failure Rate: Percentage ratio of 429/500/503 errors over sliding 5m windows.
• • Latency Histogram: Distribution across p50, p95, and p99 percentile buckets.
• • Cost Counters: Cumulative input tokens, output tokens, and cache hit ratios.

Core Benefit: Extremely low storage overhead, real-time live dashboard rendering, and reliable mathematical alerting rules.

3.3 Traces — “Where was latency spent or where did it break?”

A trace captures the chronological execution lifecycle of a single request across distributed services, represented as a tree of timed Spans:

Trace: req-98b7c4-fa21 [Total: 3,420ms] [Status: OK]
│
├── [Span 1] HTTP POST /v1/chat [Duration: 3,420ms]
│   │
│   ├── [Span 2] auth.verify_jwt [Duration: 18ms]
│   ├── [Span 3] cache.lookup [Duration: 8ms, outcome="miss"]
│   │
│   ├── [Span 4] rag.retrieve_context [Duration: 412ms]
│   │   ├── [Span 5] vector.qdrant_search [Duration: 110ms, results=20]
│   │   ├── [Span 6] keyword.bm25_search [Duration: 42ms, results=20]
│   │   └── [Span 7] rerank.cohere_reranker [Duration: 260ms, top_k=5]
│   │
│   ├── [Span 8] ai.gateway_model_call [Duration: 2,960ms]
│   │   ├── [Span 9] provider.anthropic_attempt_1 [Duration: 1,500ms, outcome="timeout"]
│   │   └── [Span 10] provider.openai_fallback_attempt_2 [Duration: 1,460ms, outcome="success"]
│   │
│   └── [Span 11] guardrail.output_validation [Duration: 22ms, status="clean"]

Core Benefit: Spans 9 and 10 immediately prove that the primary provider timed out after 1,500ms, prompting the gateway to seamlessly execute a fallback. The root cause is identified with zero guesswork.

Step 1: Install Production Packages

pip install opentelemetry-api opentelemetry-sdk opentelemetry-instrumentation-fastapi opentelemetry-exporter-otlp prometheus-client python-json-logger

Step 2: Configure Tracer Provider (app/telemetry.py)

In development we inspect spans via console; in production OTel flushes binary batches over OTLP gRPC/HTTP directly to Jaeger, Datadog, or Grafana Tempo:

from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter

# 1. Define global service and deployment environment resource
resource = Resource.create({
    "service.name": "byteion-ai-gateway",
    "deployment.environment": "production",
    "service.version": "1.4.0",
})

# 2. Instantiate tracer provider
provider = TracerProvider(resource=resource)

# 3. Attach batch processor (Console in dev; OTLP gRPC exporter in production)
provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))

# 4. Register global tracer
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("byteion.ai.observability")

Step 3: FastAPI Application Instrumentation (app/main.py)

Generate a unique X-Request-ID via middleware, bind it to response headers, and capture execution spans:

import uuid
import logging
from time import perf_counter
from fastapi import FastAPI, HTTPException, Request
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from app.telemetry import tracer

app = FastAPI(title="Byteion AI Engine")
logger = logging.getLogger(__name__)

# Request ID correlation middleware
@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
    req_id = str(uuid.uuid4())
    request.state.request_id = req_id
    response = await call_next(request)
    response.headers["X-Request-ID"] = req_id
    return response

# Automatic FastAPI instrumentation (excluding probes and scrapers)
FastAPIInstrumentor.instrument_app(app, excluded_urls="/health,/metrics")

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

@app.post("/v1/knowledge/chat")
async def handle_knowledge_chat(request: Request):
    req_id = request.state.request_id
    start_time = perf_counter()

    # 1. Root business span initiation
    with tracer.start_as_current_span("ai.knowledge_workflow") as span:
        span.set_attribute("app.request_id", req_id)
        span.set_attribute("ai.task_type", "enterprise_kb")

        try:
            # 2. Retrieval child span
            with tracer.start_as_current_span("rag.retrieve_chunks") as r_span:
                r_span.set_attribute("rag.search_type", "hybrid_dense_sparse")
                # Invoke production vector store query
                chunks = await simulate_retrieval(req_id)
                r_span.set_attribute("rag.chunks_found", len(chunks))

            # 3. Downstream inference child span
            with tracer.start_as_current_span("ai.model_execution") as m_span:
                m_span.set_attribute("ai.model_alias", "claude-3-5-sonnet")
                m_span.set_attribute("ai.temperature", 0.2)
                ai_reply = await simulate_model_call(chunks)
                m_span.set_attribute("ai.finish_reason", "stop")

            span.set_attribute("ai.status", "success")
            return {
                "reply": ai_reply,
                "request_id": req_id,
                "latency_ms": round((perf_counter() - start_time) * 1000, 2)
            }

        except Exception as exc:
            span.record_exception(exc)
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(exc)))
            logger.error("AI workflow failed", extra={"request_id": req_id, "error": str(exc)})
            raise HTTPException(status_code=503, detail="AI Service Temporarily Degraded")

async def simulate_retrieval(req_id: str):
    return ["chunk_doc_102", "chunk_doc_501"]

async def simulate_model_call(chunks):
    return "The Byteion annual enterprise subscription quota is 10,000,000 tokens."

Secure Tool Invocation Tracing Pattern

async def execute_tool_with_tracing(tool_name: str, arguments: dict, caller_principal):
    with tracer.start_as_current_span(f"tool.{tool_name}") as span:
        span.set_attribute("tool.name", tool_name)
        span.set_attribute("tenant.id", caller_principal.tenant_id)
        
        # Privacy hygiene: Record metadata schema keys, never raw customer PII arguments
        safe_keys = list(arguments.keys())
        span.set_attribute("tool.arg_keys", safe_keys)
        
        try:
            start_t = perf_counter()
            result = await dispatch_tool(tool_name, arguments, caller_principal)
            span.set_attribute("tool.duration_ms", round((perf_counter() - start_t) * 1000, 2))
            span.set_attribute("tool.status", "success")
            return result
        except PermissionError as pe:
            span.set_attribute("tool.status", "permission_denied")
            span.record_exception(pe)
            raise
        except Exception as ex:
            span.set_attribute("tool.status", "error")
            span.record_exception(ex)
            raise

JSON Formatter Configuration (app/logging_config.py)

import logging
import sys
from pythonjsonlogger.json import JsonFormatter

def setup_production_logging():
    handler = logging.StreamHandler(sys.stdout)
    formatter = JsonFormatter("%(asctime)s %(levelname)s %(name)s %(message)s")
    handler.setFormatter(formatter)

    root_logger = logging.getLogger()
    root_logger.handlers.clear()
    root_logger.addHandler(handler)
    root_logger.setLevel(logging.INFO)

# Execute upon application bootstrap
setup_production_logging()

Emitting Structured Events from AI Endpoints

logger.info(
    "ai_request_completed",
    extra={
        "request_id": req_id,
        "route": "fast",
        "model": "gpt-4o-mini",
        "duration_ms": 780,
        "input_tokens": 420,
        "output_tokens": 85,
        "cache_hit": False,
        "fallback_triggered": False,
    }
)

Production Prometheus Metrics Definition (app/metrics.py)

from prometheus_client import Counter, Histogram

# 1. Primary Request Counter (partitioned by task, route, and outcome)
AI_REQUESTS = Counter(
    "byteion_ai_requests_total",
    "Total AI requests processed by the platform",
    ["task_type", "route", "outcome"]  # outcome: success | error | fallback
)

# 2. Latency Histogram (configured with realistic AI latency buckets)
AI_REQUEST_DURATION = Histogram(
    "byteion_ai_request_duration_seconds",
    "End-to-end AI workflow response duration in seconds",
    ["task_type", "route"],
    buckets=(0.1, 0.25, 0.5, 1.0, 2.0, 4.0, 8.0, 15.0, 30.0)
)

# 3. Token Consumption Counters
AI_INPUT_TOKENS = Counter(
    "byteion_ai_input_tokens_total",
    "Total input tokens consumed",
    ["model_alias"]
)

AI_OUTPUT_TOKENS = Counter(
    "byteion_ai_output_tokens_total",
    "Total output tokens generated",
    ["model_alias"]
)

# 4. Cache & Fallback Counters
AI_CACHE_RESULTS = Counter(
    "byteion_ai_cache_results_total",
    "Response cache lookup outcomes",
    ["result"]  # hit | miss
)

AI_FALLBACKS = Counter(
    "byteion_ai_fallbacks_total",
    "Gateway model fallback occurrences",
    ["route"]
)

Robust Observation Wrapper Pattern

from time import perf_counter
from app.metrics import AI_REQUESTS, AI_REQUEST_DURATION

async def track_ai_execution(task_type: str, route: str, execute_fn):
    start = perf_counter()
    outcome = "error"
    try:
        result = await execute_fn()
        outcome = "success"
        return result
    finally:
        duration = perf_counter() - start
        AI_REQUESTS.labels(task_type=task_type, route=route, outcome=outcome).inc()
        AI_REQUEST_DURATION.labels(task_type=task_type, route=route).observe(duration)

11.1 Exposing a Protected /metrics Endpoint

from fastapi import Response
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST

@app.get("/metrics", include_in_schema=False)
async def get_prometheus_metrics():
    # Security policy: Bind scrape route strictly to private internal VPC subnets
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)