upload: generate dtsh before segmenting input file

clients/mist_client.go

@@ -7,7 +7,13 @@ import (
 	"fmt"
 	"io"
 	"net/url"
+	"os/exec"
+	"path"
 	"sync"
+	"time"
+
+	"github.com/livepeer/catalyst-api/config"
+	"github.com/livepeer/catalyst-api/subprocess"
 )
 
 type MistAPIClient interface {
@@ -17,6 +23,7 @@ type MistAPIClient interface {
 	AddTrigger(streamName, triggerName string) error
 	DeleteTrigger(streamName, triggerName string) error
 	GetStreamInfo(streamName string) (MistStreamInfo, error)
+	CreateDTSH(destination string) error
 }
 
 type MistClient struct {
@@ -94,6 +101,32 @@ func (mc *MistClient) DeleteStream(streamName string) error {
 	return wrapErr(validateDeleteStream(mc.sendCommand(c)), streamName)
 }
 
+func (mc *MistClient) CreateDTSH(destination string) error {
+	url, err := url.Parse(destination)
+	if err != nil {
+		return err
+	}
+	url.RawQuery = ""
+	url.Fragment = ""
+	headerPrepare := exec.Command(path.Join(config.PathMistDir, "MistInMP4"), "-H", url.String(), "-g", "5")
+	if err = subprocess.LogOutputs(headerPrepare); err != nil {
+		return err
+	}
+
+	if err = headerPrepare.Start(); err != nil {
+		return err
+	}
+
+	// Make sure the command doesn't run indefinitely
+	// Tested on ~1Gb files and too < 10 seconds, so this should hopefully be plenty of time
+	timer := time.AfterFunc(5*time.Minute, func() {
+		_ = headerPrepare.Process.Kill()
+	})
+	defer timer.Stop()
+
+	return headerPrepare.Wait()
+}
+
 // AddTrigger adds a trigger `triggerName` for the stream `streamName`.
 // Note that Mist API supports only overriding the whole trigger configuration, therefore this function needs to:
 // 1. Acquire a lock

clients/mist_client_mock.go

@@ -25,3 +25,7 @@ func (s StubMistClient) AddTrigger(streamName, triggerName string) error {
 func (s StubMistClient) DeleteTrigger(streamName, triggerName string) error {
 	return nil
 }
+
+func (s StubMistClient) CreateDTSH(destination string) error {
+	return nil
+}

clients/mist_client_test.go

@@ -2,10 +2,16 @@ package clients
 
 import (
 	"errors"
+	"io/fs"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
+	"os"
+	"path"
 	"testing"
+	"time"
 
+	"github.com/livepeer/catalyst-api/config"
 	"github.com/stretchr/testify/require"
 )
 
@@ -241,6 +247,31 @@ func TestItFailsWhenMaxRetriesReached(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestItCanGenerateDTSH(t *testing.T) {
+	mc := &MistClient{
+		HttpReqUrl: "",
+	}
+
+	dir := t.TempDir()
+	config.PathMistDir = dir
+	destination := "unused"
+
+	err := mc.CreateDTSH("invalid://user:abc{DEf1=lp@example.com:5432/db?sslmode=require")
+	require.Error(t, err)
+	require.IsType(t, &url.Error{}, err)
+
+	err = mc.CreateDTSH(destination)
+	require.Error(t, err)
+	require.IsType(t, &fs.PathError{}, err)
+
+	script := path.Join(dir, "MistInMP4")
+	_ = os.WriteFile(script, []byte("#!/bin/sh\necho livepeer\n"), 0744)
+
+	err = mc.CreateDTSH(destination)
+	require.NoError(t, err)
+	time.Sleep(50 * time.Millisecond)
+}
+
 var mistResponse = `{
 	"height": 720,
 	"meta": {

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/kylelemons/godebug v1.1.0
 	github.com/livepeer/go-tools v0.0.0-20220926110222-2ebcbb5685b4
 	github.com/livepeer/livepeer-data v0.4.22
-	github.com/minio/madmin-go v1.7.4
+	github.com/minio/madmin-go v1.7.5
 	github.com/stretchr/testify v1.8.0
 	github.com/xeipuuv/gojsonschema v1.2.0
 	sigs.k8s.io/yaml v1.3.0
@@ -45,7 +45,7 @@ require (
 	github.com/lufia/plan9stats v0.0.0-20220913051719-115f729f3c8c // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.16 // indirect
-	github.com/minio/minio-go/v7 v7.0.43-0.20221021202758-c6319beb6b27 // indirect
+	github.com/minio/minio-go/v7 v7.0.43 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/philhofer/fwd v1.1.2-0.20210722190033-5c56ac6d0bb9 // indirect
@@ -59,15 +59,15 @@ require (
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/stretchr/objx v0.4.0 // indirect
 	github.com/tinylib/msgp v1.1.7-0.20220719154719-f3635b96e483 // indirect
-	github.com/tklauser/go-sysconf v0.3.10 // indirect
-	github.com/tklauser/numcpus v0.5.0 // indirect
+	github.com/tklauser/go-sysconf v0.3.11 // indirect
+	github.com/tklauser/numcpus v0.6.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.opencensus.io v0.23.0 // indirect
-	golang.org/x/crypto v0.1.0 // indirect
-	golang.org/x/net v0.1.0 // indirect
+	golang.org/x/crypto v0.2.0 // indirect
+	golang.org/x/net v0.2.0 // indirect
 	golang.org/x/oauth2 v0.1.0 // indirect
 	golang.org/x/sync v0.1.0 // indirect
-	golang.org/x/sys v0.1.0 // indirect
+	golang.org/x/sys v0.2.0 // indirect
 	golang.org/x/text v0.4.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
 	google.golang.org/api v0.98.0 // indirect

go.sum

@@ -339,10 +339,10 @@ github.com/mattn/go-isatty v0.0.16 h1:bq3VjFmv/sOjHtdEhmkEV4x1AJtvUvOJ2PFAZ5+peK
 github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
-github.com/minio/madmin-go v1.7.4 h1:xEx9P4lFGfwyg5aiEYEyfGxPLzlPIoXakMU6TULs5rE=
-github.com/minio/madmin-go v1.7.4/go.mod h1:3SO8SROxHN++tF6QxdTii2SSUaYSrr8lnE9EJWjvz0k=
-github.com/minio/minio-go/v7 v7.0.43-0.20221021202758-c6319beb6b27 h1:EqNpIeNzjGQjAe9Ih5gVW/4PccAse9+aa46hAoQSCQY=
-github.com/minio/minio-go/v7 v7.0.43-0.20221021202758-c6319beb6b27/go.mod h1:nCrRzjoSUQh8hgKKtu3Y708OLvRLtuASMg2/nvmbarw=
+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/minio-go/v7 v7.0.43 h1:14Q4lwblqTdlAmba05oq5xL0VBLHi06zS4yLnIkz6hI=
+github.com/minio/minio-go/v7 v7.0.43/go.mod h1:nCrRzjoSUQh8hgKKtu3Y708OLvRLtuASMg2/nvmbarw=
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=
 github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
@@ -436,11 +436,12 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
 github.com/tinylib/msgp v1.1.7-0.20220719154719-f3635b96e483 h1:dV39KLgsNZGri7Hn5QhHwRQzGf7kHOki2vZujFXDFhI=
 github.com/tinylib/msgp v1.1.7-0.20220719154719-f3635b96e483/go.mod h1:g7jEyb18KPe65d9RRhGw+ThaJr5duyBH8eaFgBUor7Y=
-github.com/tklauser/go-sysconf v0.3.10 h1:IJ1AZGZRWbY8T5Vfk04D9WOA5WSejdflXxP03OUqALw=
 github.com/tklauser/go-sysconf v0.3.10/go.mod h1:C8XykCvCb+Gn0oNCWPIlcb0RuglQTYaQ2hGm7jmxEFk=
+github.com/tklauser/go-sysconf v0.3.11 h1:89WgdJhk5SNwJfu+GKyYveZ4IaJ7xAkecBo+KdJV0CM=
+github.com/tklauser/go-sysconf v0.3.11/go.mod h1:GqXfhXY3kiPa0nAXPDIQIWzJbMCB7AmcWpGR8lSZfqI=
 github.com/tklauser/numcpus v0.4.0/go.mod h1:1+UI3pD8NW14VMwdgJNJ1ESk2UnwhAnz5hMwiKKqXCQ=
-github.com/tklauser/numcpus v0.5.0 h1:ooe7gN0fg6myJ0EKoTAf5hebTZrH52px3New/D9iJ+A=
-github.com/tklauser/numcpus v0.5.0/go.mod h1:OGzpTxpcIMNGYQdit2BYL1pvk/dSOaJWjKoflh+RQjo=
+github.com/tklauser/numcpus v0.6.0 h1:kebhY2Qt+3U6RNK7UqpYNA+tJ23IBEGKkB7JQBfDYms=
+github.com/tklauser/numcpus v0.6.0/go.mod h1:FEZLMke0lhOUG6w2JadTzp0a+Nl8PF/GFkQ5UVIcaL4=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f h1:J9EGpcZtP0E/raorCMxlFGSTBrsSlaDGf3jU/qvAE2c=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
@@ -477,8 +478,8 @@ golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.1.0 h1:MDRAIl0xIo9Io2xV565hzXHw3zVseKrJKodhohM5CjU=
-golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
+golang.org/x/crypto v0.2.0 h1:BRXPfhNivWL5Yq0BGQ39a2sW6t44aODpfxkWjYdzewE=
+golang.org/x/crypto v0.2.0/go.mod h1:hebNnKkNXi2UzZN1eVRvBB7co0a+JxK6XbPiWVs/3J4=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -561,8 +562,8 @@ golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su
 golang.org/x/net v0.0.0-20220607020251-c690dde0001d/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220909164309-bea034e7d591/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
-golang.org/x/net v0.1.0 h1:hZ/3BUoy5aId7sCpA/Tc5lt8DkFgdVS2onTpJsZ/fl0=
-golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
+golang.org/x/net v0.2.0 h1:sZfSu1wtKLGlWI4ZZayP0ck9Y73K1ynO6gqzTdBVdPU=
+golang.org/x/net v0.2.0/go.mod h1:KqCZLdyyvdV855qA2rE3GC2aiw5xGR5TEjj8smXukLY=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -670,8 +671,8 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220610221304-9f5ed59c137d/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=
-golang.org/x/sys v0.1.0 h1:kunALQeHf1/185U1i0GOB/fy1IPRDDpuoOOqRReG57U=
-golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.2.0 h1:ljd4t30dBnAvMZaQCevtY0xLLD0A+bRZXbgLMLU1F/A=
+golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

handlers/misttriggers/recording_end.go

@@ -4,18 +4,13 @@ import (
 	"fmt"
 	"math"
 	"net/http"
-	"net/url"
-	"os/exec"
-	"path"
 	"strconv"
 	"strings"
 
 	"github.com/livepeer/catalyst-api/cache"
 	"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/subprocess"
 	"github.com/livepeer/catalyst-api/transcode"
 )
 
@@ -144,7 +139,6 @@ func (d *MistCallbackHandlersCollection) triggerRecordingEndSegmenting(w http.Re
 			return
 		}
 
-		// When createDtsh() completes issue another callback signaling to studio playback is ready
 		defer func() {
 			// Send the success callback
 			err = clients.DefaultCallbackClient.SendTranscodeStatusCompleted(transcodeRequest.CallbackURL, inputInfo, outputs)
@@ -155,11 +149,9 @@ func (d *MistCallbackHandlersCollection) triggerRecordingEndSegmenting(w http.Re
 
 		// prepare .dtsh headers for all rendition playlists
 		for _, output := range outputs {
-			// output is multivariant playlist
-			err := createDtsh(requestID, output.Manifest)
-			if err != nil {
+			if err := d.MistClient.CreateDTSH(output.Manifest); err != nil {
 				// should not block the ingestion flow or make it fail on error.
-				log.LogError(requestID, "master createDtsh() failed", err, "destination", output.Manifest)
+				log.LogError(requestID, "CreateDTSH() for rendition failed", err, "destination", output.Manifest)
 			}
 		}
 
@@ -233,25 +225,3 @@ func ParseRecordingEndPayload(payload string) (RecordingEndPayload, error) {
 		LastMediaTimestampMillis:  LastMediaTimestampMillis,
 	}, nil
 }
-
-func createDtsh(requestID, destination string) error {
-	url, err := url.Parse(destination)
-	if err != nil {
-		return err
-	}
-	url.RawQuery = ""
-	url.Fragment = ""
-	headerPrepare := exec.Command(path.Join(config.PathMistDir, "MistInHLS"), "-H", url.String(), "-g", "5")
-	if err = subprocess.LogOutputs(headerPrepare); err != nil {
-		return err
-	}
-	if err = headerPrepare.Start(); err != nil {
-		return err
-	}
-	go func() {
-		if err := headerPrepare.Wait(); err != nil {
-			log.LogError(requestID, "createDtsh return code", err, "destination", destination)
-		}
-	}()
-	return nil
-}

handlers/misttriggers/triggers_test.go

@@ -4,12 +4,8 @@ import (
 	"bytes"
 	"encoding/json"
 	"io"
-	"io/fs"
 	"net/http"
 	"net/http/httptest"
-	"net/url"
-	"os"
-	"path"
 	"testing"
 	"time"
 
@@ -111,23 +107,6 @@ func TestRecordingCompleted(t *testing.T) {
 	}
 }
 
-func TestMistInHLSStart(t *testing.T) {
-	dir := t.TempDir()
-	config.PathMistDir = dir
-	destination := "unused"
-	err := createDtsh("testRequestID", "invalid://user:abc{DEf1=lp@example.com:5432/db?sslmode=require")
-	require.IsType(t, &url.Error{}, err)
-	err = createDtsh("testRequestID", destination)
-	require.IsType(t, &fs.PathError{}, err)
-
-	script := path.Join(dir, "MistInHLS")
-	_ = os.WriteFile(script, []byte("#!/bin/sh\necho livepeer\n"), 0744)
-
-	err = createDtsh("testRequestID", destination)
-	require.NoError(t, err)
-	time.Sleep(50 * time.Millisecond)
-}
-
 type StreamSample struct {
 	streamName string
 	expected   PipelineId