21983950160

Committed 13 Feb 2026 10:45AM UTC coverage: 64.291% (-0.2%) from 64.444%

Build # 21983950160

Build Type

push

github

Committed by

jastBytes

Commit Message

fix(deps): update github.com/mandelsoft/goutils digest to cd86279

Coverage Stats

839 of 1305 relevant lines covered (64.29%)

5.41 hits per line

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

51.89

/pkg/discovery/handler/handler.go

// Copyright 2026 BWI GmbH and Solution Arsenal contributors
// SPDX-License-Identifier: Apache-2.0

package handler

import (
        "context"
        "fmt"
        "strings"
        "sync"
        "time"

        "github.com/cenkalti/backoff/v4"
        "github.com/go-logr/logr"
        "golang.org/x/time/rate"
        "ocm.software/ocm/api/ocm"
        "ocm.software/ocm/api/ocm/extensions/repositories/ocireg"

        "go.opendefense.cloud/solar/pkg/discovery"
)

var (
        // handlerRegistry is a map of handler types to their corresponding handlers.
        handlerRegistry = make(map[HandlerType]InitHandlerFunc)
)

type InitHandlerFunc func(log logr.Logger) ComponentHandler

func RegisterComponentHandler(t HandlerType, fn InitHandlerFunc) {
        if fn == nil {
                panic("cannot register nil handler")
        }

        if _, exists := handlerRegistry[t]; exists {
                panic(fmt.Sprintf("handler %q already registered", t))
        }

        handlerRegistry[t] = fn
}

type Handler struct {
        provider    *discovery.RegistryProvider
        inputChan   <-chan discovery.ComponentVersionEvent
        outputChan  chan<- discovery.WriteAPIResourceEvent
        errChan     chan<- discovery.ErrorEvent
        logger      logr.Logger
        stopChan    chan struct{}
        wg          sync.WaitGroup
        stopped     bool
        stopMu      sync.Mutex
        handler     map[HandlerType]ComponentHandler
        rateLimiter *rate.Limiter
        backoff     backoff.BackOff
}

// Option describes the available options
// for creating the Handler.
type Option func(r *Handler)

func WithLogger(l logr.Logger) Option {
        return func(r *Handler) {
                r.logger = l
        }
}

// WithRateLimiter sets the rate limiter for the Qualifier that allows events up to the given interval and burst.
func WithRateLimiter(interval time.Duration, burst int) Option {
        return func(r *Handler) {
                r.rateLimiter = rate.NewLimiter(rate.Every(interval), burst)
        }
}

// WithExponentialBackoff sets an exponential backoff strategy for the Qualifier.
func WithExponentialBackoff(initialInterval time.Duration, maxInterval time.Duration, maxElapsedTime time.Duration) Option {
        return func(r *Handler) {
                b := backoff.NewExponentialBackOff()
                b.InitialInterval = initialInterval
                b.MaxInterval = maxInterval
                b.MaxElapsedTime = maxElapsedTime
                r.backoff = b
        }
}

func NewHandler(
        provider *discovery.RegistryProvider,
        inputChan <-chan discovery.ComponentVersionEvent,
        outputChan chan<- discovery.WriteAPIResourceEvent,
        errChan chan<- discovery.ErrorEvent,
        opts ...Option,
) *Handler {
        c := &Handler{
                provider:   provider,
                inputChan:  inputChan,
                outputChan: outputChan,
                errChan:    errChan,
                logger:     logr.Discard(),
                stopChan:   make(chan struct{}),
                handler:    make(map[HandlerType]ComponentHandler),
        }
        for _, o := range opts {
                o(c)
        }

        return c
}

func (rs *Handler) Start(ctx context.Context) error {
        rs.logger.Info("starting handler")

        rs.wg.Add(1)
        go rs.handlerLoop(ctx)

        return nil
}

// Stop gracefully stops the qualifier.
func (rs *Handler) Stop() {
        rs.stopMu.Lock()
        defer rs.stopMu.Unlock()

        if rs.stopped {
                return
        }

        rs.logger.Info("stopping handler")
        rs.stopped = true
        close(rs.stopChan)
        rs.wg.Wait()
        rs.logger.Info("handler stopped")
}

func (rs *Handler) handlerLoop(ctx context.Context) {
        defer rs.wg.Done()

        for {
                select {
                case <-rs.stopChan:
                        return
                case <-ctx.Done():
                        return
                case ev := <-rs.inputChan:
                        rs.processEvent(ctx, &ev)
                }
        }
}

// isRetryable determines if we should wait and try again
func isRetryable(err error) bool {
        msg := strings.ToLower(err.Error())
        // OCM often wraps errors, so we check the string for common rate-limit indicators
        return strings.Contains(msg, "429") ||
                strings.Contains(msg, "too many requests") ||
                strings.Contains(msg, "connection refused")
}

func (rs *Handler) processEvent(ctx context.Context, ev *discovery.ComponentVersionEvent) {
        rs.logger.Info("processing component version event", "event", ev)
        comp := ev.Component
        version := ev.Source.Version

        // Analyze resources contained in component descriptor.
        helmChartCount := 0
        handlerType := HandlerType("")

        // If rate limiter is configured, wait before making the request
        if rs.rateLimiter != nil {
                if err := rs.rateLimiter.Wait(ctx); err != nil {
                        rs.logger.Error(err, "rate limiter wait failed")
                        return
                }
        }

        // Get registry configuration
        registry := rs.provider.Get(ev.Source.Registry)
        if registry == nil {
                discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{
                        Error:     fmt.Errorf("invalid registry: %s", ev.Source.Registry),
                        Timestamp: time.Now().UTC(),
                })
                rs.logger.V(2).Info("invalid registry", "registry", ev.Source.Registry)

                return
        }

        // Create repository for the component
        baseURL := fmt.Sprintf("%s/%s", registry.GetURL(), ev.Namespace)
        octx := ocm.FromContext(ctx)
        repo, err := octx.RepositoryForSpec(ocireg.NewRepositorySpec(baseURL))
        if err != nil {
                discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{
                        Timestamp: time.Now().UTC(),
                        Error:     fmt.Errorf("failed to create repo spec: %w", err),
                })
                rs.logger.Error(err, "failed to create repo spec", "registry", ev.Source.Registry, "namespace", ev.Namespace)

                return
        }
        defer func() { _ = repo.Close() }()

        // Lookup the specific component version
        var compVersion ocm.ComponentVersionAccess
        if rs.backoff == nil {
                compVersion, err = repo.LookupComponentVersion(comp, version)
        } else {
                // If backoff is configured, use it to retry on transient errors
                operation := func() error {
                        var err error
                        compVersion, err = repo.LookupComponentVersion(comp, version)
                        if err != nil {
                                // Check if the error is a 429 or transient
                                if isRetryable(err) {
                                        return err // Returning error triggers a retry
                                }

                                return backoff.Permanent(err) // Stops retrying for 401, 404, etc.
                        }

                        return nil
                }
                err = backoff.Retry(operation, rs.backoff)
        }
        if err != nil {
                discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{
                        Timestamp: time.Now().UTC(),
                        Error:     fmt.Errorf("failed to lookup component: %w", err),
                })
                rs.logger.Error(err, "failed to lookup component", "version", version)

                return
        }
        defer func() { _ = compVersion.Close() }()

        // Count the number of Helm chart resources in the component version and determine the handler type based on that.
        for _, res := range compVersion.GetDescriptor().ComponentSpec.Resources {
                if res.Type == string(HelmResource) {
                        helmChartCount++
                }
        }

        // Classify component based on contained resources as helm chart and send it to the corresponding handler.
        if helmChartCount == 1 {
                handlerType = HelmHandler
        }

        // If no handler type could be determined, log and publish error.
        if handlerType == "" {
                // No handler found for event, log and publish error.
                rs.logger.Info("no handler found for event", "event", ev)
                discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{
                        Error:     fmt.Errorf("no handler found for event: %v", ev),
                        Timestamp: time.Now().UTC(),
                })

                return
        }

        // Process component with determined handler type.
        h, err := rs.getHandler(handlerType)
        if err != nil {
                rs.logger.Error(err, "failed to process component with handler", "handler", handlerType)
                discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{
                        Error:     fmt.Errorf("failed to process component with handler %q: %w", handlerType, err),
                        Timestamp: time.Now().UTC(),
                })

                return
        }

        // Process component with determined handler. If processing fails, log and publish error.
        resEvent, err := h.Process(ctx, ev, compVersion)
        if err != nil {
                rs.logger.Error(err, "failed to process component with handler", "handler", handlerType)
                discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{
                        Error:     fmt.Errorf("failed to process component with handler %q: %w", handlerType, err),
                        Timestamp: time.Now().UTC(),
                })

                return
        }

        // Publish processed component as API resource event.
        discovery.Publish(&rs.logger, rs.outputChan, *resEvent)
}

// getHandler returns the handler for the given type, initializing it if necessary.
func (rs *Handler) getHandler(t HandlerType) (ComponentHandler, error) {
        if rs.handler[HelmHandler] == nil {
                if initFn, ok := handlerRegistry[HelmHandler]; ok {
                        handler := initFn(rs.logger.WithValues("handler", HelmHandler))
                        rs.handler[HelmHandler] = handler

                        return handler, nil
                }
        }

        return nil, fmt.Errorf("no handler registered for type %v", t)
}

1	// Copyright 2026 BWI GmbH and Solution Arsenal contributors
2	// SPDX-License-Identifier: Apache-2.0
3
4	package handler
5
6	import (
7	"context"
8	"fmt"
9	"strings"
10	"sync"
11	"time"
12
13	"github.com/cenkalti/backoff/v4"
14	"github.com/go-logr/logr"
15	"golang.org/x/time/rate"
16	"ocm.software/ocm/api/ocm"
17	"ocm.software/ocm/api/ocm/extensions/repositories/ocireg"
18
19	"go.opendefense.cloud/solar/pkg/discovery"
20	)
21
22	var (
23	// handlerRegistry is a map of handler types to their corresponding handlers.
24	handlerRegistry = make(map[HandlerType]InitHandlerFunc)
25	)
26
27	type InitHandlerFunc func(log logr.Logger) ComponentHandler
28
29	func RegisterComponentHandler(t HandlerType, fn InitHandlerFunc) {	1✔
30	if fn == nil {	1✔
31	panic("cannot register nil handler")	×
32	}
33
34	if _, exists := handlerRegistry[t]; exists {	1✔
35	panic(fmt.Sprintf("handler %q already registered", t))	×
36	}
37
38	handlerRegistry[t] = fn	1✔
39	}
40
41	type Handler struct {
42	provider *discovery.RegistryProvider
43	inputChan <-chan discovery.ComponentVersionEvent
44	outputChan chan<- discovery.WriteAPIResourceEvent
45	errChan chan<- discovery.ErrorEvent
46	logger logr.Logger
47	stopChan chan struct{}
48	wg sync.WaitGroup
49	stopped bool
50	stopMu sync.Mutex
51	handler map[HandlerType]ComponentHandler
52	rateLimiter *rate.Limiter
53	backoff backoff.BackOff
54	}
55
56	// Option describes the available options
57	// for creating the Handler.
58	type Option func(r *Handler)
59
60	func WithLogger(l logr.Logger) Option {	1✔
61	return func(r *Handler) {	3✔
62	r.logger = l	2✔
63	}	2✔
64	}
65
66	// WithRateLimiter sets the rate limiter for the Qualifier that allows events up to the given interval and burst.
67	func WithRateLimiter(interval time.Duration, burst int) Option {	×
68	return func(r *Handler) {	×
69	r.rateLimiter = rate.NewLimiter(rate.Every(interval), burst)	×
70	}	×
71	}
72
73	// WithExponentialBackoff sets an exponential backoff strategy for the Qualifier.
74	func WithExponentialBackoff(initialInterval time.Duration, maxInterval time.Duration, maxElapsedTime time.Duration) Option {	×
75	return func(r *Handler) {	×
76	b := backoff.NewExponentialBackOff()	×
77	b.InitialInterval = initialInterval	×
78	b.MaxInterval = maxInterval	×
79	b.MaxElapsedTime = maxElapsedTime	×
80	r.backoff = b	×
81	}	×
82	}
83
84	func NewHandler(
85	provider *discovery.RegistryProvider,
86	inputChan <-chan discovery.ComponentVersionEvent,
87	outputChan chan<- discovery.WriteAPIResourceEvent,
88	errChan chan<- discovery.ErrorEvent,
89	opts ...Option,
90	) *Handler {	2✔
91	c := &Handler{	2✔
92	provider: provider,	2✔
93	inputChan: inputChan,	2✔
94	outputChan: outputChan,	2✔
95	errChan: errChan,	2✔
96	logger: logr.Discard(),	2✔
97	stopChan: make(chan struct{}),	2✔
98	handler: make(map[HandlerType]ComponentHandler),	2✔
99	}	2✔
100	for _, o := range opts {	4✔
101	o(c)	2✔
102	}	2✔
103
104	return c	2✔
105	}
106
107	func (rs *Handler) Start(ctx context.Context) error {	2✔
108	rs.logger.Info("starting handler")	2✔
109		2✔
110	rs.wg.Add(1)	2✔
111	go rs.handlerLoop(ctx)	2✔
112		2✔
113	return nil	2✔
114	}	2✔
115
116	// Stop gracefully stops the qualifier.
117	func (rs *Handler) Stop() {	3✔
118	rs.stopMu.Lock()	3✔
119	defer rs.stopMu.Unlock()	3✔
120		3✔
121	if rs.stopped {	4✔
122	return	1✔
123	}	1✔
124
125	rs.logger.Info("stopping handler")	2✔
126	rs.stopped = true	2✔
127	close(rs.stopChan)	2✔
128	rs.wg.Wait()	2✔
129	rs.logger.Info("handler stopped")	2✔
130	}
131
132	func (rs *Handler) handlerLoop(ctx context.Context) {	2✔
133	defer rs.wg.Done()	2✔
134		2✔
135	for {	5✔
136	select {	3✔
137	case <-rs.stopChan:	2✔
138	return	2✔
139	case <-ctx.Done():	×
140	return	×
141	case ev := <-rs.inputChan:	1✔
142	rs.processEvent(ctx, &ev)	1✔
143	}
144	}
145	}
146
147	// isRetryable determines if we should wait and try again
148	func isRetryable(err error) bool {	×
149	msg := strings.ToLower(err.Error())	×
150	// OCM often wraps errors, so we check the string for common rate-limit indicators	×
151	return strings.Contains(msg, "429") \|\|	×
152	strings.Contains(msg, "too many requests") \|\|	×
153	strings.Contains(msg, "connection refused")	×
154	}	×
155
156	func (rs Handler) processEvent(ctx context.Context, ev discovery.ComponentVersionEvent) {	1✔
157	rs.logger.Info("processing component version event", "event", ev)	1✔
158	comp := ev.Component	1✔
159	version := ev.Source.Version	1✔
160		1✔
161	// Analyze resources contained in component descriptor.	1✔
162	helmChartCount := 0	1✔
163	handlerType := HandlerType("")	1✔
164		1✔
165	// If rate limiter is configured, wait before making the request	1✔
166	if rs.rateLimiter != nil {	1✔
167	if err := rs.rateLimiter.Wait(ctx); err != nil {	×
168	rs.logger.Error(err, "rate limiter wait failed")	×
169	return	×
170	}	×
171	}
172
173	// Get registry configuration
174	registry := rs.provider.Get(ev.Source.Registry)	1✔
175	if registry == nil {	1✔
176	discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{	×
177	Error: fmt.Errorf("invalid registry: %s", ev.Source.Registry),	×
178	Timestamp: time.Now().UTC(),	×
179	})	×
180	rs.logger.V(2).Info("invalid registry", "registry", ev.Source.Registry)	×
181		×
182	return	×
183	}	×
184
185	// Create repository for the component
186	baseURL := fmt.Sprintf("%s/%s", registry.GetURL(), ev.Namespace)	1✔
187	octx := ocm.FromContext(ctx)	1✔
188	repo, err := octx.RepositoryForSpec(ocireg.NewRepositorySpec(baseURL))	1✔
189	if err != nil {	1✔
190	discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{	×
191	Timestamp: time.Now().UTC(),	×
192	Error: fmt.Errorf("failed to create repo spec: %w", err),	×
193	})	×
194	rs.logger.Error(err, "failed to create repo spec", "registry", ev.Source.Registry, "namespace", ev.Namespace)	×
195		×
196	return	×
197	}	×
198	defer func() { _ = repo.Close() }()	2✔
199
200	// Lookup the specific component version
201	var compVersion ocm.ComponentVersionAccess	1✔
202	if rs.backoff == nil {	2✔
203	compVersion, err = repo.LookupComponentVersion(comp, version)	1✔
204	} else {	1✔
205	// If backoff is configured, use it to retry on transient errors	×
206	operation := func() error {	×
207	var err error	×
208	compVersion, err = repo.LookupComponentVersion(comp, version)	×
209	if err != nil {	×
210	// Check if the error is a 429 or transient	×
211	if isRetryable(err) {	×
212	return err // Returning error triggers a retry	×
213	}	×
214
215	return backoff.Permanent(err) // Stops retrying for 401, 404, etc.	×
216	}
217
218	return nil	×
219	}
220	err = backoff.Retry(operation, rs.backoff)	×
221	}
222	if err != nil {	1✔
223	discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{	×
224	Timestamp: time.Now().UTC(),	×
225	Error: fmt.Errorf("failed to lookup component: %w", err),	×
226	})	×
227	rs.logger.Error(err, "failed to lookup component", "version", version)	×
228		×
229	return	×
230	}	×
231	defer func() { _ = compVersion.Close() }()	2✔
232
233	// Count the number of Helm chart resources in the component version and determine the handler type based on that.
234	for _, res := range compVersion.GetDescriptor().ComponentSpec.Resources {	6✔
235	if res.Type == string(HelmResource) {	6✔
236	helmChartCount++	1✔
237	}	1✔
238	}
239
240	// Classify component based on contained resources as helm chart and send it to the corresponding handler.
241	if helmChartCount == 1 {	2✔
242	handlerType = HelmHandler	1✔
243	}	1✔
244
245	// If no handler type could be determined, log and publish error.
246	if handlerType == "" {	1✔
247	// No handler found for event, log and publish error.	×
248	rs.logger.Info("no handler found for event", "event", ev)	×
249	discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{	×
250	Error: fmt.Errorf("no handler found for event: %v", ev),	×
251	Timestamp: time.Now().UTC(),	×
252	})	×
253		×
254	return	×
255	}	×
256
257	// Process component with determined handler type.
258	h, err := rs.getHandler(handlerType)	1✔
259	if err != nil {	1✔
260	rs.logger.Error(err, "failed to process component with handler", "handler", handlerType)	×
261	discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{	×
262	Error: fmt.Errorf("failed to process component with handler %q: %w", handlerType, err),	×
263	Timestamp: time.Now().UTC(),	×
264	})	×
265		×
266	return	×
267	}	×
268
269	// Process component with determined handler. If processing fails, log and publish error.
270	resEvent, err := h.Process(ctx, ev, compVersion)	1✔
271	if err != nil {	1✔
272	rs.logger.Error(err, "failed to process component with handler", "handler", handlerType)	×
273	discovery.Publish(&rs.logger, rs.errChan, discovery.ErrorEvent{	×
274	Error: fmt.Errorf("failed to process component with handler %q: %w", handlerType, err),	×
275	Timestamp: time.Now().UTC(),	×
276	})	×
277		×
278	return	×
279	}	×
280
281	// Publish processed component as API resource event.
282	discovery.Publish(&rs.logger, rs.outputChan, *resEvent)	1✔
283	}
284
285	// getHandler returns the handler for the given type, initializing it if necessary.
286	func (rs *Handler) getHandler(t HandlerType) (ComponentHandler, error) {	1✔
287	if rs.handler[HelmHandler] == nil {	2✔
288	if initFn, ok := handlerRegistry[HelmHandler]; ok {	2✔
289	handler := initFn(rs.logger.WithValues("handler", HelmHandler))	1✔
290	rs.handler[HelmHandler] = handler	1✔
291		1✔
292	return handler, nil	1✔
293	}	1✔
294	}
295
296	return nil, fmt.Errorf("no handler registered for type %v", t)	×
297	}

opendefensecloud / solution-arsenal / 21983950160

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