Add selfstreaming feature to the runners

pkg/runner_manager/manager.go

@@ -6,10 +6,12 @@ import (
 	"fmt"
 	log "log/slog"
 	"net"
+	"os"
 	"strconv"
 	"strings"
 	"time"
 
+	"github.com/TUM-Dev/gocast/web"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
@@ -24,6 +26,10 @@ import (
 	"github.com/tum-dev/gocast/runner/protobuf"
 )
 
+var logger = log.New(log.NewJSONHandler(os.Stdout, &log.HandlerOptions{
+	Level: log.LevelDebug,
+})).With("service", "runner_manager")
+
 // Manager manages communication with runners and handles job distribution
 type Manager struct {
 	dao        dao.DaoWrapper
@@ -46,6 +52,12 @@ func (m *Manager) TriggerDueStreams() error {
 	log.Info("Triggering due streams")
 	ctx := context.Background()
 	streams, err := m.dao.GetDueStreamsForRunners()
+	// TODO: Remove this when turning off workers
+	if web.VersionTag == "development" {
+		workerStreams := m.dao.GetDueStreamsForWorkers()
+		streams = append(streams, workerStreams...)
+	}
+
 	log.Info(fmt.Sprintf("%d streams to start for runner", len(streams)))
 	if err != nil {
 		return err
@@ -240,15 +252,85 @@ func (m *Manager) requestStreamVersion(ctx context.Context, s model.Stream, clie
 	default:
 		return nil, fmt.Errorf("invalid stream version %v", version)
 	}
+
+	outputOptions := "-c:v libx264 -preset veryfast -tune zerolatency -c:a aac -ar 44100 -b:a 128k -g 60 -bufsize 5000k -maxrate 5000k -b:v 3500k"
+	if !s.IsSelfStream() {
+		switch lh.StreamProtocol {
+		case model.RTSP:
+			outputOptions = "-c:a copy -c:v copy -rtsp_transport tcp -preset veryfast -tune zerolatency"
+		case model.SRT:
+			outputOptions = "-c:a copy -c:v copy -preset veryfast -tune zerolatency"
+		}
+	}
+
+	var input string
+	if !s.IsSelfStream() {
+		switch lh.StreamProtocol {
+		case model.RTSP:
+			input = fmt.Sprintf("rtsp://%s", ip)
+		case model.SRT:
+			input = fmt.Sprintf("srt://%s", ip)
+		default:
+			return nil, fmt.Errorf("invalid stream protocol %v", lh.StreamProtocol)
+		}
+	} else {
+		// Lecture is Selfstream
+		course, err := m.dao.CoursesDao.GetCourseById(ctx, s.CourseID)
+		if err != nil {
+			return nil, err
+		}
+		input = fmt.Sprintf("rtmp://localhost/%s-%d", course.Slug, s.ID)
+	}
 	return client.RequestStream(ctx, &protobuf.StreamRequest{
 		StreamId:            ptr.Take(uint64(s.ID)),
 		Version:             ptr.Take(version),
 		End:                 timestamppb.New(s.End),
-		FfmpegOutputOptions: ptr.Take("-c:a copy -c:v copy"),
-		Input:               ptr.Take(fmt.Sprintf("srt://%s", ip)),
+		FfmpegOutputOptions: ptr.Take(outputOptions),
+		Input:               ptr.Take(input),
 	})
 }
 
+func (m *Manager) RequestSelfStream(ctx context.Context, request *protobuf.SelfStreamRequest) (*protobuf.SelfStreamResponse, error) {
+	if _, err := m.dao.RunnerDao.Get(ctx, *request.Hostname); err != nil {
+		logger.Warn("Could not find runner for selfstream", "hostname", *request.Hostname)
+		return nil, err
+	}
+
+	if *request.StreamKey == "" {
+		return nil, errors.New("no stream key")
+	}
+	stream, err := m.dao.StreamsDao.GetStreamByKey(ctx, *request.StreamKey)
+	if err != nil {
+		return nil, err
+	}
+	course, err := m.dao.CoursesDao.GetCourseById(ctx, stream.CourseID)
+	if err != nil {
+		return nil, err
+	}
+	if *request.CourseSlug != fmt.Sprintf("%s-%d", course.Slug, stream.ID) {
+		return nil, fmt.Errorf("bad stream name, should: %s, is: %s", fmt.Sprintf("%s-%d", course.Slug, stream.ID), *request.CourseSlug)
+	}
+	// reject streams that are more than 30 minutes in the future or more than 30 minutes past
+	if !(time.Now().After(stream.Start.Add(time.Minute*-30)) && time.Now().Before(stream.End.Add(time.Minute*30))) {
+		logger.Warn("Stream rejected, time out of bounds", "streamId", stream.ID)
+		return nil, errors.New("stream rejected")
+	}
+
+	client, err := m.getClient(ctx)
+	if err != nil {
+		logger.Error("Could not get client", "err", err)
+		return nil, err
+	}
+
+	resp, err := m.requestStreamVersion(ctx, stream, client, model.LectureHall{}, protobuf.StreamVersion_STREAM_VERSION_COMBINED)
+	if err != nil && !errors.Is(err, errNotNoLectureSource) {
+		logger.Error("Could not start selfstream", "err", err)
+		return nil, err
+	}
+	log.With("stream", stream.ID, "job", resp.JobId, "version", model.COMB).Info("started Stream")
+	return &protobuf.SelfStreamResponse{}, nil
+}
+
 func dialRunner(ctx context.Context, runner model.Runner) (*grpc.ClientConn, error) {
 	return grpc.NewClient(fmt.Sprintf("%s:%d", runner.Hostname, runner.Port), grpc.WithTransportCredentials(insecure.NewCredentials()))
 }

runner/cmd/runner/main.go

@@ -56,5 +56,6 @@ func main() {
 	}
 
 	blocking := make(chan struct{})
+	//nolint:all
 	_ = <-blocking
 }

runner/handlers.go

@@ -12,9 +12,7 @@ import (
 	"github.com/tum-dev/gocast/runner/protobuf"
 )
 
-func (r *Runner) RequestStream(ctx context.Context, req *protobuf.StreamRequest) (*protobuf.StreamResponse, error) {
-	ctx = context.Background()
-
+func (r *Runner) RequestStream(_ context.Context, req *protobuf.StreamRequest) (*protobuf.StreamResponse, error) {
 	data := map[string]any{
 		"streamID":      req.GetStreamId(),
 		"streamVersion": protobuf.StreamVersion_name[int32(req.GetVersion())],

runner/hls.go

@@ -12,8 +12,8 @@ type HLSServer struct {
 	version string
 }
 
-func NewHLSServer(LiveDir string, log *slog.Logger, version string) *HLSServer {
-	return &HLSServer{fs: http.FileServer(http.Dir(LiveDir)), log: log, version: version}
+func NewHLSServer(liveDir string, log *slog.Logger, version string) *HLSServer {
+	return &HLSServer{fs: http.FileServer(http.Dir(liveDir)), log: log, version: version}
 }
 
 func (h *HLSServer) Start() error {

runner/pkg/actions/stream_end.go

@@ -3,9 +3,9 @@ package actions
 import (
 	"context"
 	"fmt"
-	"github.com/tum-dev/gocast/runner/pkg/ptr"
 	"log/slog"
 
+	"github.com/tum-dev/gocast/runner/pkg/ptr"
 	"github.com/tum-dev/gocast/runner/pkg/metrics"
 	"github.com/tum-dev/gocast/runner/protobuf"
 )

runner/pkg/netutil/netutil.go

@@ -1,14 +1,27 @@
 package netutil
 
-import "net"
+import (
+	"log/slog"
+	"net"
+	"os"
+)
+
+var logger = slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
+	Level: slog.LevelDebug,
+})).With("service", "netutil")
 
 // GetFreePort returns a free port for tcp use.
 func GetFreePort() (port int, err error) {
 	var a *net.TCPAddr
 	if a, err = net.ResolveTCPAddr("tcp", "localhost:0"); err == nil {
 		var l *net.TCPListener
 		if l, err = net.ListenTCP("tcp", a); err == nil {
-			defer l.Close()
+			defer func(l *net.TCPListener) {
+				err := l.Close()
+				if err != nil {
+					logger.Error("failed to close listener: %v", "err", err)
+				}
+			}(l)
 			return l.Addr().(*net.TCPAddr).Port, nil
 		}
 	}