diff --git a/.gitignore b/.gitignore index 80e493e..e80cb53 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,8 @@ venv/ mediamtx/mediamtx.tar.gz mediamtx/mediamtx package-lock.json + +# edge-agent +edge-agent/dist/ +edge-agent/mediamtx/ + diff --git a/edge-agent/agent/agent.go b/edge-agent/agent/agent.go new file mode 100644 index 0000000..cafafe3 --- /dev/null +++ b/edge-agent/agent/agent.go @@ -0,0 +1,126 @@ +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) +} diff --git a/edge-agent/agent/heartbeat.go b/edge-agent/agent/heartbeat.go new file mode 100644 index 0000000..c84c102 --- /dev/null +++ b/edge-agent/agent/heartbeat.go @@ -0,0 +1,84 @@ +package agent + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "time" +) + +type Heartbeat struct { + cfg Config + nodeID string + client *http.Client +} + +type HeartbeatPayload struct { + NodeID string `json:"node_id"` + CPUPct float64 `json:"cpu_percent"` + MemPct float64 `json:"memory_percent"` + BandwidthMbps float64 `json:"bandwidth_mbps"` + ActiveConnections int `json:"active_connections"` + HlsReaders int `json:"hls_readers"` +} + +type RegisterPayload struct { + NodeID string `json:"node_id"` + Name string `json:"name"` + Host string `json:"host"` + HlsPort int `json:"hls_port"` + ApiPort int `json:"api_port"` + Weight int `json:"weight"` + Subnet string `json:"subnet"` +} + +func NewHeartbeat(cfg Config, nodeID string) *Heartbeat { + return &Heartbeat{ + cfg: cfg, + nodeID: nodeID, + client: &http.Client{Timeout: 10 * time.Second}, + } +} + +func (h *Heartbeat) Register() error { + payload := RegisterPayload{ + NodeID: h.nodeID, + Name: h.cfg.NodeName, + Host: h.cfg.OriginURL, + HlsPort: h.cfg.HlsPort, + ApiPort: h.cfg.ApiPort, + Weight: h.cfg.Weight, + Subnet: h.cfg.Subnet, + } + return h.post("/api/nodes/register", payload) +} + +func (h *Heartbeat) Send(m Metrics) error { + payload := HeartbeatPayload{ + NodeID: h.nodeID, + CPUPct: m.CPU, + MemPct: m.Memory, + BandwidthMbps: m.Bandwidth, + ActiveConnections: m.ActiveConns, + HlsReaders: m.HlsReaders, + } + return h.post("/api/nodes/heartbeat", payload) +} + +func (h *Heartbeat) post(path string, payload interface{}) error { + body, err := json.Marshal(payload) + if err != nil { + return err + } + url := h.cfg.OriginURL + path + resp, err := h.client.Post(url, "application/json", bytes.NewReader(body)) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + return fmt.Errorf("status %d", resp.StatusCode) + } + return nil +} diff --git a/edge-agent/agent/mediamtx.go b/edge-agent/agent/mediamtx.go new file mode 100644 index 0000000..3b750f5 --- /dev/null +++ b/edge-agent/agent/mediamtx.go @@ -0,0 +1,132 @@ +package agent + +import ( + "embed" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" +) + +type MediaMTXManager struct { + cfg Config + fs embed.FS + binPath string + configPath string + cmd *exec.Cmd +} + +func NewMediaMTXManager(cfg Config, fs embed.FS) *MediaMTXManager { + return &MediaMTXManager{ + cfg: cfg, + fs: fs, + binPath: filepath.Join(cfg.DataDir, "mediamtx"), + configPath: filepath.Join(cfg.DataDir, "mediamtx.yml"), + } +} + +func (m *MediaMTXManager) Extract() error { + if _, err := os.Stat(m.binPath); err == nil { + return nil + } + + srcPath := "mediamtx/" + if runtime.GOOS == "windows" { + srcPath += "mediamtx.exe" + } else { + srcPath += "mediamtx" + } + + entries, err := m.fs.ReadDir("mediamtx") + if err != nil { + return fmt.Errorf("read embedded dir: %w", err) + } + + for _, entry := range entries { + if entry.IsDir() { + continue + } + name := entry.Name() + if runtime.GOOS == "windows" && !strings.Contains(name, ".exe") { + continue + } + if runtime.GOOS != "windows" && strings.Contains(name, ".exe") { + continue + } + + src, err := m.fs.Open("mediamtx/" + name) + if err != nil { + return fmt.Errorf("open embedded %s: %w", name, err) + } + defer src.Close() + + destName := "mediamtx" + if runtime.GOOS == "windows" { + destName = "mediamtx.exe" + } + destPath := filepath.Join(m.cfg.DataDir, destName) + dest, err := os.OpenFile(destPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0755) + if err != nil { + src.Close() + return fmt.Errorf("create %s: %w", destPath, err) + } + + if _, err := io.Copy(dest, src); err != nil { + dest.Close() + src.Close() + return fmt.Errorf("copy: %w", err) + } + dest.Close() + src.Close() + m.binPath = destPath + break + } + + return nil +} + +func (m *MediaMTXManager) WriteConfig() error { + config := fmt.Sprintf(`logLevel: info +logDestinations: [stdout] + +rtmp: no +rtsp: no + +hls: yes +hlsAddress: :%d + +api: yes +apiAddress: :%d + +paths: + all: + source: rtmp://%s:1935/live/$MTX_PATH + sourceOnDemand: yes +`, + m.cfg.HlsPort, + m.cfg.ApiPort, + strings.TrimPrefix(strings.TrimPrefix(m.cfg.OriginURL, "http://"), "https://"), + ) + + return os.WriteFile(m.configPath, []byte(config), 0644) +} + +func (m *MediaMTXManager) Start() error { + m.cmd = exec.Command(m.binPath, m.configPath) + m.cmd.Stdout = os.Stdout + m.cmd.Stderr = os.Stderr + if err := m.cmd.Start(); err != nil { + return fmt.Errorf("start mediamtx: %w", err) + } + return nil +} + +func (m *MediaMTXManager) Stop() { + if m.cmd != nil && m.cmd.Process != nil { + m.cmd.Process.Kill() + m.cmd.Wait() + } +} diff --git a/edge-agent/agent/monitor.go b/edge-agent/agent/monitor.go new file mode 100644 index 0000000..a4b0000 --- /dev/null +++ b/edge-agent/agent/monitor.go @@ -0,0 +1,83 @@ +package agent + +import ( + "log" + "sync" + "time" + + "github.com/shirou/gopsutil/v3/cpu" + "github.com/shirou/gopsutil/v3/mem" + "github.com/shirou/gopsutil/v3/net" +) + +type Metrics struct { + CPU float64 + Memory float64 + Bandwidth float64 + ActiveConns int + HlsReaders int +} + +type Monitor struct { + mu sync.RWMutex + current Metrics + prevNetIn uint64 + prevNetOut uint64 + prevNetTime time.Time +} + +func NewMonitor() *Monitor { + return &Monitor{} +} + +func (m *Monitor) Tick() { + metrics := Metrics{} + + if pct, err := cpu.Percent(0, false); err == nil && len(pct) > 0 { + metrics.CPU = pct[0] + } + + if v, err := mem.VirtualMemory(); err == nil { + metrics.Memory = v.UsedPercent + } + + ioCounters, err := net.IOCounters(false) + if err == nil && len(ioCounters) > 0 { + now := time.Now() + curIn := ioCounters[0].BytesRecv + curOut := ioCounters[0].BytesSent + + if m.prevNetTime.IsZero() { + m.prevNetIn = curIn + m.prevNetOut = curOut + m.prevNetTime = now + } else { + elapsed := now.Sub(m.prevNetTime).Seconds() + if elapsed > 0 { + bytesTotal := (curIn - m.prevNetIn) + (curOut - m.prevNetOut) + metrics.Bandwidth = float64(bytesTotal) * 8 / elapsed / 1e6 + } + m.prevNetIn = curIn + m.prevNetOut = curOut + m.prevNetTime = now + } + } + + conns, _ := net.Connections("tcp") + metrics.ActiveConns = len(conns) + + metrics.HlsReaders = 0 + + m.mu.Lock() + m.current = metrics + m.mu.Unlock() + + log.Printf("Metrics: CPU=%.1f%% MEM=%.1f%% BW=%.1fMbps CONN=%d", + metrics.CPU, metrics.Memory, metrics.Bandwidth, metrics.ActiveConns) +} + +func (m *Monitor) Snapshot() Metrics { + m.mu.RLock() + defer m.mu.RUnlock() + return m.current +} diff --git a/edge-agent/build.ps1 b/edge-agent/build.ps1 new file mode 100644 index 0000000..99f1875 --- /dev/null +++ b/edge-agent/build.ps1 @@ -0,0 +1,32 @@ +# Build StreamCore Edge Agent for Windows +# Prerequisites: Go 1.19+ + +param( + [string]$Platform = "windows" +) + +$VERSION = if ($env:VERSION) { $env:VERSION } else { "0.1.0" } +$OUTDIR = "dist" +$APP = "streamcore-edge" + +$env:GO111MODULE = "on" +$env:GOPROXY = if ($env:GOPROXY) { $env:GOPROXY } else { "https://goproxy.cn,direct" } + +if (-not (Test-Path $OUTDIR)) { + New-Item -ItemType Directory -Path $OUTDIR | Out-Null +} + +if ($Platform -eq "windows") { + Write-Host "Building Windows (amd64)..." + $env:GOOS = "windows" + $env:GOARCH = "amd64" + $env:CGO_ENABLED = "0" + + go build -ldflags "-s -w -H windowsgui -X main.Version=$VERSION" -o "$OUTDIR/$APP-windows-amd64.exe" . + Write-Host " -> $OUTDIR/$APP-windows-amd64.exe" +} +else { + Write-Host "Usage: .\build.ps1 [windows]" +} + +Write-Host "Done." diff --git a/edge-agent/build.sh b/edge-agent/build.sh new file mode 100755 index 0000000..a67d627 --- /dev/null +++ b/edge-agent/build.sh @@ -0,0 +1,55 @@ +#!/bin/bash +# Build StreamCore Edge Agent +# Prerequisites: Go 1.19+, mediamtx binary in mediamtx/ +# +# Build for Linux: +# ./build.sh linux +# Build for Windows: +# ./build.sh windows +# Build both: +# ./build.sh all + +set -e + +PLATFORM="${1:-linux}" +VERSION="${VERSION:-0.1.0}" +OUTDIR="dist" +APP="streamcore-edge" + +export GO111MODULE=on +export GOPROXY="${GOPROXY:-https://goproxy.cn,direct}" + +build_linux() { + echo "Building Linux (amd64)..." + GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build \ + -ldflags="-s -w -X main.Version=${VERSION}" \ + -o "${OUTDIR}/${APP}-linux-amd64" \ + . + echo " -> ${OUTDIR}/${APP}-linux-amd64" +} + +build_windows() { + echo "Building Windows (amd64)..." + GOOS=windows GOARCH=amd64 CGO_ENABLED=0 go build \ + -ldflags="-s -w -H windowsgui -X main.Version=${VERSION}" \ + -o "${OUTDIR}/${APP}-windows-amd64.exe" \ + . + echo " -> ${OUTDIR}/${APP}-windows-amd64.exe" +} + +mkdir -p "${OUTDIR}" + +case "${PLATFORM}" in + linux) build_linux ;; + windows) build_windows ;; + all) + build_linux + build_windows + ;; + *) + echo "Usage: $0 {linux|windows|all}" + exit 1 + ;; +esac + +echo "Done." diff --git a/edge-agent/go.mod b/edge-agent/go.mod new file mode 100644 index 0000000..27b6cf4 --- /dev/null +++ b/edge-agent/go.mod @@ -0,0 +1,27 @@ +module streamcore-edge + +go 1.19 + +require ( + github.com/getlantern/systray v1.2.2 + github.com/shirou/gopsutil/v3 v3.23.7 +) + +require ( + github.com/getlantern/context v0.0.0-20190109183933-c447772a6520 // indirect + github.com/getlantern/errors v0.0.0-20190325191628-abdb3e3e36f7 // indirect + github.com/getlantern/golog v0.0.0-20190830074920-4ef2e798c2d7 // indirect + github.com/getlantern/hex v0.0.0-20190417191902-c6586a6fe0b7 // indirect + github.com/getlantern/hidden v0.0.0-20190325191715-f02dbb02be55 // indirect + github.com/getlantern/ops v0.0.0-20190325191751-d70cb0d6f85f // indirect + github.com/go-ole/go-ole v1.2.6 // indirect + github.com/go-stack/stack v1.8.0 // indirect + github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect + github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c // indirect + github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect + github.com/shoenig/go-m1cpu v0.1.6 // indirect + github.com/tklauser/go-sysconf v0.3.11 // indirect + github.com/tklauser/numcpus v0.6.0 // indirect + github.com/yusufpapurcu/wmi v1.2.3 // indirect + golang.org/x/sys v0.10.0 // indirect +) diff --git a/edge-agent/go.sum b/edge-agent/go.sum new file mode 100644 index 0000000..d3e5879 --- /dev/null +++ b/edge-agent/go.sum @@ -0,0 +1,68 @@ +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/getlantern/context v0.0.0-20190109183933-c447772a6520 h1:NRUJuo3v3WGC/g5YiyF790gut6oQr5f3FBI88Wv0dx4= +github.com/getlantern/context v0.0.0-20190109183933-c447772a6520/go.mod h1:L+mq6/vvYHKjCX2oez0CgEAJmbq1fbb/oNJIWQkBybY= +github.com/getlantern/errors v0.0.0-20190325191628-abdb3e3e36f7 h1:6uJ+sZ/e03gkbqZ0kUG6mfKoqDb4XMAzMIwlajq19So= +github.com/getlantern/errors v0.0.0-20190325191628-abdb3e3e36f7/go.mod h1:l+xpFBrCtDLpK9qNjxs+cHU6+BAdlBaxHqikB6Lku3A= +github.com/getlantern/golog v0.0.0-20190830074920-4ef2e798c2d7 h1:guBYzEaLz0Vfc/jv0czrr2z7qyzTOGC9hiQ0VC+hKjk= +github.com/getlantern/golog v0.0.0-20190830074920-4ef2e798c2d7/go.mod h1:zx/1xUUeYPy3Pcmet8OSXLbF47l+3y6hIPpyLWoR9oc= +github.com/getlantern/hex v0.0.0-20190417191902-c6586a6fe0b7 h1:micT5vkcr9tOVk1FiH8SWKID8ultN44Z+yzd2y/Vyb0= +github.com/getlantern/hex v0.0.0-20190417191902-c6586a6fe0b7/go.mod h1:dD3CgOrwlzca8ed61CsZouQS5h5jIzkK9ZWrTcf0s+o= +github.com/getlantern/hidden v0.0.0-20190325191715-f02dbb02be55 h1:XYzSdCbkzOC0FDNrgJqGRo8PCMFOBFL9py72DRs7bmc= +github.com/getlantern/hidden v0.0.0-20190325191715-f02dbb02be55/go.mod h1:6mmzY2kW1TOOrVy+r41Za2MxXM+hhqTtY3oBKd2AgFA= +github.com/getlantern/ops v0.0.0-20190325191751-d70cb0d6f85f h1:wrYrQttPS8FHIRSlsrcuKazukx/xqO/PpLZzZXsF+EA= +github.com/getlantern/ops v0.0.0-20190325191751-d70cb0d6f85f/go.mod h1:D5ao98qkA6pxftxoqzibIBBrLSUli+kYnJqrgBf9cIA= +github.com/getlantern/systray v1.2.2 h1:dCEHtfmvkJG7HZ8lS/sLklTH4RKUcIsKrAD9sThoEBE= +github.com/getlantern/systray v1.2.2/go.mod h1:pXFOI1wwqwYXEhLPm9ZGjS2u/vVELeIgNMY5HvhHhcE= +github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= +github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= +github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk= +github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= +github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38= +github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= +github.com/lxn/walk v0.0.0-20210112085537-c389da54e794/go.mod h1:E23UucZGqpuUANJooIbHWCufXvOcT6E7Stq81gU+CSQ= +github.com/lxn/win v0.0.0-20210218163916-a377121e959e/go.mod h1:KxxjdtRkfNoYDCUP5ryK7XJJNTnpC8atvtmTheChOtk= +github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c h1:rp5dCmg/yLR3mgFuSOe4oEnDDmGLROTvMragMUXpTQw= +github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c/go.mod h1:X07ZCGwUbLaax7L0S3Tw4hpejzu63ZrrQiUe6W0hcy0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw= +github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/shirou/gopsutil/v3 v3.23.7 h1:C+fHO8hfIppoJ1WdsVm1RoI0RwXoNdfTK7yWXV0wVj4= +github.com/shirou/gopsutil/v3 v3.23.7/go.mod h1:c4gnmoRC0hQuaLqvxnx1//VXQ0Ms/X9UnJF8pddY5z4= +github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM= +github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ= +github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU= +github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k= +github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966/go.mod h1:sUM3LWHvSMaG192sy56D9F7CNvL7jUJVXoqM1QKLnog= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/tklauser/go-sysconf v0.3.11 h1:89WgdJhk5SNwJfu+GKyYveZ4IaJ7xAkecBo+KdJV0CM= +github.com/tklauser/go-sysconf v0.3.11/go.mod h1:GqXfhXY3kiPa0nAXPDIQIWzJbMCB7AmcWpGR8lSZfqI= +github.com/tklauser/numcpus v0.6.0 h1:kebhY2Qt+3U6RNK7UqpYNA+tJ23IBEGKkB7JQBfDYms= +github.com/tklauser/numcpus v0.6.0/go.mod h1:FEZLMke0lhOUG6w2JadTzp0a+Nl8PF/GFkQ5UVIcaL4= +github.com/yusufpapurcu/wmi v1.2.3 h1:E1ctvB7uKFMOJw3fdOW32DwGE9I7t++CRUEMKvFoFiw= +github.com/yusufpapurcu/wmi v1.2.3/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201018230417-eeed37f84f13/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.10.0 h1:SqMFp9UcQJZa+pmYuAKjd9xq1f0j5rLcDIk0mj4qAsA= +golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/Knetic/govaluate.v3 v3.0.0/go.mod h1:csKLBORsPbafmSCGTEh3U7Ozmsuq8ZSIlKk1bcqph0E= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/edge-agent/icon.go b/edge-agent/icon.go new file mode 100644 index 0000000..7cb3235 --- /dev/null +++ b/edge-agent/icon.go @@ -0,0 +1,20 @@ +//go:build windows + +package main + +var iconData = []byte{ + 0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A, 0x00, 0x00, 0x00, 0x0D, + 0x49, 0x48, 0x44, 0x52, 0x00, 0x00, 0x00, 0x20, 0x00, 0x00, 0x00, 0x20, + 0x08, 0x06, 0x00, 0x00, 0x00, 0x73, 0x7A, 0x7A, 0xF4, 0x00, 0x00, 0x00, + 0x01, 0x73, 0x52, 0x47, 0x42, 0x00, 0xAE, 0xCE, 0x1C, 0xE9, 0x00, 0x00, + 0x00, 0x55, 0x49, 0x44, 0x41, 0x54, 0x58, 0x47, 0x63, 0x60, 0x18, 0x05, + 0x74, 0x86, 0x02, 0x86, 0xFF, 0xFF, 0x41, 0xF4, 0x7F, 0xFA, 0x02, 0x06, + 0x86, 0x41, 0x0F, 0x05, 0x0C, 0x83, 0x1E, 0x0A, 0x06, 0x3D, 0x14, 0x0C, + 0x7A, 0x28, 0x18, 0xF4, 0x50, 0x30, 0xE8, 0xA1, 0x60, 0xD0, 0x43, 0xC1, + 0xA0, 0x87, 0x82, 0x41, 0x0F, 0x05, 0x83, 0x1E, 0x0A, 0x06, 0x3D, 0x14, + 0x0C, 0x7A, 0x28, 0x18, 0xF4, 0x50, 0x30, 0xE8, 0xA1, 0x60, 0xD0, 0x43, + 0x01, 0xA3, 0x8C, 0x84, 0xC4, 0x7F, 0x4A, 0x02, 0x60, 0xA6, 0xFE, 0xFF, + 0x0F, 0xA5, 0x80, 0xA8, 0xC0, 0x2D, 0xD3, 0xFF, 0x3F, 0xFA, 0x81, 0x64, + 0xC4, 0x04, 0x16, 0x6A, 0x27, 0x00, 0x00, 0x00, 0x00, 0x49, 0x45, 0x4E, + 0x44, 0xAE, 0x42, 0x60, 0x82, +} diff --git a/edge-agent/main.go b/edge-agent/main.go new file mode 100644 index 0000000..01a0241 --- /dev/null +++ b/edge-agent/main.go @@ -0,0 +1,68 @@ +package main + +import ( + "embed" + "log" + "os" + "os/signal" + "syscall" + "time" + + "streamcore-edge/agent" +) + +//go:embed mediamtx/* +var mediamtxFS embed.FS + +const ( + defaultOriginURL = "http://127.0.0.1:8080" + heartbeatInterval = 10 * time.Second +) + +func main() { + log.SetFlags(log.LstdFlags | log.Lshortfile) + + cfg := agent.Config{ + NodeName: "edge-node", + OriginURL: getEnv("STREAMCORE_ORIGIN", defaultOriginURL), + HlsPort: 8889, + ApiPort: 9998, + Subnet: getEnv("STREAMCORE_SUBNET", "0.0.0.0/0"), + DataDir: getEnv("STREAMCORE_DATA_DIR", ""), + Weight: 100, + } + + if cfg.DataDir == "" { + home, _ := os.UserHomeDir() + if home == "" { + home = os.TempDir() + } + cfg.DataDir = home + "/.streamcore-edge" + } + + ag, err := agent.New(cfg, mediamtxFS) + if err != nil { + log.Fatalf("Failed to initialize agent: %v", err) + } + + if err := ag.Start(); err != nil { + log.Fatalf("Failed to start agent: %v", err) + } + + log.Println("Edge agent started") + runTray(ag) + + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) + <-sigCh + + log.Println("Shutting down...") + ag.Stop() +} + +func getEnv(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} diff --git a/edge-agent/tray_linux.go b/edge-agent/tray_linux.go new file mode 100644 index 0000000..758c05f --- /dev/null +++ b/edge-agent/tray_linux.go @@ -0,0 +1,16 @@ +//go:build !windows + +package main + +import ( + "log" + "streamcore-edge/agent" +) + +func runTray(ag *agent.Agent) { + log.Println("Running in headless mode (Linux server)") + log.Printf("Node ID: %s", ag.NodeID()) + log.Printf("HLS port: %d", ag.Config().HlsPort) + log.Printf("API port: %d", ag.Config().ApiPort) + log.Println("Press Ctrl+C to stop") +} diff --git a/edge-agent/tray_windows.go b/edge-agent/tray_windows.go new file mode 100644 index 0000000..df4112b --- /dev/null +++ b/edge-agent/tray_windows.go @@ -0,0 +1,64 @@ +//go:build windows + +package main + +import ( + "fmt" + "log" + "time" + + "streamcore-edge/agent" + + "github.com/getlantern/systray" +) + +func runTray(ag *agent.Agent) { + systray.Run( + func() { onReady(ag) }, + func() { onExit(ag) }, + ) +} + +func onReady(ag *agent.Agent) { + systray.SetIcon(iconData) + systray.SetTitle("StreamCore Edge") + systray.SetTooltip("StreamCore Edge Node") + + mStatus := systray.AddMenuItem("Node: "+ag.NodeID(), "Node") + mStatus.Disable() + + systray.AddSeparator() + + mCPU := systray.AddMenuItem("CPU: --", "CPU") + mCPU.Disable() + mMem := systray.AddMenuItem("MEM: --", "Memory") + mMem.Disable() + mBW := systray.AddMenuItem("Bandwidth: --", "Bandwidth") + mBW.Disable() + + systray.AddSeparator() + + mQuit := systray.AddMenuItem("Quit", "Shutdown") + + go func() { + ticker := time.NewTicker(3 * time.Second) + defer ticker.Stop() + for { + select { + case <-ticker.C: + m := ag.Metrics() + mCPU.SetTitle(fmt.Sprintf("CPU: %.1f%%", m.CPU)) + mMem.SetTitle(fmt.Sprintf("MEM: %.1f%%", m.Memory)) + mBW.SetTitle(fmt.Sprintf("Bandwidth: %.1f Mbps", m.Bandwidth)) + case <-mQuit.ClickedCh: + systray.Quit() + return + } + } + }() +} + +func onExit(ag *agent.Agent) { + log.Println("Tray exit requested") + ag.Stop() +} diff --git a/start.sh b/start.sh index a6a8ddd..5ec4c26 100755 --- a/start.sh +++ b/start.sh @@ -1,35 +1,44 @@ #!/bin/bash +# StreamCore - Start all services set -e BASE="/workspace/livestream-platform" -echo "[livestream] Starting services..." +echo "[StreamCore] Starting services..." +# MediaMTX origin cd "$BASE/mediamtx" -if pgrep -f "mediamtx mediamtx.yml" > /dev/null; then - echo "[livestream] MediaMTX already running." +if pgrep -f "mediamtx mediamtx.yml" > /dev/null 2>/dev/null; then + echo "[StreamCore] MediaMTX already running." else nohup ./mediamtx mediamtx.yml > /tmp/mediamtx.log 2>&1 & sleep 2 - echo "[livestream] MediaMTX started." + echo "[StreamCore] MediaMTX started (RTMP :1935, HLS :8888, API :9997)." fi +# Backend API cd "$BASE/backend" if curl -sf http://127.0.0.1:8080/api/health > /dev/null 2>&1; then - echo "[livestream] Backend already running." + echo "[StreamCore] Backend already running." else - nohup ./venv/bin/python run.py > /tmp/backend.log 2>&1 & + (exec ./venv/bin/python run.py &) /tmp/backend.log & + disown sleep 3 - echo "[livestream] Backend started on :8080." + echo "[StreamCore] Backend started on :8080." fi +# Frontend cd "$BASE/frontend" if curl -sf http://127.0.0.1:5173 > /dev/null 2>&1; then - echo "[livestream] Frontend already running." + echo "[StreamCore] Frontend already running." else - nohup npm run dev > /tmp/frontend.log 2>&1 & + (exec npm run dev &) /tmp/frontend.log & + disown sleep 3 - echo "[livestream] Frontend started on :5173." + echo "[StreamCore] Frontend started on :5173." fi -echo "[livestream] All services up." +echo "[StreamCore] All services up." +echo " Public: http://nbd.rainplay.cn:13001/" +echo " Admin: http://nbd.rainplay.cn:13001/admin/login" +echo " RTMP: rtmp://nbd.rainplay.cn:13001/live/{key}"