完善直播节点监控与边缘代理

edge-agent/agent/agent.go

@@ -5,17 +5,19 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"sync"
 	"time"
 )
 
 type Config struct {
-	NodeName  string
-	OriginURL string
-	HlsPort   int
-	ApiPort   int
-	Subnet    string
-	DataDir   string
-	Weight    int
+	NodeName     string
+	OriginURL    string
+	DiscoveryURL string
+	HlsPort      int
+	ApiPort      int
+	Subnet       string
+	DataDir      string
+	Weight       int
 }
 
 type Agent struct {
@@ -26,6 +28,8 @@ type Agent struct {
 	monitor    *Monitor
 	nodeID     string
 	stopCh     chan struct{}
+	connected  bool
+	mu         sync.RWMutex
 }
 
 func New(cfg Config, mtxFS embed.FS) (*Agent, error) {
@@ -42,6 +46,13 @@ func (a *Agent) Start() error {
 		return fmt.Errorf("create data dir: %w", err)
 	}
 
+	a.hb = NewHeartbeat(a.cfg, a.nodeID)
+	if err := a.hb.Discover(); err != nil {
+		log.Printf("Warning: initial discovery failed: %v (will retry every 10 seconds)", err)
+	} else {
+		a.cfg.OriginURL = a.hb.OriginURL()
+	}
+
 	a.mtx = NewMediaMTXManager(a.cfg, a.mediamtxFS)
 	if err := a.mtx.Extract(); err != nil {
 		return fmt.Errorf("extract mediamtx: %w", err)
@@ -56,12 +67,10 @@ func (a *Agent) Start() error {
 	}
 
 	a.monitor = NewMonitor()
-	a.hb = NewHeartbeat(a.cfg, a.nodeID)
-
 	time.Sleep(2 * time.Second)
 
-	if err := a.hb.Register(); err != nil {
-		log.Printf("Warning: registration failed: %v (will retry)", err)
+	if err := a.connect(); err != nil {
+		log.Printf("Warning: connection failed: %v (will retry every 10 seconds)", err)
 	}
 
 	go a.heartbeatLoop()
@@ -87,8 +96,15 @@ func (a *Agent) heartbeatLoop() {
 		case <-a.stopCh:
 			return
 		case <-ticker.C:
+			if !a.Connected() {
+				if err := a.connect(); err != nil {
+					log.Printf("Connection retry failed: %v", err)
+					continue
+				}
+			}
 			metrics := a.monitor.Snapshot()
 			if err := a.hb.Send(metrics); err != nil {
+				a.setConnected(false)
 				log.Printf("Heartbeat error: %v", err)
 			}
 		}
@@ -104,12 +120,37 @@ func (a *Agent) monitorLoop() {
 			return
 		case <-ticker.C:
 			a.monitor.Tick()
+			if a.mtx != nil {
+				a.monitor.SetMediaMetrics(a.mtx.Metrics())
+			}
 		}
 	}
 }
 
-func (a *Agent) NodeID() string  { return a.nodeID }
-func (a *Agent) Config() Config  { return a.cfg }
+func (a *Agent) NodeID() string { return a.nodeID }
+func (a *Agent) Config() Config { return a.cfg }
+func (a *Agent) Connected() bool {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	return a.connected
+}
+
+func (a *Agent) setConnected(connected bool) {
+	a.mu.Lock()
+	a.connected = connected
+	a.mu.Unlock()
+}
+
+func (a *Agent) connect() error {
+	if err := a.hb.Discover(); err != nil {
+		return err
+	}
+	if err := a.hb.Register(); err != nil {
+		return err
+	}
+	a.setConnected(true)
+	return nil
+}
 func (a *Agent) Metrics() Metrics {
 	if a.monitor != nil {
 		return a.monitor.Snapshot()

edge-agent/agent/heartbeat.go

@@ -4,7 +4,9 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
+	"strings"
 	"time"
 )
 
@@ -21,6 +23,12 @@ type HeartbeatPayload struct {
 	BandwidthMbps     float64 `json:"bandwidth_mbps"`
 	ActiveConnections int     `json:"active_connections"`
 	HlsReaders        int     `json:"hls_readers"`
+	NetworkInMbps     float64 `json:"network_in_mbps"`
+	NetworkOutMbps    float64 `json:"network_out_mbps"`
+	ViewerCount       int     `json:"viewer_count"`
+	FrameDrops        int     `json:"frame_drop_indicators"`
+	PacketLoss        int     `json:"packet_loss_indicators"`
+	MediaAvailable    bool    `json:"media_metrics_available"`
 }
 
 type RegisterPayload struct {
@@ -54,6 +62,37 @@ func (h *Heartbeat) Register() error {
 	return h.post("/api/nodes/register", payload)
 }
 
+func (h *Heartbeat) Discover() error {
+	if h.cfg.DiscoveryURL == "" {
+		return nil
+	}
+	resp, err := h.client.Get(h.cfg.DiscoveryURL)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode >= 400 {
+		return fmt.Errorf("discovery status %d", resp.StatusCode)
+	}
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 4096))
+	if err != nil {
+		return err
+	}
+	host := strings.TrimSpace(string(body))
+	if host == "" {
+		return fmt.Errorf("discovery returned an empty address")
+	}
+	if !strings.HasPrefix(host, "http://") && !strings.HasPrefix(host, "https://") {
+		host = "http://" + host
+	}
+	h.cfg.OriginURL = strings.TrimRight(host, "/")
+	return nil
+}
+
+func (h *Heartbeat) OriginURL() string {
+	return h.cfg.OriginURL
+}
+
 func (h *Heartbeat) Send(m Metrics) error {
 	payload := HeartbeatPayload{
 		NodeID:            h.nodeID,
@@ -62,6 +101,12 @@ func (h *Heartbeat) Send(m Metrics) error {
 		BandwidthMbps:     m.Bandwidth,
 		ActiveConnections: m.ActiveConns,
 		HlsReaders:        m.HlsReaders,
+		NetworkInMbps:     m.NetworkIn,
+		NetworkOutMbps:    m.NetworkOut,
+		ViewerCount:       m.ViewerCount,
+		FrameDrops:        m.FrameDrops,
+		PacketLoss:        m.PacketLoss,
+		MediaAvailable:    m.MediaAvailable,
 	}
 	return h.post("/api/nodes/heartbeat", payload)
 }

edge-agent/agent/mediamtx.go

@@ -1,6 +1,7 @@
 package agent
 
 import (
+	"bufio"
 	"embed"
 	"fmt"
 	"io"
@@ -9,6 +10,7 @@ import (
 	"path/filepath"
 	"runtime"
 	"strings"
+	"sync"
 )
 
 type MediaMTXManager struct {
@@ -17,6 +19,15 @@ type MediaMTXManager struct {
 	binPath    string
 	configPath string
 	cmd        *exec.Cmd
+	mu         sync.RWMutex
+	metrics    MediaMetrics
+}
+
+type MediaMetrics struct {
+	ViewerCount int
+	FrameDrops  int
+	PacketLoss  int
+	Available   bool
 }
 
 func NewMediaMTXManager(cfg Config, fs embed.FS) *MediaMTXManager {
@@ -116,14 +127,57 @@ paths:
 
 func (m *MediaMTXManager) Start() error {
 	m.cmd = exec.Command(m.binPath, m.configPath)
-	m.cmd.Stdout = os.Stdout
-	m.cmd.Stderr = os.Stderr
+	stdout, err := m.cmd.StdoutPipe()
+	if err != nil {
+		return fmt.Errorf("capture mediamtx stdout: %w", err)
+	}
+	stderr, err := m.cmd.StderrPipe()
+	if err != nil {
+		return fmt.Errorf("capture mediamtx stderr: %w", err)
+	}
 	if err := m.cmd.Start(); err != nil {
 		return fmt.Errorf("start mediamtx: %w", err)
 	}
+	go m.parseLogs(stdout)
+	go m.parseLogs(stderr)
 	return nil
 }
 
+func (m *MediaMTXManager) parseLogs(reader io.Reader) {
+	scanner := bufio.NewScanner(reader)
+	for scanner.Scan() {
+		line := scanner.Text()
+		fmt.Fprintln(os.Stdout, line)
+		lower := strings.ToLower(line)
+		m.mu.Lock()
+		if strings.Contains(lower, "reader opened") || strings.Contains(lower, "hls client connected") {
+			m.metrics.Available = true
+			m.metrics.ViewerCount++
+		}
+		if strings.Contains(lower, "reader closed") || strings.Contains(lower, "hls client disconnected") {
+			m.metrics.Available = true
+			if m.metrics.ViewerCount > 0 {
+				m.metrics.ViewerCount--
+			}
+		}
+		if strings.Contains(lower, "frame drop") || strings.Contains(lower, "dropped frame") {
+			m.metrics.Available = true
+			m.metrics.FrameDrops++
+		}
+		if strings.Contains(lower, "packet loss") || strings.Contains(lower, "lost packet") {
+			m.metrics.Available = true
+			m.metrics.PacketLoss++
+		}
+		m.mu.Unlock()
+	}
+}
+
+func (m *MediaMTXManager) Metrics() MediaMetrics {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.metrics
+}
+
 func (m *MediaMTXManager) Stop() {
 	if m.cmd != nil && m.cmd.Process != nil {
 		m.cmd.Process.Kill()

edge-agent/agent/monitor.go

@@ -11,11 +11,17 @@ import (
 )
 
 type Metrics struct {
-	CPU        float64
-	Memory     float64
-	Bandwidth  float64
-	ActiveConns int
-	HlsReaders int
+	CPU            float64
+	Memory         float64
+	Bandwidth      float64
+	NetworkIn      float64
+	NetworkOut     float64
+	ActiveConns    int
+	HlsReaders     int
+	ViewerCount    int
+	FrameDrops     int
+	PacketLoss     int
+	MediaAvailable bool
 }
 
 type Monitor struct {
@@ -54,8 +60,9 @@ func (m *Monitor) Tick() {
 		} else {
 			elapsed := now.Sub(m.prevNetTime).Seconds()
 			if elapsed > 0 {
-				bytesTotal := (curIn - m.prevNetIn) + (curOut - m.prevNetOut)
-				metrics.Bandwidth = float64(bytesTotal) * 8 / elapsed / 1e6
+				metrics.NetworkIn = float64(curIn-m.prevNetIn) * 8 / elapsed / 1e6
+				metrics.NetworkOut = float64(curOut-m.prevNetOut) * 8 / elapsed / 1e6
+				metrics.Bandwidth = metrics.NetworkIn + metrics.NetworkOut
 			}
 			m.prevNetIn = curIn
 			m.prevNetOut = curOut
@@ -81,3 +88,13 @@ func (m *Monitor) Snapshot() Metrics {
 	defer m.mu.RUnlock()
 	return m.current
 }
+
+func (m *Monitor) SetMediaMetrics(media MediaMetrics) {
+	m.mu.Lock()
+	m.current.ViewerCount = media.ViewerCount
+	m.current.HlsReaders = media.ViewerCount
+	m.current.FrameDrops = media.FrameDrops
+	m.current.PacketLoss = media.PacketLoss
+	m.current.MediaAvailable = media.Available
+	m.mu.Unlock()
+}