""" Structured logging configuration for Orchard. This module provides: - Structured JSON logging for production environments - Request tracing via X-Request-ID header - Verification failure logging with context - Configurable log levels via environment Usage: from app.logging_config import setup_logging, get_request_id setup_logging() # Call once at app startup request_id = get_request_id() # Get current request's ID """ import logging import json import sys import uuid from datetime import datetime, timezone from typing import Optional, Any, Dict from contextvars import ContextVar from .config import get_settings # Context variable for request ID (thread-safe) _request_id_var: ContextVar[Optional[str]] = ContextVar("request_id", default=None) def get_request_id() -> Optional[str]: """Get the current request's ID from context.""" return _request_id_var.get() def set_request_id(request_id: Optional[str] = None) -> str: """ Set the request ID for the current context. If no ID provided, generates a new UUID. Returns the request ID that was set. """ if request_id is None: request_id = str(uuid.uuid4()) _request_id_var.set(request_id) return request_id def clear_request_id(): """Clear the request ID from context.""" _request_id_var.set(None) class JSONFormatter(logging.Formatter): """ JSON log formatter for structured logging. Output format: { "timestamp": "2025-01-01T00:00:00.000Z", "level": "INFO", "logger": "app.routes", "message": "Request completed", "request_id": "abc-123", "extra": {...} } """ def format(self, record: logging.LogRecord) -> str: log_entry: Dict[str, Any] = { "timestamp": datetime.now(timezone.utc).isoformat(), "level": record.levelname, "logger": record.name, "message": record.getMessage(), } # Add request ID if available request_id = get_request_id() if request_id: log_entry["request_id"] = request_id # Add exception info if present if record.exc_info: log_entry["exception"] = self.formatException(record.exc_info) # Add extra fields from record extra_fields: Dict[str, Any] = {} for key, value in record.__dict__.items(): if key not in ( "name", "msg", "args", "created", "filename", "funcName", "levelname", "levelno", "lineno", "module", "msecs", "pathname", "process", "processName", "relativeCreated", "stack_info", "exc_info", "exc_text", "thread", "threadName", "message", "asctime", ): try: json.dumps(value) # Ensure serializable extra_fields[key] = value except (TypeError, ValueError): extra_fields[key] = str(value) if extra_fields: log_entry["extra"] = extra_fields return json.dumps(log_entry) class StandardFormatter(logging.Formatter): """ Standard log formatter for development. Output format: [2025-01-01 00:00:00] INFO [app.routes] [req-abc123] Request completed """ def format(self, record: logging.LogRecord) -> str: timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") request_id = get_request_id() req_str = f" [req-{request_id[:8]}]" if request_id else "" base_msg = f"[{timestamp}] {record.levelname:5} [{record.name}]{req_str} {record.getMessage()}" if record.exc_info: base_msg += "\n" + self.formatException(record.exc_info) return base_msg def setup_logging(log_level: Optional[str] = None, json_format: Optional[bool] = None): """ Configure logging for the application. Args: log_level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL). Defaults to ORCHARD_LOG_LEVEL env var or INFO. json_format: Use JSON format. Defaults to True in production. """ settings = get_settings() # Determine log level if log_level is None: log_level = getattr(settings, "log_level", "INFO") effective_level = log_level if log_level else "INFO" level = getattr(logging, effective_level.upper(), logging.INFO) # Determine format if json_format is None: json_format = settings.is_production # Create handler handler = logging.StreamHandler(sys.stdout) handler.setLevel(level) # Set formatter if json_format: handler.setFormatter(JSONFormatter()) else: handler.setFormatter(StandardFormatter()) # Configure root logger root_logger = logging.getLogger() root_logger.setLevel(level) # Remove existing handlers root_logger.handlers.clear() root_logger.addHandler(handler) # Configure specific loggers for logger_name in ["app", "uvicorn", "uvicorn.access", "uvicorn.error"]: logger = logging.getLogger(logger_name) logger.setLevel(level) logger.handlers.clear() logger.addHandler(handler) logger.propagate = False # Quiet down noisy loggers logging.getLogger("botocore").setLevel(logging.WARNING) logging.getLogger("boto3").setLevel(logging.WARNING) logging.getLogger("urllib3").setLevel(logging.WARNING) def log_verification_failure( logger: logging.Logger, expected_hash: str, actual_hash: str, artifact_id: Optional[str] = None, s3_key: Optional[str] = None, project: Optional[str] = None, package: Optional[str] = None, size: Optional[int] = None, user_id: Optional[str] = None, source_ip: Optional[str] = None, verification_mode: Optional[str] = None, ): """ Log a verification failure with full context. This creates a structured log entry with all relevant details for debugging and alerting. """ logger.error( "Checksum verification failed", extra={ "event": "verification_failure", "expected_hash": expected_hash, "actual_hash": actual_hash, "artifact_id": artifact_id, "s3_key": s3_key, "project": project, "package": package, "size": size, "user_id": user_id, "source_ip": source_ip, "verification_mode": verification_mode, "hash_match": expected_hash == actual_hash, }, ) def log_verification_success( logger: logging.Logger, artifact_id: str, size: Optional[int] = None, verification_mode: Optional[str] = None, duration_ms: Optional[float] = None, ): """Log a successful verification.""" logger.info( f"Verification passed for artifact {artifact_id[:16]}...", extra={ "event": "verification_success", "artifact_id": artifact_id, "size": size, "verification_mode": verification_mode, "duration_ms": duration_ms, }, )