24246595219

Committed 10 Apr 2026 01:58PM UTC coverage: 59.513% (+0.1%) from 59.418%

Build # 24246595219

Build Type

Pull #6320

github

Committed by

web-flow

Commit Message

Merge 4c963ed6d into 644670bac

Pull Request Pull Request #6320: Include rule name in PR branch name to avoid collisions

Coverage Stats

10 of 10 new or added lines in 1 file covered. (100.0%)

22 existing lines in 1 file now uncovered.

19669 of 33050 relevant lines covered (59.51%)

36.62 hits per line

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

75.0

/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
        // 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
        closed  bool
}

// 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,
        }
        go func() {
                <-ctx.Done()
                eh.lock.Lock()
                defer eh.lock.Unlock()

                eh.closed = true

                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()
        if e.closed {
                e.lock.Unlock()
                shutdownCancel()
                return nil
        }
        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, DefaultExecutionTimeout)
                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	// cancels are a set of cancel functions for current entity events in flight.
41	// This allows us to cancel rule evaluation directly when terminationContext
42	// is cancelled.
43	cancels []*context.CancelFunc
44	lock sync.Mutex
45	closed bool
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 {	2✔
55	eh := &ExecutorEventHandler{	2✔
56	evt: evt,	2✔
57	wgEntityEventExecution: &sync.WaitGroup{},	2✔
58	handlerMiddleware: handlerMiddleware,	2✔
59	executor: executor,	2✔
60	}	2✔
61	go func() {	4✔
62	<-ctx.Done()	2✔
63	eh.lock.Lock()	2✔
64	defer eh.lock.Unlock()	2✔
65		2✔
66	eh.closed = true	2✔
67		2✔
68	for _, cancel := range eh.cancels {	2✔
UNCOV 69	(*cancel)()	×
UNCOV 70	}	×
71	}()
72
73	return eh	2✔
74	}
75
76	// Register implements the Consumer interface.
UNCOV 77	func (e *ExecutorEventHandler) Register(r interfaces.Registrar) {	×
UNCOV 78	r.Register(constants.TopicQueueEntityEvaluate, e.HandleEntityEvent, e.handlerMiddleware...)	×
UNCOV 79	}	×
80
81	// Wait waits for all the entity executions to finish.
82	func (e *ExecutorEventHandler) Wait() {	2✔
83	e.wgEntityEventExecution.Wait()	2✔
84	}	2✔
85
86	// HandleEntityEvent handles events coming from webhooks/signals
87	// as well as the init event.
88	func (e ExecutorEventHandler) HandleEntityEvent(msg message.Message) error {	3✔
89		3✔
90	// NOTE: we're _deliberately_ "escaping" from the parent context's Cancel/Done	3✔
91	// completion, because the default watermill behavior for both Go channels and	3✔
92	// SQL is to process messages sequentially, but we need additional parallelism	3✔
93	// beyond that. When we switch to a different message processing system, we	3✔
94	// should aim to remove this goroutine altogether and have the messaging system	3✔
95	// provide the parallelism.	3✔
96	// We _do_ still want to cancel on shutdown, however.	3✔
97	// TODO: Make this timeout configurable	3✔
98	msgCtx := context.WithoutCancel(msg.Context())	3✔
99	//nolint:gosec // this is called when we iterate over e.cancels	3✔
100	msgCtx, shutdownCancel := context.WithCancel(msgCtx)	3✔
101		3✔
102	e.lock.Lock()	3✔
103	if e.closed {	4✔
104	e.lock.Unlock()	1✔
105	shutdownCancel()	1✔
106	return nil	1✔
107	}	1✔
108	e.cancels = append(e.cancels, &shutdownCancel)	2✔
109	e.lock.Unlock()	2✔
110		2✔
111	// Let's not share memory with the caller. Note that this does not copy Context	2✔
112	msg = msg.Copy()	2✔
113		2✔
114	inf, err := entities.ParseEntityEvent(msg)	2✔
115	if err != nil {	2✔
116	return fmt.Errorf("error unmarshalling payload: %w", err)	×
UNCOV 117	}	×
118
119	e.wgEntityEventExecution.Add(1)	2✔
120	go func() {	4✔
121	defer e.wgEntityEventExecution.Done()	2✔
122	if inf.Type == pb.Entity_ENTITY_ARTIFACTS {	2✔
UNCOV 123	time.Sleep(ArtifactSignatureWaitPeriod)	×
UNCOV 124	}	×
125
126	ctx, cancel := context.WithTimeout(msgCtx, DefaultExecutionTimeout)	2✔
127	defer cancel()	2✔
128	defer func() {	4✔
129	e.lock.Lock()	2✔
130	e.cancels = slices.DeleteFunc(e.cancels, func(cf *context.CancelFunc) bool {	5✔
131	return cf == &shutdownCancel	3✔
132	})	3✔
133	e.lock.Unlock()	2✔
134	}()
135
136	ctx = engcontext.WithEntityContext(ctx, &engcontext.EntityContext{	2✔
137	Project: engcontext.Project{ID: inf.ProjectID},	2✔
138	// TODO: extract Provider name from ProviderID?	2✔
139	})	2✔
140		2✔
141	ts := minderlogger.BusinessRecord(ctx)	2✔
142	ctx = ts.WithTelemetry(ctx)	2✔
143		2✔
144	logger := zerolog.Ctx(ctx)	2✔
145	if err := inf.WithExecutionIDFromMessage(msg); err != nil {	2✔
UNCOV 146	logger.Info().	×
UNCOV 147	Str("message_id", msg.UUID).	×
UNCOV 148	Msg("message does not contain execution ID, skipping")	×
UNCOV 149	return	×
UNCOV 150	}	×
151
152	err := e.executor.EvalEntityEvent(ctx, inf)	2✔
153		2✔
154	// record telemetry regardless of error. We explicitly record telemetry	2✔
155	// here even though we also record it in the middleware because the evaluation	2✔
156	// is done in a separate goroutine which usually still runs after the middleware	2✔
157	// had already recorded the telemetry.	2✔
158	logMsg := logger.Info()	2✔
159	if err != nil {	2✔
160	logMsg = logger.Error()	×
161	}	×
162	ts.Record(logMsg).Send()	2✔
163		2✔
164	if err != nil {	2✔
UNCOV 165	logger.Info().	×
UNCOV 166	Str("project", inf.ProjectID.String()).	×
UNCOV 167	Str("provider_id", inf.ProviderID.String()).	×
UNCOV 168	Str("entity", inf.Type.String()).	×
169	Str("entity_id", inf.EntityID.String()).	×
170	Err(err).Msg("got error while evaluating entity event")	×
171	}	×
172
173	// We don't need to unset the execution ID because the event is going to be
174	// deleted from the database anyway. The aggregator will take care of that.
175	msg, err := inf.BuildMessage()	2✔
176	if err != nil {	2✔
UNCOV 177	logger.Err(err).Msg("error building message")	×
UNCOV 178	return	×
UNCOV 179	}	×
180
181	// Publish the result of the entity evaluation
182	if err := e.evt.Publish(constants.TopicQueueEntityFlush, msg); err != nil {	2✔
UNCOV 183	logger.Err(err).Msg("error publishing flush event")	×
UNCOV 184	}	×
185	}()
186
187	return nil	2✔
188	}

mindersec / minder / 24246595219

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