"""Monitoring and metrics for knowledge plane (Phase C3). Implementuje observability vrstvu založenou na: - Prometheus metrics (prometheus_client) pro /metrics endpoint - Prometheus instrumentation patterns (latency histograms, counters, summaries) Metriky (podle výzkumu: Redis LangCache, OpenTelemetry RAG pipeline, embedding stack latency benchmarks): 1. **Embedding latency**: histogram `knowledge_embedding_query_latency_seconds`, counter `knowledge_embedding_requests_total` s outcomes (success/error). 2. **Cache metrics**: countery `knowledge_embedding_cache_hits_total` / `knowledge_embedding_cache_misses_total`, `knowledge_search_cache_hits_total` / `knowledge_search_cache_misses_total`. 3. **Vector search**: histogram `knowledge_vector_search_latency_seconds`, counter `knowledge_vector_search_queries_total`. 4. **Batch processing**: countery `knowledge_batch_items_total`, `knowledge_batch_successes_total`, `knowledge_batch_failures_total`, `knowledge_batch_dead_lettered_total`. 5. **Sensitivity/PII**: countery `knowledge_sensitivity_detected_total` s sensitivity level labels. 6. **Health**: gauge `knowledge_component_health` (0/1 per component). Pattern: Prometheus metrics exporter (prometheus_client REGISTRY) + custom business metrics. Snapshot API pro CLI/health check. Použití: metrics = KnowledgeMetrics() metrics.record_embedding_latency(0.012) metrics.record_vector_search(latency=0.15, best_distance=0.42) snapshot = metrics.get_metrics_snapshot() """ from __future__ import annotations import logging import time from typing import Any from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram logger = logging.getLogger(__name__) class KnowledgeMetrics: """Central collection point for knowledge plane business metrics.""" def __init__(self, registry: CollectorRegistry | None = None) -> None: self.registry = registry or CollectorRegistry() self._start_time = time.monotonic() # ------------------------------------------------------------------ # Embedding metrics # ------------------------------------------------------------------ self.embedding_latency = Histogram( "knowledge_embedding_query_latency_seconds", "Latency of single embedding query in knowledge plane", buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0), registry=self.registry, ) self.embedding_requests_total = Counter( "knowledge_embedding_requests_total", "Total embedding requests", ["status"], registry=self.registry, ) self.embedding_batch_latency = Histogram( "knowledge_embedding_batch_latency_seconds", "Latency of batched embedding operations", buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0), registry=self.registry, ) # ------------------------------------------------------------------ # Embedding cache metrics # ------------------------------------------------------------------ self.embedding_cache_hits = Counter( "knowledge_embedding_cache_hits_total", "Total embedding cache hits", registry=self.registry, ) self.embedding_cache_misses = Counter( "knowledge_embedding_cache_misses_total", "Total embedding cache misses", registry=self.registry, ) # ------------------------------------------------------------------ # Search cache metrics # ------------------------------------------------------------------ self.search_cache_hits = Counter( "knowledge_search_cache_hits_total", "Total semantic search cache hits", registry=self.registry, ) self.search_cache_misses = Counter( "knowledge_search_cache_misses_total", "Total semantic search cache misses", registry=self.registry, ) # ------------------------------------------------------------------ # Vector search metrics # ------------------------------------------------------------------ self.vector_search_latency = Histogram( "knowledge_vector_search_latency_seconds", "Latency of vector similarity search", buckets=(0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0), registry=self.registry, ) self.vector_search_total = Counter( "knowledge_vector_search_queries_total", "Total vector search queries", registry=self.registry, ) # ------------------------------------------------------------------ # Batch processing metrics # ------------------------------------------------------------------ self.batch_items_total = Counter( "knowledge_batch_items_total", "Total batch items processed", registry=self.registry, ) self.batch_successes_total = Counter( "knowledge_batch_successes_total", "Total batch items successfully processed", registry=self.registry, ) self.batch_failures_total = Counter( "knowledge_batch_failures_total", "Total batch items failed (after retries)", registry=self.registry, ) self.batch_dead_lettered = Counter( "knowledge_batch_dead_lettered_total", "Total batch items moved to dead-letter queue", registry=self.registry, ) # ------------------------------------------------------------------ # Sensitivity / classification metrics # ------------------------------------------------------------------ self.sensitivity_detected = Counter( "knowledge_sensitivity_detected_total", "Total documents where PII/sensitivity was detected", ["sensitivity_level"], registry=self.registry, ) self.classification_latency = Histogram( "knowledge_classification_latency_seconds", "Classification latency", buckets=(0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0), registry=self.registry, ) # ------------------------------------------------------------------ # Knowledge graph metrics # ------------------------------------------------------------------ self.kg_entity_count = Gauge( "knowledge_entity_count", "Number of entities in knowledge graph", registry=self.registry, ) self.kg_edge_count = Gauge( "knowledge_entity_edge_count", "Number of edges in knowledge graph", registry=self.registry, ) self.kg_traversal_latency = Histogram( "knowledge_kg_traversal_latency_seconds", "Knowledge graph BFS traversal latency", buckets=(0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5), registry=self.registry, ) # ------------------------------------------------------------------ # Component health # ------------------------------------------------------------------ self.component_health = Gauge( "knowledge_component_health", "Component health status (1=healthy, 0=unhealthy)", ["component"], registry=self.registry, ) # ------------------------------------------------------------------ # Internal counters for snapshot (simple Python counters) # ------------------------------------------------------------------ self._internal_embedding_requests = 0 self._internal_embedding_errors = 0 self._internal_vector_searches = 0 self._internal_batch_items = 0 self._internal_batch_successes = 0 self._internal_batch_failures = 0 self._internal_batch_dead_lettered = 0 self._internal_sensitivity_detected = 0 self._internal_classification_requests = 0 self._internal_kg_traversals = 0 # ------------------------------------------------------------------ # Embedding metrics # ------------------------------------------------------------------ def record_embedding_latency(self, latency_seconds: float, *, cached: bool = False, error: bool = False) -> None: """Zaznamená latency embeddingu (případně cache hit / error).""" self.embedding_latency.observe(latency_seconds) if error: self.embedding_requests_total.labels(status="error").inc() self._internal_embedding_errors += 1 else: self.embedding_requests_total.labels(status="success").inc() self._internal_embedding_requests += 1 if cached: self.embedding_cache_hits.inc() else: self.embedding_cache_misses.inc() def record_embedding_batch_latency(self, latency_seconds: float) -> None: """Zaznamená latency batched embedding.""" self.embedding_batch_latency.observe(latency_seconds) # ------------------------------------------------------------------ # Cache metrics # ------------------------------------------------------------------ def record_search_cache_hit(self, hit: bool) -> None: """Zaznamená cache hit/miss pro semantic search.""" if hit: self.search_cache_hits.inc() else: self.search_cache_misses.inc() # ------------------------------------------------------------------ # Vector search metrics # ------------------------------------------------------------------ def record_vector_search(self, latency_seconds: float, best_distance: float | None = None) -> None: """Zaznamená vector search latency + score distribution.""" self.vector_search_latency.observe(latency_seconds) self.vector_search_total.inc() self._internal_vector_searches += 1 # ------------------------------------------------------------------ # Batch metrics # ------------------------------------------------------------------ def record_batch_item(self, *, success: bool, dead_lettered: bool) -> None: """Zaznamená batch záznam.""" self.batch_items_total.inc() self._internal_batch_items += 1 if dead_lettered: self.batch_dead_lettered.inc() self._internal_batch_dead_lettered += 1 return if success: self.batch_successes_total.inc() self._internal_batch_successes += 1 else: self.batch_failures_total.inc() self._internal_batch_failures += 1 # ------------------------------------------------------------------ # Sensitivity metrics # ------------------------------------------------------------------ def record_sensitivity(self, level: str, detected: bool) -> None: """Zaznamená detekci sensitivity/PII.""" if detected: self.sensitivity_detected.labels(sensitivity_level=level).inc() self._internal_sensitivity_detected += 1 def record_classification(self, latency_seconds: float, success: bool) -> None: """Zaznamená classification latency a výsledek.""" self.classification_latency.observe(latency_seconds) self._internal_classification_requests += 1 if not success: logger.warning("Classification failed") # ------------------------------------------------------------------ # Knowledge graph metrics # ------------------------------------------------------------------ def record_entity_count(self, count: int) -> None: """Zaznamená počet entit v grafe.""" self.kg_entity_count.set(count) def record_entity_edges(self, count: int) -> None: """Zaznamená počet hran v grafe.""" self.kg_edge_count.set(count) def record_kg_traversal(self, latency_seconds: float) -> None: """Zaznamená latency BFS traversalu v KG.""" self.kg_traversal_latency.observe(latency_seconds) self._internal_kg_traversals += 1 # ------------------------------------------------------------------ # Health # ------------------------------------------------------------------ def set_component_health(self, component: str, healthy: bool) -> None: """Nastaví health status komponenty.""" self.component_health.labels(component=component).set(1.0 if healthy else 0.0) # ------------------------------------------------------------------ # Snapshot # ------------------------------------------------------------------ def get_metrics_snapshot(self) -> dict[str, Any]: """Vrátí aktuální snapshot metrik pro CLI/health check.""" from prometheus_client import generate_latest # Helper to get Prometheus metric value via collect def _metric_value(metric: Any) -> float | None: try: collected = list(metric.collect()) if collected and collected[0].samples: last_sample = collected[0].samples[-1] return last_sample.value except (KeyError, AttributeError, IndexError): pass return None return { "start_time_monotonic": self._start_time, "embedding": { "requests_total": self._internal_embedding_requests, "errors": self._internal_embedding_errors, "cache_hits": _metric_value(self.embedding_cache_hits), "cache_misses": _metric_value(self.embedding_cache_misses), }, "vector_search": { "queries_total": self._internal_vector_searches, "latest_latency": _metric_value(self.vector_search_latency), }, "batch": { "items_total": self._internal_batch_items, "successes": self._internal_batch_successes, "failures": self._internal_batch_failures, "dead_lettered": self._internal_batch_dead_lettered, }, "sensitivity_detected": self._internal_sensitivity_detected, "kg_entities": _metric_value(self.kg_entity_count), "kg_edges": _metric_value(self.kg_edge_count), "component_health": { c: _metric_value(self.component_health.labels(component=c)) for c in ("embedding", "vector_store", "navigator") }, "prometheus_text": generate_latest(self.registry).decode("utf-8"), } def get_default_metrics() -> KnowledgeMetrics: """Vrátí globální default instance metrik.""" return KnowledgeMetrics() __all__ = ["KnowledgeMetrics", "get_default_metrics"]