Improve recording upload backoff (#42718)

Failed uploads could get in an infinite backoff of 10s instead of
linearly backing off as intended. Due to the asynchronous nature
of launching uploads by `(Uploader) Scan`, it could return without
an error which resulted in resetting the backoff to its initial
value even if previous failed uploads had incremented the backoff.
To avoid this, resetting the backoff delay was modified to only
occur if an upload completed successfully.

Additionally error messaging was attempted to be improved. Any errors
caused by the stream being terminated should now be returned instead
of a vague message.

integration/integration_test.go

@@ -524,7 +524,7 @@ func testAuditOn(t *testing.T, suite *integrationTestSuite) {
 			}
 
 			// wait for the upload of the right session to complete
-			timeoutC := time.After(10 * time.Second)
+			timeoutC := time.After(20 * time.Second)
 		loop:
 			for {
 				select {
@@ -1268,7 +1268,7 @@ func testLeafProxySessionRecording(t *testing.T, suite *integrationTestSuite) {
 
 			// Wait for the session recording to be uploaded and available
 			var uploaded bool
-			timeoutC := time.After(10 * time.Second)
+			timeoutC := time.After(20 * time.Second)
 			for !uploaded {
 				select {
 				case event := <-uploadChan:

integration/kube_integration_test.go

@@ -329,7 +329,7 @@ func testExec(t *testing.T, suite *KubeSuite, pinnedIP string, clientError strin
 
 	// verify traffic capture and upload, wait for the upload to hit
 	var sessionID string
-	timeoutC := time.After(10 * time.Second)
+	timeoutC := time.After(20 * time.Second)
 loop:
 	for {
 		select {
@@ -764,7 +764,7 @@ func testKubeTrustedClustersClientCert(t *testing.T, suite *KubeSuite) {
 
 	// verify traffic capture and upload, wait for the upload to hit
 	var sessionID string
-	timeoutC := time.After(10 * time.Second)
+	timeoutC := time.After(20 * time.Second)
 loop:
 	for {
 		select {
@@ -1038,7 +1038,7 @@ func testKubeTrustedClustersSNI(t *testing.T, suite *KubeSuite) {
 
 	// verify traffic capture and upload, wait for the upload to hit
 	var sessionID string
-	timeoutC := time.After(10 * time.Second)
+	timeoutC := time.After(20 * time.Second)
 loop:
 	for {
 		select {

lib/events/filesessions/fileasync.go

@@ -50,6 +50,8 @@ type UploaderConfig struct {
 	CorruptedDir string
 	// Clock is the clock replacement
 	Clock clockwork.Clock
+	// InitialScanDelay is how long to wait before performing the initial scan.
+	InitialScanDelay time.Duration
 	// ScanPeriod is a uploader dir scan period
 	ScanPeriod time.Duration
 	// ConcurrentUploads sets up how many parallel uploads to schedule
@@ -193,9 +195,11 @@ func (u *Uploader) Serve(ctx context.Context) error {
 
 	u.log.Infof("uploader will scan %v every %v", u.cfg.ScanDir, u.cfg.ScanPeriod.String())
 	backoff, err := retryutils.NewLinear(retryutils.LinearConfig{
-		Step:  u.cfg.ScanPeriod,
-		Max:   u.cfg.ScanPeriod * 100,
-		Clock: u.cfg.Clock,
+		First:  u.cfg.InitialScanDelay,
+		Step:   u.cfg.ScanPeriod,
+		Max:    u.cfg.ScanPeriod * 100,
+		Clock:  u.cfg.Clock,
+		Jitter: retryutils.NewSeventhJitter(),
 	})
 	if err != nil {
 		return trace.Wrap(err)
@@ -221,8 +225,7 @@ func (u *Uploader) Serve(ctx context.Context) error {
 				}
 			default:
 				backoff.Inc()
-				u.log.WithError(event.Error).Warningf(
-					"Backing off, will retry after %v.", backoff.Duration())
+				u.log.Warnf("Increasing session upload backoff due to error, will retry after %v.", backoff.Duration())
 			}
 			// forward the event to channel that used in tests
 			if u.cfg.EventsC != nil {
@@ -234,19 +237,12 @@ func (u *Uploader) Serve(ctx context.Context) error {
 			}
 		// Tick at scan period but slow down (and speeds up) on errors.
 		case <-backoff.After():
-			var failed bool
 			if _, err := u.Scan(ctx); err != nil {
 				if !errors.Is(trace.Unwrap(err), errContext) {
-					failed = true
-					u.log.WithError(err).Warningf("Uploader scan failed.")
+					backoff.Inc()
+					u.log.WithError(err).Warningf("Uploader scan failed, will retry after %v.", backoff.Duration())
 				}
 			}
-			if failed {
-				backoff.Inc()
-				u.log.Debugf("Scan failed, backing off, will retry after %v.", backoff.Duration())
-			} else {
-				backoff.ResetToDelay()
-			}
 		}
 	}
 }
@@ -532,6 +528,12 @@ func (u *Uploader) upload(ctx context.Context, up *upload) error {
 	select {
 	case <-u.closeC:
 		return trace.Errorf("operation has been canceled, uploader is closed")
+	case <-stream.Done():
+		if errStream, ok := stream.(interface{ Error() error }); ok {
+			return trace.ConnectionProblem(errStream.Error(), errStream.Error().Error())
+		}
+
+		return trace.ConnectionProblem(nil, "upload stream terminated unexpectedly")
 	case <-stream.Status():
 	case <-time.After(events.NetworkRetryDuration):
 		return trace.ConnectionProblem(nil, "timeout waiting for stream status update")

lib/events/filesessions/fileasync_test.go

@@ -483,16 +483,17 @@ func TestUploadBadSession(t *testing.T) {
 // uploaderPack reduces boilerplate required
 // to create a test
 type uploaderPack struct {
-	scanPeriod  time.Duration
-	clock       clockwork.FakeClock
-	eventsC     chan events.UploadEvent
-	memEventsC  chan events.UploadEvent
-	memUploader *eventstest.MemoryUploader
-	streamer    events.Streamer
-	scanDir     string
-	uploader    *Uploader
-	ctx         context.Context
-	cancel      context.CancelFunc
+	scanPeriod       time.Duration
+	initialScanDelay time.Duration
+	clock            clockwork.FakeClock
+	eventsC          chan events.UploadEvent
+	memEventsC       chan events.UploadEvent
+	memUploader      *eventstest.MemoryUploader
+	streamer         events.Streamer
+	scanDir          string
+	uploader         *Uploader
+	ctx              context.Context
+	cancel           context.CancelFunc
 }
 
 func (u *uploaderPack) Close(t *testing.T) {
@@ -505,13 +506,14 @@ func newUploaderPack(t *testing.T, wrapStreamer wrapStreamerFn) uploaderPack {
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	pack := uploaderPack{
-		clock:      clockwork.NewFakeClock(),
-		eventsC:    make(chan events.UploadEvent, 100),
-		memEventsC: make(chan events.UploadEvent, 100),
-		ctx:        ctx,
-		cancel:     cancel,
-		scanDir:    scanDir,
-		scanPeriod: 10 * time.Second,
+		clock:            clockwork.NewFakeClock(),
+		eventsC:          make(chan events.UploadEvent, 100),
+		memEventsC:       make(chan events.UploadEvent, 100),
+		ctx:              ctx,
+		cancel:           cancel,
+		scanDir:          scanDir,
+		scanPeriod:       10 * time.Second,
+		initialScanDelay: 10 * time.Millisecond,
 	}
 	pack.memUploader = eventstest.NewMemoryUploader(pack.memEventsC)
 
@@ -527,12 +529,13 @@ func newUploaderPack(t *testing.T, wrapStreamer wrapStreamerFn) uploaderPack {
 	}
 
 	uploader, err := NewUploader(UploaderConfig{
-		ScanDir:      pack.scanDir,
-		CorruptedDir: corruptedDir,
-		ScanPeriod:   pack.scanPeriod,
-		Streamer:     pack.streamer,
-		Clock:        pack.clock,
-		EventsC:      pack.eventsC,
+		ScanDir:          pack.scanDir,
+		CorruptedDir:     corruptedDir,
+		InitialScanDelay: pack.initialScanDelay,
+		ScanPeriod:       pack.scanPeriod,
+		Streamer:         pack.streamer,
+		Clock:            pack.clock,
+		EventsC:          pack.eventsC,
 	})
 	require.NoError(t, err)
 	pack.uploader = uploader
@@ -564,12 +567,13 @@ func runResume(t *testing.T, testCase resumeTestCase) {
 
 	scanPeriod := 10 * time.Second
 	uploader, err := NewUploader(UploaderConfig{
-		EventsC:      eventsC,
-		ScanDir:      scanDir,
-		CorruptedDir: corruptedDir,
-		ScanPeriod:   scanPeriod,
-		Streamer:     test.streamer,
-		Clock:        clock,
+		EventsC:          eventsC,
+		ScanDir:          scanDir,
+		CorruptedDir:     corruptedDir,
+		InitialScanDelay: 10 * time.Millisecond,
+		ScanPeriod:       scanPeriod,
+		Streamer:         test.streamer,
+		Clock:            clock,
 	})
 	require.NoError(t, err)
 	go uploader.Serve(ctx)

lib/service/service.go

@@ -3025,10 +3025,11 @@ func (process *TeleportProcess) initUploaderService() error {
 	corruptedDir := filepath.Join(paths[1]...)
 
 	fileUploader, err := filesessions.NewUploader(filesessions.UploaderConfig{
-		Streamer:     uploaderClient,
-		ScanDir:      uploadsDir,
-		CorruptedDir: corruptedDir,
-		EventsC:      process.Config.Testing.UploadEventsC,
+		Streamer:         uploaderClient,
+		ScanDir:          uploadsDir,
+		CorruptedDir:     corruptedDir,
+		EventsC:          process.Config.Testing.UploadEventsC,
+		InitialScanDelay: 15 * time.Second,
 	})
 	if err != nil {
 		return trace.Wrap(err)