19761192038

Committed 28 Nov 2025 10:32AM UTC coverage: 61.842% (-0.4%) from 62.247%

Build # 19761192038

Build Type

push

github

Committed by

jastBytes

Commit Message

add docs on how to define new artifact types

Coverage Stats

611 of 988 relevant lines covered (61.84%)

895.65 hits per line

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

84.72

/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: 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.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) {	2,266✔
49	log := ctrl.LoggerFrom(ctx)	2,266✔
50	ctrlResult := ctrl.Result{}	2,266✔
51		2,266✔
52	aw := &arcv1alpha1.ArtifactWorkflow{}	2,266✔
53	if err := r.Get(ctx, req.NamespacedName, aw); err != nil {	2,272✔
54	if apierrors.IsNotFound(err) {	12✔
55	// Object not found, return.	6✔
56	return ctrlResult, nil	6✔
57	}	6✔
58	return ctrlResult, errLogAndWrap(log, err, "failed to get object")	×
59	}
60
61	// Update last reconcile time
62	aw.Status.LastReconcileAt = metav1.Now()	2,260✔
63		2,260✔
64	// Handle deletion	2,260✔
65	if !aw.DeletionTimestamp.IsZero() {	2,265✔
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() {	4,510✔
87	if !slices.Contains(aw.Finalizers, artifactWorkflowFinalizer) {	2,279✔
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)	2,231✔
100	if err != nil {	2,231✔
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)) {	2,232✔
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 {	3,429✔
120	return ctrlResult, r.createArgoWorkflow(ctx, log, aw)	1,199✔
121	}	1,199✔
122
123	if aw.Status.Phase.InProgress() {	2,059✔
124	return ctrlResult, r.checkArgoWorkflow(ctx, log, aw)	1,028✔
125	}	1,028✔
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,199✔
146	srcSecret := corev1.Secret{}	1,199✔
147	if aw.Spec.SrcSecretRef.Name != "" {	2,081✔
148	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.SrcSecretRef.Name), &srcSecret); err != nil {	882✔
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,199✔
155	if aw.Spec.DstSecretRef.Name != "" {	2,081✔
156	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.DstSecretRef.Name), &dstSecret); err != nil {	882✔
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,199✔
163		1,199✔
164	if err := controllerutil.SetControllerReference(aw, wf, r.Scheme); err != nil {	1,199✔
165	return errLogAndWrap(log, err, "failed to set controller reference")	×
166	}	×
167
168	if err := r.Create(ctx, wf); client.IgnoreAlreadyExists(err) != nil {	1,372✔
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,026✔
173		1,026✔
174	aw.Status.Phase = arcv1alpha1.WorkflowPending	1,026✔
175	if err := r.Status().Update(ctx, aw); err != nil {	1,027✔
176	return errLogAndWrap(log, err, "failed to update status")	1✔
177	}	1✔
178	return nil	1,025✔
179	}
180
181	func (r ArtifactWorkflowReconciler) hydrateArgoWorkflow(aw arcv1alpha1.ArtifactWorkflow, srcSecret corev1.Secret, dstSecret corev1.Secret) *wfv1alpha1.Workflow {	1,199✔
182	srcVolume := corev1.Volume{	1,199✔
183	Name: "src-secret-vol",	1,199✔
184	VolumeSource: corev1.VolumeSource{	1,199✔
185	EmptyDir: &corev1.EmptyDirVolumeSource{},	1,199✔
186	},	1,199✔
187	}	1,199✔
188	if srcSecret.Name != "" {	2,081✔
189	srcVolume.VolumeSource = corev1.VolumeSource{	882✔
190	Secret: &corev1.SecretVolumeSource{	882✔
191	SecretName: srcSecret.Name,	882✔
192	},	882✔
193	}	882✔
194	}	882✔
195
196	dstVolume := corev1.Volume{	1,199✔
197	Name: "dst-secret-vol",	1,199✔
198	VolumeSource: corev1.VolumeSource{	1,199✔
199	EmptyDir: &corev1.EmptyDirVolumeSource{},	1,199✔
200	},	1,199✔
201	}	1,199✔
202	if dstSecret.Name != "" {	2,081✔
203	dstVolume.VolumeSource = corev1.VolumeSource{	882✔
204	Secret: &corev1.SecretVolumeSource{	882✔
205	SecretName: dstSecret.Name,	882✔
206	},	882✔
207	}	882✔
208	}	882✔
209
210	parameters := []wfv1alpha1.Parameter{}	1,199✔
211	for _, p := range aw.Spec.Parameters {	7,675✔
212	parameters = append(parameters, wfv1alpha1.Parameter{	6,476✔
213	Name: p.Name,	6,476✔
214	Value: (*wfv1alpha1.AnyString)(&p.Value),	6,476✔
215	})	6,476✔
216	}	6,476✔
217
218	wf := &wfv1alpha1.Workflow{	1,199✔
219	ObjectMeta: workflowObjectMeta(aw),	1,199✔
220	Spec: wfv1alpha1.WorkflowSpec{	1,199✔
221	WorkflowTemplateRef: &wfv1alpha1.WorkflowTemplateRef{	1,199✔
222	Name: aw.Spec.WorkflowTemplateRef.Name,	1,199✔
223	ClusterScope: aw.Spec.WorkflowTemplateRef.ClusterScope,	1,199✔
224	},	1,199✔
225	Volumes: []corev1.Volume{	1,199✔
226	srcVolume,	1,199✔
227	dstVolume,	1,199✔
228	},	1,199✔
229	Arguments: wfv1alpha1.Arguments{	1,199✔
230	Parameters: parameters,	1,199✔
231	},	1,199✔
232	},	1,199✔
233	}	1,199✔
234		1,199✔
235	return wf	1,199✔
236	}
237
238	func (r ArtifactWorkflowReconciler) checkArgoWorkflow(ctx context.Context, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) error {	1,028✔
239	wf := wfv1alpha1.Workflow{}	1,028✔
240	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Name), &wf); err != nil {	1,028✔
241	return errLogAndWrap(log, err, "failed to get workflow")	×
242	}	×
243	if aw.Status.Phase == arcv1alpha1.WorkflowPhase(wf.Status.Phase) {	1,030✔
244	return nil // nothing updated	2✔
245	}	2✔
246	aw.Status.Phase = arcv1alpha1.WorkflowPhase(wf.Status.Phase)	1,026✔
247		1,026✔
248	switch aw.Status.Phase {	1,026✔
249	case arcv1alpha1.WorkflowSucceeded:	2✔
250	aw.Status.CompletionTime = metav1.Now()	2✔
251	case arcv1alpha1.WorkflowError, arcv1alpha1.WorkflowFailed:	1✔
252	// If workflow has errored or failed, fetch logs and update status message	1✔
253	switch aw.Status.Phase {	1✔
254	case arcv1alpha1.WorkflowFailed:	1✔
255	r.generateWorkflowStatusMessage(ctx, wf, log, aw)	1✔
256	case arcv1alpha1.WorkflowError:	×
257	// TODO: Properly show why the workflow errored	×
258	aw.Status.Message = wf.Status.Message	×
259	}
260	}
261
262	if err := r.Status().Update(ctx, aw); err != nil {	1,027✔
263	return errLogAndWrap(log, err, "failed to update status")	1✔
264	}	1✔
265
266	return nil	1,025✔
267	}
268
269	func (r ArtifactWorkflowReconciler) generateWorkflowStatusMessage(ctx context.Context, wf wfv1alpha1.Workflow, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) {	1✔
270	failedNodes := []struct {	1✔
271	Name string	1✔
272	Pod string	1✔
273	Message string	1✔
274	}{}	1✔
275	for _, node := range wf.Status.Nodes {	4✔
276	if node.Phase == wfv1alpha1.NodeFailed && node.Type == wfv1alpha1.NodeTypePod {	5✔
277	nr := struct {	2✔
278	Name string	2✔
279	Pod string	2✔
280	Message string	2✔
281	}{	2✔
282	Name: node.DisplayName,	2✔
283	Pod: generatePodNameFromNodeStatus(node),	2✔
284	Message: node.Message,	2✔
285	}	2✔
286	failedNodes = append(failedNodes, nr)	2✔
287	}	2✔
288	}
289
290	for _, nr := range failedNodes {	3✔
291	logs, err := r.fetchPodLogs(ctx, aw.Namespace, nr.Pod)	2✔
292	if err != nil {	2✔
293	log.V(1).Info("failed to fetch pod logs", "pod", nr.Pod, "error", err)	×
294	continue	×
295	}
296	aw.Status.Message += fmt.Sprintf("Step '%s' failed:\n%s\nLogs:\n%s\n\n", nr.Name, nr.Message, logs)	2✔
297	}
298	}
299
300	func (r *ArtifactWorkflowReconciler) fetchPodLogs(ctx context.Context, namespace, podName string) (string, error) {	2✔
301	podLogOptions := corev1.PodLogOptions{	2✔
302	Container: "main", // Assuming the main container	2✔
303	Follow: false,	2✔
304	TailLines: sprint.ToPointer(int64(30)), // Fetch last 30 lines	2✔
305	}	2✔
306	req := r.ClientSet.CoreV1().Pods(namespace).GetLogs(podName, &podLogOptions)	2✔
307	podLogs, err := req.Stream(ctx)	2✔
308	if err != nil {	2✔
309	return "", err	×
310	}	×
311	defer sprint.PanicOnErrorFunc(podLogs.Close) // Close the stream when done	2✔
312		2✔
313	buf := new(bytes.Buffer)	2✔
314	_, err = io.Copy(buf, podLogs)	2✔
315	if err != nil {	2✔
316	return "", err	×
317	}	×
318	return buf.String(), nil	2✔
319	}
320
321	// SetupWithManager sets up the controller with the Manager.
322	func (r *ArtifactWorkflowReconciler) SetupWithManager(mgr ctrl.Manager) error {	1✔
323	return ctrl.NewControllerManagedBy(mgr).	1✔
324	For(&arcv1alpha1.ArtifactWorkflow{}).	1✔
325	Owns(&wfv1alpha1.Workflow{}).	1✔
326	Complete(r)	1✔
327	}	1✔

opendefensecloud / artifact-conduit / 19761192038

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