orchestrata_ultimate.core.logging

Reference › orchestrata_ultimate › core › logging

Centralised structured logging with correlation identifiers, in core. Three context variables, correlation_id, session_id and volk_id, carry request identity across CXP messages and volk interactions. StructuredFormatter renders a LogRecord as one JSON object, including the context variables and a fixed list of optional extra fields such as duration_ms, model, gate, cycle and escalated; StandardFormatter is the plain-text alternative. configure_logging(level, log_file, json_format) sets up the application's handlers (JSON by default, with an optional file), new_correlation_id generates and sets a fresh identifier, and get_logger(name) returns a named logger. These three functions carry @traces annotations.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

StructuredFormatter

Bases: logging.Formatter · Line 56

JSON structured log formatter with correlation IDs.

StandardFormatter

Bases: logging.Formatter · Line 93

Human-readable formatter for development use.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/logging.py

"""Centralized structured logging with correlation IDs.

Provides JSON-formatted log output with context variables for
request tracing across CXP messages and volk interactions.
"""

import json
import logging
import sys
import uuid
from contextvars import ContextVar
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

# Context variables for request tracing across CXP messages
correlation_id: ContextVar[str] = ContextVar("correlation_id", default="")
session_id: ContextVar[str] = ContextVar("session_id", default="")
volk_id: ContextVar[str] = ContextVar("volk_id", default="")

# Extra fields that StructuredFormatter extracts from LogRecords
_EXTRA_FIELDS = (
    "duration_ms",
    "method",
    "params",
    "result",
    "error",
    "prompt_length",
    "response_length",
    "history_length",
    "cycle",
    "gate",
    "gate_result",
    "round_num",
    "questions",
    "answers",
    "decided_complete",
    "issues_before",
    "issues_after",
    "model",
    "escalated",
    "volk",
    "temperature",
    "event_type",
    "client_ip",
    "auth_result",
    "scope",
    "topic",
    "subscriber_count",
    "sequence",
    "subscription_id",
)


class StructuredFormatter(logging.Formatter):
    """JSON structured log formatter with correlation IDs.

    Emits one JSON object per log line containing timestamp,
    level, logger name, message, correlation context, and any
    extra fields attached to the LogRecord.
    """

    @traces(
        spec="001", story="US-008", reqs=["REQ-001-US-008-008", "REQ-001-US-008-009"]
    )
    def format(self, record: logging.LogRecord) -> str:
        """Format a LogRecord as a JSON string.

        Args:
            record: The log record to format.

        Returns:
            A single-line JSON string.
        """
        log_entry: dict[str, Any] = {
            "timestamp": datetime.now(UTC).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "correlation_id": correlation_id.get(""),
            "session_id": session_id.get(""),
            "volk_id": volk_id.get(""),
        }
        if record.exc_info and record.exc_info[1] is not None:
            log_entry["exception"] = self.formatException(record.exc_info)
        for key in _EXTRA_FIELDS:
            if hasattr(record, key):
                log_entry[key] = getattr(record, key)
        return json.dumps(log_entry, default=str)


class StandardFormatter(logging.Formatter):
    """Human-readable formatter for development use.

    Includes timestamp, level, logger name, and correlation ID
    in a fixed-width columnar layout.
    """

    FORMAT = (
        "%(asctime)s [%(levelname)-5s] %(name)s (%(filename)s:%(lineno)d) %(message)s"
    )

    def __init__(self) -> None:
        """Initialize with the standard format string."""
        super().__init__(fmt=self.FORMAT)


@traces(spec="001", story="US-008", reqs=["REQ-001-US-008-012", "REQ-001-US-008-020"])
def configure_logging(
    level: str = "INFO",
    log_file: str = "",
    json_format: bool = True,
) -> None:
    """Configure structured logging for the entire application.

    Sets up the root logger with either JSON or human-readable
    formatting. Optionally writes to a file in addition to stderr.

    Args:
        level: Log level name (DEBUG, INFO, WARNING, ERROR).
        log_file: Path to a log file. Empty string means no file.
        json_format: True for JSON output, False for human-readable.
    """
    root = logging.getLogger()
    root.setLevel(getattr(logging, level.upper(), logging.INFO))

    # Remove (and close) existing handlers to avoid duplicates on re-configure
    # and to release per-run file handles before the log directory is cleaned up.
    for handler in root.handlers[:]:
        try:
            handler.close()
        except Exception:
            pass
        root.removeHandler(handler)

    formatter: logging.Formatter
    if json_format:
        formatter = StructuredFormatter()
    else:
        formatter = StandardFormatter()

    console = logging.StreamHandler(sys.stderr)
    console.setFormatter(formatter)
    root.addHandler(console)

    if log_file:
        file_handler = logging.FileHandler(log_file)
        file_handler.setFormatter(formatter)
        root.addHandler(file_handler)


@traces(spec="001", story="US-008", reqs=["REQ-001-US-008-010"])
def new_correlation_id() -> str:
    """Generate and set a new correlation ID.

    Returns:
        The 12-character correlation ID that was set.
    """
    cid = str(uuid.uuid4())[:12]
    correlation_id.set(cid)
    return cid


@traces(spec="105", story="US-004", reqs=["REQ-105-007"])
def get_logger(name: str) -> logging.Logger:
    """Get a logger with the given name.

    The logger inherits the root configuration set by
    configure_logging(). If configure_logging() has not been
    called, Python's default logging applies.

    Args:
        name: Logger name, typically __name__.

    Returns:
        A configured logging.Logger instance.
    """
    return logging.getLogger(name)