mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-08-09 10:23:01 +02:00
fix(rrdtool): replace rrdtool.info() with self-tracked timestamp to eliminate allocation storm
Problem
-------
update_packet_metrics() called rrdtool.info() (cached for 5 s) to get the
RRD's last_update timestamp. rrdtool.info() returns a massive Python dict:
17 data sources × 5 RRAs × ~8 fields each = ~700+ dict entries per call.
tracemalloc showed +10696 new allocations / +251 KB at this exact line,
flagged as "Investigate" in the memory diagnostics dashboard.
The rrdtool.info() approach was also unnecessarily complex: it required a
5-second secondary cache, a _pending_rrd_update buffer, and two extra
instance attributes — all to answer one question ("did we already write
this period?") that we can answer ourselves with a single integer.
Fix
---
Replace _last_rrd_info_cache / _last_rrd_info_time / _pending_rrd_update
with a single self._last_rrd_update: int = 0 that stores the timestamp of
the last successful rrdtool.update() call. The throttle check becomes:
if timestamp <= self._last_rrd_update:
return
On success: self._last_rrd_update = timestamp
Zero dict allocations per call. The only downside vs rrdtool.info() is
that _last_rrd_update resets to 0 on process restart, meaning the first
packet after a restart always triggers a write — correct behaviour.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -19,10 +19,10 @@ class RRDToolHandler:
|
||||
self.rrd_path = self.storage_dir / "metrics.rrd"
|
||||
self.available = RRDTOOL_AVAILABLE
|
||||
self._init_rrd()
|
||||
# Batch RRD updates: track pending update and last cached info
|
||||
self._pending_rrd_update = None
|
||||
self._last_rrd_info_time = 0
|
||||
self._last_rrd_info_cache = None
|
||||
# Timestamp of the last successful rrdtool.update() call (unix seconds,
|
||||
# aligned to the 60-second RRD step). Used to skip writes whose period
|
||||
# has already been committed — no rrdtool.info() call needed.
|
||||
self._last_rrd_update: int = 0
|
||||
# Read-side cache: rrdtool.fetch() returns 24 h of data and is a
|
||||
# blocking disk read. Cache the result for 60 s — matching the RRD
|
||||
# step size — so repeated dashboard refreshes don't hammer the SD card.
|
||||
@@ -81,10 +81,11 @@ class RRDToolHandler:
|
||||
logger.error(f"Failed to create RRD database: {e}")
|
||||
|
||||
def update_packet_metrics(self, record: dict, cumulative_counts: dict):
|
||||
"""Buffer packet metrics for batch RRD update instead of per-packet writes.
|
||||
|
||||
RRD uses 60-second time steps, so we batch updates within each period
|
||||
and only write when the time period changes or buffer is full.
|
||||
"""Write packet metrics to RRD, throttled to once per 60-second step.
|
||||
|
||||
RRD enforces a 60-second minimum step between updates. We track the
|
||||
last written timestamp ourselves — no rrdtool.info() call needed, which
|
||||
previously allocated thousands of Python objects per call.
|
||||
"""
|
||||
if not self.available or not self.rrd_path.exists():
|
||||
return
|
||||
@@ -92,27 +93,8 @@ class RRDToolHandler:
|
||||
try:
|
||||
timestamp = int(record.get("timestamp", time.time()))
|
||||
|
||||
# Cache RRD info for up to 5 seconds to avoid repeated rrdtool.info() calls
|
||||
now = time.time()
|
||||
if now - self._last_rrd_info_time > 5 or self._last_rrd_info_cache is None:
|
||||
try:
|
||||
self._last_rrd_info_cache = rrdtool.info(str(self.rrd_path))
|
||||
self._last_rrd_info_time = now
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to cache RRD info: {e}")
|
||||
self._last_rrd_info_cache = None
|
||||
return
|
||||
|
||||
if self._last_rrd_info_cache is None:
|
||||
return
|
||||
|
||||
last_update = int(self._last_rrd_info_cache.get("last_update", timestamp - 60))
|
||||
|
||||
# Skip if timestamp is in same or earlier time period than last update
|
||||
# (RRD step is 60 seconds)
|
||||
if timestamp <= last_update:
|
||||
# But still buffer cumulative counts for when we do update
|
||||
self._pending_rrd_update = (timestamp, cumulative_counts, record)
|
||||
# Skip if this packet falls in the same 60-second period we already wrote.
|
||||
if timestamp <= self._last_rrd_update:
|
||||
return
|
||||
|
||||
# Build update string from cumulative counts
|
||||
@@ -144,11 +126,8 @@ class RRDToolHandler:
|
||||
type_values_str = ":".join(type_values)
|
||||
values = f"{basic_values}:{type_values_str}"
|
||||
|
||||
# Write to RRD - this is now only called once per 60-second period
|
||||
rrdtool.update(str(self.rrd_path), values)
|
||||
# Invalidate cache so next period fetches fresh info
|
||||
self._last_rrd_info_cache = None
|
||||
self._pending_rrd_update = None
|
||||
self._last_rrd_update = timestamp
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to update RRD packet metrics: {e}")
|
||||
|
||||
Reference in New Issue
Block a user