Upload track debug files on pipeline freeze

pkg/pipeline/debug.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"os"
 	"path"
+	"strings"
 	"sync"
 	"time"
 
@@ -44,9 +45,9 @@ func (p *Pipeline) uploadDebugFiles() {
 	}
 
 	done := make(chan struct{})
-	var wg sync.WaitGroup
-	wg.Add(2)
 
+	var wg sync.WaitGroup
+	wg.Add(3)
 	go func() {
 		defer wg.Done()
 		p.uploadDotFile(u)
@@ -55,6 +56,10 @@ func (p *Pipeline) uploadDebugFiles() {
 		defer wg.Done()
 		p.uploadPProf(u)
 	}()
+	go func() {
+		defer wg.Done()
+		p.uploadTrackFiles(u)
+	}()
 	go func() {
 		wg.Wait()
 		close(done)
@@ -68,6 +73,32 @@ func (p *Pipeline) uploadDebugFiles() {
 	}
 }
 
+func (p *Pipeline) uploadTrackFiles(u uploader.Uploader) {
+	var dir string
+	if p.Debug.ToUploadConfig() == nil {
+		dir = p.Debug.PathPrefix
+	} else {
+		dir = p.TmpDir
+	}
+
+	files, err := os.ReadDir(dir)
+	if err != nil {
+		return
+	}
+
+	for _, f := range files {
+		if strings.HasSuffix(f.Name(), ".csv") {
+			local := path.Join(dir, f.Name())
+			storage := path.Join(p.Debug.PathPrefix, f.Name())
+			_, _, err = u.Upload(local, storage, types.OutputTypeBlob, false)
+			if err != nil {
+				logger.Errorw("failed to upload debug file", err)
+				return
+			}
+		}
+	}
+}
+
 func (p *Pipeline) uploadDotFile(u uploader.Uploader) {
 	dot := p.GetGstPipelineDebugDot()
 	p.uploadDebugFile(u, []byte(dot), ".dot")
@@ -89,20 +120,20 @@ func (p *Pipeline) uploadDebugFile(u uploader.Uploader, data []byte, fileExtensi
 
 	f, err := os.Create(local)
 	if err != nil {
-		logger.Errorw("failed to create dotfile", err)
+		logger.Errorw("failed to create debug file", err)
 		return
 	}
 	defer f.Close()
 
 	_, err = f.Write(data)
 	if err != nil {
-		logger.Errorw("failed to write dotfile", err)
+		logger.Errorw("failed to write debug file", err)
 		return
 	}
 
 	_, _, err = u.Upload(local, storage, types.OutputTypeBlob, false)
 	if err != nil {
-		logger.Errorw("failed to upload dotfile", err)
+		logger.Errorw("failed to upload debug file", err)
 		return
 	}
 }

pkg/pipeline/source/sdk.go

@@ -16,6 +16,8 @@ package source
 
 import (
 	"context"
+	"fmt"
+	"path"
 	"strings"
 	"sync"
 	"time"
@@ -284,7 +286,16 @@ func (s *SDKSource) joinRoom(p *config.PipelineConfig) error {
 		}
 		appSrc := app.SrcFromElement(src)
 
-		writer, err := sdk.NewAppWriter(track, rp, codec, appSrc, s.sync, t, writeBlanks)
+		var logFilename string
+		if p.Debug.EnableProfiling {
+			if p.Debug.ToUploadConfig() == nil {
+				logFilename = path.Join(p.Debug.PathPrefix, fmt.Sprintf("%s.csv", track.ID()))
+			} else {
+				logFilename = path.Join(p.TmpDir, fmt.Sprintf("%s.csv", track.ID()))
+			}
+		}
+
+		writer, err := sdk.NewAppWriter(track, rp, codec, appSrc, s.sync, t, writeBlanks, logFilename)
 		if err != nil {
 			logger.Errorw("could not create app writer", err)
 			onSubscribeErr = err

pkg/pipeline/source/sdk/appwriter.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"os"
 	"time"
 
 	"github.com/frostbyte73/core"
@@ -52,6 +53,7 @@ const (
 
 type AppWriter struct {
 	logger      logger.Logger
+	logFile     *os.File
 	track       *webrtc.TrackRemote
 	identity    string
 	codec       types.MimeType
@@ -86,6 +88,7 @@ func NewAppWriter(
 	sync *synchronizer.Synchronizer,
 	syncInfo *synchronizer.TrackSynchronizer,
 	writeBlanks bool,
+	logFilename string,
 ) (*AppWriter, error) {
 	w := &AppWriter{
 		logger:            logger.GetLogger().WithValues("trackID", track.ID(), "kind", track.Kind().String()),
@@ -102,6 +105,15 @@ func NewAppWriter(
 		finished:          core.NewFuse(),
 	}
 
+	if logFilename != "" {
+		f, err := os.Create(logFilename)
+		if err != nil {
+			return nil, err
+		}
+		_, _ = f.WriteString("pts,sn,ts\n")
+		w.logFile = f
+	}
+
 	var depacketizer rtp.Depacketizer
 	switch codec {
 	case types.MimeTypeVP8:
@@ -200,6 +212,10 @@ func (w *AppWriter) run() {
 		"packetLoss", fmt.Sprintf("%.2f%%", loss*100),
 	)
 
+	if w.logFile != nil {
+		_ = w.logFile.Close()
+	}
+
 	w.finished.Break()
 }
 
@@ -384,5 +400,9 @@ func (w *AppWriter) pushPacket(pkt *rtp.Packet, pts time.Duration) error {
 		w.logger.Infow("unexpected flow return", "flow", flow)
 	}
 
+	if w.logFile != nil {
+		_, _ = w.logFile.WriteString(fmt.Sprintf("%s,%d,%d\n", pts.String(), pkt.SequenceNumber, pkt.Timestamp))
+	}
+
 	return nil
 }