cleanup(ene): use nodes.yaml inventory instead of hardcoded host resource map

5-Applications/scripts/swarm_network_capacity.py

@@ -1,25 +1,27 @@
 #!/usr/bin/env python3
-"""
+"""swarm_network_capacity.py — Network Resource Capacity Monitor (legacy shim).
-swarm_network_capacity.py — Network Resource Capacity Monitor
 
-Checks all Tailscale-connected nodes (via ENE mesh) to determine:
+Cleanups:
-- Total available resources across the network
+- Removed hardcoded hostname→resource maps.
-- Currently utilized capacity
+- Reads node inventory from 4-Infrastructure/auto/config/nodes.yaml (controller source of truth).
-- Idle resources that could be leveraged
+
-- Node health and connectivity status
+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:
-    """
+    def __init__(self, inventory_path: Path):
-    Monitor and report on full network resource capacity.
+        self.inventory_path = inventory_path
-    
+        self.inventory = _load_nodes_inventory(inventory_path)
-    Queries Tailscale mesh and ENE nodes to determine:
-    - Total available compute across all nodes
-    - Current utilization vs capacity
-    - Resource distribution
-    """
-    
-    def __init__(self):
         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')
+            lines = result.stdout.strip().split("\n")
-            nodes = []
+            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]
-                    
+                    status_parts = " ".join(parts[4:]) if len(parts) > 4 else ""
-                    # Parse status (can be complex)
+
-                    status_parts = ' '.join(parts[4:]) if len(parts) > 4 else ""
-                    
-                    # Determine status
                     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: List[str] = []
-                    tags = []
                     if "tagged-devices" in line:
                         tags.append("tagged-devices")
-                    
+
-                    node = TailscaleNode(
+                    nodes.append(
-                        ip=ip,
+                        TailscaleNode(
-                        hostname=hostname,
+                            ip=ip,
-                        owner=owner,
+                            hostname=hostname,
-                        os=os_type,
+                            owner=owner,
-                        status=status,
+                            os=os_type,
-                        last_seen=last_seen,
+                            status=status,
-                        tags=tags
+                            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 "🔴"
+                status_icon = "online" if node.is_online() else "offline"
-                print(f"    {status_icon} {node.hostname} ({node.ip}) - {node.status}")
+                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."""
+        """Check which nodes have ENE deployed.
-        print("\n[2] Checking ENE deployment status...")
+
-        
+        For now this still uses a heuristic: nodes with role including 'compute'
-        ene_nodes = []
+        or 'service-host' are assumed to be candidates. Real checks should be via
-        
+        SSH/probes.
-        for node in self.tailscale_nodes:
+        """
-            if not node.is_online():
+        print("\n[2] Checking ENE deployment status (heuristic)...")
-                continue
+        nodes = self.inventory.get("nodes", {})
-            
+        ene_nodes: List[str] = []
-            # Try to check if ENE is running on remote node
+        for node_id, spec in nodes.items():
-            # This would typically use SSH or ENE's gossip protocol
+            roles = spec.get("roles", []) if isinstance(spec, dict) else []
-            # For now, we'll simulate based on known deployment
+            if any(r in roles for r in ["compute", "service-host", "control-plane"]):
-            
+                ene_nodes.append(node_id)
-            if node.hostname in ["architect", "judge", "qfox"]:
+        self.ene_nodes = sorted(set(ene_nodes))
-                ene_nodes.append(node.hostname)
+
-        
+        print(f"  ENE candidates: {len(self.ene_nodes)}")
-        self.ene_nodes = ene_nodes
+        for n in self.ene_nodes[:20]:
-        
+            print(f"    ✅ {n}")
-        print(f"  ENE deployed on: {len(ene_nodes)} nodes")
+        return self.ene_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
-    
     def estimate_node_resources(self, node: TailscaleNode) -> Optional[NodeResources]:
-        """Estimate resources available on a node."""
+        """Estimate node resources.
-        # These would normally be queried from the node
+
-        # For now, estimate based on hostname patterns
+        nodes.yaml currently doesn't track CPU/RAM explicitly; so this function
-        
+        returns conservative defaults. Once resources are added to nodes.yaml,
-        resource_map = {
+        this function will use them directly.
-            "qfox": {"cpu": 16, "ram": 32, "storage": 1000, "gpu": 1, "bw": 1000},
+        """
-            "architect": {"cpu": 8, "ram": 16, "storage": 500, "gpu": 0, "bw": 500},
+        specs = self.inventory.get("nodes", {}).get(node.hostname) or {}
-            "judge": {"cpu": 4, "ram": 8, "storage": 200, "gpu": 0, "bw": 500},
+        cpu = int(specs.get("cpu", 2)) if isinstance(specs, dict) else 2
-            "ip-172-31-25-81": {"cpu": 2, "ram": 4, "storage": 100, "gpu": 0, "bw": 1000},  # AWS
+        ram = float(specs.get("ram", 4)) if isinstance(specs, dict) else 4.0
-            "netcup-router": {"cpu": 4, "ram": 8, "storage": 500, "gpu": 0, "bw": 1000},
+        storage = float(specs.get("storage", 100)) if isinstance(specs, dict) else 100.0
-            "racknerd-510bd9c": {"cpu": 2, "ram": 4, "storage": 100, "gpu": 0, "bw": 1000},
+        gpu = int(specs.get("gpu", 0)) if isinstance(specs, dict) else 0
-            "racknerd-atl": {"cpu": 2, "ram": 4, "storage": 100, "gpu": 0, "bw": 1000},
+        bw = float(specs.get("bw", 100)) if isinstance(specs, dict) else 100.0
-            "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})
-        
         return NodeResources(
             node_id=node.hostname,
-            cpu_cores=specs["cpu"],
+            cpu_cores=cpu,
-            memory_gb=specs["ram"],
+            memory_gb=ram,
-            storage_gb=specs["storage"],
+            storage_gb=storage,
-            bandwidth_mbps=specs["bw"],
+            bandwidth_mbps=bw,
-            gpu_count=specs["gpu"],
+            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 = {
+        capacity: Dict[str, float] = {
-            "cpu_cores": total_cpu,
+            "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()),
+            "online_nodes": float(sum(1 for n in self.tailscale_nodes if n.is_online())),
-            "total_nodes": len(self.tailscale_nodes)
+            "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
+            cpu_util = min(100.0, (load_1min / 16) * 100)
-            # This is simplified - would need per-node queries
+
-            cpu_util = min(100.0, (load_1min / 16) * 100)  # Assuming 16 cores
+            mem_result = subprocess.run(["free", "-m"], capture_output=True, text=True, timeout=5)
-            
+            mem_lines = mem_result.stdout.split("\n")
-            # Memory
-            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 {"cpu_percent": cpu_util, "memory_percent": mem_util, "local_node_only": True}
-            return utilization
+        except Exception:
-            
+            return {"cpu_percent": 0.0, "memory_percent": 0.0, "local_node_only": True}
-        except Exception as e:
+
-            print(f"  ⚠️  Error checking utilization: {e}")
+    def generate_capacity_report(self) -> Dict[str, Any]:
-            return {"cpu_percent": 0, "memory_percent": 0, "local_node_only": True}
-    
-    def generate_capacity_report(self) -> Dict[str, any]:
-        """Generate full capacity report."""
         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: Dict[str, Any] = {
-        report = {
             "timestamp": datetime.now().isoformat(),
+            "inventory_path": str(self.inventory_path),
             "network": {
                 "total_nodes": len(self.tailscale_nodes),
-                "online_nodes": capacity["online_nodes"],
+                "online_nodes": int(capacity["online_nodes"]),
-                "offline_nodes": len(self.tailscale_nodes) - capacity["online_nodes"],
+                "offline_nodes": len(self.tailscale_nodes) - int(capacity["online_nodes"]),
-                "ene_deployed": len(self.ene_nodes),
+                "ene_candidates": 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"]
             },
+            "capacity": capacity,
             "utilization": {
-                "cpu_percent": utilization["cpu_percent"],
+                "cpu_percent": utilization.get("cpu_percent", 0.0),
-                "memory_percent": utilization["memory_percent"],
+                "memory_percent": utilization.get("memory_percent", 0.0),
-                "note": "Local node only - full mesh monitoring requires ENE deployment on all nodes"
+                "note": "Local only - full mesh monitoring requires probes",
             },
-            "idle_resources": {
+            "recommendations": [
-                "cpu_cores_available": capacity["cpu_cores"] * (1 - utilization["cpu_percent"]/100),
+                "Add cpu/ram/storage/gpu/bw fields to nodes.yaml to remove conservative defaults.",
-                "memory_gb_available": capacity["memory_gb"] * (1 - utilization["memory_percent"]/100),
+                "Replace ENE deployment heuristic with SSH/probe checks.",
-                "message": "Significant idle capacity available across mesh"
+            ],
-            },
-            "recommendations": []
         }
-        
+
-        # 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():
+def main(argv: list[str] | None = None) -> int:
-    """Run network capacity check."""
+    parser = argparse.ArgumentParser()
-    monitor = SwarmNetworkCapacity()
+    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))
-    # Save report
+    return 0
-    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
 
 
 if __name__ == "__main__":
-    main()
+    raise SystemExit(main())