diff --git a/backend/config.py b/backend/config.py index 88c287b..23657fe 100644 --- a/backend/config.py +++ b/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"}, diff --git a/backend/routes/nodes.py b/backend/routes/nodes.py index 123926a..7edf5bb 100644 --- a/backend/routes/nodes.py +++ b/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() diff --git a/backend/routes/streams.py b/backend/routes/streams.py index ec3cb1d..ce761e0 100644 --- a/backend/routes/streams.py +++ b/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, diff --git a/backend/services/load_balancer.py b/backend/services/load_balancer.py index 411b56d..c6bfc2f 100644 --- a/backend/services/load_balancer.py +++ b/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 diff --git a/backend/services/stream_service.py b/backend/services/stream_service.py index ff7c8cb..f1607e9 100644 --- a/backend/services/stream_service.py +++ b/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({ diff --git a/frontend/src/admin/pages/NodeManager.tsx b/frontend/src/admin/pages/NodeManager.tsx index ac1724a..7b31003 100644 --- a/frontend/src/admin/pages/NodeManager.tsx +++ b/frontend/src/admin/pages/NodeManager.tsx @@ -1,12 +1,20 @@ import { useState, useEffect } from 'react' import { - Card, Table, InputNumber, Switch, Typography, message, Space, Tag, Statistic, Row, Col, Button, + Card, Table, InputNumber, Switch, Typography, message, Space, Tag, Statistic, Row, Col, Button, Tooltip, Progress, } from 'antd' -import { CheckCircleOutlined, CloseCircleOutlined, ReloadOutlined } from '@ant-design/icons' +import { CheckCircleOutlined, CloseCircleOutlined, ReloadOutlined, DashboardOutlined } from '@ant-design/icons' import api from '../../api/client' const { Title, Text } = Typography +interface NodeMetrics { + cpu_percent: number | null + memory_percent: number | null + bandwidth_mbps: number | null + active_connections: number | null + hls_readers: number | null +} + interface NodeInfo { id: string name: string @@ -15,6 +23,10 @@ interface NodeInfo { api_port: number weight: number enabled: boolean + subnet: string + is_origin: boolean + online: boolean + metrics: NodeMetrics | null } interface StrategyInfo { @@ -24,6 +36,7 @@ interface StrategyInfo { } const strategyLabels: Record = { + 'subnet': 'Subnet-Aware', 'round-robin': 'Round-Robin', 'weighted': 'Weighted', 'least-conn': 'Least-Conn', @@ -34,7 +47,7 @@ export default function NodeManager() { const [strategy, setStrategy] = useState(null) const [loading, setLoading] = useState(true) - useEffect(() => { loadData() }, []) + useEffect(() => { loadData(); const t = setInterval(loadData, 10000); return () => clearInterval(t) }, []) async function loadData() { setLoading(true) @@ -66,42 +79,72 @@ export default function NodeManager() { } catch { message.error('Failed') } } + const onlineNodes = nodes.filter((n) => n.online).length + const columns = [ + { + title: 'Status', dataIndex: 'online', key: 'online', width: 80, + render: (online: boolean) => ( + + : } + color={online ? 'success' : 'error'} /> + + ), + }, { title: 'Name', dataIndex: 'name', key: 'name' }, - { title: 'Host', dataIndex: 'host', key: 'host' }, - { title: 'HLS Port', dataIndex: 'hls_port', key: 'hls_port', width: 100 }, - { title: 'API Port', dataIndex: 'api_port', key: 'api_port', width: 100 }, + { title: 'Host', dataIndex: 'host', key: 'host', width: 140 }, + { + title: 'Subnet', dataIndex: 'subnet', key: 'subnet', width: 140, + render: (v: string) => {v}, + }, + { + title: 'CPU', key: 'cpu', width: 80, + render: (_: any, r: NodeInfo) => r.metrics?.cpu_percent != null + ? + : --, + }, + { + title: 'RAM', key: 'ram', width: 80, + render: (_: any, r: NodeInfo) => r.metrics?.memory_percent != null + ? + : --, + }, + { + title: 'BW', key: 'bw', width: 70, + render: (_: any, r: NodeInfo) => r.metrics?.bandwidth_mbps != null + ? {r.metrics.bandwidth_mbps.toFixed(1)} Mbps + : --, + }, + { + title: 'Conn', dataIndex: 'connections', key: 'conn', width: 60, + render: (_: any, r: NodeInfo) => r.metrics?.active_connections != null + ? r.metrics.active_connections + : --, + }, { title: 'Weight', dataIndex: 'weight', key: 'weight', width: 140, render: (_: number, record: NodeInfo) => ( { const v = val || 0 setNodes((prev) => prev.map((n) => (n.id === record.id ? { ...n, weight: v } : n))) updateWeight(record.id, v) }} - size="small" - style={{ width: 90 }} + size="small" style={{ width: 80 }} + disabled={record.is_origin} /> ), }, { - title: 'Enabled', dataIndex: 'enabled', key: 'enabled', width: 80, + title: 'Enabled', dataIndex: 'enabled', key: 'enabled', width: 70, render: (_: any, record: NodeInfo) => ( toggleNode(record.id, val)} - size="small" + checked={record.enabled} onChange={(val) => toggleNode(record.id, val)} + size="small" disabled={record.is_origin} /> ), }, - { - title: 'Health', key: 'health', width: 100, - render: () => } color="success">Online, - }, ] return ( @@ -110,26 +153,26 @@ export default function NodeManager() { {strategy && ( - + - + - + - + - + - + + + + + + 0 ? '#2E7D32' : '#C62828' }} /> @@ -146,7 +189,8 @@ export default function NodeManager() { rowKey="id" loading={loading} pagination={false} - size="middle" + size="small" + scroll={{ x: 1100 }} /> diff --git a/frontend/src/admin/pages/Settings.tsx b/frontend/src/admin/pages/Settings.tsx index 59d1d46..bad1953 100644 --- a/frontend/src/admin/pages/Settings.tsx +++ b/frontend/src/admin/pages/Settings.tsx @@ -7,8 +7,9 @@ import api from '../../api/client' const { Title } = Typography const strategies = [ - { value: 'round-robin', label: 'Round-Robin' }, + { value: 'subnet', label: 'Subnet-Aware (Recommended)' }, { value: 'weighted', label: 'Weighted' }, + { value: 'round-robin', label: 'Round-Robin' }, { value: 'least-conn', label: 'Least Connections' }, ]