rafal/debug-initial-startup-time

docker/Dockerfile

@@ -68,4 +68,6 @@ COPY --from=build	/usr/bin/grpc_health_probe	/usr/local/bin/grpc_health_probe
 COPY --from=build	/src/tasmodel.pb	/tasmodel.pb
 COPY --from=build	/usr/share/misc/pci.ids	/usr/share/misc/pci.ids
 
+RUN apt update && apt install -yqq ffmpeg
+
 ENTRYPOINT	["/usr/local/bin/livepeer"]

media/mediamtx.go

@@ -6,6 +6,7 @@
 	"fmt"
 	"io"
 	"net/http"
+	"time"
 
 	"github.com/livepeer/go-livepeer/clog"
 )
@@ -27,10 +28,11 @@
 }
 
 const (
-	mediaMTXControlPort   = "9997"
-	mediaMTXControlUser   = "admin"
-	MediaMTXWebrtcSession = "webrtcSession"
-	MediaMTXRtmpConn      = "rtmpConn"
+	mediaMTXControlPort    = "9997"
+	mediaMTXControlTimeout = 30 * time.Second
+	mediaMTXControlUser    = "admin"
+	MediaMTXWebrtcSession  = "webrtcSession"
+	MediaMTXRtmpConn       = "rtmpConn"
 )
 
 func MediamtxSourceTypeToString(s string) (string, error) {
@@ -44,7 +46,7 @@
 	}
 }
 
 func getApiPath(sourceType string) (string, error) {
 	var apiPath string
 	switch sourceType {
 	case MediaMTXWebrtcSession:
@@ -64,7 +66,9 @@
 		return err
 	}
 
-	req, err := http.NewRequest(http.MethodPost, fmt.Sprintf("http://%s:%s/v3/%s/kick/%s", mc.host, mediaMTXControlPort, apiPath, mc.sourceID), nil)
+	ctx, cancel := context.WithTimeout(context.Background(), mediaMTXControlTimeout)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, fmt.Sprintf("http://%s:%s/v3/%s/kick/%s", mc.host, mediaMTXControlPort, apiPath, mc.sourceID), nil)
 	if err != nil {
 		return fmt.Errorf("failed to create kick request: %w", err)
 	}
@@ -85,7 +89,9 @@
 	if err != nil {
 		return false, err
 	}
-	req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://%s:%s/v3/%s/get/%s", mc.host, mediaMTXControlPort, apiPath, mc.sourceID), nil)
+	ctx, cancel := context.WithTimeout(context.Background(), mediaMTXControlTimeout)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://%s:%s/v3/%s/get/%s", mc.host, mediaMTXControlPort, apiPath, mc.sourceID), nil)
 	if err != nil {
 		return false, fmt.Errorf("failed to create get stream request: %w", err)
 	}

media/rtmp2segment.go

@@ -11,14 +11,14 @@
 	"log/slog"
 	"math/rand"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"strings"
 	"sync"
 	"syscall"
 	"time"
 
 	"github.com/livepeer/go-livepeer/clog"
-	"github.com/livepeer/lpms/ffmpeg"
 	"golang.org/x/sys/unix"
 )
 
@@ -49,17 +49,18 @@
 			clog.Errorf(ctx, "Stopping segmentation, input stream does not exist. in=%s err=%s", in, err)
 			break
 		}
-		ffmpeg.FfmpegSetLogLevel(ffmpeg.FFLogWarning)
-		_, err = ffmpeg.Transcode3(&ffmpeg.TranscodeOptionsIn{
-			Fname: in,
-		}, []ffmpeg.TranscodeOptions{{
-			Oname:        outFilePattern,
-			AudioEncoder: ffmpeg.ComponentOptions{Name: "copy"},
-			VideoEncoder: ffmpeg.ComponentOptions{Name: "copy"},
-			Muxer:        ffmpeg.ComponentOptions{Name: "segment"},
-		}})
+		cmd := exec.Command("ffmpeg",
+			"-i", in,
+			"-c:a", "copy",
+			"-c:v", "copy",
+			"-f", "segment",
+			outFilePattern,
+		)
+		output, err := cmd.CombinedOutput()
 		if err != nil {
-			clog.Errorf(ctx, "Failed to run segmentation. in=%s err=%s", in, err)
+			clog.Errorf(ctx, "Error sending RTMP out process: %v", err)
+			clog.Infof(ctx, "Process output: %s", output)
+			return
 		}
 		retryCount++
 		time.Sleep(5 * time.Second)
@@ -174,7 +175,7 @@
 	// things protected by the mutex mu
 	mu := &sync.Mutex{}
 	isComplete := false
 	var currentSegment *os.File = nil
 	pipeCompletion := make(chan bool, 1)
 
 	// Start a goroutine to wait for the completion signal
@@ -237,7 +238,7 @@
 	}
 }
 
 func readSegment(ctx context.Context, segmentHandler SegmentHandler, file *os.File, pipeName string) {
 	defer file.Close()
 	reader := bufio.NewReader(file)
 	writer := NewMediaWriter()

server/ai_live_video.go

@@ -11,6 +11,7 @@
 	"net/http"
 	"net/url"
 	"os"
+	"os/exec"
 	"strings"
 	"sync"
 	"time"
@@ -21,8 +22,6 @@
 	"github.com/livepeer/go-livepeer/monitor"
 	"github.com/livepeer/go-livepeer/trickle"
 
-	"github.com/livepeer/lpms/ffmpeg"
-
 	"github.com/dustin/go-humanize"
 )
 
@@ -211,16 +210,20 @@
 				break
 			}
 
-			_, err = ffmpeg.Transcode3(&ffmpeg.TranscodeOptionsIn{
-				Fname: fmt.Sprintf("pipe:%d", r.Fd()),
-			}, []ffmpeg.TranscodeOptions{{
-				Oname:        params.liveParams.outputRTMPURL,
-				AudioEncoder: ffmpeg.ComponentOptions{Name: "copy"},
-				VideoEncoder: ffmpeg.ComponentOptions{Name: "copy"},
-				Muxer:        ffmpeg.ComponentOptions{Name: "flv"},
-			}})
+			out := fmt.Sprintf("rtmp://%s/out%s", MediaMTXHost, params.liveParams.stream)
+			cmd := exec.Command("ffmpeg",
+				"-i", "pipe:0",
+				"-c:a", "copy",
+				"-c:v", "copy",
+				"-f", "flv",
+				out,
+			)
+			cmd.Stdin = r
+			output, err := cmd.CombinedOutput()
 			if err != nil {
-				clog.Infof(ctx, "Error sending RTMP out: %s", err)
+				clog.Errorf(ctx, "Error sending RTMP out: %v", err)
+				clog.Infof(ctx, "Process output: %s", output)
+				return
 			}
 			time.Sleep(5 * time.Second)
 		}

server/ai_mediaserver.go

@@ -364,6 +364,8 @@
 	})
 }
 
+var MediaMTXHost string
+
 func (ls *LivepeerServer) StartLiveVideo() http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()
@@ -373,6 +375,9 @@
 			http.Error(w, "Missing stream name", http.StatusBadRequest)
 			return
 		}
+		if strings.HasPrefix(streamName, "out") {
+			return
+		}
 
 		ctx = clog.AddVal(ctx, "stream", streamName)
 		sourceID := r.FormValue("source_id")
@@ -441,6 +446,7 @@
 		}
 
 		mediaMTXClient := media.NewMediaMTXClient(remoteHost, ls.mediaMTXApiPassword, sourceID, sourceType)
+		MediaMTXHost = remoteHost
 
 		if LiveAIAuthWebhookURL != nil {
 			authResp, err := authenticateAIStream(LiveAIAuthWebhookURL, ls.liveAIAuthApiKey, AIAuthRequest{