diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index c5e95350..a33220ee 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -927,37 +927,79 @@ 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 - Returns: - Active embedding model name + 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 self.bedrock_embedding_model or "bedrock-default" + return "bedrock", self.bedrock_embedding_model or "bedrock-default" if self.openai_api_key: - return self.openai_embedding_model + return "openai", self.openai_embedding_model if self.mistral_api_key: - return self.mistral_embedding_model + return "mistral", self.mistral_embedding_model if self.ollama_base_url: - return self.ollama_embedding_model + return "ollama", self.ollama_embedding_model - return f"simple-{self.simple_embedding_dimension}" + 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 + """ + # 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. + return self._detect_base_provider()[1] + + 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. + + 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 + | ollama | simple + """ + if self.embedding_provider == "gateway": + model = self.embedding_gateway_model or "" + return model.split("/", 1)[0] if "/" in model else "gateway" + + 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 f812a264..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""" @@ -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]: @@ -70,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/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..545513de 100644 --- a/nextcloud_mcp_server/document_processors/registry.py +++ b/nextcloud_mcp_server/document_processors/registry.py @@ -1,8 +1,12 @@ """Central registry for document processors.""" 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 from .base import DocumentProcessor, ProcessingResult, ProcessorError @@ -68,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: @@ -81,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). @@ -113,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. @@ -152,13 +156,103 @@ 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, + }, + record_exception=True, + ) 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", + ) + # 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 + 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..bcf7e373 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,106 @@ 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() + # 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( + pages + ) + if chars > 0: + document_chars_processed_total.labels(processor=processor, tier=tier).inc( + chars + ) + if byte_size > 0: + 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 > 0: + embedding_chunks_total.labels(kind=kind, provider=provider).inc(chunks) + if chars > 0: + 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..cfd17567 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 @@ -20,6 +19,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, @@ -35,6 +36,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. @@ -209,9 +214,16 @@ 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 + # 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") @@ -228,10 +240,12 @@ 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: + except Exception as e: if attempt < max_retries - 1: logger.warning( "Retry %s/%s for %s_%s: %s", @@ -240,6 +254,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 +271,31 @@ 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 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") 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. 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") + 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 @@ -512,12 +547,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(_ATTR_CHUNK_COUNT, len(chunks)) # Assign page numbers to chunks if page boundaries are available (PDFs) page_boundaries = file_metadata.get("page_boundaries") @@ -527,7 +565,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), }, ): @@ -583,27 +621,64 @@ 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), + _ATTR_CHUNK_COUNT: len(chunk_texts), + "vector_sync.total_chars": total_chars, + "embedding.kind": "dense", + "embedding.provider": provider, + "embedding.model": settings.get_embedding_model_name(), }, ): 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), + _ATTR_CHUNK_COUNT: len(chunk_texts), + "vector_sync.total_chars": total_chars, + "embedding.kind": "sparse", + "embedding.provider": "bm25", }, ): 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).""" @@ -617,7 +692,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), }, ): @@ -662,7 +737,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: @@ -680,7 +755,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 @@ -853,4 +928,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/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 new file mode 100644 index 00000000..be0cbf61 --- /dev/null +++ b/tests/unit/test_document_parse_metrics.py @@ -0,0 +1,352 @@ +"""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 AsyncMock, MagicMock, patch + +import pytest + +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, +) +from nextcloud_mcp_server.vector import processor as proc +from nextcloud_mcp_server.vector.scanner import DocumentTask + +pytestmark = pytest.mark.unit + +# ``metric_sample`` is provided as a shared fixture in tests/unit/conftest.py. + + +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, metric_sample): + labels = {"processor": "uttest-success", "tier": "fast"} + 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"} + ) + + record_document_parse( + "uttest-success", + "fast", + 1.23, + pages=50, + chars=1000, + byte_size=99, + status="success", + ) + + 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"} + ) == pytest.approx(before_total + 1) + # The duration histogram observed one sample. + assert ( + metric_sample( + "astrolabe_document_parse_duration_seconds_count", + {**labels, "status": "success"}, + ) + >= 1 + ) + + 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", + 0.5, + pages=10, + chars=10, + 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(before_pages) + assert metric_sample( + "astrolabe_document_chars_processed_total", labels + ) == pytest.approx(before_chars) + assert metric_sample( + "astrolabe_document_parse_total", {**labels, "status": "error"} + ) == pytest.approx(before_total + 1) + + def test_record_document_chunks(self, metric_sample): + labels = {"doc_type": "uttest-chunks"} + before = metric_sample("astrolabe_document_chunks_total", labels) + record_document_chunks("uttest-chunks", 7) + assert metric_sample( + "astrolabe_document_chunks_total", labels + ) == pytest.approx(before + 7) + + def test_vector_sync_processing_increments_documents_indexed(self, metric_sample): + labels = {"source": "uttest-doctype", "status": "success"} + before = metric_sample("astrolabe_documents_indexed_total", labels) + record_vector_sync_processing(0.1, "success", doc_type="uttest-doctype") + 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, 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 metric_sample( + "astrolabe_documents_indexed_total", labels + ) == pytest.approx(0.0) + + 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 = metric_sample("astrolabe_document_escalation_total", labels) + record_document_escalation("fast", "ocr", "empty_text") + 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) + + 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) diff --git a/tests/unit/test_embedding_metrics.py b/tests/unit/test_embedding_metrics.py new file mode 100644 index 00000000..1f165480 --- /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 nextcloud_mcp_server.config import Settings +from nextcloud_mcp_server.observability.metrics import record_embedding + +pytestmark = pytest.mark.unit + +# ``metric_sample`` is provided as a shared fixture in tests/unit/conftest.py. + + +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="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, metric_sample): + labels = {"kind": "dense", "provider": "uttest-prov"} + 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 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"} + ) == pytest.approx(before_req + 1) + assert ( + metric_sample( + "astrolabe_embedding_duration_seconds_count", + {**labels, "status": "success"}, + ) + >= 1 + ) + + 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 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)