Skip to content

[v17] Fix large event handling for Athena backend #63551

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kshi36 merged 1 commit into from
Feb 6, 2026
Merged

[v17] Fix large event handling for Athena backend #63551

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 16 additions & 34 deletions lib/events/athena/querier.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ import (
athenaTypes "github.com/aws/aws-sdk-go-v2/service/athena/types"
"github.com/aws/aws-sdk-go-v2/service/s3"
s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
"github.com/dustin/go-humanize"
"github.com/google/uuid"
"github.com/gravitational/trace"
"github.com/jonboulle/clockwork"
Expand Down Expand Up @@ -982,41 +981,24 @@ func (rb *responseBuilder) appendUntilSizeLimit(resultResp *athena.GetQueryResul
if err != nil {
return false, trace.Wrap(err, "failed to marshal event, %s", eventData)
}
if len(marshalledEvent)+rb.totalSize <= events.MaxEventBytesInResponse {
events.MetricQueriedTrimmedEvents.Inc()
// Exact rb.totalSize doesn't really matter since the response is
// already size limited.
rb.totalSize += events.MaxEventBytesInResponse
rb.output = append(rb.output, event)
return true, nil

if len(marshalledEvent)+rb.totalSize > events.MaxEventBytesInResponse {
// Failed to trim the event to size.
// Even if we fail to trim the event, we still try to return the oversized event.
rb.logger.WarnContext(context.Background(), "Failed to query event exceeding maximum response size.",
"event_type", event.GetType(),
"event_id", event.GetID(),
"event_size", len(eventData),
"event_size_after_trim", len(marshalledEvent),
)
}
events.MetricQueriedTrimmedEvents.Inc()

// Failed to trim the event to size. The only options are to return
// a response with 0 events, skip this event, or return an error.
//
// Silently skipping events is a terrible option, it's better for
// the client to get an error.
//
// Returning 0 events amounts to either skipping the event or
// getting the client stuck in a paging loop depending on what would
// be returned for the next page token.
//
// Returning a descriptive error should at least give the client a
// hint as to what has gone wrong so that an attempt can be made to
// fix it.
//
// If this condition is reached it should be considered a bug, any
// event that can possibly exceed the maximum size should implement
// TrimToMaxSize (until we can one day implement an API for storing
// and retrieving large events).
rb.logger.ErrorContext(context.Background(), "Failed to query event exceeding maximum response size.",
"event_type", event.GetType(),
"event_id", event.GetID(),
"event_size", len(eventData),
)
return true, trace.Errorf(
"%s event %s is %s and cannot be returned because it exceeds the maximum response size of %s",
event.GetType(), event.GetID(), humanize.IBytes(uint64(len(eventData))), humanize.IBytes(events.MaxEventBytesInResponse))
// Exact rb.totalSize doesn't really matter since the response is
// already size limited.
rb.totalSize += events.MaxEventBytesInResponse
rb.output = append(rb.output, event)
Copy link
Copy Markdown
Contributor Author

@kshi36 kshi36 Feb 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Manual backport because return type for branch/v17 is apievents.AuditEvent instead of events.EventFields.

return true, nil
}
rb.totalSize += len(eventData)
rb.output = append(rb.output, event)
Expand Down
18 changes: 6 additions & 12 deletions lib/events/athena/querier_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ import (
athenaTypes "github.com/aws/aws-sdk-go-v2/service/athena/types"
"github.com/aws/aws-sdk-go-v2/service/s3"
s3Types "github.com/aws/aws-sdk-go-v2/service/s3/types"
"github.com/dustin/go-humanize"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/google/uuid"
Expand Down Expand Up @@ -766,10 +765,9 @@ func Test_querier_fetchResults(t *testing.T) {
// fakeResp defines responses which will be returned based on given
// input token to GetQueryResults. Note that due to limit of GetQueryResults
// we are doing multiple calls, first always with empty token.
fakeResp map[string]eventsWithToken
wantEvents []apievents.AuditEvent
wantKeyset string
wantErrorMsg string
fakeResp map[string]eventsWithToken
wantEvents []apievents.AuditEvent
wantKeyset string
}{
{
name: "no data returned from query, return empty results",
Expand Down Expand Up @@ -802,9 +800,9 @@ func Test_querier_fetchResults(t *testing.T) {
"": {returnToken: "", events: []apievents.AuditEvent{bigUntrimmableEvent}},
},
limit: 10,
wantErrorMsg: fmt.Sprintf(
"app.create event %s is 5.0 MiB and cannot be returned because it exceeds the maximum response size of %s",
bigUntrimmableEvent.Metadata.ID, humanize.IBytes(events.MaxEventBytesInResponse)),
// we still want to receive the untrimmable event
wantEvents: []apievents.AuditEvent{bigUntrimmableEvent},
wantKeyset: mustEventToKey(t, bigUntrimmableEvent),
},
{
name: "events with trimmable event exceeding > MaxEventBytesInResponse",
Expand Down Expand Up @@ -861,10 +859,6 @@ func Test_querier_fetchResults(t *testing.T) {
},
}
gotEvents, gotKeyset, err := q.fetchResults(context.Background(), "queryid", tt.limit, tt.condition)
if tt.wantErrorMsg != "" {
require.ErrorContains(t, err, tt.wantErrorMsg)
return
}
require.NoError(t, err)
require.Empty(t, cmp.Diff(tt.wantEvents, gotEvents, cmpopts.EquateEmpty(),
// Expect the database query to be trimmed
Expand Down
Loading