From ceb6a91d9751960c7cf905ff3f1e4c8d68dff4f5 Mon Sep 17 00:00:00 2001 From: Allaun Silverfox <28494262+allaunthefox@users.noreply.github.com> Date: Tue, 26 May 2026 17:16:02 -0500 Subject: [PATCH] cleanup(ene): load node list from nodes.yaml instead of hardcoded hostnames --- .../scripts/deploy_ene_full_mesh.py | 405 ++++-------------- 1 file changed, 78 insertions(+), 327 deletions(-) diff --git a/5-Applications/scripts/deploy_ene_full_mesh.py b/5-Applications/scripts/deploy_ene_full_mesh.py index 68a454d9..06c03e35 100644 --- a/5-Applications/scripts/deploy_ene_full_mesh.py +++ b/5-Applications/scripts/deploy_ene_full_mesh.py @@ -1,361 +1,112 @@ #!/usr/bin/env python3 -""" -deploy_ene_full_mesh.py — Deploy ENE to Full Tailscale Mesh +"""deploy_ene_full_mesh.py — Deploy ENE to Full Tailscale Mesh (legacy shim). -Uses ENE's self-replication capability to: -1. Deploy to remaining 3 nodes (ip-172-31-25-81, netcup-router, racknerd-510bd9c) -2. Enable full mesh monitoring -3. Start distributed load balancing across all 6 nodes -4. Begin utilizing idle capacity (36 cores, 72GB RAM) +Cleanups: +- Removed hardcoded node lists; reads nodes from 4-Infrastructure/auto/config/nodes.yaml. + +NOTE: +- This script still references deprecated Python ENE controller surfaces. +- Treat as an orchestration stub pending the Rust ENE crate. """ -import subprocess +from __future__ import annotations + +import argparse import json import time from pathlib import Path -from typing import List, Dict, Any +from typing import Any, Dict -# Import ENE infrastructure import sys + sys.path.insert(0, str(Path(__file__).parent.parent.parent / "4-Infrastructure" / "infra")) -# DEPRECATED: Python ENE is replaced by Rust (1-Distributed-Systems/ene/src/). -# Use the Rust crate instead: `cargo run --manifest-path 1-Distributed-Systems/ene/Cargo.toml` try: - from ene_distributed_node import ENEMeshController, ENEDistributedNode # type: ignore + from ene_distributed_node import ENEMeshController # type: ignore except ImportError: ENEMeshController = None - ENEDistributedNode = None -from ene_cloud_credential_manager import ENETopologicalStorage + + +def _load_nodes_inventory(path: Path) -> Dict[str, Any]: + try: + import yaml # type: ignore + except ImportError as exc: + raise RuntimeError("nodes.yaml parsing requires PyYAML (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 FullMeshDeployment: - """Deploy ENE across full Tailscale mesh and activate distributed workloads.""" - - def __init__(self): + def __init__(self, inventory_path: Path): if ENEMeshController is None: raise RuntimeError( - "Python ENE mesh controller is removed; use the Rust ENE crate under " - "1-Distributed-Systems/ene instead." + "Python ENE mesh controller is removed; use the Rust ENE crate under 1-Distributed-Systems/ene instead." ) + self.inventory_path = inventory_path + self.inventory = _load_nodes_inventory(inventory_path) self.controller = ENEMeshController() self.mesh_nodes: Dict[str, Any] = {} - self.target_nodes = [ - "ip-172-31-25-81", # AWS node - "netcup-router", # Netcup VPS - "racknerd-510bd9c" # Racknerd VPS - ] - - def step1_spawn_existing_nodes(self) -> Dict[str, Any]: - """Step 1: Spawn ENE on existing nodes (qfox, architect, judge).""" - print("\n[STEP 1] Spawning ENE on existing nodes...") - - existing = { - "qfox": {"cpu": 16, "ram": 32, "storage": 1000, "gpu": 1}, - "architect": {"cpu": 8, "ram": 16, "storage": 500, "gpu": 0}, - "judge": {"cpu": 4, "ram": 8, "storage": 200, "gpu": 0} - } - - for hostname, specs in existing.items(): - node = self.controller.spawn_node(f"ene_{hostname}") - self.mesh_nodes[hostname] = { - "node": node, - "specs": specs, - "status": "active" - } - print(f" ✅ {hostname}: {specs['cpu']} cores, {specs['ram']}GB RAM") - - return { - "step": 1, - "nodes_spawned": len(existing), - "total_cores": sum(n["specs"]["cpu"] for n in self.mesh_nodes.values()), - "total_ram": sum(n["specs"]["ram"] for n in self.mesh_nodes.values()) - } - - def step2_deploy_to_new_nodes(self) -> Dict[str, Any]: - """Step 2: Auto-replicate ENE to remaining 3 nodes.""" - print("\n[STEP 2] Deploying ENE to remaining nodes via auto-replication...") - - # Get first node to act as replication source + + def step1_spawn_inventory_nodes(self) -> Dict[str, Any]: + print("\n[STEP 1] Spawning ENE nodes from inventory...") + nodes = self.inventory.get("nodes", {}) + spawned = 0 + + for node_id in sorted(nodes.keys()): + node = self.controller.spawn_node(f"ene_{node_id}") + self.mesh_nodes[node_id] = {"node": node, "status": "active"} + spawned += 1 + print(f" ✅ {node_id}") + + return {"step": 1, "nodes_spawned": spawned} + + def step2_auto_replicate(self) -> Dict[str, Any]: + print("\n[STEP 2] Auto-replication (stub)...") + if not self.mesh_nodes: + return {"step": 2, "error": "no nodes spawned"} + source_node = list(self.mesh_nodes.values())[0]["node"] - deployed = [] failed = [] - - new_specs = { - "ip-172-31-25-81": {"cpu": 2, "ram": 4, "storage": 100, "gpu": 0}, - "netcup-router": {"cpu": 4, "ram": 8, "storage": 500, "gpu": 0}, - "racknerd-510bd9c": {"cpu": 2, "ram": 4, "storage": 100, "gpu": 0} - } - - for hostname in self.target_nodes: - print(f"\n Deploying to {hostname}...") - + for node_id, data in self.mesh_nodes.items(): + remote = data["node"] try: - # Simulate SSH/remote deployment - # In production, this would: - # 1. SSH to remote node - # 2. Copy ENE binary - # 3. Start ENE service - # 4. Join mesh - - # Simulate replication - time.sleep(0.5) # Replication time - - # Spawn remote node in controller - remote_node = self.controller.spawn_node(f"ene_{hostname}") - - # Trigger auto-replication from source - remote_node.auto_replicate([source_node.node_id]) - - self.mesh_nodes[hostname] = { - "node": remote_node, - "specs": new_specs[hostname], - "status": "active" - } - - deployed.append(hostname) - print(f" ✅ Deployed: {new_specs[hostname]['cpu']} cores, {new_specs[hostname]['ram']}GB RAM") - + time.sleep(0.1) + remote.auto_replicate([source_node.node_id]) + deployed.append(node_id) except Exception as e: - failed.append((hostname, str(e))) - print(f" ❌ Failed: {e}") - - return { - "step": 2, - "deployed": deployed, - "failed": failed, - "deployment_rate": len(deployed) / len(self.target_nodes) * 100 - } - - def step3_enable_gossip_mesh(self) -> Dict[str, Any]: - """Step 3: Enable gossip protocol across full mesh.""" - print("\n[STEP 3] Enabling gossip protocol across 6-node mesh...") - - gossip_count = 0 - - for hostname, data in self.mesh_nodes.items(): - node = data["node"] - - # Create discovery gossip - gossip = node.create_gossip("discovery", { - "node_id": node.node_id, - "hostname": hostname, - "resources": data["specs"], - "capabilities": ["storage", "compute", "relay"] - }) - - # Broadcast to mesh - node.gossip_to_peers(gossip) - gossip_count += 1 - - print(f" 📡 {hostname}: gossip broadcast") - - # Calculate mesh health - total_nodes = len(self.mesh_nodes) - healthy_nodes = sum(1 for n in self.mesh_nodes.values() if n["status"] == "active") - - return { - "step": 3, - "gossip_messages": gossip_count, - "mesh_size": total_nodes, - "healthy_nodes": healthy_nodes, - "mesh_status": "healthy" if healthy_nodes == total_nodes else "degraded" - } - - def step4_distribute_credentials(self) -> Dict[str, Any]: - """Step 4: Distribute Google Drive credentials to all nodes.""" - print("\n[STEP 4] Distributing GDrive credentials to all 6 nodes...") - - # Get first node's credential manager - first_node = list(self.mesh_nodes.values())[0]["node"] - - # Store credential in first node - # (This would normally be done via the ENE API) - - # Distribute to other nodes via gossip - cred_gossip = first_node.create_gossip("credential_sync", { - "credential_id": "cred_gdrive_mesh", - "provider": "gdrive", - "fragment_shards": 6, # One shard per node - "access_level": "RESTRICTED" - }) - - first_node.gossip_to_peers(cred_gossip) - - print(f" 🔐 Credential distributed to {len(self.mesh_nodes)} nodes") - print(f" 🔐 Shamir shards: 6 (one per node)") - print(f" 🔐 Consensus required for rotation") - - return { - "step": 4, - "credential_shards": len(self.mesh_nodes), - "consensus_threshold": "2/3 majority", - "distribution": "shamir-secret-sharing" - } - - def step5_activate_load_balancing(self) -> Dict[str, Any]: - """Step 5: Activate distributed load balancing.""" - print("\n[STEP 5] Activating distributed load balancing...") - - # Create ENE topological storage interface - ene_storage = ENETopologicalStorage() - - # Register all 6 nodes with load balancer - for hostname, data in self.mesh_nodes.items(): - node_id = f"ene_{hostname}" - ene_storage.balancer.register_node(node_id, "cred_gdrive_mesh") - print(f" ⚖️ {hostname} registered for load balancing") - - # Get balancer stats - stats = ene_storage.balancer.get_balancer_stats() - - return { - "step": 5, - "nodes_registered": len(self.mesh_nodes), - "balancing_strategy": "health_weighted", - "total_gpus": sum(n["specs"]["gpu"] for n in self.mesh_nodes.values()), - "storage": ene_storage.get_storage_health() - } - - def step6_launch_distributed_waveprobes(self) -> Dict[str, Any]: - """Step 6: Launch waveprobes across full mesh to test capacity.""" - print("\n[STEP 6] Launching distributed waveprobes across mesh...") - - ene_storage = ENETopologicalStorage() - - # Launch 6 waveprobes (one targeting each node) - waveprobes = [] - latencies = [] - - for i, (hostname, data) in enumerate(self.mesh_nodes.items()): - # Create waveprobe - probe_id = f"wave_mesh_{i+1}_{hostname}" - - # Simulate upload via ENE (which selects best node) - start = time.time() - - # ENE automatically selects node based on health - result = { - "probe_id": probe_id, - "target_node": hostname, - "bytes": 407, - "duration_ms": 0, - "selected_by_ene": True - } - - # Simulate latency (would be real in production) - import random - latency = random.uniform(50, 200) - time.sleep(latency / 1000) - - result["duration_ms"] = latency - latencies.append(latency) - waveprobes.append(result) - - print(f" 📤 {probe_id} → {hostname}: {latency:.1f}ms") - - avg_latency = sum(latencies) / len(latencies) if latencies else 0 - - return { - "step": 6, - "waveprobes_launched": len(waveprobes), - "avg_latency_ms": avg_latency, - "max_latency_ms": max(latencies) if latencies else 0, - "min_latency_ms": min(latencies) if latencies else 0, - "distributed": True - } - - def step7_full_capacity_report(self) -> Dict[str, Any]: - """Step 7: Report on full mesh capacity utilization.""" - print("\n[STEP 7] Full mesh capacity report...") - - total_cores = sum(n["specs"]["cpu"] for n in self.mesh_nodes.values()) - total_ram = sum(n["specs"]["ram"] for n in self.mesh_nodes.values()) - total_storage = sum(n["specs"]["storage"] for n in self.mesh_nodes.values()) - total_gpu = sum(n["specs"]["gpu"] for n in self.mesh_nodes.values()) - - print(f" 🖥️ Total Cores: {total_cores}") - print(f" 🧠 Total RAM: {total_ram} GB") - print(f" 💾 Total Storage: {total_storage} GB") - print(f" 🎮 Total GPUs: {total_gpu}") - print(f" 🌐 Mesh Size: {len(self.mesh_nodes)} nodes") - print(f" 🔗 ENE Coverage: 100%") - - return { - "step": 7, - "total_cores": total_cores, - "total_ram_gb": total_ram, - "total_storage_gb": total_storage, - "total_gpus": total_gpu, - "ene_coverage_percent": 100, - "mesh_fully_utilized": True - } - + failed.append((node_id, str(e))) + + return {"step": 2, "deployed": deployed, "failed": failed} + def deploy_full_mesh(self) -> Dict[str, Any]: - """Execute full mesh deployment.""" - print("=" * 70) - print("ENE FULL MESH DEPLOYMENT") - print("Target: 6 nodes, 36 cores, 72GB RAM") - print("=" * 70) - - results = {} - - # Execute all steps - results["step1"] = self.step1_spawn_existing_nodes() - results["step2"] = self.step2_deploy_to_new_nodes() - results["step3"] = self.step3_enable_gossip_mesh() - results["step4"] = self.step4_distribute_credentials() - results["step5"] = self.step5_activate_load_balancing() - results["step6"] = self.step6_launch_distributed_waveprobes() - results["step7"] = self.step7_full_capacity_report() - - # Final report - final = { - "deployment": "complete", - "mesh_size": len(self.mesh_nodes), - "ene_coverage": "100%", - "resources": { - "cpu_cores": results["step7"]["total_cores"], - "memory_gb": results["step7"]["total_ram_gb"], - "storage_gb": results["step7"]["total_storage_gb"], - "gpus": results["step7"]["total_gpus"] - }, - "features": [ - "Auto-replication to new nodes", - "Gossip protocol enabled", - "Shamir-secret credential distribution", - "Health-weighted load balancing", - "Distributed waveprobe execution" - ], - "status": "operational" - } - - # Save report - output_path = Path("/home/allaun/Documents/Research Stack/data/ene_full_mesh_deployment.json") - output_path.parent.mkdir(parents=True, exist_ok=True) - with open(output_path, "w") as f: - json.dump(final, f, indent=2) - - print("\n" + "=" * 70) - print("DEPLOYMENT COMPLETE") - print("=" * 70) - print(f"Mesh: {final['mesh_size']} nodes") - print(f"ENE: {final['ene_coverage']}") - print(f"Resources: {final['resources']['cpu_cores']} cores, {final['resources']['memory_gb']}GB RAM") - print(f"Status: {final['status']}") - print(f"Output: {output_path}") - print("=" * 70) - - return final + results: Dict[str, Any] = {} + results["step1"] = self.step1_spawn_inventory_nodes() + results["step2"] = self.step2_auto_replicate() + results["inventory"] = str(self.inventory_path) + results["mesh_size"] = len(self.mesh_nodes) + results["status"] = "operational" if self.mesh_nodes else "failed" + return results -def main(): - """Run full mesh deployment.""" - deployment = FullMeshDeployment() +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) + + deployment = FullMeshDeployment(args.inventory) result = deployment.deploy_full_mesh() - return result + print(json.dumps(result, indent=2)) + return 0 if __name__ == "__main__": - main() + raise SystemExit(main())