diff --git a/src/cache/disk_cache.py b/src/cache/disk_cache.py index 13992dfc8..a1a525add 100644 --- a/src/cache/disk_cache.py +++ b/src/cache/disk_cache.py @@ -68,14 +68,15 @@ def get_cache_path(self, key: str) -> Optional[str]: return None return os.path.join(self.cache_dir, f"{key}.json") - def get(self, key: str, max_age: int = 300) -> Optional[Dict[str, Any]]: + def get(self, key: str, max_age: Optional[int] = 300) -> Optional[Dict[str, Any]]: """ Get data from disk cache. - + Args: key: Cache key - max_age: Maximum age in seconds - + max_age: Maximum age in seconds; None disables age-based expiry + (the record never counts as stale). Mirrors MemoryCache.get. + Returns: Cached data or None if not found or expired """ @@ -105,7 +106,13 @@ def get(self, key: str, max_age: int = 300) -> Optional[Dict[str, Any]]: record_ts = None now = time.time() - if record_ts is None or (now - record_ts) <= max_age: + # max_age=None means "never expires" (mirrors MemoryCache and the + # cache_manager docstring). Guard it explicitly — otherwise the + # comparison below raises TypeError and the record is treated as a + # miss, which silently breaks callers that persist long-lived state + # via get(key, max_age=None) (e.g. plugin health/metrics that must + # survive restarts and be read cross-process). + if record_ts is None or max_age is None or (now - record_ts) <= max_age: return record else: # Stale on disk; keep file for potential diagnostics but treat as miss diff --git a/src/cache_manager.py b/src/cache_manager.py index 723c4a5f2..5065dcdb3 100644 --- a/src/cache_manager.py +++ b/src/cache_manager.py @@ -574,9 +574,19 @@ def update_cache(self, data_type: str, data: Dict[str, Any]) -> bool: } return self.save_cache(data_type, cache_data) - def get(self, key: str, max_age: int = 300) -> Optional[Dict[str, Any]]: - """Get data from cache if it exists and is not stale.""" - cached_data = self.get_cached_data(key, max_age) + def get(self, key: str, max_age: Optional[int] = 300, + memory_ttl: Optional[int] = None) -> Optional[Dict[str, Any]]: + """Get data from cache if it exists and is not stale. + + Args: + key: Cache key + max_age: Max age (seconds) for the on-disk entry; None never expires. + memory_ttl: Max age (seconds) for the in-memory entry. Pass 0 to + bypass the memory tier and force a fresh read from disk — used by + cross-process readers that must observe another process's latest + write rather than a stale first snapshot. Defaults to max_age. + """ + cached_data = self.get_cached_data(key, max_age, memory_ttl=memory_ttl) if cached_data and 'data' in cached_data: return cached_data['data'] return cached_data diff --git a/src/display_controller.py b/src/display_controller.py index 905257e42..8ae64f34f 100644 --- a/src/display_controller.py +++ b/src/display_controller.py @@ -230,7 +230,24 @@ def _follower_gated_update(): cache_manager=self.cache_manager, font_manager=self.font_manager ) - + + # Activate the plugin health/metrics subsystem. PluginManager leaves + # health_tracker/resource_monitor as None by default; wiring real + # instances here turns on the circuit breaker (a repeatedly-failing + # plugin's update() is skipped after consecutive failures, then + # retried after a cooldown) and per-plugin execution-time metrics. + # Both persist to the shared cache so the web UI can surface them. + # Done before discovery/loading so load-time schema warnings have a + # tracker to record against. + try: + from src.plugin_system.plugin_health import PluginHealthTracker + from src.plugin_system.resource_monitor import PluginResourceMonitor + self.plugin_manager.health_tracker = PluginHealthTracker(self.cache_manager) + self.plugin_manager.resource_monitor = PluginResourceMonitor(self.cache_manager) + logger.info("Plugin health tracking and resource monitoring enabled") + except Exception as e: + logger.warning("Could not enable plugin health/resource monitoring: %s", e) + # Validate plugins after plugin manager is created try: from src.startup_validator import StartupValidator diff --git a/src/plugin_system/plugin_health.py b/src/plugin_system/plugin_health.py index 81e32dbb9..ecbec0769 100644 --- a/src/plugin_system/plugin_health.py +++ b/src/plugin_system/plugin_health.py @@ -52,11 +52,18 @@ def _get_health_key(self, plugin_id: str) -> str: """Get cache key for plugin health data.""" return f"plugin_health:{plugin_id}" - def _load_health_state(self, plugin_id: str) -> Dict[str, Any]: - """Load health state from cache or return defaults.""" + def _load_health_state(self, plugin_id: str, force_reload: bool = False) -> Dict[str, Any]: + """Load health state from cache or return defaults. + + ``force_reload=True`` bypasses the cache manager's in-memory tier so a + read-only consumer (e.g. the web process) observes the writer process's + latest persisted state instead of a stale first snapshot. + """ cache_key = self._get_health_key(plugin_id) - cached = self.cache_manager.get(cache_key, max_age=None) - + cached = self.cache_manager.get( + cache_key, max_age=None, memory_ttl=0 if force_reload else None + ) + if cached: return cached @@ -79,10 +86,17 @@ def _save_health_state(self, plugin_id: str, state: Dict[str, Any]) -> None: self.cache_manager.set(cache_key, state) # Persist indefinitely self._health_state[plugin_id] = state - def get_health_state(self, plugin_id: str) -> Dict[str, Any]: - """Get current health state for a plugin.""" - if plugin_id not in self._health_state: - self._health_state[plugin_id] = self._load_health_state(plugin_id) + def get_health_state(self, plugin_id: str, force_reload: bool = False) -> Dict[str, Any]: + """Get current health state for a plugin. + + ``force_reload=True`` re-reads the persisted state from the cache, + bypassing the in-memory copy — needed by cross-process readers that + would otherwise be pinned to the first snapshot they loaded. + """ + if force_reload or plugin_id not in self._health_state: + self._health_state[plugin_id] = self._load_health_state( + plugin_id, force_reload=force_reload + ) return self._health_state[plugin_id] def record_success(self, plugin_id: str) -> None: @@ -139,6 +153,28 @@ def record_failure(self, plugin_id: str, error: Optional[Exception] = None) -> N self._save_health_state(plugin_id, state) + def set_degraded(self, plugin_id: str, reason: Optional[str]) -> None: + """Flag (or clear) a plugin as degraded without touching the circuit breaker. + + Used for non-fatal issues — e.g. a config that no longer satisfies the + plugin's schema — that should be surfaced to the user but must NOT cause + the plugin to be skipped or counted as a runtime failure. Passing + ``reason=None`` clears the flag. The write is skipped when nothing + actually changes, so calling this on every load is cheap. + + Args: + plugin_id: Plugin identifier + reason: Human-readable reason string, or None to clear the flag + """ + state = self.get_health_state(plugin_id) + new_degraded = bool(reason) + new_reason = reason if reason else None + if state.get('degraded', False) == new_degraded and state.get('degraded_reason') == new_reason: + return # No change — avoid a redundant cache write + state['degraded'] = new_degraded + state['degraded_reason'] = new_reason + self._save_health_state(plugin_id, state) + def should_skip_plugin(self, plugin_id: str) -> bool: """ Check if plugin should be skipped due to circuit breaker. @@ -181,9 +217,13 @@ def should_skip_plugin(self, plugin_id: str) -> bool: return False - def get_health_summary(self, plugin_id: str) -> Dict[str, Any]: - """Get health summary for a plugin.""" - state = self.get_health_state(plugin_id) + def get_health_summary(self, plugin_id: str, force_reload: bool = False) -> Dict[str, Any]: + """Get health summary for a plugin. + + ``force_reload=True`` refreshes from the persisted cache first so + cross-process readers reflect the writer's latest state. + """ + state = self.get_health_state(plugin_id, force_reload=force_reload) total_calls = state.get('total_successes', 0) + state.get('total_failures', 0) success_rate = 0.0 @@ -201,6 +241,8 @@ def get_health_summary(self, plugin_id: str) -> Dict[str, Any]: 'last_failure_time': state.get('last_failure_time'), 'last_error': state.get('last_error'), 'is_healthy': state.get('circuit_state') == CircuitState.CLOSED.value, + 'degraded': state.get('degraded', False), + 'degraded_reason': state.get('degraded_reason'), 'circuit_opened_time': state.get('circuit_opened_time'), 'half_open_start_time': state.get('half_open_start_time') } diff --git a/src/plugin_system/plugin_manager.py b/src/plugin_system/plugin_manager.py index 3e677dbf1..22f8b470e 100644 --- a/src/plugin_system/plugin_manager.py +++ b/src/plugin_system/plugin_manager.py @@ -390,7 +390,15 @@ def load_plugin(self, plugin_id: str) -> bool: self.logger.error("Error validating plugin %s config: %s", plugin_id, e, exc_info=True) self.state_manager.set_state(plugin_id, PluginState.ERROR, error=e) return False - + + # Schema validation (warn/degrade only — never blocks loading). + # A config that violates the plugin's JSON schema is surfaced to the + # user (log warning + degraded flag in the health tracker) but the + # plugin still loads exactly as it does today. This deliberately does + # NOT change load_plugin()'s pass/fail behaviour for any plugin that + # loads under the current code. + self._validate_config_schema_soft(plugin_id, config) + # Store plugin instance self.plugins[plugin_id] = plugin_instance self.plugin_last_update[plugin_id] = 0.0 @@ -419,6 +427,59 @@ def load_plugin(self, plugin_id: str) -> bool: self.state_manager.set_state(plugin_id, PluginState.ERROR, error=e) return False + def _validate_config_schema_soft(self, plugin_id: str, config: Dict[str, Any]) -> None: + """Validate a plugin's config against its JSON schema — warn/degrade only. + + On a schema violation this logs a warning and marks the plugin degraded + in the health tracker (when one is wired), so the problem is visible in + the web UI. It never raises, never changes plugin state, and never + affects whether the plugin loads. ``config`` here has already been + merged with schema defaults by the caller, so fields that ship a default + never appear "missing" — only genuinely user-supplied required fields + (e.g. an API key) can trip the required-field check. + """ + try: + schema = self.schema_manager.load_schema(plugin_id) + except Exception as e: # pragma: no cover - defensive + self.logger.debug("Could not load schema for %s: %s", plugin_id, e) + return + + if not schema: + # No schema shipped — nothing to validate. Clear any stale flag. + self._set_degraded_safe(plugin_id, None) + return + + try: + is_valid, errors = self.schema_manager.validate_config_against_schema( + config, schema, plugin_id + ) + except Exception as e: # pragma: no cover - defensive + # Validation machinery itself failed — do not penalise the plugin. + self.logger.debug("Schema validation raised for %s: %s", plugin_id, e) + return + + if is_valid or not errors: + self._set_degraded_safe(plugin_id, None) + return + + summary = "; ".join(errors[:5]) + if len(errors) > 5: + summary += f" (+{len(errors) - 5} more)" + self.logger.warning( + "Plugin %s config does not match its schema (loading anyway): %s", + plugin_id, summary, + ) + self._set_degraded_safe(plugin_id, f"Config schema: {summary}") + + def _set_degraded_safe(self, plugin_id: str, reason: Optional[str]) -> None: + """Best-effort ``health_tracker.set_degraded`` that never raises.""" + if not self.health_tracker: + return + try: + self.health_tracker.set_degraded(plugin_id, reason) + except Exception as e: # pragma: no cover - defensive + self.logger.debug("Could not set degraded flag for %s: %s", plugin_id, e) + def unload_plugin(self, plugin_id: str) -> bool: """ Unload a plugin by ID. @@ -836,7 +897,7 @@ def get_plugin_health_metrics(self) -> Dict[str, Any]: # Get health tracker metrics if available if self.health_tracker: - health_info = self.health_tracker.get_plugin_health(plugin_id) + health_info = self.health_tracker.get_health_summary(plugin_id) plugin_metrics['health'] = health_info else: plugin_metrics['health'] = {'status': 'unknown'} @@ -861,7 +922,7 @@ def get_plugin_resource_metrics(self) -> Dict[str, Any]: # Get resource monitor metrics if available if self.resource_monitor: - resource_info = self.resource_monitor.get_plugin_metrics(plugin_id) + resource_info = self.resource_monitor.get_metrics_summary(plugin_id) plugin_metrics['resources'] = resource_info else: plugin_metrics['resources'] = {'status': 'unknown'} diff --git a/src/plugin_system/resource_monitor.py b/src/plugin_system/resource_monitor.py index 31352a55b..42ded4ec0 100644 --- a/src/plugin_system/resource_monitor.py +++ b/src/plugin_system/resource_monitor.py @@ -71,17 +71,32 @@ def __init__(self, cache_manager, enable_monitoring: bool = True): self.cache_manager = cache_manager self.enable_monitoring = enable_monitoring and PSUTIL_AVAILABLE self.logger = logging.getLogger(__name__) - + # Resource metrics per plugin self._metrics: Dict[str, ResourceMetrics] = {} self._limits: Dict[str, ResourceLimits] = {} - + # Thread-local storage for execution tracking self._local = threading.local() - + # Lock for thread-safe access self._lock = threading.Lock() - + + # Cache a single psutil.Process handle. Reusing the same handle is what + # lets cpu_percent() be read non-blocking (interval=None): psutil returns + # the utilisation since the *previous* call on that same object. Creating + # a fresh Process() per call would force interval-based sampling that + # blocks the caller — unacceptable on the display loop's update path. + self._process = None + if self.enable_monitoring: + try: + self._process = psutil.Process() + # Prime cpu_percent so the first real measurement returns a + # meaningful delta instead of 0.0. + self._process.cpu_percent(interval=None) + except Exception: # pragma: no cover - psutil edge cases + self._process = None + if not PSUTIL_AVAILABLE and enable_monitoring: self.logger.warning( "psutil not available - resource monitoring will be limited to execution time only" @@ -95,13 +110,21 @@ def _get_limits_key(self, plugin_id: str) -> str: """Get cache key for plugin limits.""" return f"plugin_limits:{plugin_id}" - def get_metrics(self, plugin_id: str) -> ResourceMetrics: - """Get current metrics for a plugin.""" + def get_metrics(self, plugin_id: str, force_reload: bool = False) -> ResourceMetrics: + """Get current metrics for a plugin. + + ``force_reload=True`` bypasses both the in-memory copy and the cache + manager's memory tier so a read-only consumer (e.g. the web process) + sees the writer process's latest persisted metrics rather than a stale + first snapshot. + """ with self._lock: - if plugin_id not in self._metrics: + if force_reload or plugin_id not in self._metrics: # Try to load from cache cache_key = self._get_metrics_key(plugin_id) - cached = self.cache_manager.get(cache_key, max_age=None) + cached = self.cache_manager.get( + cache_key, max_age=None, memory_ttl=0 if force_reload else None + ) if cached: metrics = ResourceMetrics(**cached) else: @@ -137,21 +160,24 @@ def get_limits(self, plugin_id: str) -> Optional[ResourceLimits]: def _get_process_memory_mb(self) -> float: """Get current process memory usage in MB.""" - if not self.enable_monitoring: + if not self.enable_monitoring or self._process is None: return 0.0 try: - process = psutil.Process() - return process.memory_info().rss / 1024 / 1024 + return self._process.memory_info().rss / 1024 / 1024 except Exception: return 0.0 - - def _get_process_cpu_percent(self, interval: float = 0.1) -> float: - """Get current process CPU usage percentage.""" - if not self.enable_monitoring: + + def _get_process_cpu_percent(self) -> float: + """Get current process CPU usage percentage (non-blocking). + + Reads cpu_percent(interval=None) against the cached process handle, so + it returns immediately with the utilisation observed since the previous + call rather than blocking to sample a fresh interval. + """ + if not self.enable_monitoring or self._process is None: return 0.0 try: - process = psutil.Process() - return process.cpu_percent(interval=interval) + return self._process.cpu_percent(interval=None) except Exception: return 0.0 @@ -281,9 +307,13 @@ def _check_limits(self, plugin_id: str, metrics: ResourceMetrics, self.logger.error(error_msg) raise ResourceLimitExceeded(error_msg) - def get_metrics_summary(self, plugin_id: str) -> Dict[str, Any]: - """Get metrics summary for a plugin.""" - metrics = self.get_metrics(plugin_id) + def get_metrics_summary(self, plugin_id: str, force_reload: bool = False) -> Dict[str, Any]: + """Get metrics summary for a plugin. + + ``force_reload=True`` refreshes from the persisted cache first so + cross-process readers reflect the writer's latest metrics. + """ + metrics = self.get_metrics(plugin_id, force_reload=force_reload) limits = self.get_limits(plugin_id) avg_execution_time = 0.0 diff --git a/test/test_cache_manager.py b/test/test_cache_manager.py index 034175e76..63a8d2e61 100644 --- a/test/test_cache_manager.py +++ b/test/test_cache_manager.py @@ -279,10 +279,23 @@ def test_get_expired(self, tmp_path): """Test getting expired cache entry.""" cache = DiskCache(cache_dir=str(tmp_path)) cache.set("test_key", {"data": "value"}) - + # Get with max_age=0 to force expiration result = cache.get("test_key", max_age=0) assert result is None + + def test_get_max_age_none_never_expires(self, tmp_path): + """max_age=None must return persisted records regardless of age. + + Regression: the age comparison raised TypeError for max_age=None, + which was swallowed and treated as a miss — silently breaking + long-lived state (plugin health/metrics) read across processes. + """ + cache = DiskCache(cache_dir=str(tmp_path)) + cache.set("test_key", {"data": "value", "timestamp": 0}) # epoch → very old + result = cache.get("test_key", max_age=None) + assert result is not None + assert result["data"] == "value" def test_get_nonexistent(self, tmp_path): """Test getting non-existent key.""" diff --git a/test/test_display_controller.py b/test/test_display_controller.py index 34d5ce262..782ce69e3 100644 --- a/test/test_display_controller.py +++ b/test/test_display_controller.py @@ -391,3 +391,16 @@ def test_inactive_hours(self, test_display_controller): controller._check_schedule() assert controller.is_display_active is False + + +class TestPluginHealthWiring: + """Phase 1: DisplayController activates the dormant plugin health/metrics + subsystem by wiring real tracker/monitor instances onto the plugin manager.""" + + def test_health_tracker_and_resource_monitor_wired(self, test_display_controller): + from src.plugin_system.plugin_health import PluginHealthTracker + from src.plugin_system.resource_monitor import PluginResourceMonitor + + pm = test_display_controller.plugin_manager + assert isinstance(pm.health_tracker, PluginHealthTracker) + assert isinstance(pm.resource_monitor, PluginResourceMonitor) diff --git a/test/test_plugin_health.py b/test/test_plugin_health.py new file mode 100644 index 000000000..fd647a4b9 --- /dev/null +++ b/test/test_plugin_health.py @@ -0,0 +1,93 @@ +""" +Tests for src/plugin_system/plugin_health.py + +Focus on the additive ``set_degraded`` mechanism used by the warn-only schema +validation path: it must surface a degraded reason without touching the circuit +breaker or causing the plugin to be skipped. +""" + +from unittest.mock import MagicMock + +from src.plugin_system.plugin_health import PluginHealthTracker, CircuitState + + +def _cache(): + cache = MagicMock() + cache.get.return_value = None + return cache + + +def test_set_degraded_marks_and_surfaces_reason(): + tracker = PluginHealthTracker(_cache()) + tracker.set_degraded("p", "bad config") + summary = tracker.get_health_summary("p") + assert summary["degraded"] is True + assert summary["degraded_reason"] == "bad config" + + +def test_set_degraded_none_clears(): + tracker = PluginHealthTracker(_cache()) + tracker.set_degraded("p", "bad config") + tracker.set_degraded("p", None) + summary = tracker.get_health_summary("p") + assert summary["degraded"] is False + assert summary["degraded_reason"] is None + + +def test_set_degraded_does_not_affect_circuit_breaker(): + tracker = PluginHealthTracker(_cache()) + tracker.set_degraded("p", "bad config") + summary = tracker.get_health_summary("p") + # Degraded is a *separate* signal from circuit health: the plugin is not + # counted as failing, the circuit stays closed, and it is not skipped. + assert summary["circuit_state"] == CircuitState.CLOSED.value + assert summary["consecutive_failures"] == 0 + assert summary["is_healthy"] is True + assert tracker.should_skip_plugin("p") is False + + +def test_set_degraded_skips_redundant_cache_write(): + cache = _cache() + tracker = PluginHealthTracker(cache) + tracker.set_degraded("p", "x") + writes_after_first = cache.set.call_count + assert writes_after_first >= 1 + tracker.set_degraded("p", "x") # unchanged → no extra write + assert cache.set.call_count == writes_after_first + + +def test_default_summary_has_degraded_fields(): + tracker = PluginHealthTracker(_cache()) + summary = tracker.get_health_summary("never-seen") + assert summary["degraded"] is False + assert summary["degraded_reason"] is None + + +def test_force_reload_refreshes_stale_in_memory_snapshot(): + """A long-lived reader (e.g. the web process) must not be pinned to the + first snapshot: force_reload re-reads persisted state and bypasses the + cache manager's memory tier so cross-process updates are visible.""" + cache = _cache() + tracker = PluginHealthTracker(cache) + + # First read snapshots an empty (healthy) state into the in-memory copy. + assert tracker.get_health_summary("p")["consecutive_failures"] == 0 + + # The display service later persists a failing/open state. + cache.get.return_value = { + "consecutive_failures": 5, + "circuit_state": "open", + "total_failures": 5, + "total_successes": 0, + } + + # A plain read is still pinned to the stale snapshot... + assert tracker.get_health_summary("p")["consecutive_failures"] == 0 + + # ...but force_reload observes the new persisted state. + fresh = tracker.get_health_summary("p", force_reload=True) + assert fresh["consecutive_failures"] == 5 + assert fresh["circuit_state"] == "open" + + # and it asked the cache to bypass the in-memory tier (memory_ttl=0). + assert any(c.kwargs.get("memory_ttl") == 0 for c in cache.get.call_args_list) diff --git a/test/test_plugin_manager_schema_soft.py b/test/test_plugin_manager_schema_soft.py new file mode 100644 index 000000000..5a77281b8 --- /dev/null +++ b/test/test_plugin_manager_schema_soft.py @@ -0,0 +1,79 @@ +""" +Tests for PluginManager._validate_config_schema_soft (Phase 1, warn-only schema +validation). + +Contract: +- A schema violation logs a warning and marks the plugin degraded in the health + tracker, but never raises and never changes load pass/fail behaviour. +- A valid config (or no schema) clears any stale degraded flag. +- The method is safe when no health tracker is wired. +""" + +import tempfile +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from src.plugin_system.plugin_manager import PluginManager + + +@pytest.fixture +def pm(): + with tempfile.TemporaryDirectory() as tmp: + manager = PluginManager(plugins_dir=str(Path(tmp) / "plugins")) + manager.schema_manager = MagicMock() + yield manager + + +def test_invalid_config_marks_degraded_without_raising(pm): + pm.health_tracker = MagicMock() + pm.schema_manager.load_schema.return_value = {"type": "object"} + pm.schema_manager.validate_config_against_schema.return_value = ( + False, + ["Missing required field: 'api_key'"], + ) + + pm._validate_config_schema_soft("youtube-stats", {}) + + pm.health_tracker.set_degraded.assert_called_once() + plugin_id, reason = pm.health_tracker.set_degraded.call_args[0] + assert plugin_id == "youtube-stats" + assert "api_key" in reason + + +def test_valid_config_clears_degraded(pm): + pm.health_tracker = MagicMock() + pm.schema_manager.load_schema.return_value = {"type": "object"} + pm.schema_manager.validate_config_against_schema.return_value = (True, []) + + pm._validate_config_schema_soft("p", {"api_key": "x"}) + + pm.health_tracker.set_degraded.assert_called_once_with("p", None) + + +def test_no_schema_clears_degraded(pm): + pm.health_tracker = MagicMock() + pm.schema_manager.load_schema.return_value = None + + pm._validate_config_schema_soft("p", {}) + + pm.health_tracker.set_degraded.assert_called_once_with("p", None) + + +def test_validation_exception_is_swallowed(pm): + pm.health_tracker = MagicMock() + pm.schema_manager.load_schema.return_value = {"type": "object"} + pm.schema_manager.validate_config_against_schema.side_effect = RuntimeError("boom") + + # Must not raise — the validation machinery failing must never break loading. + pm._validate_config_schema_soft("p", {}) + + +def test_safe_without_health_tracker(pm): + pm.health_tracker = None + pm.schema_manager.load_schema.return_value = {"type": "object"} + pm.schema_manager.validate_config_against_schema.return_value = (False, ["err"]) + + # Must not raise even though there is no tracker to record against. + pm._validate_config_schema_soft("p", {}) diff --git a/test/test_resource_monitor.py b/test/test_resource_monitor.py new file mode 100644 index 000000000..6b3dbe48a --- /dev/null +++ b/test/test_resource_monitor.py @@ -0,0 +1,129 @@ +""" +Tests for src/plugin_system/resource_monitor.py + +Focus areas: +- Execution-time metrics are captured regardless of psutil availability. +- CPU sampling is non-blocking (regression guard for the previous + ``cpu_percent(interval=0.1)`` call that blocked 100 ms per monitored call). +- Resource limits are enforced. +""" + +import time + +import pytest +from unittest.mock import MagicMock + +from src.plugin_system.resource_monitor import ( + PluginResourceMonitor, + ResourceLimits, + ResourceLimitExceeded, + PSUTIL_AVAILABLE, +) + + +def _cache(): + cache = MagicMock() + cache.get.return_value = None + return cache + + +class TestExecutionTimeMetrics: + def test_monitor_call_returns_value_and_records_call(self): + mon = PluginResourceMonitor(_cache(), enable_monitoring=False) + result = mon.monitor_call("p", lambda: 42) + assert result == 42 + metrics = mon.get_metrics("p") + assert metrics.call_count == 1 + assert metrics.total_execution_time >= 0.0 + + def test_avg_and_max_execution_time(self): + mon = PluginResourceMonitor(_cache(), enable_monitoring=False) + mon.monitor_call("p", lambda: time.sleep(0.01)) + mon.monitor_call("p", lambda: None) + summary = mon.get_metrics_summary("p") + assert summary["call_count"] == 2 + assert summary["max_execution_time"] >= summary["avg_execution_time"] >= 0.0 + + def test_exception_propagates_but_is_still_timed(self): + mon = PluginResourceMonitor(_cache(), enable_monitoring=False) + + def boom(): + raise ValueError("nope") + + with pytest.raises(ValueError): + mon.monitor_call("p", boom) + # Execution time is still recorded even when the call raised. + assert mon.get_metrics("p").execution_time >= 0.0 + + +class TestNonBlockingCpu: + def test_cpu_sampling_is_fast_when_disabled(self): + mon = PluginResourceMonitor(_cache(), enable_monitoring=False) + start = time.time() + for _ in range(50): + mon._get_process_cpu_percent() + # The old implementation blocked ~0.1s/call (~5s for 50). Non-blocking + # must complete near-instantly. + assert time.time() - start < 0.5 + assert mon._get_process_cpu_percent() == 0.0 + + @pytest.mark.skipif(not PSUTIL_AVAILABLE, reason="psutil not installed") + def test_cpu_sampling_is_fast_with_psutil(self): + mon = PluginResourceMonitor(_cache(), enable_monitoring=True) + assert mon._process is not None + start = time.time() + for _ in range(30): + mon._get_process_cpu_percent() + # 30 blocking 0.1s samples would be ~3s; non-blocking must be well under. + assert time.time() - start < 0.5 + + def test_monitor_call_does_not_block_on_cpu_sampling(self): + mon = PluginResourceMonitor(_cache()) # enable depends on psutil + start = time.time() + for _ in range(25): + mon.monitor_call("p", lambda: None) + # 25 * 0.1s = 2.5s under the old blocking bug; must be far faster now. + assert time.time() - start < 1.0 + + +class TestResourceLimits: + def test_execution_time_limit_raises(self): + mon = PluginResourceMonitor(_cache(), enable_monitoring=False) + mon.set_limits("p", ResourceLimits(max_execution_time=0.001)) + with pytest.raises(ResourceLimitExceeded): + mon.monitor_call("p", lambda: time.sleep(0.02)) + + def test_reset_metrics_clears_counts(self): + cache = _cache() + mon = PluginResourceMonitor(cache, enable_monitoring=False) + mon.monitor_call("p", lambda: None) + assert mon.get_metrics("p").call_count == 1 + mon.reset_metrics("p") + assert mon.get_metrics("p").call_count == 0 + + +class TestForceReload: + def test_force_reload_refreshes_stale_snapshot(self): + """A read-only consumer must see the writer process's latest persisted + metrics rather than a pinned first snapshot.""" + cache = MagicMock() + persisted = {"value": None} # only the metrics key returns data + + def cache_get(key, max_age=None, memory_ttl=None): + return persisted["value"] if key.startswith("plugin_metrics:") else None + + cache.get.side_effect = cache_get + mon = PluginResourceMonitor(cache, enable_monitoring=False) + + # First read snapshots empty metrics. + assert mon.get_metrics_summary("p")["call_count"] == 0 + + # The display service later persists real metrics. + persisted["value"] = {"call_count": 7, "total_execution_time": 1.4} + + # Plain read stays stale... + assert mon.get_metrics_summary("p")["call_count"] == 0 + # ...force_reload picks up the persisted values and bypasses memory. + 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) diff --git a/test/test_web_api.py b/test/test_web_api.py index 103ab9891..269a85253 100644 --- a/test/test_web_api.py +++ b/test/test_web_api.py @@ -759,3 +759,54 @@ def test_save_plugin_config_none_array_gets_default(self, client, mock_config_ma teams = soccer_cfg.get('leagues', {}).get('eng.1', {}).get('favorite_teams') assert isinstance(teams, list), f"Expected list, got: {type(teams)}" assert teams == [], f"Expected empty default list, got: {teams}" + + +class TestPluginHealthRoutes: + """Phase 1: /plugins/health and /plugins/metrics build per-installed-id so + they surface cross-process data persisted by the display service.""" + + def test_health_route_builds_per_installed_id(self, client, mock_plugin_manager): + from web_interface.blueprints.api_v3 import api_v3 + from src.plugin_system.plugin_health import PluginHealthTracker + + cache = MagicMock() + cache.get.return_value = None + api_v3.plugin_manager = mock_plugin_manager + mock_plugin_manager.plugin_manifests = {'p1': {}, 'p2': {}} + mock_plugin_manager.health_tracker = PluginHealthTracker(cache) + + resp = client.get('/api/v3/plugins/health') + assert resp.status_code == 200 + data = resp.get_json()['data'] + assert set(data.keys()) == {'p1', 'p2'} + assert data['p1']['is_healthy'] is True + assert data['p1']['degraded'] is False + + def test_health_route_reports_not_available_without_tracker(self, client, mock_plugin_manager): + from web_interface.blueprints.api_v3 import api_v3 + api_v3.plugin_manager = mock_plugin_manager + mock_plugin_manager.health_tracker = None + + resp = client.get('/api/v3/plugins/health') + assert resp.status_code == 200 + body = resp.get_json() + assert body['data'] == {} + assert 'not available' in body['message'].lower() + + def test_metrics_route_builds_per_installed_id(self, client, mock_plugin_manager): + from web_interface.blueprints.api_v3 import api_v3 + from src.plugin_system.resource_monitor import PluginResourceMonitor + + cache = MagicMock() + cache.get.return_value = None + api_v3.plugin_manager = mock_plugin_manager + mock_plugin_manager.plugin_manifests = {'p1': {}} + mock_plugin_manager.resource_monitor = PluginResourceMonitor( + cache, enable_monitoring=False + ) + + resp = client.get('/api/v3/plugins/metrics') + assert resp.status_code == 200 + data = resp.get_json()['data'] + assert 'p1' in data + assert data['p1']['call_count'] == 0 diff --git a/web_interface/app.py b/web_interface/app.py index af795a465..6b16daf2c 100644 --- a/web_interface/app.py +++ b/web_interface/app.py @@ -160,6 +160,22 @@ from src.cache_manager import CacheManager api_v3.cache_manager = CacheManager() +# Wire plugin health/metrics for the web process. The display service records +# health and execution-time metrics to the shared on-disk cache; giving the web +# process its own tracker/monitor backed by that same cache lets the health API +# routes (/api/v3/plugins/health, /plugins/metrics) read that persisted data. +# Guarded so any init failure degrades to "not available" rather than breaking +# the web server. +try: + from src.plugin_system.plugin_health import PluginHealthTracker + from src.plugin_system.resource_monitor import PluginResourceMonitor + plugin_manager.health_tracker = PluginHealthTracker(api_v3.cache_manager) + plugin_manager.resource_monitor = PluginResourceMonitor(api_v3.cache_manager) +except Exception as _hm_err: # pragma: no cover - defensive startup guard + logging.getLogger(__name__).warning( + "Could not enable plugin health/metrics for web UI: %s", _hm_err + ) + app.register_blueprint(pages_v3, url_prefix='/v3') app.register_blueprint(api_v3, url_prefix='/api/v3') diff --git a/web_interface/blueprints/api_v3.py b/web_interface/blueprints/api_v3.py index 0425590ee..e06037665 100644 --- a/web_interface/blueprints/api_v3.py +++ b/web_interface/blueprints/api_v3.py @@ -2073,6 +2073,18 @@ def _build_plugin_entry(plugin_info): return None def _build_plugin_entry_inner(plugin_info, plugin_id): + # Capture runtime state (state machine + error context) before the + # manifest merge below can shadow the 'state' key. get_all_plugin_info + # attaches this via PluginStateManager.get_state_info(); surfacing it + # lets the UI show *why* a plugin isn't running instead of just + # 'loaded: false'. + state_info = plugin_info.get('state') + plugin_state = None + plugin_error_info = None + if isinstance(state_info, dict): + plugin_state = state_info.get('state') + plugin_error_info = state_info.get('error_info') + # Re-read manifest from disk to ensure we have the latest metadata manifest_path = Path(api_v3.plugin_manager.plugins_dir) / plugin_id / "manifest.json" if manifest_path.exists(): @@ -2154,6 +2166,8 @@ def _build_plugin_entry_inner(plugin_info, plugin_id): 'enabled': enabled, 'verified': verified, 'loaded': plugin_info.get('loaded', False), + 'state': plugin_state, + 'error_info': plugin_error_info, 'last_updated': last_updated, 'last_commit': last_commit, 'last_commit_message': last_commit_message, @@ -2173,6 +2187,31 @@ def _build_plugin_entry_inner(plugin_info, plugin_id): logger.error('Error in get_installed_plugins', exc_info=True) return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details'}), 500 +def _installed_plugin_ids(): + """Best-effort list of installed plugin IDs for the web process. + + Health/metrics state is written by the separate display service to the + shared on-disk cache, so the tracker's in-memory set is empty here. We + enumerate the installed plugins and read each one's persisted summary by ID + instead of relying on the tracker's in-memory `get_all_*` view. + """ + pm = api_v3.plugin_manager + manifests = getattr(pm, 'plugin_manifests', None) + if not manifests: + # Only pay for a discovery scan when we haven't discovered anything yet; + # subsequent polls reuse the already-populated manifest map. + try: + pm.discover_plugins() + except Exception: + logger.debug('discover_plugins failed while listing plugin ids', exc_info=True) + manifests = getattr(pm, 'plugin_manifests', None) + try: + return list(manifests.keys()) if manifests else [] + except Exception: + logger.debug('listing plugin_manifests failed while building plugin ids', exc_info=True) + return [] + + @api_v3.route('/plugins/health', methods=['GET']) def get_plugin_health(): """Get health metrics for all plugins""" @@ -2188,8 +2227,23 @@ def get_plugin_health(): 'message': 'Health tracking not available' }) - # Get health summaries for all plugins - health_summaries = api_v3.plugin_manager.health_tracker.get_all_health_summaries() + tracker = api_v3.plugin_manager.health_tracker + # Build per-plugin summaries by ID so persisted (cross-process) health + # is included, then fold in any in-memory-only entries. + health_summaries = {} + for pid in _installed_plugin_ids(): + try: + # force_reload: this process only reads; bypass the in-memory + # snapshot so each poll reflects the display service's latest + # persisted state. + health_summaries[pid] = tracker.get_health_summary(pid, force_reload=True) + except Exception: + logger.debug('Could not read health summary for %s', pid, exc_info=True) + try: + for pid, summary in tracker.get_all_health_summaries().items(): + health_summaries.setdefault(pid, summary) + except Exception: + logger.debug('get_all_health_summaries failed', exc_info=True) return jsonify({ 'status': 'success', @@ -2264,8 +2318,22 @@ def get_plugin_metrics(): 'message': 'Resource monitoring not available' }) - # Get metrics summaries for all plugins - metrics_summaries = api_v3.plugin_manager.resource_monitor.get_all_metrics_summaries() + monitor = api_v3.plugin_manager.resource_monitor + # Build per-plugin summaries by ID so persisted (cross-process) metrics + # are included, then fold in any in-memory-only entries. + metrics_summaries = {} + for pid in _installed_plugin_ids(): + try: + # force_reload: read-only path — bypass the in-memory snapshot so + # each poll reflects the display service's latest persisted metrics. + metrics_summaries[pid] = monitor.get_metrics_summary(pid, force_reload=True) + except Exception: + logger.debug('Could not read metrics summary for %s', pid, exc_info=True) + try: + for pid, summary in monitor.get_all_metrics_summaries().items(): + metrics_summaries.setdefault(pid, summary) + except Exception: + logger.debug('get_all_metrics_summaries failed', exc_info=True) return jsonify({ 'status': 'success', diff --git a/web_interface/static/v3/js/plugins/api_client.js b/web_interface/static/v3/js/plugins/api_client.js index 427cc1e02..fd33ce125 100644 --- a/web_interface/static/v3/js/plugins/api_client.js +++ b/web_interface/static/v3/js/plugins/api_client.js @@ -340,11 +340,25 @@ const PluginAPI = { * @returns {Promise} Health data */ async getPluginHealth(pluginId = null) { - const endpoint = pluginId + const endpoint = pluginId ? `/plugins/health/${pluginId}` : '/plugins/health'; const response = await this.request(endpoint); return response.data || {}; + }, + + /** + * Get plugin resource metrics (execution time, memory, cpu). + * + * @param {string} pluginId - Optional plugin identifier (null for all) + * @returns {Promise} Metrics data keyed by plugin id + */ + async getPluginMetrics(pluginId = null) { + const endpoint = pluginId + ? `/plugins/metrics/${pluginId}` + : '/plugins/metrics'; + const response = await this.request(endpoint); + return response.data || {}; } }; diff --git a/web_interface/templates/v3/partials/tools.html b/web_interface/templates/v3/partials/tools.html index 7ac8cf878..83cf51b78 100644 --- a/web_interface/templates/v3/partials/tools.html +++ b/web_interface/templates/v3/partials/tools.html @@ -157,6 +157,38 @@

Services

+ +
+
+
+

Plugin Health

+

Circuit-breaker status and per-plugin update timings recorded by the display service. A plugin whose update() keeps failing is paused ("Circuit open") and retried automatically after a cooldown.

+
+ +
+ +
+ + + + + + + + + + + + + + +
PluginStatusAvg updateMax updateUpdatesLast error
Loading…
+
+
+