24002752417

Committed 05 Apr 2026 01:42PM UTC coverage: 58.301% (+0.003%) from 58.298%

Build # 24002752417

Build Type

Pull #6281

github

Committed by

web-flow

Commit Message

Merge 14718ed64 into 63a6a8e5f

Pull Request Pull Request #6281: refactor: populate provider field in entity context using provider ID fallback

Coverage Stats

28 of 34 new or added lines in 2 files covered. (82.35%)

24 existing lines in 2 files now uncovered.

19216 of 32960 relevant lines covered (58.3%)

36.47 hits per line

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

73.83

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

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

                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, 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},
                        Provider: engcontext.Provider{
                                Name: inf.ProviderID.String(), // fallback until provider lookup is available
                        },
                })

                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	}
46
47	// NewExecutorEventHandler creates the event handler for the executor
48	func NewExecutorEventHandler(
49	ctx context.Context,
50	evt interfaces.Publisher,
51	handlerMiddleware []message.HandlerMiddleware,
52	executor Executor,
53	) *ExecutorEventHandler {	1✔
54	eh := &ExecutorEventHandler{	1✔
55	evt: evt,	1✔
56	wgEntityEventExecution: &sync.WaitGroup{},	1✔
57	handlerMiddleware: handlerMiddleware,	1✔
58	executor: executor,	1✔
59	}	1✔
60	go func() {	2✔
61	<-ctx.Done()	1✔
62	eh.lock.Lock()	1✔
63	defer eh.lock.Unlock()	1✔
64		1✔
65	for _, cancel := range eh.cancels {	1✔
NEW 66	(*cancel)()	×
NEW 67	}	×
68	}()
69
70	return eh	1✔
71	}
72
73	// Register implements the Consumer interface.
UNCOV 74	func (e *ExecutorEventHandler) Register(r interfaces.Registrar) {	×
UNCOV 75	r.Register(constants.TopicQueueEntityEvaluate, e.HandleEntityEvent, e.handlerMiddleware...)	×
76	}	×
77
78	// Wait waits for all the entity executions to finish.
79	func (e *ExecutorEventHandler) Wait() {	1✔
80	e.wgEntityEventExecution.Wait()	1✔
81	}	1✔
82
83	// HandleEntityEvent handles events coming from webhooks/signals
84	// as well as the init event.
85	func (e ExecutorEventHandler) HandleEntityEvent(msg message.Message) error {	2✔
86		2✔
87	// NOTE: we're _deliberately_ "escaping" from the parent context's Cancel/Done	2✔
88	// completion, because the default watermill behavior for both Go channels and	2✔
89	// SQL is to process messages sequentially, but we need additional parallelism	2✔
90	// beyond that. When we switch to a different message processing system, we	2✔
91	// should aim to remove this goroutine altogether and have the messaging system	2✔
92	// provide the parallelism.	2✔
93	// We _do_ still want to cancel on shutdown, however.	2✔
94	// TODO: Make this timeout configurable	2✔
95	msgCtx := context.WithoutCancel(msg.Context())	2✔
96	//nolint:gosec // this is called when we iterate over e.cancels	2✔
97	msgCtx, shutdownCancel := context.WithCancel(msgCtx)	2✔
98		2✔
99	e.lock.Lock()	2✔
100	e.cancels = append(e.cancels, &shutdownCancel)	2✔
101	e.lock.Unlock()	2✔
102		2✔
103	// Let's not share memory with the caller. Note that this does not copy Context	2✔
104	msg = msg.Copy()	2✔
105		2✔
106	inf, err := entities.ParseEntityEvent(msg)	2✔
107	if err != nil {	2✔
UNCOV 108	return fmt.Errorf("error unmarshalling payload: %w", err)	×
UNCOV 109	}	×
110
111	e.wgEntityEventExecution.Add(1)	2✔
112	go func() {	4✔
113	defer e.wgEntityEventExecution.Done()	2✔
114	if inf.Type == pb.Entity_ENTITY_ARTIFACTS {	2✔
UNCOV 115	time.Sleep(ArtifactSignatureWaitPeriod)	×
UNCOV 116	}	×
117
118	ctx, cancel := context.WithTimeout(msgCtx, DefaultExecutionTimeout)	2✔
119	defer cancel()	2✔
120	defer func() {	4✔
121	e.lock.Lock()	2✔
122	e.cancels = slices.DeleteFunc(e.cancels, func(cf *context.CancelFunc) bool {	4✔
123	return cf == &shutdownCancel	2✔
124	})	2✔
125	e.lock.Unlock()	2✔
126	}()
127
128	ctx = engcontext.WithEntityContext(ctx, &engcontext.EntityContext{	2✔
129	Project: engcontext.Project{ID: inf.ProjectID},	2✔
130	Provider: engcontext.Provider{	2✔
131	Name: inf.ProviderID.String(), // fallback until provider lookup is available	2✔
132	},	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✔
UNCOV 154	logMsg = logger.Error()	×
UNCOV 155	}	×
156	ts.Record(logMsg).Send()	2✔
157		2✔
158	if err != nil {	2✔
159	logger.Info().	×
160	Str("project", inf.ProjectID.String()).	×
161	Str("provider_id", inf.ProviderID.String()).	×
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")	×
NEW 172	return	×
NEW 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 / 24002752417

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