fix(observability): address second review round
- Failed deletes no longer bump astrolabe_documents_indexed_total: the outer except in process_document now gates doc_type on operation != "delete", so a delete error is counted as processed-error but not as an indexing event. Added test_failed_delete_is_processed_but_not_indexed. - registry parse span: pass record_exception=True explicitly (matches instrument_tool) and add a structured logger.warning on the parse-error path (processor/tier/byte_size/duration_ms) for a Loki-aggregatable failed-parse signal. - test_error_does_not_increment_throughput: snapshot-before/delta pattern instead of absolute 0.0 (counters are global singletons). - config: document the deliberate gateway asymmetry between get_embedding_model_name() (no gateway branch) and get_embedding_provider_family() (short-circuits on gateway). - Cleanup in touched scope: narrow `except (HTTPStatusError, Exception)` to `except Exception` (drop now-unused import); convert registry signatures from Optional[...] to `... | None`. Refs Deck #175, PR #831. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
2e49e442f4
commit
68c9e20636
@@ -931,6 +931,12 @@ class Settings:
|
|||||||
Returns:
|
Returns:
|
||||||
Active embedding model name
|
Active embedding model name
|
||||||
"""
|
"""
|
||||||
|
# NOTE: there is intentionally no "gateway" branch here. When
|
||||||
|
# EMBEDDING_PROVIDER=gateway this falls through to the underlying
|
||||||
|
# provider's model (used for the Qdrant collection name), whereas
|
||||||
|
# get_embedding_provider_family() short-circuits to the gateway-routed
|
||||||
|
# family. Keep that asymmetry in mind before joining metrics/labels
|
||||||
|
# derived from these two methods.
|
||||||
if (
|
if (
|
||||||
self.aws_region
|
self.aws_region
|
||||||
or self.bedrock_embedding_model
|
or self.bedrock_embedding_model
|
||||||
|
|||||||
@@ -3,7 +3,7 @@
|
|||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from collections.abc import Awaitable, Callable
|
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.metrics import record_document_parse
|
||||||
from nextcloud_mcp_server.observability.tracing import trace_operation
|
from nextcloud_mcp_server.observability.tracing import trace_operation
|
||||||
@@ -72,7 +72,7 @@ class ProcessorRegistry:
|
|||||||
len(processor.supported_mime_types),
|
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.
|
"""Get a processor by name.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
@@ -85,7 +85,7 @@ class ProcessorRegistry:
|
|||||||
return self._processors[name][0]
|
return self._processors[name][0]
|
||||||
return None
|
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.
|
"""Find the first processor that supports the given MIME type.
|
||||||
|
|
||||||
Processors are checked in priority order (highest priority first).
|
Processors are checked in priority order (highest priority first).
|
||||||
@@ -117,12 +117,12 @@ class ProcessorRegistry:
|
|||||||
self,
|
self,
|
||||||
content: bytes,
|
content: bytes,
|
||||||
content_type: str,
|
content_type: str,
|
||||||
filename: Optional[str] = None,
|
filename: str | None = None,
|
||||||
processor_name: Optional[str] = None,
|
processor_name: str | None = None,
|
||||||
options: Optional[dict[str, Any]] = None,
|
options: dict[str, Any] | None = None,
|
||||||
progress_callback: Optional[
|
progress_callback: (
|
||||||
Callable[[float, Optional[float], Optional[str]], Awaitable[None]]
|
Callable[[float, float | None, str | None], Awaitable[None]] | None
|
||||||
] = None,
|
) = None,
|
||||||
) -> ProcessingResult:
|
) -> ProcessingResult:
|
||||||
"""Process a document using available processors.
|
"""Process a document using available processors.
|
||||||
|
|
||||||
@@ -183,6 +183,7 @@ class ProcessorRegistry:
|
|||||||
"byte_size": byte_size,
|
"byte_size": byte_size,
|
||||||
"escalated": False,
|
"escalated": False,
|
||||||
},
|
},
|
||||||
|
record_exception=True,
|
||||||
) as span:
|
) as span:
|
||||||
try:
|
try:
|
||||||
result = await processor.process(
|
result = await processor.process(
|
||||||
@@ -197,6 +198,22 @@ class ProcessorRegistry:
|
|||||||
byte_size=byte_size,
|
byte_size=byte_size,
|
||||||
status="error",
|
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 "<bytes>",
|
||||||
|
processor.name,
|
||||||
|
duration,
|
||||||
|
extra={
|
||||||
|
"processor": processor.name,
|
||||||
|
"tier": tier,
|
||||||
|
"byte_size": byte_size,
|
||||||
|
"duration_ms": round(duration * 1000, 1),
|
||||||
|
"status": "error",
|
||||||
|
},
|
||||||
|
)
|
||||||
raise
|
raise
|
||||||
|
|
||||||
duration = time.time() - start_time
|
duration = time.time() - start_time
|
||||||
|
|||||||
@@ -11,7 +11,6 @@ from typing import Any, cast
|
|||||||
import anyio
|
import anyio
|
||||||
from anyio.abc import TaskStatus
|
from anyio.abc import TaskStatus
|
||||||
from anyio.streams.memory import MemoryObjectReceiveStream
|
from anyio.streams.memory import MemoryObjectReceiveStream
|
||||||
from httpx import HTTPStatusError
|
|
||||||
from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct
|
from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct
|
||||||
|
|
||||||
from nextcloud_mcp_server.acl_hash import compute_acl_hash
|
from nextcloud_mcp_server.acl_hash import compute_acl_hash
|
||||||
@@ -246,7 +245,7 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient):
|
|||||||
)
|
)
|
||||||
return # Success
|
return # Success
|
||||||
|
|
||||||
except (HTTPStatusError, Exception) as e:
|
except Exception as e:
|
||||||
if attempt < max_retries - 1:
|
if attempt < max_retries - 1:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"Retry %s/%s for %s_%s: %s",
|
"Retry %s/%s for %s_%s: %s",
|
||||||
@@ -289,9 +288,14 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient):
|
|||||||
except Exception:
|
except Exception:
|
||||||
# Single processing-error call site: catches exhausted-retry
|
# Single processing-error call site: catches exhausted-retry
|
||||||
# re-raises, delete failures, and setup errors (get_qdrant_client /
|
# re-raises, delete failures, and setup errors (get_qdrant_client /
|
||||||
# get_settings) — each counted exactly once.
|
# get_settings) — each counted exactly once. A failed delete is not
|
||||||
|
# an indexing event either, so doc_type is omitted for deletes to
|
||||||
|
# keep them out of astrolabe_documents_indexed_total.
|
||||||
duration = time.time() - start_time
|
duration = time.time() - start_time
|
||||||
record_vector_sync_processing(duration, "error", doc_type=doc_task.doc_type)
|
indexed_doc_type = (
|
||||||
|
None if doc_task.operation == "delete" else doc_task.doc_type
|
||||||
|
)
|
||||||
|
record_vector_sync_processing(duration, "error", doc_type=indexed_doc_type)
|
||||||
raise
|
raise
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -192,6 +192,14 @@ class TestParseMetricHelpers:
|
|||||||
|
|
||||||
def test_error_does_not_increment_throughput(self, metric_sample):
|
def test_error_does_not_increment_throughput(self, metric_sample):
|
||||||
labels = {"processor": "uttest-error", "tier": "fast"}
|
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(
|
record_document_parse(
|
||||||
"uttest-error",
|
"uttest-error",
|
||||||
"fast",
|
"fast",
|
||||||
@@ -201,16 +209,17 @@ class TestParseMetricHelpers:
|
|||||||
byte_size=10,
|
byte_size=10,
|
||||||
status="error",
|
status="error",
|
||||||
)
|
)
|
||||||
|
|
||||||
# Error parses count the attempt + duration, but NOT pages/chars/bytes.
|
# Error parses count the attempt + duration, but NOT pages/chars/bytes.
|
||||||
assert metric_sample(
|
assert metric_sample(
|
||||||
"astrolabe_document_pages_processed_total", labels
|
"astrolabe_document_pages_processed_total", labels
|
||||||
) == pytest.approx(0.0)
|
) == pytest.approx(before_pages)
|
||||||
assert metric_sample(
|
assert metric_sample(
|
||||||
"astrolabe_document_chars_processed_total", labels
|
"astrolabe_document_chars_processed_total", labels
|
||||||
) == pytest.approx(0.0)
|
) == pytest.approx(before_chars)
|
||||||
assert metric_sample(
|
assert metric_sample(
|
||||||
"astrolabe_document_parse_total", {**labels, "status": "error"}
|
"astrolabe_document_parse_total", {**labels, "status": "error"}
|
||||||
) == pytest.approx(1.0)
|
) == pytest.approx(before_total + 1)
|
||||||
|
|
||||||
def test_record_document_chunks(self, metric_sample):
|
def test_record_document_chunks(self, metric_sample):
|
||||||
labels = {"doc_type": "uttest-chunks"}
|
labels = {"doc_type": "uttest-chunks"}
|
||||||
@@ -313,3 +322,31 @@ class TestProcessDocumentMetricCounting:
|
|||||||
assert metric_sample(
|
assert metric_sample(
|
||||||
"mcp_vector_sync_documents_processed_total", processed_labels
|
"mcp_vector_sync_documents_processed_total", processed_labels
|
||||||
) == pytest.approx(before_processed + 1)
|
) == 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)
|
||||||
|
|||||||
Reference in New Issue
Block a user