Monitoring and Observability¶

# Backend metrics implementation
from prometheus_client import Counter, Histogram, Gauge
import time

# Request metrics
request_count = Counter('rag_requests_total', 'Total RAG requests', ['method', 'endpoint'])
request_duration = Histogram('rag_request_duration_seconds', 'Request duration', ['method', 'endpoint'])

# RAG-specific metrics
rag_queries_total = Counter('rag_queries_total', 'Total RAG queries', ['collection', 'status'])
rag_query_duration = Histogram('rag_query_duration_seconds', 'RAG query duration', ['collection'])
vector_search_duration = Histogram('vector_search_duration_seconds', 'Vector search duration', ['collection'])
embedding_duration = Histogram('embedding_duration_seconds', 'Embedding generation duration')

# Resource metrics
active_connections = Gauge('active_connections', 'Active database connections')
cache_hit_rate = Gauge('cache_hit_rate', 'Cache hit rate')
memory_usage = Gauge('memory_usage_bytes', 'Memory usage in bytes')

# Error metrics
error_count = Counter('errors_total', 'Total errors', ['error_type', 'endpoint'])

# Health check implementation
@app.get("/health")
async def health_check():
    """Health check endpoint for monitoring"""
    try:
        # Check database connectivity
        db_status = await check_database_connection()

        # Check vector database connectivity
        vector_status = await check_vector_database_connection()

        # Check object storage connectivity
        storage_status = await check_object_storage_connection()

        # Overall health status
        overall_status = "healthy" if all([db_status, vector_status, storage_status]) else "unhealthy"

        return {
            "status": overall_status,
            "timestamp": datetime.utcnow().isoformat(),
            "checks": {
                "database": db_status,
                "vector_database": vector_status,
                "object_storage": storage_status
            }
        }
    except Exception as e:
        return {
            "status": "unhealthy",
            "timestamp": datetime.utcnow().isoformat(),
            "error": str(e)
        }

# Structured logging configuration
import logging
import json
from datetime import datetime

class StructuredLogger:
    def __init__(self, name):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)

        # Create formatter
        formatter = logging.Formatter('%(message)s')

        # Create handler
        handler = logging.StreamHandler()
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)

    def log(self, level, message, **kwargs):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": level.upper(),
            "message": message,
            "service": "rag-modulo-backend",
            **kwargs
        }
        self.logger.info(json.dumps(log_entry))

# Usage
logger = StructuredLogger(__name__)

# Log request
logger.log("info", "Request received",
          method="GET",
          path="/api/search",
          user_id="12345",
          request_id="req-123")

# Log error
logger.log("error", "Database connection failed",
          error="Connection timeout",
          database="postgresql",
          retry_count=3)

# Access log middleware
@app.middleware("http")
async def access_log_middleware(request: Request, call_next):
    start_time = time.time()

    # Process request
    response = await call_next(request)

    # Calculate duration
    duration = time.time() - start_time

    # Log access
    logger.log("info", "Request completed",
              method=request.method,
              path=request.url.path,
              status_code=response.status_code,
              duration=duration,
              user_agent=request.headers.get("user-agent"),
              ip_address=request.client.host)

    return response

# Error analysis queries
error_analysis_queries = {
    "error_rate_by_endpoint": """
        SELECT
            endpoint,
            COUNT(*) as error_count,
            COUNT(*) * 100.0 / SUM(COUNT(*)) OVER() as error_percentage
        FROM logs
        WHERE level = 'ERROR'
        AND timestamp >= NOW() - INTERVAL '1 hour'
        GROUP BY endpoint
        ORDER BY error_count DESC
    """,

    "error_trends": """
        SELECT
            DATE_TRUNC('hour', timestamp) as hour,
            COUNT(*) as error_count
        FROM logs
        WHERE level = 'ERROR'
        AND timestamp >= NOW() - INTERVAL '24 hours'
        GROUP BY hour
        ORDER BY hour
    """,

    "top_errors": """
        SELECT
            message,
            COUNT(*) as count,
            MAX(timestamp) as last_occurrence
        FROM logs
        WHERE level = 'ERROR'
        AND timestamp >= NOW() - INTERVAL '1 hour'
        GROUP BY message
        ORDER BY count DESC
        LIMIT 10
    """
}

# Performance analysis queries
performance_analysis_queries = {
    "slow_queries": """
        SELECT
            endpoint,
            AVG(duration) as avg_duration,
            MAX(duration) as max_duration,
            COUNT(*) as request_count
        FROM logs
        WHERE duration > 1.0
        AND timestamp >= NOW() - INTERVAL '1 hour'
        GROUP BY endpoint
        ORDER BY avg_duration DESC
    """,

    "response_time_trends": """
        SELECT
            DATE_TRUNC('minute', timestamp) as minute,
            AVG(duration) as avg_duration,
            PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY duration) as p95_duration
        FROM logs
        WHERE timestamp >= NOW() - INTERVAL '1 hour'
        GROUP BY minute
        ORDER BY minute
    """
}