File size: 3,089 Bytes
30e6218 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | """Prometheus metrics with bounded label cardinality.
Labels use route templates (``/api/v1/documents/{doc_id}``), never document
IDs or query text. Metrics are optional: the app works without the
``prometheus_client`` package.
"""
from __future__ import annotations
import logging
import time
from contextlib import contextmanager
from typing import Any, Iterator, Optional
logger = logging.getLogger(__name__)
try:
from prometheus_client import Counter, Gauge, Histogram, Info
AVAILABLE = True
except ImportError: # pragma: no cover
AVAILABLE = False
request_count: Any = None
request_latency: Any = None
active_requests: Any = None
stage_latency: Any = None
stage_outcomes: Any = None
documents_ready: Any = None
chunks_active: Any = None
system_info: Any = None
if AVAILABLE:
request_count = Counter(
"docintel_requests_total", "HTTP requests", ["method", "route", "status"]
)
request_latency = Histogram(
"docintel_request_duration_seconds", "HTTP request latency", ["method", "route"]
)
active_requests = Gauge("docintel_active_requests", "In-flight HTTP requests")
stage_latency = Histogram(
"docintel_stage_duration_seconds",
"Pipeline stage latency",
["stage"],
buckets=(0.005, 0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0),
)
stage_outcomes = Counter(
"docintel_stage_outcomes_total", "Pipeline stage outcomes", ["stage", "outcome"]
)
documents_ready = Gauge("docintel_documents_ready", "Documents with status READY")
chunks_active = Gauge("docintel_chunks_active", "Chunks of READY documents")
system_info = Info("docintel_build", "Build information")
def record_request(method: str, route: str, status_code: int, duration: float) -> None:
if not AVAILABLE:
return
status = (
"2xx"
if status_code < 300
else "3xx"
if status_code < 400
else "4xx"
if status_code < 500
else "5xx"
)
request_count.labels(method=method, route=route, status=status).inc()
request_latency.labels(method=method, route=route).observe(duration)
@contextmanager
def track_stage(stage: str) -> Iterator[None]:
start = time.perf_counter()
outcome = "ok"
try:
yield
except Exception:
outcome = "error"
raise
finally:
if AVAILABLE:
stage_latency.labels(stage=stage).observe(time.perf_counter() - start)
stage_outcomes.labels(stage=stage, outcome=outcome).inc()
def update_index_gauges(documents: int, chunks: int) -> None:
if AVAILABLE:
documents_ready.set(documents)
chunks_active.set(chunks)
def initialize_metrics(
app_version: str, environment: str, embedding: Optional[str], generation: str
) -> None:
if AVAILABLE:
system_info.info(
{
"version": app_version,
"environment": environment,
"embedding_provider": embedding or "none",
"generation_provider": generation,
}
)
|