Base emergency cleanup on the streams a camera is currently recording (#24022)

* gate emergency cleanup bandwidth on the streams a camera currently records

* settle bandwidth samples per stream instead of per camera

* fix mypy

frigate/storage.py

@@ -9,6 +9,10 @@ from pathlib import Path
 from peewee import SQL, Case, fn
 
 from frigate.config import FrigateConfig
+from frigate.config.camera.updater import (
+    CameraConfigUpdateEnum,
+    CameraConfigUpdateSubscriber,
+)
 from frigate.const import (
     RECORD_DIR,
     REPLAY_CAMERA_PREFIX,
@@ -24,6 +28,7 @@ bandwidth_equation = Recordings.segment_size / (
 )
 
 MAX_CALCULATED_BANDWIDTH = 10000  # 10Gb/hr
+BANDWIDTH_SAMPLE_TARGET = 50
 
 
 class StorageMaintainer(threading.Thread):
@@ -34,6 +39,41 @@ class StorageMaintainer(threading.Thread):
         self.config = config
         self.stop_event = stop_event
         self.camera_storage_stats: dict[str, dict] = {}
+        self.config_subscriber = CameraConfigUpdateSubscriber(
+            self.config,
+            self.config.cameras,
+            [CameraConfigUpdateEnum.record],
+        )
+
+    def _recording_stream_types(self, camera: str) -> tuple[str, ...]:
+        """Return the stream types the camera is currently recording."""
+        camera_config = self.config.cameras.get(camera)
+
+        if camera_config is None or not camera_config.record.enabled:
+            return ()
+
+        if camera_config.record.sub.enabled:
+            return (STREAM_TYPE_MAIN, STREAM_TYPE_SUB)
+
+        return (STREAM_TYPE_MAIN,)
+
+    def expected_hourly_bandwidth(self) -> float:
+        """Return the MB/hr the cameras are expected to write.
+
+        Only the streams a camera currently records are counted, so toggling
+        recording or sub stream recording is reflected without waiting for the
+        existing segments of a stopped stream to expire.
+        """
+        total = 0.0
+
+        for camera, stats in self.camera_storage_stats.items():
+            stream_bandwidths = stats.get("bandwidth_by_stream", {})
+            total += sum(
+                stream_bandwidths.get(stream_type, 0)
+                for stream_type in self._recording_stream_types(camera)
+            )
+
+        return round(total, 2)
 
     def _recent_stream_bandwidth(
         self, camera: str, stream_type: str, window: int
@@ -64,6 +104,32 @@ class StorageMaintainer(threading.Thread):
         avg: float | None = Recordings.select(fn.AVG(SQL("bw"))).from_(recent).scalar()
         return avg
 
+    def _stream_sample_count(self, camera: str, stream_type: str) -> int:
+        """Count a stream's non-zero segments, stopping at the sample target."""
+        count: int = (
+            Recordings.select(Recordings.id)
+            .where(
+                Recordings.camera == camera,
+                Recordings.stream_type == stream_type,
+                Recordings.segment_size > 0,
+            )
+            .limit(BANDWIDTH_SAMPLE_TARGET)
+            .count()
+        )
+        return count
+
+    def _needs_refresh(self, camera: str) -> bool:
+        """Return whether a stream the camera records still lacks samples.
+
+        Counted per stream rather than per camera: a stream that starts
+        recording later has no samples of its own yet, and a camera-wide count
+        would report it settled on the strength of another stream's history.
+        """
+        return any(
+            self._stream_sample_count(camera, stream_type) < BANDWIDTH_SAMPLE_TARGET
+            for stream_type in self._recording_stream_types(camera)
+        )
+
     def calculate_camera_bandwidth(self) -> None:
         """Calculate an average MB/hr for each camera."""
         for camera in self.config.cameras.keys():
@@ -71,56 +137,45 @@ class StorageMaintainer(threading.Thread):
             if camera.startswith(REPLAY_CAMERA_PREFIX):
                 continue
 
-            # cameras with < 50 segments should be refreshed to keep size accurate
+            if not self.camera_storage_stats.get(camera, {}).get("needs_refresh", True):
-            # when few segments are available
+                continue
-            if self.camera_storage_stats.get(camera, {}).get("needs_refresh", True):
-                self.camera_storage_stats[camera] = {
-                    "needs_refresh": (
-                        Recordings.select(Recordings.id)
-                        .where(Recordings.camera == camera, Recordings.segment_size > 0)
-                        .limit(50)
-                        .count()
-                        < 50
-                    )
-                }
 
-                # calculate MB/hr from the last 100 segments of each stream
+            # calculate MB/hr from the last 100 segments of each stream
-                # type and sum the rates; mixing streams would average small
+            # type and sum the rates; mixing streams would average small
-                # sub segments against large main segments and underestimate
+            # sub segments against large main segments and underestimate
-                # the true write rate
+            # the true write rate
-                bandwidth_by_stream: dict[str, float] = {}
+            bandwidth_by_stream: dict[str, float] = {}
-                for stream_type in (STREAM_TYPE_MAIN, STREAM_TYPE_SUB):
+            for stream_type in (STREAM_TYPE_MAIN, STREAM_TYPE_SUB):
-                    avg_bw = self._recent_stream_bandwidth(camera, stream_type, 100)
+                avg_bw = self._recent_stream_bandwidth(camera, stream_type, 100)
-                    if avg_bw is None:
+                if avg_bw is None:
-                        # the recent window can be all zero-size ingest
+                    # the recent window can be all zero-size ingest
-                        # glitches; look further back before concluding
+                    # glitches; look further back before concluding
-                        # the stream writes nothing
+                    # the stream writes nothing
-                        avg_bw = self._recent_stream_bandwidth(
+                    avg_bw = self._recent_stream_bandwidth(camera, stream_type, 1000)
-                            camera, stream_type, 1000
+                if avg_bw is not None:
-                        )
+                    bandwidth_by_stream[stream_type] = round(avg_bw * 3600, 2)
-                    if avg_bw is not None:
-                        bandwidth_by_stream[stream_type] = round(avg_bw * 3600, 2)
 
-                bandwidth = round(sum(bandwidth_by_stream.values()), 2)
+            bandwidth = round(sum(bandwidth_by_stream.values()), 2)
 
-                if bandwidth > MAX_CALCULATED_BANDWIDTH:
+            if bandwidth > MAX_CALCULATED_BANDWIDTH:
-                    logger.warning(
+                logger.warning(
-                        f"{camera} has a bandwidth of {bandwidth} MB/hr which exceeds the expected maximum. This typically indicates an issue with the cameras recordings."
+                    f"{camera} has a bandwidth of {bandwidth} MB/hr which exceeds the expected maximum. This typically indicates an issue with the cameras recordings."
-                    )
-                    # scale each stream so the per stream values still sum to
-                    # the clamped total the UI displays alongside them
-                    scale = MAX_CALCULATED_BANDWIDTH / bandwidth
-                    bandwidth_by_stream = {
-                        stream_type: round(value * scale, 2)
-                        for stream_type, value in bandwidth_by_stream.items()
-                    }
-                    bandwidth = MAX_CALCULATED_BANDWIDTH
-
-                self.camera_storage_stats[camera]["bandwidth"] = bandwidth
-                self.camera_storage_stats[camera]["bandwidth_by_stream"] = (
-                    bandwidth_by_stream
                 )
-                logger.debug(f"{camera} has a bandwidth of {bandwidth} MiB/hr.")
+                # scale each stream so the per stream values still sum to
+                # the clamped total the UI displays alongside them
+                scale = MAX_CALCULATED_BANDWIDTH / bandwidth
+                bandwidth_by_stream = {
+                    stream_type: round(value * scale, 2)
+                    for stream_type, value in bandwidth_by_stream.items()
+                }
+                bandwidth = MAX_CALCULATED_BANDWIDTH
+
+            self.camera_storage_stats[camera] = {
+                "needs_refresh": self._needs_refresh(camera),
+                "bandwidth": bandwidth,
+                "bandwidth_by_stream": bandwidth_by_stream,
+            }
+            logger.debug(f"{camera} has a bandwidth of {bandwidth} MiB/hr.")
 
     def calculate_camera_usages(self) -> dict[str, dict]:
         """Calculate the storage usage of each camera."""
@@ -175,9 +230,7 @@ class StorageMaintainer(threading.Thread):
         """Return if storage needs cleanup."""
         # currently runs cleanup if less than 1 hour of space is left
         # disk_usage should not spin up disks
-        hourly_bandwidth = sum(
+        hourly_bandwidth = self.expected_hourly_bandwidth()
-            [b["bandwidth"] for b in self.camera_storage_stats.values()]
-        )
         remaining_storage = round(shutil.disk_usage(RECORD_DIR).free / pow(2, 20), 1)
         logger.debug(
             f"Storage cleanup check: {hourly_bandwidth} hourly with remaining storage: {remaining_storage}."
@@ -188,9 +241,7 @@ class StorageMaintainer(threading.Thread):
         """Remove oldest hour of recordings."""
         logger.debug("Starting storage cleanup.")
         deleted_segments_size = 0
-        hourly_bandwidth = sum(
+        hourly_bandwidth = self.expected_hourly_bandwidth()
-            [b["bandwidth"] for b in self.camera_storage_stats.values()]
-        )
 
         recordings = (
             Recordings.select(
@@ -350,10 +401,17 @@ class StorageMaintainer(threading.Thread):
         """Check every 5 minutes if storage needs to be cleaned up."""
         if self.config.safe_mode:
             logger.info("Safe mode enabled, skipping storage maintenance")
+            self.config_subscriber.stop()
             return
 
         self.calculate_camera_bandwidth()
         while not self.stop_event.wait(300):
+            updated_topics = self.config_subscriber.check_for_updates()
+
+            for camera in updated_topics.get(CameraConfigUpdateEnum.record.name, []):
+                if camera in self.camera_storage_stats:
+                    self.camera_storage_stats[camera]["needs_refresh"] = True
+
             if not self.camera_storage_stats or True in [
                 r["needs_refresh"] for r in self.camera_storage_stats.values()
             ]:
@@ -366,4 +424,5 @@ class StorageMaintainer(threading.Thread):
                 )
                 self.reduce_storage_consumption()
 
+        self.config_subscriber.stop()
         logger.info("Exiting storage maintainer...")

frigate/test/test_storage.py

@@ -36,7 +36,10 @@ class TestHttp(unittest.TestCase):
                 "front_door": {
                     "ffmpeg": {
                         "inputs": [
-                            {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}
+                            {
+                                "path": "rtsp://10.0.0.1:554/video",
+                                "roles": ["detect", "record"],
+                            }
                         ]
                     },
                     "detect": {
@@ -44,6 +47,7 @@ class TestHttp(unittest.TestCase):
                         "width": 1920,
                         "fps": 5,
                     },
+                    "record": {"enabled": True},
                 }
             },
         }
@@ -53,7 +57,10 @@ class TestHttp(unittest.TestCase):
                 "front_door": {
                     "ffmpeg": {
                         "inputs": [
-                            {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}
+                            {
+                                "path": "rtsp://10.0.0.1:554/video",
+                                "roles": ["detect", "record"],
+                            }
                         ]
                     },
                     "detect": {
@@ -61,11 +68,15 @@ class TestHttp(unittest.TestCase):
                         "width": 1920,
                         "fps": 5,
                     },
+                    "record": {"enabled": True},
                 },
                 "back_door": {
                     "ffmpeg": {
                         "inputs": [
-                            {"path": "rtsp://10.0.0.2:554/video", "roles": ["detect"]}
+                            {
+                                "path": "rtsp://10.0.0.2:554/video",
+                                "roles": ["detect", "record"],
+                            }
                         ]
                     },
                     "detect": {
@@ -73,6 +84,7 @@ class TestHttp(unittest.TestCase):
                         "width": 1920,
                         "fps": 5,
                     },
+                    "record": {"enabled": True},
                 },
             },
         }
@@ -326,6 +338,92 @@ class TestHttp(unittest.TestCase):
         assert streams[STREAM_TYPE_SUB]["bandwidth"] is None
         assert streams[STREAM_TYPE_MAIN]["bandwidth"] == 7200
 
+    def test_expected_bandwidth_follows_recording_config(self):
+        """Only the streams a camera currently records count toward cleanup."""
+        config = FrigateConfig(**self.minimal_config)
+        storage = StorageMaintainer(config, MagicMock())
+        record = config.cameras["front_door"].record
+
+        time_keep = datetime.datetime.now().timestamp()
+        _insert_mock_recording(
+            "1234567.frontdoor",
+            os.path.join(self.test_dir, "main.tmp"),
+            time_keep,
+            time_keep + 10,
+            seg_size=20,
+            seg_dur=10,
+        )
+        _insert_mock_recording(
+            "1234568.frontdoor",
+            os.path.join(self.test_dir, "sub.tmp"),
+            time_keep,
+            time_keep + 10,
+            seg_size=2,
+            seg_dur=10,
+            stream_type=STREAM_TYPE_SUB,
+        )
+
+        storage.calculate_camera_bandwidth()
+
+        # sub segments are still on disk but the camera no longer records them
+        assert storage.expected_hourly_bandwidth() == 7200
+
+        record.sub.enabled = True
+        assert storage.expected_hourly_bandwidth() == 7920
+
+        record.enabled = False
+        assert storage.expected_hourly_bandwidth() == 0
+
+    def test_stream_stays_dirty_until_it_has_its_own_samples(self):
+        """A stream enabled before its first segment keeps the camera dirty.
+
+        The record config update can be handled on a tick before ffmpeg has
+        written anything, so a camera-wide segment count would settle the camera
+        on the strength of the other stream's history and never measure the new
+        stream at all.
+        """
+        config = FrigateConfig(**self.minimal_config)
+        storage = StorageMaintainer(config, MagicMock())
+        record = config.cameras["front_door"].record
+        stats = storage.camera_storage_stats
+
+        time_keep = datetime.datetime.now().timestamp()
+        for i in range(60):
+            _insert_mock_recording(
+                f"main_{i}.frontdoor",
+                os.path.join(self.test_dir, f"main_{i}.tmp"),
+                time_keep + i * 10,
+                time_keep + i * 10 + 10,
+                seg_size=20,
+                seg_dur=10,
+            )
+
+        storage.calculate_camera_bandwidth()
+        assert stats["front_door"]["needs_refresh"] is False
+
+        record.sub.enabled = True
+        stats["front_door"]["needs_refresh"] = True
+        storage.calculate_camera_bandwidth()
+
+        assert stats["front_door"]["needs_refresh"] is True
+        assert storage.expected_hourly_bandwidth() == 7200
+
+        for i in range(60):
+            _insert_mock_recording(
+                f"sub_{i}.frontdoor",
+                os.path.join(self.test_dir, f"sub_{i}.tmp"),
+                time_keep + 5000 + i * 10,
+                time_keep + 5000 + i * 10 + 10,
+                seg_size=2,
+                seg_dur=10,
+                stream_type=STREAM_TYPE_SUB,
+            )
+
+        storage.calculate_camera_bandwidth()
+
+        assert stats["front_door"]["needs_refresh"] is False
+        assert storage.expected_hourly_bandwidth() == 7920
+
     def test_camera_usages_with_no_recordings(self):
         """A camera with no segments reports zero usage and no streams."""
         config = FrigateConfig(**self.minimal_config)