feat(metrics): current-in-Qdrant chunk-density snapshot histogram

nextcloud_mcp_server/app.py

@@ -128,7 +128,10 @@
 )
 from nextcloud_mcp_server.server.auth_tools import register_auth_tools
 from nextcloud_mcp_server.server.oauth_tools import register_oauth_tools
-from nextcloud_mcp_server.vector.metrics_publisher import vector_sync_metrics_task
+from nextcloud_mcp_server.vector.metrics_publisher import (
+    vector_density_snapshot_task,
+    vector_sync_metrics_task,
+)
 from nextcloud_mcp_server.vector.oauth_sync import (
     ProvisionSignal,
     credential_cleanup_task,
@@ -2129,6 +2132,12 @@ async def spawn_worker(
                     shutdown_event,
                 )
 
+                # Current-corpus chunk-density snapshot on its own slower cadence
+                # (heavier collection scroll). Opt-out via
+                # VECTOR_DENSITY_SNAPSHOT_ENABLED.
+                if settings.vector_density_snapshot_enabled:
+                    await tg.start(vector_density_snapshot_task, shutdown_event)
+
                 logger.info(
                     "Background sync tasks started: 1 scanner + %s processors (queue=%s)",
                     ingest_transport.active_consumer_count,
@@ -2360,6 +2369,12 @@ async def spawn_worker(
                         shutdown_event,
                     )
 
+                    # Current-corpus chunk-density snapshot on its own slower
+                    # cadence (heavier collection scroll). Opt-out via
+                    # VECTOR_DENSITY_SNAPSHOT_ENABLED.
+                    if settings.vector_density_snapshot_enabled:
+                        await tg.start(vector_density_snapshot_task, shutdown_event)
+
                     logger.info(
                         "Background sync tasks started: 1 user manager + %s processors (queue=%s)",
                         ingest_transport.active_consumer_count,

nextcloud_mcp_server/config.py

@@ -114,6 +114,9 @@
     "vector_sync_processor_workers": 3,
     "vector_sync_queue_max_size": 10000,
     "vector_sync_metrics_refresh_interval": 20,
+    "vector_density_snapshot_enabled": True,
+    "vector_density_snapshot_interval": 300,
+    "vector_density_snapshot_max_documents": 50000,
     "vector_ram_hnsw_overhead_factor": 1.5,
     "vector_sync_user_poll_interval": 60,
     "health_ready_refresh_interval": 15,
@@ -450,6 +453,8 @@ def _resolve_settings_files() -> list[str]:
         Validator("VECTOR_SYNC_PROCESSOR_WORKERS", gte=1),
         Validator("VECTOR_SYNC_QUEUE_MAX_SIZE", gte=1),
         Validator("VECTOR_SYNC_METRICS_REFRESH_INTERVAL", gte=1),
+        Validator("VECTOR_DENSITY_SNAPSHOT_INTERVAL", gte=1),
+        Validator("VECTOR_DENSITY_SNAPSHOT_MAX_DOCUMENTS", gte=1),
         Validator("VECTOR_RAM_HNSW_OVERHEAD_FACTOR", gte=1),
         Validator("VECTOR_SYNC_USER_POLL_INTERVAL", gte=1),
         Validator("HEALTH_READY_REFRESH_INTERVAL", gte=1),
@@ -924,6 +929,14 @@ class Settings:
     # outstanding-work + indexed documents/chunks. Decoupled from the consumer
     # so the gauges are correct on every deployment mode and queue backend.
     vector_sync_metrics_refresh_interval: int = 20  # seconds
+    # Current-corpus chunk-density snapshot (vector/metrics_publisher.py:
+    # vector_density_snapshot_task). Scrolls the collection to recompute the
+    # distribution of documents CURRENTLY in Qdrant, so it runs on its own slower
+    # cadence than the count()-based gauges above. ``max_documents`` caps the
+    # scroll; hitting it sets the ``..._snapshot_truncated`` gauge (no silent cap).
+    vector_density_snapshot_enabled: bool = True
+    vector_density_snapshot_interval: int = 300  # seconds
+    vector_density_snapshot_max_documents: int = 50000
     # HNSW-graph/segment overhead multiplier applied when estimating dense-vector
     # RAM (``chunks * dim * 4 bytes * factor``). ~1.5 matches the cost-to-serve
     # note's ~6 KB / 1024-dim observation; a deployment knob because the real
@@ -1729,6 +1742,9 @@ def get_settings() -> Settings:
         "vector_sync_processor_workers": "VECTOR_SYNC_PROCESSOR_WORKERS",
         "vector_sync_queue_max_size": "VECTOR_SYNC_QUEUE_MAX_SIZE",
         "vector_sync_metrics_refresh_interval": "VECTOR_SYNC_METRICS_REFRESH_INTERVAL",
+        "vector_density_snapshot_enabled": "VECTOR_DENSITY_SNAPSHOT_ENABLED",
+        "vector_density_snapshot_interval": "VECTOR_DENSITY_SNAPSHOT_INTERVAL",
+        "vector_density_snapshot_max_documents": "VECTOR_DENSITY_SNAPSHOT_MAX_DOCUMENTS",
         "vector_ram_hnsw_overhead_factor": "VECTOR_RAM_HNSW_OVERHEAD_FACTOR",
         "vector_sync_user_poll_interval": "VECTOR_SYNC_USER_POLL_INTERVAL",
         "vector_sync_orphan_sweep_enabled": "VECTOR_SYNC_ORPHAN_SWEEP_ENABLED",

nextcloud_mcp_server/observability/metrics.py

@@ -16,14 +16,18 @@
 
 import functools
 import logging
+import threading
 import time
 
 from prometheus_client import (
+    REGISTRY,
     Counter,
     Gauge,
     Histogram,
     start_http_server,
 )
+from prometheus_client.core import GaugeHistogramMetricFamily
+from prometheus_client.registry import Collector
 
 from nextcloud_mcp_server.observability.tracing import trace_operation
 
@@ -460,11 +464,101 @@
 # tail higher and disproportionately inflate vector RAM relative to the billed
 # source bytes. Buckets straddle that band so the risky tail is visible. Recorded
 # for both index modes (density is a property of content, not of dense-vs-keyword).
+#
+# Shared with the current-corpus snapshot GaugeHistogram
+# (``astrolabe_qdrant_chunk_density_chunks_per_mb_current``) so the ingest-flow
+# panel and the current-distribution panel use identical bucket edges and are
+# directly comparable.
+CHUNK_DENSITY_BUCKETS = (1, 5, 10, 20, 40, 60, 91, 120, 160, 200, 300, 500)
+
 document_chunk_density_chunks_per_mb = Histogram(
     "astrolabe_document_chunk_density_chunks_per_mb",
     "Chunks produced per MB of source content, per embedded document",
     ["doc_type"],
-    buckets=(1, 5, 10, 20, 40, 60, 91, 120, 160, 200, 300, 500),
+    buckets=CHUNK_DENSITY_BUCKETS,
+)
+
+# -----------------------------------------------------------------------------
+# Current-corpus chunk-density snapshot (GaugeHistogram).
+#
+# Unlike the ingest-time Histogram above — which accumulates one observation per
+# document as it is embedded and never decrements — this is a *snapshot* of the
+# density distribution of the documents CURRENTLY resident in Qdrant, recomputed
+# periodically by scrolling the collection (see
+# ``vector.metrics_publisher.vector_density_snapshot_task``). A GaugeHistogram is
+# the correct Prometheus type: its buckets rise and fall as the corpus changes.
+#
+# Fed forward-only: only documents whose Qdrant payload carries
+# ``payload_keys.SOURCE_BYTES`` contribute. Documents indexed before that key
+# shipped (or otherwise missing a usable source size) are counted separately in
+# ``chunk_density_uncovered_documents`` so the snapshot's coverage is explicit.
+# -----------------------------------------------------------------------------
+QDRANT_CHUNK_DENSITY_CURRENT_METRIC = (
+    "astrolabe_qdrant_chunk_density_chunks_per_mb_current"
+)
+
+
+class _ChunkDensitySnapshotCollector(Collector):
+    """Custom collector exposing the current-corpus density as a GaugeHistogram.
+
+    Holds the most recent snapshot, keyed by ``doc_type``. ``update`` replaces the
+    whole snapshot atomically (a fresh scroll produces a complete new picture);
+    ``collect`` yields one GaugeHistogram sample set per ``doc_type``. Emits
+    nothing until the first snapshot lands, so a scrape before the publisher's
+    first pass simply omits the metric rather than reporting a misleading zero.
+
+    ``_snapshot`` maps ``doc_type -> (cumulative_buckets, gsum)`` where
+    ``cumulative_buckets`` is a list of ``(le_str, cumulative_count)`` including
+    the terminal ``"+Inf"`` bucket, matching Prometheus cumulative-bucket
+    semantics. ``gcount`` is the ``+Inf`` count, so it is not stored separately.
+    """
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._snapshot: dict[str, tuple[list[tuple[str, float]], float]] = {}
+
+    def update(
+        self, snapshot: dict[str, tuple[list[tuple[str, float]], float]]
+    ) -> None:
+        with self._lock:
+            self._snapshot = snapshot
+
+    def collect(self):
+        with self._lock:
+            snapshot = self._snapshot
+        if not snapshot:
+            return
+        family = GaugeHistogramMetricFamily(
+            QDRANT_CHUNK_DENSITY_CURRENT_METRIC,
+            "Chunks per MB of source content across documents currently in Qdrant "
+            "(snapshot, recomputed periodically)",
+            labels=["doc_type"],
+        )
+        for doc_type, (buckets, gsum) in snapshot.items():
+            family.add_metric([doc_type], buckets, gsum_value=gsum)
+        yield family
+
+
+chunk_density_snapshot_collector = _ChunkDensitySnapshotCollector()
+REGISTRY.register(chunk_density_snapshot_collector)
+
+# Documents currently in Qdrant that could NOT be placed in the density snapshot
+# because they carry no usable source-byte size (payload predates
+# payload_keys.SOURCE_BYTES, or the value is missing/non-positive). Makes the
+# forward-only coverage gap visible instead of silently shrinking the histogram.
+chunk_density_uncovered_documents = Gauge(
+    "astrolabe_qdrant_chunk_density_uncovered_documents",
+    "Documents in Qdrant excluded from the chunk-density snapshot (no source_bytes)",
+    ["doc_type"],
+)
+
+# 1 when the last density snapshot stopped early at the scan cap
+# (vector_density_snapshot_max_documents) and so covers only a prefix of the
+# collection; 0 when the whole collection was scanned. Alertable so a truncated
+# snapshot is never mistaken for a complete one.
+chunk_density_snapshot_truncated = Gauge(
+    "astrolabe_qdrant_chunk_density_snapshot_truncated",
+    "1 if the last chunk-density snapshot hit the document scan cap (partial)",
 )
 
 documents_indexed_total = Counter(
@@ -1024,6 +1118,62 @@ def record_chunk_density(doc_type: str, chunk_count: int, source_bytes: int) ->
         )
 
 
+def density_bucket_index(chunks_per_mb: float) -> int:
+    """Index into a per-bucket tally (``CHUNK_DENSITY_BUCKETS`` + overflow slot).
+
+    Returns the position of the first bucket whose upper edge is ``>=`` the value,
+    or ``len(CHUNK_DENSITY_BUCKETS)`` (the trailing ``"+Inf"`` overflow slot) when
+    the value exceeds every finite edge. The companion tally therefore has length
+    ``len(CHUNK_DENSITY_BUCKETS) + 1``. Shared by the snapshot publisher so its
+    bucketing matches these exact edges.
+    """
+    for idx, edge in enumerate(CHUNK_DENSITY_BUCKETS):
+        if chunks_per_mb <= edge:
+            return idx
+    return len(CHUNK_DENSITY_BUCKETS)
+
+
+def update_qdrant_chunk_density_snapshot(
+    per_doc_type: dict[str, tuple[list[float], float]],
+    *,
+    uncovered: dict[str, int] | None = None,
+    truncated: bool = False,
+) -> None:
+    """Publish one current-corpus chunk-density snapshot (GaugeHistogram + coverage).
+
+    ``per_doc_type`` maps ``doc_type -> (bucket_counts, gsum)`` where
+    ``bucket_counts`` is a NON-cumulative per-bucket tally aligned to
+    ``CHUNK_DENSITY_BUCKETS`` with one trailing overflow (``"+Inf"``) slot
+    (length ``len(CHUNK_DENSITY_BUCKETS) + 1``, as produced via
+    ``density_bucket_index``), and ``gsum`` is the sum of observed densities. The
+    tally is converted to Prometheus cumulative ``(le, count)`` buckets here — the
+    single place cumulative-bucket semantics live — and the GaugeHistogram
+    snapshot is swapped atomically.
+
+    ``uncovered`` (doc_type -> count of docs with no usable source size) and
+    ``truncated`` (scan hit the document cap) update the companion coverage
+    gauges. The uncovered gauge is fully reset each snapshot so a doc_type that
+    falls back to zero uncovered does not leave a stale series.
+    """
+    edges = [str(b) for b in CHUNK_DENSITY_BUCKETS] + ["+Inf"]
+    snapshot: dict[str, tuple[list[tuple[str, float]], float]] = {}
+    for doc_type, (bucket_counts, gsum) in per_doc_type.items():
+        cumulative: list[tuple[str, float]] = []
+        running = 0.0
+        for le, count in zip(edges, bucket_counts):
+            running += count
+            cumulative.append((le, running))
+        snapshot[doc_type] = (cumulative, gsum)
+    chunk_density_snapshot_collector.update(snapshot)
+
+    # Reset then repopulate so a doc_type absent this round drops to no series.
+    chunk_density_uncovered_documents.clear()
+    for doc_type, count in (uncovered or {}).items():
+        chunk_density_uncovered_documents.labels(doc_type=doc_type).set(count)
+
+    chunk_density_snapshot_truncated.set(1 if truncated else 0)
+
+
 # =============================================================================
 # Decorator for Automatic Tool Instrumentation
 # =============================================================================