From 4395fe2417f08b77c4c6b0836220b3c3d1e70b4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9dric=20Verstraeten?= Date: Wed, 5 Aug 2026 20:53:37 +0200 Subject: [PATCH] Add optional MoQ live-stream publisher Adds a dedicated MoQ build path that publishes H.264 live streams to a configurable relay, with retry handling, stream selection, Annex B framing, and Docker packaging. Standard builds retain a no-op implementation. --- Dockerfile.moq | 74 ++++++++++ machinery/go.mod | 2 + machinery/go.sum | 4 + machinery/src/cloud/livemoq/annexb.go | 29 ++++ machinery/src/cloud/livemoq/annexb_test.go | 47 ++++++ machinery/src/cloud/livemoq_disabled.go | 8 ++ machinery/src/cloud/livemoq_enabled.go | 159 +++++++++++++++++++++ machinery/src/components/kerberos.go | 4 + 8 files changed, 327 insertions(+) create mode 100644 Dockerfile.moq create mode 100644 machinery/src/cloud/livemoq/annexb.go create mode 100644 machinery/src/cloud/livemoq/annexb_test.go create mode 100644 machinery/src/cloud/livemoq_disabled.go create mode 100644 machinery/src/cloud/livemoq_enabled.go diff --git a/Dockerfile.moq b/Dockerfile.moq new file mode 100644 index 0000000..6f5fcfe --- /dev/null +++ b/Dockerfile.moq @@ -0,0 +1,74 @@ +ARG GO_IMAGE=golang:1.24-trixie +ARG RUNTIME_IMAGE=debian:trixie-slim +ARG VERSION=0.0.0 + +FROM ${GO_IMAGE} AS build-machinery + +ARG VERSION +ENV CGO_ENABLED=1 + +RUN apt-get update && apt-get install -y --no-install-recommends \ + build-essential \ + ca-certificates \ + git \ + libavcodec-dev \ + libavutil-dev \ + libswscale-dev \ + pkg-config \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /src +COPY machinery ./machinery +COPY .git ./.git + +RUN cd machinery && \ + go mod download && \ + if [ -z "${VERSION}" ] || [ "${VERSION}" = "0.0.0" ]; then \ + VERSION=$(git -C /src describe --tags --always 2>/dev/null || echo "0.0.0"); \ + fi && \ + go build \ + -tags moq,timetzdata,netgo,osusergo \ + -ldflags "-s -w -X github.com/kerberos-io/agent/machinery/src/utils.VERSION=${VERSION}" \ + -o /out/main \ + ./main.go && \ + cp -r data /out/data && \ + printf '%s' "${VERSION}" > /out/version && \ + mkdir -p /out/data/cloud /out/data/snapshots /out/data/log /out/data/recordings /out/data/capture-test /out/data/config + +FROM node:22-alpine AS build-ui + +WORKDIR /src/ui +COPY ui ./ +RUN mkdir -p /src/machinery && \ + yarn config set network-timeout 300000 && \ + yarn install --frozen-lockfile && \ + yarn build + +FROM ${RUNTIME_IMAGE} + +RUN apt-get update && apt-get install -y --no-install-recommends \ + ca-certificates \ + curl \ + ffmpeg \ + libcap2-bin \ + libstdc++6 \ + && rm -rf /var/lib/apt/lists/* && \ + groupadd --system kerberosio && \ + useradd --system --gid kerberosio --groups video --create-home agent + +COPY --from=build-machinery --chown=root:root /out/main /home/agent/main +COPY --from=build-machinery --chown=agent:kerberosio /out/data /home/agent/data +COPY --from=build-machinery --chown=root:root /out/version /home/agent/version +COPY --from=build-ui --chown=agent:kerberosio /src/machinery/www /home/agent/www + +RUN cp /home/agent/data/config/config.json /home/agent/data/config.template.json && \ + chown agent:kerberosio /home/agent/data/config.template.json && \ + setcap 'cap_net_bind_service=+ep' /home/agent/main && \ + /home/agent/main -action version + +USER agent +WORKDIR /home/agent + +EXPOSE 80 +HEALTHCHECK CMD curl --fail http://localhost:80 || exit 1 +CMD ["./main", "-action", "run", "-port", "80"] \ No newline at end of file diff --git a/machinery/go.mod b/machinery/go.mod index 1364d91..496bb53 100644 --- a/machinery/go.mod +++ b/machinery/go.mod @@ -26,6 +26,7 @@ require ( github.com/kerberos-io/joy4 v1.0.64 github.com/kerberos-io/onvif v1.2.2 github.com/minio/minio-go/v6 v6.0.57 + github.com/moq-dev/moq-go v0.5.7 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 github.com/op/go-logging v0.0.0-20160315200505-970db520ece7 github.com/pion/interceptor v0.1.40 @@ -95,6 +96,7 @@ require ( github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect github.com/montanaflynn/stats v0.7.1 // indirect + github.com/moq-dev/moq-go-ffi v0.3.7 // indirect github.com/nxadm/tail v1.4.11 // indirect github.com/pelletier/go-toml/v2 v2.2.3 // indirect github.com/pion/datachannel v1.5.10 // indirect diff --git a/machinery/go.sum b/machinery/go.sum index 8e11c52..bb469ba 100644 --- a/machinery/go.sum +++ b/machinery/go.sum @@ -845,6 +845,10 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE= github.com/montanaflynn/stats v0.7.1/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow= +github.com/moq-dev/moq-go v0.5.7 h1:LfFpgAU8FRMcnU85L5Lb03HzxhYbQ2+BhT/pJQdjy5U= +github.com/moq-dev/moq-go v0.5.7/go.mod h1:5K8zjKKjWe5lzfCtlfxoAfdCr3KX6EBuTOZsz0WBnzw= +github.com/moq-dev/moq-go-ffi v0.3.7 h1:+xwPOzTJHvB0tuTnW6znbvvOgT4yLnN7sMqXXZYh02M= +github.com/moq-dev/moq-go-ffi v0.3.7/go.mod h1:zxpOlUetvaoxWBnbXTdILtLLUytaoVuLvO36lftwWO0= github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6Oo2LfFZAehjjQMERAvZLEDnQ= github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646/go.mod h1:jpp1/29i3P1S/RLdc7JQKbRpFeM1dOBd8T9ki5s+AY8= github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno= diff --git a/machinery/src/cloud/livemoq/annexb.go b/machinery/src/cloud/livemoq/annexb.go new file mode 100644 index 0000000..52c68d9 --- /dev/null +++ b/machinery/src/cloud/livemoq/annexb.go @@ -0,0 +1,29 @@ +package livemoq + +import "strings" + +var annexBStartCode = []byte{0x00, 0x00, 0x00, 0x01} + +// EnsureAnnexB restores the start code stripped by the Agent capture queue. +func EnsureAnnexB(payload []byte) []byte { + if hasAnnexBStartCode(payload) { + return payload + } + + framed := make([]byte, 0, len(annexBStartCode)+len(payload)) + framed = append(framed, annexBStartCode...) + return append(framed, payload...) +} + +func BroadcastPath(prefix string, deviceKey string) string { + prefix = strings.Trim(prefix, "/") + if prefix == "" { + prefix = "devices" + } + return prefix + "/" + strings.Trim(deviceKey, "/") + "/live.hang" +} + +func hasAnnexBStartCode(payload []byte) bool { + return len(payload) >= 4 && payload[0] == 0 && payload[1] == 0 && + ((payload[2] == 0 && payload[3] == 1) || payload[2] == 1) +} diff --git a/machinery/src/cloud/livemoq/annexb_test.go b/machinery/src/cloud/livemoq/annexb_test.go new file mode 100644 index 0000000..7ba2ae7 --- /dev/null +++ b/machinery/src/cloud/livemoq/annexb_test.go @@ -0,0 +1,47 @@ +package livemoq + +import ( + "bytes" + "testing" +) + +func TestEnsureAnnexB(t *testing.T) { + tests := []struct { + name string + payload []byte + want []byte + }{ + { + name: "missing start code", + payload: []byte{0x41, 0x01}, + want: []byte{0x00, 0x00, 0x00, 0x01, 0x41, 0x01}, + }, + { + name: "four byte start code", + payload: []byte{0x00, 0x00, 0x00, 0x01, 0x65}, + want: []byte{0x00, 0x00, 0x00, 0x01, 0x65}, + }, + { + name: "three byte start code", + payload: []byte{0x00, 0x00, 0x01, 0x41}, + want: []byte{0x00, 0x00, 0x01, 0x41}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := EnsureAnnexB(test.payload); !bytes.Equal(got, test.want) { + t.Fatalf("EnsureAnnexB() = %x, want %x", got, test.want) + } + }) + } +} + +func TestBroadcastPath(t *testing.T) { + if got := BroadcastPath("/devices/", "/camera-1/"); got != "devices/camera-1/live.hang" { + t.Fatalf("BroadcastPath() = %q", got) + } + if got := BroadcastPath("", "camera-1"); got != "devices/camera-1/live.hang" { + t.Fatalf("BroadcastPath() default = %q", got) + } +} diff --git a/machinery/src/cloud/livemoq_disabled.go b/machinery/src/cloud/livemoq_disabled.go new file mode 100644 index 0000000..40e7a55 --- /dev/null +++ b/machinery/src/cloud/livemoq_disabled.go @@ -0,0 +1,8 @@ +//go:build !moq + +package cloud + +import "github.com/kerberos-io/agent/machinery/src/models" + +// StartLiveStreamMoQ is disabled in the standard Agent build. +func StartLiveStreamMoQ(_ *models.Configuration, _ *models.Communication, _ bool) {} diff --git a/machinery/src/cloud/livemoq_enabled.go b/machinery/src/cloud/livemoq_enabled.go new file mode 100644 index 0000000..b275b1a --- /dev/null +++ b/machinery/src/cloud/livemoq_enabled.go @@ -0,0 +1,159 @@ +//go:build moq + +package cloud + +import ( + "context" + "fmt" + "os" + "strings" + "time" + + "github.com/kerberos-io/agent/machinery/src/cloud/livemoq" + "github.com/kerberos-io/agent/machinery/src/log" + "github.com/kerberos-io/agent/machinery/src/models" + "github.com/kerberos-io/agent/machinery/src/packets" + "github.com/moq-dev/moq-go/moq" +) + +const ( + defaultMoQRelayURL = "https://relay.uug.ai/anon" + minMoQRetryDelay = time.Second + maxMoQRetryDelay = 30 * time.Second +) + +type liveMoQConfig struct { + relayURL string + broadcast string + quality string + sourceLabel string + queue *packets.Queue +} + +// StartLiveStreamMoQ starts the publisher only in the dedicated MoQ build and +// only when explicitly enabled by the deployment. +func StartLiveStreamMoQ(configuration *models.Configuration, communication *models.Communication, subStreamEnabled bool) { + if os.Getenv("AGENT_LIVE_MOQ_ENABLED") != "true" { + return + } + + config := configuration.Config + if config.Offline == "true" || config.Capture.Liveview == "false" { + log.Log.Info("cloud.StartLiveStreamMoQ(): disabled by Agent live-view configuration") + return + } + if config.Key == "" { + log.Log.Warning("cloud.StartLiveStreamMoQ(): AGENT_KEY is required") + return + } + + quality := os.Getenv("AGENT_LIVE_MOQ_QUALITY") + if quality == "" { + quality = models.StreamQualityAuto + } + useSub := models.SelectSubStreamForQuality(config, quality, subStreamEnabled) + queue := communication.Queue + sourceLabel := "main" + if useSub && communication.SubQueue != nil { + queue = communication.SubQueue + sourceLabel = "sub" + } + if queue == nil { + log.Log.Warning("cloud.StartLiveStreamMoQ(): selected packet queue is unavailable") + return + } + + relayURL := os.Getenv("AGENT_LIVE_MOQ_URL") + if relayURL == "" { + relayURL = defaultMoQRelayURL + } + publisherConfig := liveMoQConfig{ + relayURL: relayURL, + broadcast: livemoq.BroadcastPath(os.Getenv("AGENT_LIVE_MOQ_BROADCAST_PREFIX"), config.Key), + quality: quality, + sourceLabel: sourceLabel, + queue: queue, + } + + ctx := context.Background() + if communication.Context != nil { + ctx = *communication.Context + } + go runLiveStreamMoQ(ctx, publisherConfig) +} + +func runLiveStreamMoQ(ctx context.Context, config liveMoQConfig) { + log.Log.Info(fmt.Sprintf( + "cloud.runLiveStreamMoQ(): publishing %s stream (quality=%s) to %s/%s", + config.sourceLabel, config.quality, strings.TrimRight(config.relayURL, "/"), config.broadcast, + )) + + retryDelay := minMoQRetryDelay + for ctx.Err() == nil { + connectedAt := time.Now() + err := publishLiveStreamMoQ(ctx, config) + if ctx.Err() != nil { + return + } + log.Log.Warning("cloud.runLiveStreamMoQ(): publisher stopped: " + err.Error()) + if time.Since(connectedAt) >= time.Minute { + retryDelay = minMoQRetryDelay + } + + timer := time.NewTimer(retryDelay) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + if retryDelay < maxMoQRetryDelay { + retryDelay *= 2 + if retryDelay > maxMoQRetryDelay { + retryDelay = maxMoQRetryDelay + } + } + } +} + +func publishLiveStreamMoQ(ctx context.Context, config liveMoQConfig) error { + client, err := moq.Dial(ctx, config.relayURL) + if err != nil { + return fmt.Errorf("connect to relay: %w", err) + } + defer client.Close() + + broadcast, err := client.CreateBroadcast(config.broadcast) + if err != nil { + return fmt.Errorf("create broadcast: %w", err) + } + defer broadcast.Finish() + + stream, err := broadcast.PublishMediaStream("avc3") + if err != nil { + return fmt.Errorf("create H.264 media stream: %w", err) + } + defer stream.Finish() + + cursor := config.queue.Latest() + writing := false + for { + packet, err := cursor.ReadPacket() + if err != nil { + return fmt.Errorf("read packet: %w", err) + } + if !packet.IsVideo || len(packet.Data) == 0 || !strings.EqualFold(packet.Codec, "H264") { + continue + } + if !writing { + if !packet.IsKeyFrame { + continue + } + writing = true + log.Log.Info("cloud.publishLiveStreamMoQ(): first H.264 keyframe received; broadcast is live") + } + if err := stream.Write(livemoq.EnsureAnnexB(packet.Data)); err != nil { + return fmt.Errorf("write H.264 access unit: %w", err) + } + } +} diff --git a/machinery/src/components/kerberos.go b/machinery/src/components/kerberos.go index 5637824..68544e2 100644 --- a/machinery/src/components/kerberos.go +++ b/machinery/src/components/kerberos.go @@ -305,6 +305,10 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu // watching. go cloud.HandleLiveStreamHLS(configuration, communication, mqttClient, subStreamEnabled) + // MoQ is available only in the dedicated CGO/glibc build. The standard + // static Alpine build resolves this hook to a no-op. + cloud.StartLiveStreamMoQ(configuration, communication, subStreamEnabled) + // Handle livestream HD (high resolution over WEBRTC). Both the main and sub // stream are exposed as separate broadcasters so a viewer can request the // high (main) or low (sub) resolution per peer connection; "auto" prefers the