19671042949

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

Build # 19671042949

Build Type

push

github

Committed by

jastBytes

Commit Message

show missing types

Coverage Stats

527 of 828 relevant lines covered (63.65%)

902.84 hits per line

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

77.43

/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,434✔
66	log := ctrl.LoggerFrom(ctx)	2,434✔
67		2,434✔
68	// Fetch the Order instance	2,434✔
69	order := &arcv1alpha1.Order{}	2,434✔
70	if err := r.Get(ctx, req.NamespacedName, order); err != nil {	2,436✔
71	if apierrors.IsNotFound(err) {	4✔
72	// Object not found, return. Created objects are automatically garbage collected.	2✔
73	return ctrl.Result{}, nil	2✔
74	}	2✔
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,434✔
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✔
94	return ctrl.Result{}, errLogAndWrap(log, err, "failed to update order status")	×
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✔
108	return ctrl.Result{}, errLogAndWrap(log, err, "failed to remove finalizer")	×
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,860✔
116	if !slices.Contains(order.Finalizers, orderFinalizer) {	2,440✔
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✔
120	return ctrl.Result{}, errLogAndWrap(log, err, "failed to add finalizer")	×
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,441✔
129	order.Status.ArtifactWorkflows = map[string]arcv1alpha1.OrderArtifactWorkflowStatus{}	21✔
130	}	21✔
131
132	// Before we compare to our status, let's fetch all necessary information
133	// to compute desired state:
134	desiredAWs := map[string]desiredAW{}	2,420✔
135	for i, artifact := range order.Spec.Artifacts {	6,093✔
136	daw, err := r.computeDesiredAW(ctx, log, order, &artifact, i)	3,673✔
137	if err != nil {	3,706✔
138	r.Recorder.Event(order, corev1.EventTypeWarning, "ComputationFailed", fmt.Sprintf("Failed to compute desired artifact workflow for artifact index %d: %v", i, err))	33✔
139	order.Status.Message = fmt.Sprintf("Failed to compute desired artifact workflow for artifact index %d: %v", i, err)	33✔
140	if err := r.Status().Update(ctx, order); err != nil {	33✔
141	return ctrl.Result{}, errLogAndWrap(log, err, "failed to update status")	×
142	}	×
143	return ctrl.Result{}, errLogAndWrap(log, err, "failed to compute desired artifact workflow")	33✔
144	}
145	desiredAWs[daw.sha] = *daw	3,640✔
146	}
147
148	// List missing artifact workflows
149	createAWs := []string{}	2,387✔
150	for sha := range desiredAWs {	6,022✔
151	_, exists := order.Status.ArtifactWorkflows[sha]	3,635✔
152	if exists {	7,253✔
153	continue	3,618✔
154	}
155	createAWs = append(createAWs, sha)	17✔
156	}
157
158	// Find obsolete artifact workflows
159	deleteAWs := []string{}	2,387✔
160	for sha := range order.Status.ArtifactWorkflows {	6,006✔
161	_, exists := desiredAWs[sha]	3,619✔
162	if exists {	7,237✔
163	continue	3,618✔
164	}
165	deleteAWs = append(deleteAWs, sha)	1✔
166	}
167
168	// Create missing artifact workflows
169	for _, sha := range createAWs {	2,404✔
170	daw := desiredAWs[sha]	17✔
171	aw, err := r.hydrateArtifactWorkflow(&daw)	17✔
172	if err != nil {	17✔
173	r.Recorder.Event(order, corev1.EventTypeWarning, "HydrationFailed", fmt.Sprintf("Failed to hydrate artifact workflow for artifact index %d: %v", daw.index, err))	×
174	return ctrl.Result{}, errLogAndWrap(log, err, "failed to hydrate artifact workflow")	×
175	}	×
176
177	// Set owner references
178	if err := controllerutil.SetControllerReference(order, aw, r.Scheme); err != nil {	17✔
179	r.Recorder.Event(order, corev1.EventTypeWarning, "HydrationFailed", fmt.Sprintf("Failed to set controller reference for artifact workflow: %v", err))	×
180	return ctrl.Result{}, errLogAndWrap(log, err, "failed to set controller reference")	×
181	}	×
182
183	// Create artifact workflow
184	if err := r.Create(ctx, aw); err != nil {	18✔
185	if apierrors.IsAlreadyExists(err) {	2✔
186	// Already created by a previous reconcile — that's fine	1✔
187	continue	1✔
188	}
189	r.Recorder.Event(order, corev1.EventTypeWarning, "CreationFailed", fmt.Sprintf("Failed to create artifact workflow for artifact index %d: %v", daw.index, err))	×
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 {	2,388✔
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✔
209	r.Recorder.Event(order, corev1.EventTypeWarning, "DeletionFailed", fmt.Sprintf("Failed to delete obsolete artifact workflow '%s': %v", sha, err))	×
210	return ctrl.Result{}, errLogAndWrap(log, err, "failed to delete artifact workflow")	×
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	2,387✔
220	for sha, daw := range desiredAWs {	6,022✔
221	if slices.Contains(createAWs, sha) {	3,652✔
222	// If it was just created we skip the update	17✔
223	continue	17✔
224	}
225	aw := arcv1alpha1.ArtifactWorkflow{}	3,618✔
226	if err := r.Get(ctx, namespacedName(daw.objectMeta.Namespace, daw.objectMeta.Name), &aw); err != nil {	3,618✔
227	return ctrl.Result{}, errLogAndWrap(log, err, "failed to get artifact workflow")	×
228	}	×
229	if order.Status.ArtifactWorkflows[sha].Phase != aw.Status.Phase {	5,174✔
230	awStatus := order.Status.ArtifactWorkflows[sha]	1,556✔
231	awStatus.Phase = aw.Status.Phase	1,556✔
232	order.Status.ArtifactWorkflows[sha] = awStatus	1,556✔
233	anyPhaseChanged = true	1,556✔
234	}	1,556✔
235	}
236
237	// Update status
238	if len(createAWs) > 0 \|\| len(deleteAWs) > 0 \|\| anyPhaseChanged {	3,910✔
239	log.V(1).Info("Updating order status")	1,523✔
240	// Make sure ArtifactIndex is up to date	1,523✔
241	for sha, daw := range desiredAWs {	3,862✔
242	aws := order.Status.ArtifactWorkflows[sha]	2,339✔
243	aws.ArtifactIndex = daw.index	2,339✔
244	order.Status.ArtifactWorkflows[sha] = aws	2,339✔
245	}	2,339✔
246	if err := r.Status().Update(ctx, order); err != nil {	1,566✔
247	return ctrl.Result{}, errLogAndWrap(log, err, "failed to update status")	43✔
248	}	43✔
249	}
250
251	return ctrl.Result{}, nil	2,344✔
252	}
253
254	func (r OrderReconciler) hydrateArtifactWorkflow(daw desiredAW) (*arcv1alpha1.ArtifactWorkflow, error) {	17✔
255	params, err := dawToParameters(daw)	17✔
256	if err != nil {	17✔
257	return nil, err	×
258	}	×
259
260	// Next we create the ArtifactWorkflow instance
261	aw := &arcv1alpha1.ArtifactWorkflow{	17✔
262	ObjectMeta: daw.objectMeta,	17✔
263	Spec: arcv1alpha1.ArtifactWorkflowSpec{	17✔
264	Type: daw.artifact.Type,	17✔
265	Parameters: params,	17✔
266	SrcSecretRef: daw.srcEndpoint.Spec.SecretRef,	17✔
267	DstSecretRef: daw.dstEndpoint.Spec.SecretRef,	17✔
268	},	17✔
269	}	17✔
270		17✔
271	return aw, nil	17✔
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
286	requests := make([]reconcile.Request, len(resourcesReferencingEndpoint.Items))	×
287	for i, item := range resourcesReferencingEndpoint.Items {	×
288	log := ctrl.LoggerFrom(ctx)	×
289	log.V(1).Info("Generating reconcile request for resource because referenced endpoint has changed...")	×
290	requests[i] = reconcile.Request{	×
291	NamespacedName: types.NamespacedName{	×
292	Name: item.GetName(),	×
293	Namespace: item.GetNamespace(),	×
294	},	×
295	}	×
296	}	×
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,673✔
301	log = log.WithValues("artifactIndex", i)	3,673✔
302		3,673✔
303	// We need the referenced src- and dst-endpoints for the artifact	3,673✔
304	srcRefName := artifact.SrcRef.Name	3,673✔
305	if srcRefName == "" {	4,210✔
306	srcRefName = order.Spec.Defaults.SrcRef.Name	537✔
307	}	537✔
308	dstRefName := artifact.DstRef.Name	3,673✔
309	if dstRefName == "" {	4,390✔
310	dstRefName = order.Spec.Defaults.DstRef.Name	717✔
311	}	717✔
312	srcEndpoint := &arcv1alpha1.Endpoint{}	3,673✔
313	if err := r.Get(ctx, namespacedName(order.Namespace, srcRefName), srcEndpoint); err != nil {	3,673✔
314	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Failed to fetch source endpoint '%s': %v", srcRefName, err))	×
315	return nil, errLogAndWrap(log, err, "failed to fetch endpoint for source")	×
316	}	×
317	dstEndpoint := &arcv1alpha1.Endpoint{}	3,673✔
318	if err := r.Get(ctx, namespacedName(order.Namespace, dstRefName), dstEndpoint); err != nil {	3,673✔
319	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Failed to fetch destination endpoint '%s': %v", dstRefName, err))	×
320	return nil, errLogAndWrap(log, err, "failed to fetch endpoint for destination")	×
321	}	×
322
323	// Validate that the endpoint usage is correct
324	if srcEndpoint.Spec.Usage != arcv1alpha1.EndpointUsagePullOnly && srcEndpoint.Spec.Usage != arcv1alpha1.EndpointUsageAll {	3,673✔
325	err := fmt.Errorf("endpoint '%s' usage '%s' is not compatible with source usage", srcEndpoint.Name, srcEndpoint.Spec.Usage)	×
326	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Source endpoint '%s' has incompatible usage '%s'", srcEndpoint.Name, srcEndpoint.Spec.Usage))	×
327	return nil, errLogAndWrap(log, err, "artifact validation failed")	×
328	}	×
329	if dstEndpoint.Spec.Usage != arcv1alpha1.EndpointUsagePushOnly && dstEndpoint.Spec.Usage != arcv1alpha1.EndpointUsageAll {	3,673✔
330	err := fmt.Errorf("endpoint '%s' usage '%s' is not compatible with destination usage", dstEndpoint.Name, dstEndpoint.Spec.Usage)	×
331	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidEndpoint", fmt.Sprintf("Destination endpoint '%s' has incompatible usage '%s'", dstEndpoint.Name, dstEndpoint.Spec.Usage))	×
332	return nil, errLogAndWrap(log, err, "artifact validation failed")	×
333	}	×
334
335	// Validate against ArtifactType rules
336	artifactType := &arcv1alpha1.ArtifactType{}	3,673✔
337	if err := r.Get(ctx, namespacedName(order.Namespace, artifact.Type), artifactType); client.IgnoreNotFound(err) != nil {	3,673✔
338	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidArtifactType", fmt.Sprintf("Failed to fetch ArtifactType '%s': %v", artifact.Type, err))	×
339	return nil, errLogAndWrap(log, err, "failed to fetch referenced ArtifactType")	×
340	}	×
341	var (	3,673✔
342	artifactTypeGen int64	3,673✔
343	artifactTypeSpec *arcv1alpha1.ArtifactTypeSpec	3,673✔
344	)	3,673✔
345	if artifactType.Name == "" { // was not found, let's check ClusterArtifactType	7,154✔
346	clusterArtifactType := &arcv1alpha1.ClusterArtifactType{}	3,481✔
347	if err := r.Get(ctx, namespacedName("", artifact.Type), clusterArtifactType); err != nil {	3,505✔
348	return nil, errLogAndWrap(log, err, "failed to fetch ArtifactType or ClusterArtifactType")	24✔
349	}	24✔
350	artifactTypeSpec = &clusterArtifactType.Spec	3,457✔
351	artifactTypeGen = clusterArtifactType.Generation	3,457✔
352	} else {	192✔
353	artifactTypeSpec = &artifactType.Spec	192✔
354	artifactTypeGen = artifactType.Generation	192✔
355	}	192✔
356
357	if len(artifactTypeSpec.Rules.SrcTypes) > 0 && !slices.Contains(artifactTypeSpec.Rules.SrcTypes, srcEndpoint.Spec.Type) {	3,658✔
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,640✔
363	err := fmt.Errorf("destination endpoint type '%s' is not allowed by ArtifactType rules", dstEndpoint.Spec.Type)	×
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))	×
365	return nil, errLogAndWrap(log, err, "artifact validation failed")	×
366	}	×
367
368	// Next, we need the secret contents
369	srcSecret := &corev1.Secret{}	3,640✔
370	if srcEndpoint.Spec.SecretRef.Name != "" {	7,080✔
371	if err := r.Get(ctx, namespacedName(order.Namespace, srcEndpoint.Spec.SecretRef.Name), srcSecret); err != nil {	3,440✔
372	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch source secret '%s': %v", srcEndpoint.Spec.SecretRef.Name, err))	×
373	return nil, errLogAndWrap(log, err, "failed to fetch secret for source")	×
374	}	×
375	}
376
377	dstSecret := &corev1.Secret{}	3,640✔
378	if dstEndpoint.Spec.SecretRef.Name != "" {	7,080✔
379	if err := r.Get(ctx, namespacedName(order.Namespace, dstEndpoint.Spec.SecretRef.Name), dstSecret); err != nil {	3,440✔
380	r.Recorder.Event(order, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch destination secret '%s': %v", dstEndpoint.Spec.SecretRef.Name, err))	×
381	return nil, errLogAndWrap(log, err, "failed to fetch secret for destination")	×
382	}	×
383	}
384
385	// Create a hash based on all related data for idempotency and compute the workflow name
386	h := sha256.New()	3,640✔
387	data := []any{	3,640✔
388	order.Namespace,	3,640✔
389	artifact.Type, artifact.Spec.Raw, artifactTypeGen,	3,640✔
390	srcEndpoint.Name, srcEndpoint.Generation,	3,640✔
391	dstEndpoint.Name, dstEndpoint.Generation,	3,640✔
392	srcSecret.Name, srcSecret.Generation,	3,640✔
393	dstSecret.Name, dstSecret.Generation,	3,640✔
394	}	3,640✔
395	jsonData, err := json.Marshal(data)	3,640✔
396	if err != nil {	3,640✔
397	return nil, errLogAndWrap(log, err, "failed to marshal artifact workflow data")	×
398	}	×
399	h.Write(jsonData)	3,640✔
400	sha := hex.EncodeToString(h.Sum(nil))[:16]	3,640✔
401		3,640✔
402	// We gave all the information to further process this artifact workflow.	3,640✔
403	// Let's store it to compare it to the current status!	3,640✔
404	return &desiredAW{	3,640✔
405	index: i,	3,640✔
406	objectMeta: awObjectMeta(order, sha),	3,640✔
407	artifact: artifact,	3,640✔
408	srcEndpoint: srcEndpoint,	3,640✔
409	dstEndpoint: dstEndpoint,	3,640✔
410	srcSecret: srcSecret,	3,640✔
411	dstSecret: dstSecret,	3,640✔
412	sha: sha,	3,640✔
413	}, nil	3,640✔
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 / 19671042949

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