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)