Add the analytics report transport

frigate/analytics/transport.py

@@ -0,0 +1,54 @@
+"""POST a report to the ingest endpoint."""
+
+import logging
+from enum import Enum
+
+import requests
+
+from frigate.version import VERSION
+
+logger = logging.getLogger(__name__)
+
+TIMEOUT_S = 30
+
+
+class SendOutcome(Enum):
+    accepted = "accepted"
+    rejected = "rejected"
+    rate_limited = "rate_limited"
+    failed = "failed"
+
+
+def send_report(url: str, body: str) -> SendOutcome:
+    """Send one report. Never raises; the outcome says what happened."""
+    try:
+        # a redirect would turn the POST into a GET, so it counts as a failure
+        response = requests.post(
+            url,
+            data=body.encode(),
+            headers={
+                "Content-Type": "application/json",
+                "User-Agent": f"Frigate/{VERSION}",
+            },
+            timeout=TIMEOUT_S,
+            allow_redirects=False,
+        )
+    except requests.RequestException as err:
+        logger.warning("Unable to send the analytics report: %s", err)
+        return SendOutcome.failed
+
+    status = response.status_code
+
+    if 200 <= status < 300:
+        return SendOutcome.accepted
+
+    if status == 400:
+        logger.warning("The analytics report was rejected: %s", response.text[:200])
+        return SendOutcome.rejected
+
+    if status == 429:
+        logger.debug("The analytics endpoint is rate limiting this install")
+        return SendOutcome.rate_limited
+
+    logger.warning("The analytics endpoint returned %s", status)
+    return SendOutcome.failed

frigate/const.py

@@ -101,6 +101,9 @@ MAX_WAL_SIZE = 10  # MB
 
 DEFAULT_FFMPEG_VERSION = os.environ.get("DEFAULT_FFMPEG_VERSION", "")
 INCLUDED_FFMPEG_VERSIONS = os.environ.get("INCLUDED_FFMPEG_VERSIONS", "").split(":")
+ANALYTICS_URL = os.environ.get(
+    "FRIGATE_ANALYTICS_URL", "https://analytics.frigate.video/report"
+)
 LIBAVFORMAT_VERSION_MAJOR = int(os.environ.get("LIBAVFORMAT_VERSION_MAJOR", "59"))
 FFMPEG_HWACCEL_NVIDIA = "preset-nvidia"
 FFMPEG_HWACCEL_VAAPI = "preset-vaapi"

frigate/test/test_analytics_transport.py

@@ -0,0 +1,60 @@
+"""Tests for sending a report."""
+
+import unittest
+from unittest.mock import Mock, patch
+
+import requests
+
+from frigate.analytics.transport import TIMEOUT_S, SendOutcome, send_report
+from frigate.version import VERSION
+
+URL = "https://example.test/report"
+
+
+def post(status: int = 204, text: str = "", side_effect: Exception | None = None):
+    return patch(
+        "frigate.analytics.transport.requests.post",
+        return_value=Mock(status_code=status, text=text),
+        side_effect=side_effect,
+    )
+
+
+class TestSendReport(unittest.TestCase):
+    def test_posts_the_body_as_json_with_a_user_agent(self):
+        with post() as mock_post:
+            outcome = send_report(URL, '{"a":1}')
+
+        self.assertIs(outcome, SendOutcome.accepted)
+        args, kwargs = mock_post.call_args
+        self.assertEqual(args[0], URL)
+        self.assertEqual(kwargs["data"], b'{"a":1}')
+        self.assertEqual(kwargs["headers"]["Content-Type"], "application/json")
+        self.assertEqual(kwargs["headers"]["User-Agent"], f"Frigate/{VERSION}")
+        self.assertFalse(kwargs["allow_redirects"])
+        self.assertEqual(kwargs["timeout"], TIMEOUT_S)
+
+    def test_maps_statuses_to_outcomes(self):
+        cases = (
+            (200, SendOutcome.accepted),
+            (400, SendOutcome.rejected),
+            (429, SendOutcome.rate_limited),
+            (302, SendOutcome.failed),
+            (500, SendOutcome.failed),
+        )
+
+        for status, outcome in cases:
+            with self.subTest(status=status), post(status):
+                self.assertIs(send_report(URL, "{}"), outcome)
+
+    def test_logs_the_rejection_reason(self):
+        with (
+            post(400, "install_id: bad"),
+            self.assertLogs("frigate.analytics.transport", "WARNING") as logs,
+        ):
+            send_report(URL, "{}")
+
+        self.assertIn("install_id: bad", logs.output[0])
+
+    def test_a_network_error_fails_without_raising(self):
+        with post(side_effect=requests.ConnectionError("down")):
+            self.assertIs(send_report(URL, "{}"), SendOutcome.failed)