From 5d205fcaab6f7018e68ac9ac6718897223f36f86 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Tue, 2 Jun 2026 17:46:20 +0200 Subject: [PATCH 1/4] 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 + ) From 2e49e442f4ce94cdc28774a2eaadbb415fcea589 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Tue, 2 Jun 2026 18:15:50 +0200 Subject: [PATCH 2/4] fix(observability): address PR review + SonarCloud findings Reviewer findings: - Fix double-count of exhausted-retry failures: the inner final-retry branch and the outer except both recorded a processing error. Consolidate to the outer handler (single call site); inner branch keeps only the Qdrant-upsert error metric. Regression test added. - Deletes are no longer counted as indexing events: the delete success path drops doc_type so astrolabe_documents_indexed_total is not inflated. Regression test added. - Reuse the already-resolved `settings` in _index_document instead of a second get_settings() call. - Use explicit `> 0` guards in record_document_parse / record_embedding instead of truthiness checks. SonarCloud: - S1244 (BUG): replace float `==` equality in metric tests with pytest.approx. - S5332 (hotspot): use https in the gateway-URL test fixture. - S1192: extract the repeated "vector_sync.chunk_count" span-attribute literal into a module constant. Review nit: move the duplicated `_sample` test helper into a shared `metric_sample` fixture in tests/unit/conftest.py. Refs Deck #175, PR #831. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/observability/metrics.py | 10 +- nextcloud_mcp_server/vector/processor.py | 38 +++-- tests/unit/conftest.py | 15 ++ tests/unit/test_document_parse_metrics.py | 158 +++++++++++++----- tests/unit/test_embedding_metrics.py | 50 +++--- 5 files changed, 183 insertions(+), 88 deletions(-) diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 4f4c9fcb..5cc00e6c 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -534,15 +534,15 @@ def record_document_parse( ).observe(duration) document_parse_total.labels(processor=processor, tier=tier, status=status).inc() if status == "success": - if pages: + if pages > 0: document_pages_processed_total.labels(processor=processor, tier=tier).inc( pages ) - if chars: + if chars > 0: document_chars_processed_total.labels(processor=processor, tier=tier).inc( chars ) - if byte_size: + if byte_size > 0: document_bytes_processed_total.labels(processor=processor, tier=tier).inc( byte_size ) @@ -587,9 +587,9 @@ def record_embedding( ).observe(duration) embedding_requests_total.labels(kind=kind, provider=provider, status=status).inc() if status == "success": - if chunks: + if chunks > 0: embedding_chunks_total.labels(kind=kind, provider=provider).inc(chunks) - if chars: + if chars > 0: embedding_chars_total.labels(kind=kind, provider=provider).inc(chars) diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 94837071..dd207af0 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -37,6 +37,10 @@ from nextcloud_mcp_server.vector.scanner import DocumentTask logger = logging.getLogger(__name__) +# Shared span-attribute key (avoids duplicating the string literal across the +# many vector_sync spans that report a chunk count). +_ATTR_CHUNK_COUNT = "vector_sync.chunk_count" + def assign_page_numbers(chunks, page_boundaries): """Assign page numbers to chunks based on page boundaries. @@ -218,12 +222,12 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): }, ) - # Record successful deletion metrics + # Record successful deletion metrics. A delete is not an + # indexing event, so doc_type is intentionally omitted here to + # keep it out of astrolabe_documents_indexed_total. duration = time.time() - start_time record_qdrant_operation("delete", "success") - record_vector_sync_processing( - duration, "success", doc_type=doc_task.doc_type - ) + record_vector_sync_processing(duration, "success") return # Handle indexing with retry @@ -276,16 +280,16 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): "status": "error", }, ) - # Record failed processing metrics - duration = time.time() - start_time + # Record the failed Qdrant upsert. The processing-error + # metric is recorded once by the outer handler below, so + # exhausted-retry failures aren't double-counted. record_qdrant_operation("upsert", "error") - record_vector_sync_processing( - duration, "error", doc_type=doc_task.doc_type - ) raise except Exception: - # Catch any other unexpected errors + # Single processing-error call site: catches exhausted-retry + # re-raises, delete failures, and setup errors (get_qdrant_client / + # get_settings) — each counted exactly once. duration = time.time() - start_time record_vector_sync_processing(duration, "error", doc_type=doc_task.doc_type) raise @@ -547,7 +551,7 @@ async def _index_document( 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)) + chunk_span.set_attribute(_ATTR_CHUNK_COUNT, len(chunks)) # Assign page numbers to chunks if page boundaries are available (PDFs) page_boundaries = file_metadata.get("page_boundaries") @@ -557,7 +561,7 @@ async def _index_document( with trace_operation( "vector_sync.assign_page_numbers", attributes={ - "vector_sync.chunk_count": len(chunks), + _ATTR_CHUNK_COUNT: len(chunks), "vector_sync.page_count": len(page_boundaries_list), }, ): @@ -618,7 +622,7 @@ async def _index_document( with trace_operation( "vector_sync.embed_dense", attributes={ - "vector_sync.chunk_count": len(chunk_texts), + _ATTR_CHUNK_COUNT: len(chunk_texts), "vector_sync.total_chars": total_chars, "embedding.kind": "dense", "embedding.provider": provider, @@ -650,7 +654,7 @@ async def _index_document( with trace_operation( "vector_sync.embed_sparse", attributes={ - "vector_sync.chunk_count": len(chunk_texts), + _ATTR_CHUNK_COUNT: len(chunk_texts), "embedding.kind": "sparse", "embedding.provider": "bm25", "embedding.batch_size": len(chunk_texts), @@ -685,7 +689,7 @@ async def _index_document( with trace_operation( "vector_sync.compute_chunk_bboxes", attributes={ - "vector_sync.chunk_count": len(chunks), + _ATTR_CHUNK_COUNT: len(chunks), "vector_sync.pdf_size": len(content_bytes), }, ): @@ -730,7 +734,7 @@ async def _index_document( "vector_sync.parallel_processing", attributes={ "vector_sync.is_pdf": is_pdf, - "vector_sync.chunk_count": len(chunks), + _ATTR_CHUNK_COUNT: len(chunks), }, ): async with anyio.create_task_group() as tg: @@ -748,7 +752,7 @@ async def _index_document( # PIPELINE_TIER is "fast"; ACL hash records at least the owner principal # (full share enumeration is a follow-up — a missing/partial acl_hash is # safe because the query-side pre-filter only applies when present + enabled). - _embedding_identity = get_settings().get_embedding_model_name() + _embedding_identity = settings.get_embedding_model_name() _acl_hash = compute_acl_hash([("user", doc_task.user_id)]) # Surface deck card data quality issues at indexing time rather than diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 51ba62a8..67fd5531 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -8,6 +8,21 @@ import pytest from tests.fixtures.storage_backend import storage_backend # noqa: F401 +@pytest.fixture +def metric_sample(): + """Return a callable reading a Prometheus sample value (0.0 if absent). + + Shared across the metric unit tests so the helper isn't duplicated per + module. + """ + from prometheus_client import REGISTRY + + def _sample(name: str, labels: dict[str, str]) -> float: + return REGISTRY.get_sample_value(name, labels) or 0.0 + + return _sample + + @pytest.fixture(autouse=True) def _reload_dynaconf_after_test(): """Ensure dynaconf cache is clean between tests. diff --git a/tests/unit/test_document_parse_metrics.py b/tests/unit/test_document_parse_metrics.py index e27167ee..d1d82c96 100644 --- a/tests/unit/test_document_parse_metrics.py +++ b/tests/unit/test_document_parse_metrics.py @@ -13,10 +13,9 @@ Covers two layers: from __future__ import annotations from typing import Any -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest -from prometheus_client import REGISTRY from nextcloud_mcp_server.document_processors.base import ( DocumentProcessor, @@ -30,13 +29,12 @@ from nextcloud_mcp_server.observability.metrics import ( record_document_parse, record_vector_sync_processing, ) +from nextcloud_mcp_server.vector import processor as proc +from nextcloud_mcp_server.vector.scanner import DocumentTask 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 +# ``metric_sample`` is provided as a shared fixture in tests/unit/conftest.py. class _FakeProcessor(DocumentProcessor): @@ -152,12 +150,12 @@ class TestRegistryParseInstrumentation: class TestParseMetricHelpers: - def test_success_increments_throughput_counters(self): + def test_success_increments_throughput_counters(self, metric_sample): 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( + before_pages = metric_sample("astrolabe_document_pages_processed_total", labels) + before_chars = metric_sample("astrolabe_document_chars_processed_total", labels) + before_bytes = metric_sample("astrolabe_document_bytes_processed_total", labels) + before_total = metric_sample( "astrolabe_document_parse_total", {**labels, "status": "success"} ) @@ -171,28 +169,28 @@ class TestParseMetricHelpers: 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( + assert metric_sample( + "astrolabe_document_pages_processed_total", labels + ) == pytest.approx(before_pages + 50) + assert metric_sample( + "astrolabe_document_chars_processed_total", labels + ) == pytest.approx(before_chars + 1000) + assert metric_sample( + "astrolabe_document_bytes_processed_total", labels + ) == pytest.approx(before_bytes + 99) + assert metric_sample( "astrolabe_document_parse_total", {**labels, "status": "success"} - ) == (before_total + 1) + ) == pytest.approx(before_total + 1) # The duration histogram observed one sample. assert ( - _sample( + metric_sample( "astrolabe_document_parse_duration_seconds_count", {**labels, "status": "success"}, ) >= 1 ) - def test_error_does_not_increment_throughput(self): + def test_error_does_not_increment_throughput(self, metric_sample): labels = {"processor": "uttest-error", "tier": "fast"} record_document_parse( "uttest-error", @@ -204,36 +202,114 @@ class TestParseMetricHelpers: 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 - ) + assert metric_sample( + "astrolabe_document_pages_processed_total", labels + ) == pytest.approx(0.0) + assert metric_sample( + "astrolabe_document_chars_processed_total", labels + ) == pytest.approx(0.0) + assert metric_sample( + "astrolabe_document_parse_total", {**labels, "status": "error"} + ) == pytest.approx(1.0) - def test_record_document_chunks(self): + def test_record_document_chunks(self, metric_sample): labels = {"doc_type": "uttest-chunks"} - before = _sample("astrolabe_document_chunks_total", labels) + before = metric_sample("astrolabe_document_chunks_total", labels) record_document_chunks("uttest-chunks", 7) - assert _sample("astrolabe_document_chunks_total", labels) == before + 7 + assert metric_sample( + "astrolabe_document_chunks_total", labels + ) == pytest.approx(before + 7) - def test_vector_sync_processing_increments_documents_indexed(self): + def test_vector_sync_processing_increments_documents_indexed(self, metric_sample): labels = {"source": "uttest-doctype", "status": "success"} - before = _sample("astrolabe_documents_indexed_total", labels) + before = metric_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 + assert metric_sample( + "astrolabe_documents_indexed_total", labels + ) == pytest.approx(before + 1) - def test_vector_sync_processing_without_doc_type_is_noop_for_indexed(self): + def test_vector_sync_processing_without_doc_type_is_noop_for_indexed( + self, metric_sample + ): # 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 + assert metric_sample( + "astrolabe_documents_indexed_total", labels + ) == pytest.approx(0.0) - def test_record_document_escalation(self): + def test_record_document_escalation(self, metric_sample): # 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) + before = metric_sample("astrolabe_document_escalation_total", labels) record_document_escalation("fast", "ocr", "empty_text") - assert _sample("astrolabe_document_escalation_total", labels) == before + 1 + assert metric_sample( + "astrolabe_document_escalation_total", labels + ) == pytest.approx(before + 1) + + +class TestProcessDocumentMetricCounting: + """Regression tests for the error/delete counting fixes from PR #831 review.""" + + async def test_exhausted_retries_count_error_once(self, metric_sample): + # The inner final-retry branch and the outer except both used to record + # a processing error, double-counting exhausted-retry failures. + task = DocumentTask( + user_id="u", doc_id="1", doc_type="note", operation="index", modified_at=0 + ) + err_labels = {"status": "error"} + indexed_labels = {"source": "note", "status": "error"} + before_processed = metric_sample( + "mcp_vector_sync_documents_processed_total", err_labels + ) + before_indexed = metric_sample( + "astrolabe_documents_indexed_total", indexed_labels + ) + + with ( + patch.object( + proc, "get_qdrant_client", new=AsyncMock(return_value=MagicMock()) + ), + patch.object( + proc, "_index_document", new=AsyncMock(side_effect=RuntimeError("boom")) + ), + patch.object(proc.anyio, "sleep", new=AsyncMock()), # skip backoff + ): + with pytest.raises(RuntimeError): + await proc.process_document(task, MagicMock()) + + assert metric_sample( + "mcp_vector_sync_documents_processed_total", err_labels + ) == pytest.approx(before_processed + 1) + assert metric_sample( + "astrolabe_documents_indexed_total", indexed_labels + ) == pytest.approx(before_indexed + 1) + + async def test_delete_is_processed_but_not_indexed(self, metric_sample): + # A delete is processed but is NOT an indexing event, so it must not + # touch astrolabe_documents_indexed_total. + task = DocumentTask( + user_id="u", doc_id="2", doc_type="note", operation="delete", modified_at=0 + ) + indexed_labels = {"source": "note", "status": "success"} + processed_labels = {"status": "success"} + before_indexed = metric_sample( + "astrolabe_documents_indexed_total", indexed_labels + ) + before_processed = metric_sample( + "mcp_vector_sync_documents_processed_total", processed_labels + ) + + qmock = MagicMock() + qmock.delete = AsyncMock() + with patch.object(proc, "get_qdrant_client", new=AsyncMock(return_value=qmock)): + await proc.process_document(task, MagicMock()) + + assert metric_sample( + "astrolabe_documents_indexed_total", indexed_labels + ) == pytest.approx(before_indexed) + assert metric_sample( + "mcp_vector_sync_documents_processed_total", processed_labels + ) == pytest.approx(before_processed + 1) diff --git a/tests/unit/test_embedding_metrics.py b/tests/unit/test_embedding_metrics.py index 3e6c4a15..1f165480 100644 --- a/tests/unit/test_embedding_metrics.py +++ b/tests/unit/test_embedding_metrics.py @@ -10,16 +10,13 @@ Covers: 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 +# ``metric_sample`` is provided as a shared fixture in tests/unit/conftest.py. class TestProviderFamily: @@ -75,48 +72,51 @@ class TestProviderFamily: def test_gateway_uses_model_prefix(self): settings = Settings( embedding_provider="gateway", - embedding_gateway_url="http://gateway:8080", + embedding_gateway_url="https://gateway:8080", embedding_gateway_model="mistral/mistral-embed", ) assert settings.get_embedding_provider_family() == "mistral" class TestRecordEmbedding: - def test_dense_success_increments_throughput(self): + def test_dense_success_increments_throughput(self, metric_sample): 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( + before_chunks = metric_sample("astrolabe_embedding_chunks_total", labels) + before_chars = metric_sample("astrolabe_embedding_chars_total", labels) + before_req = metric_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( + assert metric_sample( + "astrolabe_embedding_chunks_total", labels + ) == pytest.approx(before_chunks + 12) + assert metric_sample( + "astrolabe_embedding_chars_total", labels + ) == pytest.approx(before_chars + 3400) + assert metric_sample( "astrolabe_embedding_requests_total", {**labels, "status": "success"} - ) == (before_req + 1) + ) == pytest.approx(before_req + 1) assert ( - _sample( + metric_sample( "astrolabe_embedding_duration_seconds_count", {**labels, "status": "success"}, ) >= 1 ) - def test_sparse_error_skips_throughput(self): + def test_sparse_error_skips_throughput(self, metric_sample): 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 - ) + assert metric_sample( + "astrolabe_embedding_chunks_total", labels + ) == pytest.approx(0.0) + assert metric_sample( + "astrolabe_embedding_chars_total", labels + ) == pytest.approx(0.0) + assert metric_sample( + "astrolabe_embedding_requests_total", {**labels, "status": "error"} + ) == pytest.approx(1.0) From 68c9e206361ef0caa8ed50ecfab859ebc3db5f9f Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Tue, 2 Jun 2026 21:09:50 +0200 Subject: [PATCH 3/4] fix(observability): address second review round - Failed deletes no longer bump astrolabe_documents_indexed_total: the outer except in process_document now gates doc_type on operation != "delete", so a delete error is counted as processed-error but not as an indexing event. Added test_failed_delete_is_processed_but_not_indexed. - registry parse span: pass record_exception=True explicitly (matches instrument_tool) and add a structured logger.warning on the parse-error path (processor/tier/byte_size/duration_ms) for a Loki-aggregatable failed-parse signal. - test_error_does_not_increment_throughput: snapshot-before/delta pattern instead of absolute 0.0 (counters are global singletons). - config: document the deliberate gateway asymmetry between get_embedding_model_name() (no gateway branch) and get_embedding_provider_family() (short-circuits on gateway). - Cleanup in touched scope: narrow `except (HTTPStatusError, Exception)` to `except Exception` (drop now-unused import); convert registry signatures from Optional[...] to `... | None`. Refs Deck #175, PR #831. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/config.py | 6 +++ .../document_processors/registry.py | 35 +++++++++++---- nextcloud_mcp_server/vector/processor.py | 12 ++++-- tests/unit/test_document_parse_metrics.py | 43 +++++++++++++++++-- 4 files changed, 80 insertions(+), 16 deletions(-) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index eebf3a98..6da74196 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -931,6 +931,12 @@ class Settings: Returns: Active embedding model name """ + # NOTE: there is intentionally no "gateway" branch here. When + # EMBEDDING_PROVIDER=gateway this falls through to the underlying + # provider's model (used for the Qdrant collection name), whereas + # get_embedding_provider_family() short-circuits to the gateway-routed + # family. Keep that asymmetry in mind before joining metrics/labels + # derived from these two methods. if ( self.aws_region or self.bedrock_embedding_model diff --git a/nextcloud_mcp_server/document_processors/registry.py b/nextcloud_mcp_server/document_processors/registry.py index 8f434afa..545513de 100644 --- a/nextcloud_mcp_server/document_processors/registry.py +++ b/nextcloud_mcp_server/document_processors/registry.py @@ -3,7 +3,7 @@ import logging import time from collections.abc import Awaitable, Callable -from typing import Any, Optional +from typing import Any from nextcloud_mcp_server.observability.metrics import record_document_parse from nextcloud_mcp_server.observability.tracing import trace_operation @@ -72,7 +72,7 @@ class ProcessorRegistry: len(processor.supported_mime_types), ) - def get_processor(self, name: str) -> Optional[DocumentProcessor]: + def get_processor(self, name: str) -> DocumentProcessor | None: """Get a processor by name. Args: @@ -85,7 +85,7 @@ class ProcessorRegistry: return self._processors[name][0] return None - def find_processor(self, content_type: str) -> Optional[DocumentProcessor]: + def find_processor(self, content_type: str) -> DocumentProcessor | None: """Find the first processor that supports the given MIME type. Processors are checked in priority order (highest priority first). @@ -117,12 +117,12 @@ class ProcessorRegistry: self, content: bytes, content_type: str, - filename: Optional[str] = None, - processor_name: Optional[str] = None, - options: Optional[dict[str, Any]] = None, - progress_callback: Optional[ - Callable[[float, Optional[float], Optional[str]], Awaitable[None]] - ] = None, + filename: str | None = None, + processor_name: str | None = None, + options: dict[str, Any] | None = None, + progress_callback: ( + Callable[[float, float | None, str | None], Awaitable[None]] | None + ) = None, ) -> ProcessingResult: """Process a document using available processors. @@ -183,6 +183,7 @@ class ProcessorRegistry: "byte_size": byte_size, "escalated": False, }, + record_exception=True, ) as span: try: result = await processor.process( @@ -197,6 +198,22 @@ class ProcessorRegistry: byte_size=byte_size, status="error", ) + # Structured error signal for Loki (the processor logs the + # traceback; this adds the aggregatable fields). The span + # records the exception itself via record_exception=True. + logger.warning( + "Parse failed for %s with '%s' after %.2fs", + filename or "", + processor.name, + duration, + extra={ + "processor": processor.name, + "tier": tier, + "byte_size": byte_size, + "duration_ms": round(duration * 1000, 1), + "status": "error", + }, + ) raise duration = time.time() - start_time diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index dd207af0..658b785e 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -11,7 +11,6 @@ from typing import Any, cast import anyio from anyio.abc import TaskStatus from anyio.streams.memory import MemoryObjectReceiveStream -from httpx import HTTPStatusError from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct from nextcloud_mcp_server.acl_hash import compute_acl_hash @@ -246,7 +245,7 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): ) return # Success - except (HTTPStatusError, Exception) as e: + except Exception as e: if attempt < max_retries - 1: logger.warning( "Retry %s/%s for %s_%s: %s", @@ -289,9 +288,14 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): except Exception: # Single processing-error call site: catches exhausted-retry # re-raises, delete failures, and setup errors (get_qdrant_client / - # get_settings) — each counted exactly once. + # get_settings) — each counted exactly once. A failed delete is not + # an indexing event either, so doc_type is omitted for deletes to + # keep them out of astrolabe_documents_indexed_total. duration = time.time() - start_time - record_vector_sync_processing(duration, "error", doc_type=doc_task.doc_type) + indexed_doc_type = ( + None if doc_task.operation == "delete" else doc_task.doc_type + ) + record_vector_sync_processing(duration, "error", doc_type=indexed_doc_type) raise diff --git a/tests/unit/test_document_parse_metrics.py b/tests/unit/test_document_parse_metrics.py index d1d82c96..be0cbf61 100644 --- a/tests/unit/test_document_parse_metrics.py +++ b/tests/unit/test_document_parse_metrics.py @@ -192,6 +192,14 @@ class TestParseMetricHelpers: def test_error_does_not_increment_throughput(self, metric_sample): labels = {"processor": "uttest-error", "tier": "fast"} + # Snapshot before — counters are global singletons, so assert the delta + # rather than an absolute value (consistent with the success test). + before_pages = metric_sample("astrolabe_document_pages_processed_total", labels) + before_chars = metric_sample("astrolabe_document_chars_processed_total", labels) + before_total = metric_sample( + "astrolabe_document_parse_total", {**labels, "status": "error"} + ) + record_document_parse( "uttest-error", "fast", @@ -201,16 +209,17 @@ class TestParseMetricHelpers: byte_size=10, status="error", ) + # Error parses count the attempt + duration, but NOT pages/chars/bytes. assert metric_sample( "astrolabe_document_pages_processed_total", labels - ) == pytest.approx(0.0) + ) == pytest.approx(before_pages) assert metric_sample( "astrolabe_document_chars_processed_total", labels - ) == pytest.approx(0.0) + ) == pytest.approx(before_chars) assert metric_sample( "astrolabe_document_parse_total", {**labels, "status": "error"} - ) == pytest.approx(1.0) + ) == pytest.approx(before_total + 1) def test_record_document_chunks(self, metric_sample): labels = {"doc_type": "uttest-chunks"} @@ -313,3 +322,31 @@ class TestProcessDocumentMetricCounting: assert metric_sample( "mcp_vector_sync_documents_processed_total", processed_labels ) == pytest.approx(before_processed + 1) + + async def test_failed_delete_is_processed_but_not_indexed(self, metric_sample): + # A *failed* delete also must not touch astrolabe_documents_indexed_total + # (the outer except gates doc_type on operation != "delete"). + task = DocumentTask( + user_id="u", doc_id="3", doc_type="note", operation="delete", modified_at=0 + ) + indexed_labels = {"source": "note", "status": "error"} + processed_labels = {"status": "error"} + before_indexed = metric_sample( + "astrolabe_documents_indexed_total", indexed_labels + ) + before_processed = metric_sample( + "mcp_vector_sync_documents_processed_total", processed_labels + ) + + qmock = MagicMock() + qmock.delete = AsyncMock(side_effect=RuntimeError("boom")) + with patch.object(proc, "get_qdrant_client", new=AsyncMock(return_value=qmock)): + with pytest.raises(RuntimeError): + await proc.process_document(task, MagicMock()) + + assert metric_sample( + "astrolabe_documents_indexed_total", indexed_labels + ) == pytest.approx(before_indexed) + assert metric_sample( + "mcp_vector_sync_documents_processed_total", processed_labels + ) == pytest.approx(before_processed + 1) From b779627fa649b278d290d888286ab074177b338c Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 3 Jun 2026 02:01:28 +0200 Subject: [PATCH 4/4] fix(observability): address third review round MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remaining items from the PR #831 Claude review: - processor span symmetry: add "vector_sync.total_chars" to the sparse embedding span (already on the dense span) and drop the redundant "embedding.batch_size" attribute from both spans — it always equalled vector_sync.chunk_count and would mislead once batching is split. - metrics: document the deliberate "throughput counts only on full success" contract in record_document_parse (partial extractions flagged success=False are counted as a parse-error but never inflate pages/chars/bytes throughput). - config: extract _detect_base_provider() -> (family, model) as the single source of truth for the provider-detection priority chain, shared by get_embedding_model_name() and get_embedding_provider_family(). Preserves the intentional gateway asymmetry (only the family method short-circuits). - base.py: Optional[...] -> PEP 604 `... | None`; drop now-unused import. Behavior unchanged (get_embedding_* outputs covered by test_config.py). Refs Deck #175, PR #831. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/config.py | 84 +++++++++---------- .../document_processors/base.py | 13 ++- nextcloud_mcp_server/observability/metrics.py | 4 + nextcloud_mcp_server/vector/processor.py | 3 +- 4 files changed, 49 insertions(+), 55 deletions(-) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 6da74196..92445d06 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -917,16 +917,45 @@ class Settings: self.enable_multi_user_basic_auth = resolved_mode == "multi_user_basic" self.enable_login_flow = resolved_mode == "login_flow" - def get_embedding_model_name(self) -> str: + def _detect_base_provider(self) -> tuple[str, str]: """ - Get the active embedding model name based on provider priority. + Resolve the ``(family, model)`` for the underlying embedding provider. - Priority order (same as ProviderRegistry): + Single source of truth for the provider-detection priority chain shared + by ``get_embedding_model_name`` and ``get_embedding_provider_family``: 1. Bedrock - if AWS_REGION or BEDROCK_EMBEDDING_MODEL is set 2. OpenAI - if OPENAI_API_KEY is set 3. Mistral - if MISTRAL_API_KEY is set 4. Ollama - if OLLAMA_BASE_URL is set - 5. Simple - fallback (returns "simple-{dimension}") + 5. Simple - fallback + + Does NOT handle the gateway short-circuit — callers layer that on top + as needed (see the asymmetry note on ``get_embedding_model_name``). + """ + if ( + self.aws_region + or self.bedrock_embedding_model + or self.bedrock_generation_model + ): + return "bedrock", self.bedrock_embedding_model or "bedrock-default" + + if self.openai_api_key: + return "openai", self.openai_embedding_model + + if self.mistral_api_key: + return "mistral", self.mistral_embedding_model + + if self.ollama_base_url: + return "ollama", self.ollama_embedding_model + + return "simple", f"simple-{self.simple_embedding_dimension}" + + def get_embedding_model_name(self) -> str: + """ + Get the active embedding model name based on provider priority. + + Priority order (same as ProviderRegistry): bedrock → openai → mistral → + ollama → simple (returns "simple-{dimension}"). Returns: Active embedding model name @@ -937,23 +966,7 @@ class Settings: # get_embedding_provider_family() short-circuits to the gateway-routed # family. Keep that asymmetry in mind before joining metrics/labels # derived from these two methods. - if ( - self.aws_region - or self.bedrock_embedding_model - or self.bedrock_generation_model - ): - return self.bedrock_embedding_model or "bedrock-default" - - if self.openai_api_key: - return self.openai_embedding_model - - if self.mistral_api_key: - return self.mistral_embedding_model - - if self.ollama_base_url: - return self.ollama_embedding_model - - return f"simple-{self.simple_embedding_dimension}" + return self._detect_base_provider()[1] def get_embedding_provider_family(self) -> str: """ @@ -964,14 +977,9 @@ class Settings: *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 + Gateway short-circuits to the gateway-routed family (from the model + prefix, e.g. "mistral/mistral-embed" -> "mistral"); otherwise the family + comes from the shared ``_detect_base_provider`` priority chain. Returns: Provider family: gateway-routed family | bedrock | openai | mistral @@ -981,23 +989,7 @@ class Settings: 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" + return self._detect_base_provider()[0] def get_collection_name(self) -> str: """ diff --git a/nextcloud_mcp_server/document_processors/base.py b/nextcloud_mcp_server/document_processors/base.py index 86ab1850..cb36ded2 100644 --- a/nextcloud_mcp_server/document_processors/base.py +++ b/nextcloud_mcp_server/document_processors/base.py @@ -2,7 +2,7 @@ from abc import ABC, abstractmethod from collections.abc import Awaitable, Callable -from typing import Any, Optional +from typing import Any from pydantic import BaseModel @@ -22,7 +22,7 @@ class ProcessingResult(BaseModel): success: bool = True """Whether processing succeeded""" - error: Optional[str] = None + error: str | None = None """Error message if processing failed""" @@ -83,11 +83,10 @@ class DocumentProcessor(ABC): self, content: bytes, content_type: str, - filename: Optional[str] = None, - options: Optional[dict[str, Any]] = None, - progress_callback: Optional[ - Callable[[float, Optional[float], Optional[str]], Awaitable[None]] - ] = None, + filename: str | None = None, + options: dict[str, Any] | None = None, + progress_callback: Callable[[float, float | None, str | None], Awaitable[None]] + | None = None, ) -> ProcessingResult: """Process a document and extract text. diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 5cc00e6c..bcf7e373 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -533,6 +533,10 @@ def record_document_parse( processor=processor, tier=tier, status=status ).observe(duration) document_parse_total.labels(processor=processor, tier=tier, status=status).inc() + # Throughput counters (pages/chars/bytes) accrue only on a full success. + # A partial extraction flagged success=False is recorded above as a + # parse-error but is intentionally excluded here so low-confidence output + # never inflates pipeline throughput. if status == "success": if pages > 0: document_pages_processed_total.labels(processor=processor, tier=tier).inc( diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 658b785e..cfd17567 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -631,7 +631,6 @@ async def _index_document( "embedding.kind": "dense", "embedding.provider": provider, "embedding.model": settings.get_embedding_model_name(), - "embedding.batch_size": len(chunk_texts), }, ): embedding_service = get_embedding_service() @@ -659,9 +658,9 @@ async def _index_document( "vector_sync.embed_sparse", attributes={ _ATTR_CHUNK_COUNT: len(chunk_texts), + "vector_sync.total_chars": total_chars, "embedding.kind": "sparse", "embedding.provider": "bm25", - "embedding.batch_size": len(chunk_texts), }, ): bm25_service = await get_bm25_service()