From bba63d87862fbc8ad668a94f9e704ee72a088262 Mon Sep 17 00:00:00 2001 From: Alex Fournier Date: Thu, 23 Jul 2026 14:01:41 -0700 Subject: [PATCH] fix(telemetry): prune expired local metrics Signed-off-by: Alex Fournier --- cli-config.yaml.example | 2 + docs/observability/relay-shared-metrics.md | 4 +- hermes_cli/observability/shared_metrics.py | 77 +++++++++++++++++- tests/hermes_cli/test_relay_shared_metrics.py | 78 +++++++++++++++++++ 4 files changed, 159 insertions(+), 2 deletions(-) diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 0821bcfcd2b..0140c566ed6 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1410,6 +1410,8 @@ display: # packages under $HERMES_HOME/telemetry/shared_metrics; it does not upload them. # Packages include a random profile-scoped ID that stays stable until this # directory is deleted. It is not derived from hardware, account, or host data. +# Successfully exported local history is retained for 30 days; pending deltas +# are retained until they can be exported. telemetry: shared_metrics: enabled: false diff --git a/docs/observability/relay-shared-metrics.md b/docs/observability/relay-shared-metrics.md index b9dd9bd15a6..ed9a77bd9e8 100644 --- a/docs/observability/relay-shared-metrics.md +++ b/docs/observability/relay-shared-metrics.md @@ -72,7 +72,9 @@ $HERMES_HOME/telemetry/shared_metrics/outbox/*.json The database keeps transactional aggregate and package-outbox state. Package files are immutable delta documents that conform to a closed JSON schema and -are written with atomic replacement. +are written with atomic replacement. Fully packaged aggregate rows and +successfully exported package rows and files are retained locally for 30 days. +Pending package rows and counters with unexported deltas are never pruned. Each package contains an `install_id` generated as a random UUID. Despite the schema field name, its current scope is one `HERMES_HOME`, so it is more diff --git a/hermes_cli/observability/shared_metrics.py b/hermes_cli/observability/shared_metrics.py index 0c4912a0a80..506fb4ab131 100644 --- a/hermes_cli/observability/shared_metrics.py +++ b/hermes_cli/observability/shared_metrics.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import logging import sqlite3 import uuid from collections.abc import Iterator @@ -26,6 +27,9 @@ _PACKAGE_SCHEMA_VERSION = "hermes.shared_metrics.v1" _STORE_SCHEMA_VERSION = "1" _BUSY_TIMEOUT_MS = 250 _SCHEMA_BUSY_TIMEOUT_MS = 5_000 +_LOCAL_HISTORY_RETENTION_DAYS = 30 + +logger = logging.getLogger(__name__) def _utc_now() -> datetime: @@ -110,7 +114,15 @@ class SharedMetricsStore: for _ in range(pending_periods): if self._create_package() is None: break - return self._export_pending_packages() + exported = self._export_pending_packages() + try: + self._prune_expired_history() + except Exception: + logger.warning( + "Unable to prune expired shared-metrics history", + exc_info=True, + ) + return exported def counter_snapshot(self) -> list[dict[str, Any]]: """Return cumulative counters for focused tests and local inspection.""" @@ -409,3 +421,66 @@ class SharedMetricsStore: ) exported.append(path) return exported + + def _prune_expired_history(self, *, now: datetime | None = None) -> None: + """Remove exported local history after the bounded retention window.""" + cutoff = (now or _utc_now()) - timedelta( + days=_LOCAL_HISTORY_RETENTION_DAYS + ) + cutoff_timestamp = _isoformat(cutoff) + cutoff_period = cutoff.date().isoformat() + with self._connection() as connection: + rows = connection.execute( + """ + SELECT package_id + FROM package_outbox + WHERE exported_at IS NOT NULL + AND exported_at < ? + ORDER BY exported_at, package_id + """, + (cutoff_timestamp,), + ).fetchall() + + removable_package_ids: list[str] = [] + for row in rows: + package_id = str(row["package_id"]) + try: + (self.outbox_directory / f"{package_id}.json").unlink( + missing_ok=True + ) + except OSError: + logger.warning( + "Unable to prune expired shared-metrics package %s", + package_id, + exc_info=True, + ) + continue + removable_package_ids.append(package_id) + + with self._connection() as connection: + with write_txn(connection): + for package_id in removable_package_ids: + connection.execute( + """ + DELETE FROM package_outbox + WHERE package_id = ? + AND exported_at IS NOT NULL + AND exported_at < ? + """, + (package_id, cutoff_timestamp), + ) + connection.execute( + """ + DELETE FROM counter_aggregates + WHERE period_start < ? + AND value = packaged_value + AND NOT EXISTS ( + SELECT 1 + FROM package_outbox + WHERE exported_at IS NULL + AND substr(package_outbox.period_start, 1, 10) + = counter_aggregates.period_start + ) + """, + (cutoff_period,), + ) diff --git a/tests/hermes_cli/test_relay_shared_metrics.py b/tests/hermes_cli/test_relay_shared_metrics.py index 6dfb11eec9c..058e07db7bc 100644 --- a/tests/hermes_cli/test_relay_shared_metrics.py +++ b/tests/hermes_cli/test_relay_shared_metrics.py @@ -12,6 +12,7 @@ import time import uuid from concurrent.futures import ThreadPoolExecutor from copy import deepcopy +from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace from typing import Any @@ -586,6 +587,83 @@ def test_pending_package_retry_reuses_the_same_package_and_file(tmp_path): assert list(outbox_directory.glob("*.json")) == [package_path] +def test_retention_prunes_only_expired_exported_history(tmp_path): + database_path = tmp_path / "metrics.sqlite3" + outbox_directory = tmp_path / "outbox" + store = SharedMetricsStore(database_path, outbox_directory) + + store.record_model_call(_dimensions(), "expired-version") + [expired_path] = store.create_and_export_package() + store.record_model_call(_dimensions(), "current-version") + [current_path] = store.create_and_export_package() + store.record_model_call(_dimensions(), "pending-version") + pending_package = store._create_package() + assert pending_package is not None + + with sqlite3.connect(database_path) as connection: + connection.execute( + """ + UPDATE counter_aggregates + SET period_start = '2026-05-01' + WHERE hermes_version = 'expired-version' + """ + ) + connection.execute( + """ + UPDATE package_outbox + SET period_start = '2026-05-01T00:00:00Z', + period_end = '2026-05-02T00:00:00Z', + exported_at = '2026-05-02T00:00:00Z' + WHERE package_id = ? + """, + (expired_path.stem,), + ) + + store._prune_expired_history( + now=datetime(2026, 7, 23, tzinfo=timezone.utc) + ) + + assert not expired_path.exists() + assert current_path.exists() + assert not (outbox_directory / f"{pending_package['package_id']}.json").exists() + with sqlite3.connect(database_path) as connection: + outbox_rows = connection.execute( + "SELECT package_id, exported_at FROM package_outbox ORDER BY package_id" + ).fetchall() + aggregate_versions = { + row[0] + for row in connection.execute( + "SELECT hermes_version FROM counter_aggregates" + ).fetchall() + } + assert {row[0] for row in outbox_rows} == { + current_path.stem, + pending_package["package_id"], + } + assert next( + row[1] for row in outbox_rows if row[0] == pending_package["package_id"] + ) is None + assert aggregate_versions == {"current-version", "pending-version"} + + +def test_retention_failure_does_not_fail_a_committed_export(tmp_path, monkeypatch): + store = SharedMetricsStore( + tmp_path / "metrics.sqlite3", + tmp_path / "outbox", + ) + store.record_model_call(_dimensions(), "test-version") + + def fail_pruning(): + raise OSError("retention unavailable") + + monkeypatch.setattr(store, "_prune_expired_history", fail_pruning) + + [package_path] = store.create_and_export_package() + + assert package_path.exists() + assert store.counter_snapshot()[0]["packaged_value"] == 1 + + def test_file_export_failure_retries_committed_outbox_without_duplicate_delta( tmp_path, monkeypatch ):