23982777537

Committed 04 Apr 2026 04:21PM UTC coverage: 58.3% (+0.002%) from 58.298%

Build # 23982777537

Build Type

Pull #6278

github

Committed by

web-flow

Commit Message

Merge 33f441671 into 63a6a8e5f

Pull Request Pull Request #6278: feat: make executor event handling timeout configurable

Coverage Stats

14 of 16 new or added lines in 2 files covered. (87.5%)

27 existing lines in 2 files now uncovered.

19215 of 32959 relevant lines covered (58.3%)

36.42 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.58

/internal/engine/handler.go

// SPDX-FileCopyrightText: Copyright 2024 The Minder Authors
// SPDX-License-Identifier: Apache-2.0

package engine

import (
        "context"
        "fmt"
        "slices"
        "sync"
        "time"

        "github.com/ThreeDotsLabs/watermill/message"
        "github.com/rs/zerolog"

        "github.com/mindersec/minder/internal/engine/engcontext"
        "github.com/mindersec/minder/internal/engine/entities"
        minderlogger "github.com/mindersec/minder/internal/logger"
        pb "github.com/mindersec/minder/pkg/api/protobuf/go/minder/v1"
        "github.com/mindersec/minder/pkg/eventer/constants"
        "github.com/mindersec/minder/pkg/eventer/interfaces"
)

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.
type ExecutorEventHandler struct {
        evt                    interfaces.Publisher
        handlerMiddleware      []message.HandlerMiddleware
        wgEntityEventExecution *sync.WaitGroup
        executor               Executor
        executionTimeout       time.Duration
        // 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.
        cancels []*context.CancelFunc
        lock    sync.Mutex
}

// NewExecutorEventHandler creates the event handler for the executor
func NewExecutorEventHandler(
        ctx context.Context,
        evt interfaces.Publisher,
        handlerMiddleware []message.HandlerMiddleware,
        executor Executor,
) *ExecutorEventHandler {
        eh := &ExecutorEventHandler{
                evt:                    evt,
                wgEntityEventExecution: &sync.WaitGroup{},
                handlerMiddleware:      handlerMiddleware,
                executor:               executor,
                executionTimeout:       DefaultExecutionTimeout,
        }
        go func() {
                <-ctx.Done()
                eh.lock.Lock()
                defer eh.lock.Unlock()

                for _, cancel := range eh.cancels {
                        (*cancel)()
                }
        }()

        return eh
}

// Register implements the Consumer interface.
func (e *ExecutorEventHandler) Register(r interfaces.Registrar) {
        r.Register(constants.TopicQueueEntityEvaluate, e.HandleEntityEvent, e.handlerMiddleware...)
}

// Wait waits for all the entity executions to finish.
func (e *ExecutorEventHandler) Wait() {
        e.wgEntityEventExecution.Wait()
}

// HandleEntityEvent handles events coming from webhooks/signals
// as well as the init event.
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
        // 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
        msgCtx, shutdownCancel := context.WithCancel(msgCtx)

        e.lock.Lock()
        e.cancels = append(e.cancels, &shutdownCancel)
        e.lock.Unlock()

        // Let's not share memory with the caller.  Note that this does not copy Context
        msg = msg.Copy()

        inf, err := entities.ParseEntityEvent(msg)
        if err != nil {
                return fmt.Errorf("error unmarshalling payload: %w", err)
        }

        e.wgEntityEventExecution.Add(1)
        go func() {
                defer e.wgEntityEventExecution.Done()
                if inf.Type == pb.Entity_ENTITY_ARTIFACTS {
                        time.Sleep(ArtifactSignatureWaitPeriod)
                }

                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 {
                                return cf == &shutdownCancel
                        })
                        e.lock.Unlock()
                }()

                ctx = engcontext.WithEntityContext(ctx, &engcontext.EntityContext{
                        Project: engcontext.Project{ID: inf.ProjectID},
                        // TODO: extract Provider name from ProviderID?
                })

                ts := minderlogger.BusinessRecord(ctx)
                ctx = ts.WithTelemetry(ctx)

                logger := zerolog.Ctx(ctx)
                if err := inf.WithExecutionIDFromMessage(msg); err != nil {
                        logger.Info().
                                Str("message_id", msg.UUID).
                                Msg("message does not contain execution ID, skipping")
                        return
                }

                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()
                }
                ts.Record(logMsg).Send()

                if err != nil {
                        logger.Info().
                                Str("project", inf.ProjectID.String()).
                                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")
                }

                // 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")
                }
        }()

        return nil
}

1	// SPDX-FileCopyrightText: Copyright 2024 The Minder Authors
2	// SPDX-License-Identifier: Apache-2.0
3
4	package engine
5
6	import (
7	"context"
8	"fmt"
9	"slices"
10	"sync"
11	"time"
12
13	"github.com/ThreeDotsLabs/watermill/message"
14	"github.com/rs/zerolog"
15
16	"github.com/mindersec/minder/internal/engine/engcontext"
17	"github.com/mindersec/minder/internal/engine/entities"
18	minderlogger "github.com/mindersec/minder/internal/logger"
19	pb "github.com/mindersec/minder/pkg/api/protobuf/go/minder/v1"
20	"github.com/mindersec/minder/pkg/eventer/constants"
21	"github.com/mindersec/minder/pkg/eventer/interfaces"
22	)
23
24	const (
25	// DefaultExecutionTimeout is the timeout for execution of a set
26	// of profiles on an entity.
27	DefaultExecutionTimeout = 5 * time.Minute
28	// ArtifactSignatureWaitPeriod is the waiting period for potential artifact signature to be available
29	// before proceeding with evaluation.
30	ArtifactSignatureWaitPeriod = 10 * time.Second
31	)
32
33	// ExecutorEventHandler is responsible for consuming entity events, passing
34	// entities to the executor, and then publishing the results.
35	type ExecutorEventHandler struct {
36	evt interfaces.Publisher
37	handlerMiddleware []message.HandlerMiddleware
38	wgEntityEventExecution *sync.WaitGroup
39	executor Executor
40	executionTimeout time.Duration
41	// cancels are a set of cancel functions for current entity events in flight.
42	// This allows us to cancel rule evaluation directly when terminationContext
43	// is cancelled.
44	cancels []*context.CancelFunc
45	lock sync.Mutex
46	}
47
48	// NewExecutorEventHandler creates the event handler for the executor
49	func NewExecutorEventHandler(
50	ctx context.Context,
51	evt interfaces.Publisher,
52	handlerMiddleware []message.HandlerMiddleware,
53	executor Executor,
54	) *ExecutorEventHandler {	1✔
55	eh := &ExecutorEventHandler{	1✔
56	evt: evt,	1✔
57	wgEntityEventExecution: &sync.WaitGroup{},	1✔
58	handlerMiddleware: handlerMiddleware,	1✔
59	executor: executor,	1✔
60	executionTimeout: DefaultExecutionTimeout,	1✔
61	}	1✔
62	go func() {	2✔
63	<-ctx.Done()	1✔
64	eh.lock.Lock()	1✔
65	defer eh.lock.Unlock()	1✔
66		1✔
67	for _, cancel := range eh.cancels {	1✔
NEW 68	(*cancel)()	×
UNCOV 69	}	×
70	}()
71
72	return eh	1✔
73	}
74
75	// Register implements the Consumer interface.
76	func (e *ExecutorEventHandler) Register(r interfaces.Registrar) {	×
UNCOV 77	r.Register(constants.TopicQueueEntityEvaluate, e.HandleEntityEvent, e.handlerMiddleware...)	×
UNCOV 78	}	×
79
80	// Wait waits for all the entity executions to finish.
81	func (e *ExecutorEventHandler) Wait() {	1✔
82	e.wgEntityEventExecution.Wait()	1✔
83	}	1✔
84
85	// HandleEntityEvent handles events coming from webhooks/signals
86	// as well as the init event.
87	func (e ExecutorEventHandler) HandleEntityEvent(msg message.Message) error {	2✔
88		2✔
89	// NOTE: we're _deliberately_ "escaping" from the parent context's Cancel/Done	2✔
90	// completion, because the default watermill behavior for both Go channels and	2✔
91	// SQL is to process messages sequentially, but we need additional parallelism	2✔
92	// beyond that. When we switch to a different message processing system, we	2✔
93	// should aim to remove this goroutine altogether and have the messaging system	2✔
94	// provide the parallelism.	2✔
95	// We _do_ still want to cancel on shutdown, however.	2✔
96	// TODO: Make this timeout configurable	2✔
97	msgCtx := context.WithoutCancel(msg.Context())	2✔
98	//nolint:gosec // this is called when we iterate over e.cancels	2✔
99	msgCtx, shutdownCancel := context.WithCancel(msgCtx)	2✔
100		2✔
101	e.lock.Lock()	2✔
102	e.cancels = append(e.cancels, &shutdownCancel)	2✔
103	e.lock.Unlock()	2✔
104		2✔
105	// Let's not share memory with the caller. Note that this does not copy Context	2✔
106	msg = msg.Copy()	2✔
107		2✔
108	inf, err := entities.ParseEntityEvent(msg)	2✔
109	if err != nil {	2✔
UNCOV 110	return fmt.Errorf("error unmarshalling payload: %w", err)	×
UNCOV 111	}	×
112
113	e.wgEntityEventExecution.Add(1)	2✔
114	go func() {	4✔
115	defer e.wgEntityEventExecution.Done()	2✔
116	if inf.Type == pb.Entity_ENTITY_ARTIFACTS {	2✔
UNCOV 117	time.Sleep(ArtifactSignatureWaitPeriod)	×
UNCOV 118	}	×
119
120	ctx, cancel := context.WithTimeout(msgCtx, e.executionTimeout)	2✔
121	defer cancel()	2✔
122	defer func() {	4✔
123	e.lock.Lock()	2✔
124	e.cancels = slices.DeleteFunc(e.cancels, func(cf *context.CancelFunc) bool {	5✔
125	return cf == &shutdownCancel	3✔
126	})	3✔
127	e.lock.Unlock()	2✔
128	}()
129
130	ctx = engcontext.WithEntityContext(ctx, &engcontext.EntityContext{	2✔
131	Project: engcontext.Project{ID: inf.ProjectID},	2✔
132	// TODO: extract Provider name from ProviderID?	2✔
133	})	2✔
134		2✔
135	ts := minderlogger.BusinessRecord(ctx)	2✔
136	ctx = ts.WithTelemetry(ctx)	2✔
137		2✔
138	logger := zerolog.Ctx(ctx)	2✔
139	if err := inf.WithExecutionIDFromMessage(msg); err != nil {	2✔
UNCOV 140	logger.Info().	×
UNCOV 141	Str("message_id", msg.UUID).	×
UNCOV 142	Msg("message does not contain execution ID, skipping")	×
UNCOV 143	return	×
UNCOV 144	}	×
145
146	err := e.executor.EvalEntityEvent(ctx, inf)	2✔
147		2✔
148	// record telemetry regardless of error. We explicitly record telemetry	2✔
149	// here even though we also record it in the middleware because the evaluation	2✔
150	// is done in a separate goroutine which usually still runs after the middleware	2✔
151	// had already recorded the telemetry.	2✔
152	logMsg := logger.Info()	2✔
153	if err != nil {	2✔
154	logMsg = logger.Error()	×
155	}	×
156	ts.Record(logMsg).Send()	2✔
157		2✔
158	if err != nil {	2✔
UNCOV 159	logger.Info().	×
UNCOV 160	Str("project", inf.ProjectID.String()).	×
UNCOV 161	Str("provider_id", inf.ProviderID.String()).	×
UNCOV 162	Str("entity", inf.Type.String()).	×
UNCOV 163	Str("entity_id", inf.EntityID.String()).	×
UNCOV 164	Err(err).Msg("got error while evaluating entity event")	×
UNCOV 165	}	×
166
167	// We don't need to unset the execution ID because the event is going to be
168	// deleted from the database anyway. The aggregator will take care of that.
169	msg, err := inf.BuildMessage()	2✔
170	if err != nil {	2✔
UNCOV 171	logger.Err(err).Msg("error building message")	×
172	return	×
173	}	×
174
175	// Publish the result of the entity evaluation
176	if err := e.evt.Publish(constants.TopicQueueEntityFlush, msg); err != nil {	2✔
177	logger.Err(err).Msg("error publishing flush event")	×
178	}	×
179	}()
180
181	return nil	2✔
182	}

mindersec / minder / 23982777537

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous