feat(runner): yet another runner (#1486)

go.work

@@ -7,4 +7,5 @@ use (
 	./worker
 	./worker/edge
 	vod-service
+	runner
 )

runner/.gitignore

@@ -0,0 +1 @@
+mediamtx

runner/Dockerfile

@@ -0,0 +1,27 @@
+FROM amd64/golang:1.21-alpine3.18 as builder
+
+WORKDIR /go/src/github.com/TUM-Dev/gocast/runner
+COPY . .
+
+RUN GO111MODULE=on go mod download
+# bundle version into binary if specified in build-args, dev otherwise.
+ARG version=dev
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a -ldflags "-w -extldflags '-static' -X main.V=${version}" -o /runner cmd/runner/main.go
+
+FROM bluenviron/mediamtx:1.1.0 as mediamtx
+
+FROM alpine:3.18
+ADD entrypoint.sh /entrypoint.sh
+ADD mediamtx.yml /mediamtx.yml
+RUN chmod +x /entrypoint.sh
+
+RUN apk add --no-cache \
+  ffmpeg \
+  tzdata
+
+COPY --from=builder /runner /runner
+RUN chmod +x /runner
+COPY --from=mediamtx /mediamtx /mediamtx
+RUN chmod +x /mediamtx
+
+CMD ["/entrypoint.sh"]

runner/Makefile

@@ -0,0 +1,36 @@
+.PHONY: all
+all: build
+
+VERSION := $(shell git rev-parse --short origin/HEAD)
+
+.PHONY: protoGen
+protoGen:
+	protoc ./*.proto --go-grpc_out=.. --go_out=..
+
+.PHONY: build
+build: deps
+	go build -o main -ldflags="-X 'main.VersionTag=$(VERSION)'" cmd/runner/main.go
+
+.PHONY: deps
+deps:
+	go get ./...
+
+.PHONY: install
+install:
+	mv main /bin/runner
+
+.PHONY: clean
+clean:
+	rm -f main
+
+.PHONY: test
+test:
+	go test -race ./...
+
+.PHONY: run
+run:
+	go run cmd/runner/main.go
+
+.PHONY: lint
+lint:
+	golangci-lint run

runner/cmd/runner/main.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"github.com/tum-dev/gocast/runner"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+// V (Version) is bundled into binary with -ldflags "-X ..."
+var V = "dev"
+
+func main() {
+	// ...
+
+	// Init EnvConfig
+	r := runner.NewRunner(V)
+	go r.Run()
+
+	shouldShutdown := false // set to true once we receive a shutdown signal
+
+	currentCount := 0
+	go func() {
+		for {
+			currentCount += <-r.JobCount // count Job start/stop
+			slog.Info("current action count", "count", currentCount)
+			if shouldShutdown && currentCount == 0 { // if we should shut down and no jobs are running, exit.
+				slog.Info("No actions left, shutting down")
+				os.Exit(0)
+			}
+		}
+	}()
+
+	osSignal := make(chan os.Signal, 1)
+	signal.Notify(osSignal, syscall.SIGINT, syscall.SIGTERM, syscall.SIGUSR1)
+	s := <-osSignal
+	slog.Info("Received signal", "signal", s)
+	shouldShutdown = true
+	r.Drain()
+
+	if currentCount == 0 {
+		slog.Info("No jobs left, shutting down")
+		os.Exit(0)
+	}
+
+	blocking := make(chan struct{})
+	_ = <-blocking
+
+}

runner/config/config.go

@@ -0,0 +1,26 @@
+package config
+
+import (
+	"github.com/caarlos0/env"
+	"log/slog"
+)
+
+var Config struct {
+	LogFmt       string `env:"LOG_FMT" envDefault:"txt"`
+	LogLevel     string `env:"LOG_LEVEL" envDefault:"debug"`
+	Port         int    `env:"PORT" envDefault:"0"`
+	StoragePath  string `env:"STORAGE_PATH" envDefault:"storage/mass"`
+	SegmentPath  string `env:"SEGMENT_PATH" envDefault:"storage/live"`
+	GocastServer string `env:"GOCAST_SERVER" envDefault:"localhost:50056"`
+	Hostname     string `env:"REALHOST" envDefault:"localhost"`
+	Version      string `env:"VERSION" envDefault:"dev"`
+	EdgeServer   string `env:"EDGE_SERVER" envDefault:"localhost:50057"`
+}
+
+func init() {
+	if err := env.Parse(&Config); err != nil {
+		slog.Error("error parsing envConfig", "error", err)
+	}
+
+	slog.Info("envConfig loaded", "envConfig", Config)
+}

runner/entrypoint.sh

@@ -0,0 +1,4 @@
+#!/bin/sh
+
+exec /mediamtx &
+exec /runner

runner/go.mod

@@ -0,0 +1,30 @@
+module github.com/tum-dev/gocast/runner
+
+go 1.21.0
+
+toolchain go1.21.3
+
+require (
+	github.com/caarlos0/env v3.5.0+incompatible
+	github.com/google/uuid v1.4.0
+	github.com/icza/gox v0.0.0-20230924165045-adcb03233bb5
+	github.com/shirou/gopsutil/v3 v3.24.1
+	golang.org/x/sync v0.6.0
+	google.golang.org/grpc v1.61.0
+	google.golang.org/protobuf v1.32.0
+)
+
+require (
+	github.com/go-ole/go-ole v1.3.0 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/lufia/plan9stats v0.0.0-20231016141302-07b5767bb0ed // indirect
+	github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b // indirect
+	github.com/shoenig/go-m1cpu v0.1.6 // indirect
+	github.com/tklauser/go-sysconf v0.3.13 // indirect
+	github.com/tklauser/numcpus v0.7.0 // indirect
+	github.com/yusufpapurcu/wmi v1.2.4 // indirect
+	golang.org/x/net v0.20.0 // indirect
+	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240125205218-1f4bbc51befe // indirect
+)

runner/go.sum

@@ -0,0 +1,76 @@
+github.com/caarlos0/env v3.5.0+incompatible h1:Yy0UN8o9Wtr/jGHZDpCBLpNrzcFLLM2yixi/rBrKyJs=
+github.com/caarlos0/env v3.5.0+incompatible/go.mod h1:tdCsowwCzMLdkqRYDlHpZCp2UooDD3MspDBjZ2AD02Y=
+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/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
+github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
+github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
+github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
+github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/icza/gox v0.0.0-20230924165045-adcb03233bb5 h1:K7KEFpKgVcjj98jOu2Z3xMBTtTwfYVT90Zmo3ZuWmbE=
+github.com/icza/gox v0.0.0-20230924165045-adcb03233bb5/go.mod h1:VbcN86fRkkUMPX2ufM85Um8zFndLZswoIW1eYtpAcVk=
+github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
+github.com/lufia/plan9stats v0.0.0-20231016141302-07b5767bb0ed h1:036IscGBfJsFIgJQzlui7nK1Ncm0tp2ktmPj8xO4N/0=
+github.com/lufia/plan9stats v0.0.0-20231016141302-07b5767bb0ed/go.mod h1:ilwx/Dta8jXAgpFYFvSWEMwxmbWXyiUHkd5FwyKhb5k=
+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/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
+github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b h1:0LFwY6Q3gMACTjAbMZBjXAqTOzOwFaj2Ld6cjeQ7Rig=
+github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
+github.com/shirou/gopsutil/v3 v3.24.1 h1:R3t6ondCEvmARp3wxODhXMTLC/klMa87h2PHUw5m7QI=
+github.com/shirou/gopsutil/v3 v3.24.1/go.mod h1:UU7a2MSBQa+kW1uuDq8DeEBS8kmrnQwsv2b5O513rwU=
+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/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.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.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
+github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08lq3r4=
+github.com/tklauser/go-sysconf v0.3.13/go.mod h1:zwleP4Q4OehZHGn4CYZDipCgg9usW5IJePewFCGVEa0=
+github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY=
+github.com/tklauser/numcpus v0.7.0 h1:yjuerZP127QG9m5Zh/mSO4wqurYil27tHrqwRoRjpr4=
+github.com/tklauser/numcpus v0.7.0/go.mod h1:bb6dMVcj8A42tSE7i32fsIUCbQNllK5iDguyOZRUzAY=
+github.com/yusufpapurcu/wmi v1.2.3/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
+github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
+github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
+golang.org/x/net v0.20.0 h1:aCL9BSgETF1k+blQaYUBx9hJ9LOGP3gAVemcZlf1Kpo=
+golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
+golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ=
+golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/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.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
+golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240125205218-1f4bbc51befe h1:bQnxqljG/wqi4NTXu2+DJ3n7APcEA882QZ1JvhQAq9o=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240125205218-1f4bbc51befe/go.mod h1:PAREbraiVEVGVdTZsVWjSbbTtSyGbAgIIvni8a8CD5s=
+google.golang.org/grpc v1.61.0 h1:TOvOcuXn30kRao+gfcvsebNEa5iZIiLkisYEkf7R7o0=
+google.golang.org/grpc v1.61.0/go.mod h1:VUbo7IFqmF1QtCAstipjG0GIoq49KvMe9+h1jFLBNJs=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
+google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
+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=

runner/handlers.go

@@ -0,0 +1,30 @@
+package runner
+
+import (
+	"context"
+	"github.com/tum-dev/gocast/runner/pkg/actions"
+	"github.com/tum-dev/gocast/runner/pkg/ptr"
+	"github.com/tum-dev/gocast/runner/protobuf"
+)
+
+func (r *Runner) RequestStream(ctx context.Context, req *protobuf.StreamRequest) (*protobuf.StreamResponse, error) {
+	ctx = context.Background()
+
+	data := map[string]any{
+		"streamID":   req.StreamId,
+		"streamEnd":  req.End.AsTime(),
+		"globalOpts": req.FfmpegGlobalOptions,
+		"inputOpts":  req.FfmpegInputOptions,
+		"outputOpts": req.FfmpegOutputOptions,
+		"input":      req.Input,
+	}
+
+	a := []actions.Action{
+		actions.Stream,
+	}
+
+	jID := r.RunAction(a, data)
+	r.log.Info("job added", "ID", jID)
+
+	return &protobuf.StreamResponse{JobID: ptr.Take(jID)}, nil
+}

runner/hls.go

@@ -0,0 +1,26 @@
+package runner
+
+import (
+	"log/slog"
+	"net/http"
+)
+
+type HLSServer struct {
+	log *slog.Logger
+	fs  http.Handler
+}
+
+func NewHLSServer(LiveDir string, log *slog.Logger) *HLSServer {
+	return &HLSServer{fs: http.FileServer(http.Dir(LiveDir)), log: log}
+}
+
+func (h *HLSServer) Start() error {
+	http.Handle("/", h)
+	h.log.Info("starting hls server", "port", 8187)
+	return http.ListenAndServe(":8187", h)
+}
+
+func (h *HLSServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	//w.Header().Set("Access-Control-Allow-Origin", "*")
+	h.fs.ServeHTTP(w, r)
+}