/lmcache/v1/mp_observability/config.py

# SPDX-License-Identifier: Apache-2.0

"""
Configuration for the MP-mode observability stack.
"""

# Future
from __future__ import annotations

# Standard
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
import argparse
import uuid

if TYPE_CHECKING:
    # First Party
    from lmcache.v1.mp_observability.event_bus import EventBus

# First Party
from lmcache.v1.mp_observability.subscribers.logging.lookup_hash import (
    LookupHashLogConfig,
)


@dataclass
class ObservabilityConfig:
    """Unified configuration for the EventBus-based observability system.

    Controls the EventBus, OTel metrics/tracing pipelines, and subscriber
    registration.
    """

    enabled: bool = True
    """Master switch for the EventBus."""

    max_queue_size: int = 10_000
    """Maximum events in the EventBus queue before tail-drop."""

    metrics_enabled: bool = True
    """Register metrics subscribers (OTel counters / histograms)."""

    logging_enabled: bool = True
    """Register logging subscribers."""

    tracing_enabled: bool = False
    """Register span subscribers (OTel traces)."""

    otlp_endpoint: str | None = None
    """OTLP gRPC endpoint (e.g. ``http://localhost:4317``).  When set,
    metrics and traces are pushed to an OTel collector.  When ``None``,
    metrics fall back to an in-process Prometheus ``/metrics`` endpoint."""

    prometheus_port: int = 9090
    """Port for the Prometheus /metrics endpoint.  Only used when
    ``otlp_endpoint`` is ``None`` (Prometheus pull fallback)."""

    metrics_sample_rate: float = 0.01
    """Fraction of chunks/blocks to track for lifecycle histograms (0, 1.0].
    Counters always count all events regardless of this setting."""

    lookup_hash_log: LookupHashLogConfig = field(default_factory=LookupHashLogConfig)
    """Configuration for lookup hash file logging.  Disabled by default
    (empty ``output_dir``)."""

    trace_level: str | None = None
    """If set, enables trace recording at the given level.  Currently
    only ``"storage"`` is supported.  See
    :mod:`lmcache.v1.mp_observability.trace` for details."""

    trace_output: str | None = None
    """Path to write the trace file.  When :attr:`trace_level` is set
    but this is ``None``, a timestamped path under ``$TMPDIR`` is
    minted and logged at INFO."""

    service_instance_id: str | None = None
    """OTel ``service.instance.id`` resource attribute, attached to every
    metric and span. There is no CLI flag for it: the operator-facing id is
    ``--instance-id`` (``MPServerConfig.instance_id``), which ``run_cache_server``
    projects onto this attribute so telemetry and coordinator membership share
    one id.

    ``None`` (the default) means "not set": standalone callers that build an
    ``ObservabilityConfig`` directly (e.g. the trace CLI driver) fall back to a
    random UUID v4 at ``init_observability`` time. An explicit value is
    preserved verbatim."""


DEFAULT_OBSERVABILITY_CONFIG = ObservabilityConfig(enabled=False)


def add_observability_args(
    parser: argparse.ArgumentParser,
) -> argparse.ArgumentParser:
    """Add observability configuration arguments to an existing parser.

    Args:
        parser: The argument parser to add arguments to.

    Returns:
        The same parser with observability arguments added.
    """
    group = parser.add_argument_group(
        "Observability", "Configuration for metrics, logging, and tracing"
    )
    group.add_argument(
        "--disable-observability",
        action="store_true",
        default=False,
        help="Disable the observability EventBus entirely.",
    )
    group.add_argument(
        "--disable-metrics",
        action="store_true",
        default=False,
        help="Disable metrics subscribers (OTel counters).",
    )
    group.add_argument(
        "--disable-logging",
        action="store_true",
        default=False,
        help="Disable logging subscribers.",
    )
    group.add_argument(
        "--enable-tracing",
        action="store_true",
        default=False,
        help="Enable span subscribers (OTel traces). Disabled by default.",
    )
    group.add_argument(
        "--otlp-endpoint",
        type=str,
        default=None,
        help=(
            "OTLP gRPC endpoint (e.g. http://localhost:4317). "
            "When set, metrics/traces are pushed to an OTel collector. "
            "When unset, falls back to Prometheus pull mode."
        ),
    )
    group.add_argument(
        "--event-bus-queue-size",
        type=int,
        default=10_000,
        help=(
            "Maximum number of events in the EventBus queue before "
            "tail-drop. Default is 10000."
        ),
    )
    group.add_argument(
        "--prometheus-port",
        type=int,
        default=9090,
        help=(
            "Port for the Prometheus /metrics endpoint. "
            "Only used when --otlp-endpoint is not set. Default is 9090."
        ),
    )
    group.add_argument(
        "--metrics-sample-rate",
        type=float,
        default=0.01,
        help=(
            "Fraction of chunks/blocks to track for lifecycle histograms "
            "(0, 1.0]. Counters always count all events. Default is 0.01 (1%%)."
        ),
    )

    # Lookup hash logging config
    log_group = parser.add_argument_group(
        "Lookup Hash Logging",
        "Configuration for lookup hash file logging (offline analysis)",
    )
    log_group.add_argument(
        "--lookup-hash-log-dir",
        type=str,
        default="",
        help="Directory to write lookup hash JSONL files for offline analysis. "
        "Empty string (default) disables logging.",
    )
    log_group.add_argument(
        "--lookup-hash-log-rotation-interval",
        type=int,
        default=6 * 3600,
        help="Time interval in seconds before rotating to a new log file. "
        "Default is 21600 (6 hours).",
    )
    log_group.add_argument(
        "--lookup-hash-log-rotation-max-size",
        type=int,
        default=100 * 1024 * 1024,
        help="Max file size in bytes before rotating even if the time "
        "interval has not elapsed. Default is 100MB (104857600).",
    )
    log_group.add_argument(
        "--lookup-hash-log-max-files",
        type=int,
        default=100,
        help="Max number of lookup hash log files to keep. "
        "Oldest files are deleted when this limit is exceeded. Default is 100.",
    )

    trace_group = parser.add_argument_group(
        "Trace Recording",
        "Capture LMCache operations to a binary trace file for replay "
        "(see `lmcache trace`).",
    )
    trace_group.add_argument(
        "--trace-level",
        type=str,
        choices=["storage"],
        default=None,
        help="Enable trace recording at the given level. Currently only "
        "'storage' is supported (records StorageManager public-API calls).",
    )
    trace_group.add_argument(
        "--trace-output",
        type=str,
        default=None,
        help="Path to write the trace file. Defaults to a timestamped "
        "file under $TMPDIR when --trace-level is set without an explicit "
        "output path.",
    )

    return parser


def parse_args_to_observability_config(
    args: argparse.Namespace,
) -> ObservabilityConfig:
    """Convert parsed command line arguments to an ObservabilityConfig.

    Args:
        args: Parsed arguments from the argument parser.

    Returns:
        The configuration object.
    """
    config = ObservabilityConfig(
        enabled=not args.disable_observability,
        max_queue_size=args.event_bus_queue_size,
        metrics_enabled=not args.disable_metrics,
        logging_enabled=not args.disable_logging,
        tracing_enabled=args.enable_tracing,
        otlp_endpoint=args.otlp_endpoint,
        prometheus_port=args.prometheus_port,
        metrics_sample_rate=args.metrics_sample_rate,
        lookup_hash_log=LookupHashLogConfig(
            output_dir=args.lookup_hash_log_dir,
            rotation_interval_sec=args.lookup_hash_log_rotation_interval,
            rotation_max_size=args.lookup_hash_log_rotation_max_size,
            max_files=args.lookup_hash_log_max_files,
        ),
        trace_level=args.trace_level,
        trace_output=args.trace_output,
    )

    if config.tracing_enabled and config.otlp_endpoint is None:
        raise ValueError(
            "--enable-tracing requires --otlp-endpoint to be set. "
            "Tracing needs an OTLP gRPC endpoint to export spans."
        )

    return config


def init_observability(
    obs_config: ObservabilityConfig,
    *,
    start_prometheus_http_server: bool = True,
) -> EventBus:
    """Initialize OTel providers, EventBus, and register subscribers.

    This is the single entry-point that every MP server calls at startup.
    Returns a **started** EventBus.

    Args:
        obs_config: Observability configuration.
        start_prometheus_http_server: Whether to start a standalone
            Prometheus HTTP server.  Set to ``False`` when an
            external HTTP framework already serves ``/metrics``.
    """
    # First Party
    from lmcache.v1.mp_observability.event_bus import (
        EventBusConfig,
        init_event_bus,
    )

    # Set up OTel providers BEFORE creating subscribers so that
    # module-level get_meter()/get_tracer() calls bind to the real provider
    instance_id = (
        obs_config.service_instance_id
        if obs_config.service_instance_id is not None
        else str(uuid.uuid4())
    )
    resource_attrs = {"service.instance.id": instance_id}

    if obs_config.enabled and obs_config.metrics_enabled:
        # First Party
        from lmcache.v1.mp_observability.otel_init import init_otel_metrics

        init_otel_metrics(
            otlp_endpoint=obs_config.otlp_endpoint,
            prometheus_port=obs_config.prometheus_port,
            resource_attributes=resource_attrs,
            start_http_server=start_prometheus_http_server,
        )

    if obs_config.enabled and obs_config.tracing_enabled:
        # First Party
        from lmcache.v1.mp_observability.otel_init import init_otel_tracing

        init_otel_tracing(
            otlp_endpoint=obs_config.otlp_endpoint,
            resource_attributes=resource_attrs,
        )

    bus = init_event_bus(
        EventBusConfig(
            enabled=obs_config.enabled,
            max_queue_size=obs_config.max_queue_size,
        )
    )

    if obs_config.metrics_enabled:
        # First Party
        from lmcache.v1.mp_observability.subscribers.metrics import (
            BlendMetricsSubscriber,
            EngineMetricsSubscriber,
            EventBusSelfMetricsSubscriber,
            L0L1ThroughputSubscriber,
            L0LifecycleSubscriber,
            L1EvictionLoopSubscriber,
            L1FailureMetricsSubscriber,
            L1LifecycleSubscriber,
            L1MetricsSubscriber,
            L2FailureMetricsSubscriber,
            L2MetricsSubscriber,
            L2ThroughputSubscriber,
            LookupMetricsSubscriber,
            SMLifecycleSubscriber,
            TimeoutMetricsSubscriber,
        )

        sample_rate = obs_config.metrics_sample_rate
        bus.register_subscriber(L0LifecycleSubscriber(sample_rate=sample_rate))
        bus.register_subscriber(L1MetricsSubscriber())
        bus.register_subscriber(L1LifecycleSubscriber(sample_rate=sample_rate))
        bus.register_subscriber(L1FailureMetricsSubscriber())
        bus.register_subscriber(L1EvictionLoopSubscriber())
        bus.register_subscriber(L0L1ThroughputSubscriber())
        bus.register_subscriber(L2MetricsSubscriber())
        bus.register_subscriber(L2FailureMetricsSubscriber())
        bus.register_subscriber(L2ThroughputSubscriber())
        bus.register_subscriber(LookupMetricsSubscriber())
        bus.register_subscriber(SMLifecycleSubscriber(sample_rate=sample_rate))
        bus.register_subscriber(BlendMetricsSubscriber())
        bus.register_subscriber(EngineMetricsSubscriber())
        bus.register_subscriber(EventBusSelfMetricsSubscriber(bus))
        bus.register_subscriber(TimeoutMetricsSubscriber())

    if obs_config.logging_enabled:
        # First Party
        from lmcache.v1.mp_observability.subscribers.logging import (
            BlendLoggingSubscriber,
            L1LoggingSubscriber,
            L2LoggingSubscriber,
            MPServerLoggingSubscriber,
            SMLoggingSubscriber,
            TimeoutLoggingSubscriber,
        )

        bus.register_subscriber(MPServerLoggingSubscriber())
        bus.register_subscriber(L1LoggingSubscriber())
        bus.register_subscriber(L2LoggingSubscriber())
        bus.register_subscriber(SMLoggingSubscriber())
        bus.register_subscriber(BlendLoggingSubscriber())
        bus.register_subscriber(TimeoutLoggingSubscriber())

    if obs_config.tracing_enabled:
        # First Party
        from lmcache.v1.mp_observability.subscribers.tracing import (
            BlendTracingSubscriber,
            MPServerTracingSubscriber,
            TimeoutTracingSubscriber,
            get_span_registry,
        )

        registry = get_span_registry()
        bus.register_subscriber(MPServerTracingSubscriber(registry))
        bus.register_subscriber(BlendTracingSubscriber(registry))
        bus.register_subscriber(TimeoutTracingSubscriber(registry))

    # Lookup hash file logging (independent of the logging_enabled flag —
    # it has its own enable gate via output_dir).
    if obs_config.lookup_hash_log.enabled:
        # First Party
        from lmcache.v1.mp_observability.subscribers.logging.lookup_hash import (
            LookupHashLoggingSubscriber,
        )

        bus.register_subscriber(LookupHashLoggingSubscriber(obs_config.lookup_hash_log))

    bus.start()
    return bus