19701017879

Committed 26 Nov 2025 10:45AM UTC coverage: 61.457% (-2.2%) from 63.647%

Build # 19701017879

Build Type

Pull #70

github

Committed by

web-flow

Commit Message

Merge 0b8fd311c into 9a3f70af8

Pull Request Pull Request #70: Make sure parameters from artifact workflow and artifact type can not…

Coverage Stats

46 of 46 new or added lines in 2 files covered. (100.0%)

51 existing lines in 2 files now uncovered.

582 of 947 relevant lines covered (61.46%)

887.4 hits per line

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

79.51

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

        "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/runtime"
        "k8s.io/client-go/tools/record"
        ctrl "sigs.k8s.io/controller-runtime"
        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
)

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

        // Find finished artifact workflows to clean up
        finishedAWs := []string{}
        for sha := range order.Status.ArtifactWorkflows {
                awStatus := order.Status.ArtifactWorkflows[sha]

                // Only consider succeeded workflows for TTL cleanup
                if awStatus.Phase != arcv1alpha1.WorkflowSucceeded {
                        continue
                }

                // If TTL is set, check if it has expired
                if order.Spec.TTLSecondsAfterCompletion != nil && *order.Spec.TTLSecondsAfterCompletion > 0 {
                        if time.Since(awStatus.CompletionTime.Time) > time.Duration(*order.Spec.TTLSecondsAfterCompletion)*time.Second {
                                finishedAWs = append(finishedAWs, sha)
                        }
                        continue
                }

                // No TTL set, cleanup immediately
                finishedAWs = append(finishedAWs, 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))
        }

        // Delete finished artifact workflows
        for _, sha := range finishedAWs {
                // Finished, 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 finished artifact workflow '%s': %v", sha, err))
                        return ctrl.Result{}, errLogAndWrap(log, err, "failed to delete artifact workflow")
                }

                log.V(1).Info("Deleted finished artifact workflow", "artifactWorkflow", sha)
                r.Recorder.Event(order, corev1.EventTypeNormal, "ArtifactWorkflowDeleted", fmt.Sprintf("Deleted finished 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
                        awStatus.CompletionTime = aw.Status.CompletionTime
                        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
}

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

opendefensecloud / artifact-conduit / 19701017879

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