from __future__ import annotations import logging import os import re import shutil import socket import time import threading from pathlib import Path from typing import Any from concurrent.futures import ThreadPoolExecutor, as_completed from logger import get_logger log = get_logger("metrics") # Cache storage _metrics_cache: dict[str, Any] = {} _cache_lock = threading.Lock() _updater_thread: threading.Thread | None = None _stop_event = threading.Event() # Caches for slow collectors _docker_cache_data: list[dict[str, Any]] | None = None _docker_cache_time: float = 0.0 _docker_cache_lock = threading.Lock() _ssl_cache_data: list[dict[str, Any]] | None = None _ssl_cache_time: float = 0.0 _ssl_cache_lock = threading.Lock() # Historical counters for calculating rates _last_cpu_times: tuple[float, float, float, float] | None = None _last_net_stats: dict[str, dict[str, Any]] | None = None _last_proc_stats: dict[int, dict[str, Any]] = {} _last_disk_stats: tuple[float, float] | None = None def get_uid_map() -> dict[int, str]: """Parse /etc/passwd into a uid -> username mapping.""" uid_map = {} try: with open("/etc/passwd", "r", encoding="utf-8") as f: for line in f: parts = line.strip().split(":") if len(parts) >= 3: try: uid_map[int(parts[2])] = parts[0] except ValueError: pass except Exception: pass return uid_map def run_readonly(cmd: list[str]) -> str | None: """Helper to run command and capture output (similar to discovery.system).""" import subprocess if not shutil.which(cmd[0]): return None try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=5, check=False, ) if result.returncode != 0: return None return result.stdout except Exception: return None def sanitize_cmdline(cmdline: str) -> str: """Mask sensitive parameters in process command lines.""" if not cmdline: return "" # 1. Handle key=value style assignments (e.g. --password=secret, api_key=123) pattern_eq = re.compile( r"(?i)(password|passwd|pwd|token|secret|api_key|apikey|access_token|refresh_token|authorization|bearer|--password|--token)\s*=\s*(\S+)" ) cmdline = pattern_eq.sub(r"\1=***", cmdline) # 2. Handle key value style flags/arguments (e.g. -p secret, --token secret) parts = cmdline.split() sanitized = [] skip_next = False sensitive_keys = { "--password", "--passwd", "--pwd", "--token", "--secret", "--api-key", "--api_key", "--apikey", "-p", "password", "passwd", "pwd", "token", "secret", "api_key", "apikey", "access_token", "refresh_token", "authorization", "bearer" } for i, part in enumerate(parts): if skip_next: suffix = "" if part.endswith("'"): suffix = "'" elif part.endswith('"'): suffix = '"' elif part.endswith(')'): suffix = ')' sanitized.append("***" + suffix) skip_next = False continue part_lower = part.lower() if part_lower in sensitive_keys: sanitized.append(part) if i + 1 < len(parts): skip_next = True continue if part.startswith("-p") and len(part) > 2: sanitized.append("-p***") continue matched_eq = False for k in sensitive_keys: if part_lower.startswith(k.lower() + ":"): sanitized.append(part.split(":", 1)[0] + ":***") matched_eq = True break if matched_eq: continue sanitized.append(part) return " ".join(sanitized) # --- Individual Collectors -------------------------------------------------- def collect_cpu() -> dict[str, Any]: """Collect CPU usage percent, core count, load average, steal and iowait.""" global _last_cpu_times cores = os.cpu_count() or 0 load_avg = [] try: load_avg = list(os.getloadavg()) except (OSError, AttributeError): pass usage_percent = 0.0 steal_percent = 0.0 iowait_percent = 0.0 try: if os.path.exists("/proc/stat"): with open("/proc/stat", "r", encoding="utf-8") as f: line = f.readline() if line.startswith("cpu "): parts = line.strip().split() # user, nice, sys, idle, iowait, irq, softirq, steal vals = [float(x) for x in parts[1:]] user, nice, system, idle, iowait, irq, softirq, steal = vals[:8] idle_all = idle + iowait total = sum(vals[:8]) # First run initialization if _last_cpu_times is None: _last_cpu_times = (total, idle_all, steal, iowait) time.sleep(0.05) with open("/proc/stat", "r", encoding="utf-8") as f2: line2 = f2.readline() if line2.startswith("cpu "): parts = line2.strip().split() vals = [float(x) for x in parts[1:]] user, nice, system, idle, iowait, irq, softirq, steal = vals[:8] idle_all = idle + iowait total = sum(vals[:8]) if _last_cpu_times is not None: last_total, last_idle_all, last_steal, last_iowait = _last_cpu_times total_diff = total - last_total if total_diff > 0: idle_diff = idle_all - last_idle_all usage_percent = round((1.0 - (idle_diff / total_diff)) * 100.0, 1) steal_percent = round(((steal - last_steal) / total_diff) * 100.0, 1) iowait_percent = round(((iowait - last_iowait) / total_diff) * 100.0, 1) _last_cpu_times = (total, idle_all, steal, iowait) except Exception as exc: log.debug("CPU collector error: %s", exc) return { "usage_percent": max(0.0, min(100.0, usage_percent)), "cores": cores, "load_average": load_avg, "steal_percent": max(0.0, min(100.0, steal_percent)), "iowait_percent": max(0.0, min(100.0, iowait_percent)), } def collect_ram() -> dict[str, Any]: """Collect RAM and Swap metrics from /proc/meminfo.""" ram_info = { "total": 0, "used": 0, "free": 0, "cached": 0, "buffers": 0, "swap": {"total": 0, "used": 0, "free": 0}, } try: values = {} if os.path.exists("/proc/meminfo"): with open("/proc/meminfo", "r", encoding="utf-8") as f: for line in f: parts = line.split(":") if len(parts) == 2: key = parts[0].strip() val_parts = parts[1].strip().split() if val_parts: try: values[key] = int(val_parts[0]) * 1024 # convert KiB to Bytes except ValueError: pass total = values.get("MemTotal", 0) free = values.get("MemFree", 0) cached = values.get("Cached", 0) buffers = values.get("Buffers", 0) used = max(0, total - free - buffers - cached) swap_total = values.get("SwapTotal", 0) swap_free = values.get("SwapFree", 0) swap_used = max(0, swap_total - swap_free) ram_info["total"] = total ram_info["used"] = used ram_info["free"] = free ram_info["cached"] = cached ram_info["buffers"] = buffers ram_info["swap"] = { "total": swap_total, "used": swap_used, "free": swap_free, } except Exception as exc: log.debug("RAM collector error: %s", exc) return ram_info def collect_disk() -> list[dict[str, Any]]: """Collect disk usage and inode count for each mount point.""" mounts = [] try: if os.path.exists("/proc/mounts"): with open("/proc/mounts", "r", encoding="utf-8") as f: for line in f: parts = line.strip().split() if len(parts) >= 3: dev, mount_point, fs_type = parts[0], parts[1], parts[2] is_physical = ( dev.startswith("/dev/") or fs_type in ("zfs", "btrfs", "xfs", "ext4", "ext3", "ext2", "f2fs") ) if is_physical and not mount_point.startswith(("/boot/efi", "/var/lib/docker")): try: usage = shutil.disk_usage(mount_point) statv = os.statvfs(mount_point) inodes_total = statv.f_files inodes_free = statv.f_ffree inodes_used = inodes_total - inodes_free mounts.append({ "mount_point": mount_point, "device": dev, "filesystem": fs_type, "total": usage.total, "used": usage.used, "free": usage.free, "inodes_total": inodes_total, "inodes_used": inodes_used, "inodes_free": inodes_free, }) except Exception: pass else: # Non-Linux fallback try: usage = shutil.disk_usage("/") mounts.append({ "mount_point": "/", "device": "root", "filesystem": "unknown", "total": usage.total, "used": usage.used, "free": usage.free, "inodes_total": 0, "inodes_used": 0, "inodes_free": 0, }) except Exception: pass except Exception as exc: log.debug("Disk collector error: %s", exc) return mounts def collect_network() -> list[dict[str, Any]]: """Collect network traffic details and rate metrics for each interface.""" global _last_net_stats interfaces = [] now = time.time() try: current_stats = {} if os.path.exists("/proc/net/dev"): with open("/proc/net/dev", "r", encoding="utf-8") as f: lines = f.readlines() for line in lines[2:]: parts = line.split(":") if len(parts) == 2: iface = parts[0].strip() if iface == "lo": continue cols = parts[1].strip().split() if len(cols) >= 12: try: rx_bytes = int(cols[0]) rx_packets = int(cols[1]) rx_errors = int(cols[2]) rx_drops = int(cols[3]) tx_bytes = int(cols[8]) tx_packets = int(cols[9]) tx_errors = int(cols[10]) tx_drops = int(cols[11]) current_stats[iface] = { "rx_bytes": rx_bytes, "rx_packets": rx_packets, "rx_errors": rx_errors, "rx_drops": rx_drops, "tx_bytes": tx_bytes, "tx_packets": tx_packets, "tx_errors": tx_errors, "tx_drops": tx_drops, "timestamp": now, } except ValueError: pass for iface, stats in current_stats.items(): rx_rate = 0.0 tx_rate = 0.0 if _last_net_stats and iface in _last_net_stats: last = _last_net_stats[iface] time_diff = now - last["timestamp"] if time_diff > 0: rx_rate = max(0.0, (stats["rx_bytes"] - last["rx_bytes"]) / time_diff) tx_rate = max(0.0, (stats["tx_bytes"] - last["tx_bytes"]) / time_diff) speed_mbps = -1 speed_path = f"/sys/class/net/{iface}/speed" if os.path.exists(speed_path): try: with open(speed_path, "r", encoding="utf-8") as sf: speed_mbps = int(sf.read().strip()) except Exception: pass interfaces.append({ "interface": iface, "rx_bytes": stats["rx_bytes"], "tx_bytes": stats["tx_bytes"], "rx_packets": stats["rx_packets"], "tx_packets": stats["tx_packets"], "rx_errors": stats["rx_errors"], "tx_errors": stats["tx_errors"], "rx_drops": stats["rx_drops"], "tx_drops": stats["tx_drops"], "rx_rate_bps": int(rx_rate * 8), "tx_rate_bps": int(tx_rate * 8), "speed_mbps": speed_mbps, }) _last_net_stats = current_stats except Exception as exc: log.debug("Network collector error: %s", exc) return interfaces def collect_process() -> dict[str, Any]: """Collect Top 10 CPU and RAM consuming processes.""" global _last_proc_stats top_cpu = [] top_ram = [] now = time.time() try: try: page_size = os.sysconf("SC_PAGE_SIZE") except (AttributeError, ValueError): page_size = 4096 try: clk_tck = os.sysconf("SC_CLK_TCK") except (AttributeError, ValueError): clk_tck = 100 # Retrieve boot time to calculate process start time btime = 0.0 if os.path.exists("/proc/stat"): with open("/proc/stat", "r", encoding="utf-8") as f: for line in f: if line.startswith("btime "): try: btime = float(line.strip().split()[1]) except (IndexError, ValueError): pass break uid_map = get_uid_map() # Pass 1: Scan all PIDs and parse only stat (very cheap) candidates = [] if os.path.exists("/proc"): for pid_dir in os.listdir("/proc"): if pid_dir.isdigit(): pid = int(pid_dir) proc_dir = f"/proc/{pid_dir}" try: with open(f"{proc_dir}/stat", "r", errors="ignore") as f: stat_content = f.read().strip() rpar_idx = stat_content.rfind(")") if rpar_idx != -1: name_part = stat_content[:rpar_idx] name = name_part[name_part.find("(") + 1:] cols = stat_content[rpar_idx + 2:].split() else: cols = stat_content.split() name = cols[1] if len(cols) > 1 else "unknown" cols = cols[2:] if len(cols) >= 22: utime = int(cols[11]) stime = int(cols[12]) rss = int(cols[21]) starttime = int(cols[19]) cpu_time = utime + stime ram_bytes = rss * page_size start_timestamp = btime + (starttime / clk_tck) # Calculate temporary cpu percent for sorting candidates cpu_percent = 0.0 if _last_proc_stats and pid in _last_proc_stats: last = _last_proc_stats[pid] time_diff = now - last["timestamp"] cpu_diff = cpu_time - last["cpu_time"] if time_diff > 0: cpu_percent = round((cpu_diff / clk_tck) / time_diff * 100.0, 1) candidates.append({ "pid": pid, "pid_dir": pid_dir, "proc_dir": proc_dir, "name": name, "cpu_time": cpu_time, "ram_bytes": ram_bytes, "start_time": start_timestamp, "cpu_percent": cpu_percent }) except (FileNotFoundError, PermissionError, ValueError, IndexError): continue if not candidates: return {"top_cpu": [], "top_ram": []} # Select Top 30 by CPU and Top 30 by RAM top_cpu_candidates = sorted(candidates, key=lambda x: x["cpu_percent"], reverse=True)[:30] top_ram_candidates = sorted(candidates, key=lambda x: x["ram_bytes"], reverse=True)[:30] selected_pids = set() selected_candidates = [] for c in top_cpu_candidates + top_ram_candidates: if c["pid"] not in selected_pids: selected_pids.add(c["pid"]) selected_candidates.append(c) # Pass 2: Parse expensive details only for selected candidates current_proc_stats = {} for c in selected_candidates: pid = c["pid"] proc_dir = c["proc_dir"] # Parse uid from status uid = 0 try: with open(f"{proc_dir}/status", "r", errors="ignore") as sf: for line in sf: if line.startswith("Uid:"): try: uid = int(line.strip().split()[1]) except (IndexError, ValueError): pass break except Exception: pass # Parse process io read_io_bytes = 0 write_io_bytes = 0 try: with open(f"{proc_dir}/io", "r", errors="ignore") as iof: for line in iof: if line.startswith("read_bytes:"): read_io_bytes = int(line.strip().split()[1]) elif line.startswith("write_bytes:"): write_io_bytes = int(line.strip().split()[1]) except Exception: pass cmdline = "" try: with open(f"{proc_dir}/cmdline", "r", errors="ignore") as cf: cmdline = cf.read().replace("\x00", " ").strip() except Exception: pass if cmdline: cmdline = sanitize_cmdline(cmdline) p_name = cmdline if cmdline else c["name"] # Classify process type short_name = c["name"].lower() if "nginx" in short_name: p_type = "Service" runtimes = ["Nginx"] elif "apache" in short_name or "httpd" in short_name: p_type = "Service" runtimes = ["Apache"] elif "php" in short_name: p_type = "Runtime" runtimes = ["PHP"] elif "mysql" in short_name or "mariadb" in short_name: p_type = "Database" runtimes = ["MySQL"] elif "postgres" in short_name: p_type = "Database" runtimes = ["PostgreSQL"] elif "redis" in short_name: p_type = "Database" runtimes = ["Redis"] elif "docker" in short_name: p_type = "Service" runtimes = ["Docker"] elif "fail2ban" in short_name: p_type = "Service" runtimes = ["Fail2Ban"] elif "cron" in short_name: p_type = "Service" runtimes = ["Cron"] elif "node" in short_name: p_type = "Runtime" runtimes = ["Node.js"] elif "python" in short_name: p_type = "Runtime" runtimes = ["Python"] else: p_type = "System" runtimes = [] # Calculate process Disk IO rate disk_io_rate = 0.0 if _last_proc_stats and pid in _last_proc_stats: last = _last_proc_stats[pid] time_diff = now - last["timestamp"] if time_diff > 0 and "read_io_bytes" in last: io_diff = (read_io_bytes - last["read_io_bytes"]) + (write_io_bytes - last["write_io_bytes"]) disk_io_rate = max(0.0, io_diff / time_diff) current_proc_stats[pid] = { "name": p_name, "type": p_type, "runtimes": runtimes, "user": uid_map.get(uid, str(uid)), "cpu_time": c["cpu_time"], "ram_bytes": c["ram_bytes"], "start_time": c["start_time"], "read_io_bytes": read_io_bytes, "write_io_bytes": write_io_bytes, "timestamp": now, "cpu_percent": c["cpu_percent"], "disk_io_rate": disk_io_rate, } # Build next proc stats tracking memory (needs all candidates to avoid missing CPU spikes) next_proc_stats = {} for c in candidates: pid = c["pid"] if pid in current_proc_stats: next_proc_stats[pid] = { "cpu_time": c["cpu_time"], "timestamp": now, "read_io_bytes": current_proc_stats[pid]["read_io_bytes"], "write_io_bytes": current_proc_stats[pid]["write_io_bytes"], } else: next_proc_stats[pid] = { "cpu_time": c["cpu_time"], "timestamp": now, } # Handle initial run to populate stats if not _last_proc_stats and candidates: _last_proc_stats = next_proc_stats time.sleep(0.05) return collect_process() _last_proc_stats = next_proc_stats # Build final process list proc_list = [] for pid, s in current_proc_stats.items(): elapsed = now - s["start_time"] if elapsed < 60: started = "just now" elif elapsed < 3600: started = f"{int(elapsed // 60)}m ago" elif elapsed < 86400: started = f"{int(elapsed // 3600)}h ago" else: started = f"{int(elapsed // 86400)}d ago" proc_list.append({ "id": str(pid), "pid": str(pid), "name": s["name"], "type": s["type"], "user": s["user"], "cpu": f"{s['cpu_percent']}%", "cpu_percent": s["cpu_percent"], "ram": f"{round(s['ram_bytes'] / (1024 * 1024), 1)} MB", "ram_bytes": s["ram_bytes"], "running": True, "updated": "just now", "started": started, "diskIo": f"{round(s['disk_io_rate'] / (1024 * 1024), 2)} MB/s", "description": f"{s['type']} process running under {s['user']}", "runtimes": s["runtimes"], "domains": [], }) top_cpu = sorted(proc_list, key=lambda x: x["cpu_percent"], reverse=True)[:10] top_ram = sorted(proc_list, key=lambda x: x["ram_bytes"], reverse=True)[:10] except Exception as exc: log.debug("Process collector error: %s", exc) return { "top_cpu": top_cpu, "top_ram": top_ram, } def collect_services() -> dict[str, str]: """Collect health status of common systemd services.""" services = { "nginx": "nginx.service", "apache": "apache2.service", "php": "php-fpm.service", "mysql": "mysql.service", "redis": "redis-server.service", "docker": "docker.service", "fail2ban": "fail2ban.service", "cron": "cron.service", } alt_services = { "apache": ["httpd.service"], "mysql": ["mariadb.service", "mysqld.service"], "redis": ["redis.service"], "cron": ["crond.service"], } status_map = {name: "inactive" for name in services} output = run_readonly(["systemctl", "list-units", "--type=service", "--all", "--no-legend", "--no-pager"]) if output: active_units = {} for line in output.splitlines(): fields = line.split(maxsplit=4) if len(fields) >= 4 and fields[0].endswith(".service"): unit_name = fields[0] active_state = fields[2] active_units[unit_name] = active_state for key, unit in services.items(): if unit in active_units: status_map[key] = active_units[unit] else: found = False if key in alt_services: for alt_unit in alt_services[key]: if alt_unit in active_units: status_map[key] = active_units[alt_unit] found = True break if not found and key == "php": for active_unit, state in active_units.items(): if active_unit.startswith("php") and active_unit.endswith("-fpm.service"): status_map[key] = state found = True break return status_map def collect_docker() -> list[dict[str, Any]]: """Collect details of active Docker containers.""" global _docker_cache_data, _docker_cache_time # 5 minutes TTL ttl = 300.0 now = time.time() with _docker_cache_lock: if _docker_cache_data is not None and (now - _docker_cache_time) < ttl: return _docker_cache_data if not shutil.which("docker"): return [] try: containers_json = run_readonly(["docker", "ps", "-a", "--format", "{{.ID}}\t{{.Names}}\t{{.State}}"]) if not containers_json: return [] containers_list = [] ids = [] for line in containers_json.splitlines(): parts = line.strip().split("\t") if len(parts) >= 3: cid, name, state = parts[0], parts[1], parts[2] containers_list.append({ "id": cid, "name": name, "status": state, "restart_count": 0, "cpu_percent": "0.0%", "ram_usage": "0B", }) ids.append(cid) if ids: # Get restart counts inspect_cmd = ["docker", "inspect", "--format", "{{.Id}}\t{{.RestartCount}}"] + ids inspect_output = run_readonly(inspect_cmd) restart_counts = {} if inspect_output: for line in inspect_output.splitlines(): parts = line.strip().split("\t") if len(parts) == 2: full_id, rcount = parts[0], parts[1] restart_counts[full_id[:12]] = int(rcount) # Get CPU/RAM usage stats_cmd = ["docker", "stats", "--no-stream", "--format", "{{.ID}}\t{{.CPUPerc}}\t{{.MemUsage}}"] stats_output = run_readonly(stats_cmd) docker_stats = {} if stats_output: for line in stats_output.splitlines(): parts = line.strip().split("\t") if len(parts) == 3: cid, cpu, mem = parts[0], parts[1], parts[2] mem_used = mem.split("/")[0].strip() if "/" in mem else mem docker_stats[cid] = { "cpu": cpu, "ram": mem_used, } for c in containers_list: short_id = c["id"][:12] rcount = 0 for kid, val in restart_counts.items(): if kid.startswith(short_id) or short_id.startswith(kid): rcount = val break c["restart_count"] = rcount if short_id in docker_stats: c["cpu_percent"] = docker_stats[short_id]["cpu"] c["ram_usage"] = docker_stats[short_id]["ram"] with _docker_cache_lock: _docker_cache_data = containers_list _docker_cache_time = now return containers_list except Exception as exc: log.debug("Docker collector failed, falling back to cache: %s", exc) with _docker_cache_lock: if _docker_cache_data is not None: return _docker_cache_data return [] def collect_ssl() -> list[dict[str, Any]]: """Collect SSL certificate expire dates, issuers and auto-renew status.""" global _ssl_cache_data, _ssl_cache_time # 15 minutes TTL ttl = 900.0 now = time.time() with _ssl_cache_lock: if _ssl_cache_data is not None and (now - _ssl_cache_time) < ttl: return _ssl_cache_data from datetime import datetime cert_dirs = ("/etc/letsencrypt/live", "/etc/ssl/certs") certs = [] try: found_paths = [] for directory in cert_dirs: try: base = Path(directory) if not base.is_dir(): continue if "letsencrypt" in directory: for p in base.glob("**/fullchain*"): if p.is_file(): found_paths.append(p) else: for p in list(base.glob("**/*.pem"))[:20]: if p.is_file() and "privkey" not in p.name and "key" not in p.name.lower(): found_paths.append(p) except (PermissionError, OSError) as exc: log.debug("SSL path read permission denied for %s: %s", directory, exc) continue for cert in found_paths[:30]: try: output = run_readonly(["openssl", "x509", "-noout", "-issuer", "-enddate", "-in", str(cert)]) if output: issuer = "unknown" enddate = None for line in output.splitlines(): if line.startswith("issuer="): raw_issuer = line.split("=", 1)[-1].strip() m = re.search(r"\bO\s*=\s*([^,]+)", raw_issuer) if m: issuer = m.group(1).strip() else: m_cn = re.search(r"\bCN\s*=\s*([^,]+)", raw_issuer) if m_cn: issuer = m_cn.group(1).strip() else: issuer = raw_issuer elif line.startswith("notAfter="): enddate_str = line.split("=", 1)[-1].strip() try: dt = datetime.strptime(enddate_str, "%b %d %H:%M:%S %Y %Z") enddate = dt.isoformat() + "Z" except Exception: enddate = enddate_str auto_renew = False provider = "letsencrypt" if "letsencrypt" in str(cert) else "custom" cert_name = None if provider == "letsencrypt": m = re.search(r"/etc/letsencrypt/live/([^/]+)", str(cert).replace("\\", "/")) if m: cert_name = m.group(1) renewal_conf = Path(f"/etc/letsencrypt/renewal/{cert_name}.conf") if renewal_conf.is_file(): auto_renew = True certs.append({ "path": str(cert), "expire": enddate, "issuer": issuer, "auto_renew": auto_renew, }) except (PermissionError, OSError) as exc: log.debug("Failed to read SSL cert %s: %s", cert, exc) continue with _ssl_cache_lock: _ssl_cache_data = certs _ssl_cache_time = now return certs except Exception as exc: log.debug("SSL collector failed, falling back to cache: %s", exc) with _ssl_cache_lock: if _ssl_cache_data is not None: return _ssl_cache_data return [] def collect_disk_io() -> float: """Collect global Disk IO rate in MB/s.""" global _last_disk_stats now = time.time() total_bytes = 0 try: if os.path.exists("/proc/diskstats"): with open("/proc/diskstats", "r", encoding="utf-8") as f: for line in f: parts = line.strip().split() if len(parts) >= 10: dev = parts[2] is_physical_dev = ( (dev.startswith("sd") and not dev[-1].isdigit()) or (dev.startswith("vd") and not dev[-1].isdigit()) or (dev.startswith("nvme") and "p" not in dev) ) if is_physical_dev: sectors_read = int(parts[5]) sectors_written = int(parts[9]) total_bytes += (sectors_read + sectors_written) * 512 disk_io_rate = 0.0 if _last_disk_stats is not None: last_bytes, last_time = _last_disk_stats time_diff = now - last_time if time_diff > 0: disk_io_rate = max(0.0, (total_bytes - last_bytes) / time_diff) # First run init if _last_disk_stats is None: _last_disk_stats = (total_bytes, now) time.sleep(0.05) return collect_disk_io() _last_disk_stats = (total_bytes, now) return round(disk_io_rate / (1024 * 1024), 2) except Exception as exc: log.debug("Disk IO collector error: %s", exc) return 0.0 # --- Executor & Caching ---------------------------------------------------- def collect_all(timeout: float = 4.0) -> dict[str, Any]: """Execute all collectors concurrently using a ThreadPoolExecutor with global timeout. Individually catches collector exceptions so a failure does not block/affect others. """ collectors = { "cpu": collect_cpu, "ram": collect_ram, "disk": collect_disk, "network": collect_network, "process": collect_process, "services": collect_services, "docker": collect_docker, "ssl": collect_ssl, "disk_io": collect_disk_io, } results: dict[str, Any] = {} with ThreadPoolExecutor(max_workers=len(collectors)) as executor: futures = {executor.submit(func): name for name, func in collectors.items()} try: for future in as_completed(futures, timeout=timeout): name = futures[future] try: results[name] = future.result() except Exception as exc: log.warning("Collector %s failed with exception: %s", name, exc) results[name] = None except TimeoutError: log.warning("Metrics collection exceeded global timeout of %s seconds", timeout) # Fill in any missing or pending results for future, name in futures.items(): if name not in results: if not future.done(): log.warning("Collector %s timed out", name) results[name] = None else: try: results[name] = future.result() except Exception as exc: log.warning("Collector %s failed late: %s", name, exc) results[name] = None # Get system uptime uptime = 0.0 try: if os.path.exists("/proc/uptime"): with open("/proc/uptime", "r", encoding="utf-8") as f: uptime = float(f.readline().split()[0]) except Exception: pass results["uptime"] = int(uptime) return results def _update_loop(interval: int) -> None: """Daemon thread loop that keeps the cache updated.""" log.info("Metrics cache updater thread started (interval=%ss)", interval) while not _stop_event.is_set(): try: start_time = time.time() data = collect_all(timeout=min(interval - 1, 4.0)) with _cache_lock: global _metrics_cache _metrics_cache = data elapsed = time.time() - start_time sleep_time = max(0.1, interval - elapsed) # Sleep in small chunks to detect stop signal faster for _ in range(int(sleep_time * 2)): if _stop_event.is_set(): break time.sleep(0.5) except Exception as exc: log.error("Error in metrics updater loop: %s", exc) time.sleep(5.0) def start_updater(interval: int = 15) -> None: """Start the background thread updater if it isn't running.""" global _updater_thread, _stop_event with _cache_lock: if _updater_thread is not None and _updater_thread.is_alive(): return _stop_event.clear() _updater_thread = threading.Thread( target=_update_loop, args=(interval,), name="WolfMetricsUpdater", daemon=True, ) _updater_thread.start() def stop_updater() -> None: """Stop the background thread updater.""" global _updater_thread, _stop_event _stop_event.set() if _updater_thread is not None: _updater_thread.join(timeout=2.0) _updater_thread = None def get_metrics() -> dict[str, Any]: """Retrieve metrics from the cache, or collect them synchronously if empty.""" with _cache_lock: if _metrics_cache: return _metrics_cache.copy() # Cache is empty (e.g. CLI run); collect synchronously log.debug("Metrics cache empty; performing synchronous collection") return collect_all(timeout=5.0)