diff --git a/flytepropeller/events/admin_eventsink.go b/flytepropeller/events/admin_eventsink.go index cc9c57661c3..00344af419a 100644 --- a/flytepropeller/events/admin_eventsink.go +++ b/flytepropeller/events/admin_eventsink.go @@ -66,37 +66,42 @@ func (s *adminEventSink) Sink(ctx context.Context, message proto.Message) error // Validate submission with rate limiter and send admin event if s.rateLimiter.Allow() { + var sendErr error switch eventMessage := message.(type) { case *event.WorkflowExecutionEvent: request := &admin.WorkflowExecutionEventRequest{ Event: eventMessage, } - _, err := s.adminClient.CreateWorkflowEvent(ctx, request) - if err != nil { - return errors.WrapError(err) - } + _, sendErr = s.adminClient.CreateWorkflowEvent(ctx, request) case *event.NodeExecutionEvent: request := &admin.NodeExecutionEventRequest{ Event: eventMessage, } - _, err := s.adminClient.CreateNodeEvent(ctx, request) - if err != nil { - return errors.WrapError(err) - } + _, sendErr = s.adminClient.CreateNodeEvent(ctx, request) case *event.TaskExecutionEvent: request := &admin.TaskExecutionEventRequest{ Event: eventMessage, } - _, err := s.adminClient.CreateTaskEvent(ctx, request) - if err != nil { - return errors.WrapError(err) - } + _, sendErr = s.adminClient.CreateTaskEvent(ctx, request) default: return fmt.Errorf("unknown event type [%s]", eventMessage.String()) } + + if sendErr != nil { + wrapped := errors.WrapError(sendErr) + if errors.IsAlreadyExists(wrapped) { + // Admin already holds this event, so re-sending it can never succeed. Record the id + // so later re-emits are answered by the filter.Contains check above instead of + // spending rate limiter capacity that other executions need. The check returns this + // same error code, so callers see no change. + s.filter.Add(ctx, id) + } + + return wrapped + } } else { return &errors.EventError{Code: errors.ResourceExhausted, Cause: fmt.Errorf("Admin EventSink throttling admin traffic"), Message: "Resource Exhausted"} diff --git a/flytepropeller/events/eventsink_drain_repro_test.go b/flytepropeller/events/eventsink_drain_repro_test.go new file mode 100644 index 00000000000..6281bc238d4 --- /dev/null +++ b/flytepropeller/events/eventsink_drain_repro_test.go @@ -0,0 +1,87 @@ +package events + +import ( + "context" + "testing" + + "github.com/golang/protobuf/ptypes" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/flyteorg/flyte/flyteidl/clients/go/admin/mocks" + "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin" + "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/core" + "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/event" + "github.com/flyteorg/flyte/flytestdlib/fastcheck" + "github.com/flyteorg/flyte/flytestdlib/promutils" +) + +func taskEventFor(name string) *event.TaskExecutionEvent { + return &event.TaskExecutionEvent{ + Phase: core.TaskExecution_SUCCEEDED, + OccurredAt: ptypes.TimestampNow(), + TaskId: &core.Identifier{ResourceType: core.ResourceType_TASK, Name: "task-id"}, + RetryAttempt: 1, + ParentNodeExecutionId: &core.NodeExecutionIdentifier{ + NodeId: "node-id", + ExecutionId: &core.WorkflowExecutionIdentifier{ + Project: "p", + Domain: "d", + Name: name, + }, + }, + Logs: []*core.TaskLog{{Uri: "logs.txt"}}, + } +} + +// A workflow that keeps re-emitting an event Admin has already durably recorded must not be able +// to drain the process-wide event-sink rate limiter and starve other executions. +// +// The sink is built exactly as flytepropeller builds it in NewAdminEventSink: a real +// OppoBloomFilter and a real token-bucket limiter. Only the Admin client is a stub, and it +// answers exactly as a real Admin does for an event it already holds: codes.AlreadyExists. +func TestAdminEventSink_AlreadyExistsDrainsRateLimiter(t *testing.T) { + ctx := context.Background() + scope := promutils.NewTestScope() + + filter, err := fastcheck.NewOppoBloomFilter(50000, scope.NewSubScope("filter")) + assert.NoError(t, err) + + // Capacity 10: a real deployment's shared budget, scaled down so the drain is observable. + adminClient := &mocks.AdminServiceClient{} + sink, err := NewAdminEventSink(ctx, adminClient, &Config{Rate: 1, Capacity: 10}, filter) + assert.NoError(t, err) + + stuck := taskEventFor("stuck-execution") + healthy := taskEventFor("healthy-execution") + + adminCalls := 0 + adminClient.On("CreateTaskEvent", ctx, + mock.MatchedBy(func(req *admin.TaskExecutionEventRequest) bool { + return req.GetEvent().GetParentNodeExecutionId().GetExecutionId().GetName() == "stuck-execution" + }), + ).Run(func(mock.Arguments) { adminCalls++ }). + Return(nil, status.Error(codes.AlreadyExists, "Grpc AlreadyExists error")) + + adminClient.On("CreateTaskEvent", ctx, + mock.MatchedBy(func(req *admin.TaskExecutionEventRequest) bool { + return req.GetEvent().GetParentNodeExecutionId().GetExecutionId().GetName() == "healthy-execution" + }), + ).Return(&admin.TaskExecutionEventResponse{}, nil) + + // Admin already holds this event, so re-emitting it can never succeed. Every attempt should + // be answered from the local dedup filter, not by spending a token on a doomed round trip. + for i := 0; i < 10; i++ { + _ = sink.Sink(ctx, stuck) + } + t.Logf("re-emits of an already-recorded event that reached Admin: %d/10", adminCalls) + assert.Equal(t, 1, adminCalls, + "an already-recorded event should reach Admin once; afterwards the dedup filter should answer") + + // The unrelated execution must still be able to record its event. + healthyErr := sink.Sink(ctx, healthy) + assert.NoError(t, healthyErr, + "a healthy execution was starved of rate-limiter capacity by an already-recorded event") +}