From 106e1b8d175aac63b57a3f728e4d07597422363e Mon Sep 17 00:00:00 2001 From: Divyam Talwar Date: Sat, 4 Jul 2026 01:08:01 +0530 Subject: [PATCH] Bound telemetry growth during upload outages Enforce the configured byte budget for logs/telemetry when upload attempts fail, while preserving the existing successful-upload cleanup order across logs/. Keep the cleanup pass best-effort so filesystem pruning errors do not stop future uploader retries. Constraint: Failed-upload cleanup must not become a surprise global log cleanup pass for self-hosted operators. Rejected: Run the existing global logs/ budget sweep on every uploader cycle | that would delete unrelated logs during proxy outages and broaden the PR beyond telemetry retention. Confidence: high Scope-risk: narrow Directive: Keep failure-path cleanup scoped to telemetry files unless a future PR deliberately changes the documented log-retention policy. Tested: .venv/bin/python -m pytest core/tests/unit/test_log_uploader_retention.py -q; uv run ty check core/services/log_uploader.py core/tests/unit/test_log_uploader_retention.py; .venv/bin/python -m py_compile core/services/log_uploader.py core/tests/unit/test_log_uploader_retention.py; .venv/bin/markdown-it docs/telemetry.md >/tmp/telemetry-pr11.html; git diff --check Not-tested: Full core/tests/unit run; it requires dummy POSTGRES_URI and stalled in ColPali model artifact fetching for over 1 minute. Ruff check/format unavailable because ruff is not installed in this environment. --- core/services/log_uploader.py | 56 +++++--- .../tests/unit/test_log_uploader_retention.py | 129 ++++++++++++++++++ docs/telemetry.md | 14 ++ 3 files changed, 182 insertions(+), 17 deletions(-) create mode 100644 core/tests/unit/test_log_uploader_retention.py diff --git a/core/services/log_uploader.py b/core/services/log_uploader.py index f4874329..d5ce8be2 100644 --- a/core/services/log_uploader.py +++ b/core/services/log_uploader.py @@ -101,21 +101,30 @@ def _run_forever(self) -> None: def _upload_cycle(self) -> None: with self._lock: - telemetry_paths = self._gather_telemetry_files() - if not telemetry_paths: - LOGGER.debug("No telemetry events to upload") - return - - bundle = self._build_telemetry_bundle(telemetry_paths) - if not bundle: - LOGGER.debug("Telemetry bundle contained no events") - return - - if self._post_usage_payload(bundle): - self._truncate_files(telemetry_paths) - self._enforce_local_budget() - else: - LOGGER.warning("Telemetry proxy upload failed; retaining local files") + telemetry_paths: Sequence[Path] = () + enforce_budget = False + budget_root = self.log_dir + try: + telemetry_paths = self._gather_telemetry_files() + if not telemetry_paths: + LOGGER.debug("No telemetry events to upload") + return + + bundle = self._build_telemetry_bundle(telemetry_paths) + if not bundle: + LOGGER.debug("Telemetry bundle contained no events") + return + + if self._post_usage_payload(bundle): + enforce_budget = True + self._truncate_files(telemetry_paths) + else: + enforce_budget = True + budget_root = self.telemetry_dir + LOGGER.warning("Telemetry proxy upload failed; enforcing telemetry log budget before retry") + finally: + if enforce_budget: + self._enforce_local_budget_safely(root=budget_root) def _gather_telemetry_files(self) -> List[Path]: files: List[Path] = [] @@ -271,18 +280,28 @@ def _truncate_files(self, paths: Sequence[Path]) -> None: except OSError as exc: LOGGER.warning("Unable to truncate %s: %s", path, exc) - def _enforce_local_budget(self) -> None: + def _enforce_local_budget_safely(self, *, root: Optional[Path] = None) -> None: + try: + self._enforce_local_budget(root=root) + except Exception as exc: + LOGGER.warning("Unable to enforce telemetry log budget: %s", exc, exc_info=True) + + def _enforce_local_budget(self, *, root: Optional[Path] = None) -> None: if self.max_local_bytes <= 0: return + search_root = Path(root) if root is not None else self.log_dir files: List[tuple[Path, int, float]] = [] total = 0 - for path in self.log_dir.rglob("*"): + for path in search_root.rglob("*"): if not path.is_file(): continue try: stat = path.stat() except FileNotFoundError: continue + except OSError as exc: + LOGGER.warning("Unable to stat %s while enforcing log budget: %s", path, exc) + continue total += stat.st_size files.append((path, stat.st_size, stat.st_mtime)) if total <= self.max_local_bytes: @@ -293,6 +312,9 @@ def _enforce_local_budget(self) -> None: path.unlink() except FileNotFoundError: continue + except OSError as exc: + LOGGER.warning("Unable to remove %s while enforcing log budget: %s", path, exc) + continue LOGGER.warning("Removed %s to enforce %s-byte log budget", path, self.max_local_bytes) total -= size if total <= self.max_local_bytes: diff --git a/core/tests/unit/test_log_uploader_retention.py b/core/tests/unit/test_log_uploader_retention.py new file mode 100644 index 00000000..25c92aa6 --- /dev/null +++ b/core/tests/unit/test_log_uploader_retention.py @@ -0,0 +1,129 @@ +import os +import threading +from pathlib import Path +from typing import Optional + +import pytest + +from core.services.log_uploader import LogUploader + +pytestmark = pytest.mark.unit + + +def _make_uploader(tmp_path: Path, *, max_local_bytes: int) -> LogUploader: + uploader = object.__new__(LogUploader) + uploader.log_dir = tmp_path + uploader.telemetry_dir = tmp_path / "telemetry" + uploader.telemetry_dir.mkdir(parents=True, exist_ok=True) + uploader.max_local_bytes = max_local_bytes + uploader._lock = threading.Lock() + return uploader + + +def _write_file(path: Path, content: str, *, mtime: float) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + os.utime(path, (mtime, mtime)) + return path + + +def _event_line(operation: str = "query") -> str: + return f'{{"timestamp":"2026-01-01T00:00:00+00:00","worker_pid":123,"operation":"{operation}"}}\n' + + +def test_upload_failure_enforces_budget_only_for_telemetry_files(tmp_path, monkeypatch): + uploader = _make_uploader(tmp_path, max_local_bytes=len(_event_line("newer"))) + app_log = _write_file(tmp_path / "app.log", "x" * 80, mtime=1) + old_telemetry = _write_file( + tmp_path / "telemetry" / "usage_events_worker_1.jsonl", + _event_line("older"), + mtime=2, + ) + new_telemetry = _write_file( + tmp_path / "telemetry" / "usage_events_worker_2.jsonl", + _event_line("newer"), + mtime=3, + ) + monkeypatch.setattr(LogUploader, "_post_usage_payload", lambda _self, _bundle: False) + + uploader._upload_cycle() + + assert app_log.exists() + assert not old_telemetry.exists() + assert new_telemetry.exists() + assert new_telemetry.read_text(encoding="utf-8") == _event_line("newer") + + +def test_success_truncates_uploaded_files_before_global_budget_enforcement(tmp_path, monkeypatch): + uploader = _make_uploader(tmp_path, max_local_bytes=5) + stale_log = _write_file(tmp_path / "app.log", "x" * 80, mtime=1) + telemetry_file = _write_file(tmp_path / "telemetry" / "usage_events_worker_1.jsonl", _event_line(), mtime=2) + retained_log = _write_file(tmp_path / "recent.log", "fresh", mtime=3) + monkeypatch.setattr(LogUploader, "_post_usage_payload", lambda _self, _bundle: True) + + uploader._upload_cycle() + + assert telemetry_file.exists() + assert telemetry_file.read_text(encoding="utf-8") == "" + assert not stale_log.exists() + assert retained_log.exists() + assert retained_log.read_text(encoding="utf-8") == "fresh" + + +def test_cycle_without_telemetry_files_does_not_run_global_budget(tmp_path): + uploader = _make_uploader(tmp_path, max_local_bytes=5) + stale_log = _write_file(tmp_path / "app.log", "x" * 40, mtime=1) + + uploader._upload_cycle() + + assert stale_log.exists() + assert stale_log.read_text(encoding="utf-8") == "x" * 40 + + +def test_empty_bundle_does_not_run_budget_cleanup(tmp_path, monkeypatch): + uploader = _make_uploader(tmp_path, max_local_bytes=5) + app_log = _write_file(tmp_path / "app.log", "x" * 80, mtime=1) + old_telemetry = _write_file(tmp_path / "telemetry" / "usage_events_worker_1.jsonl", _event_line(), mtime=2) + telemetry_file = tmp_path / "telemetry" / "usage_events_worker_2.jsonl" + + monkeypatch.setattr(LogUploader, "_gather_telemetry_files", lambda _self: [old_telemetry, telemetry_file]) + monkeypatch.setattr(LogUploader, "_build_telemetry_bundle", lambda _self, _paths: None) + monkeypatch.setattr( + LogUploader, + "_post_usage_payload", + lambda _self, _bundle: pytest.fail("empty bundles must not be uploaded"), + ) + + uploader._upload_cycle() + + assert app_log.exists() + assert old_telemetry.exists() + assert old_telemetry.read_text(encoding="utf-8") == _event_line() + + +def test_cleanup_failure_does_not_escape_upload_cycle(tmp_path, monkeypatch, caplog): + uploader = _make_uploader(tmp_path, max_local_bytes=1) + _write_file(tmp_path / "telemetry" / "usage_events_worker_1.jsonl", _event_line(), mtime=1) + monkeypatch.setattr(LogUploader, "_post_usage_payload", lambda _self, _bundle: False) + + def fail_budget(_self: LogUploader, *, root: Optional[Path] = None) -> None: + raise OSError("permission denied") + + monkeypatch.setattr(LogUploader, "_enforce_local_budget", fail_budget) + + uploader._upload_cycle() + + assert "Unable to enforce telemetry log budget" in caplog.text + + +def test_zero_budget_does_not_delete_retained_files_on_upload_failure(tmp_path, monkeypatch): + uploader = _make_uploader(tmp_path, max_local_bytes=0) + stale_log = _write_file(tmp_path / "app.log", "x" * 80, mtime=1) + telemetry_file = _write_file(tmp_path / "telemetry" / "usage_events_worker_1.jsonl", _event_line(), mtime=2) + monkeypatch.setattr(LogUploader, "_post_usage_payload", lambda _self, _bundle: False) + + uploader._upload_cycle() + + assert stale_log.exists() + assert telemetry_file.exists() + assert telemetry_file.read_text(encoding="utf-8") == _event_line() diff --git a/docs/telemetry.md b/docs/telemetry.md index f90e7bf7..c81935f5 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -3,3 +3,17 @@ Morphik logs minimal operational metadata (operation name, status, duration, token counts) to `logs/telemetry/` so we can keep deployments healthy, then periodically uploads those JSONL files to `https://logs.morphik.ai` to avoid unbounded disk usage. Telemetry is enabled by default; set `TELEMETRY=false` in the environment if you need to disable it locally, and contact founders@morphik.ai for additional compliance questions. + +## Local retention + +Self-hosted deployments can configure `telemetry.max_local_bytes` in `morphik.toml` to bound local log +storage during telemetry uploader cycles. The retention scope depends on the uploader outcome: + +- Successful uploads: Morphik truncates uploaded telemetry files, then enforces the configured budget + across `logs/`, preserving the existing success-path cleanup behavior. +- Failed uploads: Morphik retries later and enforces the same byte budget only inside `logs/telemetry/`. + This prevents proxy outages from growing local telemetry indefinitely without deleting unrelated + application logs. Local telemetry files may be pruned oldest-first before they are uploaded if + `logs/telemetry/` exceeds the configured budget. +- No events to upload: cycles with no telemetry files, or files that produce no uploadable bundle, do not + run pruning.