Add converting after stream (MKVod action) (#1605)

* Added video conversion

* Fixed a bug

* Changed logging behaviour

* Changed logging behaviour 2

Author: SebiWrn

runner/handlers.go

@@ -30,7 +30,7 @@ func (r *Runner) RequestStream(ctx context.Context, req *protobuf.StreamRequest)
 		actions.MkVOD,
 	}
 
-	jID := r.RunAction(a, data)
+	jID := r.RunAction(a, data, r.log.With("stream_id", req.GetStreamId(), "stream_version", req.GetVersion(), "input", req.GetInput()))
 	r.log.Info("job added", "ID", jID)
 
 	return &protobuf.StreamResponse{JobId: ptr.Take(jID)}, nil

runner/pkg/actions/mkvod.go

@@ -9,10 +9,10 @@ import (
 	"log/slog"
 	"net/url"
 	"os"
+	"os/exec"
 	"path"
 	"strings"
-
-	log "github.com/sirupsen/logrus"
+	"time"
 
 	"github.com/tum-dev/gocast/runner/config"
 	"github.com/tum-dev/gocast/runner/pkg/metrics"
@@ -21,7 +21,7 @@ import (
 
 // MkVOD takes a stream that was streamed and moves the hls stream to long term storage.
 // Additionally, the playlist type is transformed from event to VOD.
-func MkVOD(_ context.Context, _ *slog.Logger, notify chan *protobuf.Notification, d map[string]any, metrics *metrics.Broker) error {
+func MkVOD(ctx context.Context, logger *slog.Logger, notify chan *protobuf.Notification, d map[string]any, metrics *metrics.Broker) error {
 	streamID, ok := d["streamID"].(uint64)
 	if !ok {
 		return AbortingError(fmt.Errorf("no stream id in context"))
@@ -35,6 +35,11 @@ func MkVOD(_ context.Context, _ *slog.Logger, notify chan *protobuf.Notification
 		return AbortingError(fmt.Errorf("no recordingDir in context"))
 	}
 
+	metrics.ConvertingProgresses.With(metrics.With().Stream(streamID).L()).Inc()
+	defer func() {
+		metrics.ConvertingProgresses.With(metrics.With().Stream(streamID).L()).Dec()
+	}()
+
 	vodDir := path.Join(config.Config.StoragePath, fmt.Sprintf("%d", streamID), streamVersion)
 	err := os.MkdirAll(vodDir, os.ModePerm)
 	if err != nil {
@@ -48,27 +53,13 @@ func MkVOD(_ context.Context, _ *slog.Logger, notify chan *protobuf.Notification
 
 	for _, entry := range recordingContent {
 		if entry.IsDir() {
-			log.Warn("found dir in recordingDir, skipping", "name", entry.Name())
+			logger.Warn("found dir in recordingDir, skipping", "name", entry.Name())
 			continue
 		}
 		if entry.Name() == "playlist.m3u8" {
-			srcPlst, err := os.Open(path.Join(recordingDir, entry.Name()))
-			if err != nil {
-				return AbortingError(fmt.Errorf("open recording playlist: %w", err))
-			}
-			dstPlstS := vodFromEventPlst(srcPlst)
-			dstPlst, err := os.Create(path.Join(vodDir, entry.Name()))
-			if err != nil {
-				return AbortingError(fmt.Errorf("create vod playlist: %w", err))
-			}
-			_, err = io.WriteString(dstPlst, dstPlstS)
-			if err != nil {
-				return fmt.Errorf("write vod to playlist: %w", err)
-			}
-			_ = dstPlst.Close()
+			err = convertStream(ctx, logger, streamID, path.Join(recordingDir, entry.Name()), vodDir, entry.Name())
 			continue
 		}
-		err = copyFile(path.Join(recordingDir, entry.Name()), path.Join(vodDir, entry.Name()))
 	}
 
 	vodUrl, err := url.JoinPath(config.Config.EdgeServer, fmt.Sprintf("%d", streamID), streamVersion, "playlist.m3u8")
@@ -87,24 +78,36 @@ func MkVOD(_ context.Context, _ *slog.Logger, notify chan *protobuf.Notification
 	return nil
 }
 
-func copyFile(sourcePath, destPath string) error {
-	inputFile, err := os.Open(sourcePath)
+func convertStream(ctx context.Context, logger *slog.Logger, streamID uint64, streamPath, vodDir string, playlistName string) error {
+	input := "-i " + streamPath
+	options := "-c copy -f hls -hls_time 240 -hls_playlist_type event -hls_flags append_list -hls_segment_filename " + path.Join(vodDir, "%05d.ts") + " " + path.Join(vodDir, playlistName)
+
+	args := strings.Split(input, " ")
+	args = append(args, strings.Split(options, " ")...)
+	command := exec.CommandContext(ctx, "ffmpeg", args...)
+
+	// give ffmpeg 10 seconds on sigterm (context cancellation) to shut down before sending sigkill.
+	command.WaitDelay = 10 * time.Second
+
+	logger.Info("starting ffmpeg", "command", command.String())
+	stderr, err := command.StderrPipe()
 	if err != nil {
-		return fmt.Errorf("open source file: %w", err)
+		return err
 	}
-	defer inputFile.Close()
+	go logCmdPipe(logger, stderr, []any{"stream", streamID, "logStream", "stderr"})
 
-	outputFile, err := os.Create(destPath)
+	stdo, err := command.StdoutPipe()
 	if err != nil {
-		return fmt.Errorf("open dest file: %w", err)
+		return err
 	}
-	defer outputFile.Close()
-
-	_, err = io.Copy(outputFile, inputFile)
+	go logCmdPipe(logger, stdo, []any{"stream", streamID, "logStream", "stdout"})
+	err = command.Run()
 	if err != nil {
-		return fmt.Errorf("copy to dest from source: %w", err)
+		logger.Error("ffmpeg command failed", "error", err)
+	} else {
+		logger.Info("ffmpeg converting completed successfully", "stream_id", streamID)
 	}
-	return nil
+	return err
 }
 
 // vodFromEventPlst modifies an HLS playlist from EVENT to VOD

runner/pkg/actions/stream.go

@@ -50,7 +50,6 @@ func Stream(ctx context.Context, log *slog.Logger, notify chan *protobuf.Notific
 	if !ok {
 		return AbortingError(fmt.Errorf("no input in context"))
 	}
-	log = log.With("stream_id", streamID)
 
 	metrics.Streams.With(metrics.With().Stream(streamID).Source(input).L()).Inc()
 	defer func() {

runner/pkg/metrics/broker.go

@@ -13,9 +13,10 @@ import (
 
 // Broker manages Prometheus metrics.
 type Broker struct {
-	port         int
-	Streams      *prometheus.GaugeVec
-	StreamErrors *prometheus.CounterVec
+	port                 int
+	Streams              *prometheus.GaugeVec
+	StreamErrors         *prometheus.CounterVec
+	ConvertingProgresses *prometheus.GaugeVec
 }
 
 // Option represents a functional option for configuring a Broker.
@@ -45,7 +46,14 @@ func NewBroker(options ...Option) *Broker {
 			Subsystem: "stream",
 			Name:      "n_errors",
 			Help:      "Number of stream ffmpeg errors",
-		}, []string{"stream_id", "source"}),
+		}, []string{"stream_id"}),
+
+		ConvertingProgresses: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: "runner",
+			Subsystem: "converting",
+			Name:      "n_converting",
+			Help:      "Number of streams currently being converted",
+		}, []string{"stream_id"}),
 	}
 	for _, option := range options {
 		option(b)

runner/runner.go

@@ -155,7 +155,7 @@ func (r *Runner) InitApiGrpc() {
 	}
 }
 
-func (r *Runner) RunAction(a []actions.Action, data map[string]any) string {
+func (r *Runner) RunAction(a []actions.Action, data map[string]any, logger *slog.Logger) string {
 	// create new context to avoid cancellation on grpc request termination
 	c, cancel := context.WithCancel(context.Background())
 	job := uuid.New().String()
@@ -169,7 +169,7 @@ func (r *Runner) RunAction(a []actions.Action, data map[string]any) string {
 		}()
 		for _, action := range a {
 			for {
-				log := r.log.With("action", getFunctionName(action)).With("job", job)
+				log := logger.With("action", getFunctionName(action)).With("job", job)
 				log.Info("running action")
 				s := time.Now()
 				err := action(c, log, r.notifications, data, r.Metrics)