19698835285

Committed 26 Nov 2025 09:26AM UTC coverage: 64.242% (+0.4%) from 63.83%

Build # 19698835285

Build Type

push

github

Committed by

jastBytes

Commit Message

refactor: standardize TTL field naming and update related logic across workflows

Coverage Stats

1 of 2 new or added lines in 1 file covered. (50.0%)

58 existing lines in 2 files now uncovered.

530 of 825 relevant lines covered (64.24%)

888.66 hits per line

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

86.9

/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=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/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)

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

        if !aw.DeletionTimestamp.IsZero() {
                log.V(1).Info("ArtifactWorkflow is being deleted")
                // Cleanup workflow, if exists
                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 ctrl.Result{}, 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 ctrl.Result{}, errLogAndWrap(log, err, "failed to remove finalizer")
                        }
                }
                return ctrl.Result{}, 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 ctrl.Result{}, errLogAndWrap(log, err, "failed to add finalizer")
                        }
                        // Return without requeue; the Update event will trigger reconciliation again
                        return ctrl.Result{}, nil
                }
        }

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

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

        return ctrl.Result{}, nil
}

func (r *ArtifactWorkflowReconciler) createArgoWorkflow(ctx context.Context, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow) (ctrl.Result, error) {
        artifactType := &arcv1alpha1.ArtifactType{}
        if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.Type), artifactType); client.IgnoreNotFound(err) != nil {
                r.Recorder.Event(aw, corev1.EventTypeWarning, "InvalidArtifactType", fmt.Sprintf("Failed to fetch artifact type '%s': %v", aw.Spec.Type, err))
                return ctrl.Result{}, errLogAndWrap(log, err, "failed to fetch referenced ArtifactType")
        }
        var artifactTypeSpec *arcv1alpha1.ArtifactTypeSpec
        if artifactType.Name == "" { // was not found, let's check ClusterArtifactType
                clusterArtifactType := &arcv1alpha1.ClusterArtifactType{}
                if err := r.Get(ctx, namespacedName("", aw.Spec.Type), clusterArtifactType); err != nil {
                        r.Recorder.Event(aw, corev1.EventTypeWarning, "InvalidArtifactType", fmt.Sprintf("Failed to fetch artifact type on cluster scope '%s': %v", aw.Spec.Type, err))
                        return ctrl.Result{}, errLogAndWrap(log, err, "failed to fetch ArtifactType or ClusterArtifactType")
                }
                artifactTypeSpec = &clusterArtifactType.Spec
                // For ClusterArtifactType we only reference ClusterWorkloadTemplates
                artifactTypeSpec.WorkflowTemplateRef.ClusterScope = true
        } else {
                artifactTypeSpec = &artifactType.Spec
        }

        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 ctrl.Result{}, 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 ctrl.Result{}, errLogAndWrap(log, err, "failed to fetch secret for destination")
                }
        }

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

        if err := controllerutil.SetControllerReference(aw, wf, r.Scheme); err != nil {
                return ctrl.Result{}, 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 ctrl.Result{}, 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 ctrl.Result{}, errLogAndWrap(log, err, "failed to update status")
        }
        return ctrl.Result{}, nil
}

func (r *ArtifactWorkflowReconciler) hydrateArgoWorkflow(aw *arcv1alpha1.ArtifactWorkflow, artifactTypeSpec *arcv1alpha1.ArtifactTypeSpec, 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),
                })
        }
        for _, p := range artifactTypeSpec.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:         artifactTypeSpec.WorkflowTemplateRef.Name,
                                ClusterScope: artifactTypeSpec.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) (ctrl.Result, error) {
        wf := wfv1alpha1.Workflow{}
        if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Name), &wf); err != nil {
                return ctrl.Result{}, errLogAndWrap(log, err, "failed to get workflow")
        }
        if aw.Status.Phase == arcv1alpha1.WorkflowPhase(wf.Status.Phase) {
                return ctrl.Result{}, nil // nothing updated
        }
        aw.Status.Phase = arcv1alpha1.WorkflowPhase(wf.Status.Phase)

        if aw.Status.Phase == arcv1alpha1.WorkflowSucceeded {
                aw.Status.CompletionTime = metav1.Now()
        }

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

opendefensecloud / artifact-conduit / 19698835285

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