diff --git a/deploy/nginx_172.31.170.47.conf b/deploy/nginx_172.31.170.47.conf index 091caba3..53d5adff 100644 --- a/deploy/nginx_172.31.170.47.conf +++ b/deploy/nginx_172.31.170.47.conf @@ -58,8 +58,8 @@ server { include fastcgi_params; } - # ── Installer assets (CSS/JS in /web/) ── - location ~ ^/(css|js|vendor)/ { + # ── Installer + Agent assets (root = /web/) ── + location ~ ^/(css|js|vendor|agent)/ { root /www/wwwroot/api.synaglobal.vip/web; try_files $uri =404; } diff --git a/deploy/nginx_api.synaglobal.vip.conf b/deploy/nginx_api.synaglobal.vip.conf index 0a319b75..627972d4 100644 --- a/deploy/nginx_api.synaglobal.vip.conf +++ b/deploy/nginx_api.synaglobal.vip.conf @@ -45,8 +45,8 @@ server { include fastcgi_params; } - # ── Installer assets (CSS/JS in /web/) ── - location ~ ^/(css|js|vendor)/ { + # ── Installer + Agent assets (root = /web/) ── + location ~ ^/(css|js|vendor|agent)/ { root /www/wwwroot/api.synaglobal.vip/web; try_files $uri =404; } diff --git a/server/application/services/sync_engine_v2.py b/server/application/services/sync_engine_v2.py index 99e7ae6b..87a6ff3c 100644 --- a/server/application/services/sync_engine_v2.py +++ b/server/application/services/sync_engine_v2.py @@ -55,6 +55,7 @@ class SyncEngineV2: operator: str = "admin", batch_size: int = 50, concurrency: int = 10, + trigger_type: str = "manual", ) -> dict: """S1: File sync using rsync over SSH""" concurrency = min(concurrency, MAX_CONCURRENT) @@ -80,7 +81,7 @@ class SyncEngineV2: server_id=server.id, source_path=source_path, target_path=target_path or server.target_path or "/tmp/sync", - trigger_type="manual", + trigger_type=trigger_type, operator=operator, status="running", sync_mode=sync_mode, diff --git a/web/agent/agent.py b/web/agent/agent.py index 6c08c90f..10c25ad1 100644 --- a/web/agent/agent.py +++ b/web/agent/agent.py @@ -1,6 +1,6 @@ """ -MultiSync Agent - Runs on each sub-server -Provides: health endpoint, heartbeat, config reload +Nexus Agent — Runs on each managed server +Provides: health endpoint, heartbeat, command execution, config reload """ import os import sys @@ -9,6 +9,7 @@ import time import asyncio import logging import platform +import subprocess from datetime import datetime from pathlib import Path from typing import Optional @@ -18,7 +19,7 @@ import psutil from fastapi import FastAPI, HTTPException, Header, Request from fastapi.responses import JSONResponse -# --- Configuration (JSON, 主服务器可推送更新) --- +# --- Configuration --- CONFIG_PATH = Path(__file__).parent / "config.json" @@ -26,7 +27,6 @@ def load_config(): if CONFIG_PATH.exists(): with open(CONFIG_PATH) as f: return json.load(f) - # Fallback to example example = Path(__file__).parent / "config.example.json" if example.exists(): with open(example) as f: @@ -36,13 +36,14 @@ def load_config(): config = load_config() API_KEY = config.get("api_key", "") +SERVER_ID = config.get("server_id", 0) CENTRAL_URL = config.get("central", {}).get("url", "") CENTRAL_API_KEY = config.get("central", {}).get("api_key", "") -HEARTBEAT_INTERVAL = config.get("heartbeat_interval", 30) +HEARTBEAT_INTERVAL = config.get("heartbeat_interval", 60) # --- Logging --- -log_file = config.get("log_file", "/var/log/multisync-agent.log") +log_file = config.get("log_file", "/var/log/nexus-agent.log") log_level = config.get("log_level", "INFO") logging.basicConfig( @@ -53,11 +54,11 @@ logging.basicConfig( logging.FileHandler(log_file, mode="a") if log_file != "stdout" else logging.StreamHandler(sys.stdout), ], ) -logger = logging.getLogger("multisync-agent") +logger = logging.getLogger("nexus-agent") # --- FastAPI App --- -app = FastAPI(title="MultiSync Agent", version="1.0.0") +app = FastAPI(title="Nexus Agent", version="2.0.0") def verify_api_key(x_api_key: str = Header(default="")): @@ -69,7 +70,7 @@ def verify_api_key(x_api_key: str = Header(default="")): @app.get("/health") async def health_check(): - """Health check endpoint - called by central server""" + """Health check endpoint — called by Nexus central server""" try: cpu_percent = psutil.cpu_percent(interval=0.5) memory = psutil.virtual_memory() @@ -79,80 +80,104 @@ async def health_check(): "hostname": platform.node(), "platform": platform.platform(), "python_version": platform.python_version(), - "cpu_percent": cpu_percent, + "cpu_usage": cpu_percent, + "mem_usage": memory.percent, + "disk_usage": round(disk.percent, 1), "memory_total_gb": round(memory.total / (1024**3), 2), "memory_used_gb": round(memory.used / (1024**3), 2), - "memory_percent": memory.percent, "disk_total_gb": round(disk.total / (1024**3), 2), "disk_used_gb": round(disk.used / (1024**3), 2), - "disk_percent": round(disk.percent, 1), "uptime_seconds": int(time.time() - psutil.boot_time()), "agent_time": datetime.utcnow().isoformat(), } + return {"status": "healthy", "system_info": system_info} + except Exception as e: + return {"status": "degraded", "error": str(e)} + + +# --- Command Execution --- + +@app.post("/exec") +async def exec_command(payload: dict, x_api_key: str = Header(default="")): + """Execute a shell command — called by Nexus to run commands on this server""" + verify_api_key(x_api_key) + + command = payload.get("command", "") + timeout = payload.get("timeout", 30) + sudo = payload.get("sudo", False) + + if not command: + raise HTTPException(status_code=400, detail="command is required") + + if sudo: + command = f"sudo {command}" + + try: + proc = await asyncio.create_subprocess_shell( + command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + exit_code = proc.returncode or 0 + return { - "status": "healthy", - "system_info": system_info, + "status": "success" if exit_code == 0 else "failed", + "stdout": stdout.decode()[:10000], + "stderr": stderr.decode()[:10000], + "exit_code": exit_code, + } + except asyncio.TimeoutError: + proc.kill() + return { + "status": "timeout", + "stdout": "", + "stderr": f"Command timed out after {timeout}s", + "exit_code": -1, } except Exception as e: - return { - "status": "degraded", - "error": str(e), - } + return {"status": "error", "stdout": "", "stderr": str(e), "exit_code": -1} +# --- Config Reload --- + @app.post("/config/reload") async def reload_config(data: dict, x_api_key: str = Header(default="")): - """主服务器推送新配置 — 更新 config.json 并重载""" + """Nexus pushes new config — update config.json and reload""" verify_api_key(x_api_key) try: now = datetime.utcnow().isoformat() data["updated_at"] = now - data["updated_by"] = "master" + data["updated_by"] = "nexus" with open(CONFIG_PATH, "w") as f: json.dump(data, f, indent=2) - # 重新加载配置 - global CENTRAL_URL, CENTRAL_API_KEY, config + + global SERVER_ID, CENTRAL_URL, CENTRAL_API_KEY, config config = data - CENTRAL_URL = config.get("central", {}).get("url", "") - CENTRAL_API_KEY = config.get("central", {}).get("api_key", "") - return {"status": "ok", "message": "Config updated", "central_url": CENTRAL_URL, "updated_at": now} + SERVER_ID = config.get("server_id", SERVER_ID) + CENTRAL_URL = config.get("central", {}).get("url", CENTRAL_URL) + CENTRAL_API_KEY = config.get("central", {}).get("api_key", CENTRAL_API_KEY) + + return {"status": "ok", "message": "Config updated", "updated_at": now} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) -@app.post("/uninstall") -async def uninstall_self(x_api_key: str = Header(default="")): - """主服务器远程触发卸载 — Agent 执行卸载脚本并停止自身""" - verify_api_key(x_api_key) - import subprocess, os as _os - script = "/opt/multisync-agent/uninstall.sh" - if _os.path.exists(script): - try: - proc = subprocess.run(["bash", script], capture_output=True, text=True, timeout=30) - return {"status": "ok", "message": "Agent uninstalled", "output": proc.stdout[:500]} - except subprocess.TimeoutExpired: - return {"status": "timeout", "message": "Uninstall script timed out"} - except Exception as e: - return {"status": "error", "message": str(e)} - else: - return {"status": "not_found", "message": f"Uninstall script not found: {script}"} - - # --- Heartbeat to Central Server --- _last_metrics = {} _last_full_sync = 0 HISTORY_SIZE = 10 -# 初始化采样窗口（避免重启后空窗等 5 分钟） _cpu_init = psutil.cpu_percent() _mem_init = psutil.virtual_memory().percent _disk_init = psutil.disk_usage("/").percent _metric_history = [{"cpu": _cpu_init, "memory": _mem_init, "disk": _disk_init}] * min(5, HISTORY_SIZE) + def _check_alert(): - """连续 HISTORY_SIZE 次 CPU/内存>80% 或磁盘>90% 返回告警标记""" + """Check if CPU/mem > 80% or disk > 90% for HISTORY_SIZE consecutive samples""" if len(_metric_history) < HISTORY_SIZE: return False, {} cpu_high = all(h["cpu"] >= 80 for h in _metric_history) @@ -166,57 +191,65 @@ def _check_alert(): async def send_heartbeat(): - """IoT 优化：变化 >10% + 连续 10 次高负载告警 + Redis 存指标""" + """Send heartbeat to Nexus central — smart: only when metrics change >10% or alert""" global _last_metrics, _last_full_sync, _metric_history while True: try: - if CENTRAL_URL: - cpu = psutil.cpu_percent() - mem = psutil.virtual_memory().percent - disk = psutil.disk_usage("/").percent + if not CENTRAL_URL or not SERVER_ID: + await asyncio.sleep(HEARTBEAT_INTERVAL) + continue - now_ts = time.time() - changed = ( - abs(cpu - _last_metrics.get("cpu", 0)) >= 10 or - abs(mem - _last_metrics.get("memory", 0)) >= 10 or - abs(disk - _last_metrics.get("disk", 0)) >= 10 - ) - force_sync = (now_ts - _last_full_sync) > 600 # 10 分钟强制全量 + cpu = psutil.cpu_percent() + mem = psutil.virtual_memory().percent + disk = psutil.disk_usage("/").percent - # 滑动窗口记录采样 - _metric_history.append({"cpu": cpu, "memory": mem, "disk": disk}) - if len(_metric_history) > HISTORY_SIZE: - _metric_history.pop(0) + now_ts = time.time() + changed = ( + abs(cpu - _last_metrics.get("cpu", 0)) >= 10 or + abs(mem - _last_metrics.get("memory", 0)) >= 10 or + abs(disk - _last_metrics.get("disk", 0)) >= 10 + ) + force_sync = (now_ts - _last_full_sync) > 600 - # 检查告警 - has_alert, alert_info = _check_alert() + _metric_history.append({"cpu": cpu, "memory": mem, "disk": disk}) + if len(_metric_history) > HISTORY_SIZE: + _metric_history.pop(0) + + has_alert, alert_info = _check_alert() + + if changed or force_sync or has_alert: + _last_metrics = {"cpu": cpu, "memory": mem, "disk": disk} + if force_sync: + _last_full_sync = now_ts + + payload = { + "server_id": SERVER_ID, + "is_online": True, + "agent_version": "2.0.0", + "system_info": { + "cpu_usage": cpu, + "mem_usage": mem, + "disk_usage": disk, + "hostname": platform.node(), + "platform": platform.platform(), + }, + } + if has_alert: + payload["alert"] = alert_info + + async with httpx.AsyncClient(timeout=10) as client: + resp = await client.post( + f"{CENTRAL_URL}/api/agent/heartbeat", + json=payload, + headers={"X-API-Key": CENTRAL_API_KEY}, + ) + if resp.status_code == 200: + logger.debug("Heartbeat sent") + else: + logger.warning(f"Heartbeat failed: {resp.status_code}") + else: + logger.debug("Metrics unchanged, skip heartbeat") - if changed or force_sync or has_alert: - _last_metrics = {"cpu": cpu, "memory": mem, "disk": disk} - if force_sync: _last_full_sync = now_ts - payload = { - "timestamp": datetime.utcnow().isoformat(), - "agent_version": "2.0.0", - "system_info": { - "hostname": platform.node(), - "cpu": cpu, "memory": mem, "disk": disk, - "history": _metric_history, - }, - } - if has_alert: - payload["alert"] = alert_info - async with httpx.AsyncClient(timeout=10) as client: - resp = await client.post( - f"{CENTRAL_URL}/api/agent/heartbeat", - json=payload, - headers={"X-API-Key": CENTRAL_API_KEY}, - ) - if resp.status_code == 200: - logger.debug("Heartbeat sent successfully") - else: - logger.warning(f"Heartbeat failed: {resp.status_code}") - else: - logger.debug("Metrics unchanged <10%, skip heartbeat") except Exception as e: logger.warning(f"Heartbeat error: {e}") @@ -224,14 +257,10 @@ async def send_heartbeat(): # --- App Lifecycle --- -# 注：TriggerFileHandler + start_watchers 已删除（触发文件机制已废弃） -# Agent 仅保留心跳上报 + 健康接口 @app.on_event("startup") async def startup(): - logger.info("MultiSync Agent starting...") - - # Start heartbeat + logger.info(f"Nexus Agent starting (server_id={SERVER_ID})") asyncio.create_task(send_heartbeat()) host = config.get("server", {}).get("host", "0.0.0.0") @@ -241,13 +270,11 @@ async def startup(): @app.on_event("shutdown") async def shutdown(): - logger.info("MultiSync Agent stopped") + logger.info("Nexus Agent stopped") -# --- Main --- - if __name__ == "__main__": + host = config.get("server", {}).get("host", "0.0.0.0") + port = config.get("server", {}).get("port", 8601) import uvicorn - host = config["server"]["host"] - port = config["server"]["port"] uvicorn.run(app, host=host, port=port) diff --git a/web/agent/agent.sh b/web/agent/agent.sh index 0eb7f2b5..c7c2e9e1 100644 --- a/web/agent/agent.sh +++ b/web/agent/agent.sh @@ -1,12 +1,13 @@ #!/bin/bash # ================================================================ -# MultiSync Shell Agent — 心跳 + 健康接口 +# Nexus Shell Agent — 心跳 + 健康接口 # 依赖: bash, curl # ================================================================ -HEARTBEAT_INTERVAL="${HEARTBEAT_INTERVAL:-30}" -LOG_FILE="${LOG_FILE:-/var/log/multisync-agent.log}" +HEARTBEAT_INTERVAL="${HEARTBEAT_INTERVAL:-60}" +LOG_FILE="${LOG_FILE:-/var/log/nexus-agent.log}" CENTRAL_URL="${CENTRAL_URL:-}" CENTRAL_API_KEY="${CENTRAL_API_KEY:-}" +SERVER_ID="${SERVER_ID:-0}" AGENT_PORT="${AGENT_PORT:-8601}" log() { @@ -36,7 +37,7 @@ heartbeat_loop() { local info=$(get_system_info) curl -s -o /dev/null -X POST -H "Content-Type: application/json" \ -H "X-API-Key: ${CENTRAL_API_KEY}" \ - -d "{\"agent_port\":${AGENT_PORT},\"system_info\":${info}}" \ + -d "{\"server_id\":${SERVER_ID},\"is_online\":true,\"system_info\":${info}}" \ "${CENTRAL_URL}/api/agent/heartbeat" 2>/dev/null if [ $? -eq 0 ]; then log "INFO" "心跳 OK" @@ -50,7 +51,7 @@ heartbeat_loop() { health_server() { while true; do if command -v nc >/dev/null 2>&1; then - printf "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n{\"status\":\"healthy\",\"agent\":\"shell\"}" \ + printf "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n{\"status\":\"healthy\",\"agent\":\"nexus-shell\"}" \ | nc -l -p "${AGENT_PORT}" -q 1 2>/dev/null else exec 3<>/dev/tcp/0.0.0.0/${AGENT_PORT} 2>/dev/null && { @@ -69,10 +70,11 @@ trap cleanup SIGINT SIGTERM main() { echo "==========================================" - echo " MultiSync Shell Agent — 心跳上报" + echo " Nexus Shell Agent — 心跳上报" echo "==========================================" check_deps log "INFO" "中央服务器: ${CENTRAL_URL}" + log "INFO" "服务器ID: ${SERVER_ID}" log "INFO" "心跳间隔: ${HEARTBEAT_INTERVAL}秒" heartbeat_loop & @@ -88,10 +90,9 @@ main() { } case "${1:-}" in - install) bash "$(dirname "$0")/install.sh" "${@:2}" ;; - stop) systemctl stop multisync-agent; echo "已停止" ;; - restart) systemctl restart multisync-agent; echo "已重启" ;; - status) systemctl status multisync-agent --no-pager 2>/dev/null || echo "未运行" ;; - log|logs) journalctl -u multisync-agent -f --no-pager ;; + stop) systemctl stop nexus-agent; echo "已停止" ;; + restart) systemctl restart nexus-agent; echo "已重启" ;; + status) systemctl status nexus-agent --no-pager 2>/dev/null || echo "未运行" ;; + log|logs) journalctl -u nexus-agent -f --no-pager ;; *) main ;; esac diff --git a/web/agent/config.example.json b/web/agent/config.example.json new file mode 100644 index 00000000..87d2cde5 --- /dev/null +++ b/web/agent/config.example.json @@ -0,0 +1,15 @@ +{ + "server_id": 0, + "central": { + "url": "https://api.synaglobal.vip", + "api_key": "YOUR_API_KEY" + }, + "api_key": "YOUR_API_KEY", + "heartbeat_interval": 60, + "log_file": "/var/log/nexus-agent.log", + "log_level": "INFO", + "server": { + "host": "0.0.0.0", + "port": 8601 + } +} diff --git a/web/agent/install.sh b/web/agent/install.sh index 7bbe6e6e..54cf4581 100644 --- a/web/agent/install.sh +++ b/web/agent/install.sh @@ -1,7 +1,7 @@ #!/bin/bash # ================================================================ -# MultiSync Python Agent Install Script -# Usage: curl -fsSL https://主服务器/agent/install.sh | bash -s -- --url https://主服务器 --key KEY +# Nexus Agent Install Script +# Usage: curl -fsSL https://api.synaglobal.vip/agent/install.sh | bash -s -- --url https://api.synaglobal.vip --key KEY --id SERVER_ID # ================================================================ set -e @@ -9,30 +9,32 @@ set -e CENTRAL_URL="" WEB_URL="" API_KEY="" +SERVER_ID="" AGENT_PORT="8601" -WATCH_DIRS="/var/www/html" -INSTALL_DIR="/opt/multisync-agent" +INSTALL_DIR="/opt/nexus-agent" while [[ $# -gt 0 ]]; do case $1 in --url) CENTRAL_URL="$2"; shift 2 ;; --web-url) WEB_URL="$2"; shift 2 ;; --key) API_KEY="$2"; shift 2 ;; + --id) SERVER_ID="$2"; shift 2 ;; --port) AGENT_PORT="$2"; shift 2 ;; - --dirs) WATCH_DIRS="$2"; shift 2 ;; *) shift ;; esac done [ -z "$CENTRAL_URL" ] && echo "ERROR: --url is required" && exit 1 +[ -z "$SERVER_ID" ] && echo "ERROR: --id is required (server ID from Nexus dashboard)" && exit 1 +[ -z "$API_KEY" ] && echo "ERROR: --key is required (API key from Nexus settings)" && exit 1 echo "" echo "==========================================" -echo " MultiSync Python Agent Install" +echo " Nexus Agent Install" echo "==========================================" -echo " Server: $CENTRAL_URL" -echo " Dirs: $WATCH_DIRS" -echo " Port: $AGENT_PORT" +echo " Server: $CENTRAL_URL" +echo " ID: $SERVER_ID" +echo " Port: $AGENT_PORT" echo "==========================================" echo "" @@ -56,7 +58,7 @@ mkdir -p "$INSTALL_DIR" # 3. Download agent.py echo "[3/5] Download agent.py..." -[ -z "$WEB_URL" ] && WEB_URL="$(echo "$CENTRAL_URL" | sed 's|:[0-9]\+||; s|^http://|https://|')" +[ -z "$WEB_URL" ] && WEB_URL="${CENTRAL_URL}" curl -fsSL "${WEB_URL}/agent/agent.py" -o "$INSTALL_DIR/agent.py" || { echo "ERROR: Cannot download from $WEB_URL/agent/agent.py" exit 1 @@ -66,21 +68,22 @@ curl -fsSL "${WEB_URL}/agent/agent.py" -o "$INSTALL_DIR/agent.py" || { echo "[4/5] Write config..." cat > "$INSTALL_DIR/config.json" << EOF { + "server_id": ${SERVER_ID}, "central": { "url": "${CENTRAL_URL}", "api_key": "${API_KEY}" }, "api_key": "${API_KEY}", - "heartbeat_interval": 30, - "watch_dirs": "${WATCH_DIRS}", - "log_file": "/var/log/multisync-agent.log", - "log_level": "INFO" + "heartbeat_interval": 60, + "log_file": "/var/log/nexus-agent.log", + "log_level": "INFO", + "server": { "host": "0.0.0.0", "port": ${AGENT_PORT} } } EOF # 5. systemd + firewall + start echo "[5/5] Setup service + firewall + start..." -cat > /etc/systemd/system/multisync-agent.service << EOF +cat > /etc/systemd/system/nexus-agent.service << EOF [Unit] -Description=MultiSync Python Agent +Description=Nexus Agent After=network.target [Service] @@ -96,7 +99,6 @@ WantedBy=multi-user.target EOF # Firewall -AGENT_PORT="${AGENT_PORT:-8601}" if command -v ufw &>/dev/null && ufw status 2>/dev/null | grep -q 'Status: active'; then ufw allow ${AGENT_PORT}/tcp 2>/dev/null && echo " ufw: opened ${AGENT_PORT}/tcp" elif command -v firewall-cmd &>/dev/null && firewall-cmd --state 2>/dev/null | grep -q 'running'; then @@ -109,12 +111,12 @@ else fi systemctl daemon-reload -systemctl enable multisync-agent -systemctl start multisync-agent +systemctl enable nexus-agent +systemctl start nexus-agent sleep 2 STATUS="running" -systemctl is-active --quiet multisync-agent || STATUS="FAILED" +systemctl is-active --quiet nexus-agent || STATUS="FAILED" HOSTNAME=$(hostname 2>/dev/null || echo "unknown") IP=$(hostname -I 2>/dev/null | awk '{print $1}' || echo "unknown") @@ -124,10 +126,11 @@ echo "==========================================" echo " Done! Status: ${STATUS}" echo "==========================================" echo "" -echo " Agent: http://${IP}:${AGENT_PORT}" +echo " Agent: http://${IP}:${AGENT_PORT}" +echo " Server ID: ${SERVER_ID}" echo "" echo " Commands:" -echo " systemctl status multisync-agent" -echo " journalctl -u multisync-agent -f" -echo " systemctl restart multisync-agent" +echo " systemctl status nexus-agent" +echo " journalctl -u nexus-agent -f" +echo " systemctl restart nexus-agent" echo ""