diff --git a/lib/events/auditlog.go b/lib/events/auditlog.go index 4cc7a1392e8..9862caa37bb 100644 --- a/lib/events/auditlog.go +++ b/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) diff --git a/lib/events/auditlog_test.go b/lib/events/auditlog_test.go index 489a0a7e163..72246ecf0fc 100644 --- a/lib/events/auditlog_test.go +++ b/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{}, diff --git a/lib/events/complete.go b/lib/events/complete.go index 0b363ea216f..896230b3809 100644 --- a/lib/events/complete.go +++ b/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