diff --git a/frigate/analytics/reporter.py b/frigate/analytics/reporter.py new file mode 100644 index 0000000000..fc4a3c93fb --- /dev/null +++ b/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) diff --git a/frigate/app.py b/frigate/app.py index 9c49cabb1b..b16fb25042 100644 --- a/frigate/app.py +++ b/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() diff --git a/frigate/test/test_analytics_reporter.py b/frigate/test/test_analytics_reporter.py new file mode 100644 index 0000000000..ca7a06ab01 --- /dev/null +++ b/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()