# docker-compose.edge.yml
version: '3.8'
services:
  ollama:
    image: ollama/ollama:latest
    container_name: edge-ollama
    ports:
      - "127.0.0.1:11434:11434"
    volumes:
      - ./models:/root/.ollama
      - ./config:/etc/ollama
    environment:
      - OLLAMA_HOST=0.0.0.0
      - OLLAMA_ORIGINS=https://your-domain.com
      - OLLAMA_KEEP_ALIVE=-1
    deploy:
      resources:
        limits:
          memory: 8G
          cpus: '4.0'
    restart: unless-stopped

  ollama-exporter:
    image: prom/node-exporter:latest
    container_name: edge-ollama-exporter
    ports:
      - "9101:9100"
    volumes:
      - /proc:/host/proc:ro
      - /sys:/host/sys:ro
    restart: unless-stopped

# Start Ollama service
docker compose -f docker-compose.edge.yml up -d

# Pull edge-friendly lightweight models
ollama pull qwen2.5:7b-instruct-q4_K_M
ollama pull nomic-embed-text:v1.0.3

# Verify service
curl http://localhost:11434/api/tags

-- /etc/nginx/lua/router.lua
local geoip = require "resty.geoip"
local health = require "health.checker"

-- Node configuration
local nodes = {
    { name = "tokyo", ip = "10.0.1.10", port = 11434, region = "ap-northeast" },
    { name = "singapore", ip = "10.0.2.10", port = 11434, region = "ap-southeast" },
    { name = "virginia", ip = "10.0.3.10", port = 11434, region = "us-east" },
}

-- Select nearest node based on client IP
local function select_node_by_geo(client_ip)
    local region = geoip.lookup(client_ip)
    local best_node = nil
    local min_latency = math.huge

    for _, node in ipairs(nodes) do
        local latency = health.get_latency(node)
        if latency < min_latency then
            min_latency = latency
            best_node = node
        end
    end

    return best_node, min_latency
end

-- Nginx config example
-- upstream ollama_cluster {
--     server 10.0.1.10:11434;
--     server 10.0.2.10:11434;
--     server 10.0.3.10:11434;
-- }

http {
    # Define backend node pool
    upstream ollama_cluster {
        least_conn;
        server 10.0.1.10:11434 weight=3;   # Tokyo - high weight
        server 10.0.2.10:11434 weight=2;   # Singapore
        server 10.0.3.10:11434 weight=1;   # Virginia
        keepalive 32;
    }

    server {
        listen 80;
        server_name api.your-edge-cluster.com;

        # Health check - auto-remove unhealthy nodes
        location /health {
            access_by_lua_block {
                local health = require "health.checker"
                local ok = health.check_all_nodes()
                if not ok then
                    ngx.status = 503
                    ngx.say("Node unavailable")
                    return ngx.exit(503)
                end
            }
            return 200 "OK";
        }

        # AI inference request routing
        location /v1/ {
            proxy_pass http://ollama_cluster;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

            # Timeout settings
            proxy_connect_timeout 5s;
            proxy_read_timeout 120s;
            proxy_send_timeout 30s;

            # Add latency headers for client monitoring
            header_filter_by_lua_block {
                local node = ngx.var.upstream_addr
                ngx.header["X-Routed-Node"] = node
                ngx.header["X-Edge-Latency"] = ngx.var.upstream_response_time
            }
        }
    }
}

#!/bin/bash
# sync_models.sh - Model Version Sync Manager

CLUSTER_NODES=("10.0.1.10" "10.0.2.10" "10.0.3.10")
MODEL_REGISTRY="https://registry.your-edge-cluster.com"
CURRENT_VERSION=$(curl -s ${MODEL_REGISTRY}/latest-version)
SYNC_LOG="/var/log/edge-model-sync.log"

log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" >> $SYNC_LOG
}

# Check current version
check_version() {
    local node=$1
    local remote_version=$(ssh root@$node "ollama list | grep qwen2.5 | awk '{print \$2}'" 2>/dev/null)
    if [[ "$remote_version" != "$CURRENT_VERSION" ]]; then
        log "Node $node version mismatch: expected $CURRENT_VERSION, got $remote_version"
        return 1
    fi
    return 0
}

# Gray release sync: sync primary first, verify, then sync others
sync_to_nodes() {
    local primary=${CLUSTER_NODES[0]}
    
    # Step 1: Sync primary node
    log "Starting gray release sync to primary node: $primary"
    ssh root@$primary "cd /opt/ollama && ollama pull qwen2.5:7b-instruct-${CURRENT_VERSION}"
    
    # Wait for verification
    sleep 30
    
    # Check primary node is running
    if curl -sf http://${primary}:11434/api/tags > /dev/null; then
        log "Primary node verified OK"
        
        # Step 2: Sync remaining nodes
        for node in "${CLUSTER_NODES[@]:1}"; do
            log "Syncing to secondary node: $node"
            ssh root@$node "cd /opt/ollama && ollama pull qwen2.5:7b-instruct-${CURRENT_VERSION}"
            
            # Verify each node
            sleep 10
            if curl -sf http://${node}:11434/api/tags > /dev/null; then
                log "Node $node synced and verified OK"
            else
                log "ERROR: Node $node failed verification, rolling back"
                rollback_node $node
            fi
        done
        
        log "Gray release completed successfully"
    else
        log "ERROR: Primary node failed verification, aborting sync"
        rollback_primary
    fi
}

# Rollback functions
rollback_node() {
    local node=$1
    local prev_version=$(curl -s ${MODEL_REGISTRY}/prev-version)
    ssh root@$node "cd /opt/ollama && ollama pull qwen2.5:7b-instruct-${prev_version}"
    log "Rolled back node $node to version $prev_version"
}

rollback_primary() {
    log "Aborting gray release - rolling back primary"
    # Rollback to previous stable version logic...
}

# Scheduled execution: check for updates daily at 3 AM
# crontab: 0 3 * * * /opt/edge/sync_models.sh
sync_to_nodes

#!/usr/bin/env python3
"""
ai_load_balancer.py - AI-Driven Adaptive Load Balancer
Dynamically adjusts node weights based on historical performance data
"""

import time
import json
import requests
from collections import deque
from datetime import datetime, timedelta

class AdaptiveLoadBalancer:
    def __init__(self, config_path="lb_config.json"):
        with open(config_path) as f:
            self.config = json.load(f)
        
        self.nodes = self.config["nodes"]
        self.latency_history = {node["id"]: deque(maxlen=100) for node in self.nodes}
        self.error_history = {node["id"]: deque(maxlen=50) for node in self.nodes}
        self.weights = {node["id"]: node.get("weight", 1.0) for node in self.nodes}
        self.score_cache = {}
        self.cache_ttl = 5  # Score cache TTL: 5 seconds
        
    def record_latency(self, node_id, latency_ms):
        """Record node latency data"""
        self.latency_history[node_id].append({
            "timestamp": datetime.now(),
            "latency": latency_ms
        })
        self.score_cache = {}  # Clear cache, force recalculation
    
    def record_error(self, node_id):
        """Record node error"""
        self.error_history[node_id].append(datetime.now())
        self.score_cache = {}
    
    def calculate_node_score(self, node_id):
        """
        Calculate comprehensive node score (0-100)
        Dimensions:
        - Average latency (40%)
        - P99 latency (25%)
        - Error rate (20%)
        - Throughput (15%)
        """
        if node_id in self.score_cache:
            return self.score_cache[node_id]
        
        history = self.latency_history[node_id]
        errors = self.error_history[node_id]
        
        if len(history) < 5:
            return 50.0  # Default score for new nodes
        
        latencies = [h["latency"] for h in history]
        avg_latency = sum(latencies) / len(latencies)
        p99_latency = sorted(latencies)[int(len(latencies) * 0.99)]
        
        # Count errors in time window
        window_start = datetime.now() - timedelta(minutes=10)
        recent_errors = sum(1 for e in errors if e > window_start)
        error_rate = recent_errors / max(len(errors), 1)
        
        # Comprehensive scoring
        latency_score = max(0, 100 - (avg_latency / 5))
        p99_score = max(0, 100 - (p99_latency / 10))
        error_score = max(0, 100 - (error_rate * 500))
        
        total_score = (
            latency_score * 0.40 +
            p99_score * 0.25 +
            error_score * 0.20 +
            (100 - min(avg_latency, 100)) * 0.15
        )
        
        self.score_cache[node_id] = total_score
        return total_score
    
    def update_weights(self):
        """Dynamically update node weights based on scores"""
        total_score = sum(self.calculate_node_score(nid) for nid in self.weights)
        
        if total_score == 0:
            return
        
        for node_id in self.weights:
            score = self.calculate_node_score(node_id)
            self.weights[node_id] = round(score / total_score, 4)
        
        # Normalize weights
        weight_sum = sum(self.weights.values())
        for node_id in self.weights:
            self.weights[node_id] /= weight_sum
        
        self.score_cache = {}
    
    def get_next_node(self):
        """Weighted random selection based on scores"""
        weighted_list = []
        for node_id, weight in self.weights.items():
            for node in self.nodes:
                if node["id"] == node_id:
                    weighted_list.append((node, weight))
                    break
        
        total_weight = sum(w for _, w in weighted_list)
        r = time.time() % total_weight
        
        cumulative = 0
        for node, weight in weighted_list:
            cumulative += weight
            if r <= cumulative:
                return node
        
        return weighted_list[-1][0]
    
    def run_monitoring_loop(self, interval=10):
        """Continuous monitoring and weight updates"""
        print(f"[{datetime.now()}] AI Load Balancer started")
        print(f"Nodes: {[n['id'] for n in self.nodes]}")
        
        while True:
            self.update_weights()
            
            print(f"\n[{datetime.now()}] Current weights:")
            for node_id, weight in self.weights.items():
                score = self.calculate_node_score(node_id)
                history = self.latency_history[node_id]
                avg_lat = sum(h["latency"] for h in history) / len(history) if history else 0
                print(f"  {node_id}: weight={weight:.4f} score={score:.1f} avg_latency={avg_lat:.1f}ms")
            
            time.sleep(interval)

if __name__ == "__main__":
    lb = AdaptiveLoadBalancer()
    lb.run_monitoring_loop()

-- /etc/nginx/lua/dynamic_upstream.lua
-- Periodically fetch latest weights from AI load balancer and update upstream

local cjson = require "cjson"
local http = require "resty.http"

-- Fetch weight data from AI LB
local function fetch_weights()
    local httpc = require "resty.http".new()
    local res, err = httpc:request_uri("http://127.0.0.1:8080/api/weights", {
        method = "GET",
        header = { ["Accept"] = "application/json" }
    })
    
    if res and res.status == 200 then
        return cjson.decode(res.body)
    end
    return nil
end

-- Dynamically set upstream weights
local function set_upstream_weights(weights)
    if not weights then return end
    
    local shm = ngx.shared.lb_weights
    if not shm then
        ngx.log(ngx.ERR, "Shared memory zone not initialized")
        return
    end
    
    for node_id, weight in pairs(weights) do
        shm:set(node_id, tostring(weight), 300)
    end
end

-- Update every 30 seconds
timer = ngx.timer.every(30, function()
    local weights = fetch_weights()
    if weights then
        set_upstream_weights(weights)
    end
end)

#!/usr/bin/env python3
"""
edge_health_monitor.py - Edge Node Health Monitoring & Self-Healing
"""

import time
import json
import requests
import subprocess
from datetime import datetime
from typing import Dict, List, Optional

class EdgeHealthMonitor:
    def __init__(self, config_path="monitor_config.json"):
        with open(config_path) as f:
            self.config = json.load(f)
        
        self.nodes = self.config["nodes"]
        self.alerts = []
        self.recovery_actions = self.config.get("recovery_actions", {})
    
    def check_node_health(self, node: dict) -> dict:
        """Comprehensive health check for a single node"""
        result = {
            "node_id": node["id"],
            "ip": node["ip"],
            "timestamp": datetime.now().isoformat(),
            "checks": {},
            "overall_status": "healthy"
        }
        
        # 1. Network connectivity
        try:
            ping = subprocess.run(
                ["ping", "-c", "3", "-W", "2", node["ip"]],
                capture_output=True, text=True, timeout=10
            )
            result["checks"]["connectivity"] = {
                "status": "up" if ping.returncode == 0 else "down",
                "details": "reachable" if ping.returncode == 0 else "unreachable"
            }
            if ping.returncode != 0:
                result["overall_status"] = "unhealthy"
        except Exception as e:
            result["checks"]["connectivity"] = {"status": "error", "details": str(e)}
            result["overall_status"] = "unhealthy"
        
        # 2. Ollama service health
        try:
            resp = requests.get(
                f"http://{node['ip']}:11434/api/tags",
                timeout=10
            )
            if resp.status_code == 200:
                models = resp.json().get("models", [])
                result["checks"]["ollama_service"] = {
                    "status": "running",
                    "models_loaded": len(models),
                    "model_names": [m["name"] for m in models]
                }
            else:
                result["checks"]["ollama_service"] = {
                    "status": "error",
                    "details": f"HTTP {resp.status_code}"
                }
                result["overall_status"] = "degraded"
        except requests.exceptions.ConnectionError:
            result["checks"]["ollama_service"] = {
                "status": "down",
                "details": "connection refused"
            }
            result["overall_status"] = "unhealthy"
        except Exception as e:
            result["checks"]["ollama_service"] = {
                "status": "error",
                "details": str(e)
            }
        
        # 3. System resources
        try:
            ssh_cmd = (f"ssh root@{node['ip']} 'free -m | awk \"/Mem:/{{print "
                       f"\\$3/\\$2*100}}' && df -m / | awk \'NR==2{{print "
                       f"\\$5}}' && top -bn1 | grep Cpu | awk \\'"
                       f"{{print \\$2}}\\'")
            proc = subprocess.run(ssh_cmd, shell=True, capture_output=True, 
                                  text=True, timeout=15)
            outputs = proc.stdout.strip().split('\n')
            
            if len(outputs) >= 3:
                mem_usage = float(outputs[0])
                disk_usage = outputs[1].rstrip('%')
                cpu_idle = outputs[2].split(',')[0].split(':')[1].strip()
                
                result["checks"]["resources"] = {
                    "memory_usage_percent": mem_usage,
                    "disk_usage_percent": float(disk_usage),
                    "cpu_idle_percent": float(cpu_idle)
                }
                
                if mem_usage > 90:
                    result["overall_status"] = "critical"
                elif mem_usage > 80:
                    result["overall_status"] = "warning"
        except Exception as e:
            result["checks"]["resources"] = {"status": "error", "details": str(e)}
        
        # 4. Inference latency test
        try:
            start = time.time()
            resp = requests.post(
                f"http://{node['ip']}:11434/api/generate",
                json={
                    "model": "qwen2.5:7b",
                    "prompt": "Say hello in 1 word",
                    "stream": False
                },
                timeout=30
            )
            latency_ms = (time.time() - start) * 1000
            
            result["checks"]["inference_latency"] = {
                "status": "ok" if resp.status_code == 200 else "error",
                "latency_ms": round(latency_ms, 2),
                "response_code": resp.status_code
            }
            
            if latency_ms > 5000:
                if result["overall_status"] == "healthy":
                    result["overall_status"] = "degraded"
        except Exception as e:
            result["checks"]["inference_latency"] = {
                "status": "error",
                "details": str(e)
            }
        
        return result
    
    def auto_remediate(self, node_id: str, issue: str):
        """Auto-execute fix based on issue type"""
        remediation_map = {
            "ollama_down": [
                f"ssh root@{{node_id}} 'systemctl restart ollama'",
                f"ssh root@{{node_id}} 'docker compose -f /opt/edge/docker-compose.edge.yml restart ollama'"
            ],
            "high_memory": [
                f"ssh root@{{node_id}} 'sync && echo 3 > /proc/sys/vm/drop_caches'",
                f"ssh root@{{node_id}} 'systemctl restart ollama'"
            ],
            "high_disk": [
                f"ssh root@{{node_id}} 'journalctl --vacuum-size=100M'",
                f"ssh root@{{node_id}} 'ollama rm unused-models'"
            ]
        }
        
        actions = remediation_map.get(issue, [])
        for action in actions:
            try:
                cmd = action.format(node_id=node_id.replace(".", "\\."))
                subprocess.run(cmd, shell=True, timeout=30, capture_output=True)
                print(f"[{datetime.now()}] Executed remediation: {cmd}")
            except Exception as e:
                print(f"[{datetime.now()}] Remediation failed: {e}")
    
    def run_full_monitor(self):
        """Full monitoring loop"""
        print(f"[{datetime.now()}] Starting edge health monitoring")
        print(f"Monitoring {len(self.nodes)} nodes...\n")
        
        while True:
            all_results = []
            critical_found = False
            
            for node in self.nodes:
                result = self.check_node_health(node)
                all_results.append(result)
                
                if result["overall_status"] == "critical":
                    critical_found = True
                    self.alerts.append({
                        "node_id": node["id"],
                        "severity": "critical",
                        "timestamp": result["timestamp"],
                        "details": result
                    })
                    print(f"🚨 CRITICAL: {node['id']} - {result['overall_status']}")
                    
                    # Auto-remediate
                    for check_name, check_result in result["checks"].items():
                        if check_result.get("status") in ["down", "error"]:
                            self.auto_remediate(node["id"], check_name)
                
                elif result["overall_status"] == "degraded":
                    print(f"⚠️  DEGRADED: {node['id']} - {result['overall_status']}")
                else:
                    print(f"✅ HEALTHY: {node['id']} - {result['overall_status']}")
            
            # Generate monitoring report
            report = {
                "timestamp": datetime.now().isoformat(),
                "nodes_checked": len(all_results),
                "healthy": sum(1 for r in all_results if r["overall_status"] == "healthy"),
                "degraded": sum(1 for r in all_results if r["overall_status"] == "degraded"),
                "critical": sum(1 for r in all_results if r["overall_status"] == "critical"),
                "unhealthy": sum(1 for r in all_results if r["overall_status"] == "unhealthy"),
            }
            
            print(f"\n📊 Cluster Report: {json.dumps(report, indent=2)}\n")
            
            time.sleep(self.config.get("check_interval", 60))

if __name__ == "__main__":
    monitor = EdgeHealthMonitor()
    monitor.run_full_monitor()

{
    "nodes": [
        {"id": "tokyo-node", "ip": "10.0.1.10", "region": "ap-northeast"},
        {"id": "singapore-node", "ip": "10.0.2.10", "region": "ap-southeast"},
        {"id": "virginia-node", "ip": "10.0.3.10", "region": "us-east"}
    ],
    "check_interval": 60,
    "recovery_actions": {
        "ollama_down": "restart_service",
        "high_memory": "clear_cache",
        "high_disk": "vacuum_logs"
    }
}

# 1. Use quantized models to reduce memory footprint
# Q4_K_M quantization reduces 7B model from ~14GB to ~5GB
ollama pull qwen2.5:7b-instruct-q4_K_M

# 2. Limit concurrent requests
# Edit /etc/ollama/config.json
cat > /etc/ollama/config.json << 'EOF'
{
    "num_parallel": 1,
    "max_requests": 4,
    "gpu_layers": 35
}
EOF
systemctl restart ollama

# 3. Set container resource limits
# Already configured in docker-compose.yml deploy.resources.limits

# 4. Enable model warmup to reduce cold-start latency
# Add to startup script
curl -s http://localhost:11434/api/generate \
  -d '{"model":"qwen2.5:7b","prompt":"test","stream":false}' > /dev/null

# 1. Enable TCP BBR congestion control
cat > /etc/sysctl.d/99-edge-tuning.conf << 'EOF'
net.core.default_qdisc = fq
net.ipv4.tcp_congestion_control = bbr
net.ipv4.tcp_max_syn_backlog = 8192
net.core.somaxconn = 65535
EOF
sysctl -p /etc/sysctl.d/99-edge-tuning.conf

# 2. Optimize SSH connection reuse (for inter-node communication)
cat >> ~/.ssh/config << 'EOF'
Host 10.0.*.*
    ControlMaster auto
    ControlPath /tmp/%r@%h:%p
    ControlPersist 600
    ServerAliveInterval 30
    ServerAliveCountMax 3
EOF

# 3. Encrypt inter-node communication with mTLS
# Generate certificates
openssl req -x509 -newkey rsa:4096 -keyout ca.key -out ca.crt \
    -days 365 -nodes -subj "/CN=edge-cluster-ca"

# cost_optimizer.py - Traffic-based elastic scaling decisions
import json

def analyze_traffic_pattern(hourly_data: list) -> dict:
    """Analyze hourly traffic patterns, predict off-peak periods"""
    total = sum(h["requests"] for h in hourly_data)
    avg = total / len(hourly_data) if hourly_data else 0
    
    low_traffic_hours = [
        h["hour"] for h in hourly_data 
        if h["requests"] < avg * 0.3
    ]
    
    return {
        "avg_hourly_requests": round(avg, 1),
        "low_traffic_hours": low_traffic_hours,
        "peak_hour": max(hourly_data, key=lambda x: x["requests"])["hour"],
        "savings_potential_pct": round(
            (len(low_traffic_hours) / 24) * 100, 1
        )
    }

# Example: Scale down non-critical nodes during 2-6 AM (low traffic)
# Saves approximately 25% of monthly costs