[#37] 增加sys/swarm_tests(cpu) ；单独构建的node bundle镜像

src/master/app/config.py

@@ -13,6 +13,8 @@ class AppConfig:
     scheduler_interval_seconds: int
     node_id_prefix: str
     auth_mode: str
+    target_prefer_net_cidrs: str
+    target_reachability_check: bool
 
 
 def _get_int_env(name: str, default: int) -> int:
@@ -27,6 +29,12 @@ def _get_int_env(name: str, default: int) -> int:
 
 def load_config() -> AppConfig:
     """读取环境变量生成配置对象，方便统一管理运行参数。"""
+    def _bool_env(name: str, default: bool) -> bool:
+        raw = os.environ.get(name)
+        if raw is None or raw.strip() == "":
+            return default
+        return raw.strip().lower() in ("1", "true", "yes", "on")
+
     return AppConfig(
         db_path=os.environ.get("DB_PATH", "/private/argus/master/db.sqlite3"),
         metric_nodes_json_path=os.environ.get(
@@ -37,4 +45,6 @@ def load_config() -> AppConfig:
         scheduler_interval_seconds=_get_int_env("SCHEDULER_INTERVAL_SECONDS", 30),
         node_id_prefix=os.environ.get("NODE_ID_PREFIX", "A"),
         auth_mode=os.environ.get("AUTH_MODE", "disabled"),
+        target_prefer_net_cidrs=os.environ.get("TARGET_PREFER_NET_CIDRS", "10.0.0.0/8,172.31.0.0/16"),
+        target_reachability_check=_bool_env("TARGET_REACHABILITY_CHECK", False),
     )

src/master/app/scheduler.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import ipaddress
 import logging
+import socket
 import threading
-from typing import Optional
+from typing import Optional, Iterable, Dict, Any, List
 
 from .config import AppConfig
 from .storage import Storage
@@ -34,10 +36,117 @@ class StatusScheduler:
         self._pending_nodes_json.set()
 
     def generate_nodes_json(self) -> None:
+        """根据在线节点生成 Prometheus 抓取目标，优先 overlay IP。
+
+        候选顺序：meta.overlay_ip > hostname A 记录（命中偏好网段）> meta.ip。
+        可选 reachability 检查：TARGET_REACHABILITY_CHECK=true 时，对 9100/9400 做一次 1s TCP 连接测试，
+        选择首个可达的候选；全部失败则按顺序取第一个并记录日志。
+        """
         with self._nodes_json_lock:
-            online_nodes = self._storage.get_online_nodes()
-            atomic_write_json(self._config.metric_nodes_json_path, online_nodes)
-            self._logger.info("nodes.json updated", extra={"count": len(online_nodes)})
+            rows = self._storage.get_online_nodes_meta()
+            prefer_cidrs = self._parse_cidrs(self._config.target_prefer_net_cidrs)
+            reachability = self._config.target_reachability_check
+
+            result: List[Dict[str, Any]] = []
+            for row in rows:
+                meta = row.get("meta", {})
+                hostname = meta.get("hostname") or row.get("name")
+                labels = row.get("labels") or []
+
+                overlay_ip = meta.get("overlay_ip")
+                legacy_ip = meta.get("ip")
+                host_candidates = self._resolve_host_ips(hostname)
+                host_pref = self._pick_by_cidrs(host_candidates, prefer_cidrs)
+
+                candidates: List[str] = []
+                for ip in [overlay_ip, host_pref, legacy_ip]:
+                    if ip and ip not in candidates:
+                        candidates.append(ip)
+
+                chosen = None
+                if reachability:
+                    ports = [9100]
+                    try:
+                        if int(meta.get("gpu_number", 0)) > 0:
+                            ports.append(9400)
+                    except Exception:
+                        pass
+                    for ip in candidates:
+                        if any(self._reachable(ip, p, 1.0) for p in ports):
+                            chosen = ip
+                            break
+                if not chosen:
+                    chosen = candidates[0] if candidates else legacy_ip
+                if not chosen:
+                    # ultimate fallback: 127.0.0.1 (should not happen)
+                    chosen = "127.0.0.1"
+                    self._logger.warning("No candidate IPs for node; falling back", extra={"node": row.get("node_id")})
+
+                if chosen and ipaddress.ip_address(chosen) in ipaddress.ip_network("172.22.0.0/16"):
+                    self._logger.warning(
+                        "Prometheus target uses docker_gwbridge address; prefer overlay",
+                        extra={"node": row.get("node_id"), "ip": chosen},
+                    )
+
+                result.append(
+                    {
+                        "node_id": row.get("node_id"),
+                        "user_id": meta.get("user"),
+                        "ip": chosen,
+                        "hostname": hostname,
+                        "labels": labels if isinstance(labels, list) else [],
+                    }
+                )
+
+            atomic_write_json(self._config.metric_nodes_json_path, result)
+            self._logger.info("nodes.json updated", extra={"count": len(result)})
+
+    # ---------------------------- helpers ----------------------------
+    @staticmethod
+    def _parse_cidrs(raw: str) -> List[ipaddress.IPv4Network]:
+        nets: List[ipaddress.IPv4Network] = []
+        for item in (x.strip() for x in (raw or "").split(",")):
+            if not item:
+                continue
+            try:
+                net = ipaddress.ip_network(item, strict=False)
+                if isinstance(net, ipaddress.IPv4Network):
+                    nets.append(net)
+            except ValueError:
+                continue
+        return nets
+
+    @staticmethod
+    def _resolve_host_ips(hostname: str) -> List[str]:
+        ips: List[str] = []
+        try:
+            infos = socket.getaddrinfo(hostname, None, family=socket.AF_INET)
+            for info in infos:
+                ip = info[4][0]
+                if ip not in ips:
+                    ips.append(ip)
+        except OSError:
+            pass
+        return ips
+
+    @staticmethod
+    def _pick_by_cidrs(candidates: Iterable[str], prefer: List[ipaddress.IPv4Network]) -> str | None:
+        for net in prefer:
+            for ip in candidates:
+                try:
+                    if ipaddress.ip_address(ip) in net:
+                        return ip
+                except ValueError:
+                    continue
+        return None
+
+    @staticmethod
+    def _reachable(ip: str, port: int, timeout: float) -> bool:
+        try:
+            with socket.create_connection((ip, port), timeout=timeout):
+                return True
+        except OSError:
+            return False
 
     # ------------------------------------------------------------------
     # internal loop

src/master/app/storage.py

@@ -324,9 +324,35 @@ class Storage:
                 {
                     "node_id": row["id"],
                     "user_id": meta.get("user"),
-                    "ip": meta.get("ip"),
+                    "ip": meta.get("ip"),  # kept for backward-compat; preferred IP selection handled in scheduler
                     "hostname": meta.get("hostname", row["name"]),
                     "labels": labels if isinstance(labels, list) else [],
                 }
             )
         return result
+
+    def get_online_nodes_meta(self) -> List[Dict[str, Any]]:
+        """返回在线节点的原始 meta 与名称、标签，交由上层选择目标 IP。
+
+        每项包含：{ node_id, name, meta, labels }
+        """
+        with self._lock:
+            cur = self._conn.execute(
+                "SELECT id, name, meta_json, labels_json FROM nodes WHERE status = ? ORDER BY id ASC",
+                ("online",),
+            )
+            rows = cur.fetchall()
+
+        result: List[Dict[str, Any]] = []
+        for row in rows:
+            meta = json.loads(row["meta_json"]) if row["meta_json"] else {}
+            labels = json.loads(row["labels_json"]) if row["labels_json"] else []
+            result.append(
+                {
+                    "node_id": row["id"],
+                    "name": row["name"],
+                    "meta": meta if isinstance(meta, dict) else {},
+                    "labels": labels if isinstance(labels, list) else [],
+                }
+            )
+        return result