fix source playback for storj recordings (#858)

livepeer · Sep 13, 2023 · 506e24f · 506e24f
1 parent 0794951
commit 506e24f
Show file tree

Hide file tree

Showing 2 changed files with 53 additions and 16 deletions.
diff --git a/pipeline/ffmpeg.go b/pipeline/ffmpeg.go
@@ -156,14 +156,34 @@ func (f *ffmpeg) sendSourcePlayback(job *JobInfo) {
 		return
 	}
 
+	sourceURL, err := url.Parse(job.SourceFile)
+	if err != nil {
+		log.LogError(job.RequestID, "unable to parse source url for source playback", err)
+		return
+	}
+
+	prefix := ""
+	if clients.IsHLSInput(sourceURL) {
+		for k, v := range f.sourcePlaybackHosts {
+			if strings.HasPrefix(job.SourceFile, k) {
+				prefix = strings.Replace(job.SourceFile, k, v, 1)
+				break
+			}
+		}
+		if prefix == "" {
+			log.Log(job.RequestID, "no source playback prefix found", "host", sourceURL.Host)
+			return
+		}
+	}
+
 	segmentingPath := strings.Split(segmentingTargetURL.Path, "/")
 	if len(segmentingPath) < 3 || segmentingPath[1] == "" {
 		log.Log(job.RequestID, "unable to find bucket for source playback", "segmentingTargetURL", segmentingTargetURL)
 		return
 	}
 	// assume bucket is second element in slice (first element should be an empty string as the path has a leading slash)
 	segmentingBucket := segmentingPath[1]
-	if job.HlsTargetURL == nil || !strings.Contains(job.HlsTargetURL.String(), "/"+segmentingBucket+"/") {
+	if (job.HlsTargetURL == nil || !strings.Contains(job.HlsTargetURL.String(), "/"+segmentingBucket+"/")) && prefix == "" {
 		log.Log(job.RequestID, "source playback not available, not a studio job", "segmentingTargetURL", segmentingTargetURL)
 		return
 	}
@@ -183,19 +203,10 @@ func (f *ffmpeg) sendSourcePlayback(job *JobInfo) {
 		return
 	}
 
-	sourceURL, err := url.Parse(job.SourceFile)
-	if err != nil {
-		log.LogError(job.RequestID, "unable to parse source url for source playback", err)
-		return
-	}
-
-	prefix := f.sourcePlaybackHosts[sourceURL.Host]
-	if clients.IsHLSInput(sourceURL) && prefix == "" {
-		log.Log(job.RequestID, "no source playback prefix found", "host", sourceURL.Host)
-		return
+	if prefix == "" {
+		prefix = "/" + path.Join(segmentingPath[2:]...)
 	}
-
-	sourceMaster.Append(prefix+"/"+path.Join(segmentingPath[2:]...), &m3u8.MediaPlaylist{}, m3u8.VariantParams{
+	sourceMaster.Append(prefix, &m3u8.MediaPlaylist{}, m3u8.VariantParams{
 		Bandwidth:  uint32(videoTrack.Bitrate),
 		Resolution: fmt.Sprintf("%dx%d", videoTrack.Width, videoTrack.Height),
 		Name:       fmt.Sprintf("%dp", videoTrack.Height),

diff --git a/pipeline/ffmpeg_test.go b/pipeline/ffmpeg_test.go
@@ -57,7 +57,9 @@ func Test_sendSourcePlayback(t *testing.T) {
 	}
 	ff := ffmpeg{
 		sourcePlaybackHosts: map[string]string{
-			"lp-us-catalyst-recordings-monster.storage.googleapis.com": "//recordings-cdn.lp-playback.monster/hls",
+			"http://lp-us-catalyst-recordings-monster.storage.googleapis.com/foo":                         "//recordings-cdn.lp-playback.monster/hls",
+			"https://link.storjshare.io/raw/jvnqoncawzmc3lb7tstb5ut3d7va/catalyst-recordings-monster/hls": "//link.storjshare.io/raw/jvnqoncawzmc3lb7tstb5ut3d7va/catalyst-recordings-monster/hls",
+			"https://link.storjshare.io/raw/jvnqoncawzmc3lb7tstb5ut3d7vb/catalyst-recordings-monster/hls": "//recordings-cdn.lp-playback.monster/hls",
 		},
 	}
 
@@ -101,12 +103,36 @@ func Test_sendSourcePlayback(t *testing.T) {
 			job: &JobInfo{
 				SegmentingTargetURL: segmentingTargetURL,
 				UploadJobPayload: UploadJobPayload{
-					SourceFile:   "http://lp-us-catalyst-recordings-monster.storage.googleapis.com/foo/bar",
+					SourceFile:   "http://lp-us-catalyst-recordings-monster.storage.googleapis.com/foo/bar/output.m3u8",
 					HlsTargetURL: mustParseUrl("/bucket/foo", t),
 				},
 			},
 			shouldWriteSourcePlaylist: true,
-			expectedRendition:         "//recordings-cdn.lp-playback.monster/hls/path",
+			expectedRendition:         "//recordings-cdn.lp-playback.monster/hls/bar/output.m3u8",
+		},
+		{
+			name: "host mapping - storj",
+			job: &JobInfo{
+				SegmentingTargetURL: "https://link.storjshare.io/raw/jvnqoncawzmc3lb7tstb5ut3d7va/catalyst-recordings-monster/hls/e88briv8dl7rzg8o-test/3c446cbe-3ca9-4eba-84a9-68b38305d67a/output.m3u8",
+				UploadJobPayload: UploadJobPayload{
+					SourceFile:   "https://link.storjshare.io/raw/jvnqoncawzmc3lb7tstb5ut3d7va/catalyst-recordings-monster/hls/e88briv8dl7rzg8o-test/3c446cbe-3ca9-4eba-84a9-68b38305d67a/output.m3u8",
+					HlsTargetURL: mustParseUrl("/bucket/foo", t),
+				},
+			},
+			shouldWriteSourcePlaylist: true,
+			expectedRendition:         "//link.storjshare.io/raw/jvnqoncawzmc3lb7tstb5ut3d7va/catalyst-recordings-monster/hls/e88briv8dl7rzg8o-test/3c446cbe-3ca9-4eba-84a9-68b38305d67a/output.m3u8",
+		},
+		{
+			name: "host mapping - storj cdn",
+			job: &JobInfo{
+				SegmentingTargetURL: "https://link.storjshare.io/raw/jvnqoncawzmc3lb7tstb5ut3d7vb/catalyst-recordings-monster/hls/e88briv8dl7rzg8o-test/3c446cbe-3ca9-4eba-84a9-68b38305d67a/output.m3u8",
+				UploadJobPayload: UploadJobPayload{
+					SourceFile:   "https://link.storjshare.io/raw/jvnqoncawzmc3lb7tstb5ut3d7vb/catalyst-recordings-monster/hls/e88briv8dl7rzg8o-test/3c446cbe-3ca9-4eba-84a9-68b38305d67a/output.m3u8",
+					HlsTargetURL: mustParseUrl("/bucket/foo", t),
+				},
+			},
+			shouldWriteSourcePlaylist: true,
+			expectedRendition:         "//recordings-cdn.lp-playback.monster/hls/e88briv8dl7rzg8o-test/3c446cbe-3ca9-4eba-84a9-68b38305d67a/output.m3u8",
 		},
 		{
 			name: "not standard bucket - no source playback",