19715181387

Committed 26 Nov 2025 07:26PM UTC coverage: 60.656% (-0.7%) from 61.352%

Build # 19715181387

Build Type

push

github

Committed by

web-flow

Commit Message

Feature/87 workflow template ref refactor (#94)

Closes #87

Coverage Stats

38 of 38 new or added lines in 3 files covered. (100.0%)

4 existing lines in 2 files now uncovered.

555 of 915 relevant lines covered (60.66%)

746.89 hits per line

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

80.62

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

opendefensecloud / artifact-conduit / 19715181387

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