Monitoring & Observability Guide¶

# Health check implementation
import http.client
import sys

def check_health() -> None:
    try:
        conn = http.client.HTTPConnection("localhost", 8000)
        conn.request("GET", "/api/health")
        response = conn.getresponse()
        if response.status == 200:
            sys.exit(0)  # Healthy
        else:
            sys.exit(1)  # Unhealthy
    except Exception:
        sys.exit(1)  # Failed to connect

from core.enhanced_logging import get_logger
from core.logging_context import log_operation, pipeline_stage_context, PipelineStage

logger = get_logger(__name__)

# Simple logging
logger.info("Starting search operation")
logger.error("Database connection failed", exc_info=True)

# With operation context
with log_operation(logger, "search_documents", "collection", collection_id, user_id=user_id):
    # All logs within this context include collection_id and user_id
    logger.info("Query rewriting started")

    # Track pipeline stages
    with pipeline_stage_context(PipelineStage.QUERY_REWRITING):
        logger.info("Original query", extra={"query": original_query})
        # ... query rewriting logic ...
        logger.info("Rewritten query", extra={"rewritten": new_query})

    with pipeline_stage_context(PipelineStage.RETRIEVAL):
        logger.info("Retrieving documents", extra={"top_k": 5})
        # ... retrieval logic ...
        logger.info("Retrieved documents", extra={"count": len(results)})

# backend/main.py
from pathlib import Path
from core.logging_utils import setup_logging

# Determine log directory (container vs local)
log_dir = Path("/app/logs") if os.getenv("CONTAINER_ENV") else Path(__file__).parent.parent / "logs"

# Initialize logging
setup_logging(log_dir)
logger = get_logger(__name__)

class PipelineStage:
    QUERY_REWRITING = "query_rewriting"
    RETRIEVAL = "retrieval"
    RERANKING = "reranking"
    PROMPT_BUILDING = "prompt_building"
    GENERATION = "generation"
    RESPONSE_GENERATION = "response_generation"
    CHAIN_OF_THOUGHT = "chain_of_thought"
    SOURCE_ATTRIBUTION = "source_attribution"

# In SearchService
async def search(self, search_input: SearchInput):
    with log_operation(self.logger, "search", "collection", search_input.collection_id):
        # Stage 1: Query rewriting
        with pipeline_stage_context(PipelineStage.QUERY_REWRITING):
            rewritten = await self._rewrite_query(search_input.question)

        # Stage 2: Retrieval
        with pipeline_stage_context(PipelineStage.RETRIEVAL):
            docs = await self._retrieve_documents(rewritten)

        # Stage 3: Generation
        with pipeline_stage_context(PipelineStage.GENERATION):
            response = await self._generate_answer(docs, rewritten)

        return response

from core.log_storage_service import LogStorageService, LogLevel

# Get log storage instance
log_storage = LogStorageService.get_instance()

# Query logs
logs = log_storage.query_logs(
    level=LogLevel.ERROR,
    entity_type="collection",
    entity_id="550e8400-e29b-41d4-a716-446655440000",
    limit=100
)

# Get recent logs
recent = log_storage.get_recent_logs(limit=50)

# Filter by time range
from datetime import datetime, timedelta
since = datetime.now() - timedelta(hours=1)
recent_hour = log_storage.query_logs(since=since)

# Future: backend/core/metrics.py
from prometheus_client import Counter, Histogram, Gauge

# Request metrics
http_requests_total = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

http_request_duration_seconds = Histogram(
    'http_request_duration_seconds',
    'HTTP request latency',
    ['method', 'endpoint']
)

# RAG-specific metrics
rag_searches_total = Counter(
    'rag_searches_total',
    'Total RAG search requests',
    ['collection', 'status']
)

rag_search_duration_seconds = Histogram(
    'rag_search_duration_seconds',
    'RAG search latency',
    ['collection', 'pipeline_stage']
)

vector_db_operations = Counter(
    'vector_db_operations_total',
    'Vector database operations',
    ['operation', 'status']
)

llm_api_calls = Counter(
    'llm_api_calls_total',
    'LLM API calls',
    ['provider', 'model', 'status']
)

active_connections = Gauge(
    'active_connections',
    'Active WebSocket connections'
)

# Middleware adds correlation_id to all requests
# backend/core/authentication_middleware.py

import uuid
from fastapi import Request

async def add_correlation_id(request: Request):
    correlation_id = request.headers.get('X-Correlation-ID', str(uuid.uuid4()))
    request.state.correlation_id = correlation_id
    logger.info("Request received", extra={"correlation_id": correlation_id})

# In any request handler
from fastapi import Request

@router.post("/search")
async def search(request: Request, search_input: SearchInput):
    correlation_id = getattr(request.state, 'correlation_id', 'unknown')
    logger.info("Search started", extra={"correlation_id": correlation_id})
    # ... process search ...
    return results

# Future: backend/core/tracing.py
from opentelemetry import trace
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider

tracer_provider = TracerProvider()
trace.set_tracer_provider(tracer_provider)

jaeger_exporter = JaegerExporter(
    agent_host_name="jaeger",
    agent_port=6831,
)

tracer = trace.get_tracer(__name__)

# Usage
with tracer.start_as_current_span("search_operation"):
    # ... search logic ...
    with tracer.start_as_current_span("query_rewrite"):
        # ... query rewriting ...
        pass

# Performance tracking in logs
import time

start_time = time.time()
# ... operation ...
elapsed_ms = (time.time() - start_time) * 1000

logger.info(
    "Operation completed",
    extra={
        "operation": "search",
        "execution_time_ms": elapsed_ms,
        "collection_id": collection_id
    }
)