Fix short segment length (#909)

* Fix short segment length * set playlist type to vod --------- Co-authored-by: Thom Shutt <[email protected]>
livepeer · Oct 12, 2023 · 33d192e · 33d192e
1 parent 4c3cd7e
commit 33d192e
Show file tree

Hide file tree

Showing 4 changed files with 41 additions and 13 deletions.
diff --git a/api/http_internal.go b/api/http_internal.go
@@ -122,7 +122,7 @@ func NewCatalystAPIRouterInternal(cli config.Cli, vodEngine *pipeline.Coordinato
 	broker.OnUserEnd(analyticsHandlers.HandleUserEnd)
 
 	// Endpoint to receive segments and manifests that ffmpeg produces
-	router.PUT("/api/ffmpeg/:id/:filename", withLogging(ffmpegSegmentingHandlers.NewFile()))
+	router.POST("/api/ffmpeg/:id/:filename", withLogging(ffmpegSegmentingHandlers.NewFile()))
 
 	// Temporary endpoint for admin queries
 	router.GET("/admin/members", withLogging(adminHandlers.MembersHandler()))

diff --git a/handlers/ffmpeg/ffmpeg.go b/handlers/ffmpeg/ffmpeg.go
@@ -1,13 +1,17 @@
 package ffmpeg
 
 import (
+	"bytes"
+	"io"
 	"net/http"
 	"regexp"
 
+	"github.com/grafov/m3u8"
 	"github.com/julienschmidt/httprouter"
 	"github.com/livepeer/catalyst-api/clients"
 	"github.com/livepeer/catalyst-api/config"
 	"github.com/livepeer/catalyst-api/errors"
+	"github.com/livepeer/catalyst-api/log"
 	"github.com/livepeer/catalyst-api/pipeline"
 )
 
@@ -32,14 +36,34 @@ func (h *HandlersCollection) NewFile() httprouter.Handle {
 			return
 		}
 
+		var body io.Reader = req.Body
+		reg := regexp.MustCompile(`[^/]+.m3u8$`)
+		if reg.MatchString(filename) {
+			// ensure that playlist type in the manifest is set to vod
+			buf := bytes.Buffer{}
+			_, err := buf.ReadFrom(req.Body)
+			if err != nil {
+				errors.WriteHTTPInternalServerError(w, "Error reading body", err)
+				return
+			}
+
+			playlist, playlistType, err := m3u8.Decode(buf, true)
+			if err != nil {
+				log.LogError(job.RequestID, "failed to parse segmented manifest", err)
+				body = &buf
+			} else if playlistType == m3u8.MEDIA {
+				mediaPl := playlist.(*m3u8.MediaPlaylist)
+				mediaPl.MediaType = m3u8.VOD
+				body = mediaPl.Encode()
+			}
+		}
 		// job.SegmentingTargetURL comes in the format the Mist wants, looking like:
 		//   protocol://abc@123:s3.com/a/b/c/<something>.m3u8
 		// but since this endpoint receives both .ts segments and m3u8 updates, we strip off the filename
 		// and pass the one ffmpeg gives us to UploadToOSURL instead
-		reg := regexp.MustCompile(`[^/]+.m3u8$`)
 		targetURLBase := reg.ReplaceAllString(job.SegmentingTargetURL, "")
 
-		if err := clients.UploadToOSURL(targetURLBase, filename, req.Body, config.SEGMENT_WRITE_TIMEOUT); err != nil {
+		if err := clients.UploadToOSURL(targetURLBase, filename, body, config.SEGMENT_WRITE_TIMEOUT); err != nil {
 			errors.WriteHTTPInternalServerError(w, "Error uploading segment", err)
 			return
 		}

diff --git a/test/steps/ffmpeg.go b/test/steps/ffmpeg.go
@@ -80,8 +80,11 @@ func (s *StepContext) TheSourceManifestIsWrittenToStorageWithinSeconds(secs, num
 	if mediaManifest.Version() != 3 {
 		return fmt.Errorf("expected manifest to be HLSv3 but got version: %d", mediaManifest.Version())
 	}
-	if mediaManifest.TargetDuration != 10.0 {
-		return fmt.Errorf("expected manifest to have a Target Duration of 10 but got: %f", mediaManifest.TargetDuration)
+	if mediaManifest.TargetDuration != 11.0 {
+		return fmt.Errorf("expected manifest to have a Target Duration of 11 but got: %f", mediaManifest.TargetDuration)
+	}
+	if mediaManifest.MediaType != m3u8.VOD {
+		return fmt.Errorf("expected manifest to have playlist type VOD but got: %v", mediaManifest.MediaType)
 	}
 
 	return nil

diff --git a/video/segment.go b/video/segment.go
@@ -2,11 +2,12 @@ package video
 
 import (
 	"fmt"
+	"strings"
 
 	ffmpeg "github.com/u2takey/ffmpeg-go"
 )
 
-// Split a source video URL into segments
+// Segment splits a source video URL into segments
 //
 // FFMPEG can use remote files, but depending on the layout of the file can get bogged
 // down and end up making multiple range requests per segment.
@@ -15,16 +16,16 @@ func Segment(sourceFilename string, outputManifestURL string, targetSegmentSize
 	// Do the segmenting, using the local file as source
 	err := ffmpeg.Input(sourceFilename).
 		Output(
-			outputManifestURL,
+			strings.Replace(outputManifestURL, ".m3u8", "", 1)+"%d.ts",
 			ffmpeg.KwArgs{
 				"c:a":               "copy",
 				"c:v":               "copy",
-				"f":                 "hls",
-				"hls_segment_type":  "mpegts",
-				"hls_playlist_type": "vod",
-				"hls_list_size":     "0",
-				"hls_time":          targetSegmentSize,
-				"method":            "PUT",
+				"f":                 "segment",
+				"segment_list":      outputManifestURL,
+				"segment_list_type": "m3u8",
+				"segment_format":    "mpegts",
+				"segment_time":      targetSegmentSize,
+				"min_seg_duration":  "2",
 			},
 		).OverWriteOutput().ErrorToStdOut().Run()
 	if err != nil {