diff --git a/shell.nix b/shell.nix index 550985f..b024b71 100644 --- a/shell.nix +++ b/shell.nix @@ -9,7 +9,6 @@ mkShell { gst_all_1.gst-plugins-base gst_all_1.gst-plugins-good gst_all_1.gst-libav # For avenc_aac - gst_all_1.gst-vaapi libcap go diff --git a/streamd/http.go b/streamd/http.go index 8da13d2..6f91319 100644 --- a/streamd/http.go +++ b/streamd/http.go @@ -2,6 +2,7 @@ package main import ( "fmt" + "html/template" "net/http" "github.com/go-gst/go-gst/gst" @@ -9,6 +10,126 @@ import ( type httpServer struct { daemonController + combPort string + presPort string + camPort string + lb *logBuffer +} + +type indexData struct { + Warnings uint64 + QosEvents map[string]uint64 + CompCallers int + PresentCallers int + CamCallers int + LoadOne float64 + LoadFive float64 + LoadFifteen float64 + MemUsedMB int64 + MemFreeMB int64 + CompPort string + PresentPort string + CamPort string +} + +var indexTmpl = template.Must(template.New("index").Parse(` + + + + +streamd + + + +

streamd

+ +

Pipeline

+ + +{{.Warnings}} +{{if .QosEvents -}} + +{{- end}} +
State● running
Warnings
QoS events{{range $k, $v := .QosEvents}}{{$k}}: {{$v}}
{{end}}
+ +

SRT Sinks

+ + + + +
Combined (port {{.CompPort}}){{.CompCallers}} caller(s)
Presentation (port {{.PresentPort}}){{.PresentCallers}} caller(s)
Camera (port {{.CamPort}}){{.CamCallers}} caller(s)
+ +

System

+ + + + +
Load average (1/5/15 min){{printf "%.2f" .LoadOne}} / {{printf "%.2f" .LoadFive}} / {{printf "%.2f" .LoadFifteen}}
Memory used{{.MemUsedMB}} MB
Memory available{{.MemFreeMB}} MB
+ +

Actions

+
+ +
+ +
+ /logs + /metrics + /graph +
+ + +`)) + +func (h *httpServer) handleIndex(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + m := h.metricsSnapshot() + data := indexData{ + Warnings: m.pipelineStats.warnings, + QosEvents: m.pipelineStats.qosEvents, + CompCallers: len(m.compSinkStats.callers), + PresentCallers: len(m.presentSinkStats.callers), + CamCallers: len(m.camSinkStats.callers), + LoadOne: m.loadAvg.One, + LoadFive: m.loadAvg.Five, + LoadFifteen: m.loadAvg.Fifteen, + MemUsedMB: int64(m.mem.MemTotal-m.mem.MemFree-m.mem.Buffers-m.mem.Cached) / 1024, + MemFreeMB: int64(m.mem.MemFree+m.mem.Buffers+m.mem.Cached) / 1024, + CompPort: h.combPort, + PresentPort: h.presPort, + CamPort: h.camPort, + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + indexTmpl.Execute(w, data) +} + +func (h *httpServer) handleRestart(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + if err := h.daemonController.restart(); err != nil { + http.Error(w, fmt.Sprintf("restart failed: %v", err), http.StatusInternalServerError) + return + } + http.Redirect(w, r, "/", http.StatusSeeOther) } func writeSRTStatsMeta(w http.ResponseWriter) { @@ -225,6 +346,9 @@ func (h *httpServer) graph(w http.ResponseWriter, r *http.Request) { } func (h *httpServer) setupHTTPHandlers() { + http.HandleFunc("/", h.handleIndex) + http.HandleFunc("/logs", h.handleLogs) http.HandleFunc("/metrics", h.metrics) http.HandleFunc("/graph", h.graph) + http.HandleFunc("/restart", h.handleRestart) } diff --git a/streamd/logs.go b/streamd/logs.go new file mode 100644 index 0000000..e411ff2 --- /dev/null +++ b/streamd/logs.go @@ -0,0 +1,61 @@ +package main + +import ( + "net/http" + "strings" + "sync" +) + +const logBufferSize = 10000 + +type logBuffer struct { + mu sync.Mutex + lines []string + pos int + full bool +} + +func newLogBuffer() *logBuffer { + return &logBuffer{lines: make([]string, logBufferSize)} +} + +// Write implements io.Writer so logBuffer can be passed to klog.SetOutput. +// klog writes one complete formatted log line per Write call. +func (lb *logBuffer) Write(p []byte) (n int, err error) { + s := string(p) + if len(s) > 0 && s[len(s)-1] == '\n' { + s = s[:len(s)-1] + } + if s == "" { + return len(p), nil + } + lb.mu.Lock() + lb.lines[lb.pos] = s + lb.pos = (lb.pos + 1) % logBufferSize + if lb.pos == 0 { + lb.full = true + } + lb.mu.Unlock() + return len(p), nil +} + +// snapshot returns buffered lines in chronological order. +func (lb *logBuffer) snapshot() []string { + lb.mu.Lock() + defer lb.mu.Unlock() + if !lb.full { + out := make([]string, lb.pos) + copy(out, lb.lines[:lb.pos]) + return out + } + out := make([]string, logBufferSize) + copy(out, lb.lines[lb.pos:]) + copy(out[logBufferSize-lb.pos:], lb.lines[:lb.pos]) + return out +} + +func (h *httpServer) handleLogs(w http.ResponseWriter, r *http.Request) { + lines := h.lb.snapshot() + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.Write([]byte(strings.Join(lines, "\n"))) +} diff --git a/streamd/main.go b/streamd/main.go index 79cb4e8..60adb07 100644 --- a/streamd/main.go +++ b/streamd/main.go @@ -4,6 +4,7 @@ import ( "context" "flag" "fmt" + "io" "net" "net/http" "os" @@ -75,6 +76,7 @@ type daemonController interface { metricsSnapshot() metrics graph(details gst.DebugGraphDetails) string srtStatistics() ([]*srtStats, error) + restart() error } func (d *daemon) srtStatistics() ([]*srtStats, error) { @@ -136,6 +138,28 @@ func (d *daemon) runPipeline() error { return nil } +func (d *daemon) restart() error { + d.mu.Lock() + oldGstPipeline := d.pipeline.pipeline + d.mu.Unlock() + + oldGstPipeline.BlockSetState(gst.StateNull) + + newP, err := newPipeline(&d.daemonConfig) + if err != nil { + return err + } + + d.mu.Lock() + d.pipeline = newP + d.metrics.pipelineStats = newPipelineStats() + d.mu.Unlock() + + d.registerBusWatch() + newP.pipeline.SetState(gst.StatePlaying) + return nil +} + func main() { d := &daemon{} @@ -154,6 +178,7 @@ func main() { flag.IntVar(&d.audioEncBitrateKbps, "audio-enc-bitrate", 96, "Video encoding bitrate in Kbps") flag.Float64Var(&d.audioAmplification, "audio-amplification", 1.0, "Audio amplifcation after conversion") flag.BoolVar(&d.hwAccel, "hw-accel", false, "Enable hardware acceleration and offload processing tasks onto the GPU or a DSP") + klog.InitFlags(nil) // register klog flags with flag.CommandLine before parsing flag.Parse() if d.listenCidr != "" { @@ -173,11 +198,24 @@ func main() { d.listenAddr = "[::]" } + lb := newLogBuffer() + // klog defaults to logtostderr=true, which writes directly to os.Stderr + // and bypasses the file sinks that SetOutput replaces. Disable it so all + // log lines go through our MultiWriter (which still writes to os.Stderr). + flag.Set("logtostderr", "false") + klog.SetOutput(io.MultiWriter(os.Stderr, lb)) + d.mainloop = glib.NewMainLoop(glib.MainContextDefault(), false) ctx, _ := signal.NotifyContext(context.Background(), os.Interrupt) // Create and start HTTP server - h := &httpServer{d} + h := &httpServer{ + daemonController: d, + combPort: d.combPort, + presPort: d.presPort, + camPort: d.camPort, + lb: lb, + } h.setupHTTPHandlers() klog.Infof("listening for HTTP at %s:%s", d.listenAddr, d.listenHTTP)