From 6676481af21676eeca7881cd64d3a8e7eb7b7d5a Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 13 Jul 2026 15:52:58 +0800 Subject: [PATCH 1/4] Expand OSS telemetry with ADR-0004 schema, heartbeats, and usage aggregates. Formalize the anonymous self-hosted emitter taxonomy so operators get reliable adoption and health signals without collecting document or user content. Co-authored-by: Cursor --- README.md | 18 + apps/api/.env.example | 12 + apps/api/main.py | 37 +- .../test_self_hosted_telemetry_contract.py | 187 +++++++- .../0004-anonymous-self-hosted-telemetry.md | 110 +++++ docs/adr/README.md | 10 + .../shared/services/telemetry/__init__.py | 11 +- .../shared/services/telemetry/aggregates.py | 398 +++++++++++++++--- .../shared/services/telemetry/config.py | 4 +- .../shared/services/telemetry/events.py | 151 ++++++- .../shared/services/telemetry/runtime.py | 185 +++++++- 11 files changed, 1059 insertions(+), 64 deletions(-) create mode 100644 docs/adr/0004-anonymous-self-hosted-telemetry.md diff --git a/README.md b/README.md index 8681c6028..442a540f5 100644 --- a/README.md +++ b/README.md @@ -244,6 +244,24 @@ make check - External dependency guide: [docs/external-services.md](docs/external-services.md) +- Architecture decisions: + [docs/adr/README.md](docs/adr/README.md) + +## Telemetry + +Self-hosted Knowhere emits **anonymous** product telemetry to PostHog so Ontos +operators can understand OSS adoption (install liveness, usage aggregates, +client/document mix). Events never include filenames, prompts, emails, IPs, or +geo. Schema and allowlists are locked in +[ADR-0004](docs/adr/0004-anonymous-self-hosted-telemetry.md). + +Telemetry is **default-on**. To opt out, set: + +```bash +TELEMETRY_ENABLED=false +``` + +Related settings live in `apps/api/.env.example` under `TELEMETRY_*`. ## Citation diff --git a/apps/api/.env.example b/apps/api/.env.example index f4dc6ce8d..04f297b40 100644 --- a/apps/api/.env.example +++ b/apps/api/.env.example @@ -30,6 +30,18 @@ TMP_PATH=/tmp/knowhere # Optional or development-only: observability and local dashboard wiring LOGFIRE_TOKEN= +# Anonymous self-hosted product telemetry (PostHog). Default-on; opt out with false. +# See docs/adr/0004-anonymous-self-hosted-telemetry.md and the README Telemetry section. +TELEMETRY_ENABLED=true +# TELEMETRY_POSTHOG_HOST=https://us.i.posthog.com +# TELEMETRY_POSTHOG_PROJECT_KEY= +# TELEMETRY_INSTALLATION_ID= +# TELEMETRY_INSTALLATION_ID_PATH=/data/secrets/telemetry-installation-id +# TELEMETRY_BATCH_SIZE=20 +# TELEMETRY_REQUEST_TIMEOUT_SECONDS=2.0 +# TELEMETRY_DEPLOYMENT_MODE=self_hosted +# TELEMETRY_AGGREGATE_INTERVAL_SECONDS=300 + # Required for local startup: database DATABASE_URL=postgresql+asyncpg://root:root123@localhost:5432/Knowhere DB_SSL_MODE=disable diff --git a/apps/api/main.py b/apps/api/main.py index 7140fab0d..3e5475661 100644 --- a/apps/api/main.py +++ b/apps/api/main.py @@ -70,10 +70,23 @@ async def lifespan(app: FastAPI): await load_rules(session) logger.info("rate limit rules loaded at startup; restart the pod to apply changes") + import time + from shared.services.telemetry.aggregates import ( start_self_hosted_aggregate_telemetry, ) - from shared.services.telemetry.runtime import start_self_hosted_telemetry + from shared.services.telemetry.runtime import ( + build_postgres_health_probe, + build_redis_health_probe, + start_self_hosted_heartbeat_telemetry, + start_self_hosted_telemetry, + ) + + telemetry_started_at = time.monotonic() + + async def _redis_ping() -> bool: + redis_service = redis_pool_manager.get_redis_service() + return await redis_service.ping() telemetry_runtime = await start_self_hosted_telemetry( settings, @@ -86,6 +99,7 @@ async def lifespan(app: FastAPI): app.state.self_hosted_telemetry_client = None app.state.self_hosted_telemetry_config = None app.state.self_hosted_aggregate_telemetry_runner = None + app.state.self_hosted_heartbeat_telemetry_runner = None else: telemetry_client, telemetry_config = telemetry_runtime app.state.self_hosted_telemetry_client = telemetry_client @@ -99,6 +113,16 @@ async def lifespan(app: FastAPI): api_metrics=app.state.self_hosted_api_telemetry_metrics, ) ) + app.state.self_hosted_heartbeat_telemetry_runner = ( + await start_self_hosted_heartbeat_telemetry( + settings, + telemetry_client=telemetry_client, + config=telemetry_config, + started_at_monotonic=telemetry_started_at, + postgres_probe=build_postgres_health_probe(get_db_context), + redis_probe=build_redis_health_probe(_redis_ping), + ) + ) mcp_server = getattr(app.state, "retrieval_mcp_server", None) mcp_session_manager = getattr(mcp_server, "session_manager", None) @@ -114,13 +138,20 @@ async def lifespan(app: FastAPI): from shared.services.telemetry.aggregates import ( stop_self_hosted_aggregate_telemetry, ) - from shared.services.telemetry.runtime import stop_self_hosted_telemetry + from shared.services.telemetry.runtime import ( + stop_self_hosted_heartbeat_telemetry, + stop_self_hosted_telemetry, + ) + await stop_self_hosted_heartbeat_telemetry( + getattr(app.state, "self_hosted_heartbeat_telemetry_runner", None) + ) await stop_self_hosted_aggregate_telemetry( getattr(app.state, "self_hosted_aggregate_telemetry_runner", None) ) await stop_self_hosted_telemetry( - getattr(app.state, "self_hosted_telemetry_client", None) + getattr(app.state, "self_hosted_telemetry_client", None), + config=getattr(app.state, "self_hosted_telemetry_config", None), ) except Exception as e: logger.error(f"self-hosted telemetry shutdown failed: {e}") diff --git a/apps/api/tests/contract/test_self_hosted_telemetry_contract.py b/apps/api/tests/contract/test_self_hosted_telemetry_contract.py index 54152c258..b974e7613 100644 --- a/apps/api/tests/contract/test_self_hosted_telemetry_contract.py +++ b/apps/api/tests/contract/test_self_hosted_telemetry_contract.py @@ -26,11 +26,22 @@ stop_self_hosted_aggregate_telemetry, ) from shared.services.telemetry.events import ( + SCHEMA_VERSION, + build_base_event_properties, + compute_success_rate, + count_to_bucket, get_allowed_telemetry_event_names, + normalize_client_name, + normalize_document_type, + normalize_source_type, sanitize_event_properties, + uptime_seconds_to_bucket, ) from shared.services.telemetry.identity import get_or_create_installation_id - +from shared.services.telemetry.runtime import ( + SelfHostedHeartbeatTelemetryRunner, + stop_self_hosted_telemetry, +) def test_installation_id_is_generated_once(tmp_path: Path) -> None: installation_id_path = tmp_path / "telemetry-installation-id" @@ -99,9 +110,175 @@ def test_aggregate_event_names_are_allowed() -> None: "self_hosted_worker_aggregate", "self_hosted_api_aggregate", "self_hosted_provider_aggregate", + "self_hosted_document_type_aggregate", + "self_hosted_client_aggregate", }.issubset(get_allowed_telemetry_event_names()) +def test_normalize_document_type_and_client_name() -> None: + assert normalize_document_type("report.PDF") == "pdf" + assert normalize_document_type("photo.jpeg") == "image" + assert normalize_document_type("notes.htm") == "html" + assert normalize_document_type("secret.xyz") == "other" + assert normalize_document_type(None) == "other" + assert normalize_client_name("node-sdk") == "node-sdk" + assert normalize_client_name("CLI") == "cli" + assert normalize_client_name("custom-bot") == "other" + assert normalize_source_type("direct_upload") == "file" + assert normalize_source_type("url") == "url" + assert normalize_source_type("demo") == "other" + + +def test_success_rate_and_count_buckets() -> None: + assert compute_success_rate(9, 1) == 0.9 + assert compute_success_rate(0, 0) == 0.0 + assert count_to_bucket(0) == "0" + assert count_to_bucket(7) == "1-10" + assert count_to_bucket(50) == "11-100" + assert count_to_bucket(101) == "100+" + assert uptime_seconds_to_bucket(30) == "0m-5m" + assert uptime_seconds_to_bucket(3600) == "1h-24h" + + +def test_usage_and_document_type_properties_strip_sensitive_values() -> None: + usage_properties = sanitize_event_properties( + "self_hosted_usage_aggregate", + { + "app_version": "1.2.3", + "window_seconds": 86_400, + "success_rate_24h": 0.9, + "source_file_jobs_24h": 2, + "email": "user@example.com", + "document_name": "private.pdf", + "source_file_name": "private.pdf", + }, + ) + document_type_properties = sanitize_event_properties( + "self_hosted_document_type_aggregate", + { + "document_type": "pdf", + "jobs_created_24h": 1, + "source_file_name": "private.pdf", + "email": "user@example.com", + }, + ) + client_properties = sanitize_event_properties( + "self_hosted_client_aggregate", + { + "created_by_client": "cli", + "jobs_created_24h": 1, + "client_version": "9.9.9", + "email": "user@example.com", + }, + ) + + assert usage_properties == { + "app_version": "1.2.3", + "window_seconds": 86_400, + "success_rate_24h": 0.9, + "source_file_jobs_24h": 2, + } + assert document_type_properties == { + "document_type": "pdf", + "jobs_created_24h": 1, + } + assert client_properties == { + "created_by_client": "cli", + "jobs_created_24h": 1, + } + + +@pytest.mark.asyncio +async def test_heartbeat_emit_once_includes_health_and_uptime( + tmp_path: Path, +) -> None: + posthog_client = _FakePostHogClient() + config = _build_config(tmp_path) + telemetry_client = TelemetryClient(config, posthog_client=posthog_client) + await telemetry_client.start() + + async def postgres_probe() -> bool: + return True + + async def redis_probe() -> bool: + return False + + runner = SelfHostedHeartbeatTelemetryRunner( + config=config, + telemetry_client=telemetry_client, + settings=_HeartbeatSettings(), + interval_seconds=60, + started_at_monotonic=0.0, + postgres_probe=postgres_probe, + redis_probe=redis_probe, + ) + await runner.emit_once() + await telemetry_client.stop() + + assert len(posthog_client.captured_events) == 1 + captured = posthog_client.captured_events[0] + assert captured.event_name == "self_hosted_instance_heartbeat" + properties = cast(dict[str, object], captured.kwargs["properties"]) + assert properties["api_healthy"] is True + assert properties["postgres_healthy"] is True + assert properties["redis_healthy"] is False + assert properties["uptime_bucket"] in { + "0m-5m", + "5m-1h", + "1h-24h", + "24h-7d", + "7d+", + } + assert properties["schema_version"] == SCHEMA_VERSION + + +@pytest.mark.asyncio +async def test_shutdown_includes_base_event_properties(tmp_path: Path) -> None: + posthog_client = _FakePostHogClient() + config = _build_config(tmp_path) + telemetry_client = TelemetryClient(config, posthog_client=posthog_client) + await telemetry_client.start() + + await stop_self_hosted_telemetry(telemetry_client, config=config) + + assert len(posthog_client.captured_events) == 1 + captured = posthog_client.captured_events[0] + assert captured.event_name == "self_hosted_instance_shutdown" + assert captured.kwargs["properties"] == { + **build_base_event_properties(config), + "$process_person_profile": False, + } + + +def test_usage_aggregate_allowlist_includes_v2_keys() -> None: + properties = sanitize_event_properties( + "self_hosted_usage_aggregate", + { + "success_rate_24h": 1.0, + "job_duration_p95_seconds_24h": 12.5, + "has_webhooks_24h": True, + "has_retrieval_24h": False, + "jobs_created_bucket": "1-10", + "pages_processed_bucket": "0", + "source_file_jobs_24h": 1, + "source_url_jobs_24h": 0, + "source_other_jobs_24h": 0, + "filename": "secret.pdf", + }, + ) + assert set(properties) == { + "success_rate_24h", + "job_duration_p95_seconds_24h", + "has_webhooks_24h", + "has_retrieval_24h", + "jobs_created_bucket", + "pages_processed_bucket", + "source_file_jobs_24h", + "source_url_jobs_24h", + "source_other_jobs_24h", + } + + def test_self_hosted_telemetry_defaults_to_enabled( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -428,6 +605,13 @@ class _AggregateSettings: TELEMETRY_AGGREGATE_INTERVAL_SECONDS: int = 60 +@dataclass(frozen=True) +class _HeartbeatSettings: + API_STANDALONE_MODE_ENABLED: bool = False + BILLING_ENABLED: bool = False + TELEMETRY_AGGREGATE_INTERVAL_SECONDS: int = 60 + + class _FailingSessionContext(AbstractAsyncContextManager[AsyncSession]): async def __aenter__(self) -> AsyncSession: raise RuntimeError("database unavailable") @@ -517,4 +701,5 @@ def _build_config( environment="production", app_env="production", service_name="knowhere-api", + schema_version=SCHEMA_VERSION, ) diff --git a/docs/adr/0004-anonymous-self-hosted-telemetry.md b/docs/adr/0004-anonymous-self-hosted-telemetry.md new file mode 100644 index 000000000..bb51f2603 --- /dev/null +++ b/docs/adr/0004-anonymous-self-hosted-telemetry.md @@ -0,0 +1,110 @@ +# 0004 Anonymous Self-Hosted Telemetry + +## Status + +Accepted + +## Context + +Self-hosted Knowhere installs already emit anonymous PostHog events so Ontos +operators can understand OSS adoption. The v1 emitter covered instance +lifecycle and coarse aggregates, but lacked: + +- a locked privacy and allowlist contract +- real periodic health heartbeats +- SaaS `/usage`-parity KPIs (success rate, p95 duration, source mix) +- document-type and client-type mix without leaking filenames or free-form metadata + +Operators need a stable schema and metric catalog before the metrics dashboard +and official clients depend on new events. + +## Decision + +### Purpose and opt-out + +Anonymous self-hosted telemetry exists for Ontos operators measuring OSS / +self-hosted adoption. It is **default-on**. Operators opt out with +`TELEMETRY_ENABLED=false`. Transport remains PostHog; Logfire/OTEL are out of +scope. + +### Privacy bounds + +Events must never include filenames, prompts, emails, IPs, geo, document +content, or arbitrary customer metadata keys. Only allowlisted scalar property +names may leave the box. Free-form `client_version` is not emitted in aggregate +events (cardinality); app version stays on base properties only. + +### Schema version + +`schema_version = 2026-07-telemetry-v2` + +### Event catalog + +Keep the existing eight event names and add two: + +| Event | Role | +| --- | --- | +| `self_hosted_instance_started` | Install boot | +| `self_hosted_instance_heartbeat` | Periodic liveness + health | +| `self_hosted_instance_shutdown` | Graceful stop (includes base props) | +| `self_hosted_usage_aggregate` | Fleet usage snapshot (24h window) | +| `self_hosted_retrieval_aggregate` | Retrieval activity | +| `self_hosted_worker_aggregate` | Worker backlog / completion | +| `self_hosted_api_aggregate` | In-process API request counters | +| `self_hosted_provider_aggregate` | Parse/retrieval provider activity | +| `self_hosted_document_type_aggregate` | Per allowlisted document type | +| `self_hosted_client_aggregate` | Per allowlisted created_by_client | + +### Allowlists + +- `document_type`: `pdf|docx|doc|xlsx|xls|pptx|ppt|csv|txt|md|html|image|other` + - Extension from `job_metadata->>'source_file_name'` only; map + `png|jpg|jpeg|gif|webp|tiff` → `image`; everything else → `other` +- `created_by_client`: `cli|node-sdk|dashboard|notebook|mcp|api|other` + - From `job_metadata #>> '{document_metadata,created_by_client}'` +- `source_type`: `file|url|other` + +### Success rate + +`success_rate_24h = done / (done + failed)` over the same 24h window, as a +float in **0–1**. Exclude non-terminal statuses. When `(done + failed) = 0`, +emit `0.0`. + +### Metric IDs (dashboard catalog) + +| Metric ID | Source | +| --- | --- | +| `oss.active_installs_7d` / `oss.active_installs_30d` | heartbeat distinct installs | +| `oss.new_installs_*` | started | +| `oss.retention.w0_w1` / `oss.retention.w0_w4` | started ∩ heartbeat | +| `oss.usage.jobs_created_24h` | usage aggregate | +| `oss.usage.jobs_done_24h` / `oss.usage.jobs_failed_24h` | usage / worker | +| `oss.usage.success_rate_24h` | usage aggregate | +| `oss.usage.pages_processed_24h` | usage aggregate | +| `oss.usage.job_duration_avg_seconds_24h` | worker aggregate | +| `oss.usage.job_duration_p95_seconds_24h` | usage aggregate | +| `oss.usage.backlog_*` | worker pending/running/… | +| `oss.usage.source_*` | `source_{file,url,other}_jobs_24h` | +| `oss.client_*` | client aggregate | +| `oss.document_type_*` | document type aggregate | +| `oss.health.*` | heartbeat health fields | +| `oss.fleet.version_*` / `oss.fleet.flags_*` | base props last-known | + +### Capability flags and buckets + +Usage aggregate may emit: + +- `has_webhooks_24h`, `has_retrieval_24h` (bool) +- `jobs_created_bucket`, `pages_processed_bucket` as + `0|1-10|11-100|100+` + +## Consequences + +- Emitter code must bump `schema_version`, extend property allowlists, emit + real heartbeats, and add document-type / client aggregates before the + metrics-dashboard Telemetry UI can rely on them. +- Official clients should populate `document_metadata.created_by_client` + (separate decision surface); until then dashboards must tolerate + `other`-heavy client mix. +- Changing allowlists or the success-rate formula requires a new ADR or an + explicit schema_version bump. diff --git a/docs/adr/README.md b/docs/adr/README.md index 3d0cb4fd0..68e15b221 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -10,3 +10,13 @@ Use this shape: - Context - Decision - Consequences + +## Index + +| ADR | Title | +| --- | --- | +| [0001](0001-keep-routes-and-worker-tasks-as-adapters.md) | Keep routes and worker tasks as adapters | +| [0002](0002-use-typed-workflow-outcomes.md) | Use typed workflow outcomes | +| [0003](0003-keep-retrieval-workflow-policy-explicit.md) | Keep retrieval workflow policy explicit | +| [0004](0004-anonymous-self-hosted-telemetry.md) | Anonymous self-hosted telemetry | +| \ No newline at end of file diff --git a/packages/shared-python/shared/services/telemetry/__init__.py b/packages/shared-python/shared/services/telemetry/__init__.py index 047ce17d6..ee2fa6003 100644 --- a/packages/shared-python/shared/services/telemetry/__init__.py +++ b/packages/shared-python/shared/services/telemetry/__init__.py @@ -1,18 +1,27 @@ """Anonymous self-hosted telemetry helpers.""" from .client import TelemetryClient -from .config import TelemetryRuntimeConfig, build_telemetry_config +from .config import SCHEMA_VERSION, TelemetryRuntimeConfig, build_telemetry_config from .events import ( + build_base_event_properties, build_instance_event_properties, get_allowed_telemetry_event_names, + normalize_client_name, + normalize_document_type, + normalize_source_type, ) from .identity import get_or_create_installation_id __all__ = [ + "SCHEMA_VERSION", "TelemetryClient", "TelemetryRuntimeConfig", "build_telemetry_config", + "build_base_event_properties", "build_instance_event_properties", "get_allowed_telemetry_event_names", "get_or_create_installation_id", + "normalize_client_name", + "normalize_document_type", + "normalize_source_type", ] diff --git a/packages/shared-python/shared/services/telemetry/aggregates.py b/packages/shared-python/shared/services/telemetry/aggregates.py index ddb5af9b7..80634ccc5 100644 --- a/packages/shared-python/shared/services/telemetry/aggregates.py +++ b/packages/shared-python/shared/services/telemetry/aggregates.py @@ -13,7 +13,15 @@ from .api_metrics import ApiRequestTelemetryMetrics, ApiRequestMetricsSnapshot from .client import TelemetryClient from .config import TelemetryRuntimeConfig, TelemetrySettings -from .events import TelemetryProperties, build_base_event_properties +from .events import ( + TelemetryProperties, + build_base_event_properties, + compute_success_rate, + count_to_bucket, + normalize_client_name, + normalize_document_type, + normalize_source_type, +) AGGREGATE_WINDOW_SECONDS = 24 * 60 * 60 AGGREGATE_ADVISORY_LOCK_ID = 0x4B4E4F5748455245 @@ -52,14 +60,14 @@ def __init__( async def emit_once(self) -> None: """Collect and emit all aggregate snapshots once.""" - event_properties = await collect_self_hosted_aggregate_event_properties( + event_captures = await collect_self_hosted_aggregate_event_captures( config=self._config, db_session_factory=self._db_session_factory, api_metrics=self._api_metrics, window_seconds=AGGREGATE_WINDOW_SECONDS, api_window_seconds=self._interval_seconds, ) - for event_name, properties in event_properties.items(): + for event_name, properties in event_captures: self._telemetry_client.capture(event_name, properties) def start(self) -> None: @@ -126,61 +134,164 @@ async def stop_self_hosted_aggregate_telemetry( await runner.stop() -async def collect_self_hosted_aggregate_event_properties( +async def collect_self_hosted_aggregate_event_captures( *, config: TelemetryRuntimeConfig, db_session_factory: DatabaseSessionFactory, api_metrics: ApiRequestTelemetryMetrics, window_seconds: int = AGGREGATE_WINDOW_SECONDS, api_window_seconds: int = AGGREGATE_WINDOW_SECONDS, -) -> dict[str, TelemetryProperties]: - """Collect aggregate event properties without including customer content.""" - event_properties = { - "self_hosted_api_aggregate": _collect_api_aggregate( - config, - api_metrics.snapshot_and_reset(), - api_window_seconds, - ), - } +) -> list[tuple[str, TelemetryProperties]]: + """Collect aggregate captures without including customer content.""" + captures: list[tuple[str, TelemetryProperties]] = [ + ( + "self_hosted_api_aggregate", + _collect_api_aggregate( + config, + api_metrics.snapshot_and_reset(), + api_window_seconds, + ), + ) + ] async with db_session_factory() as session: lock_acquired = await _try_aggregate_advisory_lock(session) if not lock_acquired: - return event_properties + return captures try: - event_properties.update( - { - "self_hosted_usage_aggregate": await _collect_usage_aggregate( - session, - config, - window_seconds, - ), - "self_hosted_retrieval_aggregate": await _collect_retrieval_aggregate( - session, - config, - window_seconds, - ), - "self_hosted_worker_aggregate": await _collect_worker_aggregate( - session, - config, - window_seconds, - ), - "self_hosted_provider_aggregate": await _collect_provider_aggregate( + captures.append( + ( + "self_hosted_usage_aggregate", + await _collect_usage_aggregate(session, config, window_seconds), + ) + ) + captures.append( + ( + "self_hosted_retrieval_aggregate", + await _collect_retrieval_aggregate( session, config, window_seconds, ), - } + ) ) - return event_properties + captures.append( + ( + "self_hosted_worker_aggregate", + await _collect_worker_aggregate(session, config, window_seconds), + ) + ) + captures.append( + ( + "self_hosted_provider_aggregate", + await _collect_provider_aggregate(session, config, window_seconds), + ) + ) + for properties in await _collect_document_type_aggregates( + session, + config, + window_seconds, + ): + captures.append(("self_hosted_document_type_aggregate", properties)) + for properties in await _collect_client_aggregates( + session, + config, + window_seconds, + ): + captures.append(("self_hosted_client_aggregate", properties)) + return captures finally: await _release_aggregate_advisory_lock(session) +async def collect_self_hosted_aggregate_event_properties( + *, + config: TelemetryRuntimeConfig, + db_session_factory: DatabaseSessionFactory, + api_metrics: ApiRequestTelemetryMetrics, + window_seconds: int = AGGREGATE_WINDOW_SECONDS, + api_window_seconds: int = AGGREGATE_WINDOW_SECONDS, +) -> dict[str, TelemetryProperties]: + """Collect singleton aggregate event properties (compat helper for tests).""" + captures = await collect_self_hosted_aggregate_event_captures( + config=config, + db_session_factory=db_session_factory, + api_metrics=api_metrics, + window_seconds=window_seconds, + api_window_seconds=api_window_seconds, + ) + event_properties: dict[str, TelemetryProperties] = {} + for event_name, properties in captures: + # Keep first capture for singleton events; multi-row events are omitted + # from this dict helper (use collect_self_hosted_aggregate_event_captures). + if event_name in { + "self_hosted_document_type_aggregate", + "self_hosted_client_aggregate", + }: + continue + event_properties[event_name] = properties + return event_properties + + async def _collect_usage_aggregate( session: AsyncSession, config: TelemetryRuntimeConfig, window_seconds: int, ) -> TelemetryProperties: + completed_jobs_24h = await _int_scalar( + session, + _windowed_count_sql("jobs", "updated_at", "status = 'done'"), + window_seconds, + ) + failed_jobs_24h = await _int_scalar( + session, + _windowed_count_sql("jobs", "updated_at", "status = 'failed'"), + window_seconds, + ) + jobs_created_24h = await _int_scalar( + session, + _windowed_count_sql("jobs", "created_at"), + window_seconds, + ) + pages_processed_24h = await _int_scalar( + session, + f""" + SELECT COALESCE(SUM(page_count), 0) + FROM jobs + WHERE updated_at >= {_window_start_expression()} + AND status = 'done' + """, + window_seconds, + ) + source_counts = await _collect_source_type_counts(session, window_seconds) + has_webhooks_24h = ( + await _int_scalar( + session, + f""" + SELECT CASE + WHEN EXISTS ( + SELECT 1 FROM jobs + WHERE created_at >= {_window_start_expression()} + AND webhook_enabled = true + ) + OR EXISTS ( + SELECT 1 FROM webhook_logs + WHERE created_at >= {_window_start_expression()} + ) + THEN 1 ELSE 0 + END + """, + window_seconds, + ) + > 0 + ) + has_retrieval_24h = ( + await _int_scalar( + session, + _windowed_count_sql("retrieval_runs", "created_at"), + window_seconds, + ) + > 0 + ) properties = _base_aggregate_properties(config, window_seconds) properties.update( { @@ -190,23 +301,31 @@ async def _collect_usage_aggregate( "SELECT COUNT(*) FROM api_keys WHERE is_active = true", ), "total_jobs": await _int_scalar(session, "SELECT COUNT(*) FROM jobs"), - "jobs_created_24h": await _int_scalar( - session, - _windowed_count_sql("jobs", "created_at"), - window_seconds, - ), + "jobs_created_24h": jobs_created_24h, + "jobs_created_bucket": count_to_bucket(jobs_created_24h), "active_jobs": await _int_scalar( session, "SELECT COUNT(*) FROM jobs WHERE status IN ('waiting-file', 'pending', 'running', 'converting')", ), - "completed_jobs_24h": await _int_scalar( - session, - _windowed_count_sql("jobs", "updated_at", "status = 'done'"), - window_seconds, + "completed_jobs_24h": completed_jobs_24h, + "failed_jobs_24h": failed_jobs_24h, + "success_rate_24h": compute_success_rate( + completed_jobs_24h, + failed_jobs_24h, ), - "failed_jobs_24h": await _int_scalar( + "job_duration_p95_seconds_24h": await _float_scalar( session, - _windowed_count_sql("jobs", "updated_at", "status = 'failed'"), + f""" + SELECT COALESCE( + percentile_cont(0.95) WITHIN GROUP ( + ORDER BY EXTRACT(EPOCH FROM (updated_at - created_at)) + ), + 0 + ) + FROM jobs + WHERE updated_at >= {_window_start_expression()} + AND status IN ('done', 'failed') + """, window_seconds, ), "total_documents": await _int_scalar( @@ -225,16 +344,8 @@ async def _collect_usage_aggregate( session, "SELECT COUNT(*) FROM job_chunks", ), - "pages_processed_24h": await _int_scalar( - session, - f""" - SELECT COALESCE(SUM(page_count), 0) - FROM jobs - WHERE updated_at >= {_window_start_expression()} - AND status = 'done' - """, - window_seconds, - ), + "pages_processed_24h": pages_processed_24h, + "pages_processed_bucket": count_to_bucket(pages_processed_24h), "credits_charged_24h": await _int_scalar( session, f""" @@ -244,11 +355,188 @@ async def _collect_usage_aggregate( """, window_seconds, ), + "has_webhooks_24h": has_webhooks_24h, + "has_retrieval_24h": has_retrieval_24h, + "source_file_jobs_24h": source_counts["file"], + "source_url_jobs_24h": source_counts["url"], + "source_other_jobs_24h": source_counts["other"], } ) return properties +async def _collect_source_type_counts( + session: AsyncSession, + window_seconds: int, +) -> dict[str, int]: + counts = {"file": 0, "url": 0, "other": 0} + result = await session.execute( + text( + f""" + SELECT source_type, COUNT(*) AS job_count + FROM jobs + WHERE created_at >= {_window_start_expression()} + GROUP BY source_type + """ + ), + {"window_seconds": window_seconds}, + ) + for row in result.mappings(): + source_type = normalize_source_type(cast(str | None, row["source_type"])) + counts[source_type] = counts.get(source_type, 0) + int(row["job_count"] or 0) + return counts + + +async def _collect_document_type_aggregates( + session: AsyncSession, + config: TelemetryRuntimeConfig, + window_seconds: int, +) -> list[TelemetryProperties]: + """Collect one aggregate per allowlisted document type with activity.""" + result = await session.execute( + text( + f""" + SELECT + job_metadata->>'source_file_name' AS source_file_name, + COUNT(*) FILTER ( + WHERE created_at >= {_window_start_expression()} + ) AS jobs_created_24h, + COUNT(*) FILTER ( + WHERE updated_at >= {_window_start_expression()} + AND status = 'done' + ) AS jobs_done_24h, + COUNT(*) FILTER ( + WHERE updated_at >= {_window_start_expression()} + AND status = 'failed' + ) AS jobs_failed_24h, + COALESCE( + SUM(page_count) FILTER ( + WHERE updated_at >= {_window_start_expression()} + AND status = 'done' + ), + 0 + ) AS pages_processed_24h + FROM jobs + WHERE created_at >= {_window_start_expression()} + OR ( + updated_at >= {_window_start_expression()} + AND status IN ('done', 'failed') + ) + GROUP BY job_metadata->>'source_file_name' + """ + ), + {"window_seconds": window_seconds}, + ) + merged: dict[str, dict[str, int]] = {} + for row in result.mappings(): + document_type = normalize_document_type( + cast(str | None, row["source_file_name"]) + ) + bucket = merged.setdefault( + document_type, + { + "jobs_created_24h": 0, + "jobs_done_24h": 0, + "jobs_failed_24h": 0, + "pages_processed_24h": 0, + }, + ) + bucket["jobs_created_24h"] += int(row["jobs_created_24h"] or 0) + bucket["jobs_done_24h"] += int(row["jobs_done_24h"] or 0) + bucket["jobs_failed_24h"] += int(row["jobs_failed_24h"] or 0) + bucket["pages_processed_24h"] += int(row["pages_processed_24h"] or 0) + + properties_list: list[TelemetryProperties] = [] + for document_type, counts in sorted(merged.items()): + if not any(counts.values()): + continue + properties = _base_aggregate_properties(config, window_seconds) + properties.update( + { + "document_type": document_type, + "jobs_created_24h": counts["jobs_created_24h"], + "jobs_done_24h": counts["jobs_done_24h"], + "jobs_failed_24h": counts["jobs_failed_24h"], + "pages_processed_24h": counts["pages_processed_24h"], + "success_rate_24h": compute_success_rate( + counts["jobs_done_24h"], + counts["jobs_failed_24h"], + ), + } + ) + properties_list.append(properties) + return properties_list + + +async def _collect_client_aggregates( + session: AsyncSession, + config: TelemetryRuntimeConfig, + window_seconds: int, +) -> list[TelemetryProperties]: + """Collect one aggregate per allowlisted created_by_client with activity.""" + result = await session.execute( + text( + f""" + SELECT + job_metadata #>> '{{document_metadata,created_by_client}}' AS created_by_client, + COUNT(*) FILTER ( + WHERE created_at >= {_window_start_expression()} + ) AS jobs_created_24h, + COUNT(*) FILTER ( + WHERE updated_at >= {_window_start_expression()} + AND status = 'done' + ) AS jobs_done_24h, + COUNT(*) FILTER ( + WHERE updated_at >= {_window_start_expression()} + AND status = 'failed' + ) AS jobs_failed_24h + FROM jobs + WHERE created_at >= {_window_start_expression()} + OR ( + updated_at >= {_window_start_expression()} + AND status IN ('done', 'failed') + ) + GROUP BY job_metadata #>> '{{document_metadata,created_by_client}}' + """ + ), + {"window_seconds": window_seconds}, + ) + merged: dict[str, dict[str, int]] = {} + for row in result.mappings(): + client_name = normalize_client_name(cast(str | None, row["created_by_client"])) + bucket = merged.setdefault( + client_name, + { + "jobs_created_24h": 0, + "jobs_done_24h": 0, + "jobs_failed_24h": 0, + }, + ) + bucket["jobs_created_24h"] += int(row["jobs_created_24h"] or 0) + bucket["jobs_done_24h"] += int(row["jobs_done_24h"] or 0) + bucket["jobs_failed_24h"] += int(row["jobs_failed_24h"] or 0) + + properties_list: list[TelemetryProperties] = [] + for client_name, counts in sorted(merged.items()): + if not any(counts.values()): + continue + properties = _base_aggregate_properties(config, window_seconds) + properties.update( + { + "created_by_client": client_name, + "jobs_created_24h": counts["jobs_created_24h"], + "jobs_done_24h": counts["jobs_done_24h"], + "jobs_failed_24h": counts["jobs_failed_24h"], + "success_rate_24h": compute_success_rate( + counts["jobs_done_24h"], + counts["jobs_failed_24h"], + ), + } + ) + properties_list.append(properties) + return properties_list + + async def _collect_retrieval_aggregate( session: AsyncSession, config: TelemetryRuntimeConfig, diff --git a/packages/shared-python/shared/services/telemetry/config.py b/packages/shared-python/shared/services/telemetry/config.py index 328f27060..d318f6921 100644 --- a/packages/shared-python/shared/services/telemetry/config.py +++ b/packages/shared-python/shared/services/telemetry/config.py @@ -6,6 +6,8 @@ from pathlib import Path from typing import Protocol +SCHEMA_VERSION = "2026-07-telemetry-v2" + class TelemetrySettings(Protocol): """Subset of app settings required by anonymous telemetry.""" @@ -41,7 +43,7 @@ class TelemetryRuntimeConfig: environment: str app_env: str service_name: str - schema_version: str = "2026-06-telemetry-v1" + schema_version: str = SCHEMA_VERSION @property def is_ready(self) -> bool: diff --git a/packages/shared-python/shared/services/telemetry/events.py b/packages/shared-python/shared/services/telemetry/events.py index 287e8ab98..ff3ad8b90 100644 --- a/packages/shared-python/shared/services/telemetry/events.py +++ b/packages/shared-python/shared/services/telemetry/events.py @@ -4,13 +4,55 @@ import os from collections.abc import Mapping +from pathlib import PurePosixPath from typing import TypeAlias, cast -from .config import TelemetryRuntimeConfig +from .config import SCHEMA_VERSION, TelemetryRuntimeConfig TelemetryPropertyValue: TypeAlias = str | int | float | bool | None TelemetryProperties: TypeAlias = dict[str, TelemetryPropertyValue] +DOCUMENT_TYPES = frozenset( + { + "pdf", + "docx", + "doc", + "xlsx", + "xls", + "pptx", + "ppt", + "csv", + "txt", + "md", + "html", + "image", + "other", + } +) + +CLIENT_NAMES = frozenset( + { + "cli", + "node-sdk", + "dashboard", + "notebook", + "mcp", + "api", + "other", + } +) + +SOURCE_TYPES = frozenset({"file", "url", "other"}) + +_IMAGE_EXTENSIONS = frozenset({"png", "jpg", "jpeg", "gif", "webp", "tiff"}) +_HTML_EXTENSIONS = frozenset({"html", "htm"}) + +_COUNT_BUCKETS = ( + (0, "0"), + (10, "1-10"), + (100, "11-100"), +) + _BASE_PROPERTY_NAMES = frozenset( { "app_env", @@ -66,8 +108,17 @@ "completed_jobs_24h", "credits_charged_24h", "failed_jobs_24h", + "has_retrieval_24h", + "has_webhooks_24h", + "job_duration_p95_seconds_24h", "jobs_created_24h", + "jobs_created_bucket", "pages_processed_24h", + "pages_processed_bucket", + "source_file_jobs_24h", + "source_other_jobs_24h", + "source_url_jobs_24h", + "success_rate_24h", "total_document_chunks", "total_documents", "total_job_chunks", @@ -128,6 +179,27 @@ "webhook_delivery_failures_24h", } ), + "self_hosted_document_type_aggregate": _AGGREGATE_PROPERTY_NAMES + | frozenset( + { + "document_type", + "jobs_created_24h", + "jobs_done_24h", + "jobs_failed_24h", + "pages_processed_24h", + "success_rate_24h", + } + ), + "self_hosted_client_aggregate": _AGGREGATE_PROPERTY_NAMES + | frozenset( + { + "created_by_client", + "jobs_created_24h", + "jobs_done_24h", + "jobs_failed_24h", + "success_rate_24h", + } + ), } @@ -136,6 +208,83 @@ def get_allowed_telemetry_event_names() -> frozenset[str]: return frozenset(_EVENT_PROPERTY_NAMES.keys()) +def normalize_document_type(extension_or_filename: str | None) -> str: + """Map a filename or extension to an allowlisted document_type.""" + if not extension_or_filename: + return "other" + raw = extension_or_filename.strip().lower() + if not raw: + return "other" + # Accept either "pdf" or "report.pdf" / "path/report.PDF". + if "/" in raw or "\\" in raw or "." in raw: + suffix = PurePosixPath(raw.replace("\\", "/")).suffix + extension = suffix.lstrip(".") + else: + extension = raw.lstrip(".") + if not extension: + return "other" + if extension in _IMAGE_EXTENSIONS: + return "image" + if extension in _HTML_EXTENSIONS: + return "html" + if extension in DOCUMENT_TYPES: + return extension + return "other" + + +def normalize_client_name(raw: str | None) -> str: + """Map a created_by_client value to an allowlisted client name.""" + if raw is None: + return "other" + normalized = raw.strip().lower() + if normalized in CLIENT_NAMES: + return normalized + return "other" + + +def normalize_source_type(raw: str | None) -> str: + """Map a job source_type value to an allowlisted source_type.""" + if raw is None: + return "other" + normalized = raw.strip().lower() + if normalized == "direct_upload": + return "file" + if normalized in SOURCE_TYPES: + return normalized + return "other" + + +def compute_success_rate(done: int, failed: int) -> float: + """Return done / (done + failed) as a float in 0–1.""" + terminal = max(done, 0) + max(failed, 0) + if terminal == 0: + return 0.0 + return max(done, 0) / terminal + + +def count_to_bucket(count: int) -> str: + """Bucket a non-negative count into 0|1-10|11-100|100+.""" + value = max(count, 0) + for upper_bound, label in _COUNT_BUCKETS: + if value <= upper_bound: + return label + return "100+" + + +def uptime_seconds_to_bucket(uptime_seconds: float) -> str: + """Bucket process uptime for heartbeat events.""" + seconds = max(uptime_seconds, 0.0) + if seconds < 5 * 60: + return "0m-5m" + if seconds < 60 * 60: + return "5m-1h" + if seconds < 24 * 60 * 60: + return "1h-24h" + if seconds < 7 * 24 * 60 * 60: + return "24h-7d" + return "7d+" + + def build_instance_event_properties( config: TelemetryRuntimeConfig, *, diff --git a/packages/shared-python/shared/services/telemetry/runtime.py b/packages/shared-python/shared/services/telemetry/runtime.py index d1bba0ed8..184970cef 100644 --- a/packages/shared-python/shared/services/telemetry/runtime.py +++ b/packages/shared-python/shared/services/telemetry/runtime.py @@ -2,15 +2,108 @@ from __future__ import annotations +import asyncio +import time +from collections.abc import Awaitable, Callable +from contextlib import AbstractAsyncContextManager from pathlib import Path +from typing import Protocol from loguru import logger +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession from .client import TelemetryClient from .config import TelemetryRuntimeConfig, TelemetrySettings, build_telemetry_config -from .events import build_instance_event_properties +from .events import ( + build_base_event_properties, + build_instance_event_properties, + uptime_seconds_to_bucket, +) from .identity import get_or_create_installation_id +HealthProbe = Callable[[], Awaitable[bool]] + + +class DatabaseSessionFactory(Protocol): + """Factory for app-owned async database sessions.""" + + def __call__(self) -> AbstractAsyncContextManager[AsyncSession]: + """Return an async context manager yielding an AsyncSession.""" + raise NotImplementedError + + +class TelemetryHeartbeatSettings(TelemetrySettings, Protocol): + TELEMETRY_AGGREGATE_INTERVAL_SECONDS: int + + +class SelfHostedHeartbeatTelemetryRunner: + """Periodically emits real health heartbeats for anonymous telemetry.""" + + def __init__( + self, + *, + config: TelemetryRuntimeConfig, + telemetry_client: TelemetryClient, + settings: TelemetrySettings, + interval_seconds: int, + started_at_monotonic: float, + postgres_probe: HealthProbe | None = None, + redis_probe: HealthProbe | None = None, + ) -> None: + self._config = config + self._telemetry_client = telemetry_client + self._settings = settings + self._interval_seconds = interval_seconds + self._started_at_monotonic = started_at_monotonic + self._postgres_probe = postgres_probe + self._redis_probe = redis_probe + self._task: asyncio.Task[None] | None = None + + async def emit_once(self) -> None: + """Probe dependencies and capture one heartbeat event.""" + postgres_healthy = await _run_health_probe(self._postgres_probe, default=False) + redis_healthy = await _run_health_probe(self._redis_probe, default=False) + uptime_seconds = time.monotonic() - self._started_at_monotonic + self._telemetry_client.capture( + "self_hosted_instance_heartbeat", + build_instance_event_properties( + self._config, + api_standalone_mode_enabled=self._settings.API_STANDALONE_MODE_ENABLED, + billing_enabled=self._settings.BILLING_ENABLED, + api_healthy=True, + postgres_healthy=postgres_healthy, + redis_healthy=redis_healthy, + uptime_bucket=uptime_seconds_to_bucket(uptime_seconds), + ), + ) + + def start(self) -> None: + """Start the periodic heartbeat loop.""" + if self._task is not None: + return + self._task = asyncio.create_task( + self._run(), + name="self-hosted-heartbeat-telemetry", + ) + + async def stop(self) -> None: + """Stop the periodic heartbeat loop.""" + task = self._task + if task is None: + return + task.cancel() + await asyncio.gather(task, return_exceptions=True) + self._task = None + + async def _run(self) -> None: + while True: + await asyncio.sleep(self._interval_seconds) + try: + await self.emit_once() + except Exception as exc: + logger.warning(f"anonymous heartbeat telemetry failed: {exc}") + async def start_self_hosted_telemetry( settings: TelemetrySettings, @@ -70,11 +163,99 @@ async def start_self_hosted_telemetry( return telemetry_client, config +async def start_self_hosted_heartbeat_telemetry( + settings: TelemetryHeartbeatSettings, + *, + telemetry_client: TelemetryClient | None, + config: TelemetryRuntimeConfig | None, + started_at_monotonic: float | None = None, + postgres_probe: HealthProbe | None = None, + redis_probe: HealthProbe | None = None, +) -> SelfHostedHeartbeatTelemetryRunner | None: + """Start periodic real heartbeats when the base self-hosted client is active.""" + if telemetry_client is None or config is None: + return None + interval_seconds = max(settings.TELEMETRY_AGGREGATE_INTERVAL_SECONDS, 60) + runner = SelfHostedHeartbeatTelemetryRunner( + config=config, + telemetry_client=telemetry_client, + settings=settings, + interval_seconds=interval_seconds, + started_at_monotonic=( + time.monotonic() if started_at_monotonic is None else started_at_monotonic + ), + postgres_probe=postgres_probe, + redis_probe=redis_probe, + ) + try: + runner.start() + except Exception as exc: + logger.warning(f"anonymous heartbeat telemetry start failed: {exc}") + return None + logger.info("anonymous self-hosted heartbeat telemetry scheduled") + return runner + + +async def stop_self_hosted_heartbeat_telemetry( + runner: SelfHostedHeartbeatTelemetryRunner | None, +) -> None: + """Stop heartbeat telemetry if it was started.""" + if runner is None: + return + await runner.stop() + + async def stop_self_hosted_telemetry( telemetry_client: TelemetryClient | None, + *, + config: TelemetryRuntimeConfig | None = None, ) -> None: """Flush and stop anonymous self-hosted telemetry.""" if telemetry_client is None: return - telemetry_client.capture("self_hosted_instance_shutdown", {}) + shutdown_properties = ( + build_base_event_properties(config) if config is not None else {} + ) + telemetry_client.capture("self_hosted_instance_shutdown", shutdown_properties) await telemetry_client.stop() + + +def build_postgres_health_probe( + db_session_factory: DatabaseSessionFactory, +) -> HealthProbe: + """Return a probe that runs SELECT 1 against Postgres.""" + + async def _probe() -> bool: + try: + async with db_session_factory() as session: + result = await session.execute(text("SELECT 1")) + return result.scalar_one_or_none() == 1 + except Exception: + return False + + return _probe + + +def build_redis_health_probe(redis_ping: Callable[[], Awaitable[bool]]) -> HealthProbe: + """Return a probe that pings Redis through the provided callable.""" + + async def _probe() -> bool: + try: + return bool(await redis_ping()) + except Exception: + return False + + return _probe + + +async def _run_health_probe( + probe: HealthProbe | None, + *, + default: bool, +) -> bool: + if probe is None: + return default + try: + return bool(await probe()) + except Exception: + return False From 4361e45936e2bda92c315a6673ec6173ffacdc9c Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 13 Jul 2026 16:11:32 +0800 Subject: [PATCH 2/4] Potential fix for pull request finding 'CodeQL / Unused import' Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com> --- packages/shared-python/shared/services/telemetry/events.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/shared-python/shared/services/telemetry/events.py b/packages/shared-python/shared/services/telemetry/events.py index ff3ad8b90..44ce071cb 100644 --- a/packages/shared-python/shared/services/telemetry/events.py +++ b/packages/shared-python/shared/services/telemetry/events.py @@ -7,7 +7,7 @@ from pathlib import PurePosixPath from typing import TypeAlias, cast -from .config import SCHEMA_VERSION, TelemetryRuntimeConfig +from .config import TelemetryRuntimeConfig TelemetryPropertyValue: TypeAlias = str | int | float | bool | None TelemetryProperties: TypeAlias = dict[str, TelemetryPropertyValue] From b4a9403f418181a748089b5d98f8163f3b315f7d Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 13 Jul 2026 16:26:52 +0800 Subject: [PATCH 3/4] Rename telemetry events from self_hosted_* to oss_*. Co-authored-by: Cursor --- .../test_self_hosted_telemetry_contract.py | 60 +++++++++---------- .../0004-anonymous-self-hosted-telemetry.md | 22 +++---- .../shared/services/telemetry/aggregates.py | 18 +++--- .../shared/services/telemetry/events.py | 20 +++---- .../shared/services/telemetry/runtime.py | 8 +-- 5 files changed, 64 insertions(+), 64 deletions(-) diff --git a/apps/api/tests/contract/test_self_hosted_telemetry_contract.py b/apps/api/tests/contract/test_self_hosted_telemetry_contract.py index b974e7613..a58401f19 100644 --- a/apps/api/tests/contract/test_self_hosted_telemetry_contract.py +++ b/apps/api/tests/contract/test_self_hosted_telemetry_contract.py @@ -86,7 +86,7 @@ def test_explicit_installation_id_must_be_uuid(tmp_path: Path) -> None: def test_telemetry_properties_strip_unknown_and_non_scalar_values() -> None: properties = sanitize_event_properties( - "self_hosted_instance_heartbeat", + "oss_instance_heartbeat", { "app_version": "1.2.3", "api_healthy": True, @@ -105,13 +105,13 @@ def test_telemetry_properties_strip_unknown_and_non_scalar_values() -> None: def test_aggregate_event_names_are_allowed() -> None: assert { - "self_hosted_usage_aggregate", - "self_hosted_retrieval_aggregate", - "self_hosted_worker_aggregate", - "self_hosted_api_aggregate", - "self_hosted_provider_aggregate", - "self_hosted_document_type_aggregate", - "self_hosted_client_aggregate", + "oss_usage_aggregate", + "oss_retrieval_aggregate", + "oss_worker_aggregate", + "oss_api_aggregate", + "oss_provider_aggregate", + "oss_document_type_aggregate", + "oss_client_aggregate", }.issubset(get_allowed_telemetry_event_names()) @@ -142,7 +142,7 @@ def test_success_rate_and_count_buckets() -> None: def test_usage_and_document_type_properties_strip_sensitive_values() -> None: usage_properties = sanitize_event_properties( - "self_hosted_usage_aggregate", + "oss_usage_aggregate", { "app_version": "1.2.3", "window_seconds": 86_400, @@ -154,7 +154,7 @@ def test_usage_and_document_type_properties_strip_sensitive_values() -> None: }, ) document_type_properties = sanitize_event_properties( - "self_hosted_document_type_aggregate", + "oss_document_type_aggregate", { "document_type": "pdf", "jobs_created_24h": 1, @@ -163,7 +163,7 @@ def test_usage_and_document_type_properties_strip_sensitive_values() -> None: }, ) client_properties = sanitize_event_properties( - "self_hosted_client_aggregate", + "oss_client_aggregate", { "created_by_client": "cli", "jobs_created_24h": 1, @@ -217,7 +217,7 @@ async def redis_probe() -> bool: assert len(posthog_client.captured_events) == 1 captured = posthog_client.captured_events[0] - assert captured.event_name == "self_hosted_instance_heartbeat" + assert captured.event_name == "oss_instance_heartbeat" properties = cast(dict[str, object], captured.kwargs["properties"]) assert properties["api_healthy"] is True assert properties["postgres_healthy"] is True @@ -243,7 +243,7 @@ async def test_shutdown_includes_base_event_properties(tmp_path: Path) -> None: assert len(posthog_client.captured_events) == 1 captured = posthog_client.captured_events[0] - assert captured.event_name == "self_hosted_instance_shutdown" + assert captured.event_name == "oss_instance_shutdown" assert captured.kwargs["properties"] == { **build_base_event_properties(config), "$process_person_profile": False, @@ -252,7 +252,7 @@ async def test_shutdown_includes_base_event_properties(tmp_path: Path) -> None: def test_usage_aggregate_allowlist_includes_v2_keys() -> None: properties = sanitize_event_properties( - "self_hosted_usage_aggregate", + "oss_usage_aggregate", { "success_rate_24h": 1.0, "job_duration_p95_seconds_24h": 12.5, @@ -309,7 +309,7 @@ def test_self_hosted_telemetry_env_can_disable_and_override_key( def test_aggregate_properties_strip_sensitive_values() -> None: properties = sanitize_event_properties( - "self_hosted_usage_aggregate", + "oss_usage_aggregate", { "app_version": "1.2.3", "window_seconds": 86_400, @@ -361,8 +361,8 @@ async def test_api_aggregate_uses_interval_window_when_global_lock_unavailable( api_window_seconds=300, ) - assert set(properties) == {"self_hosted_api_aggregate"} - api_properties = properties["self_hosted_api_aggregate"] + assert set(properties) == {"oss_api_aggregate"} + api_properties = properties["oss_api_aggregate"] assert api_properties["window_seconds"] == 300 assert api_properties["api_requests_total"] == 1 assert api_properties["api_requests_2xx"] == 1 @@ -375,7 +375,7 @@ def test_telemetry_client_filters_posthog_sdk_properties_after_capture( sanitized_message = telemetry_client._sanitize_posthog_message( { - "event": "self_hosted_api_aggregate", + "event": "oss_api_aggregate", "properties": { "app_version": "1.2.3", "api_requests_total": 1, @@ -420,7 +420,7 @@ async def test_telemetry_client_sends_anonymous_posthog_capture( await telemetry_client.start() queued = telemetry_client.capture( - "self_hosted_instance_heartbeat", + "oss_instance_heartbeat", { "app_version": "1.2.3", "api_healthy": True, @@ -432,7 +432,7 @@ async def test_telemetry_client_sends_anonymous_posthog_capture( assert queued is True assert len(posthog_client.captured_events) == 1 captured_event = posthog_client.captured_events[0] - assert captured_event.event_name == "self_hosted_instance_heartbeat" + assert captured_event.event_name == "oss_instance_heartbeat" assert captured_event.kwargs["distinct_id"] == ( "550e8400-e29b-41d4-a716-446655440000" ) @@ -455,7 +455,7 @@ async def test_telemetry_client_sends_aggregate_events(tmp_path: Path) -> None: await telemetry_client.start() queued = telemetry_client.capture( - "self_hosted_api_aggregate", + "oss_api_aggregate", { "app_version": "1.2.3", "window_seconds": 86_400, @@ -469,7 +469,7 @@ async def test_telemetry_client_sends_aggregate_events(tmp_path: Path) -> None: assert queued is True assert len(posthog_client.captured_events) == 1 captured_event = posthog_client.captured_events[0] - assert captured_event.event_name == "self_hosted_api_aggregate" + assert captured_event.event_name == "oss_api_aggregate" assert captured_event.kwargs["properties"] == { "app_version": "1.2.3", "window_seconds": 86_400, @@ -516,7 +516,7 @@ async def test_telemetry_client_respects_batch_size(tmp_path: Path) -> None: await telemetry_client.start() for index in range(3): telemetry_client.capture( - "self_hosted_instance_heartbeat", + "oss_instance_heartbeat", { "app_version": f"1.2.{index}", }, @@ -524,9 +524,9 @@ async def test_telemetry_client_respects_batch_size(tmp_path: Path) -> None: await telemetry_client.stop() assert [event.event_name for event in posthog_client.captured_events] == [ - "self_hosted_instance_heartbeat", - "self_hosted_instance_heartbeat", - "self_hosted_instance_heartbeat", + "oss_instance_heartbeat", + "oss_instance_heartbeat", + "oss_instance_heartbeat", ] assert posthog_client.flush_count == 1 @@ -542,7 +542,7 @@ async def test_telemetry_client_flush_before_start_does_not_deadlock( ) telemetry_client.capture( - "self_hosted_instance_heartbeat", + "oss_instance_heartbeat", { "app_version": "1.2.3", }, @@ -550,7 +550,7 @@ async def test_telemetry_client_flush_before_start_does_not_deadlock( await telemetry_client.flush() await telemetry_client.start() telemetry_client.capture( - "self_hosted_instance_heartbeat", + "oss_instance_heartbeat", { "app_version": "1.2.4", }, @@ -570,7 +570,7 @@ async def test_telemetry_client_does_not_restart_after_stop(tmp_path: Path) -> N await telemetry_client.start() telemetry_client.capture( - "self_hosted_instance_heartbeat", + "oss_instance_heartbeat", { "app_version": "1.2.3", }, @@ -578,7 +578,7 @@ async def test_telemetry_client_does_not_restart_after_stop(tmp_path: Path) -> N await telemetry_client.stop() await telemetry_client.start() queued_after_stop = telemetry_client.capture( - "self_hosted_instance_heartbeat", + "oss_instance_heartbeat", { "app_version": "1.2.4", }, diff --git a/docs/adr/0004-anonymous-self-hosted-telemetry.md b/docs/adr/0004-anonymous-self-hosted-telemetry.md index bb51f2603..d13e257a9 100644 --- a/docs/adr/0004-anonymous-self-hosted-telemetry.md +++ b/docs/adr/0004-anonymous-self-hosted-telemetry.md @@ -40,20 +40,20 @@ events (cardinality); app version stays on base properties only. ### Event catalog -Keep the existing eight event names and add two: +Event names use the `oss_` prefix (not `self_hosted_`). Catalog: | Event | Role | | --- | --- | -| `self_hosted_instance_started` | Install boot | -| `self_hosted_instance_heartbeat` | Periodic liveness + health | -| `self_hosted_instance_shutdown` | Graceful stop (includes base props) | -| `self_hosted_usage_aggregate` | Fleet usage snapshot (24h window) | -| `self_hosted_retrieval_aggregate` | Retrieval activity | -| `self_hosted_worker_aggregate` | Worker backlog / completion | -| `self_hosted_api_aggregate` | In-process API request counters | -| `self_hosted_provider_aggregate` | Parse/retrieval provider activity | -| `self_hosted_document_type_aggregate` | Per allowlisted document type | -| `self_hosted_client_aggregate` | Per allowlisted created_by_client | +| `oss_instance_started` | Install boot | +| `oss_instance_heartbeat` | Periodic liveness + health | +| `oss_instance_shutdown` | Graceful stop (includes base props) | +| `oss_usage_aggregate` | Fleet usage snapshot (24h window) | +| `oss_retrieval_aggregate` | Retrieval activity | +| `oss_worker_aggregate` | Worker backlog / completion | +| `oss_api_aggregate` | In-process API request counters | +| `oss_provider_aggregate` | Parse/retrieval provider activity | +| `oss_document_type_aggregate` | Per allowlisted document type | +| `oss_client_aggregate` | Per allowlisted created_by_client | ### Allowlists diff --git a/packages/shared-python/shared/services/telemetry/aggregates.py b/packages/shared-python/shared/services/telemetry/aggregates.py index 80634ccc5..80dfe9ec7 100644 --- a/packages/shared-python/shared/services/telemetry/aggregates.py +++ b/packages/shared-python/shared/services/telemetry/aggregates.py @@ -145,7 +145,7 @@ async def collect_self_hosted_aggregate_event_captures( """Collect aggregate captures without including customer content.""" captures: list[tuple[str, TelemetryProperties]] = [ ( - "self_hosted_api_aggregate", + "oss_api_aggregate", _collect_api_aggregate( config, api_metrics.snapshot_and_reset(), @@ -160,13 +160,13 @@ async def collect_self_hosted_aggregate_event_captures( try: captures.append( ( - "self_hosted_usage_aggregate", + "oss_usage_aggregate", await _collect_usage_aggregate(session, config, window_seconds), ) ) captures.append( ( - "self_hosted_retrieval_aggregate", + "oss_retrieval_aggregate", await _collect_retrieval_aggregate( session, config, @@ -176,13 +176,13 @@ async def collect_self_hosted_aggregate_event_captures( ) captures.append( ( - "self_hosted_worker_aggregate", + "oss_worker_aggregate", await _collect_worker_aggregate(session, config, window_seconds), ) ) captures.append( ( - "self_hosted_provider_aggregate", + "oss_provider_aggregate", await _collect_provider_aggregate(session, config, window_seconds), ) ) @@ -191,13 +191,13 @@ async def collect_self_hosted_aggregate_event_captures( config, window_seconds, ): - captures.append(("self_hosted_document_type_aggregate", properties)) + captures.append(("oss_document_type_aggregate", properties)) for properties in await _collect_client_aggregates( session, config, window_seconds, ): - captures.append(("self_hosted_client_aggregate", properties)) + captures.append(("oss_client_aggregate", properties)) return captures finally: await _release_aggregate_advisory_lock(session) @@ -224,8 +224,8 @@ async def collect_self_hosted_aggregate_event_properties( # Keep first capture for singleton events; multi-row events are omitted # from this dict helper (use collect_self_hosted_aggregate_event_captures). if event_name in { - "self_hosted_document_type_aggregate", - "self_hosted_client_aggregate", + "oss_document_type_aggregate", + "oss_client_aggregate", }: continue event_properties[event_name] = properties diff --git a/packages/shared-python/shared/services/telemetry/events.py b/packages/shared-python/shared/services/telemetry/events.py index 44ce071cb..533266f76 100644 --- a/packages/shared-python/shared/services/telemetry/events.py +++ b/packages/shared-python/shared/services/telemetry/events.py @@ -89,8 +89,8 @@ ) _EVENT_PROPERTY_NAMES: dict[str, frozenset[str]] = { - "self_hosted_instance_started": frozenset(), - "self_hosted_instance_heartbeat": frozenset( + "oss_instance_started": frozenset(), + "oss_instance_heartbeat": frozenset( { "api_healthy", "postgres_healthy", @@ -98,8 +98,8 @@ "uptime_bucket", } ), - "self_hosted_instance_shutdown": frozenset(), - "self_hosted_usage_aggregate": _AGGREGATE_PROPERTY_NAMES + "oss_instance_shutdown": frozenset(), + "oss_usage_aggregate": _AGGREGATE_PROPERTY_NAMES | frozenset( { "active_api_keys", @@ -126,7 +126,7 @@ "total_users", } ), - "self_hosted_retrieval_aggregate": _AGGREGATE_PROPERTY_NAMES + "oss_retrieval_aggregate": _AGGREGATE_PROPERTY_NAMES | frozenset( { "retrieval_cache_hits_24h", @@ -140,7 +140,7 @@ "retrieval_tokens_24h", } ), - "self_hosted_worker_aggregate": _AGGREGATE_PROPERTY_NAMES + "oss_worker_aggregate": _AGGREGATE_PROPERTY_NAMES | frozenset( { "job_duration_avg_seconds_24h", @@ -152,7 +152,7 @@ "jobs_waiting_file", } ), - "self_hosted_api_aggregate": _AGGREGATE_PROPERTY_NAMES + "oss_api_aggregate": _AGGREGATE_PROPERTY_NAMES | frozenset( { "api_latency_avg_ms", @@ -164,7 +164,7 @@ "api_requests_total", } ), - "self_hosted_provider_aggregate": _AGGREGATE_PROPERTY_NAMES + "oss_provider_aggregate": _AGGREGATE_PROPERTY_NAMES | frozenset( { "parse_agent_errors_24h", @@ -179,7 +179,7 @@ "webhook_delivery_failures_24h", } ), - "self_hosted_document_type_aggregate": _AGGREGATE_PROPERTY_NAMES + "oss_document_type_aggregate": _AGGREGATE_PROPERTY_NAMES | frozenset( { "document_type", @@ -190,7 +190,7 @@ "success_rate_24h", } ), - "self_hosted_client_aggregate": _AGGREGATE_PROPERTY_NAMES + "oss_client_aggregate": _AGGREGATE_PROPERTY_NAMES | frozenset( { "created_by_client", diff --git a/packages/shared-python/shared/services/telemetry/runtime.py b/packages/shared-python/shared/services/telemetry/runtime.py index 184970cef..5a63fab4a 100644 --- a/packages/shared-python/shared/services/telemetry/runtime.py +++ b/packages/shared-python/shared/services/telemetry/runtime.py @@ -66,7 +66,7 @@ async def emit_once(self) -> None: redis_healthy = await _run_health_probe(self._redis_probe, default=False) uptime_seconds = time.monotonic() - self._started_at_monotonic self._telemetry_client.capture( - "self_hosted_instance_heartbeat", + "oss_instance_heartbeat", build_instance_event_properties( self._config, api_standalone_mode_enabled=self._settings.API_STANDALONE_MODE_ENABLED, @@ -145,9 +145,9 @@ async def start_self_hosted_telemetry( api_standalone_mode_enabled=settings.API_STANDALONE_MODE_ENABLED, billing_enabled=settings.BILLING_ENABLED, ) - telemetry_client.capture("self_hosted_instance_started", base_properties) + telemetry_client.capture("oss_instance_started", base_properties) telemetry_client.capture( - "self_hosted_instance_heartbeat", + "oss_instance_heartbeat", build_instance_event_properties( config, api_standalone_mode_enabled=settings.API_STANDALONE_MODE_ENABLED, @@ -216,7 +216,7 @@ async def stop_self_hosted_telemetry( shutdown_properties = ( build_base_event_properties(config) if config is not None else {} ) - telemetry_client.capture("self_hosted_instance_shutdown", shutdown_properties) + telemetry_client.capture("oss_instance_shutdown", shutdown_properties) await telemetry_client.stop() From d6e3f42393d737929cef64cf013b9adea18f9c84 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 13 Jul 2026 17:19:38 +0800 Subject: [PATCH 4/4] Fix telemetry contract test SCHEMA_VERSION import. Co-authored-by: Cursor --- apps/api/tests/contract/test_self_hosted_telemetry_contract.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/apps/api/tests/contract/test_self_hosted_telemetry_contract.py b/apps/api/tests/contract/test_self_hosted_telemetry_contract.py index a58401f19..69e352ae9 100644 --- a/apps/api/tests/contract/test_self_hosted_telemetry_contract.py +++ b/apps/api/tests/contract/test_self_hosted_telemetry_contract.py @@ -17,7 +17,7 @@ BaseConfig, ) from shared.services.telemetry.client import TelemetryClient -from shared.services.telemetry.config import TelemetryRuntimeConfig +from shared.services.telemetry.config import SCHEMA_VERSION, TelemetryRuntimeConfig from shared.services.telemetry.api_metrics import ApiRequestTelemetryMetrics from shared.services.telemetry.aggregates import ( TelemetryAggregateSettings, @@ -26,7 +26,6 @@ stop_self_hosted_aggregate_telemetry, ) from shared.services.telemetry.events import ( - SCHEMA_VERSION, build_base_event_properties, compute_success_rate, count_to_bucket,