mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
322 lines
11 KiB
Python
322 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""swarm_network_capacity.py — Network Resource Capacity Monitor (legacy shim).
|
|
|
|
Cleanups:
|
|
- Removed hardcoded hostname→resource maps.
|
|
- Reads node inventory from 4-Infrastructure/auto/config/nodes.yaml (controller source of truth).
|
|
|
|
NOTE: This is a legacy monitoring surface; treat results as advisory.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
@dataclass
|
|
class TailscaleNode:
|
|
ip: str
|
|
hostname: str
|
|
owner: str
|
|
os: str
|
|
status: str # online, offline, idle
|
|
last_seen: Optional[str]
|
|
tags: List[str]
|
|
|
|
def is_online(self) -> bool:
|
|
return self.status in {"online", "idle"}
|
|
|
|
|
|
@dataclass
|
|
class NodeResources:
|
|
node_id: str
|
|
cpu_cores: int
|
|
memory_gb: float
|
|
storage_gb: float
|
|
bandwidth_mbps: float
|
|
gpu_count: int
|
|
ene_enabled: bool
|
|
utilization_percent: float
|
|
|
|
|
|
def _load_nodes_inventory(path: Path) -> Dict[str, Any]:
|
|
"""Load nodes.yaml.
|
|
|
|
Tries PyYAML; if missing, raises a clear error.
|
|
"""
|
|
try:
|
|
import yaml # type: ignore
|
|
except ImportError as exc:
|
|
raise RuntimeError(
|
|
"nodes.yaml parsing requires PyYAML. Install with: pip install pyyaml"
|
|
) from exc
|
|
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
if not isinstance(data, dict) or "nodes" not in data:
|
|
raise ValueError(f"Invalid nodes inventory file: {path}")
|
|
return data
|
|
|
|
|
|
class SwarmNetworkCapacity:
|
|
def __init__(self, inventory_path: Path):
|
|
self.inventory_path = inventory_path
|
|
self.inventory = _load_nodes_inventory(inventory_path)
|
|
self.tailscale_nodes: List[TailscaleNode] = []
|
|
self.ene_nodes: List[str] = []
|
|
self.local_ip: Optional[str] = None
|
|
|
|
def discover_tailscale_mesh(self) -> List[TailscaleNode]:
|
|
print("\n[1] Discovering Tailscale mesh nodes...")
|
|
try:
|
|
result = subprocess.run(
|
|
["tailscale", "status"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
|
|
lines = result.stdout.strip().split("\n")
|
|
nodes: List[TailscaleNode] = []
|
|
|
|
for line in lines:
|
|
if not line.strip():
|
|
continue
|
|
|
|
parts = line.split()
|
|
if len(parts) >= 4:
|
|
ip = parts[0]
|
|
hostname = parts[1]
|
|
owner = parts[2]
|
|
os_type = parts[3]
|
|
status_parts = " ".join(parts[4:]) if len(parts) > 4 else ""
|
|
|
|
if "offline" in status_parts.lower():
|
|
status = "offline"
|
|
elif "idle" in status_parts.lower():
|
|
status = "idle"
|
|
else:
|
|
status = "online"
|
|
|
|
last_seen = None
|
|
if "last seen" in status_parts:
|
|
match = re.search(r"last seen ([^,]+)", status_parts)
|
|
if match:
|
|
last_seen = match.group(1)
|
|
|
|
tags: List[str] = []
|
|
if "tagged-devices" in line:
|
|
tags.append("tagged-devices")
|
|
|
|
nodes.append(
|
|
TailscaleNode(
|
|
ip=ip,
|
|
hostname=hostname,
|
|
owner=owner,
|
|
os=os_type,
|
|
status=status,
|
|
last_seen=last_seen,
|
|
tags=tags,
|
|
)
|
|
)
|
|
|
|
self.tailscale_nodes = nodes
|
|
|
|
try:
|
|
ip_result = subprocess.run(
|
|
["tailscale", "ip", "-4"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
)
|
|
self.local_ip = ip_result.stdout.strip()
|
|
except Exception:
|
|
pass
|
|
|
|
print(f" Found: {len(nodes)} Tailscale nodes")
|
|
online = sum(1 for n in nodes if n.is_online())
|
|
print(f" Online: {online}/{len(nodes)}")
|
|
|
|
for node in nodes:
|
|
status_icon = "online" if node.is_online() else "offline"
|
|
print(f" [{status_icon}] {node.hostname} ({node.ip})")
|
|
|
|
return nodes
|
|
|
|
except Exception as e:
|
|
print(f" ⚠️ Error discovering mesh: {e}")
|
|
return []
|
|
|
|
def check_ene_deployment(self) -> List[str]:
|
|
"""Check which nodes have ENE deployed.
|
|
|
|
For now this still uses a heuristic: nodes with role including 'compute'
|
|
or 'service-host' are assumed to be candidates. Real checks should be via
|
|
SSH/probes.
|
|
"""
|
|
print("\n[2] Checking ENE deployment status (heuristic)...")
|
|
nodes = self.inventory.get("nodes", {})
|
|
ene_nodes: List[str] = []
|
|
for node_id, spec in nodes.items():
|
|
roles = spec.get("roles", []) if isinstance(spec, dict) else []
|
|
if any(r in roles for r in ["compute", "service-host", "control-plane"]):
|
|
ene_nodes.append(node_id)
|
|
self.ene_nodes = sorted(set(ene_nodes))
|
|
|
|
print(f" ENE candidates: {len(self.ene_nodes)}")
|
|
for n in self.ene_nodes[:20]:
|
|
print(f" ✅ {n}")
|
|
return self.ene_nodes
|
|
|
|
def estimate_node_resources(self, node: TailscaleNode) -> Optional[NodeResources]:
|
|
"""Estimate node resources.
|
|
|
|
nodes.yaml currently doesn't track CPU/RAM explicitly; so this function
|
|
returns conservative defaults. Once resources are added to nodes.yaml,
|
|
this function will use them directly.
|
|
"""
|
|
specs = self.inventory.get("nodes", {}).get(node.hostname) or {}
|
|
cpu = int(specs.get("cpu", 2)) if isinstance(specs, dict) else 2
|
|
ram = float(specs.get("ram", 4)) if isinstance(specs, dict) else 4.0
|
|
storage = float(specs.get("storage", 100)) if isinstance(specs, dict) else 100.0
|
|
gpu = int(specs.get("gpu", 0)) if isinstance(specs, dict) else 0
|
|
bw = float(specs.get("bw", 100)) if isinstance(specs, dict) else 100.0
|
|
|
|
return NodeResources(
|
|
node_id=node.hostname,
|
|
cpu_cores=cpu,
|
|
memory_gb=ram,
|
|
storage_gb=storage,
|
|
bandwidth_mbps=bw,
|
|
gpu_count=gpu,
|
|
ene_enabled=node.hostname in self.ene_nodes,
|
|
utilization_percent=0.0,
|
|
)
|
|
|
|
def calculate_total_capacity(self) -> Dict[str, float]:
|
|
print("\n[3] Calculating total network capacity...")
|
|
|
|
total_cpu = 0
|
|
total_ram = 0.0
|
|
total_storage = 0.0
|
|
total_gpu = 0
|
|
total_bw = 0.0
|
|
|
|
for node in self.tailscale_nodes:
|
|
if not node.is_online():
|
|
continue
|
|
|
|
resources = self.estimate_node_resources(node)
|
|
if resources:
|
|
total_cpu += resources.cpu_cores
|
|
total_ram += resources.memory_gb
|
|
total_storage += resources.storage_gb
|
|
total_gpu += resources.gpu_count
|
|
total_bw += resources.bandwidth_mbps
|
|
|
|
capacity: Dict[str, float] = {
|
|
"cpu_cores": float(total_cpu),
|
|
"memory_gb": total_ram,
|
|
"storage_gb": total_storage,
|
|
"gpu_count": float(total_gpu),
|
|
"bandwidth_mbps": total_bw,
|
|
"online_nodes": float(sum(1 for n in self.tailscale_nodes if n.is_online())),
|
|
"total_nodes": float(len(self.tailscale_nodes)),
|
|
}
|
|
|
|
print(f" CPU Cores: {int(total_cpu)}")
|
|
print(f" Memory: {total_ram:.1f} GB")
|
|
print(f" Storage: {total_storage:.1f} GB")
|
|
print(f" GPUs: {int(total_gpu)}")
|
|
print(f" Bandwidth: {total_bw:.0f} Mbps")
|
|
|
|
return capacity
|
|
|
|
def check_current_utilization(self) -> Dict[str, float]:
|
|
print("\n[4] Checking current utilization (local node only)...")
|
|
|
|
try:
|
|
result = subprocess.run(["cat", "/proc/loadavg"], capture_output=True, text=True, timeout=5)
|
|
load_parts = result.stdout.strip().split()
|
|
load_1min = float(load_parts[0]) if load_parts else 0.0
|
|
|
|
cpu_util = min(100.0, (load_1min / 16) * 100)
|
|
|
|
mem_result = subprocess.run(["free", "-m"], capture_output=True, text=True, timeout=5)
|
|
mem_lines = mem_result.stdout.split("\n")
|
|
mem_used = 0
|
|
mem_total = 1
|
|
for line in mem_lines:
|
|
if line.startswith("Mem:"):
|
|
parts = line.split()
|
|
mem_total = int(parts[1])
|
|
mem_used = int(parts[2])
|
|
break
|
|
|
|
mem_util = (mem_used / mem_total) * 100 if mem_total > 0 else 0
|
|
|
|
print(f" Local CPU: {cpu_util:.1f}%")
|
|
print(f" Local Memory: {mem_util:.1f}%")
|
|
|
|
return {"cpu_percent": cpu_util, "memory_percent": mem_util, "local_node_only": True}
|
|
except Exception:
|
|
return {"cpu_percent": 0.0, "memory_percent": 0.0, "local_node_only": True}
|
|
|
|
def generate_capacity_report(self) -> Dict[str, Any]:
|
|
print("\n" + "=" * 70)
|
|
print("SWARM NETWORK CAPACITY REPORT")
|
|
print("=" * 70)
|
|
|
|
self.discover_tailscale_mesh()
|
|
self.check_ene_deployment()
|
|
capacity = self.calculate_total_capacity()
|
|
utilization = self.check_current_utilization()
|
|
|
|
report: Dict[str, Any] = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"inventory_path": str(self.inventory_path),
|
|
"network": {
|
|
"total_nodes": len(self.tailscale_nodes),
|
|
"online_nodes": int(capacity["online_nodes"]),
|
|
"offline_nodes": len(self.tailscale_nodes) - int(capacity["online_nodes"]),
|
|
"ene_candidates": len(self.ene_nodes),
|
|
},
|
|
"capacity": capacity,
|
|
"utilization": {
|
|
"cpu_percent": utilization.get("cpu_percent", 0.0),
|
|
"memory_percent": utilization.get("memory_percent", 0.0),
|
|
"note": "Local only - full mesh monitoring requires probes",
|
|
},
|
|
"recommendations": [
|
|
"Add cpu/ram/storage/gpu/bw fields to nodes.yaml to remove conservative defaults.",
|
|
"Replace ENE deployment heuristic with SSH/probe checks.",
|
|
],
|
|
}
|
|
|
|
return report
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(
|
|
"--inventory",
|
|
type=Path,
|
|
default=Path("4-Infrastructure/auto/config/nodes.yaml"),
|
|
help="Path to nodes.yaml",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
|
|
monitor = SwarmNetworkCapacity(args.inventory)
|
|
report = monitor.generate_capacity_report()
|
|
print(json.dumps(report, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|