Skip to content
Merged
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
2 changes: 1 addition & 1 deletion src/common/scroll_helper.py
Original file line number Diff line number Diff line change
Expand Up @@ -328,7 +328,7 @@ def update_scroll_position(self) -> None:
elapsed_time = current_time - (self.scroll_start_time or current_time)
# The image already includes display_width padding, so we only need total_scroll_width
required_total_distance = self.total_scroll_width
self.logger.info(
self.logger.debug(
"Scroll progress: elapsed=%.2fs, target=%.2fs, total_scrolled=%.0f/%d px (%.1f%%)",
elapsed_time,
self.calculated_duration,
Expand Down
85 changes: 84 additions & 1 deletion src/logging_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,12 @@ def setup_logging(
# Console handler (always add)
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(level)
console_handler.setFormatter(formatter)
# Under systemd, tag each line so the journal records the real severity
# rather than filing everything as informational. The file handler below
# keeps the plain formatter: the prefix is meaningful to journald and noise
# anywhere else.
console_handler.setFormatter(
JournalPriorityFormatter(formatter) if _under_systemd() else formatter)
root_logger.addHandler(console_handler)

# File handler (if specified)
Expand All @@ -145,6 +150,84 @@ def setup_logging(
sys.stderr.write(f"Warning: Could not set up file logging to {log_file}: {e}\n")


#: syslog priorities, which is what systemd parses from a "<N>" prefix on
#: stdout. Mapped from Python's levels.
_SYSLOG_PRIORITY = {
logging.CRITICAL: 2, # LOG_CRIT
logging.ERROR: 3, # LOG_ERR
logging.WARNING: 4, # LOG_WARNING
logging.INFO: 6, # LOG_INFO
logging.DEBUG: 7, # LOG_DEBUG
}


class JournalPriorityFormatter(logging.Formatter):
"""Wraps a formatter, prefixing each line with its syslog priority.

Under systemd everything this process writes to stdout lands in the journal
as PRIORITY=6, whatever the Python level was. Measured on a live rig: 55
ERROR lines and 13 WARNING lines in a day, every one of them recorded as
informational, so `journalctl -p err -u ledmatrix` returned nothing at all
while errors were being logged. Anyone triaging has to grep the message
text instead, which is both slower and wrong -- a search for "oom" matches
the radar logging "zoom=9".

systemd reads a leading "<N>" on each line and uses it as the priority
(sd-daemon(3)), so this needs no extra dependency. Multi-line records get
the prefix on every line, since the journal splits them and an unprefixed
continuation would fall back to the default.
"""

def __init__(self, inner: logging.Formatter):
super().__init__()
self._inner = inner

@property
def inner(self) -> logging.Formatter:
"""The formatter doing the actual work.

Whether journald tagging is applied depends on JOURNAL_STREAM, so it is
on under systemd and off in a terminal -- and anything asserting which
formatter setup_logging() selected would otherwise get a different
answer in CI than on a developer's machine. Exposing the inner one lets
those checks stay about format_type, which is what they mean.
"""
return self._inner

def format(self, record: logging.LogRecord) -> str:
text = self._inner.format(record)
prefix = f"<{_SYSLOG_PRIORITY.get(record.levelno, 6)}>"
return "\n".join(prefix + line for line in text.split("\n"))


def _under_systemd() -> bool:
"""True when stdout really is the journal.

systemd sets JOURNAL_STREAM to "dev:ino" for services whose output it
captures. Presence alone is not enough to act on: the variable is
inherited by child processes and survives redirection, so a subprocess
whose stdout is a pipe or a file still sees it and would emit the "<N>"
priority prefixes as literal noise into that output. systemd's own
guidance is to fstat the descriptor and compare st_dev/st_ino, which is
what distinguishes "the journal is somewhere in my ancestry" from "my
stdout is the journal".
"""
declared = os.environ.get("JOURNAL_STREAM")
if not declared:
return False
try:
dev_text, ino_text = declared.split(":", 1)
declared_ids = (int(dev_text), int(ino_text))
except (ValueError, AttributeError):
return False
try:
stat_result = os.fstat(sys.stdout.fileno())
except (OSError, ValueError, AttributeError):
# No usable stdout: captured by pytest, detached, or already closed.
return False
return (stat_result.st_dev, stat_result.st_ino) == declared_ids


class PluginLoggerAdapter(logging.LoggerAdapter):
"""LoggerAdapter that stamps every record with its plugin_id.

Expand Down
29 changes: 25 additions & 4 deletions src/plugin_system/plugin_health.py
Original file line number Diff line number Diff line change
Expand Up @@ -178,11 +178,21 @@ def get_health_state(self, plugin_id: str, force_reload: bool = False) -> Dict[s
)
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
Expand All @@ -198,9 +208,20 @@ def record_success(self, plugin_id: str) -> None:
# Shouldn't happen, but handle it
state['circuit_state'] = CircuitState.CLOSED.value
state['circuit_opened_time'] = None

self._save_health_state(plugin_id, state)


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

def record_failure(self, plugin_id: str, error: Optional[Exception] = None) -> None:
"""Record a failed plugin execution."""
state = self.get_health_state(plugin_id)
Expand Down
137 changes: 123 additions & 14 deletions src/plugin_system/resource_monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
import logging
import threading
from typing import Dict, Optional, Any, Callable
from dataclasses import dataclass, field
from dataclasses import dataclass, field, fields

try:
import psutil
Expand Down 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 @@ -102,6 +120,66 @@ def __init__(self, cache_manager, enable_monitoring: bool = True):
"psutil not available - resource monitoring will be limited to execution time only"
)

def _metrics_from_cache(self, plugin_id: str, cached: Any) -> "ResourceMetrics":
"""Build metrics from a cached record, ignoring anything unrecognised.

ResourceMetrics(**cached) raises TypeError on a single unexpected key,
and that exception escapes into plugin_manager, which reports it as
"plugin <id> operation failed". Every plugin fails, and the plugin
system never finishes initialising.

Seen on a live rig: every plugin failing with

ResourceMetrics.__init__() got an unexpected keyword argument
'consecutive_failures'

which is a plugin_health field, not a metrics one. How a health-shaped
record came to sit under a plugin_metrics key on that machine is not
established -- a restored backup that mixed two machines' caches is the
likeliest explanation -- but the loader should not be brittle enough for
it to matter. plugin_health already repairs its records field by field
rather than trusting whatever is on disk; this does the same.

Unknown keys are dropped and named once, so a genuine schema change is
visible in the log instead of silently discarded.
"""
if not isinstance(cached, dict):
self.logger.warning(
"Ignoring cached metrics for %s: expected a mapping, got %s",
plugin_id, type(cached).__name__)
return ResourceMetrics()

known = {f.name for f in fields(ResourceMetrics)}
unknown = sorted(set(cached) - known)
if unknown:
self.logger.warning(
"Dropping unrecognised field(s) from cached metrics for %s: %s",
plugin_id, ", ".join(unknown))
# A dataclass does not enforce its annotations, so
# ResourceMetrics(call_count="not a number") builds happily and only
# blows up later, deep inside monitor_call ("can only concatenate str
# (not \"int\") to str"). Coerce here, where there is still a cache
# key to name in the warning.
declared = {f.name: f.type for f in fields(ResourceMetrics)}
usable = {}
for key, value in cached.items():
if key not in known:
continue
try:
usable[key] = int(value) if declared[key] in ('int', int) else float(value)
except (TypeError, ValueError):
self.logger.warning(
"Cached metrics for %s have a bad %s (%r); starting fresh",
plugin_id, key, value)
return ResourceMetrics()
try:
return ResourceMetrics(**usable)
except (TypeError, ValueError) as e:
self.logger.warning(
"Cached metrics for %s unusable (%s); starting fresh",
plugin_id, e)
return ResourceMetrics()

def _get_metrics_key(self, plugin_id: str) -> str:
"""Get cache key for plugin metrics."""
return f"plugin_metrics:{plugin_id}"
Expand All @@ -126,7 +204,7 @@ def get_metrics(self, plugin_id: str, force_reload: bool = False) -> ResourceMet
cache_key, max_age=None, memory_ttl=0 if force_reload else None
)
if cached:
metrics = ResourceMetrics(**cached)
metrics = self._metrics_from_cache(plugin_id, cached)
else:
metrics = ResourceMetrics()
self._metrics[plugin_id] = metrics
Expand Down Expand Up @@ -232,18 +310,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 +431,52 @@ 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.
#
# The sentinel for "never written" is None, not 0.0. monotonic() is
# time since boot on Linux, and systemd starts this service *at* boot,
# so `now - 0.0 < 30` was true for the first half-minute of every
# single run -- the throttle swallowed the very first snapshot, which
# is the one that matters most after a restart.
now = time.monotonic()
last_written = self._metrics_persisted_at.get(plugin_id)
if (not force and last_written is not None
and now - last_written < _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)

Loading
Loading