Skip to content

Feat/http maintainance #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
joschahenningsen wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Feat/http maintainance #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c5eb23d
feat(http): Add simple maintainance page
joschahenningsen Aug 9, 2026
36f799d
chore(nix): drop gst_all_1.gst-vaap
joschahenningsen Aug 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion shell.nix
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
hmelder marked this conversation as resolved.

libcap
go
Expand Down
124 changes: 124 additions & 0 deletions streamd/http.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,134 @@ package main

import (
"fmt"
"html/template"
"net/http"

"github.com/go-gst/go-gst/gst"
)

type httpServer struct {
daemonController
combPort string
presPort string
camPort string
Comment on lines +13 to +15

@hmelder hmelder Aug 13, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You can retrieve the values from the daemonConfig struct which is the first member of daemon, and used during construction of the httpServer instance in main. Perhaps add a helper function to daemonController so that we can avoid duplicating the {comb,pres,cam}Port fields.

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(`<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>streamd</title>
<style>
* { box-sizing: border-box; }
body { font-family: monospace; max-width: 860px; margin: 2em auto; padding: 0 1em; color: #222; }
h1 { margin-bottom: 0.2em; }
h2 { font-size: 1em; margin: 1.4em 0 0.4em; text-transform: uppercase; letter-spacing: 0.05em; color: #555; }
table { border-collapse: collapse; width: 100%; margin-bottom: 0.5em; }
th, td { padding: 5px 10px; text-align: left; border: 1px solid #ddd; }
th { background: #f5f5f5; font-weight: normal; width: 40%; }
.ok { color: #2a2; }
.warn { color: #c80; }
.nav { margin-top: 2em; }
.nav a { margin-right: 1em; }
button { padding: 6px 16px; cursor: pointer; font-family: monospace; }
.danger { background: #fee; border: 1px solid #c00; color: #c00; }
.danger:hover { background: #c00; color: #fff; }
</style>
</head>
<body>
<h1>streamd</h1>

<h2>Pipeline</h2>
<table>
<tr><th>State</th><td><span class="ok">&#9679; running</span></td></tr>
<tr><th>Warnings</th><td{{if .Warnings}} class="warn"{{end}}>{{.Warnings}}</td></tr>
{{if .QosEvents -}}
<tr><th>QoS events</th><td class="warn">{{range $k, $v := .QosEvents}}{{$k}}: {{$v}}<br>{{end}}</td></tr>
{{- end}}
</table>

<h2>SRT Sinks</h2>
<table>
<tr><th>Combined (port {{.CompPort}})</th><td>{{.CompCallers}} caller(s)</td></tr>
<tr><th>Presentation (port {{.PresentPort}})</th><td>{{.PresentCallers}} caller(s)</td></tr>
<tr><th>Camera (port {{.CamPort}})</th><td>{{.CamCallers}} caller(s)</td></tr>
</table>

<h2>System</h2>
<table>
<tr><th>Load average (1/5/15 min)</th><td>{{printf "%.2f" .LoadOne}} / {{printf "%.2f" .LoadFive}} / {{printf "%.2f" .LoadFifteen}}</td></tr>
<tr><th>Memory used</th><td>{{.MemUsedMB}} MB</td></tr>
<tr><th>Memory available</th><td>{{.MemFreeMB}} MB</td></tr>
</table>

<h2>Actions</h2>
<form method="POST" action="/restart" onsubmit="return confirm('Restart the pipeline?')">
<button type="submit" class="danger">Restart pipeline</button>
</form>

<div class="nav">
<a href="/logs">/logs</a>
<a href="/metrics">/metrics</a>
<a href="/graph?details=states">/graph</a>
</div>
</body>
</html>
`))

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) {
Expand Down Expand Up @@ -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)
}
61 changes: 61 additions & 0 deletions streamd/logs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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")))
}
40 changes: 39 additions & 1 deletion streamd/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"flag"
"fmt"
"io"
"net"
"net/http"
"os"
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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()

@hmelder hmelder Aug 13, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Before creating a new pipeline, I think you'll need to unregister the bus with p.GetBus().removeWatch. Can you add a unregisterBusWatch method next to registerBusWatch (

LectureMix/streamd/gstreamer_bus.go

Line 25 in d210cde

return p.GetBus().AddWatch(func(msg *gst.Message) bool {
)?

@hmelder hmelder Aug 13, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For reference: https://gstreamer.freedesktop.org/documentation/gstreamer/gstbus.html?gi-language=c#gst_bus_remove_watch

@hmelder hmelder Aug 13, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Have you tested, whether restarting of the pipeline works, and that all objects are gc'd eventually?

@joschahenningsen joschahenningsen Aug 13, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My sanity check wasn't very thorough, I did it with a local test source which worked well.

newP.pipeline.SetState(gst.StatePlaying)
return nil
Comment on lines +148 to +160

@hmelder hmelder Aug 13, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not call runPipeline directly? Make sure to move gst.Init(&os.Args) in runPipeline to main, because this must be only called once. My bad.

}

func main() {
d := &daemon{}

Expand All @@ -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 != "" {
Expand All @@ -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)
Expand Down