19670817240

Committed 25 Nov 2025 01:16PM UTC coverage: 63.889% (-0.2%) from 64.132%

Build # 19670817240

Build Type

push

github

Committed by

jastBytes

Commit Message

Enhance OrderStatus structure by adding Message field and updating related conversion functions

Coverage Stats

84 of 113 new or added lines in 1 file covered. (74.34%)

62 existing lines in 2 files now uncovered.

529 of 828 relevant lines covered (63.89%)

793.71 hits per line

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

78.13

/pkg/controller/order_controller.go

// Copyright 2025 BWI GmbH and Artifact Conduit contributors
// SPDX-License-Identifier: Apache-2.0

package controller

import (
        "context"
        "crypto/sha256"
        "encoding/hex"
        "encoding/json"
        "fmt"
        "slices"

        "github.com/go-logr/logr"
        arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1"
        corev1 "k8s.io/api/core/v1"
        apierrors "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/fields"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/tools/record"
        ctrl "sigs.k8s.io/controller-runtime"
        "sigs.k8s.io/controller-runtime/pkg/builder"
        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
        "sigs.k8s.io/controller-runtime/pkg/handler"
        "sigs.k8s.io/controller-runtime/pkg/predicate"
        "sigs.k8s.io/controller-runtime/pkg/reconcile"
)

const (
        orderFinalizer = "arc.bwi.de/order-finalizer"
)

// OrderReconciler reconciles a Order object
type OrderReconciler struct {
        client.Client
        Scheme   *runtime.Scheme
        Recorder record.EventRecorder
}

type desiredAW struct {
        index       int
        objectMeta  metav1.ObjectMeta
        artifact    *arcv1alpha1.OrderArtifact
        srcEndpoint *arcv1alpha1.Endpoint
        dstEndpoint *arcv1alpha1.Endpoint
        srcSecret   *corev1.Secret
        dstSecret   *corev1.Secret
        sha         string
}

//+kubebuilder:rbac:groups=arc.bwi.de,resources=endpoints,verbs=get;list;watch
//+kubebuilder:rbac:groups=arc.bwi.de,resources=artifacttypes,verbs=get;list;watch
//+kubebuilder:rbac:groups=arc.bwi.de,resources=clusterartifacttypes,verbs=get;list;watch
//+kubebuilder:rbac:groups=arc.bwi.de,resources=artifactworkflows,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=arc.bwi.de,resources=orders,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=arc.bwi.de,resources=orders/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=arc.bwi.de,resources=orders/finalizers,verbs=update
//+kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=core,resources=events,verbs=create;patch

// Reconcile moves the current state of the cluster closer to the desired state
func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
        log := ctrl.LoggerFrom(ctx)

        // Fetch the Order instance
        order := &arcv1alpha1.Order{}
        if err := r.Get(ctx, req.NamespacedName, order); err != nil {
                if apierrors.IsNotFound(err) {
                        // Object not found, return. Created objects are automatically garbage collected.
                        return ctrl.Result{}, nil
                }
                return ctrl.Result{}, errLogAndWrap(log, err, "failed to get object")
        }

        // Handle deletion: cleanup artifact workflows, then remove finalizer
        if !order.DeletionTimestamp.IsZero() {
                log.V(1).Info("Order is being deleted")
                r.Recorder.Event(order, corev1.EventTypeWarning, "Deleting", "Order is being deleted, cleaning up artifact workflows")

                // Cleanup all artifact workflows
                if len(order.Status.ArtifactWorkflows) > 0 {
                        for sha := range order.Status.ArtifactWorkflows {
                                // Remove Secret and ArtifactWorkflow
                                aw := &arcv1alpha1.ArtifactWorkflow{
                                        ObjectMeta: awObjectMeta(order, sha),
                                }
                                _ = r.Delete(ctx, aw) // Ignore errors
                                delete(order.Status.ArtifactWorkflows, sha)
                        }
                        if err := r.Status().Update(ctx, order); err != nil {
                                return ctrl.Result{}, errLogAndWrap(log, err, "failed to update order status")
                        }
                        log.V(1).Info("Order artifact workflows cleaned up")

                        // Requeue until all artifact workflows are gone
                        return ctrl.Result{}, nil
                }
                // All artifact workflows are gone, remove finalizer
                if slices.Contains(order.Finalizers, orderFinalizer) {
                        log.V(1).Info("No artifact workflows, removing finalizer from Order")
                        order.Finalizers = slices.DeleteFunc(order.Finalizers, func(f string) bool {
                                return f == orderFinalizer
                        })
                        if err := r.Update(ctx, order); err != nil {
                                return ctrl.Result{}, errLogAndWrap(log, err, "failed to remove finalizer")
                        }
                }
                return ctrl.Result{}, nil
        }

        // Add finalizer if not present and not deleting
        if order.DeletionTimestamp.IsZero() {
                if !slices.Contains(order.Finalizers, orderFinalizer) {
                        log.V(1).Info("Adding finalizer to Order")
                        order.Finalizers = append(order.Finalizers, orderFinalizer)
                        if err := r.Update(ctx, order); err != nil {
                                return ctrl.Result{}, errLogAndWrap(log, err, "failed to add finalizer")
                        }
                        // Return without requeue; the Update event will trigger reconciliation again
                        return ctrl.Result{}, nil
                }
        }

        // Make sure status is initialized
        if order.Status.ArtifactWorkflows == nil {
                order.Status.ArtifactWorkflows = map[string]arcv1alpha1.OrderArtifactWorkflowStatus{}
        }

        // Before we compare to our status, let's fetch all necessary information
        // to compute desired state:
        desiredAWs := map[string]desiredAW{}
        for i, artifact := range order.Spec.Artifacts {
                daw, err := r.computeDesiredAW(ctx, log, order, &artifact, i)
                if err != nil {
                        r.Recorder.Event(order, corev1.EventTypeWarning, "ComputationFailed", fmt.Sprintf("Failed to compute desired artifact workflow for artifact index %d: %v", i, err))
                        order.Status.Message = fmt.Sprintf("Failed to compute desired artifact workflow for artifact index %d: %v", i, err)
                        if err := r.Status().Update(ctx, order); err != nil {
                                return ctrl.Result{}, errLogAndWrap(log, err, "failed to update status")
                        }
                        return ctrl.Result{}, errLogAndWrap(log, err, "failed to compute desired artifact workflow")
                }
                desiredAWs[daw.sha] = *daw
        }

        // List missing artifact workflows
        createAWs := []string{}
        for sha := range desiredAWs {
                _, exists := order.Status.ArtifactWorkflows[sha]
                if exists {
                        continue
                }
                createAWs = append(createAWs, sha)
        }

        // Find obsolete artifact workflows
        deleteAWs := []string{}
        for sha := range order.Status.ArtifactWorkflows {
                _, exists := desiredAWs[sha]
                if exists {
                        continue
                }
                deleteAWs = append(deleteAWs, sha)
        }

        // Create missing artifact workflows
        for _, sha := range createAWs {
                daw := desiredAWs[sha]
                aw, err := r.hydrateArtifactWorkflow(&daw)
                if err != nil {
                        r.Recorder.Event(order, corev1.EventTypeWarning, "HydrationFailed", fmt.Sprintf("Failed to hydrate artifact workflow for artifact index %d: %v", daw.index, err))
                        return ctrl.Result{}, errLogAndWrap(log, err, "failed to hydrate artifact workflow")
                }

                // Set owner references
                if err := controllerutil.SetControllerReference(order, aw, r.Scheme); err != nil {
                        r.Recorder.Event(order, corev1.EventTypeWarning, "HydrationFailed", fmt.Sprintf("Failed to set controller reference for artifact workflow: %v", err))
                        return ctrl.Result{}, errLogAndWrap(log, err, "failed to set controller reference")
                }

                // Create artifact workflow
                if err := r.Create(ctx, aw); err != nil {
                        if apierrors.IsAlreadyExists(err) {
                                // Already created by a previous reconcile — that's fine
                                continue
                        }
                        r.Recorder.Event(order, corev1.EventTypeWarning, "CreationFailed", fmt.Sprintf("Failed to create artifact workflow for artifact index %d: %v", daw.index, err))
                        return ctrl.Result{}, errLogAndWrap(log, err, "failed to create artifact workflow")
                }

                // Update status
                order.Status.ArtifactWorkflows[sha] = arcv1alpha1.OrderArtifactWorkflowStatus{
                        ArtifactIndex: daw.index,
                        Phase:         arcv1alpha1.WorkflowUnknown,
                }

                r.Recorder.Event(order, corev1.EventTypeNormal, "ArtifactWorkflowCreated", fmt.Sprintf("Created artifact workflow '%s' for artifact index %d", aw.Name, daw.index))
                log.V(1).Info("Created artifact workflow", "artifactWorkflow", aw.Name)
        }

        // Delete obsolete artifact workflows
        for _, sha := range deleteAWs {
                // Does not exist anymore, let's clean up!
                if err := r.Delete(ctx, &arcv1alpha1.ArtifactWorkflow{
                        ObjectMeta: awObjectMeta(order, sha),
                }); client.IgnoreNotFound(err) != nil {
                        r.Recorder.Event(order, corev1.EventTypeWarning, "DeletionFailed", fmt.Sprintf("Failed to delete obsolete artifact workflow '%s': %v", sha, err))
                        return ctrl.Result{}, errLogAndWrap(log, err, "failed to delete artifact workflow")
                }

                // Update status
                delete(order.Status.ArtifactWorkflows, sha)
                log.V(1).Info("Deleted obsolete artifact workflow", "artifactWorkflow", sha)
                r.Recorder.Event(order, corev1.EventTypeNormal, "ArtifactWorkflowDeleted", fmt.Sprintf("Deleted obsolete artifact workflow '%s'", sha))
        }

        anyPhaseChanged := false
        for sha, daw := range desiredAWs {
                if slices.Contains(createAWs, sha) {
                        // If it was just created we skip the update
                        continue
                }
                aw := arcv1alpha1.ArtifactWorkflow{}
                if err := r.Get(ctx, namespacedName(daw.objectMeta.Namespace, daw.objectMeta.Name), &aw); err != nil {
                        return ctrl.Result{}, errLogAndWrap(log, err, "failed to get artifact workflow")
                }
                if order.Status.ArtifactWorkflows[sha].Phase != aw.Status.Phase {
                        awStatus := order.Status.ArtifactWorkflows[sha]
                        awStatus.Phase = aw.Status.Phase
                        order.Status.ArtifactWorkflows[sha] = awStatus
                        anyPhaseChanged = true
                }
        }

        // Update status
        if len(createAWs) > 0 || len(deleteAWs) > 0 || anyPhaseChanged {
                log.V(1).Info("Updating order status")
                // Make sure ArtifactIndex is up to date
                for sha, daw := range desiredAWs {
                        aws := order.Status.ArtifactWorkflows[sha]
                        aws.ArtifactIndex = daw.index
                        order.Status.ArtifactWorkflows[sha] = aws
                }
                if err := r.Status().Update(ctx, order); err != nil {
                        return ctrl.Result{}, errLogAndWrap(log, err, "failed to update status")
                }
        }

        return ctrl.Result{}, nil
}

func (r *OrderReconciler) hydrateArtifactWorkflow(daw *desiredAW) (*arcv1alpha1.ArtifactWorkflow, error) {
        params, err := dawToParameters(daw)
        if err != nil {
                return nil, err
        }

        // Next we create the ArtifactWorkflow instance
        aw := &arcv1alpha1.ArtifactWorkflow{
                ObjectMeta: daw.objectMeta,
                Spec: arcv1alpha1.ArtifactWorkflowSpec{
                        Type:         daw.artifact.Type,
                        Parameters:   params,
                        SrcSecretRef: daw.srcEndpoint.Spec.SecretRef,
                        DstSecretRef: daw.dstEndpoint.Spec.SecretRef,
                },
        }

        return aw, nil
}

// generateReconcileRequestsForEndpoint generates reconcile requests for all Endpoints referenced by an Order
func (r *OrderReconciler) generateReconcileRequestsForEndpoint(ctx context.Context, endpoint client.Object) []reconcile.Request {
        resourcesReferencingEndpoint := &arcv1alpha1.OrderList{}
        listOps := &client.ListOptions{
                FieldSelector: fields.SelectorFromSet(fields.Set{".spec.srcRef.name": endpoint.GetName(), ".spec.dstRef.name": endpoint.GetName()}),
                Namespace:     endpoint.GetNamespace(),
        }
        err := r.List(ctx, resourcesReferencingEndpoint, listOps)
        if err != nil {
                return []reconcile.Request{}
        }

        requests := make([]reconcile.Request, len(resourcesReferencingEndpoint.Items))
        for i, item := range resourcesReferencingEndpoint.Items {
                log := ctrl.LoggerFrom(ctx)
                log.V(1).Info("Generating reconcile request for resource because referenced endpoint has changed...")
                requests[i] = reconcile.Request{
                        NamespacedName: types.NamespacedName{
                                Name:      item.GetName(),
                                Namespace: item.GetNamespace(),
                        },
                }
        }
        return requests
}

func (r *OrderReconciler) computeDesiredAW(ctx context.Context, log logr.Logger, order *arcv1alpha1.Order, artifact *arcv1alpha1.OrderArtifact, i int) (*desiredAW, error) {
        log = log.WithValues("artifactIndex", i)

        // We need the referenced src- and dst-endpoints for the artifact
        srcRefName := artifact.SrcRef.Name
        if srcRefName == "" {
                srcRefName = order.Spec.Defaults.SrcRef.Name
        }
        dstRefName := artifact.DstRef.Name
        if dstRefName == "" {
                dstRefName = order.Spec.Defaults.DstRef.Name
        }
        srcEndpoint := &arcv1alpha1.Endpoint{}
        if err := r.Get(ctx, namespacedName(order.Namespace, srcRefName), srcEndpoint); err != nil {
                r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Failed to fetch source endpoint '%s': %v", srcRefName, err))
                return nil, errLogAndWrap(log, err, "failed to fetch endpoint for source")
        }
        dstEndpoint := &arcv1alpha1.Endpoint{}
        if err := r.Get(ctx, namespacedName(order.Namespace, dstRefName), dstEndpoint); err != nil {
                r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Failed to fetch destination endpoint '%s': %v", dstRefName, err))
                return nil, errLogAndWrap(log, err, "failed to fetch endpoint for destination")
        }

        // Validate that the endpoint usage is correct
        if srcEndpoint.Spec.Usage != arcv1alpha1.EndpointUsagePullOnly && srcEndpoint.Spec.Usage != arcv1alpha1.EndpointUsageAll {
                err := fmt.Errorf("endpoint '%s' usage '%s' is not compatible with source usage", srcEndpoint.Name, srcEndpoint.Spec.Usage)
                r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Source endpoint '%s' has incompatible usage '%s'", srcEndpoint.Name, srcEndpoint.Spec.Usage))
                return nil, errLogAndWrap(log, err, "artifact validation failed")
        }
        if dstEndpoint.Spec.Usage != arcv1alpha1.EndpointUsagePushOnly && dstEndpoint.Spec.Usage != arcv1alpha1.EndpointUsageAll {
                err := fmt.Errorf("endpoint '%s' usage '%s' is not compatible with destination usage", dstEndpoint.Name, dstEndpoint.Spec.Usage)
                r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Destination endpoint '%s' has incompatible usage '%s'", dstEndpoint.Name, dstEndpoint.Spec.Usage))
                return nil, errLogAndWrap(log, err, "artifact validation failed")
        }

        // Validate against ArtifactType rules
        artifactType := &arcv1alpha1.ArtifactType{}
        if err := r.Get(ctx, namespacedName(order.Namespace, artifact.Type), artifactType); client.IgnoreNotFound(err) != nil {
                r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidArtifactType", fmt.Sprintf("Failed to fetch ArtifactType '%s': %v", artifact.Type, err))
                return nil, errLogAndWrap(log, err, "failed to fetch referenced ArtifactType")
        }
        var (
                artifactTypeGen  int64
                artifactTypeSpec *arcv1alpha1.ArtifactTypeSpec
        )
        if artifactType.Name == "" { // was not found, let's check ClusterArtifactType
                clusterArtifactType := &arcv1alpha1.ClusterArtifactType{}
                if err := r.Get(ctx, namespacedName("", artifact.Type), clusterArtifactType); err != nil {
                        return nil, errLogAndWrap(log, err, "failed to fetch ArtifactType or ClusterArtifactType")
                }
                artifactTypeSpec = &clusterArtifactType.Spec
                artifactTypeGen = clusterArtifactType.Generation
        } else {
                artifactTypeSpec = &artifactType.Spec
                artifactTypeGen = artifactType.Generation
        }

        if len(artifactTypeSpec.Rules.SrcTypes) > 0 && !slices.Contains(artifactTypeSpec.Rules.SrcTypes, srcEndpoint.Spec.Type) {
                err := fmt.Errorf("source endpoint type '%s' is not allowed by ArtifactType rules", srcEndpoint.Spec.Type)
                r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidArtifactType", fmt.Sprintf("Source endpoint type '%s' is not allowed by ArtifactType '%s' rules", srcEndpoint.Spec.Type, artifact.Type))
                return nil, errLogAndWrap(log, err, "artifact validation failed")
        }
        if len(artifactTypeSpec.Rules.DstTypes) > 0 && !slices.Contains(artifactTypeSpec.Rules.DstTypes, dstEndpoint.Spec.Type) {
                err := fmt.Errorf("destination endpoint type '%s' is not allowed by ArtifactType rules", dstEndpoint.Spec.Type)
                r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidArtifactType", fmt.Sprintf("Destination endpoint type '%s' is not allowed by ArtifactType '%s' rules", dstEndpoint.Spec.Type, artifact.Type))
                return nil, errLogAndWrap(log, err, "artifact validation failed")
        }

        // Next, we need the secret contents
        srcSecret := &corev1.Secret{}
        if srcEndpoint.Spec.SecretRef.Name != "" {
                if err := r.Get(ctx, namespacedName(order.Namespace, srcEndpoint.Spec.SecretRef.Name), srcSecret); err != nil {
                        r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch source secret '%s': %v", srcEndpoint.Spec.SecretRef.Name, err))
                        return nil, errLogAndWrap(log, err, "failed to fetch secret for source")
                }
        }

        dstSecret := &corev1.Secret{}
        if dstEndpoint.Spec.SecretRef.Name != "" {
                if err := r.Get(ctx, namespacedName(order.Namespace, dstEndpoint.Spec.SecretRef.Name), dstSecret); err != nil {
                        r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch destination secret '%s': %v", dstEndpoint.Spec.SecretRef.Name, err))
                        return nil, errLogAndWrap(log, err, "failed to fetch secret for destination")
                }
        }

        // Create a hash based on all related data for idempotency and compute the workflow name
        h := sha256.New()
        data := []any{
                order.Namespace,
                artifact.Type, artifact.Spec.Raw, artifactTypeGen,
                srcEndpoint.Name, srcEndpoint.Generation,
                dstEndpoint.Name, dstEndpoint.Generation,
                srcSecret.Name, srcSecret.Generation,
                dstSecret.Name, dstSecret.Generation,
        }
        jsonData, err := json.Marshal(data)
        if err != nil {
                return nil, errLogAndWrap(log, err, "failed to marshal artifact workflow data")
        }
        h.Write(jsonData)
        sha := hex.EncodeToString(h.Sum(nil))[:16]

        // We gave all the information to further process this artifact workflow.
        // Let's store it to compare it to the current status!
        return &desiredAW{
                index:       i,
                objectMeta:  awObjectMeta(order, sha),
                artifact:    artifact,
                srcEndpoint: srcEndpoint,
                dstEndpoint: dstEndpoint,
                srcSecret:   srcSecret,
                dstSecret:   dstSecret,
                sha:         sha,
        }, nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *OrderReconciler) SetupWithManager(mgr ctrl.Manager) error {
        return ctrl.NewControllerManagedBy(mgr).
                For(&arcv1alpha1.Order{}).
                Watches(
                        &arcv1alpha1.Endpoint{},
                        handler.EnqueueRequestsFromMapFunc(r.generateReconcileRequestsForEndpoint),
                        builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}),
                ).
                Owns(&arcv1alpha1.ArtifactWorkflow{}).
                Complete(r)
}

1	// Copyright 2025 BWI GmbH and Artifact Conduit contributors
2	// SPDX-License-Identifier: Apache-2.0
3
4	package controller
5
6	import (
7	"context"
8	"crypto/sha256"
9	"encoding/hex"
10	"encoding/json"
11	"fmt"
12	"slices"
13
14	"github.com/go-logr/logr"
15	arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1"
16	corev1 "k8s.io/api/core/v1"
17	apierrors "k8s.io/apimachinery/pkg/api/errors"
18	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
19	"k8s.io/apimachinery/pkg/fields"
20	"k8s.io/apimachinery/pkg/runtime"
21	"k8s.io/apimachinery/pkg/types"
22	"k8s.io/client-go/tools/record"
23	ctrl "sigs.k8s.io/controller-runtime"
24	"sigs.k8s.io/controller-runtime/pkg/builder"
25	"sigs.k8s.io/controller-runtime/pkg/client"
26	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
27	"sigs.k8s.io/controller-runtime/pkg/handler"
28	"sigs.k8s.io/controller-runtime/pkg/predicate"
29	"sigs.k8s.io/controller-runtime/pkg/reconcile"
30	)
31
32	const (
33	orderFinalizer = "arc.bwi.de/order-finalizer"
34	)
35
36	// OrderReconciler reconciles a Order object
37	type OrderReconciler struct {
38	client.Client
39	Scheme *runtime.Scheme
40	Recorder record.EventRecorder
41	}
42
43	type desiredAW struct {
44	index int
45	objectMeta metav1.ObjectMeta
46	artifact *arcv1alpha1.OrderArtifact
47	srcEndpoint *arcv1alpha1.Endpoint
48	dstEndpoint *arcv1alpha1.Endpoint
49	srcSecret *corev1.Secret
50	dstSecret *corev1.Secret
51	sha string
52	}
53
54	//+kubebuilder:rbac:groups=arc.bwi.de,resources=endpoints,verbs=get;list;watch
55	//+kubebuilder:rbac:groups=arc.bwi.de,resources=artifacttypes,verbs=get;list;watch
56	//+kubebuilder:rbac:groups=arc.bwi.de,resources=clusterartifacttypes,verbs=get;list;watch
57	//+kubebuilder:rbac:groups=arc.bwi.de,resources=artifactworkflows,verbs=get;list;watch;create;update;patch;delete
58	//+kubebuilder:rbac:groups=arc.bwi.de,resources=orders,verbs=get;list;watch;create;update;patch;delete
59	//+kubebuilder:rbac:groups=arc.bwi.de,resources=orders/status,verbs=get;update;patch
60	//+kubebuilder:rbac:groups=arc.bwi.de,resources=orders/finalizers,verbs=update
61	//+kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch;create;update;patch;delete
62	// +kubebuilder:rbac:groups=core,resources=events,verbs=create;patch
63
64	// Reconcile moves the current state of the cluster closer to the desired state
65	func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	2,013✔
66	log := ctrl.LoggerFrom(ctx)	2,013✔
67		2,013✔
68	// Fetch the Order instance	2,013✔
69	order := &arcv1alpha1.Order{}	2,013✔
70	if err := r.Get(ctx, req.NamespacedName, order); err != nil {	2,016✔
71	if apierrors.IsNotFound(err) {	6✔
72	// Object not found, return. Created objects are automatically garbage collected.	3✔
73	return ctrl.Result{}, nil	3✔
74	}	3✔
UNCOV 75	return ctrl.Result{}, errLogAndWrap(log, err, "failed to get object")	×
76	}
77
78	// Handle deletion: cleanup artifact workflows, then remove finalizer
79	if !order.DeletionTimestamp.IsZero() {	2,012✔
80	log.V(1).Info("Order is being deleted")	2✔
81	r.Recorder.Event(order, corev1.EventTypeWarning, "Deleting", "Order is being deleted, cleaning up artifact workflows")	2✔
82		2✔
83	// Cleanup all artifact workflows	2✔
84	if len(order.Status.ArtifactWorkflows) > 0 {	3✔
85	for sha := range order.Status.ArtifactWorkflows {	3✔
86	// Remove Secret and ArtifactWorkflow	2✔
87	aw := &arcv1alpha1.ArtifactWorkflow{	2✔
88	ObjectMeta: awObjectMeta(order, sha),	2✔
89	}	2✔
90	_ = r.Delete(ctx, aw) // Ignore errors	2✔
91	delete(order.Status.ArtifactWorkflows, sha)	2✔
92	}	2✔
93	if err := r.Status().Update(ctx, order); err != nil {	1✔
UNCOV 94	return ctrl.Result{}, errLogAndWrap(log, err, "failed to update order status")	×
UNCOV 95	}	×
96	log.V(1).Info("Order artifact workflows cleaned up")	1✔
97		1✔
98	// Requeue until all artifact workflows are gone	1✔
99	return ctrl.Result{}, nil	1✔
100	}
101	// All artifact workflows are gone, remove finalizer
102	if slices.Contains(order.Finalizers, orderFinalizer) {	2✔
103	log.V(1).Info("No artifact workflows, removing finalizer from Order")	1✔
104	order.Finalizers = slices.DeleteFunc(order.Finalizers, func(f string) bool {	2✔
105	return f == orderFinalizer	1✔
106	})	1✔
107	if err := r.Update(ctx, order); err != nil {	1✔
UNCOV 108	return ctrl.Result{}, errLogAndWrap(log, err, "failed to remove finalizer")	×
UNCOV 109	}	×
110	}
111	return ctrl.Result{}, nil	1✔
112	}
113
114	// Add finalizer if not present and not deleting
115	if order.DeletionTimestamp.IsZero() {	4,016✔
116	if !slices.Contains(order.Finalizers, orderFinalizer) {	2,018✔
117	log.V(1).Info("Adding finalizer to Order")	10✔
118	order.Finalizers = append(order.Finalizers, orderFinalizer)	10✔
119	if err := r.Update(ctx, order); err != nil {	10✔
UNCOV 120	return ctrl.Result{}, errLogAndWrap(log, err, "failed to add finalizer")	×
UNCOV 121	}	×
122	// Return without requeue; the Update event will trigger reconciliation again
123	return ctrl.Result{}, nil	10✔
124	}
125	}
126
127	// Make sure status is initialized
128	if order.Status.ArtifactWorkflows == nil {	2,020✔
129	order.Status.ArtifactWorkflows = map[string]arcv1alpha1.OrderArtifactWorkflowStatus{}	22✔
130	}	22✔
131
132	// Before we compare to our status, let's fetch all necessary information
133	// to compute desired state:
134	desiredAWs := map[string]desiredAW{}	1,998✔
135	for i, artifact := range order.Spec.Artifacts {	5,207✔
136	daw, err := r.computeDesiredAW(ctx, log, order, &artifact, i)	3,209✔
137	if err != nil {	3,243✔
138	r.Recorder.Event(order, corev1.EventTypeWarning, "ComputationFailed", fmt.Sprintf("Failed to compute desired artifact workflow for artifact index %d: %v", i, err))	34✔
139	order.Status.Message = fmt.Sprintf("Failed to compute desired artifact workflow for artifact index %d: %v", i, err)	34✔
140	if err := r.Status().Update(ctx, order); err != nil {	35✔
141	return ctrl.Result{}, errLogAndWrap(log, err, "failed to update status")	1✔
142	}	1✔
143	return ctrl.Result{}, errLogAndWrap(log, err, "failed to compute desired artifact workflow")	33✔
144	}
145	desiredAWs[daw.sha] = *daw	3,175✔
146	}
147
148	// List missing artifact workflows
149	createAWs := []string{}	1,964✔
150	for sha := range desiredAWs {	5,135✔
151	_, exists := order.Status.ArtifactWorkflows[sha]	3,171✔
152	if exists {	6,324✔
153	continue	3,153✔
154	}
155	createAWs = append(createAWs, sha)	18✔
156	}
157
158	// Find obsolete artifact workflows
159	deleteAWs := []string{}	1,964✔
160	for sha := range order.Status.ArtifactWorkflows {	5,118✔
161	_, exists := desiredAWs[sha]	3,154✔
162	if exists {	6,307✔
163	continue	3,153✔
164	}
165	deleteAWs = append(deleteAWs, sha)	1✔
166	}
167
168	// Create missing artifact workflows
169	for _, sha := range createAWs {	1,982✔
170	daw := desiredAWs[sha]	18✔
171	aw, err := r.hydrateArtifactWorkflow(&daw)	18✔
172	if err != nil {	18✔
UNCOV 173	r.Recorder.Event(order, corev1.EventTypeWarning, "HydrationFailed", fmt.Sprintf("Failed to hydrate artifact workflow for artifact index %d: %v", daw.index, err))	×
UNCOV 174	return ctrl.Result{}, errLogAndWrap(log, err, "failed to hydrate artifact workflow")	×
UNCOV 175	}	×
176
177	// Set owner references
178	if err := controllerutil.SetControllerReference(order, aw, r.Scheme); err != nil {	18✔
UNCOV 179	r.Recorder.Event(order, corev1.EventTypeWarning, "HydrationFailed", fmt.Sprintf("Failed to set controller reference for artifact workflow: %v", err))	×
UNCOV 180	return ctrl.Result{}, errLogAndWrap(log, err, "failed to set controller reference")	×
UNCOV 181	}	×
182
183	// Create artifact workflow
184	if err := r.Create(ctx, aw); err != nil {	20✔
185	if apierrors.IsAlreadyExists(err) {	4✔
186	// Already created by a previous reconcile — that's fine	2✔
187	continue	2✔
188	}
UNCOV 189	r.Recorder.Event(order, corev1.EventTypeWarning, "CreationFailed", fmt.Sprintf("Failed to create artifact workflow for artifact index %d: %v", daw.index, err))	×
UNCOV 190	return ctrl.Result{}, errLogAndWrap(log, err, "failed to create artifact workflow")	×
191	}
192
193	// Update status
194	order.Status.ArtifactWorkflows[sha] = arcv1alpha1.OrderArtifactWorkflowStatus{	16✔
195	ArtifactIndex: daw.index,	16✔
196	Phase: arcv1alpha1.WorkflowUnknown,	16✔
197	}	16✔
198		16✔
199	r.Recorder.Event(order, corev1.EventTypeNormal, "ArtifactWorkflowCreated", fmt.Sprintf("Created artifact workflow '%s' for artifact index %d", aw.Name, daw.index))	16✔
200	log.V(1).Info("Created artifact workflow", "artifactWorkflow", aw.Name)	16✔
201	}
202
203	// Delete obsolete artifact workflows
204	for _, sha := range deleteAWs {	1,965✔
205	// Does not exist anymore, let's clean up!	1✔
206	if err := r.Delete(ctx, &arcv1alpha1.ArtifactWorkflow{	1✔
207	ObjectMeta: awObjectMeta(order, sha),	1✔
208	}); client.IgnoreNotFound(err) != nil {	1✔
UNCOV 209	r.Recorder.Event(order, corev1.EventTypeWarning, "DeletionFailed", fmt.Sprintf("Failed to delete obsolete artifact workflow '%s': %v", sha, err))	×
UNCOV 210	return ctrl.Result{}, errLogAndWrap(log, err, "failed to delete artifact workflow")	×
UNCOV 211	}	×
212
213	// Update status
214	delete(order.Status.ArtifactWorkflows, sha)	1✔
215	log.V(1).Info("Deleted obsolete artifact workflow", "artifactWorkflow", sha)	1✔
216	r.Recorder.Event(order, corev1.EventTypeNormal, "ArtifactWorkflowDeleted", fmt.Sprintf("Deleted obsolete artifact workflow '%s'", sha))	1✔
217	}
218
219	anyPhaseChanged := false	1,964✔
220	for sha, daw := range desiredAWs {	5,135✔
221	if slices.Contains(createAWs, sha) {	3,189✔
222	// If it was just created we skip the update	18✔
223	continue	18✔
224	}
225	aw := arcv1alpha1.ArtifactWorkflow{}	3,153✔
226	if err := r.Get(ctx, namespacedName(daw.objectMeta.Namespace, daw.objectMeta.Name), &aw); err != nil {	3,153✔
UNCOV 227	return ctrl.Result{}, errLogAndWrap(log, err, "failed to get artifact workflow")	×
UNCOV 228	}	×
229	if order.Status.ArtifactWorkflows[sha].Phase != aw.Status.Phase {	4,433✔
230	awStatus := order.Status.ArtifactWorkflows[sha]	1,280✔
231	awStatus.Phase = aw.Status.Phase	1,280✔
232	order.Status.ArtifactWorkflows[sha] = awStatus	1,280✔
233	anyPhaseChanged = true	1,280✔
234	}	1,280✔
235	}
236
237	// Update status
238	if len(createAWs) > 0 \|\| len(deleteAWs) > 0 \|\| anyPhaseChanged {	3,223✔
239	log.V(1).Info("Updating order status")	1,259✔
240	// Make sure ArtifactIndex is up to date	1,259✔
241	for sha, daw := range desiredAWs {	3,311✔
242	aws := order.Status.ArtifactWorkflows[sha]	2,052✔
243	aws.ArtifactIndex = daw.index	2,052✔
244	order.Status.ArtifactWorkflows[sha] = aws	2,052✔
245	}	2,052✔
246	if err := r.Status().Update(ctx, order); err != nil {	1,289✔
247	return ctrl.Result{}, errLogAndWrap(log, err, "failed to update status")	30✔
248	}	30✔
249	}
250
251	return ctrl.Result{}, nil	1,934✔
252	}
253
254	func (r OrderReconciler) hydrateArtifactWorkflow(daw desiredAW) (*arcv1alpha1.ArtifactWorkflow, error) {	18✔
255	params, err := dawToParameters(daw)	18✔
256	if err != nil {	18✔
UNCOV 257	return nil, err	×
UNCOV 258	}	×
259
260	// Next we create the ArtifactWorkflow instance
261	aw := &arcv1alpha1.ArtifactWorkflow{	18✔
262	ObjectMeta: daw.objectMeta,	18✔
263	Spec: arcv1alpha1.ArtifactWorkflowSpec{	18✔
264	Type: daw.artifact.Type,	18✔
265	Parameters: params,	18✔
266	SrcSecretRef: daw.srcEndpoint.Spec.SecretRef,	18✔
267	DstSecretRef: daw.dstEndpoint.Spec.SecretRef,	18✔
268	},	18✔
269	}	18✔
270		18✔
271	return aw, nil	18✔
272	}
273
274	// generateReconcileRequestsForEndpoint generates reconcile requests for all Endpoints referenced by an Order
275	func (r *OrderReconciler) generateReconcileRequestsForEndpoint(ctx context.Context, endpoint client.Object) []reconcile.Request {	29✔
276	resourcesReferencingEndpoint := &arcv1alpha1.OrderList{}	29✔
277	listOps := &client.ListOptions{	29✔
278	FieldSelector: fields.SelectorFromSet(fields.Set{".spec.srcRef.name": endpoint.GetName(), ".spec.dstRef.name": endpoint.GetName()}),	29✔
279	Namespace: endpoint.GetNamespace(),	29✔
280	}	29✔
281	err := r.List(ctx, resourcesReferencingEndpoint, listOps)	29✔
282	if err != nil {	58✔
283	return []reconcile.Request{}	29✔
284	}	29✔
285
UNCOV 286	requests := make([]reconcile.Request, len(resourcesReferencingEndpoint.Items))	×
UNCOV 287	for i, item := range resourcesReferencingEndpoint.Items {	×
UNCOV 288	log := ctrl.LoggerFrom(ctx)	×
UNCOV 289	log.V(1).Info("Generating reconcile request for resource because referenced endpoint has changed...")	×
UNCOV 290	requests[i] = reconcile.Request{	×
UNCOV 291	NamespacedName: types.NamespacedName{	×
UNCOV 292	Name: item.GetName(),	×
UNCOV 293	Namespace: item.GetNamespace(),	×
UNCOV 294	},	×
UNCOV 295	}	×
UNCOV 296	}	×
UNCOV 297	return requests	×
298	}
299
300	func (r OrderReconciler) computeDesiredAW(ctx context.Context, log logr.Logger, order arcv1alpha1.Order, artifact arcv1alpha1.OrderArtifact, i int) (desiredAW, error) {	3,209✔
301	log = log.WithValues("artifactIndex", i)	3,209✔
302		3,209✔
303	// We need the referenced src- and dst-endpoints for the artifact	3,209✔
304	srcRefName := artifact.SrcRef.Name	3,209✔
305	if srcRefName == "" {	3,723✔
306	srcRefName = order.Spec.Defaults.SrcRef.Name	514✔
307	}	514✔
308	dstRefName := artifact.DstRef.Name	3,209✔
309	if dstRefName == "" {	3,896✔
310	dstRefName = order.Spec.Defaults.DstRef.Name	687✔
311	}	687✔
312	srcEndpoint := &arcv1alpha1.Endpoint{}	3,209✔
313	if err := r.Get(ctx, namespacedName(order.Namespace, srcRefName), srcEndpoint); err != nil {	3,209✔
NEW 314	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Failed to fetch source endpoint '%s': %v", srcRefName, err))	×
NEW 315	return nil, errLogAndWrap(log, err, "failed to fetch endpoint for source")	×
NEW 316	}	×
317	dstEndpoint := &arcv1alpha1.Endpoint{}	3,209✔
318	if err := r.Get(ctx, namespacedName(order.Namespace, dstRefName), dstEndpoint); err != nil {	3,209✔
NEW 319	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Failed to fetch destination endpoint '%s': %v", dstRefName, err))	×
NEW 320	return nil, errLogAndWrap(log, err, "failed to fetch endpoint for destination")	×
NEW 321	}	×
322
323	// Validate that the endpoint usage is correct
324	if srcEndpoint.Spec.Usage != arcv1alpha1.EndpointUsagePullOnly && srcEndpoint.Spec.Usage != arcv1alpha1.EndpointUsageAll {	3,209✔
NEW 325	err := fmt.Errorf("endpoint '%s' usage '%s' is not compatible with source usage", srcEndpoint.Name, srcEndpoint.Spec.Usage)	×
NEW 326	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Source endpoint '%s' has incompatible usage '%s'", srcEndpoint.Name, srcEndpoint.Spec.Usage))	×
NEW 327	return nil, errLogAndWrap(log, err, "artifact validation failed")	×
NEW 328	}	×
329	if dstEndpoint.Spec.Usage != arcv1alpha1.EndpointUsagePushOnly && dstEndpoint.Spec.Usage != arcv1alpha1.EndpointUsageAll {	3,209✔
NEW 330	err := fmt.Errorf("endpoint '%s' usage '%s' is not compatible with destination usage", dstEndpoint.Name, dstEndpoint.Spec.Usage)	×
NEW 331	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Destination endpoint '%s' has incompatible usage '%s'", dstEndpoint.Name, dstEndpoint.Spec.Usage))	×
NEW 332	return nil, errLogAndWrap(log, err, "artifact validation failed")	×
NEW 333	}	×
334
335	// Validate against ArtifactType rules
336	artifactType := &arcv1alpha1.ArtifactType{}	3,209✔
337	if err := r.Get(ctx, namespacedName(order.Namespace, artifact.Type), artifactType); client.IgnoreNotFound(err) != nil {	3,209✔
NEW 338	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidArtifactType", fmt.Sprintf("Failed to fetch ArtifactType '%s': %v", artifact.Type, err))	×
NEW 339	return nil, errLogAndWrap(log, err, "failed to fetch referenced ArtifactType")	×
NEW 340	}	×
341	var (	3,209✔
342	artifactTypeGen int64	3,209✔
343	artifactTypeSpec *arcv1alpha1.ArtifactTypeSpec	3,209✔
344	)	3,209✔
345	if artifactType.Name == "" { // was not found, let's check ClusterArtifactType	6,228✔
346	clusterArtifactType := &arcv1alpha1.ClusterArtifactType{}	3,019✔
347	if err := r.Get(ctx, namespacedName("", artifact.Type), clusterArtifactType); err != nil {	3,044✔
348	return nil, errLogAndWrap(log, err, "failed to fetch ArtifactType or ClusterArtifactType")	25✔
349	}	25✔
350	artifactTypeSpec = &clusterArtifactType.Spec	2,994✔
351	artifactTypeGen = clusterArtifactType.Generation	2,994✔
352	} else {	190✔
353	artifactTypeSpec = &artifactType.Spec	190✔
354	artifactTypeGen = artifactType.Generation	190✔
355	}	190✔
356
357	if len(artifactTypeSpec.Rules.SrcTypes) > 0 && !slices.Contains(artifactTypeSpec.Rules.SrcTypes, srcEndpoint.Spec.Type) {	3,193✔
358	err := fmt.Errorf("source endpoint type '%s' is not allowed by ArtifactType rules", srcEndpoint.Spec.Type)	9✔
359	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidArtifactType", fmt.Sprintf("Source endpoint type '%s' is not allowed by ArtifactType '%s' rules", srcEndpoint.Spec.Type, artifact.Type))	9✔
360	return nil, errLogAndWrap(log, err, "artifact validation failed")	9✔
361	}	9✔
362	if len(artifactTypeSpec.Rules.DstTypes) > 0 && !slices.Contains(artifactTypeSpec.Rules.DstTypes, dstEndpoint.Spec.Type) {	3,175✔
NEW 363	err := fmt.Errorf("destination endpoint type '%s' is not allowed by ArtifactType rules", dstEndpoint.Spec.Type)	×
NEW 364	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidArtifactType", fmt.Sprintf("Destination endpoint type '%s' is not allowed by ArtifactType '%s' rules", dstEndpoint.Spec.Type, artifact.Type))	×
NEW 365	return nil, errLogAndWrap(log, err, "artifact validation failed")	×
NEW 366	}	×
367
368	// Next, we need the secret contents
369	srcSecret := &corev1.Secret{}	3,175✔
370	if srcEndpoint.Spec.SecretRef.Name != "" {	6,171✔
371	if err := r.Get(ctx, namespacedName(order.Namespace, srcEndpoint.Spec.SecretRef.Name), srcSecret); err != nil {	2,996✔
NEW 372	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch source secret '%s': %v", srcEndpoint.Spec.SecretRef.Name, err))	×
NEW 373	return nil, errLogAndWrap(log, err, "failed to fetch secret for source")	×
NEW 374	}	×
375	}
376
377	dstSecret := &corev1.Secret{}	3,175✔
378	if dstEndpoint.Spec.SecretRef.Name != "" {	6,171✔
379	if err := r.Get(ctx, namespacedName(order.Namespace, dstEndpoint.Spec.SecretRef.Name), dstSecret); err != nil {	2,996✔
NEW 380	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch destination secret '%s': %v", dstEndpoint.Spec.SecretRef.Name, err))	×
NEW 381	return nil, errLogAndWrap(log, err, "failed to fetch secret for destination")	×
NEW 382	}	×
383	}
384
385	// Create a hash based on all related data for idempotency and compute the workflow name
386	h := sha256.New()	3,175✔
387	data := []any{	3,175✔
388	order.Namespace,	3,175✔
389	artifact.Type, artifact.Spec.Raw, artifactTypeGen,	3,175✔
390	srcEndpoint.Name, srcEndpoint.Generation,	3,175✔
391	dstEndpoint.Name, dstEndpoint.Generation,	3,175✔
392	srcSecret.Name, srcSecret.Generation,	3,175✔
393	dstSecret.Name, dstSecret.Generation,	3,175✔
394	}	3,175✔
395	jsonData, err := json.Marshal(data)	3,175✔
396	if err != nil {	3,175✔
NEW 397	return nil, errLogAndWrap(log, err, "failed to marshal artifact workflow data")	×
NEW 398	}	×
399	h.Write(jsonData)	3,175✔
400	sha := hex.EncodeToString(h.Sum(nil))[:16]	3,175✔
401		3,175✔
402	// We gave all the information to further process this artifact workflow.	3,175✔
403	// Let's store it to compare it to the current status!	3,175✔
404	return &desiredAW{	3,175✔
405	index: i,	3,175✔
406	objectMeta: awObjectMeta(order, sha),	3,175✔
407	artifact: artifact,	3,175✔
408	srcEndpoint: srcEndpoint,	3,175✔
409	dstEndpoint: dstEndpoint,	3,175✔
410	srcSecret: srcSecret,	3,175✔
411	dstSecret: dstSecret,	3,175✔
412	sha: sha,	3,175✔
413	}, nil	3,175✔
414	}
415
416	// SetupWithManager sets up the controller with the Manager.
417	func (r *OrderReconciler) SetupWithManager(mgr ctrl.Manager) error {	1✔
418	return ctrl.NewControllerManagedBy(mgr).	1✔
419	For(&arcv1alpha1.Order{}).	1✔
420	Watches(	1✔
421	&arcv1alpha1.Endpoint{},	1✔
422	handler.EnqueueRequestsFromMapFunc(r.generateReconcileRequestsForEndpoint),	1✔
423	builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}),	1✔
424	).	1✔
425	Owns(&arcv1alpha1.ArtifactWorkflow{}).	1✔
426	Complete(r)	1✔
427	}	1✔

opendefensecloud / artifact-conduit / 19670817240

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