diff --git a/frigate/storage.py b/frigate/storage.py index e1875cac90..aa959b61ec 100644 --- a/frigate/storage.py +++ b/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 - # when few segments are available - 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 - ) - } + if not self.camera_storage_stats.get(camera, {}).get("needs_refresh", True): + continue - # calculate MB/hr from the last 100 segments of each stream - # type and sum the rates; mixing streams would average small - # sub segments against large main segments and underestimate - # the true write rate - bandwidth_by_stream: dict[str, float] = {} - for stream_type in (STREAM_TYPE_MAIN, STREAM_TYPE_SUB): - avg_bw = self._recent_stream_bandwidth(camera, stream_type, 100) - if avg_bw is None: - # the recent window can be all zero-size ingest - # glitches; look further back before concluding - # the stream writes nothing - avg_bw = self._recent_stream_bandwidth( - camera, stream_type, 1000 - ) - if avg_bw is not None: - bandwidth_by_stream[stream_type] = round(avg_bw * 3600, 2) + # calculate MB/hr from the last 100 segments of each stream + # type and sum the rates; mixing streams would average small + # sub segments against large main segments and underestimate + # the true write rate + bandwidth_by_stream: dict[str, float] = {} + for stream_type in (STREAM_TYPE_MAIN, STREAM_TYPE_SUB): + avg_bw = self._recent_stream_bandwidth(camera, stream_type, 100) + if avg_bw is None: + # the recent window can be all zero-size ingest + # glitches; look further back before concluding + # the stream writes nothing + avg_bw = self._recent_stream_bandwidth(camera, stream_type, 1000) + 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: - 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." - ) - # 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 + if bandwidth > MAX_CALCULATED_BANDWIDTH: + 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." ) - 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( - [b["bandwidth"] for b in self.camera_storage_stats.values()] - ) + hourly_bandwidth = self.expected_hourly_bandwidth() 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( - [b["bandwidth"] for b in self.camera_storage_stats.values()] - ) + hourly_bandwidth = self.expected_hourly_bandwidth() 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...") diff --git a/frigate/test/test_storage.py b/frigate/test/test_storage.py index a5c5a84763..67b2db15bb 100644 --- a/frigate/test/test_storage.py +++ b/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)