admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(observability): astrolabe_* metrics + traces for the document pipeline

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-02 17:46:20 +02:00
co-authored by Claude Opus 4.8
parent 7e4b83dc94
commit 5d205fcaab
8 changed files with 799 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/config.py
+44
View File
@@ -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.
nextcloud_mcp_server/document_processors/base.py
+13
View File
@@ -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]:
nextcloud_mcp_server/document_processors/pymupdf.py
+8
View File
@@ -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(
nextcloud_mcp_server/document_processors/registry.py
+82 -5
View File
@@ -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 "<bytes>",
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()
nextcloud_mcp_server/observability/metrics.py
+209 -1
View File
@@ -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
# =============================================================================
nextcloud_mcp_server/vector/processor.py
+82 -8
View File
@@ -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",
},
)
tests/unit/test_document_parse_metrics.py
+239
View File
@@ -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
tests/unit/test_embedding_metrics.py
+122
View File
@@ -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
)