负载均衡重构: 子网感知选择 + 节点健康监控 + 节点注册上报API

- 新增 subnet-aware 负载策略: 根据客户端IP匹配最近/16子网节点
- 节点注册端点 (POST /api/nodes/register): 边缘节点自动注册
- 节点心跳端点 (POST /api/nodes/heartbeat): 上报CPU/RAM/带宽/连接数
- 健康汇总 (GET /api/nodes/health): 服务端汇总所有节点状态
- 配置节点新增subnet/is_origin字段
- 前端NodeManager: 实时健康状态/CPU/RAM进度条/带宽/连接数
- X-Forwarded-For/X-Real-IP透传支持客户端IP

backend/config.py

@@ -12,13 +12,20 @@ class Settings(BaseSettings):
     media_origin_hls_port: int = 8888
 
     edge_nodes: list[dict] = [
-        {"id": "edge-1", "name": "Edge Node 1", "host": "127.0.0.1",
-         "hls_port": 8889, "api_port": 9998, "weight": 100, "enabled": True},
-        {"id": "edge-2", "name": "Edge Node 2", "host": "127.0.0.1",
-         "hls_port": 8890, "api_port": 9999, "weight": 50, "enabled": True},
+        {
+            "id": "default",
+            "name": "Origin (Default)",
+            "host": "127.0.0.1",
+            "hls_port": 8888,
+            "api_port": 9997,
+            "weight": 100,
+            "enabled": True,
+            "subnet": "0.0.0.0/0",
+            "is_origin": True,
+        },
     ]
 
-    load_balance_strategy: Literal["round-robin", "weighted", "least-conn"] = "weighted"
+    load_balance_strategy: Literal["round-robin", "weighted", "least-conn", "subnet"] = "subnet"
 
     qualities: list[dict] = [
         {"label": "1080p", "width": 1920, "height": 1080, "bitrate": "5000k"},

backend/routes/nodes.py

@@ -1,5 +1,6 @@
 from fastapi import APIRouter, HTTPException, Depends
 from pydantic import BaseModel
+from typing import Optional
 from config import settings
 from routes.auth import require_auth
 from services.load_balancer import lb
@@ -11,6 +12,7 @@ router = APIRouter(prefix="/api/nodes", tags=["nodes"])
 class NodeUpdate(BaseModel):
     weight: int
     enabled: bool | None = None
+    subnet: str | None = None
 
 
 class StrategyUpdate(BaseModel):
@@ -18,18 +20,42 @@ class StrategyUpdate(BaseModel):
     multicast_enabled: bool | None = None
 
 
+class NodeHeartbeat(BaseModel):
+    node_id: str
+    cpu_percent: float | None = None
+    memory_percent: float | None = None
+    bandwidth_mbps: float | None = None
+    active_connections: int | None = None
+    hls_readers: int | None = None
+
+
+class NodeRegister(BaseModel):
+    node_id: str
+    name: str
+    host: str
+    hls_port: int
+    api_port: int
+    weight: int = 100
+    subnet: str = "0.0.0.0/0"
+
+
 @router.get("")
 async def list_nodes(_: bool = Depends(require_auth)):
     nodes = []
     for n in settings.edge_nodes:
+        nid = n["id"]
         nodes.append({
-            "id": n["id"],
+            "id": nid,
             "name": n["name"],
             "host": n["host"],
             "hls_port": n["hls_port"],
             "api_port": n["api_port"],
             "weight": n["weight"],
             "enabled": n.get("enabled", True),
+            "subnet": n.get("subnet", "0.0.0.0/0"),
+            "is_origin": n.get("is_origin", False),
+            "online": lb.is_online(nid),
+            "metrics": lb._node_health.get(nid),
         })
     return nodes
 
@@ -45,7 +71,7 @@ async def get_strategy(_: bool = Depends(require_auth)):
 
 @router.put("/strategy")
 async def update_strategy(body: StrategyUpdate, _: bool = Depends(require_auth)):
-    if body.strategy in ("round-robin", "weighted", "least-conn"):
+    if body.strategy in ("round-robin", "weighted", "least-conn", "subnet"):
         settings.load_balance_strategy = body.strategy
     if body.multicast_enabled is not None:
         settings.multicast_enabled = body.multicast_enabled
@@ -60,8 +86,54 @@ async def update_node(node_id: str, body: NodeUpdate, _: bool = Depends(require_
             n["weight"] = body.weight
             if body.enabled is not None:
                 n["enabled"] = body.enabled
+            if body.subnet is not None:
+                n["subnet"] = body.subnet
             found = True
             break
     if not found:
         raise HTTPException(status_code=404, detail="Node not found")
     return {"ok": True}
+
+
+@router.post("/register")
+async def register_node(body: NodeRegister):
+    for n in settings.edge_nodes:
+        if n["id"] == body.node_id:
+            n.update({
+                "name": body.name,
+                "host": body.host,
+                "hls_port": body.hls_port,
+                "api_port": body.api_port,
+                "weight": body.weight,
+                "subnet": body.subnet,
+            })
+            return {"ok": True, "action": "updated"}
+    settings.edge_nodes.append({
+        "id": body.node_id,
+        "name": body.name,
+        "host": body.host,
+        "hls_port": body.hls_port,
+        "api_port": body.api_port,
+        "weight": body.weight,
+        "enabled": True,
+        "subnet": body.subnet,
+        "is_origin": False,
+    })
+    return {"ok": True, "action": "registered"}
+
+
+@router.post("/heartbeat")
+async def node_heartbeat(body: NodeHeartbeat):
+    lb.report_health(body.node_id, {
+        "cpu_percent": body.cpu_percent,
+        "memory_percent": body.memory_percent,
+        "bandwidth_mbps": body.bandwidth_mbps,
+        "active_connections": body.active_connections,
+        "hls_readers": body.hls_readers,
+    })
+    return {"ok": True}
+
+
+@router.get("/health")
+async def get_all_health():
+    return lb.get_health_summary()

backend/routes/streams.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, HTTPException, Depends, Query
+from fastapi import APIRouter, HTTPException, Depends, Query, Request
 from sqlalchemy.ext.asyncio import AsyncSession
 from pydantic import BaseModel
 from typing import Optional
@@ -26,6 +26,18 @@ class StreamUpdate(BaseModel):
     scheduled_at: Optional[str] = None
 
 
+def _get_client_ip(request: Request) -> str | None:
+    forwarded = request.headers.get("X-Forwarded-For")
+    if forwarded:
+        return forwarded.split(",")[0].strip()
+    client = request.headers.get("X-Real-IP")
+    if client:
+        return client
+    if request.client:
+        return request.client.host
+    return None
+
+
 @router.get("")
 async def list_streams(db: AsyncSession = Depends(get_db)):
     svc = StreamService(db)
@@ -114,17 +126,19 @@ async def delete_stream(stream_id: int, db: AsyncSession = Depends(get_db),
 
 
 @router.get("/{stream_id}/watch")
-async def get_watch_urls(stream_id: int, db: AsyncSession = Depends(get_db)):
+async def get_watch_urls(stream_id: int, request: Request, db: AsyncSession = Depends(get_db)):
     svc = StreamService(db)
     stream = await svc.get_stream(stream_id)
     if not stream:
         raise HTTPException(status_code=404, detail="Stream not found")
-    urls = await svc.get_watch_urls(stream.stream_key)
+    client_ip = _get_client_ip(request)
+    urls = await svc.get_watch_urls(stream.stream_key, client_ip)
     return {
         "stream_id": stream.id,
         "title": stream.title,
         "status": stream.status.value,
         "multicast_available": urls["multicast_available"],
+        "selected_node": urls.get("selected_node"),
         "lines": urls["lines"],
         "qualities": urls["qualities"],
     }
@@ -153,12 +167,13 @@ async def stream_end(stream_id: int, db: AsyncSession = Depends(get_db),
 
 
 @router.get("/detail/{stream_key}")
-async def get_stream_by_key(stream_key: str, db: AsyncSession = Depends(get_db)):
+async def get_stream_by_key(stream_key: str, request: Request, db: AsyncSession = Depends(get_db)):
     svc = StreamService(db)
     s = await svc.get_stream_by_key(stream_key)
     if not s:
         raise HTTPException(status_code=404, detail="Stream not found")
-    urls = await svc.get_watch_urls(stream_key)
+    client_ip = _get_client_ip(request)
+    urls = await svc.get_watch_urls(stream_key, client_ip)
     return {
         "stream_id": s.id,
         "title": s.title,

backend/services/load_balancer.py

@@ -1,43 +1,140 @@
-import json
-import asyncio
-import subprocess
+import ipaddress
+import time
 from config import settings
 
 
 class LoadBalancer:
     def __init__(self):
         self._rr_index = 0
-        self._conn_counts: dict[str, int] = {n["id"]: 0 for n in settings.edge_nodes}
+        self._conn_counts: dict[str, int] = {}
+        self._node_health: dict[str, dict] = {}
 
     def get_enabled_nodes(self):
         return [n for n in settings.edge_nodes if n.get("enabled", True)]
 
-    def select_nodes(self, strategy: str | None = None):
-        nodes = self.get_enabled_nodes()
-        strategy = strategy or settings.load_balance_strategy
+    def get_node_by_id(self, node_id: str):
+        for n in settings.edge_nodes:
+            if n["id"] == node_id:
+                return n
+        return None
 
+    def select_best_node(self, client_ip: str | None = None):
+        nodes = self.get_enabled_nodes()
+        if not nodes:
+            return None
+
+        if client_ip:
+            best = self._find_subnet_match(client_ip, nodes)
+            if best:
+                return best
+
+        strategy = settings.load_balance_strategy
         if strategy == "round-robin":
-            if not nodes:
-                return []
-            chosen = nodes[self._rr_index % len(nodes)]
+            online = [n for n in nodes if self.is_online(n["id"])]
+            if not online:
+                online = nodes
+            chosen = online[self._rr_index % len(online)]
             self._rr_index += 1
-            return [chosen]
+            return chosen
         elif strategy == "weighted":
-            total = sum(n["weight"] for n in nodes)
-            if total == 0:
-                return nodes
-            import random
-            r = random.uniform(0, total)
-            cumulative = 0
-            for n in nodes:
-                cumulative += n["weight"]
-                if r <= cumulative:
-                    return [n]
-            return [nodes[-1]]
+            return self._weighted_select(nodes)
         elif strategy == "least-conn":
-            sorted_nodes = sorted(nodes, key=lambda n: self._conn_counts.get(n["id"], 0))
-            return [sorted_nodes[0]]
-        return nodes
+            return self._least_conn_select(nodes)
+        else:
+            return self._weighted_select(nodes)
+
+    def _find_subnet_match(self, client_ip: str, nodes: list[dict]):
+        try:
+            ip = ipaddress.ip_address(client_ip)
+        except ValueError:
+            return None
+
+        best = None
+        best_prefix = -1
+
+        for node in nodes:
+            if not self.is_online(node["id"]):
+                continue
+
+            subnet_str = node.get("subnet", "0.0.0.0/0")
+            if not subnet_str:
+                continue
+
+            try:
+                subnet = ipaddress.ip_network(subnet_str, strict=False)
+            except ValueError:
+                continue
+
+            if ip in subnet:
+                if subnet.prefixlen > best_prefix:
+                    best_prefix = subnet.prefixlen
+                    best = node
+
+        return best
+
+    def _weighted_select(self, nodes: list[dict]):
+        online = [n for n in nodes if self.is_online(n["id"])]
+        if not online:
+            online = nodes
+        total = sum(n["weight"] for n in online)
+        if total == 0:
+            return online[0] if online else None
+        import random
+        r = random.uniform(0, total)
+        cumulative = 0
+        for n in online:
+            cumulative += n["weight"]
+            if r <= cumulative:
+                return n
+        return online[-1]
+
+    def _least_conn_select(self, nodes: list[dict]):
+        online = [n for n in nodes if self.is_online(n["id"])]
+        if not online:
+            online = nodes
+        sorted_nodes = sorted(online, key=lambda n: self._conn_counts.get(n["id"], 0))
+        return sorted_nodes[0] if sorted_nodes else None
+
+    def report_health(self, node_id: str, metrics: dict):
+        self._node_health[node_id] = {
+            "last_seen": time.time(),
+            "online": True,
+            **metrics,
+        }
+
+    def is_online(self, node_id: str) -> bool:
+        h = self._node_health.get(node_id)
+        if h is None:
+            if any(n["id"] == node_id and n.get("is_origin") for n in settings.edge_nodes):
+                return True
+            return False
+        if time.time() - h.get("last_seen", 0) > 30:
+            return False
+        return h.get("online", False)
+
+    def get_health_summary(self):
+        result = {}
+        for node in settings.edge_nodes:
+            nid = node["id"]
+            h = self._node_health.get(nid)
+            result[nid] = {
+                "name": node["name"],
+                "host": node["host"],
+                "online": self.is_online(nid),
+                "weight": node["weight"],
+                "enabled": node.get("enabled", True),
+                "subnet": node.get("subnet"),
+                "is_origin": node.get("is_origin", False),
+                "last_seen": h.get("last_seen") if h else None,
+                "metrics": {
+                    "cpu_percent": h.get("cpu_percent"),
+                    "memory_percent": h.get("memory_percent"),
+                    "bandwidth_mbps": h.get("bandwidth_mbps"),
+                    "active_connections": h.get("active_connections"),
+                    "hls_readers": h.get("hls_readers"),
+                } if h else None,
+            }
+        return result
 
     def increment_connection(self, node_id: str):
         self._conn_counts[node_id] = self._conn_counts.get(node_id, 0) + 1

backend/services/stream_service.py

@@ -69,12 +69,6 @@ class StreamService:
                 status=StreamStatus.live,
                 started_at=datetime.utcnow(),
             )
-            assigned = [n["id"] for n in lb.get_enabled_nodes()]
-            await self.update_stream(
-                stream.id,
-                assigned_nodes=json.dumps(assigned),
-                multicast_enabled=multicast_detector.is_available(),
-            )
             rtmp_url = media_service.get_rtmp_url(stream_key)
             transcode_mgr.start_transcode(stream_key, rtmp_url)
 
@@ -89,20 +83,22 @@ class StreamService:
             )
             transcode_mgr.stop_transcode(stream_key)
 
-    async def get_watch_urls(self, stream_key: str):
-        nodes = lb.get_enabled_nodes()
+    async def get_watch_urls(self, stream_key: str, client_ip: str | None = None):
+        best_node = lb.select_best_node(client_ip)
+        if not best_node:
+            best_node = lb.get_enabled_nodes()[0] if lb.get_enabled_nodes() else None
+
         lines = []
-        for node in nodes:
+        if best_node:
+            base_hls = f"/hls/{stream_key}"
             line_info = {
-                "node_id": node["id"],
-                "node_name": node["name"],
-                "weight": node["weight"],
+                "node_id": best_node["id"],
+                "node_name": best_node["name"],
+                "weight": best_node["weight"],
+                "type": "hls",
+                "url": f"{base_hls}/index.m3u8",
                 "qualities": [],
             }
-            base_hls = f"/hls/{stream_key}"
-            line_info["type"] = "hls"
-            line_info["url"] = f"{base_hls}/index.m3u8"
-
             for quality in settings.qualities:
                 label = quality["label"]
                 line_info["qualities"].append({