Add cucumber test

clients/manifest.go

@@ -56,6 +56,7 @@ func DownloadRenditionManifest(requestID, sourceManifestOSURL string) (m3u8.Medi
 	return *mediaPlaylist, nil
 }
 
+// TODO only call with backup on first call from coordinator. return the url of the playlist used so that we can replace the job input url
 func downloadManifestWithBackup(requestID, sourceManifestOSURL string) (m3u8.Playlist, m3u8.ListType, error) {
 	var playlist, playlistBackup m3u8.Playlist
 	var playlistType, playlistTypeBackup m3u8.ListType

go.mod

@@ -24,6 +24,7 @@ require (
 	github.com/livepeer/m3u8 v0.11.1
 	github.com/mileusna/useragent v1.3.4
 	github.com/minio/madmin-go v1.7.5
+	github.com/minio/minio-go/v7 v7.0.45
 	github.com/mmcloughlin/geohash v0.10.0
 	github.com/peterbourgon/ff/v3 v3.4.0
 	github.com/pquerna/cachecontrol v0.2.0
@@ -52,6 +53,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/cucumber/gherkin/go/v26 v26.2.0 // indirect
 	github.com/cucumber/messages/go/v21 v21.0.1 // indirect
+	github.com/dustin/go-humanize v1.0.0 // indirect
 	github.com/eventials/go-tus v0.0.0-20220610120217-05d0564bb571 // indirect
 	github.com/fatih/color v1.13.0 // indirect
 	github.com/go-logr/logr v1.2.4 // indirect
@@ -111,7 +113,7 @@ require (
 	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
 	github.com/miekg/dns v1.1.50 // indirect
-	github.com/minio/minio-go/v7 v7.0.45 // indirect
+	github.com/minio/md5-simd v1.1.2 // indirect
 	github.com/minio/sha256-simd v1.0.1 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
@@ -138,6 +140,7 @@ require (
 	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
 	github.com/secure-io/sio-go v0.3.1 // indirect
 	github.com/shoenig/go-m1cpu v0.1.6 // indirect
+	github.com/sirupsen/logrus v1.9.0 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/stretchr/objx v0.5.0 // indirect

go.sum

@@ -110,6 +110,8 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0 h1:HbphB4TFFXpv7MNrT52FGrrgVXF1owhMVTHFZIlnvd4=
 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0/go.mod h1:DZGJHZMqrU4JJqFAWUS2UO1+lbSKsdiOoYi9Zzey7Fc=
 github.com/disintegration/imaging v1.6.2/go.mod h1:44/5580QXChDfwIclfc/PCwrr44amcmDAg8hxG0Ewe4=
+github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
+github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
@@ -408,6 +410,7 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o
 github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/klauspost/compress v1.16.3 h1:XuJt9zzcnaz6a16/OU53ZjWp/v7/42WcR5t2a0PcNQY=
 github.com/klauspost/compress v1.16.3/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.4 h1:acbojRNwl3o09bUq+yDCtZFc1aiwaAAxtcn8YkZXnvk=
 github.com/klauspost/cpuid/v2 v2.2.4/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY=
@@ -489,6 +492,8 @@ github.com/mileusna/useragent v1.3.4/go.mod h1:3d8TOmwL/5I8pJjyVDteHtgDGcefrFUX4
 github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1/go.mod h1:pD8RvIylQ358TN4wwqatJ8rNavkEINozVn9DtGI3dfQ=
 github.com/minio/madmin-go v1.7.5 h1:IF8j2HR0jWc7msiOcy0KJ8EyY7Q3z+j+lsmSDksQm+I=
 github.com/minio/madmin-go v1.7.5/go.mod h1:3SO8SROxHN++tF6QxdTii2SSUaYSrr8lnE9EJWjvz0k=
+github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
+github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
 github.com/minio/minio-go/v7 v7.0.45 h1:g4IeM9M9pW/Lo8AGGNOjBZYlvmtlE1N5TQEYWXRWzIs=
 github.com/minio/minio-go/v7 v7.0.45/go.mod h1:nCrRzjoSUQh8hgKKtu3Y708OLvRLtuASMg2/nvmbarw=
 github.com/minio/sha256-simd v0.0.0-20190131020904-2d45a736cd16/go.mod h1:2FMWW+8GMoPweT6+pI63m9YE3Lmw4J71hV56Chs1E/U=
@@ -651,6 +656,8 @@ github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeV
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
+github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=
+github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/assertions v1.2.0 h1:42S6lae5dvLc7BrLu/0ugRtcFVjoJNMC/N3yZFZkDFs=
 github.com/smartystreets/assertions v1.2.0/go.mod h1:tcbTF8ujkAEcZ8TElKY+i30BzYlVhC/LOxJk7iOWnoo=
@@ -948,6 +955,7 @@ golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220708085239-5a0f0661e09d/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

pipeline/coordinator.go

@@ -290,6 +290,10 @@ func (c *Coordinator) StartUploadJob(p UploadJobPayload) {
 
 		// Update osTransferURL if needed
 		if clients.IsHLSInput(sourceURL) {
+			// TODO get hls manifest with backup and replace input url with the one returned
+			// TODO check all segments whether they're available on main url (the manifest being used could be the backup so segment check needs to handle both primary and backup urls)
+			// TODO if any were not available on main url then write a new manifest with absolute segment urls pointing to wherever the segment is available (primary or backup)
+
 			// Currently we only clip an HLS source (e.g recordings or transcoded asset)
 			if p.ClipStrategy.Enabled {
 				err := backoff.Retry(func() error {
@@ -323,8 +327,8 @@ func (c *Coordinator) StartUploadJob(p UploadJobPayload) {
 		if p.C2PA {
 			si.C2PA = c.C2PA
 		}
-		si.SourceFile = osTransferURL.String()  // OS URL used by mist
-		si.SignedSourceURL = signedNewSourceURL // http(s) URL used by mediaconvert
+		si.SourceFile = osTransferURL.String()  // OS URL used by ffmpeg pipeline
+		si.SignedSourceURL = signedNewSourceURL // http(s) URL used by mediaconvert pipeline
 		si.InputFileInfo = inputVideoProbe
 		si.GenerateMP4 = ShouldGenerateMP4(sourceURL, p.Mp4TargetURL, p.FragMp4TargetURL, p.Mp4OnlyShort, si.InputFileInfo.Duration)
 		si.DownloadDone = time.Now()

test/features/vod.feature

@@ -7,6 +7,7 @@ Feature: VOD Streaming
     Given the VOD API is running
     And the Client app is authenticated
     And an object store is available
+#    And a fallback object
     And Studio API server is running at "localhost:13000"
     And a Broadcaster is running at "localhost:18935"
     And Mediaconvert is running at "localhost:11111"
@@ -77,8 +78,8 @@ Feature: VOD Streaming
     Then I get an HTTP response with code "200"
     And I receive a Request ID in the response body
     And my "successful" vod request metrics get recorded
-    And the Broadcaster receives "3" segments for transcoding within "10" seconds
-    And "3" transcoded segments and manifests have been written to disk for profiles "270p0,low-bitrate" within "30" seconds
+    And the Broadcaster receives "<segment_count>" segments for transcoding within "10" seconds
+    And "<segment_count>" transcoded segments and manifests have been written to disk for profiles "270p0,low-bitrate" within "30" seconds
     And a source copy has not been written to disk
     And I receive a "success" callback within "30" seconds
     And thumbnails are written to storage within "10" seconds
@@ -91,15 +92,16 @@ Feature: VOD Streaming
       | source_codec_audio       | aac             |
       | source_codec_video       | h264            |
       | source_duration          | 30000           |
-      | source_segment_count     | 3               |
+      | source_segment_count     | <segment_count> |
       | state                    | completed       |
-      | transcoded_segment_count | 3               |
+      | transcoded_segment_count | <segment_count> |
 
     Examples:
-      | payload                                                                     |
-      | a valid ffmpeg upload vod request with a source manifest                    |
-      | a valid ffmpeg upload vod request with a source manifest and source copying |
-      | a valid ffmpeg upload vod request with a source manifest and thumbnails     |
+      | payload                                                                     | segment_count |
+      | a valid ffmpeg upload vod request with a source manifest                    | 3             |
+      | a valid ffmpeg upload vod request with a source manifest and source copying | 3             |
+      | a valid ffmpeg upload vod request with a source manifest and thumbnails     | 3             |
+      | a valid ffmpeg upload vod request with a source manifest from object store  | 4             |
 
   Scenario Outline: Submit an audio-only asset for ingestion
     When I submit to the internal "/api/vod" endpoint with "<payload>"

test/fixtures/rec-fallback-bucket/tiny.m3u8

@@ -0,0 +1,13 @@
+#EXTM3U
+#EXT-X-VERSION:3
+#EXT-X-MEDIA-SEQUENCE:0
+#EXT-X-TARGETDURATION:10
+#EXTINF:10.000000,
+seg-0.ts
+#EXTINF:10.000000,
+seg-1.ts
+#EXTINF:10.000000,
+seg-2.ts
+#EXTINF:10.000000,
+seg-3.ts
+#EXT-X-ENDLIST

test/steps/http.go

@@ -163,6 +163,10 @@ func (s *StepContext) postRequest(baseURL, endpoint, payload string, headers map
 	}
 	if strings.HasPrefix(payload, "a valid ffmpeg upload vod request with a source manifest") {
 		req.URL = "file://" + filepath.Join(sourceManifestDir, "tiny.m3u8")
+		if strings.Contains(payload, "from object store") {
+			req.URL = "http://" + minioAddress + "/rec-bucket/tiny.m3u8"
+		}
+
 		req.PipelineStrategy = "catalyst_ffmpeg"
 		req.OutputLocations = []OutputLocation{
 			{
@@ -220,10 +224,10 @@ func (s *StepContext) StartApp() error {
 		"-cluster-addr=127.0.0.1:19935",
 		"-broadcaster-url=http://127.0.0.1:18935",
 		`-metrics-db-connection-string=`+DB_CONNECTION_STRING,
-		"-private-bucket",
-		"fixtures/playback-bucket",
+		"-private-bucket=fixtures/playback-bucket",
 		"-gate-url=http://localhost:13000/api/access-control/gate",
 		"-external-transcoder=mediaconverthttp://examplekey:[email protected]:11111?region=us-east-1&role=arn:aws:iam::exampleaccountid:examplerole&s3_aux_bucket=s3://example-bucket",
+		"-storage-fallback-urls=http://127.0.0.1:9000/rec-bucket=http://127.0.0.1:9000/rec-fallback-bucket",
 		"-source-output",
 		s.SourceOutputDir,
 		"-no-mist",

test/steps/init.go

@@ -1,12 +1,16 @@
 package steps
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/minio/minio-go/v7"
+	"github.com/minio/minio-go/v7/pkg/credentials"
 	"io"
 	"net/http"
 	"os"
 	"os/exec"
+	"path"
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
@@ -62,7 +66,7 @@ func WaitForStartup(url string) {
 }
 
 func (s *StepContext) StartObjectStore() error {
-	app := exec.Command("./minio", "--address "+minioAddress, "server", fmt.Sprint(os.TempDir(), "/minio"))
+	app := exec.Command("./minio", "server", "--address", minioAddress, path.Join(os.TempDir(), "catalyst-minio"))
 	outfile, err := os.Create("logs/minio.log")
 	if err != nil {
 		return err
@@ -74,12 +78,78 @@ func (s *StepContext) StartObjectStore() error {
 		return err
 	}
 
-	madmin, err := madmin.New(minioAddress, "minioadmin", "minioadmin", false)
+	admin, err := madmin.New(minioAddress, "minioadmin", "minioadmin", false)
 	if err != nil {
 		return err
 	}
+	s.MinioAdmin = admin
 
-	s.MinioAdmin = madmin
+	minioClient, err := minio.New(minioAddress, &minio.Options{
+		Creds:  credentials.NewStaticV4("minioadmin", "minioadmin", ""),
+		Secure: false,
+	})
+	if err != nil {
+		return err
+	}
+
+	// TODO replace with a proper check for minio startup
+	time.Sleep(time.Second * 5)
+
+	ctx := context.Background()
+
+	// Create buckets if they do not exist.
+	buckets := []string{"rec-bucket", "rec-fallback-bucket"}
+	for _, bucket := range buckets {
+		exists, err := minioClient.BucketExists(ctx, bucket)
+		if err != nil {
+			return fmt.Errorf("failed to check if bucket exists: %w", err)
+		}
+		if exists {
+			continue
+		}
+		err = minioClient.MakeBucket(ctx, bucket, minio.MakeBucketOptions{})
+		if err != nil {
+			return err
+		}
+	}
+
+	// Set bucket policy to allow anonymous download.
+	for _, bucket := range buckets {
+		policy := fmt.Sprintf(`{
+			"Version": "2012-10-17",
+			"Statement": [
+				{
+					"Effect": "Allow",
+					"Principal": {"AWS": ["*"]},
+					"Action": ["s3:GetObject"],
+					"Resource": ["arn:aws:s3:::%s/*"]
+				}
+			]
+		}`, bucket)
+
+		err = minioClient.SetBucketPolicy(ctx, bucket, policy)
+		if err != nil {
+			return err
+		}
+	}
+
+	// populate recording bucket
+	files := []string{"fixtures/tiny.m3u8", "fixtures/seg-0.ts", "fixtures/seg-1.ts", "fixtures/seg-2.ts"}
+	for _, file := range files {
+		_, err := minioClient.FPutObject(ctx, "rec-bucket", path.Base(file), file, minio.PutObjectOptions{})
+		if err != nil {
+			return err
+		}
+	}
+
+	// populate recording fallback bucket
+	files = []string{"fixtures/rec-fallback-bucket/tiny.m3u8", "fixtures/rec-fallback-bucket/seg-3.ts"}
+	for _, file := range files {
+		_, err := minioClient.FPutObject(ctx, "rec-fallback-bucket", path.Base(file), file, minio.PutObjectOptions{})
+		if err != nil {
+			return err
+		}
+	}
 
 	return nil
 }