From 748349a7b9f4104681b53e3f8044b432c4fa417c Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Wed, 28 Oct 2026 04:59:34 -0400 Subject: [PATCH] style: clear whitespace, unused-import and blank-line lint Strip trailing whitespace, drop unused imports/variables/global statements, split multi-imports, fix E3xx/E1xx spacing and E712 comparisons. No behaviour change; only E501 long lines remain in flake8. Co-Authored-By: Claude Sonnet 5.5 --- hbd/__init__.py | 2 +- hbd/client/__init__.py | 2 +- hbd/client/config.py | 5 +- hbd/client/main.py | 119 +++++++------ hbd/client/plugin.py | 154 ++++++++--------- hbd/client/plugins/cpu_monitor.py | 38 ++--- hbd/client/plugins/disk_monitor.py | 62 +++---- hbd/client/plugins/filesystem_info.py | 48 +++--- hbd/client/plugins/memory_monitor.py | 37 ++-- hbd/client/plugins/nagios_runner.py | 52 +++--- hbd/client/plugins/network_monitor.py | 84 ++++----- hbd/client/plugins/os_info.py | 35 ++-- hbd/common/__init__.py | 2 +- hbd/common/proto.py | 25 ++- hbd/server/__init__.py | 2 +- hbd/server/cli.py | 14 +- hbd/server/config.py | 69 ++++---- hbd/server/configio.py | 1 + hbd/server/data.py | 8 +- hbd/server/hbdclass.py | 68 ++++---- hbd/server/http.py | 44 ++--- hbd/server/journal.py | 104 ++++++------ hbd/server/main.py | 39 +++-- hbd/server/monitor.py | 7 +- hbd/server/notify.py | 8 +- hbd/server/settings.py | 2 +- hbd/server/threshold.py | 217 ++++++++++++------------ hbd/server/udp.py | 22 +-- tests/test_configio.py | 2 - tests/test_http_config_api.py | 1 - tests/test_http_users_me.py | 6 +- tests/test_notification_channels_api.py | 1 - tests/test_oauth.py | 2 - tests/test_settings_sections.py | 1 - tests/test_threshold.py | 141 ++++++++------- 35 files changed, 705 insertions(+), 719 deletions(-) diff --git a/hbd/__init__.py b/hbd/__init__.py index c9c7b57..e0d56a9 100644 --- a/hbd/__init__.py +++ b/hbd/__init__.py @@ -4,7 +4,7 @@ This package contains both the heartbeat client (hbc) and server (hbd) component organized into separate subpackages: - hbd.client: Client component with system monitoring plugins -- hbd.server: Server/daemon component with web UI and notifications +- hbd.server: Server/daemon component with web UI and notifications - hbd.common: Shared utilities and protocol definitions Install options: diff --git a/hbd/client/__init__.py b/hbd/client/__init__.py index 17a306a..8e0d8ee 100644 --- a/hbd/client/__init__.py +++ b/hbd/client/__init__.py @@ -1,3 +1,3 @@ """HeartBeat Client (hbc) - System monitoring client.""" -from hbd import __version__ +from hbd import __version__ # noqa: F401 diff --git a/hbd/client/config.py b/hbd/client/config.py index f8c7765..90d698e 100644 --- a/hbd/client/config.py +++ b/hbd/client/config.py @@ -2,7 +2,6 @@ import logging import os -import logging logger = logging.getLogger(__name__) @@ -37,10 +36,10 @@ def load_config(path=None): """Load configuration from a YAML file and merge with client defaults. If YAML is not available or the file does not exist, defaults are returned. - + Args: path: Path to YAML config file (default: ~/.hbc.yaml) - + Returns: Dictionary with configuration """ diff --git a/hbd/client/main.py b/hbd/client/main.py index 29b9d4c..881fc18 100644 --- a/hbd/client/main.py +++ b/hbd/client/main.py @@ -16,7 +16,7 @@ import sys import time from logging.handlers import SysLogHandler from pathlib import Path -from typing import Dict, List, Optional +from typing import List, Optional # Import protocol and config from .config import load_config @@ -46,20 +46,20 @@ allow_remote_command = False class AsyncConnection: """Async UDP connection to a heartbeat server.""" - + def __init__(self, conn_id: int, addr: str, port: int, af: int, name: str): self.conn_id = conn_id self.addr = addr self.port = port self.af = af self.name = name - + self.ackcount = 0 self.lastack = 0.0 self.send_count = 0 self.lastsend = 0.0 self.rtts = [0.0] - + self.transport: Optional[asyncio.DatagramTransport] = None self.protocol: Optional[asyncio.DatagramProtocol] = None self._dead = False @@ -89,7 +89,7 @@ class AsyncConnection: except Exception as e: self.logger.error(f"Failed to open connection: {e}") return False - + def close(self): """Close the connection.""" if self.transport: @@ -104,10 +104,10 @@ class AsyncConnection: self.af = af self.addr = addr self.logger = logging.getLogger(f"hbc.conn.{addr}") - + async def sendto(self, msg: dict, msg_id: str = "HTB"): """Send a message to the server. - + Args: msg: Message dictionary msg_id: Message ID (HTB, PLG, etc.) @@ -121,36 +121,36 @@ class AsyncConnection: if not self.transport: self.logger.error("Cannot send - no transport") return - + # Add standard fields msg["name"] = shortname(self.name) msg["id"] = self.conn_id msg["time"] = time.time() - + # Encode message data = dicttos(msg_id, msg) - + # Send self.transport.sendto(data, (self.addr, self.port)) self.send_count += 1 self.lastsend = time.time() - + self.logger.debug(f"Sent {msg_id} message ({len(data)} bytes)") - + def handle_ack(self, msg: dict, now: float): """Handle ACK message from server. - + RTT is calculated as: (time ACK received) - (time HTB sent) """ self.lastack = now - + # Calculate RTT: time ACK received minus time HTB sent rtt = (now - self.lastsend) * 1000.0 # Convert to ms - + self.rtts.append(rtt) if len(self.rtts) > 10: self.rtts.pop(0) - + self.ackcount += 1 self.logger.debug(f"ACK received, RTT: {rtt:.1f}ms") if msg.get("request_update"): @@ -160,11 +160,11 @@ class AsyncConnection: class HeartbeatProtocol(asyncio.DatagramProtocol): """Protocol handler for incoming UDP messages.""" - + def __init__(self, connection: AsyncConnection): self.connection = connection self.logger = logging.getLogger("hbc.protocol") - + def datagram_received(self, data: bytes, addr): """Handle incoming datagram.""" try: @@ -172,10 +172,10 @@ class HeartbeatProtocol(asyncio.DatagramProtocol): if not msg: self.logger.warning(f"Failed to parse message from {addr}") return - + now = time.time() msg_id = msg.get("ID") - + if msg_id == "ACK": self.connection.handle_ack(msg, now) elif msg_id == "CMD": @@ -186,10 +186,10 @@ class HeartbeatProtocol(asyncio.DatagramProtocol): asyncio.create_task(handle_update(self.connection, msg)) else: self.logger.warning(f"Unknown message type: {msg_id}") - + except Exception as e: self.logger.error(f"Error processing datagram: {e}", exc_info=True) - + def error_received(self, exc): """Handle protocol errors — close transport so the heartbeat sender retries.""" self.logger.warning(f"Protocol error on {self.connection.addr}: {exc} — will retry") @@ -230,7 +230,7 @@ async def handle_command(conn: AsyncConnection, msg: dict): except Exception as e: result = str(e) status = "Error" - + # Send response response = { "service": "command", @@ -446,14 +446,14 @@ async def plugin_collector_interval( interval: int ): """Collect plugins on a specific interval. - + Args: conn: Connection to send on plugins: List of plugins to collect interval: Collection interval in seconds """ logger = logging.getLogger(f"hbc.plugins.{interval}s") - + while running: for plugin in plugins: try: @@ -471,7 +471,7 @@ async def plugin_collector_interval( f"Error collecting {plugin.name}: {e}", exc_info=True ) - + # Wait for next interval or shutdown event try: if shutdown_event: @@ -498,11 +498,11 @@ def stop(): """Stop the event loop.""" global running running = False - + # Set shutdown event to wake up sleeping tasks if shutdown_event: shutdown_event.set() - + # Cancel all active tasks for task in active_tasks: if not task.done(): @@ -555,7 +555,7 @@ async def cleanup(connections: List[AsyncConnection]): """Cleanup connections on shutdown.""" logger = logging.getLogger("hbc.cleanup") logger.info("Cleaning up connections") - + target = next((c for c in connections if c.transport), connections[0] if connections else None) # A SIGHUP restart is not a host shutdown, so don't announce one. if target and send_shutdown and not dorestart: @@ -565,35 +565,35 @@ async def cleanup(connections: List[AsyncConnection]): logger.error(f"Error sending shutdown: {e}") for conn in connections: conn.close() - + # Give messages time to send await asyncio.sleep(0.5) async def async_main(args, config): """Async main function.""" - global running, shutdown_event, active_tasks, send_shutdown, allow_remote_command + global shutdown_event, active_tasks, send_shutdown, allow_remote_command # Create shutdown event shutdown_event = asyncio.Event() active_tasks = [] allow_remote_command = bool(config.get("allow_remote_command", False)) - + logger = logging.getLogger("hbc.main") - + # Setup iam = socket.gethostname() if args.name: iam = args.name - + hb_hosts = args.hosts hb_port = config.get("hb_port", PORT) interval = config.get("interval", INTERVAL) - + logger.info(f"hbc {__version__} on {iam} -> {hb_hosts} port={hb_port}, interval={interval}s") if allow_remote_command: logger.warning("allow_remote_command is true — CMD packets from the server will be executed") - + af_filter = (socket.AF_INET if getattr(args, "ipv4_only", False) else socket.AF_INET6 if getattr(args, "ipv6_only", False) else 0) @@ -632,7 +632,7 @@ async def async_main(args, config): return 1 logger.info(f"Created {len(connections)} connections") - + # Send boot/message if requested send_shutdown = False if args.boot or args.message: @@ -643,27 +643,27 @@ async def async_main(args, config): if args.message: boot_msg["service"] = "service" boot_msg["msg"] = args.message - + boot_msg["acks"] = 0 target = next((c for c in connections if c.transport), connections[0]) await target.sendto(boot_msg) - + if args.message and not args.daemon: # Message-only mode await cleanup(connections) return 0 - + # Load plugins registry = PluginRegistry() loader = PluginLoader(registry) - + plugin_dir = Path(__file__).parent / "plugins" if plugin_dir.exists(): count = await loader.load_from_directory(plugin_dir, config) logger.info(f"Loaded {count} plugins") else: logger.warning(f"Plugin directory not found: {plugin_dir}") - + # Setup signal handlers loop = asyncio.get_event_loop() for sig in (signal.SIGTERM, signal.SIGINT): @@ -675,7 +675,7 @@ async def async_main(args, config): stop() loop.add_signal_handler(signal.SIGHUP, _sighup) - + # Start async tasks # Per-protocol internet presence probe (shared with the network_monitor plugin) net_cfg = dict((config.get("plugins") or {}).get("network_monitor") @@ -690,7 +690,7 @@ async def async_main(args, config): for conn in connections: task = asyncio.create_task(heartbeat_sender(conn, interval, connectivity)) active_tasks.append(task) - + active_tasks.append(asyncio.create_task( dns_refresher(connections, hb_hosts, hb_port, af_filter, iam, interval, connectivity, conn_id))) @@ -698,18 +698,18 @@ async def async_main(args, config): if connections and registry.get_enabled(): task = asyncio.create_task(plugin_collector(connections[0], registry)) active_tasks.append(task) - + # Wait for stop or tasks to complete try: await asyncio.gather(*active_tasks, return_exceptions=True) except asyncio.CancelledError: logger.info("Tasks cancelled") - + # Cleanup logger.info("Shutting down...") await cleanup(connections) await loader.unload_all() - + return 0 @@ -727,11 +727,11 @@ def daemonize( except OSError as e: sys.stderr.write(f"fork #1 failed: {e}\n") os._exit(1) - + os.chdir(working_dir) os.setsid() os.umask(0) - + try: pid = os.fork() if pid > 0: @@ -739,14 +739,14 @@ def daemonize( except OSError as e: sys.stderr.write(f"fork #2 failed: {e}\n") sys.exit(1) - + sys.stdout.flush() sys.stderr.flush() - + si = open(stdin, "r") so = open(stdout, "a+") se = open(stderr, "a+") - + os.dup2(si.fileno(), sys.stdin.fileno()) os.dup2(so.fileno(), sys.stdout.fileno()) os.dup2(se.fileno(), sys.stderr.fileno()) @@ -838,18 +838,17 @@ def build_parser(): def main(argv=None): """Main entry point.""" - global running, dorestart - + parser = build_parser() args = parser.parse_args(argv) - + # Setup logging log_level = logging.WARNING if args.verbose: log_level = logging.INFO if args.debug: log_level = logging.DEBUG - + logging.basicConfig( level=log_level, format="%(asctime)s %(name)s %(levelname)s: %(message)s", @@ -858,14 +857,14 @@ def main(argv=None): # Load config config = load_config(args.configfile) - + # Daemonize if requested if args.daemon: logging.info("Daemonizing...") daemonize() _reconfigure_logging_for_daemon(log_level) logging.info(f"hbc starting, sending heartbeat to {', '.join(args.hosts)}") - + # Run async main try: exit_code = asyncio.run(async_main(args, config)) @@ -875,7 +874,7 @@ def main(argv=None): except Exception as e: logging.error(f"Fatal error: {e}", exc_info=True) exit_code = 1 - + # Handle restart if dorestart: logging.info("Restarting...") @@ -883,7 +882,7 @@ def main(argv=None): # a SIGHUP restart is not a host reboot. restart_argv = [sys.argv[0]] + [a for a in sys.argv[1:] if a not in ("-b", "--boot")] os.execv(restart_argv[0], restart_argv) - + sys.exit(exit_code) diff --git a/hbd/client/plugin.py b/hbd/client/plugin.py index 1cce13a..7c5f41d 100644 --- a/hbd/client/plugin.py +++ b/hbd/client/plugin.py @@ -31,13 +31,13 @@ class Plugin(ABC): enabled: Whether plugin is active (can be disabled via config) skip_reason: Set by plugin before returning False from initialize(); causes loader to log INFO instead of WARNING. """ - + name: str = "" version: str = "1.0.0" description: str = "" interval: int = 0 enabled: bool = True - + def __init__(self, config: Optional[Dict[str, Any]] = None): """Initialize plugin with optional configuration. @@ -48,51 +48,51 @@ class Plugin(ABC): self.logger = logging.getLogger(f"plugin.{self.name}") self._initialized = False self.skip_reason: Optional[str] = None - + @abstractmethod async def initialize(self) -> bool: """Initialize plugin (load resources, check dependencies). - + Called once when plugin is loaded. Plugins should validate dependencies (e.g., check if psutil is available) and prepare any resources. - + Returns: True if initialization succeeded, False otherwise """ pass - + @abstractmethod async def collect(self) -> Dict[str, Any]: """Collect data from the system. - + This is the main method called on each collection interval. Should return a dictionary of key-value pairs representing the collected data. - + Keys should be strings (metric names). Values can be: - Scalars: int, float, str, bool - Lists/dicts (will be serialized appropriately) - + Returns: Dictionary of collected metrics, or empty dict on error """ pass - + async def cleanup(self) -> None: """Cleanup plugin resources before shutdown. - + Called when plugin is being unloaded or on system shutdown. Override to release resources, close connections, etc. """ pass - + def validate_data(self, data: Dict[str, Any]) -> bool: """Validate collected data before sending to server. - + Override to implement custom validation logic. - + Args: data: Data returned from collect() - + Returns: True if data is valid, False otherwise """ @@ -101,46 +101,46 @@ class Plugin(ABC): class InfoPlugin(Plugin): """Plugin for collecting static or rarely-changing information. - + InfoPlugins collect data that doesn't change frequently: - OS name and version - Hardware specifications (CPU model, RAM size) - Network interface MAC addresses - + Characteristics: - interval = 0 (collected once at startup by default) - Can specify interval > 0 for periodic refresh (e.g., check for hardware changes) - Data is cached and reused until next collection """ - + interval: int = 0 # Collect once at startup - + def __init__(self, config: Optional[Dict[str, Any]] = None): super().__init__(config) self._cached_data: Optional[Dict[str, Any]] = None - + async def get_cached_data(self) -> Optional[Dict[str, Any]]: """Get cached data if available (avoids re-collection). - + Returns: Cached data dict, or None if not yet collected """ return self._cached_data - + async def collect(self) -> Dict[str, Any]: """Collect and cache static information.""" if self._cached_data is None: self._cached_data = await self._collect_info() return self._cached_data - + @abstractmethod async def _collect_info(self) -> Dict[str, Any]: """Internal method to perform actual data collection. - + Override this method instead of collect() for InfoPlugins. """ pass - + def invalidate_cache(self) -> None: """Force re-collection on next collect() call.""" self._cached_data = None @@ -148,33 +148,33 @@ class InfoPlugin(Plugin): class MonitorPlugin(Plugin): """Plugin for collecting periodic monitoring data. - + MonitorPlugins collect time-series metrics that change frequently: - CPU usage percentage - Memory consumption - Disk I/O statistics - Network traffic - + Characteristics: - interval > 0 (e.g., 30 seconds for CPU, 60 for disk) - Collected continuously on schedule - Data includes timestamps for time-series tracking """ - + interval: int = 30 # Default: collect every 30 seconds - + def __init__(self, config: Optional[Dict[str, Any]] = None): super().__init__(config) self._last_reading: Optional[Dict[str, Any]] = None - + def get_last_reading(self) -> Optional[Dict[str, Any]]: """Get the last collected reading. - + Returns: Last reading dict with timestamp, or None if not yet collected """ return self._last_reading - + async def collect(self) -> Dict[str, Any]: """Collect monitoring data and store as last reading.""" data = await self._collect_metrics() @@ -184,11 +184,11 @@ class MonitorPlugin(Plugin): data['_timestamp'] = time.time() self._last_reading = data return data - + @abstractmethod async def _collect_metrics(self) -> Dict[str, Any]: """Internal method to perform actual metric collection. - + Override this method instead of collect() for MonitorPlugins. """ pass @@ -196,38 +196,38 @@ class MonitorPlugin(Plugin): class PluginRegistry: """Registry for managing loaded plugins. - + Maintains a collection of loaded plugins and provides methods to query plugins by name, type, or interval. """ - + def __init__(self): self._plugins: Dict[str, Plugin] = {} self.logger = logging.getLogger("plugin.registry") - + def register(self, plugin: Plugin) -> bool: """Register a plugin instance. - + Args: plugin: Plugin instance to register - + Returns: True if registered successfully, False if name conflict """ if plugin.name in self._plugins: self.logger.error(f"Plugin '{plugin.name}' already registered") return False - + self._plugins[plugin.name] = plugin self.logger.info(f"Registered plugin: {plugin.name} v{plugin.version}") return True - + def unregister(self, name: str) -> bool: """Unregister a plugin by name. - + Args: name: Plugin name to unregister - + Returns: True if unregistered, False if not found """ @@ -236,43 +236,43 @@ class PluginRegistry: self.logger.info(f"Unregistered plugin: {name}") return True return False - + def get(self, name: str) -> Optional[Plugin]: """Get plugin by name. - + Args: name: Plugin name - + Returns: Plugin instance or None if not found """ return self._plugins.get(name) - + def get_all(self) -> List[Plugin]: """Get all registered plugins.""" return list(self._plugins.values()) - + def get_enabled(self) -> List[Plugin]: """Get all enabled plugins.""" return [p for p in self._plugins.values() if p.enabled] - + def get_by_type(self, plugin_type: Type[Plugin]) -> List[Plugin]: """Get all plugins of a specific type. - + Args: plugin_type: Plugin class (InfoPlugin or MonitorPlugin) - + Returns: List of plugins matching the type """ return [p for p in self._plugins.values() if isinstance(p, plugin_type)] - + def get_by_interval(self, interval: int) -> List[Plugin]: """Get all plugins with a specific collection interval. - + Args: interval: Interval in seconds (0 for one-time collection) - + Returns: List of plugins with matching interval """ @@ -281,51 +281,51 @@ class PluginRegistry: class PluginLoader: """Load plugins from filesystem and instantiate them. - + Scans plugin directories for Python modules containing Plugin subclasses, loads them dynamically, and registers them with the PluginRegistry. """ - + def __init__(self, registry: PluginRegistry): self.registry = registry self.logger = logging.getLogger("plugin.loader") self._loaded_modules: Dict[str, Any] = {} - + async def load_from_directory( self, directory: Path, config: Optional[Dict[str, Any]] = None ) -> int: """Load all plugins from a directory. - + Scans for .py files, imports them, finds Plugin subclasses, instantiates them with config, initializes, and registers. - + Args: directory: Path to plugin directory config: Configuration dict (may contain per-plugin config) - + Returns: Number of plugins successfully loaded """ if not directory.exists() or not directory.is_dir(): self.logger.warning(f"Plugin directory not found: {directory}") return 0 - + loaded_count = 0 raw_config = config or {} # Per-plugin config lives under the 'plugins' key or at top-level. # CLIENT_DEFAULTS seeds "plugins": {} so the key always exists; check # both the subdict and top-level so that either layout in .hbc.yaml works. plugins_subconfig = raw_config.get("plugins", {}) - + # Scan for Python files for plugin_file in directory.glob("*.py"): if plugin_file.name.startswith("_"): continue # Skip __init__.py and private modules - + self.logger.debug(f"Processing plugin file: {plugin_file.name}") - + try: # Load module dynamically module_name = f"plugins.{plugin_file.stem}" @@ -333,17 +333,17 @@ class PluginLoader: if not spec or not spec.loader: self.logger.warning(f"Could not create spec for {plugin_file}") continue - + module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) self._loaded_modules[module_name] = module - + self.logger.debug(f"Loaded module: {module_name}") - + # Track which plugin classes we've already processed to avoid duplicates processed_classes = set() - + # Find Plugin subclasses in module for name, obj in inspect.getmembers(module, inspect.isclass): # Skip base classes and non-Plugin classes @@ -353,15 +353,15 @@ class PluginLoader: if not issubclass(obj, Plugin): self.logger.debug(f"Skipping non-Plugin class: {name}") continue - + # Skip if we've already processed this class (handles module-level aliases) if id(obj) in processed_classes: self.logger.debug(f"Skipping duplicate reference to: {obj.__name__}") continue processed_classes.add(id(obj)) - + self.logger.debug(f"Found plugin class: {name}") - + # Instantiate plugin with config — check plugins subdict first, # then top-level keys (e.g. nagios_runner: ... at root of config). plugin_instance_config = dict(plugins_subconfig.get(obj.name) or raw_config.get(obj.name) or {}) @@ -369,7 +369,7 @@ class PluginLoader: if "owner" in raw_config and "owner" not in plugin_instance_config: plugin_instance_config["owner"] = raw_config["owner"] plugin = obj(config=plugin_instance_config) - + # Initialize plugin try: initialized = await plugin.initialize() @@ -389,7 +389,7 @@ class PluginLoader: exc_info=True ) continue - + # Register with registry if self.registry.register(plugin): loaded_count += 1 @@ -397,15 +397,15 @@ class PluginLoader: f"Loaded plugin: {plugin.name} v{plugin.version} " f"(interval: {plugin.interval}s)" ) - + except Exception as e: self.logger.error( f"Error loading plugin from {plugin_file}: {e}", exc_info=True ) - + return loaded_count - + async def unload_all(self) -> None: """Unload all plugins and cleanup resources.""" for plugin in self.registry.get_all(): @@ -417,7 +417,7 @@ class PluginLoader: exc_info=True ) self.registry.unregister(plugin.name) - + # Remove loaded modules for module_name in self._loaded_modules: if module_name in sys.modules: diff --git a/hbd/client/plugins/cpu_monitor.py b/hbd/client/plugins/cpu_monitor.py index b75779e..e8f64af 100644 --- a/hbd/client/plugins/cpu_monitor.py +++ b/hbd/client/plugins/cpu_monitor.py @@ -5,8 +5,6 @@ load average, and process counts. """ from typing import Any, Dict, Optional -import sys -from pathlib import Path # Import from parent package from hbd.client.plugin import MonitorPlugin @@ -14,7 +12,7 @@ from hbd.client.plugin import MonitorPlugin class CPUMonitorPlugin(MonitorPlugin): """Monitor CPU usage and load. - + Collects: - Overall CPU usage percentage - Per-core CPU usage (if enabled in config) @@ -22,28 +20,28 @@ class CPUMonitorPlugin(MonitorPlugin): - Process count - CPU frequency (if available) """ - + name = "cpu_monitor" version = "1.0.0" description = "CPU usage and load monitoring" interval = 300 # MonitorPlugin: collect every 5 minutes by default - + def __init__(self, config: Optional[Dict[str, Any]] = None): super().__init__(config) self.psutil = None self.per_core = config.get("per_core", False) if config else False self.interval = config.get("interval", 300) if config else 300 - + async def initialize(self) -> bool: """Initialize the CPU monitor plugin. - + Checks if psutil is available. - + Returns: True if psutil is available, False otherwise """ self.logger.info(f"Initializing {self.name} plugin") - + try: import psutil self.psutil = psutil @@ -54,23 +52,23 @@ class CPUMonitorPlugin(MonitorPlugin): "psutil module not available. Install with: pip install psutil" ) return False - + async def _collect_metrics(self) -> Dict[str, Any]: """Collect CPU metrics. - + Returns: Dictionary with CPU metrics """ if not self.psutil: return {} - + try: data = {} - + # Overall CPU usage percentage (non-blocking, interval=1 for accuracy) # Note: first call to cpu_percent() returns 0.0, subsequent calls work correctly data["cpu_percent"] = self.psutil.cpu_percent(interval=1) - + # Per-core CPU usage (if enabled) if self.per_core: per_core_percents = self.psutil.cpu_percent(interval=0, percpu=True) @@ -79,7 +77,7 @@ class CPUMonitorPlugin(MonitorPlugin): else: # Just report core count data["cpu_core_count"] = self.psutil.cpu_count() - + # Load average (Unix-like systems only) try: load_avg = self.psutil.getloadavg() @@ -89,13 +87,13 @@ class CPUMonitorPlugin(MonitorPlugin): except (AttributeError, OSError): # Not available on Windows pass - + # Process count try: data["process_count"] = len(self.psutil.pids()) except Exception as e: self.logger.warning(f"Could not get process count: {e}") - + # CPU frequency (if available) try: freq = self.psutil.cpu_freq() @@ -107,7 +105,7 @@ class CPUMonitorPlugin(MonitorPlugin): # Not available on all systems, or may fail on FreeBSD with sysctl issues self.logger.debug(f"CPU frequency not available: {e}") pass - + # CPU times (user, system, idle, etc.) try: cpu_times = self.psutil.cpu_times_percent(interval=0) @@ -125,12 +123,12 @@ class CPUMonitorPlugin(MonitorPlugin): data["uptime_seconds"] = int(time.time() - self.psutil.boot_time()) except Exception as e: self.logger.debug(f"Could not get uptime: {e}") - + self.logger.debug( f"Collected CPU metrics: {data.get('cpu_percent', 'N/A')}% usage" ) return data - + except Exception as e: self.logger.error(f"Error collecting CPU metrics: {e}", exc_info=True) return {} diff --git a/hbd/client/plugins/disk_monitor.py b/hbd/client/plugins/disk_monitor.py index c3f86c3..1b7389a 100644 --- a/hbd/client/plugins/disk_monitor.py +++ b/hbd/client/plugins/disk_monitor.py @@ -5,7 +5,7 @@ Collects disk usage and I/O statistics using psutil. """ import logging -from typing import Dict, Any, Optional, List +from typing import Dict, Any, Optional try: import psutil @@ -20,27 +20,27 @@ logger = logging.getLogger(__name__) class DiskMonitorPlugin(MonitorPlugin): """ Monitor disk usage and I/O statistics. - + Collects: - Disk partition information - Disk usage per partition (total, used, free, percent) - Disk I/O counters (read/write bytes, read/write count) - Disk I/O time statistics - + Configuration: interval: Collection interval in seconds (default: 300) partitions: List of mount points to monitor (default: all) include_io: Include disk I/O statistics (default: True) exclude_types: List of filesystem types to exclude (default: tmpfs, devtmpfs, squashfs) """ - + name = "disk_monitor" interval = 300 # Collect every 5 minutes by default - + def __init__(self, config: Optional[Dict[str, Any]] = None): """ Initialize the disk monitor plugin. - + Args: config: Optional configuration dict with keys: - interval: Collection interval in seconds (default: 300) @@ -53,34 +53,34 @@ class DiskMonitorPlugin(MonitorPlugin): self.include_io = self.config.get('include_io', True) self.exclude_types = set(self.config.get('exclude_types', ['tmpfs', 'devtmpfs', 'squashfs'])) self.interval = self.config.get('interval', 300) - + if psutil is None: raise ImportError("psutil library is required for disk_monitor plugin") - + # Store previous I/O counters for delta calculation self._prev_io = {} - + async def initialize(self): """Initialize the plugin (check psutil availability).""" if psutil is None: logger.error("psutil not available - disk_monitor cannot run") return False - + logger.info(f"Disk monitor initialized (interval: {self.interval}s, io: {self.include_io})") - + # Initialize I/O counters if available if self.include_io: try: self._prev_io = psutil.disk_io_counters(perdisk=True) except Exception as e: logger.warning(f"Could not initialize disk I/O counters: {e}") - + return True - + async def collect(self) -> Dict[str, Any]: """ Collect current disk statistics. - + Returns: Dictionary with disk metrics organized by partition: - partitions: Dict of partition data, keyed by mount point @@ -103,7 +103,7 @@ class DiskMonitorPlugin(MonitorPlugin): if psutil is None: logger.error("psutil not available") return {} - + try: data = await self._collect_metrics() logger.debug(f"Collected disk metrics: {len(data.get('partitions', {}))} partitions") @@ -111,24 +111,24 @@ class DiskMonitorPlugin(MonitorPlugin): except Exception as e: logger.error(f"Error collecting disk metrics: {e}") return {"error": str(e)} - + async def _collect_metrics(self) -> Dict[str, Any]: """Collect disk metrics from psutil.""" metrics = {} - + # Collect partition usage partitions_data = {} partitions = psutil.disk_partitions(all=False) - + for partition in partitions: # Skip unwanted filesystem types if partition.fstype in self.exclude_types: continue - + # Skip if we're only monitoring specific partitions if self.partitions and partition.mountpoint not in self.partitions: continue - + try: usage = psutil.disk_usage(partition.mountpoint) partitions_data[partition.mountpoint] = { @@ -145,15 +145,15 @@ class DiskMonitorPlugin(MonitorPlugin): except Exception as e: logger.warning(f"Error reading {partition.mountpoint}: {e}") continue - + metrics['partitions'] = partitions_data - + # Collect I/O statistics if self.include_io: try: io_counters = psutil.disk_io_counters(perdisk=True) io_data = {} - + for disk_name, counters in io_counters.items(): disk_stats = { 'read_count': counters.read_count, @@ -161,7 +161,7 @@ class DiskMonitorPlugin(MonitorPlugin): 'read_bytes': counters.read_bytes, 'write_bytes': counters.write_bytes, } - + # Add time statistics if available if hasattr(counters, 'read_time'): disk_stats['read_time'] = counters.read_time @@ -169,7 +169,7 @@ class DiskMonitorPlugin(MonitorPlugin): disk_stats['write_time'] = counters.write_time if hasattr(counters, 'busy_time'): disk_stats['busy_time'] = counters.busy_time - + # Calculate deltas from previous collection if disk_name in self._prev_io: prev = self._prev_io[disk_name] @@ -177,19 +177,19 @@ class DiskMonitorPlugin(MonitorPlugin): disk_stats['write_bytes_delta'] = counters.write_bytes - prev.write_bytes disk_stats['read_count_delta'] = counters.read_count - prev.read_count disk_stats['write_count_delta'] = counters.write_count - prev.write_count - + io_data[disk_name] = disk_stats - + metrics['io_counters'] = io_data - + # Store current counters for next delta calculation self._prev_io = io_counters - + except Exception as e: logger.warning(f"Could not collect disk I/O statistics: {e}") - + return metrics - + async def cleanup(self): """Cleanup (nothing to do for this plugin).""" logger.info("Disk monitor cleanup") diff --git a/hbd/client/plugins/filesystem_info.py b/hbd/client/plugins/filesystem_info.py index e75463a..c652b25 100644 --- a/hbd/client/plugins/filesystem_info.py +++ b/hbd/client/plugins/filesystem_info.py @@ -20,29 +20,29 @@ logger = logging.getLogger(__name__) class FilesystemInfoPlugin(InfoPlugin): """ Collect filesystem and partition information. - + This is an InfoPlugin that collects static information once during startup. - + By default, only reports physical mounted filesystems (e.g., ext4, xfs, btrfs). Set include_pseudo=True to also include pseudo filesystems (proc, sysfs, tmpfs, etc.). - + Collects: - List of mounted filesystems - Partition details (device, mount point, filesystem type, options) - Filesystem capabilities and features - + Configuration: include_pseudo: Include pseudo/virtual filesystems (default: False) exclude_types: List of additional filesystem types to exclude (default: []) """ - + name = "filesystem_info" interval = 0 # InfoPlugin - collect once - + def __init__(self, config: Optional[Dict[str, Any]] = None): """ Initialize the filesystem info plugin. - + Args: config: Optional configuration dict with keys: - include_pseudo: Include pseudo/virtual filesystems (default: False) @@ -53,25 +53,25 @@ class FilesystemInfoPlugin(InfoPlugin): # By default, no exclusions since all=False filters most pseudo filesystems # Users can add specific types to exclude if needed self.exclude_types = set(self.config.get('exclude_types', [])) - + if psutil is None: raise ImportError("psutil library is required for filesystem_info plugin") - + async def initialize(self): """Initialize the plugin (check psutil availability).""" if psutil is None: logger.error("psutil not available - filesystem_info cannot run") return False - + logger.info(f"Filesystem info initialized (pseudo: {self.include_pseudo})") return True - + async def collect(self) -> Dict[str, Any]: """ Collect filesystem information. - + Returns only physical mounted filesystems by default. - + Returns: Dictionary with filesystem data: - filesystems: List of filesystem dictionaries: @@ -87,7 +87,7 @@ class FilesystemInfoPlugin(InfoPlugin): if psutil is None: logger.error("psutil not available") return {} - + try: data = await self._collect_info() logger.info(f"Collected filesystem info: {len(data.get('filesystems', []))} filesystems") @@ -95,30 +95,30 @@ class FilesystemInfoPlugin(InfoPlugin): except Exception as e: logger.error(f"Error collecting filesystem info: {e}") return {"error": str(e)} - + async def _collect_info(self) -> Dict[str, Any]: """Collect filesystem information from psutil.""" info = {} filesystems = [] filesystem_types = set() - + # Get mounted disk partitions # all=False returns only physical devices (real mounted filesystems) # all=True would include pseudo filesystems (proc, sysfs, etc.) partitions = psutil.disk_partitions(all=self.include_pseudo) - + for partition in partitions: # Additional filtering if exclude_types is specified if partition.fstype in self.exclude_types: continue - + fs_info = { 'device': partition.device, 'mountpoint': partition.mountpoint, 'fstype': partition.fstype, 'opts': partition.opts, } - + # Try to get filesystem capabilities try: # Get path configuration for this mount point @@ -141,14 +141,14 @@ class FilesystemInfoPlugin(InfoPlugin): pass except Exception as e: logger.debug(f"Could not get pathconf for {partition.mountpoint}: {e}") - + filesystems.append(fs_info) filesystem_types.add(partition.fstype) - + info['filesystems'] = filesystems info['filesystem_types'] = sorted(list(filesystem_types)) info['mount_count'] = len(filesystems) - + # Add some additional filesystem statistics try: # Get boot time (useful for determining filesystem mount times) @@ -156,9 +156,9 @@ class FilesystemInfoPlugin(InfoPlugin): info['boot_time'] = boot_time except Exception as e: logger.debug(f"Could not get boot time: {e}") - + return info - + async def cleanup(self): """Cleanup (nothing to do for this plugin).""" logger.info("Filesystem info cleanup") diff --git a/hbd/client/plugins/memory_monitor.py b/hbd/client/plugins/memory_monitor.py index 13cc9c3..b5e0be3 100644 --- a/hbd/client/plugins/memory_monitor.py +++ b/hbd/client/plugins/memory_monitor.py @@ -32,31 +32,32 @@ def _zfs_arc_bytes() -> int: pass return 0 + logger = logging.getLogger(__name__) class MemoryMonitorPlugin(MonitorPlugin): """ Monitor memory and swap usage. - + Collects: - Physical memory (RAM) usage and statistics - Virtual memory details - Swap memory usage and statistics - Memory available for applications - + Configuration: interval: Collection interval in seconds (default: 300) include_swap: Include swap statistics (default: True) """ - + name = "memory_monitor" interval = 300 # Collect every 5 minutes by default - + def __init__(self, config: Optional[Dict[str, Any]] = None): """ Initialize the memory monitor plugin. - + Args: config: Optional configuration dict with keys: - interval: Collection interval in seconds (default: 300) @@ -65,23 +66,23 @@ class MemoryMonitorPlugin(MonitorPlugin): super().__init__(config) self.include_swap = self.config.get('include_swap', True) self.interval = self.config.get('interval', 300) - + if psutil is None: raise ImportError("psutil library is required for memory_monitor plugin") - + async def initialize(self): """Initialize the plugin (check psutil availability).""" if psutil is None: logger.error("psutil not available - memory_monitor cannot run") return False - + logger.info(f"Memory monitor initialized (interval: {self.interval}s, swap: {self.include_swap})") return True - + async def collect(self) -> Dict[str, Any]: """ Collect current memory statistics. - + Returns: Dictionary with memory metrics: - memory_total: Total physical RAM in bytes @@ -104,7 +105,7 @@ class MemoryMonitorPlugin(MonitorPlugin): if psutil is None: logger.error("psutil not available") return {} - + try: data = await self._collect_metrics() logger.debug(f"Collected memory metrics: {len(data)} fields") @@ -112,11 +113,11 @@ class MemoryMonitorPlugin(MonitorPlugin): except Exception as e: logger.error(f"Error collecting memory metrics: {e}") return {"error": str(e)} - + async def _collect_metrics(self) -> Dict[str, Any]: """Collect memory metrics from psutil.""" metrics = {} - + # Virtual (physical) memory statistics vmem = psutil.virtual_memory() @@ -134,7 +135,7 @@ class MemoryMonitorPlugin(MonitorPlugin): metrics['memory_used'] = used metrics['memory_free'] = vmem.free metrics['memory_percent'] = percent - + # Platform-specific memory details if hasattr(vmem, 'active'): metrics['memory_active'] = vmem.active @@ -146,7 +147,7 @@ class MemoryMonitorPlugin(MonitorPlugin): metrics['memory_cached'] = vmem.cached if hasattr(vmem, 'shared'): metrics['memory_shared'] = vmem.shared - + # Swap memory statistics if self.include_swap: try: @@ -155,7 +156,7 @@ class MemoryMonitorPlugin(MonitorPlugin): metrics['swap_used'] = swap.used metrics['swap_free'] = swap.free metrics['swap_percent'] = swap.percent - + # Swap in/out counters (may not be available on all platforms) if hasattr(swap, 'sin'): metrics['swap_sin'] = swap.sin @@ -163,9 +164,9 @@ class MemoryMonitorPlugin(MonitorPlugin): metrics['swap_sout'] = swap.sout except Exception as e: logger.warning(f"Could not collect swap statistics: {e}") - + return metrics - + async def cleanup(self): """Cleanup (nothing to do for this plugin).""" logger.info("Memory monitor cleanup") diff --git a/hbd/client/plugins/nagios_runner.py b/hbd/client/plugins/nagios_runner.py index d91aecb..5f1ff6f 100644 --- a/hbd/client/plugins/nagios_runner.py +++ b/hbd/client/plugins/nagios_runner.py @@ -43,10 +43,10 @@ STATUS_NAMES = { class NagiosRunnerPlugin(MonitorPlugin): """Run Nagios-compatible monitoring plugins. - + This plugin executes external Nagios plugins and collects their output, including status codes, messages, and performance data. - + Configuration: interval: Collection interval in seconds (default: 300) commands: List of command definitions with 'name' and 'command' keys @@ -62,20 +62,20 @@ class NagiosRunnerPlugin(MonitorPlugin): - name: check_load command: /usr/lib/nagios/plugins/check_load -w 5,4,3 -c 10,8,6 """ - + name = "nagios_runner" version = "1.0.0" description = "Execute Nagios-compatible monitoring plugins" interval = 300 # MonitorPlugin: collect every 5 minutes by default - + def __init__(self, config: Optional[Dict[str, Any]] = None): super().__init__(config) - + # Extract configuration self.commands: List[Dict[str, str]] = config.get("commands", []) if config else [] self.timeout: int = config.get("timeout", 30) if config else 30 self.interval = config.get("interval", 300) if config else 300 - + async def initialize(self) -> bool: """Initialize the Nagios runner plugin. @@ -117,10 +117,10 @@ class NagiosRunnerPlugin(MonitorPlugin): ) return True - + async def _collect_metrics(self) -> Dict[str, Any]: """Collect metrics from all configured Nagios plugins. - + Returns: Dictionary with results from all plugins """ @@ -159,7 +159,7 @@ class NagiosRunnerPlugin(MonitorPlugin): results[f"{name}_output"] = str(e) return results - + async def _run_nagios_plugin( self, command: str @@ -210,33 +210,33 @@ class NagiosRunnerPlugin(MonitorPlugin): except Exception as e: self.logger.error(f"Error executing command: {e}") return NAGIOS_UNKNOWN, f"Execution error: {str(e)}", {} - + def _parse_perfdata(self, output: str) -> Dict[str, Any]: """Parse Nagios performance data from plugin output. - + Nagios performance data format: 'label'=value[UOM];[warn];[crit];[min];[max] - + Multiple metrics separated by spaces. - + Args: output: Plugin output string - + Returns: Dictionary of metric_name: value """ perfdata = {} - + # Performance data comes after the pipe character if '|' not in output: return perfdata - + perf_section = output.split('|', 1)[1].strip() - + # Regex to match performance data format # Matches: 'label'=value or label=value perf_regex = r"'?([^'=]+)'?=([\d.]+)([a-zA-Z%]*);?([\d.]*);?([\d.]*);?([\d.]*);?([\d.]*)" - + for match in re.finditer(perf_regex, perf_section): label = match.group(1).strip() value_str = match.group(2) @@ -245,43 +245,43 @@ class NagiosRunnerPlugin(MonitorPlugin): crit = match.group(5) min_val = match.group(6) max_val = match.group(7) - + # Convert value to float try: value = float(value_str) except ValueError: continue - + # Store the value perfdata[label] = value - + # Optionally store UOM as separate field if uom: perfdata[f"{label}_uom"] = uom - + # Store thresholds if present if warn: try: perfdata[f"{label}_warn"] = float(warn) except ValueError: pass - + if crit: try: perfdata[f"{label}_crit"] = float(crit) except ValueError: pass - + if min_val: try: perfdata[f"{label}_min"] = float(min_val) except ValueError: pass - + if max_val: try: perfdata[f"{label}_max"] = float(max_val) except ValueError: pass - + return perfdata diff --git a/hbd/client/plugins/network_monitor.py b/hbd/client/plugins/network_monitor.py index 9ac654f..804a866 100644 --- a/hbd/client/plugins/network_monitor.py +++ b/hbd/client/plugins/network_monitor.py @@ -5,7 +5,7 @@ Collects network interface statistics and connection information using psutil. """ import logging -from typing import Dict, Any, Optional, List +from typing import Dict, Any, Optional try: import psutil @@ -21,13 +21,13 @@ logger = logging.getLogger(__name__) class NetworkMonitorPlugin(MonitorPlugin): """ Monitor network interface statistics and connections. - + Collects: - Network interface I/O counters (bytes sent/received, packets, errors, drops) - Per-interface statistics - Network connection counts by state - Interface addresses and configuration - + Configuration: interval: Collection interval in seconds (default: 300) interfaces: List of interfaces to monitor (default: all) @@ -38,14 +38,14 @@ class NetworkMonitorPlugin(MonitorPlugin): probe_v6: IPv6 address to ping (default: 2606:4700:4700::1111) probe_interval: Seconds between ICMP probes while healthy (default: 60) """ - + name = "network_monitor" interval = 300 # Collect every 5 minutes by default - + def __init__(self, config: Optional[Dict[str, Any]] = None): """ Initialize the network monitor plugin. - + Args: config: Optional configuration dict with keys: - interval: Collection interval in seconds (default: 300) @@ -58,34 +58,34 @@ class NetworkMonitorPlugin(MonitorPlugin): self.include_connections = self.config.get('include_connections', True) self.include_addresses = self.config.get('include_addresses', False) self.interval = self.config.get('interval', 300) - + if psutil is None: raise ImportError("psutil library is required for network_monitor plugin") - + # Store previous I/O counters for delta calculation self._prev_io = {} - + async def initialize(self): """Initialize the plugin (check psutil availability).""" if psutil is None: logger.error("psutil not available - network_monitor cannot run") return False - + logger.info(f"Network monitor initialized (interval: {self.interval}s, " - f"connections: {self.include_connections})") - + f"connections: {self.include_connections})") + # Initialize I/O counters try: self._prev_io = psutil.net_io_counters(pernic=True) except Exception as e: logger.warning(f"Could not initialize network I/O counters: {e}") - + return True - + async def collect(self) -> Dict[str, Any]: """ Collect current network statistics. - + Returns: Dictionary with network metrics: - interfaces: Dict of interface statistics, keyed by interface name @@ -113,7 +113,7 @@ class NetworkMonitorPlugin(MonitorPlugin): if psutil is None: logger.error("psutil not available") return {} - + try: data = await self._collect_metrics() logger.debug(f"Collected network metrics: {len(data.get('interfaces', {}))} interfaces") @@ -121,21 +121,21 @@ class NetworkMonitorPlugin(MonitorPlugin): except Exception as e: logger.error(f"Error collecting network metrics: {e}") return {"error": str(e)} - + async def _collect_metrics(self) -> Dict[str, Any]: """Collect network metrics from psutil.""" metrics = {} - + # Collect per-interface I/O counters try: io_counters = psutil.net_io_counters(pernic=True) interfaces_data = {} - + for iface_name, counters in io_counters.items(): # Skip if we're only monitoring specific interfaces if self.interfaces and iface_name not in self.interfaces: continue - + iface_stats = { 'bytes_sent': counters.bytes_sent, 'bytes_recv': counters.bytes_recv, @@ -146,7 +146,7 @@ class NetworkMonitorPlugin(MonitorPlugin): 'dropin': counters.dropin, 'dropout': counters.dropout, } - + # Calculate deltas from previous collection if iface_name in self._prev_io: prev = self._prev_io[iface_name] @@ -154,46 +154,46 @@ class NetworkMonitorPlugin(MonitorPlugin): iface_stats['bytes_recv_delta'] = counters.bytes_recv - prev.bytes_recv iface_stats['packets_sent_delta'] = counters.packets_sent - prev.packets_sent iface_stats['packets_recv_delta'] = counters.packets_recv - prev.packets_recv - + interfaces_data[iface_name] = iface_stats - + metrics['interfaces'] = interfaces_data - + # Store current counters for next delta calculation self._prev_io = io_counters - + except Exception as e: logger.warning(f"Could not collect network I/O counters: {e}") - + # Collect connection statistics if self.include_connections: try: connections = psutil.net_connections(kind='inet') conn_stats = {} - + # Count connections by state for conn in connections: state = conn.status conn_stats[state] = conn_stats.get(state, 0) + 1 - + metrics['connections'] = conn_stats - + except (PermissionError, psutil.AccessDenied): logger.debug("Permission denied for net_connections (requires root/admin)") except Exception as e: logger.warning(f"Could not collect connection statistics: {e}") - + # Collect interface addresses if self.include_addresses: try: addresses = psutil.net_if_addrs() addr_data = {} - + for iface_name, addrs in addresses.items(): # Skip if we're only monitoring specific interfaces if self.interfaces and iface_name not in self.interfaces: continue - + iface_addrs = [] for addr in addrs: addr_info = { @@ -205,33 +205,33 @@ class NetworkMonitorPlugin(MonitorPlugin): if addr.broadcast: addr_info['broadcast'] = addr.broadcast iface_addrs.append(addr_info) - + addr_data[iface_name] = iface_addrs - + metrics['addresses'] = addr_data - + except Exception as e: logger.warning(f"Could not collect interface addresses: {e}") - + # Add interface stats (up/down status, speed, mtu) try: if_stats = psutil.net_if_stats() stats_data = {} - + for iface_name, stats in if_stats.items(): # Skip if we're only monitoring specific interfaces if self.interfaces and iface_name not in self.interfaces: continue - + stats_data[iface_name] = { 'isup': stats.isup, 'duplex': str(stats.duplex) if hasattr(stats, 'duplex') else None, 'speed': stats.speed, 'mtu': stats.mtu, } - + metrics['interface_stats'] = stats_data - + except Exception as e: logger.warning(f"Could not collect interface stats: {e}") @@ -239,9 +239,9 @@ class NetworkMonitorPlugin(MonitorPlugin): shared = get_shared() if shared is not None: metrics['connectivity'] = shared.as_metric() - + return metrics - + async def cleanup(self): """Cleanup (nothing to do for this plugin).""" logger.info("Network monitor cleanup") diff --git a/hbd/client/plugins/os_info.py b/hbd/client/plugins/os_info.py index ae8897b..7b0e57a 100644 --- a/hbd/client/plugins/os_info.py +++ b/hbd/client/plugins/os_info.py @@ -5,7 +5,6 @@ kernel, architecture, and distribution details. """ import platform -import sys from pathlib import Path from typing import Any, Dict, Optional @@ -15,7 +14,7 @@ from hbd.client.plugin import InfoPlugin class OSInfoPlugin(InfoPlugin): """Collect operating system information. - + This plugin gathers static OS information that rarely changes: - OS name and version - Kernel version @@ -23,27 +22,27 @@ class OSInfoPlugin(InfoPlugin): - Distribution details (for Linux) - Python version (used by hbc) """ - + name = "os_info" version = "1.0.0" description = "Operating system and platform information" interval = 0 # InfoPlugin: collect once at startup - + def __init__(self, config: Optional[Dict[str, Any]] = None): super().__init__(config) - + async def initialize(self) -> bool: """Initialize the OS info plugin. - + Returns: True (always succeeds - platform module is stdlib) """ self.logger.info(f"Initializing {self.name} plugin") return True - + async def _collect_info(self) -> Dict[str, Any]: """Collect OS information. - + Returns: Dictionary with OS details """ @@ -65,15 +64,15 @@ class OSInfoPlugin(InfoPlugin): if self.config.get("owner"): self.logger.debug(f"Adding owner from config: {self.config['owner']}") data["owner"] = self.config["owner"] - + # Add Linux-specific distribution info if platform.system() == "Linux": data.update(self._get_linux_distro()) - + # Add macOS-specific info elif platform.system() == "Darwin": data["macos_version"] = platform.mac_ver()[0] - + # Add Windows-specific info elif platform.system() == "Windows": win_ver = platform.win32_ver() @@ -81,22 +80,22 @@ class OSInfoPlugin(InfoPlugin): data["windows_version"] = win_ver[1] data["windows_sp"] = win_ver[2] data["windows_type"] = win_ver[3] - + self.logger.debug(f"Collected OS info: {data['system']} {data['release']}") return data - + except Exception as e: self.logger.error(f"Error collecting OS info: {e}", exc_info=True) return {} - + def _get_linux_distro(self) -> Dict[str, str]: """Get Linux distribution information. - + Returns: Dictionary with distribution details """ distro_info = {} - + # Try reading /etc/os-release (standard on modern Linux) os_release = Path("/etc/os-release") if os_release.exists(): @@ -121,7 +120,7 @@ class OSInfoPlugin(InfoPlugin): distro_info["distro_pretty_name"] = value except Exception as e: self.logger.warning(f"Could not read /etc/os-release: {e}") - + # Fallback: try lsb_release (older systems) elif Path("/etc/lsb-release").exists(): try: @@ -138,5 +137,5 @@ class OSInfoPlugin(InfoPlugin): distro_info["distro_name"] = value except Exception as e: self.logger.warning(f"Could not read /etc/lsb-release: {e}") - + return distro_info diff --git a/hbd/common/__init__.py b/hbd/common/__init__.py index 950da80..c3ea630 100644 --- a/hbd/common/__init__.py +++ b/hbd/common/__init__.py @@ -1,3 +1,3 @@ """Common utilities shared between hbc and hbd.""" -from hbd import __version__ +from hbd import __version__ # noqa: F401 diff --git a/hbd/common/proto.py b/hbd/common/proto.py index d847366..9db1720 100644 --- a/hbd/common/proto.py +++ b/hbd/common/proto.py @@ -8,17 +8,17 @@ Message Types: PLG: Plugin data message (client -> server) """ -from typing import Dict, Any, Union +from typing import Dict, Any import json import zlib def encode_value(v: Any) -> str: """Encode a value for protocol transmission. - + Args: v: Value to encode (int, float, str, bool, list, dict, etc.) - + Returns: String representation suitable for protocol """ @@ -35,23 +35,23 @@ def encode_value(v: Any) -> str: def decode_value(val: str) -> Any: """Decode a value from protocol format. - + Args: val: String value from protocol - + Returns: Decoded Python object """ if not val: return val - + # Check for JSON-encoded complex types if val.startswith("@"): try: return json.loads(val[1:]) except Exception: return val[1:] # Return as string without @ - + # Try numeric conversion (avoid eval to prevent SyntaxWarnings on version strings) if val[0].isdigit() or (val[0] == '-' and len(val) > 1 and val[1].isdigit()): try: @@ -63,7 +63,7 @@ def decode_value(val: str) -> Any: except ValueError: pass return val - + return val @@ -135,12 +135,12 @@ def oldmtodict(msg: bytes): def encode_plugin_data(plugin_name: str, data: Dict[str, Any]) -> bytes: """Encode plugin data into a PLG message. - + Args: plugin_name: Name of the plugin (e.g., "os_info", "cpu_monitor") data: Plugin data dictionary compress: Whether to compress the payload - + Returns: Encoded message bytes """ @@ -151,12 +151,11 @@ def encode_plugin_data(plugin_name: str, data: Dict[str, Any]) -> bytes: def decode_plugin_data(msg: bytes) -> Dict[str, Any]: """Decode a PLG message into plugin data. - + Args: msg: Raw message bytes - + Returns: Dictionary with 'ID', 'plugin', and plugin data fields """ return stodict(msg) - diff --git a/hbd/server/__init__.py b/hbd/server/__init__.py index ff3d59b..88ca02b 100644 --- a/hbd/server/__init__.py +++ b/hbd/server/__init__.py @@ -1,3 +1,3 @@ """HeartBeat Daemon (hbd) - Server/daemon component.""" -from hbd import __version__ +from hbd import __version__ # noqa: F401 diff --git a/hbd/server/cli.py b/hbd/server/cli.py index 96f687a..8505aa4 100644 --- a/hbd/server/cli.py +++ b/hbd/server/cli.py @@ -120,7 +120,7 @@ def cmd_passwd(args): def cmd_notify(args): """Send a test message via a single notification channel.""" from .config import load_config - from .notify import Notification, _dispatch_to_channel, setup + from .notify import Notification, setup config = load_config(args.configfile) setup(config) @@ -193,7 +193,9 @@ def _read_pid(configfile) -> int | None: def cmd_stop(args): - import os, signal as _signal, time + import os + import signal as _signal + import time pid = _read_pid(args.configfile) if pid is None: sys.exit(1) @@ -212,7 +214,8 @@ def cmd_stop(args): def cmd_reload(args): - import os, signal as _signal + import os + import signal as _signal pid = _read_pid(args.configfile) if pid is None: sys.exit(1) @@ -222,7 +225,10 @@ def cmd_reload(args): def cmd_restart(args): - import os, signal as _signal, time, subprocess + import os + import signal as _signal + import time + import subprocess pid = _read_pid(args.configfile) if pid is not None: print(f"Stopping hbd (pid {pid})...") diff --git a/hbd/server/config.py b/hbd/server/config.py index 22ee227..81d6e47 100644 --- a/hbd/server/config.py +++ b/hbd/server/config.py @@ -43,30 +43,30 @@ SERVER_DEFAULTS = { # Host management "hosts": {}, # Unified host definitions - "dyndomains": ["example.org"], # Domains to update via nsupdate when a host with dyndns: true is updated + "dyndomains": ["example.org"], # Domains to update via nsupdate when a host with dyndns: true is updated # DNS updates - "nsupdate_bin": "/usr/bin/nsupdate", # Path to nsupdate binary - + "nsupdate_bin": "/usr/bin/nsupdate", # Path to nsupdate binary + # WebSocket settings - "ws_port": 50005, + "ws_port": 50005, "wss_port": None, "cert_path": "/usr/local/etc/ssl/", "wss_pem": "fullchain.pem", "wss_key": "privkey.pem", - + # Message journal configuration "journal_enabled": True, "journal_dir": "/var/log/heartbeat", "journal_file": "messages.journal", "journal_max_size": 100 * 1024 * 1024, # 100MB "journal_max_backups": 10, - + # Runtime flags "foreground": False, "verbose": False, "debug": 0, - + # Plugin/threshold configs (for clients reporting to this server) "plugins": {}, "thresholds": {}, @@ -76,11 +76,11 @@ THRESHOLD_DEFAULTS = { 'thresholds': { 'cpu_monitor': { 'cpu_percent': { - 'warning': 80.0, + 'warning': 80.0, 'critical': 90.0 } }, - 'memory_monitor': { + 'memory_monitor': { 'memory_percent': { 'warning': 85.0, 'critical': 95.0 @@ -90,7 +90,7 @@ THRESHOLD_DEFAULTS = { 'critical': 75.0 } }, - 'disk_monitor': { + 'disk_monitor': { 'partitions': { '/': { 'percent': { @@ -100,18 +100,18 @@ THRESHOLD_DEFAULTS = { } } }, - 'rtt': { + 'rtt': { 'warning': 200, 'critical': 250.0, 'count': 3 # Optional: number of consecutive breaches before alerting }, - 'nagios_runner': { + 'nagios_runner': { 'status_code': { 'display': '{check_name} {output}', 'operator': "nagios" } }, - 'zfs_monitor': { + 'zfs_monitor': { 'pools': { '*': { 'status': { @@ -137,10 +137,10 @@ def load_config(path=None): """Load configuration from a YAML file and merge with server defaults. If YAML is not available or the file does not exist, defaults are returned. - + Args: path: Path to YAML config file (default: ~/.hb.yaml) - + Returns: Dictionary with configuration """ @@ -165,16 +165,16 @@ def load_config(path=None): class ReloadableConfig: """Thread-safe/async-safe configuration wrapper that supports runtime reloading. - + This class wraps the configuration dictionary and provides: - Thread-safe config reloading via SIGHUP - Backward-compatible dict-like access - Async lock to prevent concurrent reloads """ - + def __init__(self, initial_config, config_path=None): """Initialize with initial configuration. - + Args: initial_config: Initial configuration dictionary config_path: Path to config file for reloading (optional) @@ -183,65 +183,62 @@ class ReloadableConfig: self._config_path = config_path self._lock = asyncio.Lock() self._logger = logging.getLogger(__name__) - + async def reload(self, config_path=None): """Reload configuration from file. - + Args: config_path: Path to config file (uses stored path if not provided) - + Returns: New configuration dictionary - + Raises: Exception if reload fails (keeps existing config) """ path = config_path or self._config_path if not path: raise ValueError("No config path specified for reload") - + async with self._lock: try: # Load new config new_config = load_config(path) - - # Store old config for rollback if needed - old_config = self._config - + # Update config self._config = new_config self._logger.info(f"Configuration reloaded from {path}") - + return new_config except Exception as e: self._logger.error(f"Failed to reload config from {path}: {e}", exc_info=True) # Keep existing config on error raise - + def get(self, key, default=None): """Get a config value (dict-compatible).""" return self._config.get(key, default) - + def __getitem__(self, key): """Get a config value via subscript (dict-compatible).""" return self._config[key] - + def __contains__(self, key): """Check if key exists (dict-compatible).""" return key in self._config - + def keys(self): """Return config keys (dict-compatible).""" return self._config.keys() - + def items(self): """Return config items (dict-compatible).""" return self._config.items() - + def values(self): """Return config values (dict-compatible).""" return self._config.values() - + @property def config(self): """Get the underlying config dict (for components that need full dict).""" @@ -319,7 +316,7 @@ def get_host_access(config, hostname) -> dict: """ host_cfg = get_host_config(config, hostname) - owner = host_cfg.get("owner") # or get_default_owner(config) + owner = host_cfg.get("owner") # or get_default_owner(config) managers = host_cfg.get("managers", []) if isinstance(managers, str): diff --git a/hbd/server/configio.py b/hbd/server/configio.py index 1793b4f..eb57b08 100644 --- a/hbd/server/configio.py +++ b/hbd/server/configio.py @@ -15,6 +15,7 @@ def _make_yaml() -> YAML: y.preserve_quotes = True return y + # Top-level keys managed by the 'server' logical section _SERVER_KEYS = [ "hbd_port", "hbd_host", "ws_port", "wss_port", "hb_port", diff --git a/hbd/server/data.py b/hbd/server/data.py index 99a785a..c3de820 100644 --- a/hbd/server/data.py +++ b/hbd/server/data.py @@ -1,12 +1,14 @@ msgs = [] # in-memory list of recent messages for new websocket clients; also logged to file via notify.eventlog + + class Data: def __init__(self, config): self.config = config self.data = {} - + def update(self, new_data): self.data.update(new_data) - + def get(self, key, default=None): - return self.data.get(key, default) \ No newline at end of file + return self.data.get(key, default) diff --git a/hbd/server/hbdclass.py b/hbd/server/hbdclass.py index 0211c4f..b3e4ade 100644 --- a/hbd/server/hbdclass.py +++ b/hbd/server/hbdclass.py @@ -43,7 +43,7 @@ class Connection: self.statetime = self.lastbeat self.deltastatetime = "computed" self.state = Connection.UNKNOWN - + # Timer-based reachability monitoring self.overdue_timer = None self.overdue_callback = None @@ -196,39 +196,39 @@ class Connection: if self.host.isDynDns(): Host.dnsQ.put((self.host.name, self.addr)) return r - + def reset_overdue_timer(self, timeout_seconds, callback): """Reset the overdue timer for this connection. - + Cancels any existing timer and sets a new one that will mark the connection as overdue if no heartbeat arrives before timeout. - + Args: timeout_seconds: Seconds before marking as overdue callback: Async function to call when timer expires """ import asyncio - + # Cancel existing timer if any if self.overdue_timer and not self.overdue_timer.cancelled(): self.overdue_timer.cancel() - + # Store parameters for later reference self.timeout_duration = timeout_seconds self.overdue_callback = callback - + # Create new timer async def timer_expired(): await callback(self) - + try: loop = asyncio.get_event_loop() - self.overdue_timer = loop.call_later(timeout_seconds, - lambda: asyncio.create_task(timer_expired())) + self.overdue_timer = loop.call_later(timeout_seconds, + lambda: asyncio.create_task(timer_expired())) except RuntimeError: # No event loop running yet pass - + def cancel_overdue_timer(self): """Cancel the overdue timer if it exists and clear all timer references.""" if self.overdue_timer: @@ -241,37 +241,37 @@ class Connection: self.overdue_timer = None self.overdue_callback = None self.timeout_duration = None - + def get_avg_rtt(self): """Get average RTT from recent samples.""" valid_rtts = [r for r in self.rtts if r > 0] if valid_rtts: return sum(valid_rtts) / len(valid_rtts) return 0 - + def get_current_rtt(self): """Get most recent RTT value.""" return self.rtts[-1] if self.rtts else 0 - + def check_rtt_threshold(self, warning_threshold=None, critical_threshold=None): """Check if RTT exceeds thresholds. - + Args: warning_threshold: RTT in ms for warning level critical_threshold: RTT in ms for critical level - + Returns: Tuple of (level, rtt_value) where level is None, 'WARNING', or 'CRITICAL' """ rtt = self.get_current_rtt() if rtt <= 0: return (None, rtt) - + if critical_threshold and rtt > critical_threshold: return ('CRITICAL', rtt) elif warning_threshold and rtt > warning_threshold: return ('WARNING', rtt) - + return (None, rtt) @@ -328,7 +328,7 @@ class Host: d["dyn"] = str(self.dyn) d["num"] = self.num d["status"] = self.status() - + # Add alert counts (split by acknowledged status) warning_unacked = 0 warning_acked = 0 @@ -347,12 +347,12 @@ class Host: critical_acked += 1 else: critical_unacked += 1 - + d["alert_warning_unacked"] = warning_unacked d["alert_warning_acked"] = warning_acked d["alert_critical_unacked"] = critical_unacked d["alert_critical_acked"] = critical_acked - + for c in ["IPv4", "IPv6"]: if c in self.connections: cs = self.connections[c].statedict() @@ -409,7 +409,7 @@ class Host: ddict[d] = cl else: ddict[d] = self.__dict__[d] - + # Add alert counts (computed from alert_states) warning_unacked = 0 warning_acked = 0 @@ -428,7 +428,7 @@ class Host: critical_acked += 1 else: critical_unacked += 1 - + ddict["alert_warning_unacked"] = warning_unacked ddict["alert_warning_acked"] = warning_acked ddict["alert_critical_unacked"] = critical_unacked @@ -554,7 +554,7 @@ class Host: def add_plugin_data(self, plugin_name, data, timestamp=None): """Store plugin data with timestamp. - + Args: plugin_name: Name of the plugin (e.g., "cpu_monitor") data: Dict of plugin data @@ -562,24 +562,24 @@ class Host: """ if timestamp is None: timestamp = time.time() - + if plugin_name not in self.plugin_data: self.plugin_data[plugin_name] = [] - + # Add new data self.plugin_data[plugin_name].append((timestamp, data)) - + # Enforce retention limit (keep last N samples) if len(self.plugin_data[plugin_name]) > self.plugin_retention: self.plugin_data[plugin_name] = self.plugin_data[plugin_name][-self.plugin_retention:] - + def get_plugin_data(self, plugin_name, limit=None): """Retrieve plugin data for a specific plugin. - + Args: plugin_name: Name of the plugin limit: Optional limit on number of recent samples to return - + Returns: List of (timestamp, data) tuples, most recent last """ @@ -587,19 +587,19 @@ class Host: if limit and len(data) > limit: return data[-limit:] return data - + def get_latest_plugin_data(self, plugin_name): """Get the most recent plugin data for a plugin. - + Args: plugin_name: Name of the plugin - + Returns: (timestamp, data) tuple or None if no data """ data = self.plugin_data.get(plugin_name, []) return data[-1] if data else None - + def get_all_plugin_data(self): """Get all plugin data for this host. diff --git a/hbd/server/http.py b/hbd/server/http.py index 2fd217a..70b18b6 100644 --- a/hbd/server/http.py +++ b/hbd/server/http.py @@ -109,6 +109,7 @@ def _insert_threshold_metric(thresholds: dict, metric_path: str, values: dict) - else: thresholds[plugin].setdefault(intermediate, {})[leaf] = cfg + def _render_template(html_str: str, **context) -> str: tmpl = jinja2.Template(html_str) return tmpl.render(**context) @@ -563,7 +564,7 @@ async def start( if err: return err hostname = request.match_info.get("hostname") - + if hostname not in hbdclass.Host.hosts: return web.json_response({"error": f"Host '{hostname}' not found"}, status=404) @@ -582,7 +583,7 @@ async def start( "data": data, "sample_count": len(samples), } - + return web.json_response({ "hostname": hostname, "plugins": plugins_summary, @@ -602,23 +603,23 @@ async def start( host = hbdclass.Host.hosts[hostname] if not _can_view_host(user, host): return web.json_response({"error": "Forbidden"}, status=403) - + # Get limit from query parameter limit = request.rel_url.query.get("limit", "10") try: limit = int(limit) except ValueError: limit = 10 - + # Get plugin data samples = host.get_plugin_data(plugin_name, limit=limit) - + if not samples: return web.json_response( {"error": f"No data for plugin '{plugin_name}' on host '{hostname}'"}, status=404 ) - + # Format samples formatted_samples = [ { @@ -627,7 +628,7 @@ async def start( } for ts, data in samples ] - + return web.json_response({ "hostname": hostname, "plugin": plugin_name, @@ -648,17 +649,17 @@ async def start( host = hbdclass.Host.hosts[hostname] if not _can_view_host(user, host): return web.json_response({"error": "Forbidden"}, status=403) - + # Get alert states alerts = [] for metric_path, alert_state in host.alert_states.items(): alerts.append(alert_state.to_dict()) - + # Get summary if threshold_checker available summary = {"ok": 0, "warning": 0, "critical": 0, "unknown": 0} if threshold_checker: summary = threshold_checker.get_alert_summary(host.alert_states) - + return web.json_response({ "hostname": hostname, "alerts": alerts, @@ -684,31 +685,31 @@ async def start( state for state in host.alert_states.values() if state.level != AlertLevel.OK ] - + for alert in active_alerts: alert_dict = alert.to_dict() alert_dict["hostname"] = hostname all_alerts.append(alert_dict) - + # Sort by level (critical first) then by hostname level_order = {"CRITICAL": 0, "WARNING": 1, "UNKNOWN": 2, "OK": 3} all_alerts.sort( key=lambda a: (level_order.get(a["level"], 99), a["hostname"], a["metric_path"]) ) - + # Get summary counts summary = {"critical": 0, "warning": 0, "unknown": 0, "total": len(all_alerts)} for alert in all_alerts: level = alert["level"].lower() if level in summary: summary[level] += 1 - + return web.json_response({ "alerts": all_alerts, "summary": summary, "host_count": len(hbdclass.Host.hosts), }) - + async def api_acknowledge_alert(request): """Acknowledge an alert to stop reminder notifications.""" user, err = _require_auth(request) @@ -721,22 +722,22 @@ async def start( {"error": "Invalid JSON in request body"}, status=400 ) - + hostname = data.get("hostname") metric_path = data.get("metric_path") - + if not hostname or not metric_path: return web.json_response( {"error": "Missing required fields: hostname and metric_path"}, status=400 ) - + if hostname not in hbdclass.Host.hosts: return web.json_response( {"error": f"Host '{hostname}' not found"}, status=404 ) - + host = hbdclass.Host.hosts[hostname] if not _can_view_host(user, host): return web.json_response({"error": "Forbidden"}, status=403) @@ -746,10 +747,10 @@ async def start( {"error": f"Alert '{metric_path}' not found for host '{hostname}'"}, status=404 ) - + alert_state = host.alert_states[metric_path] alert_state.acknowledge() - + return web.json_response({ "success": True, "hostname": hostname, @@ -1132,7 +1133,6 @@ async def start( elif is_mon: monitored.append(hostname) - # Resolve notification channel configs for display notif_channels = [] if current_user: diff --git a/hbd/server/journal.py b/hbd/server/journal.py index 5b25030..7bc68d7 100644 --- a/hbd/server/journal.py +++ b/hbd/server/journal.py @@ -19,14 +19,14 @@ logger = logging.getLogger(__name__) class MessageJournal: """ Journal logger for heartbeat messages with size-based rotation. - + Features: - Logs all received messages in JSON format - Automatic rotation when file size exceeds threshold - Keeps configurable number of rotated logs - Thread-safe and async-safe operation - Configurable log directory and file naming - + Configuration: journal_dir: Directory for journal files (default: /var/log/heartbeat/) journal_file: Base filename (default: messages.journal) @@ -34,69 +34,69 @@ class MessageJournal: max_backups: Number of backup files to keep (default: 10) enabled: Enable/disable journaling (default: True) """ - + def __init__(self, config: Optional[Dict[str, Any]] = None): """ Initialize the message journal. - + Args: config: Configuration dictionary with journal settings """ self.config = config or {} - + # Configuration options self.journal_dir = Path(self.config.get('journal_dir', '/var/log/heartbeat')) self.journal_file = self.config.get('journal_file', 'messages.journal') self.max_size = self.config.get('journal_max_size', 100 * 1024 * 1024) # 100MB default self.max_backups = self.config.get('journal_max_backups', 10) self.enabled = self.config.get('journal_enabled', True) - + # Runtime state self._file_handle = None self._current_size = 0 self._lock = asyncio.Lock() self._initialized = False - + # Full path to current journal file self.journal_path = self.journal_dir / self.journal_file - + async def initialize(self) -> bool: """ Initialize the journal. - + Creates journal directory if needed and opens the journal file. - + Returns: True if initialization successful, False otherwise """ if not self.enabled: logger.info("Message journal disabled in configuration") return True - + try: # Create journal directory if it doesn't exist self.journal_dir.mkdir(parents=True, exist_ok=True) - + # Open journal file in append mode self._file_handle = open(self.journal_path, 'a', encoding='utf-8') - + # Get current file size try: self._current_size = os.path.getsize(self.journal_path) except OSError: self._current_size = 0 - + self._initialized = True logger.info(f"Message journal initialized: {self.journal_path} " - f"(current size: {self._current_size:,} bytes, " - f"max: {self.max_size:,} bytes)") + f"(current size: {self._current_size:,} bytes, " + f"max: {self.max_size:,} bytes)") return True - + except Exception as e: logger.error(f"Failed to initialize message journal: {e}") self.enabled = False return False - + async def log_message( self, msg: Dict[str, Any], @@ -105,7 +105,7 @@ class MessageJournal: ): """ Log a received message to the journal. - + Args: msg: Parsed message dictionary addr: Source address (ip, port) tuple @@ -113,19 +113,19 @@ class MessageJournal: """ if not self.enabled or not self._initialized: return - + # Skip HTB (heartbeat) messages - too verbose msg_id = msg.get('ID', '') if msg_id == 'HTB': return - + async with self._lock: try: # Prepare journal entry if timestamp is None: import time timestamp = time.time() - + entry = { 'timestamp': timestamp, 'datetime': datetime.fromtimestamp(timestamp).isoformat(), @@ -133,21 +133,21 @@ class MessageJournal: 'source_port': addr[1] if isinstance(addr, (tuple, list)) and len(addr) > 1 else None, 'message': msg } - + # Serialize to JSON (one line per entry) json_line = json.dumps(entry, separators=(',', ':')) + '\n' json_bytes = json_line.encode('utf-8') - + # Check if rotation is needed if self._current_size + len(json_bytes) > self.max_size: await self._rotate() - + # Write to journal if self._file_handle: self._file_handle.write(json_line) self._file_handle.flush() # Ensure data is written self._current_size += len(json_bytes) - + logger.debug(f"Logged message from {addr[0]}: {msg.get('ID', 'UNKNOWN')}") except Exception as e: @@ -181,7 +181,7 @@ class MessageJournal: async def _rotate(self): """ Rotate the journal file. - + Renames current file with timestamp, opens new file, and removes old backups exceeding max_backups limit. """ @@ -190,25 +190,25 @@ class MessageJournal: if self._file_handle: self._file_handle.close() self._file_handle = None - + # Generate backup filename with timestamp timestamp_str = datetime.now().strftime('%Y%m%d-%H%M%S') backup_name = f"{self.journal_file}.{timestamp_str}" backup_path = self.journal_dir / backup_name - + # Rename current file to backup if self.journal_path.exists(): self.journal_path.rename(backup_path) logger.info(f"Rotated journal: {backup_path} " - f"(size: {self._current_size:,} bytes)") - + f"(size: {self._current_size:,} bytes)") + # Open new journal file self._file_handle = open(self.journal_path, 'a', encoding='utf-8') self._current_size = 0 - + # Clean up old backups await self._cleanup_old_backups() - + except Exception as e: logger.error(f"Error rotating journal: {e}") # Try to reopen the file even if rotation failed @@ -217,18 +217,18 @@ class MessageJournal: except Exception as e2: logger.error(f"Failed to reopen journal after rotation error: {e2}") self.enabled = False - + async def _cleanup_old_backups(self): """ Remove old backup files exceeding max_backups limit. - + Keeps only the most recent backups based on filename (which includes timestamp). """ try: # Find all backup files backup_pattern = f"{self.journal_file}.*" backup_files = sorted(self.journal_dir.glob(backup_pattern)) - + # Remove oldest backups if we have too many if len(backup_files) > self.max_backups: files_to_remove = backup_files[:len(backup_files) - self.max_backups] @@ -238,10 +238,10 @@ class MessageJournal: logger.info(f"Removed old backup: {backup_file.name}") except Exception as e: logger.warning(f"Failed to remove old backup {backup_file}: {e}") - + except Exception as e: logger.error(f"Error cleaning up old backups: {e}") - + async def log_threshold_event( self, host_name: str, @@ -253,7 +253,7 @@ class MessageJournal: ): """ Log a threshold state change event. - + Args: host_name: Name of the host metric_path: Full metric path (e.g., "cpu_monitor.cpu_percent") @@ -264,11 +264,11 @@ class MessageJournal: """ if not self.enabled or not self._initialized: return - + try: if timestamp is None: timestamp = __import__('time').time() - + event = { 'timestamp': timestamp, 'iso_time': datetime.fromtimestamp(timestamp).isoformat(), @@ -279,30 +279,30 @@ class MessageJournal: 'new_level': new_level, 'value': value, } - + async with self._lock: if not self._file_handle: return - + # Check if rotation is needed if self._current_size >= self.max_size: await self._rotate() - + # Write event line = json.dumps(event) + '\n' self._file_handle.write(line) self._file_handle.flush() - + # Update size self._current_size += len(line.encode('utf-8')) - + except Exception as e: logger.error(f"Error logging threshold event: {e}") - + async def close(self): """ Close the journal and release resources. - + Should be called during shutdown. """ async with self._lock: @@ -315,11 +315,11 @@ class MessageJournal: finally: self._file_handle = None self._initialized = False - + def get_stats(self) -> Dict[str, Any]: """ Get journal statistics. - + Returns: Dictionary with journal stats """ @@ -423,10 +423,10 @@ _journal_instance: Optional[MessageJournal] = None def get_journal(config: Optional[Dict[str, Any]] = None) -> MessageJournal: """ Get or create the global journal instance. - + Args: config: Configuration dictionary (only used on first call) - + Returns: MessageJournal instance """ diff --git a/hbd/server/main.py b/hbd/server/main.py index 2a2a10e..c5251c7 100644 --- a/hbd/server/main.py +++ b/hbd/server/main.py @@ -5,8 +5,6 @@ import logging import socket import time import signal -import sys -import ssl from . import __version__ from . import udp @@ -23,6 +21,7 @@ eventlog = notify_mod.eventlog # shared runtime collections and helpers + def save_state(config, hbdclass): """Save current state to pickle file. Safe to call at any time.""" import pickle @@ -68,21 +67,21 @@ def cleanup_function(config, hbdclass): async def reload_configuration(config_obj, config_path, components): """Reload configuration and update all components. - + Args: config_obj: ReloadableConfig instance config_path: Path to config file components: Dict with threshold_checker and other components - + Returns: True if reload succeeded, False otherwise """ try: logger.info("Starting configuration reload...") - + # Reload config file new_config = await config_obj.reload(config_path) - + # Update notify module notify_mod.reload_config(new_config) @@ -103,23 +102,23 @@ async def reload_configuration(config_obj, config_path, components): if 'threshold_checker' in components: components['threshold_checker'].reload(new_config) components['threshold_checker'].purge_stale_alerts(hbdclass) - + # Note: Changes to the following require restart: # - hb_port, hbd_port, ws_port (already bound) # - SSL certificates (already loaded) # - pickfile (already opened) # - journal settings (journal already initialized) - + # These are reloadable and effective immediately: # - notification_channels # - threshold_configs # - hosts (watchhosts, dyndns, notification_channels) # - grace period (used on next heartbeat) # - debug/verbose flags (used on next message) - + logger.info("Configuration reload completed successfully") return True - + except Exception as e: logger.error("=" * 60) logger.error(f"Failed to reload configuration: {e}", exc_info=True) @@ -142,7 +141,7 @@ async def _run_async(config, config_path=None): sig_name = signal.Signals(signum).name if hasattr(signal, "Signals") else signum logger.info(f"Received {sig_name}, initiating shutdown...") loop.call_soon_threadsafe(shutdown_event.set) - + def reload_handler(signum, frame): sig_name = signal.Signals(signum).name if hasattr(signal, "Signals") else signum logger.info(f"Received {sig_name}, initiating config reload...") @@ -181,7 +180,7 @@ async def _run_async(config, config_path=None): journal=msg_journal, ) logger.info("Threshold checker initialized") - + # Components dict for reload orchestration components = { 'threshold_checker': threshold_checker, @@ -324,7 +323,7 @@ async def _run_async(config, config_path=None): ], return_when=asyncio.FIRST_COMPLETED ) - + # Check which event was triggered if shutdown_event.is_set(): logger.info("Shutdown signal received, stopping services...") @@ -332,24 +331,24 @@ async def _run_async(config, config_path=None): for task in pending: task.cancel() break - + if reload_event.is_set(): # Clear the event for next reload reload_event.clear() - + # Cancel pending wait tasks for task in pending: task.cancel() - + # Perform reload if config_path is available if config_path: await reload_configuration(config, config_path, components) else: logger.warning("Cannot reload: no config path available") - + # Continue main loop continue - + except Exception as e: logger.exception("Error in main loop: %s", e) finally: @@ -381,7 +380,7 @@ async def _run_async(config, config_path=None): logger.warning("Timeout waiting for tasks to cancel") except Exception as e: logger.debug("Exception during task cancellation: %s", e) - + # Close message journal try: await msg_journal.close() @@ -487,7 +486,7 @@ def run(config, config_path=None): """Start the hbd service (blocking). Manually manages the event loop to ensure clean shutdown. - + Args: config: Configuration dictionary config_path: Path to config file (for reload support) diff --git a/hbd/server/monitor.py b/hbd/server/monitor.py index 6986cb4..dc8e63a 100644 --- a/hbd/server/monitor.py +++ b/hbd/server/monitor.py @@ -1,15 +1,13 @@ """Monitor helper for heartbeat daemon. This module provides monitoring tasks for the heartbeat daemon. -The primary reachability monitoring is now event-driven (timers set/reset +The primary reachability monitoring is now event-driven (timers set/reset on HTB arrival in udp.py) rather than periodic polling. This module can be extended for additional monitoring tasks. """ from __future__ import annotations -import asyncio -import time from . import notify as notify_mod DROPOVERDUE = 7 * 24 * 3600 @@ -18,11 +16,10 @@ eventlog = notify_mod.eventlog async def cleanup_connections(hbdclass): """Clean up connection timers on shutdown. - + Cancels all active overdue timers to prevent callbacks after shutdown. """ for hostname, host in list(hbdclass.Host.hosts.items()): for conn_type, conn in host.connections.items(): if hasattr(conn, 'cancel_overdue_timer'): conn.cancel_overdue_timer() - diff --git a/hbd/server/notify.py b/hbd/server/notify.py index b643fc6..0dfcc22 100644 --- a/hbd/server/notify.py +++ b/hbd/server/notify.py @@ -21,7 +21,7 @@ import smtplib import subprocess import time import sys -from dataclasses import dataclass, field +from dataclasses import dataclass from typing import Optional from . import data @@ -49,6 +49,7 @@ logf = None _LEVEL_ORDER = {"RECOVER": 0, "INFO": 0, "WARNING": 1, "CRITICAL": 2} + def _level_value(level: str) -> int: return _LEVEL_ORDER.get(level.upper(), 0) @@ -103,7 +104,6 @@ def initlog(logfile): def closelog(): - global logf if logf and logf != sys.stderr: try: logf.close() @@ -331,8 +331,6 @@ async def _send_sms_voipms_async(channel_cfg: dict, notif: Notification) -> bool return False - - async def _send_matrix_async(channel_cfg: dict, notif: Notification) -> bool: """Send a Matrix message using matrix-nio.""" try: @@ -341,7 +339,6 @@ async def _send_matrix_async(channel_cfg: dict, notif: Notification) -> bool: logger.error("matrix-nio not installed; pip install matrix-nio") return False - from nio import AsyncClient homeserver = channel_cfg.get("homeserver", "") access_token = channel_cfg.get("access_token", "") room_id = channel_cfg.get("room_id", "") @@ -359,7 +356,6 @@ async def _send_matrix_async(channel_cfg: dict, notif: Notification) -> bool: client = AsyncClient(homeserver) client.access_token = access_token try: - from nio import RoomSendResponse content = { "msgtype": "m.text", "body": text, diff --git a/hbd/server/settings.py b/hbd/server/settings.py index b969201..fe047f0 100644 --- a/hbd/server/settings.py +++ b/hbd/server/settings.py @@ -420,7 +420,7 @@ def get_settings_sections(config: dict, threshold_checker=None, user=None) -> li "fields": [ field("journal_enabled", "Enabled", "boolean", "Turn journalling on or off.", editable=True), - field("journal_dir", "Journal directory","path", + field("journal_dir", "Journal directory", "path", "Directory where journal files are written.", editable=True), field("journal_file", "Journal filename", "text", "Base filename for the journal (rotated copies get a numeric suffix)."), diff --git a/hbd/server/threshold.py b/hbd/server/threshold.py index 77a4454..6eb3f42 100644 --- a/hbd/server/threshold.py +++ b/hbd/server/threshold.py @@ -13,13 +13,14 @@ import asyncio import logging import time from enum import Enum -from typing import Dict, List, Any, Optional, Tuple, Callable +from typing import Dict, List, Any, Optional, Tuple from . import notify as notify_mod from .config import THRESHOLD_DEFAULTS logger = logging.getLogger(__name__) eventlog = notify_mod.eventlog + class AlertLevel(Enum): """Alert severity levels.""" OK = 0 @@ -64,30 +65,30 @@ class AlertState: self.acknowledged_at = None # Timestamp when acknowledged self.consecutive_count = 0 # Consecutive exceedances while still OK (for count gating) self.pending_since: Optional[float] = None # non-None while waiting out grace period before notifying - + def update( - self, - level: AlertLevel, + self, + level: AlertLevel, value: Any, threshold_value: Optional[float] = None, operator: Optional[str] = None ) -> bool: """ Update alert state. - + Args: level: New alert level value: Current metric value threshold_value: The threshold value that was exceeded (if applicable) operator: The comparison operator (>, <, >=, etc.) - + Returns: True if state changed (notification needed), False otherwise """ now = time.time() self.last_check = now self.last_value = value - + # Update threshold info when alert is active if level != AlertLevel.OK: self.threshold_value = threshold_value @@ -96,7 +97,7 @@ class AlertState: # Clear threshold info when returning to OK self.threshold_value = None self.operator = None - + # Check if state changed if level != self.level: logger.info( @@ -116,13 +117,13 @@ class AlertState: self.acknowledged = False self.acknowledged_at = None return True - + return False - + def to_dict(self) -> dict: """Convert alert state to dictionary for serialization.""" import math - + # Helper to sanitize numeric values for JSON (handle inf/nan) def sanitize_value(val): if isinstance(val, float): @@ -131,7 +132,7 @@ class AlertState: if math.isnan(val): return None return val - + result = { "metric_path": self.metric_path, "level": self.level.name, @@ -141,11 +142,11 @@ class AlertState: "notification_count": self.notification_count, "acknowledged": self.acknowledged, } - + # Include acknowledgment timestamp if acknowledged if self.acknowledged_at is not None: result["acknowledged_at"] = self.acknowledged_at - + # Include threshold info if available if self.threshold_value is not None: result["threshold_value"] = sanitize_value(self.threshold_value) @@ -164,7 +165,7 @@ class AlertState: result["recovery_threshold"] = round(self.threshold_value + ha, 4) return result - + def __setstate__(self, state): """Restore from pickle, backfilling fields added after the pickle was written.""" self.__dict__.update(state) @@ -182,6 +183,7 @@ class AlertState: def __str__(self): return self.to_dict().__str__() + class ThresholdConfig: """Configuration for a single threshold check.""" @@ -218,7 +220,7 @@ class ThresholdConfig: self.display = display self.count = max(1, int(count)) self.grace = float(grace) if grace is not None else None - + # Parse operator try: self.operator = ComparisonOperator(operator) @@ -229,7 +231,7 @@ class ThresholdConfig: metric_path ) self.operator = ComparisonOperator.GT - + def evaluate(self, value: float) -> AlertLevel: """ Evaluate a value against this threshold. @@ -271,7 +273,7 @@ class ThresholdConfig: return AlertLevel.WARNING return AlertLevel.OK - + def evaluate_with_hysteresis( self, value: float, @@ -279,11 +281,11 @@ class ThresholdConfig: ) -> AlertLevel: """ Evaluate with hysteresis to prevent flapping. - + Args: value: Current metric value current_level: Current alert level - + Returns: New alert level considering hysteresis """ @@ -296,7 +298,7 @@ class ThresholdConfig: # If no hysteresis, return new level if self.hysteresis == 0.0: return new_level - + # If improving (going to a lower severity), apply hysteresis if new_level.value < current_level.value: # For recovery, value must be better by hysteresis amount @@ -306,10 +308,10 @@ class ThresholdConfig: threshold = self.warning else: return new_level - + # Calculate hysteresis threshold hysteresis_amount = abs(threshold * self.hysteresis) - + if self.operator in [ComparisonOperator.GT, ComparisonOperator.GTE]: # For "greater than" thresholds, value must go below by hysteresis recovery_threshold = threshold - hysteresis_amount @@ -322,9 +324,9 @@ class ThresholdConfig: if value <= recovery_threshold: # Not enough improvement, keep current level return current_level - + return new_level - + def _compare(self, value: float, threshold: float) -> bool: """Perform comparison based on operator.""" if self.operator == ComparisonOperator.GT: @@ -344,7 +346,7 @@ class ThresholdConfig: class ThresholdChecker: """Main threshold checking and alerting system.""" - + def __init__( self, config: Dict[str, Any], @@ -353,7 +355,7 @@ class ThresholdChecker: ): """ Initialize threshold checker. - + Args: config: Threshold configuration dictionary from YAML renotify_interval: Seconds between repeat notifications (default: 1 hour) @@ -373,14 +375,14 @@ class ThresholdChecker: # Default config name to use when no mapping exists self.default_config = "default" - + self.renotify_interval = renotify_interval self.grace_seconds: float = float(config.get("grace", 2)) self.journal = journal # Parse configuration self._parse_config(config) - + total_thresholds = sum(len(cfg) for cfg in self.threshold_configs.values()) if total_thresholds == 0 and len(self.thresholds) > 0: # Backward compatibility: using single threshold set @@ -392,18 +394,18 @@ class ThresholdChecker: len(self.threshold_configs), total_thresholds ) - + def reload(self, config: Dict[str, Any]): """Reload threshold configuration from new config dict. - + This clears all existing thresholds and re-parses from the new configuration. Alert states are preserved to maintain hysteresis across reloads. - + Args: config: New configuration dictionary """ logger.info("Reloading threshold configuration...") - + # Clear old configuration self.threshold_configs.clear() self.threshold_raw_configs.clear() @@ -413,13 +415,13 @@ class ThresholdChecker: # Parse new configuration self._parse_config(config) - + total_thresholds = sum(len(cfg) for cfg in self.threshold_configs.values()) if total_thresholds == 0 and len(self.thresholds) > 0: total_thresholds = len(self.thresholds) - + logger.info("Threshold configuration reloaded: %d total thresholds", total_thresholds) - + def _parse_config(self, config: Dict[str, Any]): """Parse threshold configuration from YAML structure. @@ -449,7 +451,7 @@ class ThresholdChecker: self._parse_legacy_config(config) else: logger.info("No thresholds configured") - + def _parse_multi_config(self, config: Dict[str, Any]): """Parse multiple named threshold configurations.""" threshold_configs = config.get("threshold_configs", {}) @@ -545,34 +547,34 @@ class ThresholdChecker: if isinstance(legacy_mapping, dict): for host_name, value in legacy_mapping.items(): self.host_config_mapping[host_name] = _normalise(value) - + # Set default config (first one alphabetically or explicitly set) self.default_config = config.get("default_threshold_config", "default") if self.default_config not in self.threshold_configs and self.threshold_configs: # Use first available config as default self.default_config = sorted(self.threshold_configs.keys())[0] logger.info("Using '%s' as default threshold config", self.default_config) - + logger.info( "Loaded %d threshold configurations with %d host mappings", len(self.threshold_configs), len(self.host_config_mapping) ) - + def _parse_legacy_config(self, config: Dict[str, Any]): """Parse legacy single threshold configuration for backward compatibility.""" if not config or "thresholds" not in config: logger.info("No thresholds configured") return - + thresholds_config = config["thresholds"] - + for plugin_name, plugin_thresholds in thresholds_config.items(): if not isinstance(plugin_thresholds, dict): continue - + self._parse_plugin_thresholds(plugin_name, plugin_thresholds, target_dict=self.thresholds) - + def _parse_plugin_thresholds( self, plugin_name: str, @@ -580,7 +582,7 @@ class ThresholdChecker: target_dict: Optional[Dict[str, ThresholdConfig]] = None ): """Parse thresholds for a specific plugin. - + Args: plugin_name: Name of the plugin thresholds: Threshold configuration dictionary @@ -588,7 +590,7 @@ class ThresholdChecker: """ if target_dict is None: target_dict = self.thresholds - + # Special handling for RTT thresholds (per-host) if plugin_name == "rtt": self._parse_rtt_thresholds(thresholds, target_dict) @@ -606,7 +608,7 @@ class ThresholdChecker: for metric_name, threshold_config in thresholds.items(): if not isinstance(threshold_config, dict): continue - + # Handle nested metrics (e.g., partitions./.percent or pools.*.status) if metric_name == "partitions": self._parse_partition_thresholds(plugin_name, threshold_config, target_dict) @@ -614,9 +616,9 @@ class ThresholdChecker: if metric_name == "pools": self._parse_pool_thresholds(plugin_name, threshold_config, target_dict) continue - + metric_path = f"{plugin_name}.{metric_name}" - + # Extract threshold values warning = threshold_config.get("warning") critical = threshold_config.get("critical") @@ -643,7 +645,7 @@ class ThresholdChecker: display=display, grace=grace, ) - + target_dict[metric_path] = threshold logger.debug( "Registered threshold for %s: warn=%s, crit=%s, op=%s", @@ -652,7 +654,7 @@ class ThresholdChecker: critical, operator ) - + def _parse_partition_thresholds( self, plugin_name: str, @@ -660,7 +662,7 @@ class ThresholdChecker: target_dict: Optional[Dict[str, ThresholdConfig]] = None ): """Parse partition-specific thresholds for disk monitoring. - + Args: plugin_name: Name of the plugin partitions: Partition threshold configuration @@ -668,18 +670,18 @@ class ThresholdChecker: """ if target_dict is None: target_dict = self.thresholds - + for partition, metrics in partitions.items(): if not isinstance(metrics, dict): continue - + for metric_name, threshold_config in metrics.items(): if not isinstance(threshold_config, dict): continue - + # Create metric path like "disk_monitor./dev/sda1.percent" metric_path = f"{plugin_name}.{partition}.{metric_name}" - + warning = threshold_config.get("warning") critical = threshold_config.get("critical") operator = threshold_config.get("operator", ">") @@ -700,7 +702,7 @@ class ThresholdChecker: display=display, grace=grace, ) - + target_dict[metric_path] = threshold def _parse_pool_thresholds( @@ -761,26 +763,26 @@ class ThresholdChecker: target_dict: Optional[Dict[str, ThresholdConfig]] = None ): """Parse RTT thresholds (network latency thresholds). - + RTT thresholds are configured as: thresholds: rtt: warning: 100.0 # ms critical: 500.0 # ms - + Args: rtt_thresholds: RTT threshold configuration target_dict: Dictionary to store parsed thresholds """ if target_dict is None: target_dict = self.thresholds - + if not isinstance(rtt_thresholds, dict): return - + # Metric path is simply "rtt" (not per-host) metric_path = "rtt" - + warning = rtt_thresholds.get("warning") critical = rtt_thresholds.get("critical") operator = rtt_thresholds.get("operator", ">") @@ -813,7 +815,7 @@ class ThresholdChecker: critical, count, ) - + def get_thresholds_for_host(self, host_name: str) -> Dict[str, ThresholdConfig]: """Get the effective threshold configuration for a host. @@ -864,7 +866,7 @@ class ThresholdChecker: else: result.update(raw) return result - + def check_value( self, host_name: str, @@ -874,30 +876,30 @@ class ThresholdChecker: ) -> Optional[Tuple[AlertLevel, AlertLevel]]: """ Check a single value against configured threshold. - + Args: host_name: Name of the host metric_path: Full metric path (e.g., "rtt.hostname") value: The metric value to check alert_states: Host's alert_states dictionary - + Returns: Tuple of (old_level, new_level) if state changed, None otherwise """ # Get host-specific thresholds thresholds = self.get_thresholds_for_host(host_name) - + if metric_path not in thresholds: return None - + threshold = thresholds[metric_path] - + # Get or create alert state if metric_path not in alert_states: alert_states[metric_path] = AlertState(metric_path) - + alert_state = alert_states[metric_path] - + # Evaluate threshold with hysteresis new_level = threshold.evaluate_with_hysteresis( value, @@ -946,6 +948,7 @@ class ThresholdChecker: self._check_pending_or_renotify(host_name, alert_state, metric_path, value, threshold, None) return None + def _find_threshold( self, thresholds: Dict[str, "ThresholdConfig"], metric_path: str ) -> Tuple[Optional["ThresholdConfig"], Optional[str]]: @@ -985,21 +988,21 @@ class ThresholdChecker: ) -> list: """ Check plugin data against configured thresholds. - + Args: host_name: Name of the host plugin_name: Name of the plugin data: Plugin data dictionary alert_states: Host's alert_states dictionary - + Returns: List of (metric_path, old_level, new_level, value) tuples for state changes """ state_changes = [] - + # Get host-specific thresholds thresholds = self.get_thresholds_for_host(host_name) - + # Check flat metrics for metric_name, value in data.items(): metric_path = f"{plugin_name}.{metric_name}" @@ -1045,9 +1048,9 @@ class ThresholdChecker: alert_states, state_changes ) - + return state_changes - + def _check_nested_metrics( self, host_name: str, @@ -1059,7 +1062,7 @@ class ThresholdChecker: """Check nested metrics like partition-specific thresholds.""" # Get host-specific thresholds thresholds = self.get_thresholds_for_host(host_name) - + # ZFS pool health checks if plugin_name == "zfs_monitor" and "pools" in data: pools = data["pools"] @@ -1103,29 +1106,29 @@ class ThresholdChecker: partitions = data["partitions"] if not isinstance(partitions, dict): return - + for partition, metrics in partitions.items(): if not isinstance(metrics, dict): continue - + for metric_name, value in metrics.items(): metric_path = f"{plugin_name}.{partition}.{metric_name}" - + if metric_path not in thresholds: continue - + threshold = thresholds[metric_path] - + if metric_path not in alert_states: alert_states[metric_path] = AlertState(metric_path) - + alert_state = alert_states[metric_path] - + new_level = threshold.evaluate_with_hysteresis( value, alert_state.level ) - + # Determine which threshold was exceeded threshold_value = None if new_level == AlertLevel.CRITICAL and threshold.critical is not None: @@ -1141,7 +1144,7 @@ class ThresholdChecker: self._apply_grace(host_name, alert_state, metric_path, old_level, new_level, value, threshold, data) elif new_level != AlertLevel.OK: self._check_pending_or_renotify(host_name, alert_state, metric_path, value, threshold, data) - + def _trigger_notification( self, host_name: str, @@ -1155,7 +1158,7 @@ class ThresholdChecker: metric_name: Optional[str] = None, ): """Trigger a notification for an alert state change. - + Args: host_name: Name of the host metric_path: Full metric path @@ -1171,7 +1174,7 @@ class ThresholdChecker: threshold_value = threshold.critical elif new_level == AlertLevel.WARNING and threshold.warning is not None: threshold_value = threshold.warning - + # Format operator symbol op_symbol = threshold.operator.value @@ -1223,7 +1226,7 @@ class ThresholdChecker: formatted_threshold_msg = _fmt() if has_display and new_level != AlertLevel.OK else None return lvl, message, formatted_threshold_msg - + def _send_notification( self, host_name: str, @@ -1254,7 +1257,7 @@ class ThresholdChecker: service=short_path, ), )) - + # Log to journal if self.journal is not None: try: @@ -1270,7 +1273,7 @@ class ThresholdChecker: logger.debug(f"Failed to log threshold event to journal: {e}") # Log to eventlog as well eventlog(host_name, lvl, message, service="threshold") - + def _format_display( self, display_format: str, @@ -1330,7 +1333,7 @@ class ThresholdChecker: status = plugin_data.get(f"{check_name}_status") if status is not None: format_context['status'] = status - + try: # Format the display string return display_format.format(**format_context) @@ -1349,7 +1352,7 @@ class ThresholdChecker: e ) return f"(threshold: {op_symbol} {threshold_value})" - + def _apply_grace( self, host_name: str, @@ -1472,7 +1475,7 @@ class ThresholdChecker: metric_name: Optional[str] = None, ): """Check if we should send a repeat notification. - + Args: host_name: Name of the host alert_state: Current alert state @@ -1487,16 +1490,16 @@ class ThresholdChecker: # Skip reminders if alert has been acknowledged if alert_state.acknowledged: return - + now = time.time() - + # Check if we should re-notify if alert_state.last_notification is None: # First notification already sent during state change alert_state.last_notification = now alert_state.notification_count = 1 return - + if (now - alert_state.last_notification) >= self.renotify_interval: # Determine which threshold is active threshold_value = None @@ -1504,7 +1507,7 @@ class ThresholdChecker: threshold_value = threshold.critical elif alert_state.level == AlertLevel.WARNING and threshold.warning is not None: threshold_value = threshold.warning - + # Format operator symbol op_symbol = threshold.operator.value short_path = (metric_path.partition(".")[2] or metric_path).removesuffix("_status_code") @@ -1541,7 +1544,7 @@ class ThresholdChecker: logger.info("Re-notification sent: %s", message) alert_state.last_notification = now alert_state.notification_count += 1 - + def purge_stale_alerts(self, hbdclass) -> None: """Remove alert states that have no matching threshold configuration. @@ -1578,10 +1581,10 @@ class ThresholdChecker: def get_active_alerts(self, alert_states: Dict[str, AlertState]) -> list: """ Get all currently active (non-OK) alerts. - + Args: alert_states: Host's alert_states dictionary - + Returns: List of AlertState objects that are not OK """ @@ -1589,19 +1592,19 @@ class ThresholdChecker: state for state in alert_states.values() if state.level != AlertLevel.OK ] - + def get_alert_summary(self, alert_states: Dict[str, AlertState]) -> Dict[str, int]: """ Get summary counts of alert levels. - + Args: alert_states: Host's alert_states dictionary - + Returns: Dictionary with counts: {"ok": N, "warning": N, "critical": N} """ summary = {"ok": 0, "warning": 0, "critical": 0, "unknown": 0} - + for state in alert_states.values(): if state.level == AlertLevel.OK: summary["ok"] += 1 @@ -1611,5 +1614,5 @@ class ThresholdChecker: summary["critical"] += 1 elif state.level == AlertLevel.UNKNOWN: summary["unknown"] += 1 - + return summary diff --git a/hbd/server/udp.py b/hbd/server/udp.py index 50a7e29..07cea9f 100644 --- a/hbd/server/udp.py +++ b/hbd/server/udp.py @@ -23,14 +23,14 @@ _STALE_INTERVAL_MULTIPLIER = 3 # SO_TIMESTAMP: kernel attaches a struct timeval to each received datagram. # Supported on Linux, FreeBSD, and macOS. The constant is not exposed by -# Python's socket module on all platforms +# Python's socket module on all platforms platform = platform_system() if platform == "Darwin": _SO_TIMESTAMP = 1024 # SO_TIMESTAMP on macOS (not in Python's socket module) elif platform == "Linux": _SO_TIMESTAMP = 29 # Linux value (not in older Python versions) elif platform == "FreeBSD": - _SO_TIMESTAMP = 32 # FreeBSD value (not in older Python versions) + _SO_TIMESTAMP = 32 # FreeBSD value (not in older Python versions) else: logger.warning("SO_TIMESTAMP may not be supported on this platform (%s)", platform) _SO_TIMESTAMP = None @@ -71,6 +71,7 @@ class RecvmsgTransport: Exposes the same sendto() / close() interface as asyncio's DatagramTransport so the rest of the code does not need to know which path is in use. """ + def __init__(self, loop, sock): self._loop = loop self._sock = sock @@ -410,7 +411,7 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): if not msg: return now = ctx.get("recv_ts") or time.time() - + # Log message to journal msg_journal = ctx.get("msg_journal") if msg_journal: @@ -421,7 +422,7 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): loop.create_task(msg_journal.log_message(msg, addr, now)) except Exception as e: logger.debug(f"Failed to log message to journal: {e}") - + # Only named HTB/PLG datagrams come from clients; anything else (e.g. an internet scanner's # SIP probe) is journaled above but must not sign on a host. if msg.get("ID") not in ("HTB", "PLG") or not msg.get("name"): @@ -432,7 +433,6 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): hbdcls = ctx.get("hbdclass") msg_to_websockets = ctx.get("msg_to_websockets") DEBUG = ctx.get("DEBUG", 0) - verbose = ctx.get("verbose", False) # normalize addr (ip, port) ip = addr[0] if isinstance(addr, (list, tuple)) else addr @@ -457,7 +457,7 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): else: host = hbdcls.Host.hosts[uname] newh = False - + cid = msg.get("id", 0) try: rtt = float(msg.get("rtt")) @@ -523,14 +523,14 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): # If os_info reports an owner and none is configured server-side, apply it if plugin_name == "os_info": - config_owner = config_mod.get_host_access(cfg, uname).get("owner") + config_owner = config_mod.get_host_access(cfg, uname).get("owner") default_owner = config_mod.get_default_owner(cfg) inferred_owner = config_owner or plugin_data.get("owner") or default_owner host.owner = inferred_owner logger.info(f"owner for {uname} is {host.owner}") if DEBUG > 1: print(f"Stored plugin data for {uname}: {plugin_name}") - + # Check thresholds if checker is available threshold_checker = ctx.get("threshold_checker") if threshold_checker: @@ -545,7 +545,7 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): print(f"Threshold state changes for {uname}: {state_changes}") except Exception as e: logger.error(f"Error checking thresholds for {uname}.{plugin_name}: {e}") - + # Notify websockets of plugin update if msg_to_websockets: try: @@ -671,13 +671,13 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict): timeout_seconds = interval + grace on_overdue, _ = _make_timer_callbacks(uname, host, ctx) conn.reset_overdue_timer(timeout_seconds, on_overdue) - + # Check RTT thresholds using the threshold checker threshold_checker = ctx.get("threshold_checker") if threshold_checker and rtt and rtt > 0: # Metric path for RTT is simply "rtt" metric_path = "rtt" - + # Check against configured thresholds (handles alerts, notifications, etc.) threshold_checker.check_value( host_name=uname, diff --git a/tests/test_configio.py b/tests/test_configio.py index 066137a..c75d220 100644 --- a/tests/test_configio.py +++ b/tests/test_configio.py @@ -1,5 +1,3 @@ -import glob -import os import pytest from hbd.server import configio diff --git a/tests/test_http_config_api.py b/tests/test_http_config_api.py index 83f5224..240e97d 100644 --- a/tests/test_http_config_api.py +++ b/tests/test_http_config_api.py @@ -1,5 +1,4 @@ """Tests for the config read/write API helpers in http.py.""" -import pytest from hbd.server import http diff --git a/tests/test_http_users_me.py b/tests/test_http_users_me.py index 33aa39c..bcdcfb7 100644 --- a/tests/test_http_users_me.py +++ b/tests/test_http_users_me.py @@ -1,5 +1,4 @@ """Tests for PUT /api/0/users/me logic.""" -import pytest from hbd.server import users as users_mod @@ -87,7 +86,6 @@ def test_put_users_me_notification_channels(tmp_path): def test_visible_channels_excludes_private_from_others(): """Private channels owned by another user must not appear in the visible set.""" - from hbd.server import settings as settings_mod config = { "notification_channels": { @@ -105,7 +103,7 @@ def test_visible_channels_excludes_private_from_others(): self.admin = admin alice = FakeUser("alice") - bob = FakeUser("bob") + bob = FakeUser("bob") admin = FakeUser("admin", admin=True) # Simulate _visible_channels_for_user logic (mirrors http.py implementation) @@ -119,5 +117,5 @@ def test_visible_channels_excludes_private_from_others(): } assert visible(alice) == {"public_ch", "alice_priv"} - assert visible(bob) == {"public_ch", "bob_priv"} + assert visible(bob) == {"public_ch", "bob_priv"} assert visible(admin) == {"public_ch", "alice_priv", "bob_priv"} diff --git a/tests/test_notification_channels_api.py b/tests/test_notification_channels_api.py index 79fa653..129292b 100644 --- a/tests/test_notification_channels_api.py +++ b/tests/test_notification_channels_api.py @@ -1,5 +1,4 @@ """Tests for notification channel CRUD via configio helpers and visibility logic.""" -import pytest from hbd.server import configio, settings as settings_mod diff --git a/tests/test_oauth.py b/tests/test_oauth.py index bd0bc60..0008044 100644 --- a/tests/test_oauth.py +++ b/tests/test_oauth.py @@ -37,7 +37,6 @@ def reset_users_dict(): users_mod.users = original - def test_make_state_returns_unique_tokens(): s1 = oauth.make_state() s2 = oauth.make_state() @@ -124,7 +123,6 @@ def test_provision_oauth_user_survives_config_reload(): assert "oauthonly" in users_mod.users - # --------------------------------------------------------------------------- # Integration-style tests: callback logic chain # --------------------------------------------------------------------------- diff --git a/tests/test_settings_sections.py b/tests/test_settings_sections.py index 7e1ed3b..89fad2a 100644 --- a/tests/test_settings_sections.py +++ b/tests/test_settings_sections.py @@ -1,4 +1,3 @@ -import pytest from hbd.server import settings as settings_mod CFG = { diff --git a/tests/test_threshold.py b/tests/test_threshold.py index 95a3720..f120170 100644 --- a/tests/test_threshold.py +++ b/tests/test_threshold.py @@ -13,13 +13,12 @@ Tests cover: """ import sys -import time from pathlib import Path # Add parent directory to path for imports sys.path.insert(0, str(Path(__file__).parent.parent)) -from hbd.server.threshold import ( +from hbd.server.threshold import ( # noqa: E402 ThresholdChecker, ThresholdConfig, AlertLevel, @@ -31,37 +30,37 @@ from hbd.server.threshold import ( def test_threshold_config_basic(): """Test basic threshold configuration.""" print("Test 1: Basic threshold configuration...") - + config = ThresholdConfig( metric_path="cpu_monitor.cpu_percent", warning=80.0, critical=90.0, operator=">", ) - + # Test below warning result = config.evaluate(50.0) assert result == AlertLevel.OK, f"Expected OK, got {result}" - + # Test at warning result = config.evaluate(80.0) assert result == AlertLevel.OK, f"Expected OK at boundary, got {result}" - + # Test above warning but below critical result = config.evaluate(85.0) assert result == AlertLevel.WARNING, f"Expected WARNING, got {result}" - + # Test above critical result = config.evaluate(95.0) assert result == AlertLevel.CRITICAL, f"Expected CRITICAL, got {result}" - + print(" ✓ Basic threshold configuration works") def test_threshold_operators(): """Test all comparison operators.""" print("\nTest 2: Comparison operators...") - + # Greater than operator config_gt = ThresholdConfig( metric_path="test.metric", @@ -71,7 +70,7 @@ def test_threshold_operators(): ) assert config_gt.evaluate(85.0) == AlertLevel.WARNING assert config_gt.evaluate(75.0) == AlertLevel.OK - + # Less than operator (for inverse thresholds like available memory) config_lt = ThresholdConfig( metric_path="memory.available_mb", @@ -82,7 +81,7 @@ def test_threshold_operators(): assert config_lt.evaluate(800) == AlertLevel.WARNING, "Should warn when below 1000" assert config_lt.evaluate(400) == AlertLevel.CRITICAL, "Should be critical when below 500" assert config_lt.evaluate(1500) == AlertLevel.OK, "Should be OK when above 1000" - + # Greater than or equal config_gte = ThresholdConfig( metric_path="test.metric", @@ -91,7 +90,7 @@ def test_threshold_operators(): ) assert config_gte.evaluate(80.0) == AlertLevel.WARNING assert config_gte.evaluate(79.9) == AlertLevel.OK - + # Less than or equal config_lte = ThresholdConfig( metric_path="test.metric", @@ -100,14 +99,14 @@ def test_threshold_operators(): ) assert config_lte.evaluate(20.0) == AlertLevel.WARNING assert config_lte.evaluate(20.1) == AlertLevel.OK - + print(" ✓ All comparison operators work correctly") def test_hysteresis(): """Test hysteresis to prevent flapping.""" print("\nTest 3: Hysteresis...") - + config = ThresholdConfig( metric_path="cpu_monitor.cpu_percent", warning=80.0, @@ -115,87 +114,87 @@ def test_hysteresis(): operator=">", hysteresis=0.1, # 10% hysteresis ) - + # Start at OK, go to WARNING result = config.evaluate_with_hysteresis(85.0, AlertLevel.OK) assert result == AlertLevel.WARNING, "Should enter WARNING state" - + # Try to recover with insufficient improvement (within hysteresis) # Warning threshold is 80, hysteresis is 10%, so need to go below 80 - 8 = 72 result = config.evaluate_with_hysteresis(77.0, AlertLevel.WARNING) assert result == AlertLevel.WARNING, "Should stay in WARNING (hysteresis)" - + # Recover with sufficient improvement result = config.evaluate_with_hysteresis(70.0, AlertLevel.WARNING) assert result == AlertLevel.OK, "Should recover to OK" - + # Test critical hysteresis result = config.evaluate_with_hysteresis(95.0, AlertLevel.WARNING) assert result == AlertLevel.CRITICAL, "Should escalate to CRITICAL" - + # Try to recover from critical with insufficient improvement # Critical threshold is 90, hysteresis is 10%, so need to go below 90 - 9 = 81 result = config.evaluate_with_hysteresis(85.0, AlertLevel.CRITICAL) assert result == AlertLevel.CRITICAL, "Should stay CRITICAL (hysteresis, still above 81)" - + # Sufficient improvement to drop from CRITICAL (below 81) result = config.evaluate_with_hysteresis(75.0, AlertLevel.CRITICAL) assert result == AlertLevel.OK, "Should drop to OK (below warning threshold)" - + # Now test dropping from CRITICAL to WARNING result = config.evaluate_with_hysteresis(95.0, AlertLevel.OK) assert result == AlertLevel.CRITICAL, "Should go to CRITICAL" - + # Drop to between warning and critical, but still in hysteresis zone result = config.evaluate_with_hysteresis(82.0, AlertLevel.CRITICAL) assert result == AlertLevel.CRITICAL, "Should stay CRITICAL (in hysteresis)" - + # Drop below critical hysteresis but still above warning threshold # At 80.1, we're above WARNING (80) so should evaluate to WARNING result = config.evaluate_with_hysteresis(80.5, AlertLevel.CRITICAL) assert result == AlertLevel.WARNING, "Should drop to WARNING when below critical hysteresis" - + print(" ✓ Hysteresis prevents flapping") def test_alert_state(): """Test alert state tracking.""" print("\nTest 4: Alert state tracking...") - + alert = AlertState("cpu_monitor.cpu_percent") - + # Initial state assert alert.level == AlertLevel.OK assert alert.notification_count == 0 - + # Update to WARNING - should trigger notification changed = alert.update(AlertLevel.WARNING, 85.0) - assert changed == True, "State change should return True" + assert changed is True, "State change should return True" assert alert.level == AlertLevel.WARNING assert alert.last_value == 85.0 - + # Update with same level - no notification changed = alert.update(AlertLevel.WARNING, 86.0) - assert changed == False, "No state change should return False" + assert changed is False, "No state change should return False" assert alert.last_value == 86.0 - + # Escalate to CRITICAL changed = alert.update(AlertLevel.CRITICAL, 95.0) - assert changed == True, "Escalation should trigger notification" + assert changed is True, "Escalation should trigger notification" assert alert.level == AlertLevel.CRITICAL - + # Recover to OK changed = alert.update(AlertLevel.OK, 50.0) - assert changed == True, "Recovery should trigger notification" + assert changed is True, "Recovery should trigger notification" assert alert.level == AlertLevel.OK - + print(" ✓ Alert state tracking works correctly") def test_threshold_checker_parsing(): """Test parsing threshold configuration from YAML structure.""" print("\nTest 5: Configuration parsing...") - + config = { "thresholds": { "cpu_monitor": { @@ -239,9 +238,9 @@ def test_threshold_checker_parsing(): }, } } - + checker = ThresholdChecker(config) - + # Verify thresholds were parsed assert "cpu_monitor.cpu_percent" in checker.thresholds assert "cpu_monitor.load_1min" in checker.thresholds @@ -249,11 +248,11 @@ def test_threshold_checker_parsing(): assert "memory_monitor.available_mb" in checker.thresholds assert "disk_monitor./.percent" in checker.thresholds assert "disk_monitor./home.percent" in checker.thresholds - + # Verify operators were parsed correctly assert checker.thresholds["cpu_monitor.cpu_percent"].operator == ComparisonOperator.GT assert checker.thresholds["memory_monitor.available_mb"].operator == ComparisonOperator.LT - + print(f" ✓ Parsed {len(checker.thresholds)} thresholds correctly") @@ -269,7 +268,7 @@ def _checker_with_notifications(config, notifications, monkeypatch): def test_check_plugin_data(monkeypatch): """Test checking plugin data against thresholds.""" print("\nTest 6: Plugin data checking...") - + config = { "thresholds": { "cpu_monitor": { @@ -284,53 +283,53 @@ def test_check_plugin_data(monkeypatch): }, } } - + notifications = [] checker = _checker_with_notifications(config, notifications, monkeypatch) alert_states = {} - + # First check - OK plugin_data = { "cpu_percent": 50.0, "load_1min": 2.0, } - + state_changes = checker.check_plugin_data( host_name="testhost", plugin_name="cpu_monitor", data=plugin_data, alert_states=alert_states, ) - + assert len(state_changes) == 0, "No thresholds violated, no state changes" assert len(notifications) == 0, "No notifications should be sent" - + # Second check - WARNING plugin_data = { "cpu_percent": 85.0, "load_1min": 2.0, } - + state_changes = checker.check_plugin_data( host_name="testhost", plugin_name="cpu_monitor", data=plugin_data, alert_states=alert_states, ) - + assert len(state_changes) == 1, "One metric should change state" assert state_changes[0][0] == "cpu_monitor.cpu_percent" assert state_changes[0][2] == AlertLevel.WARNING assert len(notifications) == 1, "One notification should be sent" assert "WARNING" in notifications[0] assert "testhost" in notifications[0] - + # Third check - CRITICAL plugin_data = { "cpu_percent": 95.0, "load_1min": 9.0, } - + notifications.clear() state_changes = checker.check_plugin_data( host_name="testhost", @@ -338,16 +337,16 @@ def test_check_plugin_data(monkeypatch): data=plugin_data, alert_states=alert_states, ) - + assert len(state_changes) == 2, "Two metrics should change state" assert len(notifications) == 2, "Two notifications should be sent" - + # Fourth check - Recovery plugin_data = { "cpu_percent": 50.0, "load_1min": 1.0, } - + notifications.clear() state_changes = checker.check_plugin_data( host_name="testhost", @@ -355,18 +354,18 @@ def test_check_plugin_data(monkeypatch): data=plugin_data, alert_states=alert_states, ) - + assert len(state_changes) == 2, "Two metrics should recover" assert len(notifications) == 2, "Two recovery notifications" assert any("RECOVER" in n for n in notifications), "Should have recovery notification" - + print(" ✓ Plugin data checking and notifications work") def test_nested_metrics(monkeypatch): """Test checking nested metrics like disk partitions.""" print("\nTest 7: Nested metrics (partitions)...") - + config = { "thresholds": { "disk_monitor": { @@ -387,11 +386,11 @@ def test_nested_metrics(monkeypatch): }, } } - + notifications = [] checker = _checker_with_notifications(config, notifications, monkeypatch) alert_states = {} - + plugin_data = { "partitions": { "/": { @@ -404,26 +403,26 @@ def test_nested_metrics(monkeypatch): }, }, } - + state_changes = checker.check_plugin_data( host_name="testhost", plugin_name="disk_monitor", data=plugin_data, alert_states=alert_states, ) - + assert len(state_changes) == 1, "One partition should trigger alert" assert "/home" in state_changes[0][0], "Should be /home partition" assert state_changes[0][2] == AlertLevel.WARNING assert len(notifications) == 1 - + print(" ✓ Nested metric checking works") def test_alert_summary(): """Test getting alert summaries.""" print("\nTest 8: Alert summaries...") - + config = { "thresholds": { "cpu_monitor": { @@ -435,31 +434,31 @@ def test_alert_summary(): }, } } - + checker = ThresholdChecker(config) alert_states = {} - + # Create some alert states plugin_data = { "cpu_percent": 85.0, # WARNING "load_1min": 9.0, # CRITICAL } checker.check_plugin_data("testhost", "cpu_monitor", plugin_data, alert_states) - + plugin_data = { "percent": 96.0, # CRITICAL } checker.check_plugin_data("testhost", "memory_monitor", plugin_data, alert_states) - + # Get summary summary = checker.get_alert_summary(alert_states) assert summary["warning"] == 1, "Should have 1 warning" assert summary["critical"] == 2, "Should have 2 critical" - + # Get active alerts active = checker.get_active_alerts(alert_states) assert len(active) == 3, "Should have 3 active alerts" - + print(" ✓ Alert summaries work correctly") @@ -468,7 +467,7 @@ def run_all_tests(): print("=" * 70) print("THRESHOLD SYSTEM TEST SUITE") print("=" * 70) - + try: test_threshold_config_basic() test_threshold_operators() @@ -478,12 +477,12 @@ def run_all_tests(): test_check_plugin_data() test_nested_metrics() test_alert_summary() - + print("\n" + "=" * 70) print("✓ ALL TESTS PASSED") print("=" * 70) return 0 - + except AssertionError as e: print(f"\n✗ TEST FAILED: {e}") import traceback