Structured Logging¶

{
  "timestamp": "2024-01-15T10:30:45.123456+00:00",
  "level": "INFO",
  "component": "backend.services.detection_service",
  "message": "Detection completed",
  "hostname": "backend-pod-abc123",
  "container_id": "a1b2c3d4e5f6",
  "app_version": "1.2.3",
  "environment": "production",
  "request_id": "req-550e8400-e29b-41d4-a716-446655440000",
  "correlation_id": "corr-abc123",
  "trace_id": "0123456789abcdef0123456789abcdef",
  "span_id": "0123456789abcdef",
  "camera_id": "front_door",
  "detection_count": 3,
  "duration_ms": 245.7
}

2024-01-15 10:30:45 | INFO     | backend.services.detection | trace_id=abc123... span_id=def456... | Detection completed

CONSOLE_FORMAT = (
    "%(asctime)s | %(levelname)-8s | %(name)s | "
    "trace_id=%(trace_id)s span_id=%(span_id)s | %(message)s"
)

# From backend/core/logging.py:262-316
@contextmanager
def log_context(**kwargs: Any) -> Generator[None]:
    """Context manager for enriched logging within a scope."""
    current = _log_context.get() or {}
    merged = {**current, **kwargs}
    token = _log_context.set(merged)
    try:
        yield
    finally:
        _log_context.reset(token)

from backend.core.logging import log_context, get_logger

logger = get_logger(__name__)

async def process_detection(camera_id: str, image_path: str):
    with log_context(camera_id=camera_id, operation="detect"):
        logger.info("Starting detection")  # Includes camera_id and operation
        result = await detect_objects(image_path)
        logger.info("Detection complete", extra={"detection_count": len(result)})

# From backend/core/logging.py:214-259
def get_current_trace_context() -> dict[str, str | None]:
    """Get the current OpenTelemetry trace context for log correlation."""
    result: dict[str, str | None] = {"trace_id": None, "span_id": None}
    try:
        span = _get_otel_current_span()
        if span is None:
            return result
        span_context = span.get_span_context()
        if not span_context or not span_context.is_valid:
            return result
        result["trace_id"] = format(span_context.trace_id, "032x")
        result["span_id"] = format(span_context.span_id, "016x")
    except (ImportError, AttributeError, TypeError):
        pass
    return result

logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
logging.getLogger("watchdog").setLevel(logging.WARNING)

# From backend/core/logging.py:318-377
def redact_url(url: str) -> str:
    """Redact sensitive information from a URL for safe logging."""
    parsed = urlparse(url)
    if not parsed.password:
        return url
    redacted_netloc = f"{parsed.username}:[REDACTED]@{parsed.hostname}"
    # ... reconstruct URL with redacted password

SENSITIVE_FIELD_NAMES = frozenset({
    "password", "secret", "key", "token", "credential",
    "api_key", "api_keys", "admin_api_key", "yolo26_api_key",
    "nemotron_api_key", "smtp_password", "database_url", "redis_url",
})

# From backend/core/logging.py:1041-1081
def sanitize_log_value(value: Any) -> str:
    """Sanitize a value for safe inclusion in log messages."""
    str_value = str(value)
    # Remove newlines (log injection), null bytes (truncation)
    sanitized = str_value.replace("\n", " ").replace("\r", " ").replace("\x00", "")
    # Remove control characters
    sanitized = "".join(char if ord(char) >= 32 or char == "\t" else " " for char in sanitized)
    return sanitized

# From backend/core/logging.py:1002-1038
def mask_ip(ip: str) -> str:
    """Mask an IP address for safe logging."""
    if ":" in ip:  # IPv6
        parts = ip.split(":")
        return f"{parts[0]}:xxxx:xxxx:xxxx"
    # IPv4
    parts = ip.split(".")
    return f"{parts[0]}.xxx.xxx.xxx"

console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(log_level)
console_formatter = TraceContextFormatter(CONSOLE_FORMAT)
console_handler.setFormatter(console_formatter)
root_logger.addHandler(console_handler)

file_handler = RotatingFileHandler(
    filename=str(log_path),
    maxBytes=settings.log_file_max_bytes,
    backupCount=settings.log_file_backup_count,
    encoding="utf-8",
)

# From backend/core/logging.py:782-800
log_entry = Log(
    timestamp=datetime.now(UTC),
    level=record.levelname,
    component=record.name,
    message=self.format(record),
    camera_id=getattr(record, "camera_id", None),
    event_id=getattr(record, "event_id", None),
    request_id=getattr(record, "request_id", None),
    # ... additional fields
)
session.add(log_entry)
session.commit()

# From backend/core/logging.py:1111-1162
def log_exception_with_context(
    logger: logging.Logger,
    exception: Exception,
    message: str,
    *,
    include_traceback: bool = True,
    level: int = logging.ERROR,
    **extra_context: Any,
) -> None:
    """Log an exception with structured context for debugging."""
    extra: dict[str, Any] = {
        "error_type": type(exception).__name__,
        "error_message_sanitized": sanitize_error(exception),
        **extra_context,
    }
    logger.log(level, message, extra=extra, exc_info=include_traceback)

log_error(
    logger,
    "Image processing timed out",
    error=e,
    extra={"camera_id": "front_door", "timeout_ms": 5000},
)