Send the daily analytics report from the main process

frigate/analytics/reporter.py

@@ -0,0 +1,149 @@
+"""Send one analytics report a day while the admin has opted in."""
+
+import logging
+import random
+import threading
+import time
+from collections.abc import Callable
+from multiprocessing.synchronize import Event as MpEvent
+from typing import TYPE_CHECKING
+
+from frigate.analytics.context import ReportContext
+from frigate.analytics.report import build_report
+from frigate.analytics.state import (
+    STATE_PATH,
+    AnalyticsState,
+    delete_state,
+    load_state,
+    new_state,
+    save_state,
+)
+from frigate.analytics.transport import SendOutcome, send_report
+from frigate.config.holder import ConfigHolder
+from frigate.const import ANALYTICS_URL
+from frigate.notices import raise_notice, resolve_notice
+
+if TYPE_CHECKING:
+    from frigate.notices.registry import NoticeRegistry
+    from frigate.stats.emitter import StatsEmitter
+
+logger = logging.getLogger(__name__)
+
+WAKE_S = 10 * 60
+INTERVAL_S = 24 * 60 * 60
+JITTER_S = 60 * 60
+FIRST_DELAY_S = (15 * 60, 45 * 60)
+PROMPT_KIND = "analytics_prompt"
+
+
+class AnalyticsReporter(threading.Thread):
+    """Reads the live config on every wake, so a settings save needs no restart."""
+
+    def __init__(
+        self,
+        config_holder: ConfigHolder,
+        stats_emitter: "StatsEmitter",
+        notice_registry: "NoticeRegistry",
+        stop_event: MpEvent | threading.Event,
+        *,
+        state_path: str = STATE_PATH,
+        url: str = ANALYTICS_URL,
+        send: Callable[[str, str], SendOutcome] = send_report,
+        clock: Callable[[], float] = time.time,
+        rng: random.Random | None = None,
+    ) -> None:
+        super().__init__(name="analytics_reporter", daemon=True)
+        self.config_holder = config_holder
+        self.stats_emitter = stats_emitter
+        self.notice_registry = notice_registry
+        self.stop_event = stop_event
+        self.state_path = state_path
+        self.url = url
+        self.send = send
+        self.clock = clock
+        self.rng = rng or random.Random()
+        self.first_due = clock() + self.rng.uniform(*FIRST_DELAY_S)
+        self.interval = self._next_interval()
+        self.opted_in: bool | None = None
+        self.warned_unwritable = False
+
+    def _next_interval(self) -> float:
+        return INTERVAL_S + self.rng.uniform(-JITTER_S, JITTER_S)
+
+    def run(self) -> None:
+        while True:
+            try:
+                self.tick()
+            except Exception:
+                logger.exception("Analytics reporter failed")
+
+            if self.stop_event.wait(WAKE_S):
+                break
+
+    def tick(self) -> None:
+        config = self.config_holder.config
+
+        # safe mode parses a default config where analytics reads as off, and
+        # handling that as an opt-out would delete the install ID
+        if config.safe_mode:
+            return
+
+        opted_in = config.telemetry.analytics
+
+        if opted_in != self.opted_in:
+            self.opted_in = opted_in
+
+            if opted_in:
+                resolve_notice(PROMPT_KIND)
+            else:
+                raise_notice(PROMPT_KIND)
+                delete_state(self.state_path)
+
+        if not opted_in:
+            return
+
+        now = self.clock()
+
+        if now < self.first_due:
+            return
+
+        state = load_state(self.state_path) or new_state()
+
+        if not self._due(state.last_attempt_at, now):
+            return
+
+        # saved before sending, so a failing endpoint or a crash mid-send still
+        # waits a full interval; without saved state every boot would send
+        # under a new install ID
+        if not save_state(AnalyticsState(state.install_id, now), self.state_path):
+            if not self.warned_unwritable:
+                logger.warning(
+                    "Analytics is on, but %s can't be written, so no report is sent",
+                    self.state_path,
+                )
+                self.warned_unwritable = True
+
+            return
+
+        self.interval = self._next_interval()
+        self._send(state.install_id, now)
+
+    def _due(self, last_attempt_at: float, now: float) -> bool:
+        # a last attempt stamped in the future came from a wrong clock
+        return (
+            now >= last_attempt_at + self.interval or last_attempt_at > now + INTERVAL_S
+        )
+
+    def _send(self, install_id: str, now: float) -> None:
+        notice_stats = self.notice_registry.stats()
+        ctx = ReportContext(
+            config=self.config_holder.config,
+            stats=self.stats_emitter.get_latest_stats(),
+            notice_stats=notice_stats,
+        )
+        body = build_report(ctx, install_id, sent_at=int(now)).model_dump_json()
+
+        if self.send(self.url, body) is SendOutcome.accepted:
+            self.notice_registry.mark_reported(notice_stats)
+            logger.info("Sent the daily analytics report")
+            logger.debug("Analytics report: %s", body)

frigate/app.py

@@ -15,6 +15,7 @@ import uvicorn
 from peewee_migrate import Router
 from playhouse.sqlite_ext import SqliteExtDatabase
 
+from frigate.analytics.reporter import AnalyticsReporter
 from frigate.api.auth import hash_password
 from frigate.api.fastapi_app import create_fastapi_app
 from frigate.camera import CameraMetrics, PTZMetrics
@@ -529,6 +530,15 @@ class FrigateApp:
         )
         self.stats_emitter.start()
 
+    def start_analytics_reporter(self) -> None:
+        self.analytics_reporter = AnalyticsReporter(
+            self.config_holder,
+            self.stats_emitter,
+            self.notice_registry,
+            self.stop_event,
+        )
+        self.analytics_reporter.start()
+
     def start_watchdog(self) -> None:
         self.frigate_watchdog = FrigateWatchdog(self.detectors, self.stop_event)
 
@@ -680,6 +690,7 @@ class FrigateApp:
         self.start_audio_processor()
         self.start_storage_maintainer()
         self.start_stats_emitter()
+        self.start_analytics_reporter()
         self.start_timeline_processor()
         self.start_event_processor()
         self.start_event_cleanup()
@@ -779,6 +790,8 @@ class FrigateApp:
         self.event_cleanup.join()
         self.record_cleanup.join()
         self.stats_emitter.join()
+        # a send in flight can hold the thread for the whole request timeout
+        self.analytics_reporter.join(timeout=5)
         self.frigate_watchdog.join()
         self.camera_maintainer.join()
         self.db.stop()

frigate/test/test_analytics_reporter.py

@@ -0,0 +1,192 @@
+"""Tests for the analytics reporter's schedule and consent handling."""
+
+import os
+import tempfile
+import threading
+import unittest
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from frigate.analytics import reporter as reporter_module
+from frigate.analytics.reporter import (
+    FIRST_DELAY_S,
+    INTERVAL_S,
+    JITTER_S,
+    PROMPT_KIND,
+    WAKE_S,
+    AnalyticsReporter,
+)
+from frigate.analytics.state import AnalyticsState, load_state, save_state
+from frigate.analytics.transport import SendOutcome
+from frigate.test.analytics_helpers import make_config
+
+SNAPSHOT = [
+    {
+        "kind": "detector_stuck",
+        "occurrences": 3,
+        "dismissals": 0,
+        "reported_occurrences": 1,
+        "reported_dismissals": 0,
+    }
+]
+
+
+class TestAnalyticsReporter(unittest.TestCase):
+    def setUp(self):
+        self.dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.dir.cleanup)
+        self.path = os.path.join(self.dir.name, ".analytics.json")
+        self.now = 1_790_000_000.0
+        self.send = Mock(return_value=SendOutcome.accepted)
+        self.registry = Mock()
+        self.registry.stats.return_value = SNAPSHOT
+        self.stats = Mock()
+        self.stats.get_latest_stats.return_value = {}
+        self.holder = SimpleNamespace(
+            config=make_config({"telemetry": {"analytics": True}})
+        )
+
+        for name in ("raise_notice", "resolve_notice"):
+            patcher = patch.object(reporter_module, name)
+            setattr(self, name, patcher.start())
+            self.addCleanup(patcher.stop)
+
+        built = Mock()
+        built.model_dump_json.return_value = '{"report": 1}'
+        patcher = patch.object(reporter_module, "build_report", return_value=built)
+        self.build_report = patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def reporter(self, path: str | None = None) -> AnalyticsReporter:
+        # uniform(a, b) returns a: the shortest startup delay and interval
+        rng = Mock()
+        rng.uniform.side_effect = lambda low, high: low
+
+        return AnalyticsReporter(
+            self.holder,
+            self.stats,
+            self.registry,
+            threading.Event(),
+            state_path=path or self.path,
+            url="https://example.test/report",
+            send=self.send,
+            clock=lambda: self.now,
+            rng=rng,
+        )
+
+    def advance(self, seconds: float) -> None:
+        self.now += seconds
+
+    def test_waits_for_the_startup_delay_then_sends(self):
+        reporter = self.reporter()
+
+        reporter.tick()
+        self.send.assert_not_called()
+
+        self.advance(FIRST_DELAY_S[0])
+        reporter.tick()
+
+        self.send.assert_called_once_with(
+            "https://example.test/report", '{"report": 1}'
+        )
+        self.registry.mark_reported.assert_called_once_with(SNAPSHOT)
+        self.assertEqual(load_state(self.path).last_attempt_at, self.now)
+
+    def test_sends_once_per_interval(self):
+        reporter = self.reporter()
+        self.advance(FIRST_DELAY_S[0])
+        reporter.tick()
+
+        self.advance(WAKE_S)
+        reporter.tick()
+        self.assertEqual(self.send.call_count, 1)
+
+        self.advance(INTERVAL_S - JITTER_S)
+        reporter.tick()
+        self.assertEqual(self.send.call_count, 2)
+
+    def test_a_restart_inside_the_interval_does_not_send(self):
+        save_state(AnalyticsState("a" * 32, self.now), self.path)
+        reporter = self.reporter()
+
+        self.advance(FIRST_DELAY_S[0])
+        reporter.tick()
+
+        self.send.assert_not_called()
+
+    def test_a_failed_send_keeps_watermarks_and_waits_a_full_interval(self):
+        self.send.return_value = SendOutcome.failed
+        reporter = self.reporter()
+        self.advance(FIRST_DELAY_S[0])
+
+        reporter.tick()
+        self.advance(WAKE_S)
+        reporter.tick()
+
+        self.assertEqual(self.send.call_count, 1)
+        self.registry.mark_reported.assert_not_called()
+
+    def test_opting_out_raises_the_prompt_once_and_deletes_the_state(self):
+        self.holder.config = make_config()
+        save_state(AnalyticsState("a" * 32, 0.0), self.path)
+        reporter = self.reporter()
+
+        reporter.tick()
+        reporter.tick()
+
+        self.raise_notice.assert_called_once_with(PROMPT_KIND)
+        self.assertFalse(os.path.exists(self.path))
+        self.send.assert_not_called()
+
+    def test_opting_in_resolves_the_prompt(self):
+        self.holder.config = make_config()
+        reporter = self.reporter()
+        reporter.tick()
+
+        self.holder.config = make_config({"telemetry": {"analytics": True}})
+        reporter.tick()
+
+        self.resolve_notice.assert_called_once_with(PROMPT_KIND)
+
+    def test_safe_mode_touches_nothing(self):
+        self.holder.config = make_config({"safe_mode": True})
+        save_state(AnalyticsState("a" * 32, 0.0), self.path)
+        reporter = self.reporter()
+
+        self.advance(FIRST_DELAY_S[0])
+        reporter.tick()
+
+        self.raise_notice.assert_not_called()
+        self.assertTrue(os.path.exists(self.path))
+        self.send.assert_not_called()
+
+    def test_an_unwritable_config_dir_skips_sending_and_warns_once(self):
+        reporter = self.reporter(os.path.join(self.dir.name, "missing", "x.json"))
+        self.advance(FIRST_DELAY_S[0])
+
+        with self.assertLogs("frigate.analytics.reporter", "WARNING") as logs:
+            reporter.tick()
+            self.advance(WAKE_S)
+            reporter.tick()
+
+        self.send.assert_not_called()
+        self.assertEqual(len(logs.output), 1)
+
+    def test_a_last_attempt_in_the_future_is_ignored(self):
+        save_state(AnalyticsState("a" * 32, self.now + 10 * INTERVAL_S), self.path)
+        reporter = self.reporter()
+
+        self.advance(FIRST_DELAY_S[0])
+        reporter.tick()
+
+        self.send.assert_called_once()
+
+    def test_the_loop_survives_a_failing_tick(self):
+        reporter = self.reporter()
+        reporter.stop_event.set()
+
+        with (
+            patch.object(reporter, "tick", side_effect=RuntimeError("boom")),
+            self.assertLogs("frigate.analytics.reporter", "ERROR"),
+        ):
+            reporter.run()