Add debug logs to investigate crashing fra-ai

Makefile

@@ -4,7 +4,7 @@ GO_BUILD_DIR?="./"
 MOCKGEN=go run github.com/golang/mock/mockgen
 ABIGEN=go run github.com/ethereum/go-ethereum/cmd/abigen
 
-all: net/lp_rpc.pb.go net/redeemer.pb.go net/redeemer_mock.pb.go core/test_segment.go eth/contracts/chainlink/AggregatorV3Interface.go livepeer livepeer_cli livepeer_router livepeer_bench
+all: net/lp_rpc.pb.go net/redeemer.pb.go net/redeemer_mock.pb.go core/test_segment.go eth/contracts/chainlink/AggregatorV3Interface.go livepeer livepeer_cli livepeer_router livepeer_bench livepeer_ffmpeg
 
 net/lp_rpc.pb.go: net/lp_rpc.proto
 	protoc -I=. --go_out=. --go-grpc_out=. $^
@@ -108,6 +108,8 @@ livepeer_bench:
 
 livepeer_router:
 	GO111MODULE=on CGO_ENABLED=1 CC="$(cc)" CGO_CFLAGS="$(cgo_cflags)" CGO_LDFLAGS="$(cgo_ldflags) ${CGO_LDFLAGS}" go build -o $(GO_BUILD_DIR) -ldflags="$(ldflags)" cmd/livepeer_router/*.go
+livepeer_ffmpeg:
+	GO111MODULE=on CGO_ENABLED=1 CC="$(cc)" CGO_CFLAGS="$(cgo_cflags)" CGO_LDFLAGS="$(cgo_ldflags) ${CGO_LDFLAGS}" go build -o $(GO_BUILD_DIR) -ldflags="$(ldflags)" cmd/livepeer_ffmpeg/*.go
 
 docker:
 	docker buildx build --build-arg='BUILD_TAGS=mainnet,experimental' -f docker/Dockerfile .

cmd/livepeer_ffmpeg/livepeer_ffmpeg.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/livepeer/lpms/ffmpeg"
+)
+
+func main() {
+	if len(os.Args) < 3 {
+		fmt.Println("Usage: livepeer_ffmpeg <input_file> <output_pattern>")
+		os.Exit(1)
+	}
+
+	in := os.Args[1]
+	outFilePattern := os.Args[2]
+
+	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"},
+	}})
+	if err != nil {
+		fmt.Printf("Failed to run segmentation. in=%s err=%s\n", in, err)
+		os.Exit(1)
+	}
+}

docker/Dockerfile

@@ -53,7 +53,7 @@ RUN	./install_ffmpeg.sh \
 
 COPY	.	.
 
-RUN	make livepeer livepeer_cli livepeer_bench livepeer_router
+RUN	make livepeer livepeer_cli livepeer_bench livepeer_router livepeer_ffmpeg
 
 FROM --platform=$TARGETPLATFORM	nvidia/cuda:12.0.0-cudnn8-runtime-ubuntu20.04 AS livepeer-amd64-base
 
@@ -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 net-tools lsof -y
+
 ENTRYPOINT	["/usr/local/bin/livepeer"]

media/mediamtx.go

@@ -6,6 +6,7 @@
 	"fmt"
 	"io"
 	"net/http"
+	"time"
 
 	"github.com/livepeer/go-livepeer/clog"
 )
@@ -64,12 +65,17 @@
 		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(), 10*time.Second)
+	defer cancel()
+
+	clog.Infof(ctx, "Sending req to MediaMTX")
+	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)
 	}
 	req.SetBasicAuth(mediaMTXControlUser, mc.apiPassword)
 	resp, err := http.DefaultClient.Do(req)
+	clog.Infof(ctx, "Done Sending req to MediaMTX")
 	if err != nil {
 		return fmt.Errorf("failed to kick connection: %w", err)
 	}
@@ -85,12 +91,16 @@
 	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)
+	clog.Infof(context.Background(), "StreamExist: Sending request to MediaMTX, sourceID=%v", mc.sourceID)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	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)
 	}
 	req.SetBasicAuth(mediaMTXControlUser, mc.apiPassword)
 	resp, err := http.DefaultClient.Do(req)
+	clog.Infof(context.Background(), "StreamExist: Done Sending request to MediaMTX, sourceID=%v", mc.sourceID)
 	if err != nil {
 		return false, fmt.Errorf("failed to get stream: %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"
 )
 
@@ -30,6 +30,12 @@
 }
 
 func (ms *MediaSegmenter) RunSegmentation(ctx context.Context, in string, segmentHandler SegmentHandler) {
+	defer func() {
+		if r := recover(); r != nil {
+			clog.Warningf(ctx, "Recovered from panic:")
+		}
+	}()
+
 	outFilePattern := filepath.Join(ms.Workdir, randomString()+"-%d.ts")
 	completionSignal := make(chan bool, 1)
 	wg := &sync.WaitGroup{}
@@ -41,6 +47,7 @@
 
 	retryCount := 0
 	for {
+		clog.Infof(ctx, "Checking if stream exists, sourceID=%v, retryCount=%v", ms.MediaMTXClient.sourceID, retryCount)
 		streamExists, err := ms.MediaMTXClient.StreamExists()
 		if err != nil {
 			clog.Errorf(ctx, "StreamExists check failed. err=%s", err)
@@ -49,18 +56,16 @@
 			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("/usr/local/bin/livepeer_ffmpeg", in, outFilePattern)
+		_, err = cmd.CombinedOutput()
+		if err != nil {
+			clog.Errorf(ctx, "Failed to run segmentation process. in=%s err=%s", in, err)
+		}
+		err = cmd.Wait()
 		if err != nil {
-			clog.Errorf(ctx, "Failed to run segmentation. in=%s err=%s", in, err)
+			clog.Errorf(ctx, "Failed to run segmentation process. in=%s err=%s", in, err)
 		}
+		clog.Infof(ctx, "Done with segmentation process")
 		retryCount++
 		time.Sleep(5 * time.Second)
 	}