From 5d205fcaab6f7018e68ac9ac6718897223f36f86 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Tue, 2 Jun 2026 17:46:20 +0200 Subject: [PATCH] feat(observability): astrolabe_* metrics + traces for the document pipeline Make per-tier bottlenecks in the document-processing pipeline (scan -> fetch -> parse -> chunk -> embed -> Qdrant upsert) visible via metrics, traces, and structured logs. Today the document_processors layer emits only a logger.info line: no metric, no span, and page counts live only inside a log string. The single processing-duration histogram is unlabeled and whole-document, so it cannot isolate parse vs embed vs upsert. New astrolabe_* metric family (distinct from the mcp_* protocol metrics): - astrolabe_document_parse_{duration_seconds,total} + pages/chars/bytes counters recorded at the ProcessorRegistry.process() boundary (covers all current and future processors uniformly) - astrolabe_document_escalation_total (dormant; tiered-pipeline readiness) - astrolabe_embedding_{duration_seconds,requests_total,chunks_total,chars_total} - astrolabe_document_chunks_total, astrolabe_documents_indexed_total{source,status} Tracing: new document_processor.parse child span + enriched embed/chunk span attributes (provider/model/batch_size/chunk_count). Structured logs gain a consistent field vocabulary (doc_id, doc_type, processor, tier, pages, chars, byte_size, chunks, duration_ms, status) so Loki can aggregate without regex. Tier-readiness: processor/tier are labels from day one and a tier property is added to DocumentProcessor, so adding docling/OCR/LLM tiers later is additive (new label values, never new metrics). Tenant comes from the kube namespace label; mime_type/model are span attributes only (cardinality). Existing mcp_vector_sync_*/mcp_qdrant_* are left untouched. Refs Deck #175 (superset of #173 Phase 2). Dashboard/recording-rules follow-up tracked on #175 for homelab-argocd. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/config.py | 44 ++++ .../document_processors/base.py | 13 + .../document_processors/pymupdf.py | 8 + .../document_processors/registry.py | 87 ++++++- nextcloud_mcp_server/observability/metrics.py | 210 ++++++++++++++- nextcloud_mcp_server/vector/processor.py | 90 ++++++- tests/unit/test_document_parse_metrics.py | 239 ++++++++++++++++++ tests/unit/test_embedding_metrics.py | 122 +++++++++ 8 files changed, 799 insertions(+), 14 deletions(-) create mode 100644 tests/unit/test_document_parse_metrics.py create mode 100644 tests/unit/test_embedding_metrics.py diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index d4e60c9e..eebf3a98 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -949,6 +949,50 @@ class Settings: return f"simple-{self.simple_embedding_dimension}" + def get_embedding_provider_family(self) -> str: + """ + Get the active dense-embedding provider family (a low-cardinality label). + + This is the single source of truth for the ``provider`` metric label and + the ``embedding.provider`` span attribute. It returns the provider + *family* (e.g. "bedrock"), never the model name, to keep metric + cardinality bounded. + + Priority mirrors ``get_embedding_model_name`` / ProviderRegistry: + 1. Gateway - if EMBEDDING_PROVIDER=gateway (family from the model prefix, + e.g. "mistral/mistral-embed" -> "mistral") + 2. Bedrock - if AWS_REGION or BEDROCK_EMBEDDING_MODEL is set + 3. OpenAI - if OPENAI_API_KEY is set + 4. Mistral - if MISTRAL_API_KEY is set + 5. Ollama - if OLLAMA_BASE_URL is set + 6. Simple - fallback + + Returns: + Provider family: gateway-routed family | bedrock | openai | mistral + | ollama | simple + """ + if self.embedding_provider == "gateway": + model = self.embedding_gateway_model or "" + return model.split("/", 1)[0] if "/" in model else "gateway" + + if ( + self.aws_region + or self.bedrock_embedding_model + or self.bedrock_generation_model + ): + return "bedrock" + + if self.openai_api_key: + return "openai" + + if self.mistral_api_key: + return "mistral" + + if self.ollama_base_url: + return "ollama" + + return "simple" + def get_collection_name(self) -> str: """ Get Qdrant collection name. diff --git a/nextcloud_mcp_server/document_processors/base.py b/nextcloud_mcp_server/document_processors/base.py index f812a264..86ab1850 100644 --- a/nextcloud_mcp_server/document_processors/base.py +++ b/nextcloud_mcp_server/document_processors/base.py @@ -56,6 +56,19 @@ class DocumentProcessor(ABC): """Unique identifier for this processor (e.g., 'unstructured', 'tesseract').""" pass + @property + def tier(self) -> str: + """Extraction tier this processor belongs to (escalation ladder). + + Used as the ``tier`` label/attribute in observability so that adding new + extraction tiers later (docling, OCR, LLM) is purely additive. Vocabulary + (cheapest first): ``fast`` -> ``structured`` -> ``ocr`` -> ``llm``. + + Defaults to ``"fast"``; override in processors that belong to a higher + tier. + """ + return "fast" + @property @abstractmethod def supported_mime_types(self) -> set[str]: diff --git a/nextcloud_mcp_server/document_processors/pymupdf.py b/nextcloud_mcp_server/document_processors/pymupdf.py index b15aafca..96e464fd 100644 --- a/nextcloud_mcp_server/document_processors/pymupdf.py +++ b/nextcloud_mcp_server/document_processors/pymupdf.py @@ -181,6 +181,14 @@ class PyMuPDFProcessor(DocumentProcessor): metadata["page_count"], len(md_text), metadata.get("image_count", 0), + extra={ + "processor": self.name, + "tier": self.tier, + "pages": metadata["page_count"], + "chars": len(md_text), + "images": metadata.get("image_count", 0), + "byte_size": len(content), + }, ) return ProcessingResult( diff --git a/nextcloud_mcp_server/document_processors/registry.py b/nextcloud_mcp_server/document_processors/registry.py index 4bed37fe..8f434afa 100644 --- a/nextcloud_mcp_server/document_processors/registry.py +++ b/nextcloud_mcp_server/document_processors/registry.py @@ -1,9 +1,13 @@ """Central registry for document processors.""" import logging +import time from collections.abc import Awaitable, Callable from typing import Any, Optional +from nextcloud_mcp_server.observability.metrics import record_document_parse +from nextcloud_mcp_server.observability.tracing import trace_operation + from .base import DocumentProcessor, ProcessingResult, ProcessorError logger = logging.getLogger(__name__) @@ -152,13 +156,86 @@ class ProcessorRegistry: f"Registered processors: {', '.join(self.list_processors())}" ) - logger.info("Processing with '%s' processor", processor.name) - - # Process - return await processor.process( - content, content_type, filename, options, progress_callback + tier = processor.tier + logger.info( + "Processing with '%s' processor", + processor.name, + extra={ + "processor": processor.name, + "tier": tier, + "mime_type": content_type, + }, ) + # Process (instrumented: per-processor span + parse metrics). + # NOTE: when the tiered pipeline (docling/OCR/LLM) lands, escalation + # decisions are recorded here via record_document_escalation() and an + # add_span_event("document.escalation", ...) -- the escalated=False + # attribute and the metric are wired ahead of that. + byte_size = len(content) + start_time = time.time() + with trace_operation( + "document_processor.parse", + attributes={ + "processor.name": processor.name, + "processor.tier": tier, + "mime_type": content_type, + "byte_size": byte_size, + "escalated": False, + }, + ) as span: + try: + result = await processor.process( + content, content_type, filename, options, progress_callback + ) + except Exception: + duration = time.time() - start_time + record_document_parse( + processor.name, + tier, + duration, + byte_size=byte_size, + status="error", + ) + raise + + duration = time.time() - start_time + pages = int(result.metadata.get("page_count", 0) or 0) + chars = len(result.text) + status = "success" if result.success else "error" + record_document_parse( + processor.name, + tier, + duration, + pages=pages, + chars=chars, + byte_size=byte_size, + status=status, + ) + if span is not None: + span.set_attribute("page_count", pages) + span.set_attribute("char_count", chars) + span.set_attribute("processor.success", result.success) + + logger.info( + "Parsed %s with '%s': %s pages, %s chars in %.2fs", + filename or "", + processor.name, + pages, + chars, + duration, + extra={ + "processor": processor.name, + "tier": tier, + "pages": pages, + "chars": chars, + "byte_size": byte_size, + "duration_ms": round(duration * 1000, 1), + "status": status, + }, + ) + return result + # Global registry instance _registry = ProcessorRegistry() diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 5dcccaae..4f4c9fcb 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -175,6 +175,109 @@ qdrant_operations_total = Counter( ], # operation: upsert | search | delete; status: success | error ) +# ============================================================================= +# Astrolabe Document-Processing Pipeline Metrics +# ============================================================================= +# +# Product-signal metrics for the document-processing pipeline +# (scan -> fetch -> parse -> chunk -> embed -> Qdrant upsert). These use the +# ``astrolabe_`` prefix to distinguish the indexing/product pipeline from the +# ``mcp_`` protocol metrics above. The tenant dimension is NOT a label here -- +# it is supplied by the Kubernetes ``namespace`` label at scrape time. +# +# Tiered-pipeline readiness: ``processor`` and ``tier`` are labels from day one +# so that adding new extraction tiers (docling, OCR, LLM) later is purely +# additive (new label values), never new metric names. +# tier vocabulary (escalation ladder): fast -> structured -> ocr -> llm +# +# Cardinality rule: ``mime_type`` and embedding ``model`` are span attributes +# only, never metric labels. + +# --- Parse tier (recorded at the ProcessorRegistry.process() boundary) -------- + +document_parse_duration_seconds = Histogram( + "astrolabe_document_parse_duration_seconds", + "Document text-extraction (parse) duration in seconds", + ["processor", "tier", "status"], # status: success | error + # Buckets reach 300s: large PDFs exceed the 60s ceiling of the whole-doc + # histogram, which would otherwise pile every large parse into +Inf. + buckets=(0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0), +) + +document_parse_total = Counter( + "astrolabe_document_parse_total", + "Total document parse attempts", + ["processor", "tier", "status"], # status: success | error +) + +document_pages_processed_total = Counter( + "astrolabe_document_pages_processed_total", + "Total document pages processed (page-rate signal)", + ["processor", "tier"], +) + +document_chars_processed_total = Counter( + "astrolabe_document_chars_processed_total", + "Total characters extracted from documents", + ["processor", "tier"], +) + +document_bytes_processed_total = Counter( + "astrolabe_document_bytes_processed_total", + "Total bytes of source documents parsed", + ["processor", "tier"], +) + +# --- Escalation (tiered-pipeline readiness; ~0 until extra tiers exist) -------- + +document_escalation_total = Counter( + "astrolabe_document_escalation_total", + "Total document parse escalations between tiers", + # reason: low_confidence | empty_text | unsupported | error | forced + ["from_tier", "to_tier", "reason"], +) + +# --- Embedding stages --------------------------------------------------------- + +embedding_duration_seconds = Histogram( + "astrolabe_embedding_duration_seconds", + "Embedding batch duration in seconds", + ["kind", "provider", "status"], # kind: dense | sparse + buckets=(0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0), +) + +embedding_requests_total = Counter( + "astrolabe_embedding_requests_total", + "Total embedding batch calls", + ["kind", "provider", "status"], # one per embed_batch / encode_batch call +) + +embedding_chunks_total = Counter( + "astrolabe_embedding_chunks_total", + "Total chunks embedded", + ["kind", "provider"], +) + +embedding_chars_total = Counter( + "astrolabe_embedding_chars_total", + "Total characters embedded", + ["kind", "provider"], +) + +# --- Chunking & indexed-by-type ----------------------------------------------- + +document_chunks_total = Counter( + "astrolabe_document_chunks_total", + "Total chunks produced by the chunker", + ["doc_type"], +) + +documents_indexed_total = Counter( + "astrolabe_documents_indexed_total", + "Total documents indexed, by source type", + ["source", "status"], # source: note | file | deck_card | news_item +) + # ============================================================================= # Database Metrics # ============================================================================= @@ -363,16 +466,25 @@ def record_vector_sync_scan(documents_found: int) -> None: vector_sync_documents_scanned_total.inc(documents_found) -def record_vector_sync_processing(duration: float, status: str = "success") -> None: +def record_vector_sync_processing( + duration: float, status: str = "success", doc_type: str | None = None +) -> None: """ Record document processing with duration and status. Args: duration: Processing duration in seconds status: "success" or "error" + doc_type: Optional document source type (note, file, deck_card, + news_item). When supplied, also increments the per-type + ``astrolabe_documents_indexed_total`` counter. The legacy + ``mcp_vector_sync_documents_processed_total`` counter is always + incremented for backward compatibility. """ vector_sync_documents_processed_total.labels(status=status).inc() vector_sync_processing_duration_seconds.observe(duration) + if doc_type is not None: + documents_indexed_total.labels(source=doc_type, status=status).inc() def record_qdrant_operation(operation: str, status: str = "success") -> None: @@ -396,6 +508,102 @@ def update_vector_sync_queue_size(size: int) -> None: vector_sync_queue_size.set(size) +def record_document_parse( + processor: str, + tier: str, + duration: float, + pages: int = 0, + chars: int = 0, + byte_size: int = 0, + status: str = "success", +) -> None: + """ + Record a document parse (text extraction) at the processor boundary. + + Args: + processor: Processor name (e.g. "pymupdf", "unstructured", "tesseract") + tier: Extraction tier (fast | structured | ocr | llm) + duration: Parse duration in seconds + pages: Number of pages parsed (0 if not page-based) + chars: Number of characters extracted + byte_size: Size of the source document in bytes + status: "success" or "error" + """ + document_parse_duration_seconds.labels( + processor=processor, tier=tier, status=status + ).observe(duration) + document_parse_total.labels(processor=processor, tier=tier, status=status).inc() + if status == "success": + if pages: + document_pages_processed_total.labels(processor=processor, tier=tier).inc( + pages + ) + if chars: + document_chars_processed_total.labels(processor=processor, tier=tier).inc( + chars + ) + if byte_size: + document_bytes_processed_total.labels(processor=processor, tier=tier).inc( + byte_size + ) + + +def record_document_escalation(from_tier: str, to_tier: str, reason: str) -> None: + """ + Record a document parse escalation between tiers. + + Args: + from_tier: Tier that could not satisfactorily parse the document + to_tier: Tier the document was escalated to + reason: low_confidence | empty_text | unsupported | error | forced + """ + document_escalation_total.labels( + from_tier=from_tier, to_tier=to_tier, reason=reason + ).inc() + + +def record_embedding( + kind: str, + provider: str, + duration: float, + chunks: int = 0, + chars: int = 0, + status: str = "success", +) -> None: + """ + Record an embedding batch call. + + Args: + kind: "dense" or "sparse" + provider: Provider family (bedrock | openai | mistral | ollama | simple + for dense; "bm25" for sparse) + duration: Batch duration in seconds + chunks: Number of chunks embedded + chars: Total characters embedded + status: "success" or "error" + """ + embedding_duration_seconds.labels( + kind=kind, provider=provider, status=status + ).observe(duration) + embedding_requests_total.labels(kind=kind, provider=provider, status=status).inc() + if status == "success": + if chunks: + embedding_chunks_total.labels(kind=kind, provider=provider).inc(chunks) + if chars: + embedding_chars_total.labels(kind=kind, provider=provider).inc(chars) + + +def record_document_chunks(doc_type: str, count: int) -> None: + """ + Record the number of chunks produced for a document. + + Args: + doc_type: Document source type (note, file, deck_card, news_item) + count: Number of chunks produced + """ + document_chunks_total.labels(doc_type=doc_type).inc(count) + + # ============================================================================= # Decorator for Automatic Tool Instrumentation # ============================================================================= diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 4e808bf2..94837071 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -20,6 +20,8 @@ from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.document_processors import get_registry from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_service from nextcloud_mcp_server.observability.metrics import ( + record_document_chunks, + record_embedding, record_qdrant_operation, record_vector_sync_processing, update_vector_sync_queue_size, @@ -209,12 +211,19 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): doc_task.doc_type, doc_task.doc_id, doc_task.user_id, + extra={ + "doc_id": doc_task.doc_id, + "doc_type": doc_task.doc_type, + "status": "success", + }, ) # Record successful deletion metrics duration = time.time() - start_time record_qdrant_operation("delete", "success") - record_vector_sync_processing(duration, "success") + record_vector_sync_processing( + duration, "success", doc_type=doc_task.doc_type + ) return # Handle indexing with retry @@ -228,7 +237,9 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): # Record successful processing metrics duration = time.time() - start_time record_qdrant_operation("upsert", "success") - record_vector_sync_processing(duration, "success") + record_vector_sync_processing( + duration, "success", doc_type=doc_task.doc_type + ) return # Success except (HTTPStatusError, Exception) as e: @@ -240,6 +251,13 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): doc_task.doc_type, doc_task.doc_id, e, + extra={ + "doc_id": doc_task.doc_id, + "doc_type": doc_task.doc_type, + "attempt": attempt + 1, + "max_retries": max_retries, + "status": "retry", + }, ) await anyio.sleep(retry_delay) retry_delay *= 2 # Exponential backoff @@ -250,17 +268,26 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): doc_task.doc_id, max_retries, e, + extra={ + "doc_id": doc_task.doc_id, + "doc_type": doc_task.doc_type, + "attempt": max_retries, + "max_retries": max_retries, + "status": "error", + }, ) # Record failed processing metrics duration = time.time() - start_time record_qdrant_operation("upsert", "error") - record_vector_sync_processing(duration, "error") + record_vector_sync_processing( + duration, "error", doc_type=doc_task.doc_type + ) raise except Exception: # Catch any other unexpected errors duration = time.time() - start_time - record_vector_sync_processing(duration, "error") + record_vector_sync_processing(duration, "error", doc_type=doc_task.doc_type) raise @@ -512,12 +539,15 @@ async def _index_document( "vector_sync.chunk_size": settings.document_chunk_size, "vector_sync.overlap": settings.document_chunk_overlap, }, - ): + ) as chunk_span: chunker = DocumentChunker( chunk_size=settings.document_chunk_size, overlap=settings.document_chunk_overlap, ) chunks = await chunker.chunk_text(content) + record_document_chunks(doc_task.doc_type, len(chunks)) + if chunk_span is not None: + chunk_span.set_attribute("vector_sync.chunk_count", len(chunks)) # Assign page numbers to chunks if page boundaries are available (PDFs) page_boundaries = file_metadata.get("page_boundaries") @@ -583,27 +613,65 @@ async def _index_document( async def generate_dense_embeddings(): """Generate dense embeddings (I/O bound - external API call).""" nonlocal dense_embeddings + provider = settings.get_embedding_provider_family() + total_chars = sum(len(t) for t in chunk_texts) with trace_operation( "vector_sync.embed_dense", attributes={ "vector_sync.chunk_count": len(chunk_texts), - "vector_sync.total_chars": sum(len(t) for t in chunk_texts), + "vector_sync.total_chars": total_chars, + "embedding.kind": "dense", + "embedding.provider": provider, + "embedding.model": settings.get_embedding_model_name(), + "embedding.batch_size": len(chunk_texts), }, ): embedding_service = get_embedding_service() - dense_embeddings = await embedding_service.embed_batch(chunk_texts) + embed_start = time.time() + try: + dense_embeddings = await embedding_service.embed_batch(chunk_texts) + except Exception: + record_embedding( + "dense", provider, time.time() - embed_start, status="error" + ) + raise + record_embedding( + "dense", + provider, + time.time() - embed_start, + chunks=len(chunk_texts), + chars=total_chars, + ) async def generate_sparse_embeddings(): """Generate sparse embeddings (BM25 for keyword matching).""" nonlocal sparse_embeddings + total_chars = sum(len(t) for t in chunk_texts) with trace_operation( "vector_sync.embed_sparse", attributes={ "vector_sync.chunk_count": len(chunk_texts), + "embedding.kind": "sparse", + "embedding.provider": "bm25", + "embedding.batch_size": len(chunk_texts), }, ): bm25_service = await get_bm25_service() - sparse_embeddings = await bm25_service.encode_batch(chunk_texts) + embed_start = time.time() + try: + sparse_embeddings = await bm25_service.encode_batch(chunk_texts) + except Exception: + record_embedding( + "sparse", "bm25", time.time() - embed_start, status="error" + ) + raise + record_embedding( + "sparse", + "bm25", + time.time() - embed_start, + chunks=len(chunk_texts), + chars=total_chars, + ) async def generate_highlights(): """Compute chunk bounding boxes for PDF chunks (CPU-bound, no rendering).""" @@ -853,4 +921,10 @@ async def _index_document( doc_task.doc_id, doc_task.user_id, len(chunks), + extra={ + "doc_id": doc_task.doc_id, + "doc_type": doc_task.doc_type, + "chunks": len(chunks), + "status": "success", + }, ) diff --git a/tests/unit/test_document_parse_metrics.py b/tests/unit/test_document_parse_metrics.py new file mode 100644 index 00000000..e27167ee --- /dev/null +++ b/tests/unit/test_document_parse_metrics.py @@ -0,0 +1,239 @@ +"""Unit tests for document-parse instrumentation. + +Covers two layers: +1. The ``ProcessorRegistry.process()`` boundary — that it records a parse metric + (success and error) and opens a ``document_processor.parse`` span with the + expected attributes, while preserving the existing re-raise on failure. +2. The ``record_document_parse`` / ``record_document_chunks`` / + ``record_vector_sync_processing`` helpers — that they increment the right + ``astrolabe_*`` Prometheus series (and that an error parse does NOT bump the + throughput counters). +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from prometheus_client import REGISTRY + +from nextcloud_mcp_server.document_processors.base import ( + DocumentProcessor, + ProcessingResult, + ProcessorError, +) +from nextcloud_mcp_server.document_processors.registry import ProcessorRegistry +from nextcloud_mcp_server.observability.metrics import ( + record_document_chunks, + record_document_escalation, + record_document_parse, + record_vector_sync_processing, +) + +pytestmark = pytest.mark.unit + + +def _sample(name: str, labels: dict[str, str]) -> float: + """Return a Prometheus sample value, treating 'never observed' as 0.""" + return REGISTRY.get_sample_value(name, labels) or 0.0 + + +class _FakeProcessor(DocumentProcessor): + """Minimal processor for exercising the registry instrumentation.""" + + def __init__( + self, + *, + result: ProcessingResult | None = None, + exc: Exception | None = None, + proc_name: str = "pymupdf", + proc_tier: str = "fast", + ): + self._result = result + self._exc = exc + self._name = proc_name + self._tier = proc_tier + + @property + def name(self) -> str: + return self._name + + @property + def tier(self) -> str: + return self._tier + + @property + def supported_mime_types(self) -> set[str]: + return {"application/pdf"} + + async def process( + self, + content: bytes, + content_type: str, + filename: str | None = None, + options: dict[str, Any] | None = None, + progress_callback=None, + ) -> ProcessingResult: + if self._exc is not None: + raise self._exc + assert self._result is not None + return self._result + + async def health_check(self) -> bool: + return True + + +@pytest.fixture +def mock_tracer(): + """Patch trace_operation in the registry; expose the yielded span.""" + with patch( + "nextcloud_mcp_server.document_processors.registry.trace_operation" + ) as mock_trace: + span = MagicMock() + mock_trace.return_value.__enter__ = MagicMock(return_value=span) + mock_trace.return_value.__exit__ = MagicMock(return_value=False) + mock_trace.span = span + yield mock_trace + + +class TestRegistryParseInstrumentation: + async def test_success_records_metric_and_span(self, mock_tracer): + result = ProcessingResult( + text="x" * 1000, + metadata={"page_count": 50, "file_size": 99}, + processor="pymupdf", + ) + registry = ProcessorRegistry() + registry.register(_FakeProcessor(result=result)) + + with patch( + "nextcloud_mcp_server.document_processors.registry.record_document_parse" + ) as mock_record: + out = await registry.process( + b"%PDF-1.7", "application/pdf", filename="x.pdf" + ) + + assert out is result + + # Metric recorded with parsed pages/chars and success status. + mock_record.assert_called_once() + args = mock_record.call_args.args + kwargs = mock_record.call_args.kwargs + assert args[0] == "pymupdf" # processor + assert args[1] == "fast" # tier + assert kwargs["pages"] == 50 + assert kwargs["chars"] == 1000 + assert kwargs["status"] == "success" + + # Span opened with the parse name + identifying attributes. + assert mock_tracer.call_args.args[0] == "document_processor.parse" + attrs = mock_tracer.call_args.kwargs["attributes"] + assert attrs["processor.name"] == "pymupdf" + assert attrs["processor.tier"] == "fast" + assert attrs["mime_type"] == "application/pdf" + assert attrs["escalated"] is False + # Post-parse attributes set on the span. + mock_tracer.span.set_attribute.assert_any_call("page_count", 50) + mock_tracer.span.set_attribute.assert_any_call("char_count", 1000) + + async def test_error_records_error_metric_and_reraises(self, mock_tracer): + registry = ProcessorRegistry() + registry.register(_FakeProcessor(exc=ProcessorError("boom"))) + + with patch( + "nextcloud_mcp_server.document_processors.registry.record_document_parse" + ) as mock_record: + with pytest.raises(ProcessorError): + await registry.process(b"data", "application/pdf") + + mock_record.assert_called_once() + assert mock_record.call_args.kwargs["status"] == "error" + + +class TestParseMetricHelpers: + def test_success_increments_throughput_counters(self): + labels = {"processor": "uttest-success", "tier": "fast"} + before_pages = _sample("astrolabe_document_pages_processed_total", labels) + before_chars = _sample("astrolabe_document_chars_processed_total", labels) + before_bytes = _sample("astrolabe_document_bytes_processed_total", labels) + before_total = _sample( + "astrolabe_document_parse_total", {**labels, "status": "success"} + ) + + record_document_parse( + "uttest-success", + "fast", + 1.23, + pages=50, + chars=1000, + byte_size=99, + status="success", + ) + + assert _sample("astrolabe_document_pages_processed_total", labels) == ( + before_pages + 50 + ) + assert _sample("astrolabe_document_chars_processed_total", labels) == ( + before_chars + 1000 + ) + assert _sample("astrolabe_document_bytes_processed_total", labels) == ( + before_bytes + 99 + ) + assert _sample( + "astrolabe_document_parse_total", {**labels, "status": "success"} + ) == (before_total + 1) + # The duration histogram observed one sample. + assert ( + _sample( + "astrolabe_document_parse_duration_seconds_count", + {**labels, "status": "success"}, + ) + >= 1 + ) + + def test_error_does_not_increment_throughput(self): + labels = {"processor": "uttest-error", "tier": "fast"} + record_document_parse( + "uttest-error", + "fast", + 0.5, + pages=10, + chars=10, + byte_size=10, + status="error", + ) + # Error parses count the attempt + duration, but NOT pages/chars/bytes. + assert _sample("astrolabe_document_pages_processed_total", labels) == 0.0 + assert _sample("astrolabe_document_chars_processed_total", labels) == 0.0 + assert ( + _sample("astrolabe_document_parse_total", {**labels, "status": "error"}) + == 1.0 + ) + + def test_record_document_chunks(self): + labels = {"doc_type": "uttest-chunks"} + before = _sample("astrolabe_document_chunks_total", labels) + record_document_chunks("uttest-chunks", 7) + assert _sample("astrolabe_document_chunks_total", labels) == before + 7 + + def test_vector_sync_processing_increments_documents_indexed(self): + labels = {"source": "uttest-doctype", "status": "success"} + before = _sample("astrolabe_documents_indexed_total", labels) + record_vector_sync_processing(0.1, "success", doc_type="uttest-doctype") + assert _sample("astrolabe_documents_indexed_total", labels) == before + 1 + + def test_vector_sync_processing_without_doc_type_is_noop_for_indexed(self): + # Without doc_type, the per-type counter must not be touched (the legacy + # mcp_* counter still increments, but that is out of scope here). + labels = {"source": "uttest-absent", "status": "success"} + record_vector_sync_processing(0.1, "success") + assert _sample("astrolabe_documents_indexed_total", labels) == 0.0 + + def test_record_document_escalation(self): + # Dormant until the tiered pipeline lands; pin its correctness now so the + # first docling/OCR/LLM caller gets a working counter. + labels = {"from_tier": "fast", "to_tier": "ocr", "reason": "empty_text"} + before = _sample("astrolabe_document_escalation_total", labels) + record_document_escalation("fast", "ocr", "empty_text") + assert _sample("astrolabe_document_escalation_total", labels) == before + 1 diff --git a/tests/unit/test_embedding_metrics.py b/tests/unit/test_embedding_metrics.py new file mode 100644 index 00000000..3e6c4a15 --- /dev/null +++ b/tests/unit/test_embedding_metrics.py @@ -0,0 +1,122 @@ +"""Unit tests for embedding observability. + +Covers: +1. ``Settings.get_embedding_provider_family()`` — the single source of truth for + the ``provider`` metric label / span attribute — across provider configs. +2. The ``record_embedding`` helper — that it increments the right + ``astrolabe_embedding_*`` series and skips the throughput counters on error. +""" + +from __future__ import annotations + +import pytest +from prometheus_client import REGISTRY + +from nextcloud_mcp_server.config import Settings +from nextcloud_mcp_server.observability.metrics import record_embedding + +pytestmark = pytest.mark.unit + + +def _sample(name: str, labels: dict[str, str]) -> float: + return REGISTRY.get_sample_value(name, labels) or 0.0 + + +class TestProviderFamily: + """Provider-family detection mirrors ProviderRegistry priority.""" + + def test_bedrock(self): + assert ( + Settings(aws_region="us-east-1").get_embedding_provider_family() + == "bedrock" + ) + + def test_openai(self): + settings = Settings( + openai_api_key="sk-test", + aws_region=None, + bedrock_embedding_model=None, + bedrock_generation_model=None, + ) + assert settings.get_embedding_provider_family() == "openai" + + def test_mistral(self): + settings = Settings( + mistral_api_key="m-test", + aws_region=None, + bedrock_embedding_model=None, + bedrock_generation_model=None, + openai_api_key=None, + ) + assert settings.get_embedding_provider_family() == "mistral" + + def test_ollama(self): + settings = Settings( + ollama_base_url="http://localhost:11434", + aws_region=None, + bedrock_embedding_model=None, + bedrock_generation_model=None, + openai_api_key=None, + mistral_api_key=None, + ) + assert settings.get_embedding_provider_family() == "ollama" + + def test_simple_fallback(self): + settings = Settings( + aws_region=None, + bedrock_embedding_model=None, + bedrock_generation_model=None, + openai_api_key=None, + mistral_api_key=None, + ollama_base_url=None, + ) + assert settings.get_embedding_provider_family() == "simple" + + def test_gateway_uses_model_prefix(self): + settings = Settings( + embedding_provider="gateway", + embedding_gateway_url="http://gateway:8080", + embedding_gateway_model="mistral/mistral-embed", + ) + assert settings.get_embedding_provider_family() == "mistral" + + +class TestRecordEmbedding: + def test_dense_success_increments_throughput(self): + labels = {"kind": "dense", "provider": "uttest-prov"} + before_chunks = _sample("astrolabe_embedding_chunks_total", labels) + before_chars = _sample("astrolabe_embedding_chars_total", labels) + before_req = _sample( + "astrolabe_embedding_requests_total", {**labels, "status": "success"} + ) + + record_embedding("dense", "uttest-prov", 0.42, chunks=12, chars=3400) + + assert _sample("astrolabe_embedding_chunks_total", labels) == ( + before_chunks + 12 + ) + assert _sample("astrolabe_embedding_chars_total", labels) == ( + before_chars + 3400 + ) + assert _sample( + "astrolabe_embedding_requests_total", {**labels, "status": "success"} + ) == (before_req + 1) + assert ( + _sample( + "astrolabe_embedding_duration_seconds_count", + {**labels, "status": "success"}, + ) + >= 1 + ) + + def test_sparse_error_skips_throughput(self): + labels = {"kind": "sparse", "provider": "bm25-uttest"} + record_embedding( + "sparse", "bm25-uttest", 0.1, chunks=5, chars=100, status="error" + ) + assert _sample("astrolabe_embedding_chunks_total", labels) == 0.0 + assert _sample("astrolabe_embedding_chars_total", labels) == 0.0 + assert ( + _sample("astrolabe_embedding_requests_total", {**labels, "status": "error"}) + == 1.0 + )