19737406833

Committed 27 Nov 2025 01:11PM UTC coverage: 61.735%. Remained the same

Build # 19737406833

Build Type

push

github

Committed by

jastBytes

Commit Message

docs: more docs

Coverage Stats

605 of 980 relevant lines covered (61.73%)

1179.95 hits per line

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

84.91

/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.bwi.de/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.bwi.de,resources=clusterartifacttypes,verbs=get;list;watch
//+kubebuilder:rbac:groups=arc.bwi.de,resources=artifactworkflows/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=arc.bwi.de,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

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

opendefensecloud / artifact-conduit / 19737406833

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