Source code for lalandre_core.utils.metrics_utils

"""Shared Prometheus metric helpers reused across services."""

from typing import Any, Optional

LATENCY_BUCKETS = (
    0.01,
    0.025,
    0.05,
    0.1,
    0.25,
    0.5,
    1.0,
    2.5,
    5.0,
    10.0,
    30.0,
    60.0,
)

TOP_K_BUCKETS = (1, 3, 5, 10, 20, 30, 50, 100)




def status_class(status_code: int) -> str:
    """Collapse an HTTP status code into its class label such as ``2xx``."""
    return f"{int(status_code) // 100}xx"





def normalize_label(value: Any) -> str:
    """Normalize arbitrary metric label values into a lowercase token."""
    text = str(value or "unknown").strip().lower()
    if not text:
        return "unknown"
    return text.replace(" ", "_").replace("-", "_")





def normalize_search_mode(mode: Optional[str]) -> str:
    """Normalize a search mode label to one of the supported metric values."""
    normalized = normalize_label(mode)
    if normalized in {"semantic", "lexical", "hybrid"}:
        return normalized
    return "unknown"





def normalize_granularity(granularity: Optional[str]) -> str:
    """Normalize a granularity label to one of the metric-safe values."""
    normalized = normalize_label(granularity or "default")
    if normalized in {"subdivisions", "chunks", "all"}:
        return normalized
    return "default"





def classify_error(exc_or_reason: Any) -> tuple[str, str]:
    """Classify an error into (provider, error_type) for metrics labeling."""
    text = str(exc_or_reason or "").lower()

    provider = "unknown"
    if "qdrant" in text or "vector" in text:
        provider = "qdrant"
    elif "neo4j" in text or "cypher" in text:
        provider = "neo4j"
    elif "postgres" in text or "sqlalchemy" in text or "bm25" in text:
        provider = "postgres"
    elif "rerank" in text or "cross-encoder" in text or "sentence-transformer" in text:
        provider = "reranker"
    elif "mistral" in text or "openai" in text or "llamaindex" in text or "langchain" in text or "llm" in text:
        provider = "llm"

    error_type = "runtime_error"
    if "timeout" in text:
        error_type = "timeout"
    elif "rate limit" in text or "429" in text:
        error_type = "rate_limit"
    elif "connect" in text or "connection" in text or "unreachable" in text:
        error_type = "connection"
    elif "unauthorized" in text or "forbidden" in text or "401" in text or "403" in text:
        error_type = "auth"
    elif "502" in text or "503" in text or "504" in text:
        error_type = "upstream"
    elif "dimension" in text or "invalid" in text or "validation" in text:
        error_type = "validation"
    elif isinstance(exc_or_reason, BaseException):
        error_type = normalize_label(exc_or_reason.__class__.__name__)

    return provider, error_type