feat(runner/cleanup): start cleanup and vod sanity check action

runner/handlers.go

@@ -28,6 +28,8 @@ func (r *Runner) RequestStream(ctx context.Context, req *protobuf.StreamRequest)
 		actions.Stream,
 		actions.StreamEnd,
 		actions.MkVOD,
+		actions.ProbeVodHealth,
+		actions.Cleanup,
 	}
 
 	jID := r.RunAction(a, data)
@@ -37,10 +39,14 @@ func (r *Runner) RequestStream(ctx context.Context, req *protobuf.StreamRequest)
 }
 
 func (r *Runner) RequestStreamEnd(_ context.Context, req *protobuf.StreamEndRequest) (*protobuf.StreamEndResponse, error) {
-	cancel, ok := r.jobs[req.GetJobId()]
-	if ok {
-		cancel()
-		return nil, nil
+	contexts, ok := r.jobs[req.GetJobId()]
+	if !ok {
+		return nil, status.Errorf(codes.NotFound, "stream not found")
 	}
-	return nil, status.Errorf(codes.NotFound, "stream not found")
+	cf, err := contexts.GetStreamCancelFunc()
+	if err != nil {
+		return nil, status.Errorf(codes.NotFound, "action cancelation not possible: %s", err)
+	}
+	cf()
+	return nil, nil
 }

runner/pkg/actions/cleanup.go

@@ -0,0 +1,112 @@
+package actions
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"math"
+	"os"
+	"os/exec"
+	"path"
+
+	"github.com/tum-dev/gocast/runner/config"
+	"github.com/tum-dev/gocast/runner/pkg/metrics"
+	"github.com/tum-dev/gocast/runner/pkg/probe"
+	"github.com/tum-dev/gocast/runner/protobuf"
+)
+
+// Cleanup is an action that removes all contents in recordingDir if vodHealthy is true.
+// otherwise, it uses the mv command, to move the broken live recording to {mass_dir}/broken.
+func Cleanup(ctx context.Context, log *slog.Logger, _ chan *protobuf.Notification, d map[string]any, metrics *metrics.Broker) error {
+	recDir, ok := d["recordingDir"].(string)
+	if !ok {
+		return AbortingError(fmt.Errorf("no recordingDir in context"))
+	}
+	vodHealthy, ok := d["vodHealthy"].(bool)
+	if !ok {
+		return AbortingError(fmt.Errorf("no recordingDir in context"))
+	}
+	if vodHealthy {
+		err := os.RemoveAll(recDir)
+		if err != nil {
+			return AbortingError(fmt.Errorf("remove vod dir: %w", err))
+		}
+		return nil
+	}
+	// vod unhealthy, move to mass:
+	dst := path.Join(recDir, config.Config.StoragePath, "broken")
+	err := os.MkdirAll(dst, os.ModePerm)
+	if err != nil && !os.IsExist(err) {
+		return AbortingError(fmt.Errorf("create broken vod directory: %w", err))
+	}
+	log.Info("moving broken vod", "scr", recDir, "dst", dst)
+	cmd := exec.CommandContext(ctx, "mv", dst)
+	err = cmd.Start()
+	if err != nil {
+		return AbortingError(fmt.Errorf("start move broken recording command: %w", err))
+	}
+	err = cmd.Wait()
+	if err != nil {
+		return AbortingError(fmt.Errorf("wait move broken recording command: %w", err))
+	}
+	return nil
+}
+
+// ProbeVodHealth checks the duration of the live and vod playlists and sets d["vodHealthy"] to false if it detects
+// a difference of the stream lengths of more than 10 seconds or if any of the probes fail or return no streams.
+func ProbeVodHealth(ctx context.Context, log *slog.Logger, _ chan *protobuf.Notification, d map[string]any, metrics *metrics.Broker) error {
+	recDir, ok := d["recordingDir"].(string)
+	if !ok {
+		return AbortingError(fmt.Errorf("no recordingDir in context"))
+	}
+	vodDir, ok := d["vodDir"].(string)
+	if !ok {
+		return AbortingError(fmt.Errorf("no vodDir in context"))
+	}
+
+	probeLive, err := probe.Probe(ctx, path.Join(recDir, "playlist.m3u8"))
+	if err != nil {
+		log.Error("probe livestream", "err", err)
+		d["vodHealthy"] = false
+		return nil
+	}
+	probeVod, err := probe.Probe(ctx, path.Join(vodDir, "playlist.m3u8"))
+	if err != nil {
+		log.Error("probe vod", "err", err)
+		d["vodHealthy"] = false
+		return nil
+	}
+
+	if len(probeLive.Streams) == 0 {
+		log.Error("live streams # == 0")
+		d["vodHealthy"] = false
+		return nil
+	}
+
+	if len(probeVod.Streams) == 0 {
+		log.Error("vod streams # == 0")
+		d["vodHealthy"] = false
+		return nil
+	}
+
+	durationLive, err := probeLive.Streams[0].DurationFloat()
+	if err != nil {
+		log.Error("parse live stream duration", "err", err)
+		d["vodHealthy"] = false
+		return nil
+	}
+	durationVod, err := probeVod.Streams[0].DurationFloat()
+	if err != nil {
+		log.Error("parse vod stream duration", "err", err)
+		d["vodHealthy"] = false
+		return nil
+	}
+	if math.Max(durationVod, durationLive)-math.Min(durationVod, durationLive) > 10 {
+		log.Error("vod and live durations of by > 10s", "vodDuration", durationVod, "liveDuration", durationLive)
+		d["vodHealthy"] = false
+		return nil
+	}
+	log.Info("vod is healthy")
+	d["vodHealthy"] = true
+	return nil
+}

runner/pkg/actions/mkvod.go

@@ -36,8 +36,10 @@ func MkVOD(_ context.Context, _ *slog.Logger, notify chan *protobuf.Notification
 	}
 
 	vodDir := path.Join(config.Config.StoragePath, fmt.Sprintf("%d", streamID), streamVersion)
+	d["vodDir"] = vodDir
+
 	err := os.MkdirAll(vodDir, os.ModePerm)
-	if err != nil {
+	if err != nil && !os.IsExist(err) {
 		return AbortingError(fmt.Errorf("create VOD directory: %w", err))
 	}
 

runner/pkg/actions/stream.go

@@ -84,7 +84,14 @@ func Stream(ctx context.Context, log *slog.Logger, notify chan *protobuf.Notific
 	args = append(args, strings.Split(outputOpts, " ")...)
 	args = append(args, strings.Split(`-f hls -hls_time 2 -hls_playlist_type event -hls_flags append_list -hls_segment_filename `+liveRecDir+"/%05d.ts "+liveRecDir+"/playlist.m3u8", " ")...)
 
-	command := exec.CommandContext(ctx, "ffmpeg", args...)
+	// remove empty args
+	args2 := make([]string, 0, len(args))
+	for i := range args {
+		if args[i] != "" {
+			args2 = append(args2, args[i])
+		}
+
+	command := exec.CommandContext(ctx, "ffmpeg", args2...)
 	// give ffmpeg 10 seconds on sigterm (context cancellation) to shut down before sending sigkill.
 	command.WaitDelay = 10 * time.Second
 

runner/pkg/probe/probe.go

@@ -0,0 +1,129 @@
+// Package probe provides tooling to probe videos using ffprobe
+package probe
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os/exec"
+	"strconv"
+)
+
+// Probe gets a P from input
+func Probe(ctx context.Context, input string) (P, error) {
+	var res P
+	cmd := exec.CommandContext(ctx, "ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", input)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		return res, err
+	}
+	err = json.Unmarshal(out, &res)
+	if err != nil {
+		return res, fmt.Errorf("unmarshal: %w", err)
+	}
+	return res, nil
+}
+
+// P is the Probe result
+type P struct {
+	Streams []Stream `json:"streams"`
+	Format  struct {
+		Filename       string `json:"filename"`
+		NbStreams      int    `json:"nb_streams"`
+		NbPrograms     int    `json:"nb_programs"`
+		NbStreamGroups int    `json:"nb_stream_groups"`
+		FormatName     string `json:"format_name"`
+		FormatLongName string `json:"format_long_name"`
+		StartTime      string `json:"start_time"`
+		Duration       string `json:"duration"`
+		Size           string `json:"size"`
+		BitRate        string `json:"bit_rate"`
+		ProbeScore     int    `json:"probe_score"`
+		Tags           struct {
+			Encoder string `json:"encoder"`
+		} `json:"tags"`
+	} `json:"format"`
+}
+
+type Stream struct {
+	Index              int    `json:"index"`
+	CodecName          string `json:"codec_name"`
+	CodecLongName      string `json:"codec_long_name"`
+	Profile            string `json:"profile,omitempty"`
+	CodecType          string `json:"codec_type"`
+	CodecTagString     string `json:"codec_tag_string"`
+	CodecTag           string `json:"codec_tag"`
+	Width              int    `json:"width,omitempty"`
+	Height             int    `json:"height,omitempty"`
+	CodedWidth         int    `json:"coded_width,omitempty"`
+	CodedHeight        int    `json:"coded_height,omitempty"`
+	ClosedCaptions     int    `json:"closed_captions,omitempty"`
+	FilmGrain          int    `json:"film_grain,omitempty"`
+	HasBFrames         int    `json:"has_b_frames,omitempty"`
+	SampleAspectRatio  string `json:"sample_aspect_ratio,omitempty"`
+	DisplayAspectRatio string `json:"display_aspect_ratio,omitempty"`
+	PixFmt             string `json:"pix_fmt,omitempty"`
+	Level              int    `json:"level,omitempty"`
+	ColorRange         string `json:"color_range,omitempty"`
+	ColorSpace         string `json:"color_space,omitempty"`
+	ColorTransfer      string `json:"color_transfer,omitempty"`
+	ColorPrimaries     string `json:"color_primaries,omitempty"`
+	ChromaLocation     string `json:"chroma_location,omitempty"`
+	FieldOrder         string `json:"field_order,omitempty"`
+	Refs               int    `json:"refs,omitempty"`
+	IsAvc              string `json:"is_avc,omitempty"`
+	NalLengthSize      string `json:"nal_length_size,omitempty"`
+	RFrameRate         string `json:"r_frame_rate"`
+	AvgFrameRate       string `json:"avg_frame_rate"`
+	TimeBase           string `json:"time_base"`
+	StartPts           int    `json:"start_pts"`
+	StartTime          string `json:"start_time"`
+	BitsPerRawSample   string `json:"bits_per_raw_sample,omitempty"`
+	ExtradataSize      int    `json:"extradata_size,omitempty"`
+	Disposition        struct {
+		Default         int `json:"default"`
+		Dub             int `json:"dub"`
+		Original        int `json:"original"`
+		Comment         int `json:"comment"`
+		Lyrics          int `json:"lyrics"`
+		Karaoke         int `json:"karaoke"`
+		Forced          int `json:"forced"`
+		HearingImpaired int `json:"hearing_impaired"`
+		VisualImpaired  int `json:"visual_impaired"`
+		CleanEffects    int `json:"clean_effects"`
+		AttachedPic     int `json:"attached_pic"`
+		TimedThumbnails int `json:"timed_thumbnails"`
+		NonDiegetic     int `json:"non_diegetic"`
+		Captions        int `json:"captions"`
+		Descriptions    int `json:"descriptions"`
+		Metadata        int `json:"metadata"`
+		Dependent       int `json:"dependent"`
+		StillImage      int `json:"still_image"`
+		Multilayer      int `json:"multilayer"`
+	} `json:"disposition"`
+	Tags struct {
+		BPS                      string `json:"BPS"`
+		DURATION                 string `json:"DURATION"`
+		NUMBEROFFRAMES           string `json:"NUMBER_OF_FRAMES"`
+		NUMBEROFBYTES            string `json:"NUMBER_OF_BYTES"`
+		STATISTICSWRITINGAPP     string `json:"_STATISTICS_WRITING_APP"`
+		STATISTICSWRITINGDATEUTC string `json:"_STATISTICS_WRITING_DATE_UTC"`
+		STATISTICSTAGS           string `json:"_STATISTICS_TAGS"`
+		Language                 string `json:"language,omitempty"`
+		Title                    string `json:"title,omitempty"`
+	} `json:"tags"`
+	SampleFmt      string `json:"sample_fmt,omitempty"`
+	SampleRate     string `json:"sample_rate,omitempty"`
+	Channels       int    `json:"channels,omitempty"`
+	ChannelLayout  string `json:"channel_layout,omitempty"`
+	BitsPerSample  int    `json:"bits_per_sample,omitempty"`
+	InitialPadding int    `json:"initial_padding,omitempty"`
+	BitRate        string `json:"bit_rate,omitempty"`
+	DurationTs     int    `json:"duration_ts,omitempty"`
+	Duration       string `json:"duration,omitempty"`
+}
+
+// DurationFloat returns the duration of the given Stream in seconds.
+func (s Stream) DurationFloat() (float64, error) {
+	return strconv.ParseFloat(s.Duration, 64)
+}

runner/pkg/probe/probe_test.go

@@ -0,0 +1,27 @@
+package probe
+
+import (
+	"context"
+	"testing"
+)
+
+func TestProbe(t *testing.T) {
+	p, err := Probe(context.Background(), "../../testvid.mp4")
+	if err != nil {
+		t.Errorf("got error %v", err)
+		t.FailNow()
+	}
+	if len(p.Streams) != 1 {
+		t.Errorf("ffprobe len(streams), want: 1, got: %d", len(p.Streams))
+
+	}
+	if p.Streams[0].CodecName != "h264" {
+		t.Errorf("ffprobe stream codec name, want: h264, got: %s", p.Streams[0].CodecName)
+	}
+	if duration, err := p.Streams[0].DurationFloat(); err != nil {
+		t.Errorf("ffprobe get Duration: %v", err)
+	} else if duration != 1 {
+		t.Errorf("ffprobe stream duration, want: 1, got: %f", duration)
+	}
+
+}