live/edge-agent/agent/agent.go

package agent

import (
	"embed"
	"fmt"
	"log"
	"os"
	"time"
)

type Config struct {
	NodeName  string
	OriginURL string
	HlsPort   int
	ApiPort   int
	Subnet    string
	DataDir   string
	Weight    int
}

type Agent struct {
	cfg        Config
	mediamtxFS embed.FS
	mtx        *MediaMTXManager
	hb         *Heartbeat
	monitor    *Monitor
	nodeID     string
	stopCh     chan struct{}
}

func New(cfg Config, mtxFS embed.FS) (*Agent, error) {
	return &Agent{
		cfg:        cfg,
		mediamtxFS: mtxFS,
		nodeID:     generateNodeID(cfg),
		stopCh:     make(chan struct{}),
	}, nil
}

func (a *Agent) Start() error {
	if err := os.MkdirAll(a.cfg.DataDir, 0755); err != nil {
		return fmt.Errorf("create data dir: %w", err)
	}

	a.mtx = NewMediaMTXManager(a.cfg, a.mediamtxFS)
	if err := a.mtx.Extract(); err != nil {
		return fmt.Errorf("extract mediamtx: %w", err)
	}

	if err := a.mtx.WriteConfig(); err != nil {
		return fmt.Errorf("write config: %w", err)
	}

	if err := a.mtx.Start(); err != nil {
		return fmt.Errorf("start mediamtx: %w", err)
	}

	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)
	}

	go a.heartbeatLoop()
	go a.monitorLoop()

	return nil
}

func (a *Agent) Stop() {
	log.Println("Stopping agent...")
	close(a.stopCh)
	if a.mtx != nil {
		a.mtx.Stop()
	}
	log.Println("Agent stopped")
}

func (a *Agent) heartbeatLoop() {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-a.stopCh:
			return
		case <-ticker.C:
			metrics := a.monitor.Snapshot()
			if err := a.hb.Send(metrics); err != nil {
				log.Printf("Heartbeat error: %v", err)
			}
		}
	}
}

func (a *Agent) monitorLoop() {
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-a.stopCh:
			return
		case <-ticker.C:
			a.monitor.Tick()
		}
	}
}

func (a *Agent) NodeID() string  { return a.nodeID }
func (a *Agent) Config() Config  { return a.cfg }
func (a *Agent) Metrics() Metrics {
	if a.monitor != nil {
		return a.monitor.Snapshot()
	}
	return Metrics{}
}

func generateNodeID(cfg Config) string {
	host, _ := os.Hostname()
	if host == "" {
		host = "unknown"
	}
	return fmt.Sprintf("%s-%s", host, cfg.NodeName)
}