Files
LEDMatrix/src/plugin_system/plugin_manager.py
Chuck 2c2fca2219 fix(web): use HTMX for Plugin Manager tab loading (#294)
* fix: auto-repair missing plugins and graceful config fallback

Plugins whose directories are missing (failed update, migration, etc.)
now get automatically reinstalled from the store on startup. The config
endpoint no longer returns a hard 500 when a schema is unavailable —
it falls back to conservative key-name-based masking so the settings
page stays functional.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: handle ledmatrix- prefix in plugin updates and reconciliation

The store registry uses unprefixed IDs (e.g., 'weather') while older
installs used prefixed config keys (e.g., 'ledmatrix-weather'). Both
update_plugin() and auto-repair now try the unprefixed ID as a fallback
when the prefixed one isn't found in the registry.

Also filters system config keys (schedule, display, etc.) from
reconciliation to avoid false positives.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: address code review findings for plugin auto-repair

- Move backup-folder filter from _get_config_state to _get_disk_state
  where the artifact actually lives
- Run startup reconciliation in a background thread so requests aren't
  blocked by plugin reinstallation
- Set _reconciliation_done only after success so failures allow retries
- Replace print() with proper logger in reconciliation
- Wrap load_schema in try/except so exceptions fall through to
  conservative masking instead of 500
- Handle list values in _conservative_mask_config for nested secrets
- Remove duplicate import re

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: add thread-safe locking to PluginManager and fix reconciliation retry

PluginManager thread safety:
- Add RLock protecting plugin_manifests and plugin_directories
- Build scan results locally in _scan_directory_for_plugins, then update
  shared state under lock
- Protect reads in get_plugin_info, get_all_plugin_info,
  get_plugin_directory, get_plugin_display_modes, find_plugin_for_mode
- Protect manifest mutation in reload_plugin
- Prevents races between background reconciliation thread and request
  handlers reading plugin state

Reconciliation retry:
- Clear _reconciliation_started on exception so next request retries
- Check result.reconciliation_successful before marking done
- Reset _reconciliation_started on non-success results to allow retry

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(web): use HTMX for Plugin Manager tab loading instead of custom fetch

The Plugin Manager tab was the only tab using a custom window.loadPluginsTab()
function with plain fetch() instead of HTMX. This caused a race condition where
plugins_manager.js listened for htmx:afterSwap to initialize, but that event
never fired for the custom fetch. Users had to navigate to a plugin config tab
and back to trigger initialization.

Changes:
- Switch plugins tab to hx-get/hx-trigger="revealed" matching all other tabs
- Remove ~560 lines of dead code (script extraction for a partial with no scripts,
  nested retry intervals, inline HTML card rendering fallbacks)
- Add simple loadPluginsDirect() fallback for when HTMX fails to load
- Remove typeof htmx guard on afterSwap listener so it registers unconditionally
- Tighten afterSwap target check to avoid spurious re-init from other tab swaps

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: address CodeRabbit review findings across plugin system

- plugin_manager.py: clear plugin_manifests/plugin_directories before update
  to prevent ghost entries for uninstalled plugins persisting across scans
- state_reconciliation.py: remove 'enabled' key check that skipped legacy
  plugin configs, default to enabled=True matching PluginManager.load_plugin
- app.py: add threading.Lock around reconciliation start guard to prevent
  race condition spawning duplicate threads; add -> None return annotation
- store_manager.py: use resolved registry ID (alt_id) instead of original
  plugin_id when reinstalling during monorepo migration
- base.html: check Response.ok in loadPluginsDirect fallback; trigger
  fallback on tab click when HTMX unavailable; remove active-tab check
  from 5-second timeout so content preloads regardless

Skipped: api_v3.py secret redaction suggestion — the caller at line 2539
already tries schema-based mask_secret_fields() before falling back to
_conservative_mask_config, making the suggested change redundant.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: skip backup dirs in plugin discovery and fix HTMX event syntax

- plugin_manager.py: skip directories containing '.standalone-backup-'
  during discovery scan, matching state_reconciliation.py behavior and
  preventing backup manifests from overwriting live plugin entries
- base.html: fix hx-on::htmx:response-error → hx-on::response-error
  (the :: shorthand already adds the htmx: prefix, so the original
  syntax resolved to htmx:htmx:response-error making the handler dead)

Skipped findings:
- web-ui-info in _SYSTEM_CONFIG_KEYS: it's a real plugin with manifest.json
  and config entry, not a system key
- store_manager config key migration: valid feature request for handling
  ledmatrix- prefix rename, but new functionality outside this PR scope

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(web): add fetch timeout to loadPluginsDirect fallback

Add AbortController with 10s timeout so a hanging fetch doesn't leave
data-loaded set and block retries. Timer is cleared in both success
and error paths.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-26 15:21:33 -04:00

826 lines
32 KiB
Python

"""
Plugin Manager
Manages plugin discovery, loading, and lifecycle for the LEDMatrix system.
Handles dynamic plugin loading from the plugins/ directory.
API Version: 1.0.0
"""
import os
import json
import importlib
import importlib.util
import sys
import subprocess
import time
import threading
from pathlib import Path
from typing import Dict, List, Optional, Any
import logging
from src.exceptions import PluginError
from src.logging_config import get_logger
from src.plugin_system.plugin_loader import PluginLoader
from src.plugin_system.plugin_executor import PluginExecutor
from src.plugin_system.plugin_state import PluginStateManager, PluginState
from src.plugin_system.schema_manager import SchemaManager
from src.common.permission_utils import (
ensure_directory_permissions,
get_plugin_dir_mode
)
class PluginManager:
"""
Manages plugin discovery, loading, and lifecycle.
The PluginManager is responsible for:
- Discovering plugins in the plugins/ directory
- Loading plugin modules and instantiating plugin classes
- Managing plugin lifecycle (load, unload, reload)
- Providing access to loaded plugins
- Maintaining plugin manifests
Uses composition with specialized components:
- PluginLoader: Handles module loading and dependency installation
- PluginExecutor: Handles plugin execution with timeout and error isolation
- PluginStateManager: Manages plugin state machine
"""
def __init__(self, plugins_dir: str = "plugins",
config_manager: Optional[Any] = None,
display_manager: Optional[Any] = None,
cache_manager: Optional[Any] = None,
font_manager: Optional[Any] = None) -> None:
"""
Initialize the Plugin Manager.
Args:
plugins_dir: Path to the plugins directory
config_manager: Configuration manager instance
display_manager: Display manager instance
cache_manager: Cache manager instance
font_manager: Font manager instance
"""
self.plugins_dir: Path = Path(plugins_dir)
self.config_manager: Optional[Any] = config_manager
self.display_manager: Optional[Any] = display_manager
self.cache_manager: Optional[Any] = cache_manager
self.font_manager: Optional[Any] = font_manager
self.logger: logging.Logger = get_logger(__name__)
# Initialize plugin system components
self.plugin_loader = PluginLoader(logger=self.logger)
self.plugin_executor = PluginExecutor(default_timeout=30.0, logger=self.logger)
self.state_manager = PluginStateManager(logger=self.logger)
self.schema_manager = SchemaManager(plugins_dir=self.plugins_dir, logger=self.logger)
# Lock protecting plugin_manifests and plugin_directories from
# concurrent mutation (background reconciliation) and reads (requests).
self._discovery_lock = threading.RLock()
# Active plugins
self.plugins: Dict[str, Any] = {}
self.plugin_manifests: Dict[str, Dict[str, Any]] = {}
self.plugin_modules: Dict[str, Any] = {}
self.plugin_last_update: Dict[str, float] = {}
# Health tracking (optional, set by display_controller if available)
self.health_tracker = None
self.resource_monitor = None
# Ensure plugins directory exists with proper permissions
try:
ensure_directory_permissions(self.plugins_dir, get_plugin_dir_mode())
except (OSError, PermissionError) as e:
self.logger.error("Could not create plugins directory %s: %s", self.plugins_dir, e, exc_info=True)
raise PluginError(f"Could not create plugins directory: {self.plugins_dir}", context={'error': str(e)}) from e
def _scan_directory_for_plugins(self, directory: Path) -> List[str]:
"""
Scan a directory for plugins.
Args:
directory: Directory to scan
Returns:
List of plugin IDs found
"""
plugin_ids = []
if not directory.exists():
return plugin_ids
# Build new state locally before acquiring lock
new_manifests: Dict[str, Dict[str, Any]] = {}
new_directories: Dict[str, Path] = {}
try:
for item in directory.iterdir():
if not item.is_dir():
continue
# Skip backup directories so they don't overwrite live entries
if '.standalone-backup-' in item.name:
continue
manifest_path = item / "manifest.json"
if manifest_path.exists():
try:
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
plugin_id = manifest.get('id')
if plugin_id:
plugin_ids.append(plugin_id)
new_manifests[plugin_id] = manifest
new_directories[plugin_id] = item
except (json.JSONDecodeError, PermissionError, OSError) as e:
self.logger.warning("Error reading manifest from %s: %s", manifest_path, e, exc_info=True)
continue
except (OSError, PermissionError) as e:
self.logger.error("Error scanning directory %s: %s", directory, e, exc_info=True)
# Replace shared state under lock so uninstalled plugins don't linger
with self._discovery_lock:
self.plugin_manifests.clear()
self.plugin_manifests.update(new_manifests)
if not hasattr(self, 'plugin_directories'):
self.plugin_directories = {}
else:
self.plugin_directories.clear()
self.plugin_directories.update(new_directories)
return plugin_ids
def discover_plugins(self) -> List[str]:
"""
Discover all plugins in the plugins directory.
Also checks for potential config key collisions and logs warnings.
Returns:
List of plugin IDs
"""
self.logger.info("Discovering plugins in %s", self.plugins_dir)
plugin_ids = self._scan_directory_for_plugins(self.plugins_dir)
self.logger.info("Discovered %d plugin(s)", len(plugin_ids))
# Check for config key collisions
collisions = self.schema_manager.detect_config_key_collisions(plugin_ids)
for collision in collisions:
self.logger.warning(
"Config collision detected: %s",
collision.get('message', str(collision))
)
return plugin_ids
def _get_dependency_marker_path(self, plugin_id: str) -> Path:
"""Get path to dependency installation marker file."""
plugin_dir = self.plugins_dir / plugin_id
if not plugin_dir.exists():
# Try with ledmatrix- prefix
plugin_dir = self.plugins_dir / f"ledmatrix-{plugin_id}"
return plugin_dir / ".dependencies_installed"
def _check_dependencies_installed(self, plugin_id: str) -> bool:
"""Check if dependencies are already installed for a plugin."""
marker_path = self._get_dependency_marker_path(plugin_id)
return marker_path.exists()
def _mark_dependencies_installed(self, plugin_id: str) -> None:
"""Mark dependencies as installed for a plugin."""
marker_path = self._get_dependency_marker_path(plugin_id)
try:
marker_path.touch()
# Set proper file permissions after creating marker
from src.common.permission_utils import (
ensure_file_permissions,
get_plugin_file_mode
)
ensure_file_permissions(marker_path, get_plugin_file_mode())
except (OSError, PermissionError) as e:
self.logger.warning("Could not create dependency marker for %s: %s", plugin_id, e)
def _remove_dependency_marker(self, plugin_id: str) -> None:
"""Remove dependency installation marker."""
marker_path = self._get_dependency_marker_path(plugin_id)
try:
if marker_path.exists():
marker_path.unlink()
except (OSError, PermissionError) as e:
self.logger.warning("Could not remove dependency marker for %s: %s", plugin_id, e)
def _install_plugin_dependencies(self, requirements_file: Path) -> bool:
"""
Install plugin dependencies from requirements.txt.
Args:
requirements_file: Path to requirements.txt
Returns:
True if installation succeeded or not needed, False on error
"""
try:
self.logger.info("Installing dependencies from %s", requirements_file)
result = subprocess.run(
[sys.executable, "-m", "pip", "install", "--break-system-packages", "--no-cache-dir", "-r", str(requirements_file)],
capture_output=True,
text=True,
timeout=300,
check=False
)
if result.returncode == 0:
self.logger.info("Dependencies installed successfully")
return True
else:
self.logger.warning("Dependency installation returned non-zero exit code: %s", result.stderr)
return False
except subprocess.TimeoutExpired:
self.logger.error("Dependency installation timed out")
return False
except FileNotFoundError as e:
self.logger.warning("Command not found: %s. Skipping dependency installation", e)
return True
except (BrokenPipeError, OSError) as e:
# Handle broken pipe errors (errno 32) which can occur during pip downloads
# Often caused by network interruptions or output buffer issues
if isinstance(e, OSError) and e.errno == 32:
self.logger.error(
"Broken pipe error during dependency installation. "
"This usually indicates a network interruption or pip output buffer issue. "
"Try installing again or check your network connection."
)
else:
self.logger.error("OS error during dependency installation: %s", e)
return False
except Exception as e:
self.logger.error("Unexpected error installing dependencies: %s", e, exc_info=True)
return True
def load_plugin(self, plugin_id: str) -> bool:
"""
Load a plugin by ID.
This method:
1. Checks if plugin is already loaded
2. Validates the manifest exists
3. Uses PluginLoader to import module and instantiate plugin
4. Validates the plugin configuration
5. Stores the plugin instance
6. Updates plugin state
Args:
plugin_id: Plugin identifier
Returns:
True if loaded successfully, False otherwise
"""
if plugin_id in self.plugins:
self.logger.warning("Plugin %s already loaded", plugin_id)
return True
manifest = self.plugin_manifests.get(plugin_id)
if not manifest:
self.logger.error("No manifest found for plugin: %s", plugin_id)
self.state_manager.set_state(plugin_id, PluginState.ERROR)
return False
try:
# Update state to LOADED
self.state_manager.set_state(plugin_id, PluginState.LOADED)
# Find plugin directory using PluginLoader
plugin_directories = getattr(self, 'plugin_directories', None)
plugin_dir = self.plugin_loader.find_plugin_directory(
plugin_id,
self.plugins_dir,
plugin_directories
)
if plugin_dir is None:
self.logger.error("Plugin directory not found: %s", plugin_id)
self.logger.error("Searched in: %s", self.plugins_dir)
self.state_manager.set_state(plugin_id, PluginState.ERROR)
return False
# Update mapping if found via search
if plugin_directories is None or plugin_id not in plugin_directories:
if not hasattr(self, 'plugin_directories'):
self.plugin_directories = {}
self.plugin_directories[plugin_id] = plugin_dir
# Get plugin config
if self.config_manager:
full_config = self.config_manager.load_config()
config = full_config.get(plugin_id, {})
else:
config = {}
# Check if plugin has a config schema
schema_path = self.schema_manager.get_schema_path(plugin_id)
if schema_path is None:
# Schema file doesn't exist
self.logger.warning(
f"Plugin '{plugin_id}' has no config_schema.json - configuration will not be validated. "
f"Consider adding a schema file for better error detection and user experience."
)
else:
# Schema file exists, try to load it
schema = self.schema_manager.load_schema(plugin_id)
if schema is None:
# Schema exists but couldn't be loaded (likely invalid JSON or schema)
self.logger.warning(
f"Plugin '{plugin_id}' has a config_schema.json but it could not be loaded. "
f"The schema may be invalid. Please verify the schema file at: {schema_path}"
)
# Merge config with schema defaults to ensure all defaults are applied
try:
defaults = self.schema_manager.generate_default_config(plugin_id, use_cache=True)
config = self.schema_manager.merge_with_defaults(config, defaults)
self.logger.debug(f"Merged config with schema defaults for {plugin_id}")
except Exception as e:
self.logger.warning(f"Could not apply schema defaults for {plugin_id}: {e}")
# Continue with original config if defaults can't be applied
# Use PluginLoader to load plugin
plugin_instance, module = self.plugin_loader.load_plugin(
plugin_id=plugin_id,
manifest=manifest,
plugin_dir=plugin_dir,
config=config,
display_manager=self.display_manager,
cache_manager=self.cache_manager,
plugin_manager=self,
install_deps=True
)
# Store module
self.plugin_modules[plugin_id] = module
# Validate configuration
if hasattr(plugin_instance, 'validate_config'):
try:
if not plugin_instance.validate_config():
self.logger.error("Plugin %s configuration validation failed", plugin_id)
self.state_manager.set_state(plugin_id, PluginState.ERROR)
return False
except Exception as e:
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
# Store plugin instance
self.plugins[plugin_id] = plugin_instance
self.plugin_last_update[plugin_id] = 0.0
# Update state based on enabled status
if config.get('enabled', True):
self.state_manager.set_state(plugin_id, PluginState.ENABLED)
# Call on_enable if plugin is enabled
if hasattr(plugin_instance, 'on_enable'):
plugin_instance.on_enable()
else:
self.state_manager.set_state(plugin_id, PluginState.DISABLED)
self.logger.info("Loaded plugin: %s", plugin_id)
return True
except PluginError as e:
self.logger.error("Plugin error loading %s: %s", plugin_id, e, exc_info=True)
self.state_manager.set_state(plugin_id, PluginState.ERROR, error=e)
return False
except Exception as e:
self.logger.error("Unexpected error loading plugin %s: %s", plugin_id, e, exc_info=True)
self.state_manager.set_state(plugin_id, PluginState.ERROR, error=e)
return False
def unload_plugin(self, plugin_id: str) -> bool:
"""
Unload a plugin by ID.
Args:
plugin_id: Plugin identifier
Returns:
True if unloaded successfully, False otherwise
"""
if plugin_id not in self.plugins:
self.logger.warning("Plugin %s not loaded", plugin_id)
return False
try:
plugin = self.plugins[plugin_id]
# Call cleanup if available
if hasattr(plugin, 'cleanup'):
try:
plugin.cleanup()
except Exception as e:
self.logger.warning("Error during plugin cleanup: %s", e)
# Call on_disable if available
if hasattr(plugin, 'on_disable'):
try:
plugin.on_disable()
except Exception as e:
self.logger.warning("Error during plugin on_disable: %s", e)
# Remove from active plugins
del self.plugins[plugin_id]
if plugin_id in self.plugin_last_update:
del self.plugin_last_update[plugin_id]
# Remove main module from sys.modules if present
module_name = f"plugin_{plugin_id.replace('-', '_')}"
sys.modules.pop(module_name, None)
# Delegate sub-module and cached-module cleanup to the loader
self.plugin_loader.unregister_plugin_modules(plugin_id)
# Remove from plugin_modules
self.plugin_modules.pop(plugin_id, None)
# Update state
self.state_manager.set_state(plugin_id, PluginState.UNLOADED)
self.state_manager.clear_state(plugin_id)
self.logger.info("Unloaded plugin: %s", plugin_id)
return True
except Exception as e:
self.logger.error("Error unloading plugin %s: %s", plugin_id, e, exc_info=True)
self.state_manager.set_state(plugin_id, PluginState.ERROR, error=e)
return False
def reload_plugin(self, plugin_id: str) -> bool:
"""
Reload a plugin (unload and load).
Args:
plugin_id: Plugin identifier
Returns:
True if reloaded successfully, False otherwise
"""
self.logger.info("Reloading plugin: %s", plugin_id)
# Unload first
if plugin_id in self.plugins:
if not self.unload_plugin(plugin_id):
return False
# Re-discover to get updated manifest
manifest_path = self.plugins_dir / plugin_id / "manifest.json"
if manifest_path.exists():
try:
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
with self._discovery_lock:
self.plugin_manifests[plugin_id] = manifest
except Exception as e:
self.logger.error("Error reading manifest: %s", e, exc_info=True)
return False
return self.load_plugin(plugin_id)
def get_plugin(self, plugin_id: str) -> Optional[Any]:
"""
Get a loaded plugin instance by ID.
Args:
plugin_id: Plugin identifier
Returns:
Plugin instance or None if not loaded
"""
return self.plugins.get(plugin_id)
def get_all_plugins(self) -> Dict[str, Any]:
"""
Get all loaded plugins.
Returns:
Dict of plugin_id: plugin_instance
"""
return self.plugins.copy()
def get_enabled_plugins(self) -> List[str]:
"""
Get list of enabled plugin IDs.
Returns:
List of plugin IDs that are currently enabled
"""
return [pid for pid, plugin in self.plugins.items() if plugin.enabled]
def get_plugin_info(self, plugin_id: str) -> Optional[Dict[str, Any]]:
"""
Get information about a plugin (manifest + runtime info).
Args:
plugin_id: Plugin identifier
Returns:
Dict with plugin information or None if not found
"""
with self._discovery_lock:
manifest = self.plugin_manifests.get(plugin_id)
if not manifest:
return None
info = manifest.copy()
# Add runtime information if plugin is loaded
plugin = self.plugins.get(plugin_id)
if plugin:
info['loaded'] = True
if hasattr(plugin, 'get_info'):
info['runtime_info'] = plugin.get_info()
else:
info['loaded'] = False
# Add state information
info['state'] = self.state_manager.get_state_info(plugin_id)
return info
def get_all_plugin_info(self) -> List[Dict[str, Any]]:
"""
Get information about all plugins.
Returns:
List of plugin info dictionaries
"""
with self._discovery_lock:
pids = list(self.plugin_manifests.keys())
return [info for info in [self.get_plugin_info(pid) for pid in pids] if info]
def get_plugin_directory(self, plugin_id: str) -> Optional[str]:
"""
Get the directory path for a plugin.
Args:
plugin_id: Plugin identifier
Returns:
Directory path as string or None if not found
"""
with self._discovery_lock:
if hasattr(self, 'plugin_directories') and plugin_id in self.plugin_directories:
return str(self.plugin_directories[plugin_id])
plugin_dir = self.plugins_dir / plugin_id
if plugin_dir.exists():
return str(plugin_dir)
plugin_dir = self.plugins_dir / f"ledmatrix-{plugin_id}"
if plugin_dir.exists():
return str(plugin_dir)
return None
def get_plugin_display_modes(self, plugin_id: str) -> List[str]:
"""
Get display modes provided by a plugin.
Args:
plugin_id: Plugin identifier
Returns:
List of display mode names
"""
with self._discovery_lock:
manifest = self.plugin_manifests.get(plugin_id)
if not manifest:
return []
display_modes = manifest.get('display_modes', [])
if isinstance(display_modes, list):
return display_modes
return []
def find_plugin_for_mode(self, mode: str) -> Optional[str]:
"""
Find which plugin provides a given display mode.
Args:
mode: Display mode identifier
Returns:
Plugin identifier or None if not found.
"""
normalized_mode = mode.strip().lower()
with self._discovery_lock:
manifests_snapshot = dict(self.plugin_manifests)
for plugin_id, manifest in manifests_snapshot.items():
display_modes = manifest.get('display_modes')
if isinstance(display_modes, list) and display_modes:
if any(m.lower() == normalized_mode for m in display_modes):
return plugin_id
return None
def _get_plugin_update_interval(self, plugin_id: str, plugin_instance: Any) -> Optional[float]:
"""
Get the update interval for a plugin.
Args:
plugin_id: Plugin identifier
plugin_instance: Plugin instance
Returns:
Update interval in seconds or None if not configured
"""
# Check manifest first
manifest = self.plugin_manifests.get(plugin_id, {})
update_interval = manifest.get('update_interval')
if update_interval:
try:
return float(update_interval)
except (ValueError, TypeError):
pass
# Check plugin config
if self.config_manager:
try:
config = self.config_manager.get_config()
plugin_config = config.get(plugin_id, {})
update_interval = plugin_config.get('update_interval')
if update_interval:
try:
return float(update_interval)
except (ValueError, TypeError):
pass
except Exception as e:
self.logger.debug("Could not get update interval from config: %s", e)
# Default: 60 seconds
return 60.0
def run_scheduled_updates(self, current_time: Optional[float] = None) -> None:
"""
Trigger plugin updates based on their defined update intervals.
Includes health tracking and circuit breaker logic.
Uses PluginExecutor for safe execution with timeout.
"""
if current_time is None:
current_time = time.time()
for plugin_id, plugin_instance in list(self.plugins.items()):
if not getattr(plugin_instance, "enabled", True):
continue
if not hasattr(plugin_instance, "update"):
continue
# Check circuit breaker before attempting update
if self.health_tracker and self.health_tracker.should_skip_plugin(plugin_id):
continue
# Check if plugin can execute
if not self.state_manager.can_execute(plugin_id):
continue
interval = self._get_plugin_update_interval(plugin_id, plugin_instance)
if interval is None:
continue
last_update = self.plugin_last_update.get(plugin_id, 0.0)
if last_update == 0.0 or (current_time - last_update) >= interval:
# Update state to RUNNING
self.state_manager.set_state(plugin_id, PluginState.RUNNING)
try:
# Use PluginExecutor for safe execution
success = False
if self.resource_monitor:
# If resource monitor exists, wrap the call
def monitored_update():
self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
success = self.plugin_executor.execute_update(
type('obj', (object,), {'update': monitored_update})(),
plugin_id
)
else:
success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
if success:
self.plugin_last_update[plugin_id] = current_time
self.state_manager.record_update(plugin_id)
# Update state back to ENABLED
self.state_manager.set_state(plugin_id, PluginState.ENABLED)
# Record success
if self.health_tracker:
self.health_tracker.record_success(plugin_id)
else:
# Execution failed (timeout or error)
self.state_manager.set_state(plugin_id, PluginState.ERROR)
if self.health_tracker:
self.health_tracker.record_failure(plugin_id, Exception("Plugin execution failed"))
except Exception as exc: # pylint: disable=broad-except
self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
self.state_manager.set_state(plugin_id, PluginState.ERROR, error=exc)
# Record failure
if self.health_tracker:
self.health_tracker.record_failure(plugin_id, exc)
def update_all_plugins(self) -> None:
"""
Update all enabled plugins.
Calls update() on each enabled plugin using PluginExecutor.
"""
for plugin_id, plugin_instance in list(self.plugins.items()):
if not getattr(plugin_instance, "enabled", True):
continue
if not hasattr(plugin_instance, "update"):
continue
# Check if plugin can execute
if not self.state_manager.can_execute(plugin_id):
continue
# Update state to RUNNING
self.state_manager.set_state(plugin_id, PluginState.RUNNING)
try:
success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
if success:
self.plugin_last_update[plugin_id] = time.time()
self.state_manager.record_update(plugin_id)
# Update state back to ENABLED
self.state_manager.set_state(plugin_id, PluginState.ENABLED)
else:
# Execution failed
self.state_manager.set_state(plugin_id, PluginState.ERROR)
except Exception as exc: # pylint: disable=broad-except
self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
self.state_manager.set_state(plugin_id, PluginState.ERROR, error=exc)
def get_plugin_health_metrics(self) -> Dict[str, Any]:
"""
Get health metrics for all plugins.
Returns:
Dictionary mapping plugin_id to health metrics
"""
metrics = {}
for plugin_id in self.plugins.keys():
plugin_metrics = {}
# Get state information
state_info = self.state_manager.get_state_info(plugin_id)
plugin_metrics.update(state_info)
# Get health tracker metrics if available
if self.health_tracker:
health_info = self.health_tracker.get_plugin_health(plugin_id)
plugin_metrics['health'] = health_info
else:
plugin_metrics['health'] = {'status': 'unknown'}
metrics[plugin_id] = plugin_metrics
return metrics
def get_plugin_resource_metrics(self) -> Dict[str, Any]:
"""
Get resource usage metrics for all plugins.
Returns:
Dictionary mapping plugin_id to resource metrics
"""
metrics = {}
for plugin_id in self.plugins.keys():
plugin_metrics = {}
# Get state information
state_info = self.state_manager.get_state_info(plugin_id)
plugin_metrics.update(state_info)
# Get resource monitor metrics if available
if self.resource_monitor:
resource_info = self.resource_monitor.get_plugin_metrics(plugin_id)
plugin_metrics['resources'] = resource_info
else:
plugin_metrics['resources'] = {'status': 'unknown'}
metrics[plugin_id] = plugin_metrics
return metrics
def get_plugin_state(self, plugin_id: str) -> Dict[str, Any]:
"""
Get comprehensive state information for a plugin.
Args:
plugin_id: Plugin identifier
Returns:
Dictionary with state information
"""
return self.state_manager.get_state_info(plugin_id)