diff --git a/5-Applications/scripts/swarm_network_capacity.py b/5-Applications/scripts/swarm_network_capacity.py index 11825d74..0205236a 100644 --- a/5-Applications/scripts/swarm_network_capacity.py +++ b/5-Applications/scripts/swarm_network_capacity.py @@ -1,25 +1,27 @@ #!/usr/bin/env python3 -""" -swarm_network_capacity.py — Network Resource Capacity Monitor +"""swarm_network_capacity.py — Network Resource Capacity Monitor (legacy shim). -Checks all Tailscale-connected nodes (via ENE mesh) to determine: -- Total available resources across the network -- Currently utilized capacity -- Idle resources that could be leveraged -- Node health and connectivity status +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. """ -import subprocess +from __future__ import annotations + +import argparse import json import re +import subprocess from dataclasses import dataclass -from typing import List, Dict, Optional from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional @dataclass class TailscaleNode: - """Remote node in the Tailscale mesh.""" ip: str hostname: str owner: str @@ -27,14 +29,13 @@ class TailscaleNode: status: str # online, offline, idle last_seen: Optional[str] tags: List[str] - + def is_online(self) -> bool: - return self.status == "online" or self.status == "idle" + return self.status in {"online", "idle"} @dataclass class NodeResources: - """Resource capacity of a node.""" node_id: str cpu_cores: int memory_gb: float @@ -45,187 +46,172 @@ class NodeResources: 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: - """ - Monitor and report on full network resource capacity. - - Queries Tailscale mesh and ENE nodes to determine: - - Total available compute across all nodes - - Current utilization vs capacity - - Resource distribution - """ - - def __init__(self): + 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]: - """Discover all nodes in the Tailscale mesh.""" print("\n[1] Discovering Tailscale mesh nodes...") - try: result = subprocess.run( ["tailscale", "status"], capture_output=True, text=True, - timeout=10 + timeout=10, ) - - lines = result.stdout.strip().split('\n') - nodes = [] - + + lines = result.stdout.strip().split("\n") + nodes: List[TailscaleNode] = [] + for line in lines: if not line.strip(): continue - - # Parse tailscale status line - # Format: 100.x.x.x hostname owner@ os status + parts = line.split() if len(parts) >= 4: ip = parts[0] hostname = parts[1] owner = parts[2] os_type = parts[3] - - # Parse status (can be complex) - status_parts = ' '.join(parts[4:]) if len(parts) > 4 else "" - - # Determine status + 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" - - # Extract last seen if offline + last_seen = None if "last seen" in status_parts: - match = re.search(r'last seen ([^,]+)', status_parts) + match = re.search(r"last seen ([^,]+)", status_parts) if match: last_seen = match.group(1) - - # Extract tags - tags = [] + + tags: List[str] = [] if "tagged-devices" in line: tags.append("tagged-devices") - - node = TailscaleNode( - ip=ip, - hostname=hostname, - owner=owner, - os=os_type, - status=status, - last_seen=last_seen, - tags=tags + + nodes.append( + TailscaleNode( + ip=ip, + hostname=hostname, + owner=owner, + os=os_type, + status=status, + last_seen=last_seen, + tags=tags, + ) ) - nodes.append(node) - + self.tailscale_nodes = nodes - - # Get local IP + try: ip_result = subprocess.run( ["tailscale", "ip", "-4"], capture_output=True, text=True, - timeout=5 + timeout=5, ) self.local_ip = ip_result.stdout.strip() - except: + 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 = "🟢" if node.is_online() else "🔴" - print(f" {status_icon} {node.hostname} ({node.ip}) - {node.status}") - + 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.""" - print("\n[2] Checking ENE deployment status...") - - ene_nodes = [] - - for node in self.tailscale_nodes: - if not node.is_online(): - continue - - # Try to check if ENE is running on remote node - # This would typically use SSH or ENE's gossip protocol - # For now, we'll simulate based on known deployment - - if node.hostname in ["architect", "judge", "qfox"]: - ene_nodes.append(node.hostname) - - self.ene_nodes = ene_nodes - - print(f" ENE deployed on: {len(ene_nodes)} nodes") - for node in ene_nodes: - print(f" ✅ {node}") - - # Nodes needing ENE deployment - online_nodes = [n.hostname for n in self.tailscale_nodes if n.is_online()] - need_ene = set(online_nodes) - set(ene_nodes) - if need_ene: - print(f" Need ENE deployment: {len(need_ene)} nodes") - for node in need_ene: - print(f" ⚠️ {node}") - - return ene_nodes - + """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 resources available on a node.""" - # These would normally be queried from the node - # For now, estimate based on hostname patterns - - resource_map = { - "qfox": {"cpu": 16, "ram": 32, "storage": 1000, "gpu": 1, "bw": 1000}, - "architect": {"cpu": 8, "ram": 16, "storage": 500, "gpu": 0, "bw": 500}, - "judge": {"cpu": 4, "ram": 8, "storage": 200, "gpu": 0, "bw": 500}, - "ip-172-31-25-81": {"cpu": 2, "ram": 4, "storage": 100, "gpu": 0, "bw": 1000}, # AWS - "netcup-router": {"cpu": 4, "ram": 8, "storage": 500, "gpu": 0, "bw": 1000}, - "racknerd-510bd9c": {"cpu": 2, "ram": 4, "storage": 100, "gpu": 0, "bw": 1000}, - "racknerd-atl": {"cpu": 2, "ram": 4, "storage": 100, "gpu": 0, "bw": 1000}, - "desktop-0u2ceal": {"cpu": 8, "ram": 16, "storage": 500, "gpu": 1, "bw": 100}, - } - - specs = resource_map.get(node.hostname, {"cpu": 2, "ram": 4, "storage": 100, "gpu": 0, "bw": 100}) - + """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=specs["cpu"], - memory_gb=specs["ram"], - storage_gb=specs["storage"], - bandwidth_mbps=specs["bw"], - gpu_count=specs["gpu"], + 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 # Would need actual monitoring + utilization_percent=0.0, ) - + def calculate_total_capacity(self) -> Dict[str, float]: - """Calculate total network capacity.""" 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 @@ -233,182 +219,104 @@ class SwarmNetworkCapacity: total_storage += resources.storage_gb total_gpu += resources.gpu_count total_bw += resources.bandwidth_mbps - - capacity = { - "cpu_cores": total_cpu, + + capacity: Dict[str, float] = { + "cpu_cores": float(total_cpu), "memory_gb": total_ram, "storage_gb": total_storage, - "gpu_count": total_gpu, + "gpu_count": float(total_gpu), "bandwidth_mbps": total_bw, - "online_nodes": sum(1 for n in self.tailscale_nodes if n.is_online()), - "total_nodes": len(self.tailscale_nodes) + "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: {total_cpu}") + + print(f" CPU Cores: {int(total_cpu)}") print(f" Memory: {total_ram:.1f} GB") print(f" Storage: {total_storage:.1f} GB") - print(f" GPUs: {total_gpu}") + print(f" GPUs: {int(total_gpu)}") print(f" Bandwidth: {total_bw:.0f} Mbps") - + return capacity - + def check_current_utilization(self) -> Dict[str, float]: - """Check current resource utilization.""" print("\n[4] Checking current utilization (local node only)...") - + try: - # Get local CPU/memory - result = subprocess.run( - ["cat", "/proc/loadavg"], - capture_output=True, - text=True, - timeout=5 - ) + 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 - - # Estimate CPU utilization from load - # This is simplified - would need per-node queries - cpu_util = min(100.0, (load_1min / 16) * 100) # Assuming 16 cores - - # Memory - mem_result = subprocess.run( - ["free", "-m"], - capture_output=True, - text=True, - timeout=5 - ) - - mem_lines = mem_result.stdout.split('\n') + + 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:'): + 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 - - utilization = { - "cpu_percent": cpu_util, - "memory_percent": mem_util, - "local_node_only": True - } - + print(f" Local CPU: {cpu_util:.1f}%") print(f" Local Memory: {mem_util:.1f}%") - print(f" ⚠️ Remote utilization requires ENE monitoring") - - return utilization - - except Exception as e: - print(f" ⚠️ Error checking utilization: {e}") - return {"cpu_percent": 0, "memory_percent": 0, "local_node_only": True} - - def generate_capacity_report(self) -> Dict[str, any]: - """Generate full capacity report.""" + + 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) - - # Gather data + self.discover_tailscale_mesh() self.check_ene_deployment() capacity = self.calculate_total_capacity() utilization = self.check_current_utilization() - - # Calculate utilization vs capacity - report = { + + report: Dict[str, Any] = { "timestamp": datetime.now().isoformat(), + "inventory_path": str(self.inventory_path), "network": { "total_nodes": len(self.tailscale_nodes), - "online_nodes": capacity["online_nodes"], - "offline_nodes": len(self.tailscale_nodes) - capacity["online_nodes"], - "ene_deployed": len(self.ene_nodes), - "ene_coverage": len(self.ene_nodes) / capacity["online_nodes"] * 100 if capacity["online_nodes"] > 0 else 0 - }, - "capacity": { - "cpu_cores": capacity["cpu_cores"], - "memory_gb": capacity["memory_gb"], - "storage_gb": capacity["storage_gb"], - "gpu_count": capacity["gpu_count"], - "bandwidth_mbps": capacity["bandwidth_mbps"] + "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["cpu_percent"], - "memory_percent": utilization["memory_percent"], - "note": "Local node only - full mesh monitoring requires ENE deployment on all nodes" + "cpu_percent": utilization.get("cpu_percent", 0.0), + "memory_percent": utilization.get("memory_percent", 0.0), + "note": "Local only - full mesh monitoring requires probes", }, - "idle_resources": { - "cpu_cores_available": capacity["cpu_cores"] * (1 - utilization["cpu_percent"]/100), - "memory_gb_available": capacity["memory_gb"] * (1 - utilization["memory_percent"]/100), - "message": "Significant idle capacity available across mesh" - }, - "recommendations": [] + "recommendations": [ + "Add cpu/ram/storage/gpu/bw fields to nodes.yaml to remove conservative defaults.", + "Replace ENE deployment heuristic with SSH/probe checks.", + ], } - - # Add recommendations - ene_coverage = report["network"]["ene_coverage"] - if ene_coverage < 100: - report["recommendations"].append( - f"Deploy ENE to {capacity['online_nodes'] - len(self.ene_nodes)} remaining nodes for full mesh monitoring" - ) - - if utilization["cpu_percent"] < 50: - report["recommendations"].append( - "CPU utilization low - swarm can scale up workloads" - ) - - if utilization["memory_percent"] < 50: - report["recommendations"].append( - "Memory available - can distribute more tasks across mesh" - ) - - report["recommendations"].append( - "Consider load balancing across all online nodes via ENE" - ) - - # Print summary - print("\n" + "=" * 70) - print("SUMMARY") - print("=" * 70) - print(f"Network: {report['network']['online_nodes']}/{report['network']['total_nodes']} nodes online") - print(f"ENE Coverage: {ene_coverage:.1f}% ({report['network']['ene_deployed']} nodes)") - print(f"Total CPU: {report['capacity']['cpu_cores']} cores") - print(f"Total Memory: {report['capacity']['memory_gb']:.1f} GB") - print(f"Total Storage: {report['capacity']['storage_gb']:.1f} GB") - print(f"\nUtilization (local only):") - print(f" CPU: {report['utilization']['cpu_percent']:.1f}%") - print(f" Memory: {report['utilization']['memory_percent']:.1f}%") - print(f"\nRecommendations:") - for rec in report["recommendations"]: - print(f" • {rec}") - - print("\n" + "=" * 70) - + return report -def main(): - """Run network capacity check.""" - monitor = SwarmNetworkCapacity() +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() - - # Save report - import json - from pathlib import Path - - output_path = Path("/home/allaun/Documents/Research Stack/data/swarm_network_capacity.json") - output_path.parent.mkdir(parents=True, exist_ok=True) - with open(output_path, "w") as f: - json.dump(report, f, indent=2) - - print(f"Report saved: {output_path}") - - return report + print(json.dumps(report, indent=2)) + return 0 if __name__ == "__main__": - main() + raise SystemExit(main())