Eos timer (#541)

* Add EOS timeout

* add log for finding debug files

* more logging

* add nil check

pkg/pipeline/controller.go

@@ -58,6 +58,7 @@ type Controller struct {
 	limitTimer *time.Timer
 	playing    core.Fuse
 	eos        core.Fuse
+	eosTimer   *time.Timer
 	stopped    core.Fuse
 }
 
@@ -231,6 +232,7 @@ func (c *Controller) Run(ctx context.Context) *livekit.EgressInfo {
 		return c.Info
 	}
 
+	logger.Debugw("closing sinks")
 	for _, si := range c.sinks {
 		for _, s := range si {
 			if err := s.Close(); err != nil {
@@ -378,7 +380,7 @@ func (c *Controller) SendEOS(ctx context.Context) {
 	defer span.End()
 
 	c.eos.Once(func() {
-		logger.Debugw("Sending EOS")
+		logger.Debugw("sending EOS")
 
 		if c.limitTimer != nil {
 			c.limitTimer.Stop()
@@ -405,7 +407,12 @@ func (c *Controller) SendEOS(ctx context.Context) {
 
 		case livekit.EgressStatus_EGRESS_ENDING,
 			livekit.EgressStatus_EGRESS_LIMIT_REACHED:
-			go c.p.SendEOS()
+			go func() {
+				c.eosTimer = time.AfterFunc(time.Second*30, func() {
+					c.OnError(errors.ErrPipelineFrozen)
+				})
+				c.p.SendEOS()
+			}()
 		}
 
 		if c.SourceType == types.SourceTypeWeb {
@@ -430,6 +437,8 @@ func (c *Controller) Close() {
 	if c.SourceType == types.SourceTypeSDK || !c.eos.IsBroken() {
 		c.updateDuration(c.src.GetEndedAt())
 	}
+
+	logger.Debugw("closing source")
 	c.src.Close()
 
 	now := time.Now().UnixNano()
@@ -518,6 +527,7 @@ func (c *Controller) updateStartTime(startedAt int64) {
 
 		case types.EgressTypeSegments:
 			o[0].(*config.SegmentConfig).SegmentsInfo.StartedAt = startedAt
+
 		case types.EgressTypeImages:
 			for _, c := range o {
 				c.(*config.ImageConfig).ImagesInfo.StartedAt = startedAt
@@ -563,6 +573,7 @@ func (c *Controller) updateDuration(endedAt int64) {
 			}
 			segmentsInfo.EndedAt = endedAt
 			segmentsInfo.Duration = endedAt - segmentsInfo.StartedAt
+
 		case types.EgressTypeImages:
 			for _, c := range o {
 				imageInfo := c.(*config.ImageConfig).ImagesInfo

pkg/pipeline/debug.go

@@ -68,6 +68,7 @@ func (c *Controller) uploadDebugFiles() {
 
 	select {
 	case <-done:
+		logger.Infow("debug files uploaded")
 		return
 	case <-time.After(time.Second * 3):
 		logger.Errorw("failed to upload debug files", errors.New("timed out"))

pkg/pipeline/sink/uploader/uploader.go

@@ -26,6 +26,7 @@ import (
 	"github.com/livekit/egress/pkg/stats"
 	"github.com/livekit/egress/pkg/types"
 	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
 )
 
 const (
@@ -83,11 +84,12 @@ type remoteUploader struct {
 func (u *remoteUploader) Upload(localFilepath, storageFilepath string, outputType types.OutputType, deleteAfterUpload bool, fileType string) (string, int64, error) {
 	start := time.Now()
 	location, size, err := u.upload(localFilepath, storageFilepath, outputType)
-	elapsed := time.Since(start).Milliseconds()
+	elapsed := time.Since(start)
+	logger.Debugw("upload complete", "fileType", fileType, "time", elapsed.String())
 
 	// success
 	if err == nil {
-		u.monitor.IncUploadCountSuccess(fileType, float64(elapsed))
+		u.monitor.IncUploadCountSuccess(fileType, float64(elapsed.Milliseconds()))
 		if deleteAfterUpload {
 			_ = os.Remove(localFilepath)
 		}
@@ -96,7 +98,7 @@ func (u *remoteUploader) Upload(localFilepath, storageFilepath string, outputTyp
 	}
 
 	// failure
-	u.monitor.IncUploadCountFailure(fileType, float64(elapsed))
+	u.monitor.IncUploadCountFailure(fileType, float64(elapsed.Milliseconds()))
 	if u.backup != "" {
 		stat, err := os.Stat(localFilepath)
 		if err != nil {

pkg/pipeline/watch.go

@@ -121,6 +121,9 @@ func (c *Controller) messageWatch(msg *gst.Message) bool {
 	switch msg.Type() {
 	case gst.MessageEOS:
 		logger.Infow("EOS received")
+		if c.eosTimer != nil {
+			c.eosTimer.Stop()
+		}
 		c.p.Stop()
 		return false
 	case gst.MessageWarning: