fix(telemetry): report indexed document counts (#16038)

.agents/feature-map/components/observability.md

@@ -431,6 +431,12 @@ reads. The intervals are constants in `utils/fleet_telemetry.py`
 (`COLLECTION_INTERVAL_SECONDS` and the others); the fleet service cannot change
 them.
 
+Connector document counts use `db/document.py:get_document_counts_for_cc_pairs`,
+the same query as the connector admin page. It counts indexed document links for
+each connector and credential pair. Pending links do not count. The collector
+batches the capped inventory under its statement timeout; it does not read the
+legacy `ConnectorCredentialPair.total_docs_indexed` counter.
+
 **Identity.** `utils/fleet_telemetry.py:deployment_key` returns the stored key
 (`db/fleet_enrollment.py:get_or_create_deployment_key`). Under `MULTI_TENANT`, it
 derives the key from `WEB_DOMAIN`, and each tenant reports as a separate customer

backend/onyx/db/fleet_telemetry.py

@@ -7,6 +7,7 @@ from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session
 from sqlalchemy.sql.elements import KeyedColumnElement
 
+from onyx.db.document import get_document_counts_for_cc_pairs
 from onyx.db.enums import (
     AccessType,
     AccountType,
@@ -30,6 +31,7 @@ from onyx.db.models import (
     SyncRecord,
     User,
 )
+from onyx.server.documents.models import ConnectorCredentialPairIdentifier
 
 # Rows per read.
 ROW_LIMIT: int = 1000
@@ -67,9 +69,9 @@ def connector_rows(db_session: Session) -> list[dict[str, Any]]:
         select(
             ConnectorCredentialPair.id.label("cc_pair_id"),
             ConnectorCredentialPair.connector_id,
+            ConnectorCredentialPair.credential_id,
             Connector.source,
             ConnectorCredentialPair.status,
-            ConnectorCredentialPair.total_docs_indexed.label("doc_count"),
             ConnectorCredentialPair.last_successful_index_time.label("last_success_at"),
             Connector.refresh_freq.label("refresh_seconds"),
             Connector.prune_freq.label("prune_seconds"),
@@ -85,7 +87,26 @@ def connector_rows(db_session: Session) -> list[dict[str, Any]]:
         .order_by(ConnectorCredentialPair.id)
         .limit(ROW_LIMIT)
     )
-    return [dict(row) for row in db_session.execute(statement).mappings()]
+    rows = [dict(row) for row in db_session.execute(statement).mappings()]
+    # Match the admin count: only documents indexed for this exact pair.
+    counts = {
+        (connector_id, credential_id): count
+        for connector_id, credential_id, count in get_document_counts_for_cc_pairs(
+            db_session,
+            [
+                ConnectorCredentialPairIdentifier(
+                    connector_id=row["connector_id"],
+                    credential_id=row["credential_id"],
+                )
+                for row in rows
+            ],
+        )
+    }
+    for row in rows:
+        row["doc_count"] = counts.get(
+            (row["connector_id"], row.pop("credential_id")), 0
+        )
+    return rows
 
 
 def has_indexed_connectors__no_commit(db_session: Session) -> bool:

backend/tests/external_dependency_unit/telemetry/test_fleet_collector.py

@@ -26,6 +26,7 @@ from onyx.db.engine.sql_engine import get_session_with_tenant
 from onyx.db.enums import SyncType
 from onyx.db.fleet_telemetry import (
     attempt_rows,
+    connector_rows,
     email_domain_rows,
     has_indexed_connectors__no_commit,
     job_rows,
@@ -216,6 +217,12 @@ def test_a_slow_read_is_stopped(schema: str) -> None:
 def test_one_pass_sends_only_safe_events(
     schema: str, monkeypatch: pytest.MonkeyPatch
 ) -> None:
+    with get_session_with_tenant(tenant_id=schema) as db_session:
+        _add_document_link(db_session, 1, 1, "indexed-document")
+        mark_document_as_indexed_for_cc_pair__no_commit(
+            db_session, 1, 1, ["indexed-document"]
+        )
+        db_session.commit()
     sender = make_sender(capacity=256)
     monkeypatch.setattr(collector, "get_sender", lambda: sender)
     monkeypatch.setattr(collector, "get_cache_backend", Mock(return_value=FakeCache()))
@@ -240,6 +247,10 @@ def test_one_pass_sends_only_safe_events(
     )
     assert failed["error_code"] == "auth"
     assert by_type["connector"][0]["metadata"]["permission_sync_enabled"] is True
+    assert {data["cc_pair_id"]: data["doc_count"] for data in by_type["connector"]} == {
+        1: 1,
+        2: 0,
+    }
     assert "PRIVATE" not in json.dumps(events)
 
 
@@ -279,10 +290,12 @@ def test_indexed_connector_presence_follows_successful_indexing(schema: str) ->
         assert not has_indexed_connectors__no_commit(db_session)
         _add_document_link(db_session, 2, 1, "indexed-document")
         assert not has_indexed_connectors__no_commit(db_session)
+        assert [row["doc_count"] for row in connector_rows(db_session)] == [0, 0]
         mark_document_as_indexed_for_cc_pair__no_commit(
             db_session, 2, 1, ["indexed-document"]
         )
         assert has_indexed_connectors__no_commit(db_session)
+        assert [row["doc_count"] for row in connector_rows(db_session)] == [0, 1]
         # Indexing updates the link, not the pair's legacy counter.
         assert (
             db_session.scalar(
@@ -298,6 +311,38 @@ def test_indexed_connector_presence_follows_successful_indexing(schema: str) ->
             )
         )
         assert not has_indexed_connectors__no_commit(db_session)
+        assert [row["doc_count"] for row in connector_rows(db_session)] == [0, 0]
+
+
+def test_connector_counts_are_scoped_to_each_credential_pair(schema: str) -> None:
+    with get_session_with_tenant(tenant_id=schema) as db_session:
+        limit_statement_time(db_session)
+        db_session.execute(
+            text(
+                f'INSERT INTO "{schema}".connector_credential_pair '
+                "(id,name,status,connector_id,credential_id,access_type,total_docs_indexed) "
+                "VALUES (3,'Other credential','ACTIVE',1,2,'PUBLIC',0)"
+            )
+        )
+        for connector_id, credential_id, doc_id in [
+            (1, 1, "shared-document"),
+            (1, 2, "shared-document"),
+            (1, 2, "second-document"),
+            (2, 1, "other-connector"),
+            (1, 99, "unconfigured-credential"),
+        ]:
+            _add_document_link(db_session, connector_id, credential_id, doc_id)
+            mark_document_as_indexed_for_cc_pair__no_commit(
+                db_session, connector_id, credential_id, [doc_id]
+            )
+        _add_document_link(db_session, 1, 1, "pending-document")
+        rows = connector_rows(db_session)
+        assert {row["cc_pair_id"]: row["doc_count"] for row in rows} == {
+            1: 1,
+            2: 1,
+            3: 2,
+        }
+        assert all("credential_id" not in row for row in rows)
 
 
 def test_raw_library_upload_does_not_count_as_indexed(

deployment/fleet-telemetry.md

@@ -46,6 +46,10 @@ The server emits one query event when execution ends, even if the client disconn
 Cancellation freezes usage at the server's stop point. Search-mode requests leave this chat classification unset.
 Outcome remains separate from tool usage. Deploy the receiving server's optional `tool_usage` field before this instrumentation.
 
+Connector document counts use the same query as the connector admin page. They count indexed
+document links for each connector and credential pair, excluding pending links. Counts follow
+indexing and deletion directly; they do not use the pair's legacy stored counter.
+
 Only the API reports deployment type, using `DISABLE_VECTOR_DB` to distinguish Lite from Standard.
 Workers and standalone collectors omit type because their settings can differ from the API's.
 The connector flag checks indexed document links for existing connector/credential pairs, excluding