""" MotorSignal Operations Infrastructure Metrics Collector. Safe, read-only system resource monitoring, service status, network totals, and deployment health foundations. """ from dataclasses import dataclass from datetime import datetime, timezone import json import os import re import shutil import subprocess import time from typing import Any, Dict, List, Optional, Tuple from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen REVISION_HEX_REGEX = re.compile(r"^[0-9a-fA-F]{40}$") MAX_VERSION_RESPONSE_BYTES = 4096 DEFAULT_WEBSITE_INTERNAL_URL = "http://127.0.0.1:3001" # Fixed mapping from public logical service ID to actual systemd unit name SERVICE_SYSTEMD_UNITS: Dict[str, str] = { "motorsignal-website": "motorsignal-website", "motorsignal-api": "motorsignal-api", "nginx": "nginx", "postgresql": "postgresql@14-main", } ALLOWED_SYSTEMD_SERVICES: Tuple[str, ...] = tuple(SERVICE_SYSTEMD_UNITS.keys()) SERVICE_METADATA: Dict[str, Dict[str, str]] = { "motorsignal-website": { "display_name": "MotorSignal Website", "description": "Next.js frontend application runtime", }, "motorsignal-api": { "display_name": "MotorSignal API", "description": "FastAPI core backend service & ingress", }, "nginx": { "display_name": "Nginx Reverse Proxy", "description": "Edge TLS termination & reverse proxy routing", }, "postgresql": { "display_name": "PostgreSQL Database", "description": "Primary relational database persistence", }, } # In-memory previous CPU stat cache for non-blocking delta calculation _prev_cpu_stat: Optional[Tuple[float, int, int]] = None # (timestamp, idle_ticks, total_ticks) @dataclass class OpsDeploymentEvent: """Foundational data model for deployment events (Phase 2 readiness).""" id: str source_revision: str initiating_operator_id: str started_at: datetime completed_at: Optional[datetime] status: str # 'pending' | 'in_progress' | 'success' | 'failed' | 'rolled_back' release_identifier: Optional[str] = None failure_category: Optional[str] = None previous_revision: Optional[str] = None resulting_revision: Optional[str] = None def _read_proc_file(path: str) -> Optional[str]: """Safely read a /proc file if it exists.""" try: if os.path.exists(path): with open(path, "r", encoding="utf-8") as f: return f.read() except Exception: pass return None def get_cpu_percent() -> Optional[float]: """ Calculate overall CPU percent. Uses /proc/stat deltas on Linux. Degrades safely to None if unavailable. """ global _prev_cpu_stat stat_content = _read_proc_file("/proc/stat") if stat_content: for line in stat_content.splitlines(): if line.startswith("cpu "): parts = line.split() if len(parts) >= 5: try: # parts: cpu, user, nice, system, idle, iowait, irq, softirq, steal, guest, guest_nice fields = [int(p) for p in parts[1:]] idle = fields[3] + (fields[4] if len(fields) > 4 else 0) total = sum(fields) now = time.time() if _prev_cpu_stat is not None: prev_time, prev_idle, prev_total = _prev_cpu_stat delta_time = now - prev_time delta_total = total - prev_total delta_idle = idle - prev_idle if 0.1 <= delta_time <= 60.0 and delta_total > 0: _prev_cpu_stat = (now, idle, total) used_pct = ((delta_total - delta_idle) / delta_total) * 100.0 return round(max(0.0, min(100.0, used_pct)), 1) # Sample quick delta if no previous sample exists time.sleep(0.05) stat_content_2 = _read_proc_file("/proc/stat") if stat_content_2: for line2 in stat_content_2.splitlines(): if line2.startswith("cpu "): fields2 = [int(p) for p in line2.split()[1:]] idle2 = fields2[3] + (fields2[4] if len(fields2) > 4 else 0) total2 = sum(fields2) now2 = time.time() _prev_cpu_stat = (now2, idle2, total2) delta_total = total2 - total delta_idle = idle2 - idle if delta_total > 0: used_pct = ((delta_total - delta_idle) / delta_total) * 100.0 return round(max(0.0, min(100.0, used_pct)), 1) _prev_cpu_stat = (now, idle, total) except Exception: pass break return None def get_load_averages() -> Optional[Tuple[float, float, float]]: """Return 1m, 5m, 15m load averages via os.getloadavg().""" try: if hasattr(os, "getloadavg"): avg = os.getloadavg() return (round(avg[0], 2), round(avg[1], 2), round(avg[2], 2)) except Exception: pass return None def get_memory_metrics() -> Optional[Dict[str, Any]]: """ Parse /proc/meminfo for RAM metrics. Returns total, used, available in bytes, and used percent. """ mem_content = _read_proc_file("/proc/meminfo") if mem_content: mem_info: Dict[str, int] = {} for line in mem_content.splitlines(): parts = line.split(":") if len(parts) == 2: key = parts[0].strip() val_parts = parts[1].strip().split() if val_parts and val_parts[0].isdigit(): mem_info[key] = int(val_parts[0]) * 1024 # kB to bytes if "MemTotal" in mem_info: total = mem_info["MemTotal"] avail = mem_info.get( "MemAvailable", mem_info.get("MemFree", 0) + mem_info.get("Buffers", 0) + mem_info.get("Cached", 0) ) used = max(0, total - avail) pct = round((used / total) * 100.0, 1) if total > 0 else 0.0 return { "total": total, "used": used, "available": avail, "percent": pct, } return None def get_swap_metrics() -> Optional[Dict[str, Any]]: """ Parse /proc/meminfo for Swap metrics. Returns total, used in bytes, and used percent. """ mem_content = _read_proc_file("/proc/meminfo") if mem_content: mem_info: Dict[str, int] = {} for line in mem_content.splitlines(): parts = line.split(":") if len(parts) == 2: key = parts[0].strip() val_parts = parts[1].strip().split() if val_parts and val_parts[0].isdigit(): mem_info[key] = int(val_parts[0]) * 1024 if "SwapTotal" in mem_info: total = mem_info["SwapTotal"] free = mem_info.get("SwapFree", 0) used = max(0, total - free) pct = round((used / total) * 100.0, 1) if total > 0 else 0.0 return { "total": total, "used": used, "percent": pct, } return None def get_disk_metrics(mount_point: str = "/") -> Optional[Dict[str, Any]]: """ Calculate disk usage for the root filesystem. Includes warning (>= 80%) and critical (>= 90%) threshold flags. """ try: # On Windows during local testing, "/" maps to current drive, but check valid path path = mount_point if os.name == "nt" and mount_point == "/": path = os.environ.get("SystemDrive", "C:") + "\\" total, used, free = shutil.disk_usage(path) pct = round((used / total) * 100.0, 1) if total > 0 else 0.0 return { "total": total, "used": used, "available": free, "percent": pct, "warning": pct >= 80.0, "critical": pct >= 90.0, } except Exception: return None def get_uptime() -> Tuple[Optional[int], Optional[str]]: """ Parse /proc/uptime for server uptime. Returns (uptime_seconds, uptime_formatted). """ uptime_content = _read_proc_file("/proc/uptime") if uptime_content: parts = uptime_content.split() if parts: try: seconds = int(float(parts[0])) days = seconds // 86400 hours = (seconds % 86400) // 3600 minutes = (seconds % 3600) // 60 formatted = f"{days}d {hours}h {minutes}m" return seconds, formatted except Exception: pass return None, None def get_system_resources() -> Dict[str, Any]: """Collect overall server system resource metrics.""" uptime_sec, uptime_fmt = get_uptime() return { "cpu_percent": get_cpu_percent(), "load_avg": get_load_averages(), "memory": get_memory_metrics(), "swap": get_swap_metrics(), "disk": get_disk_metrics("/"), "uptime_seconds": uptime_sec, "uptime_formatted": uptime_fmt, "collected_at": datetime.now(timezone.utc).isoformat(), } def get_service_status(service_name: str) -> str: """ Query systemctl is-active for an allowed logical service name. Maps public logical service ID to actual fixed systemd unit (e.g. postgresql -> postgresql@14-main). Strictly forbids arbitrary service names, direct unit overrides, and shell interpolation. """ unit_name = SERVICE_SYSTEMD_UNITS.get(service_name) if not unit_name: return "unknown" try: res = subprocess.run( ["systemctl", "is-active", unit_name], capture_output=True, text=True, timeout=2.0, shell=False, ) out = res.stdout.strip().lower() if out == "active": return "active" if out == "inactive": return "inactive" if out == "failed": return "failed" return "unknown" except Exception: return "unknown" def get_services_health() -> List[Dict[str, Any]]: """Inspect all allowed core services.""" results = [] for svc in ALLOWED_SYSTEMD_SERVICES: meta = SERVICE_METADATA.get(svc, {}) status = get_service_status(svc) results.append({ "name": svc, "display_name": meta.get("display_name", svc), "description": meta.get("description", ""), "status": status, }) return results def get_network_totals() -> Dict[str, Any]: """ Parse /proc/net/dev to sum cumulative received and transmitted bytes across all non-loopback network interfaces since boot. """ dev_content = _read_proc_file("/proc/net/dev") rx_bytes = 0 tx_bytes = 0 found_any = False if dev_content: for line in dev_content.splitlines(): if ":" in line: parts = line.split(":", 1) iface = parts[0].strip() if iface == "lo": continue stats = parts[1].split() if len(stats) >= 9: try: rx = int(stats[0]) tx = int(stats[8]) rx_bytes += rx tx_bytes += tx found_any = True except Exception: pass return { "bytes_received": rx_bytes if found_any else None, "bytes_transmitted": tx_bytes if found_any else None, "label": "Cumulative Total Since Boot", "collected_at": datetime.now(timezone.utc).isoformat(), } def get_website_revision(website_url: Optional[str] = None) -> Optional[str]: """ Retrieve deployed website revision from the internal website /api/version endpoint. Parses JSON defensively with a strict response-size bound. Accepts revision only if it matches exactly 40 lowercase or uppercase hexadecimal characters. Never exposes malformed content or connection errors. Preserves graceful degradation if endpoint is absent, unreachable, malformed, or unmerged. Retains environment-variable fallback only for deliberately supported API-service configuration. """ base_url = website_url or os.environ.get("WEBSITE_INTERNAL_URL", DEFAULT_WEBSITE_INTERNAL_URL) endpoint = f"{base_url.rstrip('/')}/api/version" try: req = Request( endpoint, headers={ "User-Agent": "MotorSignal-Ops-Collector/1.0", "Accept": "application/json", }, ) with urlopen(req, timeout=1.5) as resp: status = getattr(resp, "status", getattr(resp, "code", None)) if status != 200: return None raw_bytes = resp.read(MAX_VERSION_RESPONSE_BYTES + 1) if len(raw_bytes) > MAX_VERSION_RESPONSE_BYTES: # Oversized response, reject defensively return None try: text = raw_bytes.decode("utf-8") data = json.loads(text) except Exception: # Malformed JSON or non-utf8, reject defensively return None if isinstance(data, dict): rev = data.get("revision") if isinstance(rev, str) and REVISION_HEX_REGEX.match(rev.strip()): return rev.strip() except Exception: # Never expose connection errors or malformed content pass # Deliberately supported API-service configuration fallback for env_var in ("DEPLOYED_REVISION", "GIT_SHA"): env_val = os.environ.get(env_var, "").strip() if env_val and REVISION_HEX_REGEX.match(env_val): return env_val return None def get_deployment_health( database_ok: bool = True, website_url: Optional[str] = None, ) -> Dict[str, Any]: """ Collect deployment runtime health and version foundations. """ target_website_url = website_url or os.environ.get("WEBSITE_INTERNAL_URL", DEFAULT_WEBSITE_INTERNAL_URL) # 1. Deployed Git Revision from internal website /api/version git_rev = get_website_revision(target_website_url) short_rev = git_rev[:7] if git_rev else None rev_status = "known" if git_rev else "unavailable" # 2. Local Website Health check website_health = "unknown" try: req = Request( f"{target_website_url.rstrip('/')}/api/health", headers={"User-Agent": "MotorSignal-Ops-Collector/1.0"}, ) with urlopen(req, timeout=1.5) as resp: status = getattr(resp, "status", getattr(resp, "code", None)) if status == 200: website_health = "available" elif status and 500 <= status < 600: website_health = "degraded" else: website_health = "available" except HTTPError as e: if 500 <= e.code < 600: website_health = "degraded" elif e.code == 404: # Route missing: fallback to probing root page try: req_root = Request( target_website_url.rstrip('/'), headers={"User-Agent": "MotorSignal-Ops-Collector/1.0"}, ) with urlopen(req_root, timeout=1.5) as resp_root: status_root = getattr(resp_root, "status", getattr(resp_root, "code", None)) if status_root in (200, 301, 302, 307, 308): website_health = "available" else: website_health = "degraded" except HTTPError as e_root: if 500 <= e_root.code < 600: website_health = "degraded" else: website_health = "unavailable" except Exception: website_health = "unavailable" else: website_health = "degraded" except (URLError, TimeoutError): website_health = "unavailable" except Exception: website_health = "unavailable" # If git revision was successfully fetched from website and website_health was unknown/unavailable, # mark website available (preserve degraded if explicitly 5xx) if git_rev and website_health in ("unknown", "unavailable"): website_health = "available" # 3. API Health api_health = "available" if database_ok else "degraded" return { "git_revision": git_rev, "short_revision": short_rev, "revision_status": rev_status, "website_health": website_health, "api_health": api_health, "history_status": "unavailable", "last_successful_deployment": None, } def get_infrastructure_metrics(database_ok: bool = True) -> Dict[str, Any]: """Aggregate all read-only infrastructure and deployment metrics.""" return { "resources": get_system_resources(), "services": get_services_health(), "network": get_network_totals(), "deployment": get_deployment_health(database_ok=database_ok), "collected_at": datetime.now(timezone.utc).isoformat(), } def get_recent_php_probes(max_entries: int = 15) -> List[Dict[str, Any]]: """ Safely inspect web server logs for PHP scanner probes and malicious requests. Degrades safely to empty list if logs are absent or unreadable. """ log_candidates = [ "/var/log/nginx/access.log", "/var/log/nginx/error.log", ] probes: List[Dict[str, Any]] = [] log_line_regex = re.compile( r'^(?P\S+)\s+\S+\s+\S+\s+\[(?P