auditlog: fix panic during concurrent streams of the same session (#15361)

The code attempts to wait on an existing download if one is already
in progress rather than starting a concurrent download of the same
session. If this code path runs, we incorrectly defer a call to a
nil function, triggering a panic.

This bug was introduced in #7360.

lib/events/auditlog.go

@@ -900,8 +900,10 @@ func (l *AuditLog) StreamSessionEvents(ctx context.Context, sessionID session.ID
 			e <- trace.BadParameter("audit log is closing, aborting the download")
 			return c, e
 		}
+	} else {
+		defer cancel()
 	}
-	defer cancel()
+
 	rawSession, err := os.OpenFile(tarballPath, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0640)
 	if err != nil {
 		e <- trace.Wrap(err)

lib/events/auditlog_test.go

@@ -19,8 +19,10 @@ package events
 import (
 	"context"
 	"encoding/json"
+	"io"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -28,6 +30,7 @@ import (
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/api/types/events"
 	"github.com/gravitational/teleport/lib/events/eventstest"
+	"github.com/gravitational/teleport/lib/session"
 	"github.com/gravitational/teleport/lib/utils"
 
 	"github.com/jonboulle/clockwork"
@@ -97,6 +100,53 @@ func TestLogRotation(t *testing.T) {
 	}
 }
 
+func TestConcurrentStreaming(t *testing.T) {
+	uploader := NewMemoryUploader()
+	alog, err := NewAuditLog(AuditLogConfig{
+		DataDir:       t.TempDir(),
+		Clock:         clockwork.NewFakeClock(),
+		ServerID:      "remote",
+		UploadHandler: uploader,
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { alog.Close() })
+
+	ctx := context.Background()
+	sid := session.ID("abc123")
+
+	// upload a bogus session so that we can try to stream its events
+	// (this is not valid protobuf, so the stream is not expected to succeed)
+	_, err = uploader.Upload(ctx, sid, io.NopCloser(strings.NewReader(`asdfasdfasdfasdfasdef`)))
+	require.NoError(t, err)
+
+	// run multiple concurrent streams, which forces the second one to wait
+	// on the download that the first one started
+	streams := 2
+	errors := make(chan error, streams)
+	for i := 0; i < streams; i++ {
+		go func() {
+			eventsC, errC := alog.StreamSessionEvents(ctx, sid, 0)
+			for {
+				select {
+				case err := <-errC:
+					errors <- err
+				case _, ok := <-eventsC:
+					if !ok {
+						errors <- nil
+						return
+					}
+				}
+			}
+		}()
+	}
+
+	// This test just verifies that the streamer does not panic when multiple
+	// concurrent streams are waiting on the same download to complete.
+	for i := 0; i < streams; i++ {
+		<-errors
+	}
+}
+
 func TestExternalLog(t *testing.T) {
 	m := &mockAuditLog{
 		emitter: eventstest.MockEmitter{},

lib/events/complete.go

@@ -207,6 +207,17 @@ func (u *UploadCompleter) checkUploads(ctx context.Context) error {
 
 		uploadData := u.cfg.Uploader.GetUploadMetadata(upload.SessionID)
 
+		// It's possible that we don't have a session ID here. For example,
+		// an S3 multipart upload may have been completed by another auth
+		// server, in which case the API returns an empty key, leaving us
+		// no way to derive the session ID from the upload.
+		//
+		// If this is the case, there's no work left to do, and we can
+		// proceed to the next upload.
+		if uploadData.SessionID == "" {
+			continue
+		}
+
 		// Schedule a background operation to check for (and emit) a session end event.
 		// This is necessary because we'll need to download the session in order to
 		// enumerate its events, and the S3 API takes a little while after the upload