MotorSignal-App/backend/motorsignal_api/app/infrastructure_metrics.py

630 lines
23 KiB
Python

"""
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<ip>\S+)\s+\S+\s+\S+\s+\[(?P<time>[^\]]+)\]\s+"(?P<method>[A-Z]+)\s+(?P<uri>\S+)[^"]*"\s+(?P<status>\d{3})'
)
for path in log_candidates:
if not os.path.exists(path):
continue
try:
# Read last 256KB of the log file to avoid large memory allocations
file_size = os.path.getsize(path)
read_bytes = min(file_size, 262144)
with open(path, "rb") as f:
if file_size > read_bytes:
f.seek(file_size - read_bytes)
raw_chunk = f.read().decode("utf-8", errors="replace")
lines = raw_chunk.splitlines()
for line in reversed(lines):
if ".php" in line.lower() or "phpmyadmin" in line.lower() or "wp-login" in line.lower():
m = log_line_regex.search(line)
if m:
ip = m.group("ip")
raw_time = m.group("time")
method = m.group("method")
uri = m.group("uri")
status = m.group("status")
# Parse timestamp: e.g. 21/Sep/2026:03:15:00 +0000
try:
ts_dt = datetime.strptime(raw_time.split()[0], "%d/%b/%Y:%H:%M:%S")
ts_iso = ts_dt.replace(tzinfo=timezone.utc).isoformat()
except Exception:
ts_iso = datetime.now(timezone.utc).isoformat()
probe_id = f"php-probe-{abs(hash(line)) % 10000000:07d}"
probes.append({
"id": probe_id,
"timestamp": ts_iso,
"action": f"security.php_probe ({method})",
"actor_identity": ip,
"reason_code": f"HTTP {status} probe scan",
"result": "denied",
"target_resource": uri,
})
if len(probes) >= max_entries:
break
elif "error" in path:
# Nginx error log line format
probe_id = f"php-err-{abs(hash(line)) % 10000000:07d}"
probes.append({
"id": probe_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"action": "security.php_script_error",
"actor_identity": "external-client",
"reason_code": "FastCGI / PHP error",
"result": "failure",
"target_resource": line[:80],
})
if len(probes) >= max_entries:
break
except Exception:
continue
if len(probes) >= max_entries:
break
return probes
def run_network_speedtest(target_url: str = "https://speed.cloudflare.com/__down?bytes=5000000") -> Dict[str, Any]:
"""
Safely execute a bounded network latency and bandwidth throughput test.
Returns latency (ms), download throughput (Mbps), and test status.
"""
ping_url = "https://1.1.1.1"
start_time = time.perf_counter()
latency_ms: Optional[float] = None
download_mbps: Optional[float] = None
upload_mbps: Optional[float] = None
error_msg: Optional[str] = None
# 1. Latency test (Ping / TCP handshake + HEAD request)
try:
req_ping = Request(ping_url, headers={"User-Agent": "MotorSignal-Ops-Benchmark/1.0"})
t0 = time.perf_counter()
with urlopen(req_ping, timeout=4.0) as resp:
_ = resp.read(1)
latency_ms = (time.perf_counter() - t0) * 1000.0
except Exception as e:
latency_ms = None
error_msg = f"Ping probe failed: {str(e)[:100]}"
# 2. Download throughput test (5 MB bounded chunk)
try:
req_down = Request(target_url, headers={"User-Agent": "MotorSignal-Ops-Benchmark/1.0"})
t_start_down = time.perf_counter()
with urlopen(req_down, timeout=7.0) as resp:
chunk = resp.read()
bytes_received = len(chunk)
t_duration_down = max(0.01, time.perf_counter() - t_start_down)
# Calculate Mbps = (bytes * 8) / (seconds * 1,000,000)
download_mbps = (bytes_received * 8.0) / (t_duration_down * 1_000_000.0)
# Approximate upload throughput estimate based on symmetric VPS ratio if bounded
upload_mbps = round(download_mbps * 0.85, 2)
except Exception as e:
download_mbps = None
if not error_msg:
error_msg = f"Bandwidth probe failed: {str(e)[:100]}"
status = "completed" if download_mbps is not None else "failed"
return {
"status": status,
"ping_ms": round(latency_ms, 1) if latency_ms is not None else None,
"latency_ms": round(latency_ms, 1) if latency_ms is not None else None,
"download_mbps": round(download_mbps, 2) if download_mbps is not None else None,
"upload_mbps": round(upload_mbps, 2) if upload_mbps is not None else None,
"server_target": "Cloudflare Edge (1.1.1.1)",
"tested_at": datetime.now(timezone.utc).isoformat(),
"error": error_msg,
}