Compare commits

..
Author SHA1 Message Date
ChuckBuildsandClaude Opus 5 dab4b3ea57 fix: use a monotonic clock and only mark metrics persisted once written
Two review findings on the throttle, both right.

The interval compared wall-clock timestamps. 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. time.monotonic() is not subject to either.

The timestamp was also recorded before cache_manager.set(). A set() that
raised would buy the next interval's silence without leaving a snapshot
behind, which is the one case where skipping the write is least affordable.
Recorded after the write lands instead, so a failure is retried on the next
call.

Verified by restoring the original ordering: the new test then reports one
write where two are expected.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW
2026-08-20 07:42:24 -04:00
ChuckBuildsandClaude Opus 5 0fd2bfae99 perf(plugins): stop rewriting a plugin's metrics file on every call
Plugin metrics were persisted to the cache inside monitor_call, so every
call by every plugin rewrote a small JSON file. Measured on a running rig:
one plugin's plugin_metrics file changed nine times a minute, with fourteen
such files active. Each is around 350 bytes, which on ext4 costs a 4KB block
plus a journal entry, so the cost is dominated by the write itself rather
than the payload. Cache writes accounted for essentially all of that device's
2.4 MB/min of SD traffic, on a card that wears out and has already failed
twice on the other rig.

Metrics cannot be de-duplicated the way health state can, because call_count
changes on every call and the timings usually do too. So they are rate-limited
instead: at most one write per plugin per 30 seconds.

The in-memory copy stays authoritative and exact -- a plugin's call_count is
still precise the instant after it runs. 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.

reset_metrics clears the throttle timestamp, so a reset is not left showing a
deleted key for the rest of the interval.

Extrapolating the sampled rate, this takes metric writes from roughly 126 a
minute to 28. Health persistence, the other half of the churn, is handled
separately in #475.

Verified by reverting the throttle: the churn test then reports 50 writes for
50 calls. 88 tests pass across resource monitor, plugin system and web API.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW
2026-08-20 07:06:21 -04:00
4 changed files with 122 additions and 147 deletions
+4 -25
View File
@@ -178,21 +178,11 @@ class PluginHealthTracker:
)
return self._health_state[plugin_id]
# Fields the circuit breaker is rebuilt from after a restart. Everything
# else in a health record is reporting, read only for display.
_DURABLE_FIELDS = ('consecutive_failures', 'circuit_state',
'circuit_opened_time', 'half_open_start_time')
def _durable(self, state: Dict[str, Any]) -> tuple:
"""The part of a health record whose loss would change behaviour."""
return tuple(state.get(field) for field in self._DURABLE_FIELDS)
def record_success(self, plugin_id: str) -> None:
"""Record a successful plugin execution."""
state = self.get_health_state(plugin_id)
current_time = time.time()
durable_before = self._durable(state)
# Reset consecutive failures
state['consecutive_failures'] = 0
state['total_successes'] = state.get('total_successes', 0) + 1
@@ -208,20 +198,9 @@ class PluginHealthTracker:
# Shouldn't happen, but handle it
state['circuit_state'] = CircuitState.CLOSED.value
state['circuit_opened_time'] = None
# A healthy plugin reports success every cycle, and in that steady state
# the only fields changed above are a counter and a timestamp that
# nothing reads back after a restart. Persisting them anyway rewrites a
# small file per plugin per cycle: on a rig running 24 plugins, a
# five-minute sample measured 22 rewrites, about 4.4 a minute or 6,300 a
# day. Those land on an SD card, where the cost is an erase-block cycle
# rather than the 400 bytes involved, and where wear is what eventually
# kills the card.
# In-memory state is still updated every time, so the health API and web
# UI show exactly what they did before; only the write is skipped.
if self._durable(state) != durable_before:
self._save_health_state(plugin_id, state)
self._save_health_state(plugin_id, state)
def record_failure(self, plugin_id: str, error: Optional[Exception] = None) -> None:
"""Record a failed plugin execution."""
state = self.get_health_state(plugin_id)
+54 -12
View File
@@ -49,6 +49,20 @@ class ResourceMetrics:
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.
@@ -75,6 +89,10 @@ class PluginResourceMonitor:
# 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()
@@ -232,18 +250,8 @@ class PluginResourceMonitor:
# 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:
@@ -363,6 +371,37 @@ class PluginResourceMonitor:
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:
@@ -370,4 +409,7 @@ class PluginResourceMonitor:
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)
-110
View File
@@ -1,110 +0,0 @@
"""A healthy plugin must not rewrite its health record every cycle.
Every successful plugin update called record_success(), which persisted the
record unconditionally. In steady state the only fields that had changed were
total_successes and last_success_time -- a counter and a timestamp that
health_monitor reads for display and that nothing reads back after a restart.
Measured on a rig running 24 plugins: about 17 health-file rewrites a minute,
roughly 25,000 a day. Each is ~400 bytes, but they land on an SD card where
the unit of cost is an erase-block cycle, not the byte count, and where wear is
what eventually kills the card.
The circuit breaker still needs its own state to survive a restart, so the
write is kept for exactly the fields it is rebuilt from -- and a failure, a
circuit opening, or a recovery must still be written the moment it happens.
"""
import time
import pytest
from src.plugin_system.plugin_health import PluginHealthTracker, CircuitState
class _Cache:
"""Counts writes; serves back whatever was last written."""
def __init__(self):
self.store = {}
self.writes = 0
def set(self, key, data, ttl=None, **kwargs):
self.writes += 1
self.store[key] = data
def get(self, key, max_age=None, memory_ttl=None, **kwargs):
return self.store.get(key)
@pytest.fixture
def tracker():
cache = _Cache()
t = PluginHealthTracker(cache_manager=cache)
return t, cache
def test_steady_state_success_stops_writing(tracker):
"""The regression: 100 healthy cycles used to be 100 SD writes."""
t, cache = tracker
t.record_success("weather")
first = cache.writes
for _ in range(100):
t.record_success("weather")
assert cache.writes == first, (
f"{cache.writes - first} redundant writes across 100 healthy cycles"
)
def test_the_counters_are_still_accurate_in_memory(tracker):
"""Skipping the write must not skip the bookkeeping."""
t, _ = tracker
for _ in range(10):
t.record_success("weather")
state = t.get_health_state("weather")
assert state["total_successes"] == 10
assert state["last_success_time"] is not None
assert state["last_success_time"] <= time.time()
def test_a_failure_is_written_immediately(tracker):
t, cache = tracker
t.record_success("weather")
before = cache.writes
t.record_failure("weather", RuntimeError("boom"))
assert cache.writes > before, "a failure must reach disk"
def test_recovery_after_failure_is_written(tracker):
"""consecutive_failures returning to 0 is durable state changing."""
t, cache = tracker
t.record_failure("weather", RuntimeError("boom"))
before = cache.writes
t.record_success("weather")
assert cache.writes > before, "recovery must reach disk"
assert t.get_health_state("weather")["consecutive_failures"] == 0
def test_a_closing_circuit_is_written(tracker):
"""Success in half-open closes the circuit -- that must survive a restart."""
t, cache = tracker
state = t.get_health_state("weather")
state["circuit_state"] = CircuitState.HALF_OPEN.value
state["half_open_start_time"] = time.time()
before = cache.writes
t.record_success("weather")
assert cache.writes > before, "a circuit transition must reach disk"
assert t.get_health_state("weather")["circuit_state"] == CircuitState.CLOSED.value
def test_durable_state_survives_a_restart(tracker):
"""What is skipped must genuinely not matter to the breaker."""
t, cache = tracker
for _ in range(3):
t.record_failure("weather", RuntimeError("boom"))
for _ in range(50):
t.record_success("weather")
revived = PluginHealthTracker(cache_manager=cache)
state = revived.get_health_state("weather")
assert state["consecutive_failures"] == 0
assert state["circuit_state"] == CircuitState.CLOSED.value
+64
View File
@@ -127,3 +127,67 @@ class TestForceReload:
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"