Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 54 additions & 12 deletions src/plugin_system/resource_monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,20 @@ def update_average_execution_time(self):
self.total_execution_time = self.total_execution_time / self.call_count


#: How often a plugin's metrics are written to the cache, in seconds.
#:
#: Persisting on every call meant a small file rewritten roughly nine times a
#: minute per plugin. On a rig with fourteen active plugins that was ~126
#: writes a minute for metrics alone, and since each ~350-byte file costs a
#: 4KB block plus an ext4 journal entry, it dominated the device's write
#: volume -- on an SD card, which wears out.
#:
#: The in-memory copy stays authoritative and exact; only the cross-process
#: snapshot the web UI reads is delayed, and telemetry up to half a minute old
#: is still a fair description of a long-running plugin.
_METRICS_PERSIST_INTERVAL = 30.0


class PluginResourceMonitor:
"""
Monitors resource usage for plugins.
Expand All @@ -75,6 +89,10 @@ def __init__(self, cache_manager, enable_monitoring: bool = True):
# Resource metrics per plugin
self._metrics: Dict[str, ResourceMetrics] = {}
self._limits: Dict[str, ResourceLimits] = {}
# When each plugin's metrics last reached the cache. Metrics change on
# every call, so they cannot be de-duplicated the way health state can;
# they are rate-limited instead. See _METRICS_PERSIST_INTERVAL.
self._metrics_persisted_at: Dict[str, float] = {}

# Thread-local storage for execution tracking
self._local = threading.local()
Expand Down Expand Up @@ -232,18 +250,8 @@ def monitor_call(self, plugin_id: str, func: Callable, *args, **kwargs) -> Any:
# CPU is harder to measure per-call, so we track it separately
metrics.cpu_percent = self._get_process_cpu_percent()

# Persist metrics
cache_key = self._get_metrics_key(plugin_id)
self.cache_manager.set(cache_key, {
'memory_mb': metrics.memory_mb,
'cpu_percent': metrics.cpu_percent,
'execution_time': metrics.execution_time,
'call_count': metrics.call_count,
'total_execution_time': metrics.total_execution_time,
'max_execution_time': metrics.max_execution_time,
'min_execution_time': metrics.min_execution_time if metrics.min_execution_time != float('inf') else 0.0,
'last_update_time': metrics.last_update_time
})
# Persist metrics, at most once per interval per plugin.
self._persist_metrics(plugin_id, metrics)

# Check limits
if limits:
Expand Down Expand Up @@ -363,11 +371,45 @@ def get_all_metrics_summaries(self) -> Dict[str, Dict[str, Any]]:
summaries[plugin_id] = self.get_metrics_summary(plugin_id)
return summaries

def _persist_metrics(self, plugin_id: str, metrics: ResourceMetrics,
force: bool = False) -> None:
"""Write a plugin's metrics to the cache, at most once per interval.

Caller must hold ``self._lock``.
"""
# Monotonic, not wall clock: these devices have no RTC, so the clock
# jumps by however far off boot-time was the moment NTP first syncs.
# A forward jump would allow an early write, a backward one would
# stall the snapshot well past the interval.
now = time.monotonic()
if not force and now - self._metrics_persisted_at.get(plugin_id, 0.0) \
< _METRICS_PERSIST_INTERVAL:
return
cache_key = self._get_metrics_key(plugin_id)
self.cache_manager.set(cache_key, {
'memory_mb': metrics.memory_mb,
'cpu_percent': metrics.cpu_percent,
'execution_time': metrics.execution_time,
'call_count': metrics.call_count,
'total_execution_time': metrics.total_execution_time,
'max_execution_time': metrics.max_execution_time,
'min_execution_time': (metrics.min_execution_time
if metrics.min_execution_time != float('inf')
else 0.0),
'last_update_time': metrics.last_update_time,
})
# Only after the write lands. Marking it first would mean a failed
# set() bought the next interval's silence without leaving a snapshot.
self._metrics_persisted_at[plugin_id] = now

def reset_metrics(self, plugin_id: str) -> None:
"""Reset metrics for a plugin."""
with self._lock:
if plugin_id in self._metrics:
self._metrics[plugin_id] = ResourceMetrics()
cache_key = self._get_metrics_key(plugin_id)
self.cache_manager.delete(cache_key)
# Let the next call persist immediately rather than leaving the
# deleted key absent for the rest of the interval.
self._metrics_persisted_at.pop(plugin_id, None)

64 changes: 64 additions & 0 deletions test/test_resource_monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,3 +127,67 @@ def cache_get(key, max_age=None, memory_ttl=None):
fresh = mon.get_metrics_summary("p", force_reload=True)
assert fresh["call_count"] == 7
assert any(c.kwargs.get("memory_ttl") == 0 for c in cache.get.call_args_list)


class TestMetricsPersistenceChurn:
"""Metrics are telemetry; writing them on every call wore the SD card.

Each write is a ~350-byte file, which on ext4 costs a 4KB block plus a
journal entry. At roughly nine calls a minute per plugin across fourteen
plugins it dominated the device's write volume.
"""

def test_repeated_calls_persist_once_per_interval(self):
cache = _cache()
mon = PluginResourceMonitor(cache, enable_monitoring=False)
for _ in range(50):
mon.monitor_call("p", lambda: None)
writes = [c for c in cache.set.call_args_list
if c.args and str(c.args[0]).startswith("plugin_metrics:")]
assert len(writes) == 1, (
f"50 calls produced {len(writes)} metric writes; expected 1")

def test_the_interval_elapsing_allows_the_next_write(self, monkeypatch):
import src.plugin_system.resource_monitor as rm
cache = _cache()
mon = PluginResourceMonitor(cache, enable_monitoring=False)
mon.monitor_call("p", lambda: None)
# pretend the interval has passed
mon._metrics_persisted_at["p"] -= rm._METRICS_PERSIST_INTERVAL + 1
mon.monitor_call("p", lambda: None)
writes = [c for c in cache.set.call_args_list
if c.args and str(c.args[0]).startswith("plugin_metrics:")]
assert len(writes) == 2

def test_in_memory_metrics_stay_exact_while_writes_are_skipped(self):
mon = PluginResourceMonitor(_cache(), enable_monitoring=False)
for _ in range(20):
mon.monitor_call("p", lambda: None)
assert mon.get_metrics("p").call_count == 20

def test_reset_lets_the_next_call_persist_immediately(self):
cache = _cache()
mon = PluginResourceMonitor(cache, enable_monitoring=False)
mon.monitor_call("p", lambda: None)
mon.reset_metrics("p")
mon.monitor_call("p", lambda: None)
writes = [c for c in cache.set.call_args_list
if c.args and str(c.args[0]).startswith("plugin_metrics:")]
assert len(writes) == 2, "reset should clear the throttle timestamp"

def test_a_failed_write_does_not_buy_the_next_interval_of_silence(self):
"""A set() that raises must not count as having persisted.

Marking the timestamp before the write would leave no snapshot in the
cache and still suppress the next 30 seconds of attempts.
"""
cache = _cache()
cache.set.side_effect = [OSError("disk full"), None]
mon = PluginResourceMonitor(cache, enable_monitoring=False)
with pytest.raises(OSError):
mon.monitor_call("p", lambda: None)
# the very next call must try again rather than skip the interval
mon.monitor_call("p", lambda: None)
writes = [c for c in cache.set.call_args_list
if c.args and str(c.args[0]).startswith("plugin_metrics:")]
assert len(writes) == 2, "a failed write should be retried, not skipped"
Loading