21986023208

Committed 13 Feb 2026 11:59AM UTC coverage: 60.654% (-3.6%) from 64.291%

Build # 21986023208

Build Type

push

github

Committed by

jastBytes

Commit Message

feat: Remove optional message field from ErrorEvent in event handling

Coverage Stats

797 of 1314 relevant lines covered (60.65%)

5.24 hits per line

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

57.41

/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"
        "time"

        "github.com/cenkalti/backoff/v4"
        "github.com/go-logr/logr"
        "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 {
        *discovery.Runner[discovery.ComponentVersionEvent, discovery.WriteAPIResourceEvent]
        provider *discovery.RegistryProvider
        handler  map[HandlerType]ComponentHandler
}

func NewHandlerOptions(opts ...discovery.RunnerOption[discovery.ComponentVersionEvent, discovery.WriteAPIResourceEvent]) []discovery.RunnerOption[discovery.ComponentVersionEvent, discovery.WriteAPIResourceEvent] {
        return opts
}

func NewHandler(
        provider *discovery.RegistryProvider,
        in <-chan discovery.ComponentVersionEvent,
        out chan<- discovery.WriteAPIResourceEvent,
        err chan<- discovery.ErrorEvent,
        opts ...discovery.RunnerOption[discovery.ComponentVersionEvent, discovery.WriteAPIResourceEvent],
) *Handler {
        p := &Handler{
                provider: provider,
                handler:  make(map[HandlerType]ComponentHandler),
        }
        p.Runner = discovery.NewRunner(p, in, out, err)
        for _, opt := range opts {
                opt(p.Runner)
        }

        return p
}

// 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) Process(ctx context.Context, ev discovery.ComponentVersionEvent) ([]discovery.WriteAPIResourceEvent, error) {
        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("")

        // Exit early on deletion
        if ev.Source.Type == discovery.EventDeleted {
                return []discovery.WriteAPIResourceEvent{{
                        Source:    ev,
                        Timestamp: time.Now().UTC(),
                }}, nil
        }

        // Get registry configuration
        registry := rs.provider.Get(ev.Source.Registry)
        if registry == nil {
                rs.Logger().V(2).Info("invalid registry", "registry", ev.Source.Registry)
                return nil, fmt.Errorf("invalid registry: %s", ev.Source.Registry)
        }

        // 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 {
                rs.Logger().Error(err, "failed to create repo spec", "registry", ev.Source.Registry, "repository", ev.Source.Repository)
                return nil, fmt.Errorf("failed to create repository spec: %w", err)
        }
        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 {
                rs.Logger().Error(err, "failed to lookup component", "version", version)
                return nil, fmt.Errorf("failed to lookup component version %s: %w", version, err)
        }
        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)
                return nil, fmt.Errorf("no handler found for component version event: %v", ev)
        }

        // 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)
                return nil, fmt.Errorf("failed to process component with handler %q: %w", handlerType, err)
        }

        // 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)
                return nil, fmt.Errorf("failed to process component with handler %q: %w", handlerType, err)
        }

        return []discovery.WriteAPIResourceEvent{*resEvent}, nil
}

// 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	"time"
11
12	"github.com/cenkalti/backoff/v4"
13	"github.com/go-logr/logr"
14	"ocm.software/ocm/api/ocm"
15	"ocm.software/ocm/api/ocm/extensions/repositories/ocireg"
16
17	"go.opendefense.cloud/solar/pkg/discovery"
18	)
19
20	var (
21	// handlerRegistry is a map of handler types to their corresponding handlers.
22	handlerRegistry = make(map[HandlerType]InitHandlerFunc)
23	)
24
25	type InitHandlerFunc func(log logr.Logger) ComponentHandler
26
27	func RegisterComponentHandler(t HandlerType, fn InitHandlerFunc) {	1✔
28	if fn == nil {	1✔
29	panic("cannot register nil handler")	×
30	}
31
32	if _, exists := handlerRegistry[t]; exists {	1✔
33	panic(fmt.Sprintf("handler %q already registered", t))	×
34	}
35
36	handlerRegistry[t] = fn	1✔
37	}
38
39	type Handler struct {
40	*discovery.Runner[discovery.ComponentVersionEvent, discovery.WriteAPIResourceEvent]
41	provider *discovery.RegistryProvider
42	handler map[HandlerType]ComponentHandler
43	}
44
45	func NewHandlerOptions(opts ...discovery.RunnerOption[discovery.ComponentVersionEvent, discovery.WriteAPIResourceEvent]) []discovery.RunnerOption[discovery.ComponentVersionEvent, discovery.WriteAPIResourceEvent] {	1✔
46	return opts	1✔
47	}	1✔
48
49	func NewHandler(
50	provider *discovery.RegistryProvider,
51	in <-chan discovery.ComponentVersionEvent,
52	out chan<- discovery.WriteAPIResourceEvent,
53	err chan<- discovery.ErrorEvent,
54	opts ...discovery.RunnerOption[discovery.ComponentVersionEvent, discovery.WriteAPIResourceEvent],
55	) *Handler {	2✔
56	p := &Handler{	2✔
57	provider: provider,	2✔
58	handler: make(map[HandlerType]ComponentHandler),	2✔
59	}	2✔
60	p.Runner = discovery.NewRunner(p, in, out, err)	2✔
61	for _, opt := range opts {	4✔
62	opt(p.Runner)	2✔
63	}	2✔
64
65	return p	2✔
66	}
67
68	// isRetryable determines if we should wait and try again
69	func isRetryable(err error) bool {	×
70	msg := strings.ToLower(err.Error())	×
71	// OCM often wraps errors, so we check the string for common rate-limit indicators	×
72	return strings.Contains(msg, "429") \|\|	×
73	strings.Contains(msg, "too many requests") \|\|	×
74	strings.Contains(msg, "connection refused")	×
75	}	×
76
77	func (rs *Handler) Process(ctx context.Context, ev discovery.ComponentVersionEvent) ([]discovery.WriteAPIResourceEvent, error) {	1✔
78	rs.Logger().Info("processing component version event", "event", ev)	1✔
79	comp := ev.Component	1✔
80	version := ev.Source.Version	1✔
81		1✔
82	// Analyze resources contained in component descriptor.	1✔
83	helmChartCount := 0	1✔
84	handlerType := HandlerType("")	1✔
85		1✔
86	// Exit early on deletion	1✔
87	if ev.Source.Type == discovery.EventDeleted {	1✔
88	return []discovery.WriteAPIResourceEvent{{	×
89	Source: ev,	×
90	Timestamp: time.Now().UTC(),	×
91	}}, nil	×
92	}	×
93
94	// Get registry configuration
95	registry := rs.provider.Get(ev.Source.Registry)	1✔
96	if registry == nil {	1✔
97	rs.Logger().V(2).Info("invalid registry", "registry", ev.Source.Registry)	×
98	return nil, fmt.Errorf("invalid registry: %s", ev.Source.Registry)	×
99	}	×
100
101	// Create repository for the component
102	baseURL := fmt.Sprintf("%s/%s", registry.GetURL(), ev.Namespace)	1✔
103	octx := ocm.FromContext(ctx)	1✔
104	repo, err := octx.RepositoryForSpec(ocireg.NewRepositorySpec(baseURL))	1✔
105	if err != nil {	1✔
106	rs.Logger().Error(err, "failed to create repo spec", "registry", ev.Source.Registry, "repository", ev.Source.Repository)	×
107	return nil, fmt.Errorf("failed to create repository spec: %w", err)	×
108	}	×
109	defer func() { _ = repo.Close() }()	2✔
110
111	// Lookup the specific component version
112	var compVersion ocm.ComponentVersionAccess	1✔
113	if rs.Backoff() == nil {	2✔
114	compVersion, err = repo.LookupComponentVersion(comp, version)	1✔
115	} else {	1✔
116	// If backoff is configured, use it to retry on transient errors	×
117	operation := func() error {	×
118	var err error	×
119	compVersion, err = repo.LookupComponentVersion(comp, version)	×
120	if err != nil {	×
121	// Check if the error is a 429 or transient	×
122	if isRetryable(err) {	×
123	return err // Returning error triggers a retry	×
124	}	×
125
126	return backoff.Permanent(err) // Stops retrying for 401, 404, etc.	×
127	}
128
129	return nil	×
130	}
131	err = backoff.Retry(operation, rs.Backoff())	×
132	}
133	if err != nil {	1✔
134	rs.Logger().Error(err, "failed to lookup component", "version", version)	×
135	return nil, fmt.Errorf("failed to lookup component version %s: %w", version, err)	×
136	}	×
137	defer func() { _ = compVersion.Close() }()	2✔
138
139	// Count the number of Helm chart resources in the component version and determine the handler type based on that.
140	for _, res := range compVersion.GetDescriptor().ComponentSpec.Resources {	6✔
141	if res.Type == string(HelmResource) {	6✔
142	helmChartCount++	1✔
143	}	1✔
144	}
145
146	// Classify component based on contained resources as helm chart and send it to the corresponding handler.
147	if helmChartCount == 1 {	2✔
148	handlerType = HelmHandler	1✔
149	}	1✔
150
151	// If no handler type could be determined, log and publish error.
152	if handlerType == "" {	1✔
153	// No handler found for event, log and publish error.	×
154	rs.Logger().Info("no handler found for event", "event", ev)	×
155	return nil, fmt.Errorf("no handler found for component version event: %v", ev)	×
156	}	×
157
158	// Process component with determined handler type.
159	h, err := rs.getHandler(handlerType)	1✔
160	if err != nil {	1✔
161	rs.Logger().Error(err, "failed to process component with handler", "handler", handlerType)	×
162	return nil, fmt.Errorf("failed to process component with handler %q: %w", handlerType, err)	×
163	}	×
164
165	// Process component with determined handler. If processing fails, log and publish error.
166	resEvent, err := h.Process(ctx, &ev, compVersion)	1✔
167	if err != nil {	1✔
168	rs.Logger().Error(err, "failed to process component with handler", "handler", handlerType)	×
169	return nil, fmt.Errorf("failed to process component with handler %q: %w", handlerType, err)	×
170	}	×
171
172	return []discovery.WriteAPIResourceEvent{*resEvent}, nil	1✔
173	}
174
175	// getHandler returns the handler for the given type, initializing it if necessary.
176	func (rs *Handler) getHandler(t HandlerType) (ComponentHandler, error) {	1✔
177	if rs.handler[HelmHandler] == nil {	2✔
178	if initFn, ok := handlerRegistry[HelmHandler]; ok {	2✔
179	handler := initFn(rs.Logger().WithValues("handler", HelmHandler))	1✔
180	rs.handler[HelmHandler] = handler	1✔
181		1✔
182	return handler, nil	1✔
183	}	1✔
184	}
185
186	return nil, fmt.Errorf("no handler registered for type %v", t)	×
187	}

opendefensecloud / solution-arsenal / 21986023208

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