20033280164

Committed 08 Dec 2025 03:25PM UTC coverage: 85.694% (-0.6%) from 86.272%

Build # 20033280164

Build Type

Pull #135

github

Committed by

web-flow

Commit Message

Merge 51f6171a2 into 08dfe7ea4

Pull Request Pull Request #135: add validation to order type

Coverage Stats

593 of 692 relevant lines covered (85.69%)

1047.44 hits per line

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

86.09

/pkg/controller/artifactworkflow_controller.go

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

package controller

import (
        "bytes"
        "context"
        "fmt"
        "io"
        "slices"

        wfv1alpha1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
        "github.com/go-logr/logr"
        "github.com/jastBytes/sprint"
        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/kubernetes"
        "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 (
        artifactWorkflowFinalizer = "arc.opendefense.cloud/artifact-workflow-finalizer"
)

// ArtifactWorkflowReconciler reconciles a ArtifactWorkflow object
type ArtifactWorkflowReconciler struct {
        client.Client
        ClientSet kubernetes.Interface
        Scheme    *runtime.Scheme
        Recorder  record.EventRecorder
}

//+kubebuilder:rbac:groups=arc.opendefense.cloud,resources=clusterartifacttypes,verbs=get;list;watch
//+kubebuilder:rbac:groups=arc.opendefense.cloud,resources=artifactworkflows/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=arc.opendefense.cloud,resources=artifactworkflows/finalizers,verbs=update
//+kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=argoproj.io,resources=workflows,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=core,resources=events,verbs=create;patch
//+kubebuilder:rbac:groups="",resources=pods;pods/log,verbs=get;list

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

        aw := &arcv1alpha1.ArtifactWorkflow{}
        if err := r.Get(ctx, req.NamespacedName, aw); err != nil {
                if apierrors.IsNotFound(err) {
                        // Object not found, return.
                        return ctrlResult, nil
                }
                return ctrlResult, errLogAndWrap(log, err, "failed to get object")
        }

        // Update last reconcile time
        aw.Status.LastReconcileAt = metav1.Now()

        // Handle deletion
        if !aw.DeletionTimestamp.IsZero() {
                log.V(1).Info("ArtifactWorkflow is being deleted")
                // Cleanup workflow, if exists
                if err := r.deleteArgoWorkflow(ctx, log, aw); err != nil {
                        return ctrlResult, errLogAndWrap(log, err, "workflow deletion failed")
                }

                // Remove finalizer
                if slices.Contains(aw.Finalizers, artifactWorkflowFinalizer) {
                        log.V(1).Info("Removing finalizer from ArtifactWorkflow")
                        aw.Finalizers = slices.DeleteFunc(aw.Finalizers, func(f string) bool {
                                return f == artifactWorkflowFinalizer
                        })
                        if err := r.Update(ctx, aw); err != nil {
                                return ctrlResult, errLogAndWrap(log, err, "failed to remove finalizer")
                        }
                }
                return ctrlResult, nil
        }

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

        // Handle force reconcile annotation
        forceAt, err := GetForceAtAnnotationValue(aw)
        if err != nil {
                log.V(1).Error(err, "Invalid force reconcile annotation, ignoring")
        }
        if !forceAt.IsZero() && (aw.Status.LastForceAt.IsZero() || forceAt.After(aw.Status.LastForceAt.Time)) {
                log.V(1).Info("Force reconcile requested")
                r.Recorder.Event(aw, corev1.EventTypeNormal, "ForceReconcile", "Force reconcile requested via annotation")
                // Delete existing workflow, if any
                if err := r.deleteArgoWorkflow(ctx, log, aw); err != nil {
                        return ctrlResult, errLogAndWrap(log, err, "failed to delete existing workflow for force reconcile")
                }
                // Reset phase so workflow gets recreated, and update last force time
                aw.Status.Phase = arcv1alpha1.WorkflowUnknown
                aw.Status.LastForceAt = metav1.Now()
                if err := r.Status().Update(ctx, aw); err != nil {
                        return ctrlResult, errLogAndWrap(log, err, "failed to update last force time")
                }
                // Return without requeue; the update event will trigger reconciliation again
                return ctrlResult, nil
        }

        if aw.Status.Phase == arcv1alpha1.WorkflowUnknown {
                return ctrlResult, r.createArgoWorkflow(ctx, log, aw)
        }

        if aw.Status.Phase.InProgress() {
                return ctrlResult, r.checkArgoWorkflow(ctx, log, aw)
        }

        return ctrlResult, nil
}

func (r *ArtifactWorkflowReconciler) deleteArgoWorkflow(ctx context.Context, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow) error {
        wf := wfv1alpha1.Workflow{
                ObjectMeta: metav1.ObjectMeta{
                        Namespace: aw.Namespace,
                        Name:      aw.Name,
                },
        }
        if err := r.Delete(ctx, &wf); client.IgnoreNotFound(err) != nil {
                r.Recorder.Event(aw, corev1.EventTypeWarning, "DeletionFailed", fmt.Sprintf("Failed to delete associated workflow '%s': %v", aw.Name, err))
                return errLogAndWrap(log, err, "workflow deletion failed")
        }
        r.Recorder.Event(aw, corev1.EventTypeNormal, "Deleted", fmt.Sprintf("Deleted workflow '%s'", aw.Name))
        return nil
}

func (r *ArtifactWorkflowReconciler) createArgoWorkflow(ctx context.Context, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow) error {
        srcSecret := corev1.Secret{}
        if aw.Spec.SrcSecretRef.Name != "" {
                if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.SrcSecretRef.Name), &srcSecret); err != nil {
                        r.Recorder.Event(aw, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch source secret '%s': %v", aw.Spec.SrcSecretRef.Name, err))
                        return errLogAndWrap(log, err, "failed to fetch secret for source")
                }
        }

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

        wf := r.hydrateArgoWorkflow(aw, &srcSecret, &dstSecret)

        if err := controllerutil.SetControllerReference(aw, wf, r.Scheme); err != nil {
                return errLogAndWrap(log, err, "failed to set controller reference")
        }

        if err := r.Create(ctx, wf); client.IgnoreAlreadyExists(err) != nil {
                r.Recorder.Event(aw, corev1.EventTypeWarning, "CreationFailed", fmt.Sprintf("Failed to create workflow '%s': %v", wf.Name, err))
                return errLogAndWrap(log, err, "failed to create argo workflow")
        }
        r.Recorder.Event(aw, corev1.EventTypeNormal, "Created", fmt.Sprintf("Created workflow '%s'", wf.Name))

        aw.Status.Phase = arcv1alpha1.WorkflowPending
        if err := r.Status().Update(ctx, aw); err != nil {
                return errLogAndWrap(log, err, "failed to update status")
        }
        return nil
}

func (r *ArtifactWorkflowReconciler) hydrateArgoWorkflow(aw *arcv1alpha1.ArtifactWorkflow, srcSecret *corev1.Secret, dstSecret *corev1.Secret) *wfv1alpha1.Workflow {
        srcVolume := corev1.Volume{
                Name: "src-secret-vol",
                VolumeSource: corev1.VolumeSource{
                        EmptyDir: &corev1.EmptyDirVolumeSource{},
                },
        }
        if srcSecret.Name != "" {
                srcVolume.VolumeSource = corev1.VolumeSource{
                        Secret: &corev1.SecretVolumeSource{
                                SecretName: srcSecret.Name,
                        },
                }
        }

        dstVolume := corev1.Volume{
                Name: "dst-secret-vol",
                VolumeSource: corev1.VolumeSource{
                        EmptyDir: &corev1.EmptyDirVolumeSource{},
                },
        }
        if dstSecret.Name != "" {
                dstVolume.VolumeSource = corev1.VolumeSource{
                        Secret: &corev1.SecretVolumeSource{
                                SecretName: dstSecret.Name,
                        },
                }
        }

        parameters := []wfv1alpha1.Parameter{}
        for _, p := range aw.Spec.Parameters {
                parameters = append(parameters, wfv1alpha1.Parameter{
                        Name:  p.Name,
                        Value: (*wfv1alpha1.AnyString)(&p.Value),
                })
        }

        wf := &wfv1alpha1.Workflow{
                ObjectMeta: workflowObjectMeta(aw),
                Spec: wfv1alpha1.WorkflowSpec{
                        WorkflowTemplateRef: &wfv1alpha1.WorkflowTemplateRef{
                                Name:         aw.Spec.WorkflowTemplateRef.Name,
                                ClusterScope: aw.Spec.WorkflowTemplateRef.ClusterScope,
                        },
                        Volumes: []corev1.Volume{
                                srcVolume,
                                dstVolume,
                        },
                        Arguments: wfv1alpha1.Arguments{
                                Parameters: parameters,
                        },
                },
        }

        return wf
}

func (r *ArtifactWorkflowReconciler) checkArgoWorkflow(ctx context.Context, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow) error {
        wf := wfv1alpha1.Workflow{}
        if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Name), &wf); err != nil {
                return errLogAndWrap(log, err, "failed to get workflow")
        }
        if aw.Status.Phase == arcv1alpha1.WorkflowPhase(wf.Status.Phase) {
                return nil // nothing updated
        }
        aw.Status.Phase = arcv1alpha1.WorkflowPhase(wf.Status.Phase)

        switch aw.Status.Phase {
        case arcv1alpha1.WorkflowSucceeded:
                aw.Status.CompletionTime = metav1.Now()
        case arcv1alpha1.WorkflowError, arcv1alpha1.WorkflowFailed:
                // If workflow has errored or failed, fetch logs and update status message
                switch aw.Status.Phase {
                case arcv1alpha1.WorkflowFailed:
                        r.generateWorkflowStatusMessage(ctx, wf, log, aw)
                case arcv1alpha1.WorkflowError:
                        // TODO: Properly show why the workflow errored
                        aw.Status.Message = wf.Status.Message
                }
        }

        if err := r.Status().Update(ctx, aw); err != nil {
                return errLogAndWrap(log, err, "failed to update status")
        }

        return nil
}

func (r *ArtifactWorkflowReconciler) generateWorkflowStatusMessage(ctx context.Context, wf wfv1alpha1.Workflow, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow) {
        failedNodes := []struct {
                Name    string
                Pod     string
                Message string
        }{}
        for _, node := range wf.Status.Nodes {
                if node.Phase == wfv1alpha1.NodeFailed && node.Type == wfv1alpha1.NodeTypePod {
                        nr := struct {
                                Name    string
                                Pod     string
                                Message string
                        }{
                                Name:    node.DisplayName,
                                Pod:     generatePodNameFromNodeStatus(node),
                                Message: node.Message,
                        }
                        failedNodes = append(failedNodes, nr)
                }
        }

        for _, nr := range failedNodes {
                logs, err := r.fetchPodLogs(ctx, aw.Namespace, nr.Pod)
                if err != nil {
                        log.V(1).Info("failed to fetch pod logs", "pod", nr.Pod, "error", err)
                        continue
                }
                aw.Status.Message += fmt.Sprintf("Step '%s' failed:\n%s\nLogs:\n%s\n\n", nr.Name, nr.Message, logs)
        }
}

func (r *ArtifactWorkflowReconciler) fetchPodLogs(ctx context.Context, namespace, podName string) (string, error) {
        podLogOptions := corev1.PodLogOptions{
                Container: "main", // Assuming the main container
                Follow:    false,
                TailLines: sprint.ToPointer(int64(30)), // Fetch last 30 lines
        }
        req := r.ClientSet.CoreV1().Pods(namespace).GetLogs(podName, &podLogOptions)
        podLogs, err := req.Stream(ctx)
        if err != nil {
                return "", err
        }
        defer sprint.PanicOnErrorFunc(podLogs.Close) // Close the stream when done

        buf := new(bytes.Buffer)
        _, err = io.Copy(buf, podLogs)
        if err != nil {
                return "", err
        }
        return buf.String(), nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *ArtifactWorkflowReconciler) SetupWithManager(mgr ctrl.Manager) error {
        return ctrl.NewControllerManagedBy(mgr).
                For(&arcv1alpha1.ArtifactWorkflow{}).
                Owns(&wfv1alpha1.Workflow{}).
                Complete(r)
}

1	// Copyright 2025 BWI GmbH and Artefact Conduit contributors
2	// SPDX-License-Identifier: Apache-2.0
3
4	package controller
5
6	import (
7	"bytes"
8	"context"
9	"fmt"
10	"io"
11	"slices"
12
13	wfv1alpha1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
14	"github.com/go-logr/logr"
15	"github.com/jastBytes/sprint"
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/kubernetes"
22	"k8s.io/client-go/tools/record"
23	ctrl "sigs.k8s.io/controller-runtime"
24	"sigs.k8s.io/controller-runtime/pkg/client"
25	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
26	)
27
28	const (
29	artifactWorkflowFinalizer = "arc.opendefense.cloud/artifact-workflow-finalizer"
30	)
31
32	// ArtifactWorkflowReconciler reconciles a ArtifactWorkflow object
33	type ArtifactWorkflowReconciler struct {
34	client.Client
35	ClientSet kubernetes.Interface
36	Scheme *runtime.Scheme
37	Recorder record.EventRecorder
38	}
39
40	//+kubebuilder:rbac:groups=arc.opendefense.cloud,resources=clusterartifacttypes,verbs=get;list;watch
41	//+kubebuilder:rbac:groups=arc.opendefense.cloud,resources=artifactworkflows/status,verbs=get;update;patch
42	//+kubebuilder:rbac:groups=arc.opendefense.cloud,resources=artifactworkflows/finalizers,verbs=update
43	//+kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch;create;update;patch;delete
44	//+kubebuilder:rbac:groups=argoproj.io,resources=workflows,verbs=get;list;watch;create;update;patch;delete
45	//+kubebuilder:rbac:groups=core,resources=events,verbs=create;patch
46	//+kubebuilder:rbac:groups="",resources=pods;pods/log,verbs=get;list
47
48	// Reconcile moves the current state of the cluster closer to the desired state
49	func (r *ArtifactWorkflowReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	1,870✔
50	log := ctrl.LoggerFrom(ctx)	1,870✔
51	ctrlResult := ctrl.Result{}	1,870✔
52		1,870✔
53	aw := &arcv1alpha1.ArtifactWorkflow{}	1,870✔
54	if err := r.Get(ctx, req.NamespacedName, aw); err != nil {	1,875✔
55	if apierrors.IsNotFound(err) {	10✔
56	// Object not found, return.	5✔
57	return ctrlResult, nil	5✔
58	}	5✔
59	return ctrlResult, errLogAndWrap(log, err, "failed to get object")	×
60	}
61
62	// Update last reconcile time
63	aw.Status.LastReconcileAt = metav1.Now()	1,865✔
64		1,865✔
65	// Handle deletion	1,865✔
66	if !aw.DeletionTimestamp.IsZero() {	1,870✔
67	log.V(1).Info("ArtifactWorkflow is being deleted")	5✔
68	// Cleanup workflow, if exists	5✔
69	if err := r.deleteArgoWorkflow(ctx, log, aw); err != nil {	5✔
70	return ctrlResult, errLogAndWrap(log, err, "workflow deletion failed")	×
71	}	×
72
73	// Remove finalizer
74	if slices.Contains(aw.Finalizers, artifactWorkflowFinalizer) {	10✔
75	log.V(1).Info("Removing finalizer from ArtifactWorkflow")	5✔
76	aw.Finalizers = slices.DeleteFunc(aw.Finalizers, func(f string) bool {	10✔
77	return f == artifactWorkflowFinalizer	5✔
78	})	5✔
79	if err := r.Update(ctx, aw); err != nil {	5✔
80	return ctrlResult, errLogAndWrap(log, err, "failed to remove finalizer")	×
81	}	×
82	}
83	return ctrlResult, nil	5✔
84	}
85
86	// Add finalizer if not present and not deleting
87	if aw.DeletionTimestamp.IsZero() {	3,720✔
88	if !slices.Contains(aw.Finalizers, artifactWorkflowFinalizer) {	1,887✔
89	log.V(1).Info("Adding finalizer to ArtifactWorkflow")	27✔
90	aw.Finalizers = append(aw.Finalizers, artifactWorkflowFinalizer)	27✔
91	if err := r.Update(ctx, aw); err != nil {	27✔
92	return ctrlResult, errLogAndWrap(log, err, "failed to add finalizer")	×
93	}	×
94	// Return without requeue; the Update event will trigger reconciliation again
95	return ctrlResult, nil	27✔
96	}
97	}
98
99	// Handle force reconcile annotation
100	forceAt, err := GetForceAtAnnotationValue(aw)	1,833✔
101	if err != nil {	1,833✔
102	log.V(1).Error(err, "Invalid force reconcile annotation, ignoring")	×
103	}	×
104	if !forceAt.IsZero() && (aw.Status.LastForceAt.IsZero() \|\| forceAt.After(aw.Status.LastForceAt.Time)) {	1,834✔
105	log.V(1).Info("Force reconcile requested")	1✔
106	r.Recorder.Event(aw, corev1.EventTypeNormal, "ForceReconcile", "Force reconcile requested via annotation")	1✔
107	// Delete existing workflow, if any	1✔
108	if err := r.deleteArgoWorkflow(ctx, log, aw); err != nil {	1✔
109	return ctrlResult, errLogAndWrap(log, err, "failed to delete existing workflow for force reconcile")	×
110	}	×
111	// Reset phase so workflow gets recreated, and update last force time
112	aw.Status.Phase = arcv1alpha1.WorkflowUnknown	1✔
113	aw.Status.LastForceAt = metav1.Now()	1✔
114	if err := r.Status().Update(ctx, aw); err != nil {	1✔
115	return ctrlResult, errLogAndWrap(log, err, "failed to update last force time")	×
116	}	×
117	// Return without requeue; the update event will trigger reconciliation again
118	return ctrlResult, nil	1✔
119	}
120
121	if aw.Status.Phase == arcv1alpha1.WorkflowUnknown {	2,847✔
122	return ctrlResult, r.createArgoWorkflow(ctx, log, aw)	1,015✔
123	}	1,015✔
124
125	if aw.Status.Phase.InProgress() {	1,630✔
126	return ctrlResult, r.checkArgoWorkflow(ctx, log, aw)	813✔
127	}	813✔
128
129	return ctrlResult, nil	4✔
130	}
131
132	func (r ArtifactWorkflowReconciler) deleteArgoWorkflow(ctx context.Context, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) error {	6✔
133	wf := wfv1alpha1.Workflow{	6✔
134	ObjectMeta: metav1.ObjectMeta{	6✔
135	Namespace: aw.Namespace,	6✔
136	Name: aw.Name,	6✔
137	},	6✔
138	}	6✔
139	if err := r.Delete(ctx, &wf); client.IgnoreNotFound(err) != nil {	6✔
140	r.Recorder.Event(aw, corev1.EventTypeWarning, "DeletionFailed", fmt.Sprintf("Failed to delete associated workflow '%s': %v", aw.Name, err))	×
141	return errLogAndWrap(log, err, "workflow deletion failed")	×
142	}	×
143	r.Recorder.Event(aw, corev1.EventTypeNormal, "Deleted", fmt.Sprintf("Deleted workflow '%s'", aw.Name))	6✔
144	return nil	6✔
145	}
146
147	func (r ArtifactWorkflowReconciler) createArgoWorkflow(ctx context.Context, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) error {	1,015✔
148	srcSecret := corev1.Secret{}	1,015✔
149	if aw.Spec.SrcSecretRef.Name != "" {	1,735✔
150	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.SrcSecretRef.Name), &srcSecret); err != nil {	720✔
151	r.Recorder.Event(aw, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch source secret '%s': %v", aw.Spec.SrcSecretRef.Name, err))	×
152	return errLogAndWrap(log, err, "failed to fetch secret for source")	×
153	}	×
154	}
155
156	dstSecret := corev1.Secret{}	1,015✔
157	if aw.Spec.DstSecretRef.Name != "" {	1,735✔
158	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.DstSecretRef.Name), &dstSecret); err != nil {	720✔
159	r.Recorder.Event(aw, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch destination secret '%s': %v", aw.Spec.DstSecretRef.Name, err))	×
160	return errLogAndWrap(log, err, "failed to fetch secret for destination")	×
161	}	×
162	}
163
164	wf := r.hydrateArgoWorkflow(aw, &srcSecret, &dstSecret)	1,015✔
165		1,015✔
166	if err := controllerutil.SetControllerReference(aw, wf, r.Scheme); err != nil {	1,015✔
167	return errLogAndWrap(log, err, "failed to set controller reference")	×
168	}	×
169
170	if err := r.Create(ctx, wf); client.IgnoreAlreadyExists(err) != nil {	1,213✔
171	r.Recorder.Event(aw, corev1.EventTypeWarning, "CreationFailed", fmt.Sprintf("Failed to create workflow '%s': %v", wf.Name, err))	198✔
172	return errLogAndWrap(log, err, "failed to create argo workflow")	198✔
173	}	198✔
174	r.Recorder.Event(aw, corev1.EventTypeNormal, "Created", fmt.Sprintf("Created workflow '%s'", wf.Name))	817✔
175		817✔
176	aw.Status.Phase = arcv1alpha1.WorkflowPending	817✔
177	if err := r.Status().Update(ctx, aw); err != nil {	822✔
178	return errLogAndWrap(log, err, "failed to update status")	5✔
179	}	5✔
180	return nil	812✔
181	}
182
183	func (r ArtifactWorkflowReconciler) hydrateArgoWorkflow(aw arcv1alpha1.ArtifactWorkflow, srcSecret corev1.Secret, dstSecret corev1.Secret) *wfv1alpha1.Workflow {	1,015✔
184	srcVolume := corev1.Volume{	1,015✔
185	Name: "src-secret-vol",	1,015✔
186	VolumeSource: corev1.VolumeSource{	1,015✔
187	EmptyDir: &corev1.EmptyDirVolumeSource{},	1,015✔
188	},	1,015✔
189	}	1,015✔
190	if srcSecret.Name != "" {	1,735✔
191	srcVolume.VolumeSource = corev1.VolumeSource{	720✔
192	Secret: &corev1.SecretVolumeSource{	720✔
193	SecretName: srcSecret.Name,	720✔
194	},	720✔
195	}	720✔
196	}	720✔
197
198	dstVolume := corev1.Volume{	1,015✔
199	Name: "dst-secret-vol",	1,015✔
200	VolumeSource: corev1.VolumeSource{	1,015✔
201	EmptyDir: &corev1.EmptyDirVolumeSource{},	1,015✔
202	},	1,015✔
203	}	1,015✔
204	if dstSecret.Name != "" {	1,735✔
205	dstVolume.VolumeSource = corev1.VolumeSource{	720✔
206	Secret: &corev1.SecretVolumeSource{	720✔
207	SecretName: dstSecret.Name,	720✔
208	},	720✔
209	}	720✔
210	}	720✔
211
212	parameters := []wfv1alpha1.Parameter{}	1,015✔
213	for _, p := range aw.Spec.Parameters {	6,631✔
214	parameters = append(parameters, wfv1alpha1.Parameter{	5,616✔
215	Name: p.Name,	5,616✔
216	Value: (*wfv1alpha1.AnyString)(&p.Value),	5,616✔
217	})	5,616✔
218	}	5,616✔
219
220	wf := &wfv1alpha1.Workflow{	1,015✔
221	ObjectMeta: workflowObjectMeta(aw),	1,015✔
222	Spec: wfv1alpha1.WorkflowSpec{	1,015✔
223	WorkflowTemplateRef: &wfv1alpha1.WorkflowTemplateRef{	1,015✔
224	Name: aw.Spec.WorkflowTemplateRef.Name,	1,015✔
225	ClusterScope: aw.Spec.WorkflowTemplateRef.ClusterScope,	1,015✔
226	},	1,015✔
227	Volumes: []corev1.Volume{	1,015✔
228	srcVolume,	1,015✔
229	dstVolume,	1,015✔
230	},	1,015✔
231	Arguments: wfv1alpha1.Arguments{	1,015✔
232	Parameters: parameters,	1,015✔
233	},	1,015✔
234	},	1,015✔
235	}	1,015✔
236		1,015✔
237	return wf	1,015✔
238	}
239
240	func (r ArtifactWorkflowReconciler) checkArgoWorkflow(ctx context.Context, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) error {	813✔
241	wf := wfv1alpha1.Workflow{}	813✔
242	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Name), &wf); err != nil {	813✔
243	return errLogAndWrap(log, err, "failed to get workflow")	×
244	}	×
245	if aw.Status.Phase == arcv1alpha1.WorkflowPhase(wf.Status.Phase) {	816✔
246	return nil // nothing updated	3✔
247	}	3✔
248	aw.Status.Phase = arcv1alpha1.WorkflowPhase(wf.Status.Phase)	810✔
249		810✔
250	switch aw.Status.Phase {	810✔
251	case arcv1alpha1.WorkflowSucceeded:	2✔
252	aw.Status.CompletionTime = metav1.Now()	2✔
253	case arcv1alpha1.WorkflowError, arcv1alpha1.WorkflowFailed:	2✔
254	// If workflow has errored or failed, fetch logs and update status message	2✔
255	switch aw.Status.Phase {	2✔
256	case arcv1alpha1.WorkflowFailed:	1✔
257	r.generateWorkflowStatusMessage(ctx, wf, log, aw)	1✔
258	case arcv1alpha1.WorkflowError:	1✔
259	// TODO: Properly show why the workflow errored	1✔
260	aw.Status.Message = wf.Status.Message	1✔
261	}
262	}
263
264	if err := r.Status().Update(ctx, aw); err != nil {	811✔
265	return errLogAndWrap(log, err, "failed to update status")	1✔
266	}	1✔
267
268	return nil	809✔
269	}
270
271	func (r ArtifactWorkflowReconciler) generateWorkflowStatusMessage(ctx context.Context, wf wfv1alpha1.Workflow, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) {	1✔
272	failedNodes := []struct {	1✔
273	Name string	1✔
274	Pod string	1✔
275	Message string	1✔
276	}{}	1✔
277	for _, node := range wf.Status.Nodes {	4✔
278	if node.Phase == wfv1alpha1.NodeFailed && node.Type == wfv1alpha1.NodeTypePod {	5✔
279	nr := struct {	2✔
280	Name string	2✔
281	Pod string	2✔
282	Message string	2✔
283	}{	2✔
284	Name: node.DisplayName,	2✔
285	Pod: generatePodNameFromNodeStatus(node),	2✔
286	Message: node.Message,	2✔
287	}	2✔
288	failedNodes = append(failedNodes, nr)	2✔
289	}	2✔
290	}
291
292	for _, nr := range failedNodes {	3✔
293	logs, err := r.fetchPodLogs(ctx, aw.Namespace, nr.Pod)	2✔
294	if err != nil {	2✔
295	log.V(1).Info("failed to fetch pod logs", "pod", nr.Pod, "error", err)	×
296	continue	×
297	}
298	aw.Status.Message += fmt.Sprintf("Step '%s' failed:\n%s\nLogs:\n%s\n\n", nr.Name, nr.Message, logs)	2✔
299	}
300	}
301
302	func (r *ArtifactWorkflowReconciler) fetchPodLogs(ctx context.Context, namespace, podName string) (string, error) {	2✔
303	podLogOptions := corev1.PodLogOptions{	2✔
304	Container: "main", // Assuming the main container	2✔
305	Follow: false,	2✔
306	TailLines: sprint.ToPointer(int64(30)), // Fetch last 30 lines	2✔
307	}	2✔
308	req := r.ClientSet.CoreV1().Pods(namespace).GetLogs(podName, &podLogOptions)	2✔
309	podLogs, err := req.Stream(ctx)	2✔
310	if err != nil {	2✔
311	return "", err	×
312	}	×
313	defer sprint.PanicOnErrorFunc(podLogs.Close) // Close the stream when done	2✔
314		2✔
315	buf := new(bytes.Buffer)	2✔
316	_, err = io.Copy(buf, podLogs)	2✔
317	if err != nil {	2✔
318	return "", err	×
319	}	×
320	return buf.String(), nil	2✔
321	}
322
323	// SetupWithManager sets up the controller with the Manager.
324	func (r *ArtifactWorkflowReconciler) SetupWithManager(mgr ctrl.Manager) error {	1✔
325	return ctrl.NewControllerManagedBy(mgr).	1✔
326	For(&arcv1alpha1.ArtifactWorkflow{}).	1✔
327	Owns(&wfv1alpha1.Workflow{}).	1✔
328	Complete(r)	1✔
329	}	1✔

opendefensecloud / artifact-conduit / 20033280164

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