nik/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-08-31 07:27:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

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

* settle bandwidth samples per stream instead of per camera

* fix mypy
This commit is contained in:
Josh Hawkins 2026-08-18 10:39:58 -05:00
parent 2dd700aa5a
commit f5e398036e
2 changed files with 212 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

163
frigate/storage.py
View File

@ -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...")

104
frigate/test/test_storage.py
View File

@ -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)