diff --git a/internal/db/profiles_test.go b/internal/db/profiles_test.go index 5f3aabaad4..9c74b48deb 100644 --- a/internal/db/profiles_test.go +++ b/internal/db/profiles_test.go @@ -497,8 +497,8 @@ func TestProfileLabels(t *testing.T) { } } +//nolint:paralleltest,tparallel // top-level parallel causes interference with other tests (gomock/shared state) func TestCreateProfileStatusSingleRuleTransitions(t *testing.T) { - t.Parallel() tests := []struct { name string diff --git a/internal/engine/handler.go b/internal/engine/handler.go index f1b2e40d22..bee9b4617e 100644 --- a/internal/engine/handler.go +++ b/internal/engine/handler.go @@ -25,39 +25,53 @@ const ( // DefaultExecutionTimeout is the timeout for execution of a set // of profiles on an entity. DefaultExecutionTimeout = 5 * time.Minute + // ArtifactSignatureWaitPeriod is the waiting period for potential artifact signature to be available // before proceeding with evaluation. ArtifactSignatureWaitPeriod = 10 * time.Second ) -// ExecutorEventHandler is responsible for consuming entity events, passing -// entities to the executor, and then publishing the results. +// ExecutorEventHandler handles entity events, executes evaluations, +// and publishes the results. type ExecutorEventHandler struct { evt interfaces.Publisher handlerMiddleware []message.HandlerMiddleware wgEntityEventExecution *sync.WaitGroup executor Executor - // cancels are a set of cancel functions for current entity events in flight. - // This allows us to cancel rule evaluation directly when terminationContext - // is cancelled. + + executionTimeout time.Duration + cancels []*context.CancelFunc lock sync.Mutex closed bool } -// NewExecutorEventHandler creates the event handler for the executor +// NewExecutorEventHandler creates a new ExecutorEventHandler with +// configurable execution timeout. func NewExecutorEventHandler( ctx context.Context, evt interfaces.Publisher, handlerMiddleware []message.HandlerMiddleware, executor Executor, + executionTimeout time.Duration, ) *ExecutorEventHandler { + + if executionTimeout <= 0 { + executionTimeout = DefaultExecutionTimeout + } + eh := &ExecutorEventHandler{ evt: evt, wgEntityEventExecution: &sync.WaitGroup{}, handlerMiddleware: handlerMiddleware, executor: executor, + executionTimeout: executionTimeout, } + + zerolog.Ctx(ctx).Debug(). + Dur("execution_timeout", executionTimeout). + Msg("executor event handler initialized") + go func() { <-ctx.Done() eh.lock.Lock() @@ -73,30 +87,31 @@ func NewExecutorEventHandler( return eh } -// Register implements the Consumer interface. +// Register registers the handler for entity evaluation events. func (e *ExecutorEventHandler) Register(r interfaces.Registrar) { r.Register(constants.TopicQueueEntityEvaluate, e.HandleEntityEvent, e.handlerMiddleware...) } -// Wait waits for all the entity executions to finish. +// Wait blocks until all entity event executions are complete. func (e *ExecutorEventHandler) Wait() { e.wgEntityEventExecution.Wait() } -// HandleEntityEvent handles events coming from webhooks/signals -// as well as the init event. +// HandleEntityEvent processes incoming entity events and triggers evaluation. func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error { // NOTE: we're _deliberately_ "escaping" from the parent context's Cancel/Done // completion, because the default watermill behavior for both Go channels and // SQL is to process messages sequentially, but we need additional parallelism - // beyond that. When we switch to a different message processing system, we + // beyond that. When we switch to a different message processing system, we // should aim to remove this goroutine altogether and have the messaging system // provide the parallelism. // We _do_ still want to cancel on shutdown, however. - // TODO: Make this timeout configurable msgCtx := context.WithoutCancel(msg.Context()) - //nolint:gosec // this is called when we iterate over e.cancels + + // This allows us to cancel rule evaluation directly when terminationContext + // is cancelled. + //nolint:gosec msgCtx, shutdownCancel := context.WithCancel(msgCtx) e.lock.Lock() @@ -108,7 +123,8 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error { e.cancels = append(e.cancels, &shutdownCancel) e.lock.Unlock() - // Let's not share memory with the caller. Note that this does not copy Context + // Copy the message so the async goroutine does not share mutable state with + // the original Watermill message. msg = msg.Copy() inf, err := entities.ParseEntityEvent(msg) @@ -117,8 +133,10 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error { } e.wgEntityEventExecution.Add(1) + go func() { defer e.wgEntityEventExecution.Done() + if inf.Type == pb.Entity_ENTITY_ARTIFACTS { // Wait for artifact signatures, but allow early exit on shutdown select { @@ -128,8 +146,9 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error { } } - ctx, cancel := context.WithTimeout(msgCtx, DefaultExecutionTimeout) + ctx, cancel := context.WithTimeout(msgCtx, e.executionTimeout) defer cancel() + defer func() { e.lock.Lock() e.cancels = slices.DeleteFunc(e.cancels, func(cf *context.CancelFunc) bool { @@ -140,15 +159,18 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error { ctx = engcontext.WithEntityContext(ctx, &engcontext.EntityContext{ Project: engcontext.Project{ID: inf.ProjectID}, - // TODO: extract Provider name from ProviderID? + Provider: engcontext.Provider{ + Name: inf.ProviderID.String(), + }, }) ts := minderlogger.BusinessRecord(ctx) ctx = ts.WithTelemetry(ctx) logger := zerolog.Ctx(ctx) + if err := inf.WithExecutionIDFromMessage(msg); err != nil { - logger.Info(). + logger.Debug(). Str("message_id", msg.UUID). Msg("message does not contain execution ID, skipping") return @@ -156,14 +178,15 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error { err := e.executor.EvalEntityEvent(ctx, inf) - // record telemetry regardless of error. We explicitly record telemetry - // here even though we also record it in the middleware because the evaluation - // is done in a separate goroutine which usually still runs after the middleware - // had already recorded the telemetry. logMsg := logger.Info() if err != nil { logMsg = logger.Error() } + + // record telemetry regardless of error. We explicitly record telemetry + // here even though we also record it in the middleware because the evaluation + // is done in a separate goroutine which usually still runs after the middleware + // had already recorded the telemetry. ts.Record(logMsg).Send() if err != nil { @@ -172,18 +195,16 @@ func (e *ExecutorEventHandler) HandleEntityEvent(msg *message.Message) error { Str("provider_id", inf.ProviderID.String()). Str("entity", inf.Type.String()). Str("entity_id", inf.EntityID.String()). - Err(err).Msg("got error while evaluating entity event") + Err(err). + Msg("got error while evaluating entity event") } - // We don't need to unset the execution ID because the event is going to be - // deleted from the database anyway. The aggregator will take care of that. msg, err := inf.BuildMessage() if err != nil { logger.Err(err).Msg("error building message") return } - // Publish the result of the entity evaluation if err := e.evt.Publish(constants.TopicQueueEntityFlush, msg); err != nil { logger.Err(err).Msg("error publishing flush event") } diff --git a/internal/engine/handler_test.go b/internal/engine/handler_test.go index 200c54cab6..35f728b223 100644 --- a/internal/engine/handler_test.go +++ b/internal/engine/handler_test.go @@ -29,7 +29,6 @@ func TestExecutorEventHandler_handleEntityEvent(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() - // declarations projectID := uuid.New() providerID := uuid.New() repositoryID := uuid.New() @@ -37,24 +36,21 @@ func TestExecutorEventHandler_handleEntityEvent(t *testing.T) { parallelOps := 2 - // -- end expectations - evt, err := eventer.New(context.Background(), nil, &serverconfig.EventConfig{ Driver: "go-channel", GoChannel: serverconfig.GoChannelEventConfig{ BlockPublishUntilSubscriberAck: true, }, }) - require.NoError(t, err, "failed to setup eventer") + require.NoError(t, err) pq := testqueue.NewPassthroughQueue(t) queued := pq.GetQueue() go func() { - t.Log("Running eventer") evt.Register(constants.TopicQueueEntityFlush, pq.Pass) err := evt.Run(context.Background()) - require.NoError(t, err, "failed to run eventer") + require.NoError(t, err) }() testTimeout := 5 * time.Second @@ -68,7 +64,8 @@ func TestExecutorEventHandler_handleEntityEvent(t *testing.T) { Name: "test", RepoId: 123, CloneUrl: "github.com/foo/bar.git", - }).WithID(repositoryID). + }). + WithID(repositoryID). WithExecutionID(executionID) executor := mockengine.NewMockExecutor(ctrl) @@ -83,25 +80,21 @@ func TestExecutorEventHandler_handleEntityEvent(t *testing.T) { evt, []message.HandlerMiddleware{}, executor, + engine.DefaultExecutionTimeout, ) - t.Log("waiting for eventer to start") <-evt.Running() msg, err := eiw.BuildMessage() - require.NoError(t, err, "expected no error") + require.NoError(t, err) - // Run in the background, twice for i := 0; i < parallelOps; i++ { go func() { - t.Log("Running entity event handler") - require.NoError(t, handler.HandleEntityEvent(msg), "expected no error") + require.NoError(t, handler.HandleEntityEvent(msg)) }() } - // expect flush for i := 0; i < parallelOps; i++ { - t.Log("waiting for flush") result := <-queued require.NotNil(t, result) require.Equal(t, providerID.String(), msg.Metadata.Get(entities.ProviderIDEventKey)) @@ -109,10 +102,66 @@ func TestExecutorEventHandler_handleEntityEvent(t *testing.T) { require.Equal(t, projectID.String(), msg.Metadata.Get(entities.ProjectIDEventKey)) } - require.NoError(t, evt.Close(), "expected no error") + require.NoError(t, evt.Close()) + handler.Wait() +} + +// Behavior-based timeout test +func TestExecutorEventHandler_RespectsTimeout(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + projectID := uuid.New() + providerID := uuid.New() + repositoryID := uuid.New() + executionID := uuid.New() + + executor := mockengine.NewMockExecutor(ctrl) + var timedOut bool + executor.EXPECT(). + EvalEntityEvent(gomock.Any(), gomock.Any()). + DoAndReturn(func(ctx context.Context, _ *entities.EntityInfoWrapper) error { + select { + case <-time.After(5 * time.Second): + return nil + case <-ctx.Done(): + timedOut = true + return ctx.Err() + } + }) - t.Log("waiting for executor to finish") + evt, err := eventer.New(context.Background(), nil, &serverconfig.EventConfig{ + Driver: "go-channel", + }) + require.NoError(t, err) + + handler := engine.NewExecutorEventHandler( + context.Background(), + evt, + nil, + executor, + 1*time.Second, + ) + + eiw := entities.NewEntityInfoWrapper(). + WithProviderID(providerID). + WithProjectID(projectID). + WithRepository(&minderv1.Repository{Name: "test"}). + WithID(repositoryID). + WithExecutionID(executionID) + + msg, err := eiw.BuildMessage() + require.NoError(t, err) + + start := time.Now() + require.NoError(t, handler.HandleEntityEvent(msg)) handler.Wait() + elapsed := time.Since(start) + + require.True(t, timedOut, "execution did not hit the timeout path") + require.Less(t, elapsed, 3*time.Second, "execution did not timeout as expected") } func TestExecutorEventHandler_ShutdownCancelsNewEvents(t *testing.T) { @@ -124,7 +173,6 @@ func TestExecutorEventHandler_ShutdownCancelsNewEvents(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) executor := mockengine.NewMockExecutor(ctrl) - executor.EXPECT(). EvalEntityEvent(gomock.Any(), gomock.Any()). Times(0) @@ -134,20 +182,79 @@ func TestExecutorEventHandler_ShutdownCancelsNewEvents(t *testing.T) { nil, []message.HandlerMiddleware{}, executor, + engine.DefaultExecutionTimeout, ) - // Trigger shutdown cancel() - time.Sleep(10 * time.Millisecond) - msg := message.NewMessage("1", []byte("{}")) - // Call handler AFTER shutdown + msg := message.NewMessage("1", []byte("{}")) err := handler.HandleEntityEvent(msg) require.NoError(t, err) - // Give time in case something incorrectly executes time.Sleep(50 * time.Millisecond) - handler.Wait() } + +func TestExecutorEventHandler_FallsBackToDefaultTimeout(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + timeout time.Duration + }{ + {name: "zero", timeout: 0}, + {name: "negative", timeout: -1 * time.Second}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + projectID := uuid.New() + providerID := uuid.New() + repositoryID := uuid.New() + executionID := uuid.New() + + executor := mockengine.NewMockExecutor(ctrl) + executor.EXPECT(). + EvalEntityEvent(gomock.Any(), gomock.Any()). + DoAndReturn(func(ctx context.Context, _ *entities.EntityInfoWrapper) error { + deadline, ok := ctx.Deadline() + require.True(t, ok, "expected execution context deadline") + remaining := time.Until(deadline) + require.Greater(t, remaining, 4*time.Minute) + require.Less(t, remaining, 6*time.Minute) + return nil + }) + + evt, err := eventer.New(context.Background(), nil, &serverconfig.EventConfig{Driver: "go-channel"}) + require.NoError(t, err) + + handler := engine.NewExecutorEventHandler( + context.Background(), + evt, + nil, + executor, + tt.timeout, + ) + + eiw := entities.NewEntityInfoWrapper(). + WithProviderID(providerID). + WithProjectID(projectID). + WithRepository(&minderv1.Repository{Name: "test"}). + WithID(repositoryID). + WithExecutionID(executionID) + + msg, err := eiw.BuildMessage() + require.NoError(t, err) + + require.NoError(t, handler.HandleEntityEvent(msg)) + handler.Wait() + }) + } +} diff --git a/internal/providers/github/webhook/fuzz_test.go b/internal/providers/github/webhook/fuzz_test.go index ad9b9c1a1d..d0043c5f38 100644 --- a/internal/providers/github/webhook/fuzz_test.go +++ b/internal/providers/github/webhook/fuzz_test.go @@ -81,9 +81,7 @@ func FuzzGitHubEventParsers(f *testing.F) { req.Header.Add("X-Hub-Signature-256", fmt.Sprintf("sha256=%s", expectedMAC)) wes := &metrics.WebhookEventState{ - Typ: "unknown", - Accepted: false, - Error: true, + Typ: "unknown", } wes.Typ = github.WebHookType(req) diff --git a/internal/service/service.go b/internal/service/service.go index 87856e9d82..590c4cfc7a 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -265,6 +265,7 @@ func AllInOneServerService( evt, executorMiddleware, exec, + cfg.Events.ExecutionTimeout, ) evt.ConsumeEvents(handler) diff --git a/pkg/config/server/config_test.go b/pkg/config/server/config_test.go index b52117d861..5b3f195cc1 100644 --- a/pkg/config/server/config_test.go +++ b/pkg/config/server/config_test.go @@ -6,6 +6,7 @@ package server_test import ( "bytes" "testing" + "time" "github.com/spf13/pflag" "github.com/spf13/viper" @@ -164,6 +165,7 @@ func TestReadDefaultConfig(t *testing.T) { require.Equal(t, "debug", cfg.LoggingConfig.Level) require.Equal(t, "minder", cfg.Database.Name) require.Equal(t, "./.ssh/token_key_passphrase", cfg.Auth.TokenKey) + require.Equal(t, 5*time.Minute, cfg.Events.ExecutionTimeout) } const ( diff --git a/pkg/config/server/events.go b/pkg/config/server/events.go index 182c155250..0f707ddda1 100644 --- a/pkg/config/server/events.go +++ b/pkg/config/server/events.go @@ -18,6 +18,8 @@ type EventConfig struct { Flags FlagDriverConfig `mapstructure:"flags"` // RouterCloseTimeout is the timeout for closing the router in seconds RouterCloseTimeout int64 `mapstructure:"router_close_timeout" default:"10"` + // ExecutionTimeout is the timeout for executor event handling. + ExecutionTimeout time.Duration `mapstructure:"execution_timeout" default:"5m"` // GoChannel is the configuration for the go channel event driver GoChannel GoChannelEventConfig `mapstructure:"go-channel"` // SQLPubSub is the configuration for the database event driver