# agent/metrics_collector.py
import psutil
import time
from datetime import datetime

class NodeMetrics:
    """Collect multi-dimensional metrics from a VPS node in real time."""
    
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.window_seconds = 60  # sliding window
        self.metrics_history = []
    
    def collect(self) -> dict:
        now = datetime.utcnow()
        
        # CPU usage (1-minute average)
        cpu_percent = psutil.cpu_percent(interval=1)
        cpu_cores = psutil.cpu_count()
        
        # Memory usage
        mem = psutil.virtual_memory()
        
        # Disk I/O
        disk_io = psutil.disk_io_counters()
        
        # Network I/O
        net_io = psutil.net_io_counters()
        
        # System load
        load_avg = psutil.getloadavg()
        
        # Process count
        processes = len(psutil.pids())
        
        return {
            "node_id": self.node_id,
            "timestamp": now.isoformat(),
            "cpu_percent": cpu_percent,
            "cpu_cores": cpu_cores,
            "memory_percent": mem.percent,
            "memory_available_mb": mem.available / 1024 / 1024,
            "disk_read_mb": disk_io.read_bytes / 1024 / 1024,
            "disk_write_mb": disk_io.write_bytes / 1024 / 1024,
            "net_sent_mb": net_io.bytes_sent / 1024 / 1024,
            "net_recv_mb": net_io.bytes_recv / 1024 / 1024,
            "load_1min": load_avg[0],
            "load_5min": load_avg[1],
            "load_15min": load_avg[2],
            "process_count": processes,
        }

# agent/health_score.py
class HealthScoreCalculator:
    """
    Node health score = f(cpu, memory, disk_io, latency, error_rate)
    Range: 0 (completely unavailable) ~ 100 (optimal state)
    """
    
    WEIGHTS = {
        "cpu": 0.20,
        "memory": 0.20,
        "disk_io": 0.15,
        "latency": 0.25,
        "error_rate": 0.20,
    }
    
    def calculate(self, metrics: dict, baseline: dict) -> float:
        """Calculate health score based on current metrics and baseline."""
        scores = {}
        
        # CPU health (lower is better)
        scores["cpu"] = max(0, 100 - metrics["cpu_percent"])
        
        # Memory health
        scores["memory"] = max(0, 100 - metrics["memory_percent"])
        
        # Disk I/O pressure
        io_pressure = (metrics["disk_read_mb"] + metrics["disk_write_mb"]) / max(baseline.get("avg_disk_io", 1), 1)
        scores["disk_io"] = max(0, 100 - min(io_pressure * 30, 100))
        
        # Latency health (reported by API Gateway)
        latency = metrics.get("avg_latency_ms", 50)
        scores["latency"] = max(0, 100 - (latency / 5))  # 250ms = 50 points
        
        # Error rate health
        error_rate = metrics.get("error_rate_percent", 0)
        scores["error_rate"] = max(0, 100 - error_rate * 10)
        
        # Weighted sum
        health_score = sum(scores[k] * self.WEIGHTS[k] for k in self.WEIGHTS)
        
        return round(health_score, 1)

# predictor/traffic_forecaster.py
from prophet import Prophet
import pandas as pd

class TrafficForecaster:
    """
    Forecast the next 30 minutes of traffic based on historical QPS data.
    """
    
    def __init__(self, window_hours=24):
        self.window_hours = window_hours
        self.models = {}  # Independent models per path/route
    
    def fit(self, df: pd.DataFrame, path: str = "default"):
        """Train the forecasting model."""
        model = Prophet(
            yearly_seasonality=True,
            weekly_seasonality=True,
            daily_seasonality=True,
            changepoint_prior_scale=0.05,
        )
        model.fit(df)
        self.models[path] = model
        return model
    
    def predict(self, path: str = "default", hours_ahead=0.5) -> dict:
        """Forecast future traffic."""
        if path not in self.models:
            return {"forecast": [], "confidence": {}}
        
        model = self.models[path]
        future = model.make_future_dataframe(periods=int(hours_ahead * 24 * 60))
        forecast = model.predict(future)
        
        recent = forecast.tail(int(hours_ahead * 12)).copy()
        
        return {
            "forecast": recent[["ds", "yhat", "yhat_lower", "yhat_upper"]].to_dict('records'),
            "predicted_peak_qps": float(recent["yhat"].max()),
            "predicted_avg_qps": float(recent["yhat"].mean()),
            "confidence_interval": {
                "lower": float(recent["yhat_lower"].min()),
                "upper": float(recent["yhat_upper"].max()),
            }
        }

# dispatcher/scheduler.py
import numpy as np
from scipy.optimize import linprog

class AIScheduler:
    """
    AI-driven intelligent scheduler.
    Computes optimal routing weights based on node health scores + predicted traffic.
    """
    
    def __init__(self, nodes: list, weights: dict = None):
        self.nodes = nodes
        self.weights = weights or {"latency": 0.4, "balance": 0.35, "cost": 0.25}
    
    def compute_weights(self, predicted_qps: float) -> dict:
        """Compute optimal weight for each VPS."""
        n = len(self.nodes)
        if n == 0:
            return {}
        
        # Latency cost: nodes with lower health scores have higher latency
        latency_cost = np.array([
            (100 - node["health_score"]) / 100 * node["current_latency_ms"]
            for node in self.nodes
        ])
        
        # Load balance cost: deviation from average load
        avg_load = np.mean([node["cpu_percent"] for node in self.nodes])
        balance_cost = np.array([
            abs(node["cpu_percent"] - avg_load) / 100
            for node in self.nodes
        ])
        
        # Cost per node
        cost_per_node = np.array([
            node.get("hourly_cost", 0.05) for node in self.nodes
        ])
        
        # Combined objective
        c = (
            self.weights["latency"] * latency_cost +
            self.weights["balance"] * balance_cost +
            self.weights["cost"] * cost_per_node
        )
        
        # Constraint: weights sum to 1
        A_eq = np.ones((1, n))
        b_eq = np.array([1.0])
        bounds = [(0, 1)] * n
        
        result = linprog(c, A_eq=A_eq, b_eq=b_eq, bounds=bounds, method='highs')
        
        if result.success:
            raw_weights = result.x
            total = raw_weights.sum()
            return {
                node["id"]: round(w / total * 100, 1)
                for node, w in zip(self.nodes, raw_weights)
                if w > 0.01
            }
        else:
            return self._fallback_weighting()
    
    def _fallback_weighting(self) -> dict:
        """Fallback: distribute proportionally by health score."""
        total_health = sum(n["health_score"] for n in self.nodes)
        return {
            node["id"]: round(node["health_score"] / total_health * 100, 1)
            for node in self.nodes
        }

-- nginx/conf.d/ai_lb.lua
local redis = require "resty.redis"
local cjson = require "cjson"

local red = redis.new()
red:set_timeout(1000)
red:connect("127.0.0.1", 6379)

local weights_json = red:get("ai_lb:weights")
if weights_json then
    local weights = cjson.decode(weights_json)
    -- weights = {vps01=28, vps02=8, vps03=42, vps04=0}
    -- Update upstream weights dynamically via Lua
end

# dispatcher/envoy_xds_client.py
class EnvoyDispatcher:
    """Update Envoy routing config dynamically via xDS protocol."""
    
    def __init__(self, xds_host="127.0.0.1", xds_port=5678):
        self.client = XdsClient(host=xds_host, port=xds_port)
    
    def update_endpoints(self, weights: dict):
        """Update backend endpoint weights."""
        endpoints = []
        for node_id, weight in weights.items():
            endpoints.append({
                "address": f"{node_id}.internal",
                "port": 8080,
                "weight": weight,
                "health_check": weight > 0,
            })
        self.client.update_cluster("backend_cluster", endpoints)
        print(f"Updated {len(endpoints)} endpoints via xDS")

# verifier/result_evaluator.py
class ResultEvaluator:
    """
    Verify the actual effect of scheduling decisions.
    A/B comparison: 5min average before scheduling vs. 5min after.
    """
    
    def evaluate(self, before: dict, after: dict) -> dict:
        return {
            "avg_latency_ms": {
                "before": before["avg_latency_ms"],
                "after": after["avg_latency_ms"],
                "change_pct": self._pct_change(before["avg_latency_ms"], after["avg_latency_ms"]),
            },
            "error_rate": {
                "before": before["error_rate_percent"],
                "after": after["error_rate_percent"],
                "change_pct": self._pct_change(before["error_rate_percent"], after["error_rate_percent"]),
            },
            "load_variance": {
                "before": before["load_variance"],
                "after": after["load_variance"],
                "change_pct": self._pct_change(before["load_variance"], after["load_variance"]),
            },
            "decision_valid": after["avg_latency_ms"] < before["avg_latency_ms"],
        }

ai-load-balancer/
├── agent/                  # VPS node collector
│   ├── metrics_collector.py
│   ├── health_score.py
│   └── agent.sh
├── predictor/              # Traffic forecasting
│   ├── forecaster.py
│   └── models/
├── dispatcher/             # Scheduling decisions
│   ├── scheduler.py
│   ├── envoy_xds_client.py
│   └── nginx_lua/
├── verifier/               # Result verification
│   └── result_evaluator.py
├── config/
│   ├── nodes.yaml
│   ├── weights.yaml
│   └── thresholds.yaml
├── orchestrator.py
└── requirements.txt

# orchestrator.py
#!/usr/bin/env python3
"""
AI Load Balancer Orchestrator
Executes a full scheduling cycle every 30 seconds.
"""

import time
import yaml
from datetime import datetime
from agent.metrics_collector import NodeMetrics
from agent.health_score import HealthScoreCalculator
from predictor.traffic_forecaster import TrafficForecaster
from dispatcher.scheduler import AIScheduler
from dispatcher.envoy_xds_client import EnvoyDispatcher
from verifier.result_evaluator import ResultEvaluator

def main():
    with open("config/nodes.yaml") as f:
        config = yaml.safe_load(f)
    
    calculators = {node["id"]: HealthScoreCalculator() for node in config["nodes"]}
    forecaster = TrafficForecaster()
    scheduler = AIScheduler(config["nodes"])
    dispatcher = EnvoyDispatcher()
    evaluator = ResultEvaluator()
    
    print(f"[{datetime.utcnow().isoformat()}] AI-LB started, monitoring {len(config['nodes'])} nodes")
    
    while True:
        try:
            # Step 1: Collect metrics from all nodes
            all_metrics = {}
            for node in config["nodes"]:
                all_metrics[node["id"]] = NodeMetrics(node["id"]).collect()
            
            # Step 2: Compute health scores
            health_scores = {
                nid: calculators[nid].calculate(metrics, {})
                for nid, metrics in all_metrics.items()
            }
            
            # Step 3: Predict future traffic
            prediction = forecaster.predict(hours_ahead=0.5)
            
            # Step 4: Compute optimal weights
            weights = scheduler.compute_weights(prediction.get("predicted_peak_qps", 0))
            
            # Step 5: Deploy configuration
            dispatcher.update_endpoints(weights)
            
            print(f"[{datetime.utcnow().isoformat()}] Weights: {weights} | "
                  f"Predicted QPS: {prediction.get('predicted_peak_qps', 'N/A')}")
            
            time.sleep(30)
            
        except Exception as e:
            print(f"[ERROR] {e}")
            time.sleep(60)

if __name__ == "__main__":
    main()

# 1. Install dependencies
pip install prophet scipy psutil pyyaml redis envoy-control

# 2. Deploy Agent on each VPS
scp agent/agent.sh user@vps-01:~/ai-lb-agent/
ssh user@vps-01 '~/ai-lb-agent/agent.sh start'

# 3. Deploy Orchestrator on the control node
cd ~/ai-load-balancer
python3 orchestrator.py &

# 4. Configure Envoy to connect to xDS
# Add to Envoy config:
# discovery_service:
#   ads:
#     address: localhost:5678