19822120502

Committed 01 Dec 2025 12:08PM UTC coverage: 65.622% (+4.3%) from 61.352%

Build # 19822120502

Build Type

Pull #96

github

Committed by

web-flow

Commit Message

Merge 71a705260 into 7683f393f

Pull Request Pull Request #96: Issue #23: Artifact Immutability

Coverage Stats

649 of 989 relevant lines covered (65.62%)

688.13 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.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")
                }
                // 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.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,071✔
49	log := ctrl.LoggerFrom(ctx)	2,071✔
50	ctrlResult := ctrl.Result{}	2,071✔
51		2,071✔
52	aw := &arcv1alpha1.ArtifactWorkflow{}	2,071✔
53	if err := r.Get(ctx, req.NamespacedName, aw); err != nil {	2,076✔
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()	2,066✔
63		2,066✔
64	// Handle deletion	2,066✔
65	if !aw.DeletionTimestamp.IsZero() {	2,071✔
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,122✔
87	if !slices.Contains(aw.Finalizers, artifactWorkflowFinalizer) {	2,087✔
88	log.V(1).Info("Adding finalizer to ArtifactWorkflow")	26✔
89	aw.Finalizers = append(aw.Finalizers, artifactWorkflowFinalizer)	26✔
90	if err := r.Update(ctx, aw); err != nil {	26✔
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	26✔
95	}
96	}
97
98	// Handle force reconcile annotation
99	forceAt, err := GetForceAtAnnotationValue(aw)	2,035✔
100	if err != nil {	2,035✔
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,036✔
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	// Reset phase so workflow gets recreated, and update last force time
111	aw.Status.Phase = arcv1alpha1.WorkflowUnknown	1✔
112	aw.Status.LastForceAt = metav1.Now()	1✔
113	if err := r.Status().Update(ctx, aw); err != nil {	1✔
114	return ctrlResult, errLogAndWrap(log, err, "failed to update last force time")	×
115	}	×
116	// Return without requeue; the update event will trigger reconciliation again
117	return ctrlResult, nil	1✔
118	}
119
120	if aw.Status.Phase == arcv1alpha1.WorkflowUnknown {	3,143✔
121	return ctrlResult, r.createArgoWorkflow(ctx, log, aw)	1,109✔
122	}	1,109✔
123
124	if aw.Status.Phase.InProgress() {	1,845✔
125	return ctrlResult, r.checkArgoWorkflow(ctx, log, aw)	920✔
126	}	920✔
127
128	return ctrlResult, nil	5✔
129	}
130
131	func (r ArtifactWorkflowReconciler) deleteArgoWorkflow(ctx context.Context, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) error {	6✔
132	wf := wfv1alpha1.Workflow{	6✔
133	ObjectMeta: metav1.ObjectMeta{	6✔
134	Namespace: aw.Namespace,	6✔
135	Name: aw.Name,	6✔
136	},	6✔
137	}	6✔
138	if err := r.Delete(ctx, &wf); client.IgnoreNotFound(err) != nil {	6✔
139	r.Recorder.Event(aw, corev1.EventTypeWarning, "DeletionFailed", fmt.Sprintf("Failed to delete associated workflow '%s': %v", aw.Name, err))	×
140	return errLogAndWrap(log, err, "workflow deletion failed")	×
141	}	×
142	r.Recorder.Event(aw, corev1.EventTypeNormal, "Deleted", fmt.Sprintf("Deleted workflow '%s'", aw.Name))	6✔
143	return nil	6✔
144	}
145
146	func (r ArtifactWorkflowReconciler) createArgoWorkflow(ctx context.Context, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) error {	1,109✔
147	srcSecret := corev1.Secret{}	1,109✔
148	if aw.Spec.SrcSecretRef.Name != "" {	1,804✔
149	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.SrcSecretRef.Name), &srcSecret); err != nil {	695✔
150	r.Recorder.Event(aw, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch source secret '%s': %v", aw.Spec.SrcSecretRef.Name, err))	×
151	return errLogAndWrap(log, err, "failed to fetch secret for source")	×
152	}	×
153	}
154
155	dstSecret := corev1.Secret{}	1,109✔
156	if aw.Spec.DstSecretRef.Name != "" {	1,804✔
157	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.DstSecretRef.Name), &dstSecret); err != nil {	695✔
158	r.Recorder.Event(aw, corev1.EventTypeWarning, "InvalidSecret", fmt.Sprintf("Failed to fetch destination secret '%s': %v", aw.Spec.DstSecretRef.Name, err))	×
159	return errLogAndWrap(log, err, "failed to fetch secret for destination")	×
160	}	×
161	}
162
163	wf := r.hydrateArgoWorkflow(aw, &srcSecret, &dstSecret)	1,109✔
164		1,109✔
165	if err := controllerutil.SetControllerReference(aw, wf, r.Scheme); err != nil {	1,109✔
166	return errLogAndWrap(log, err, "failed to set controller reference")	×
167	}	×
168
169	if err := r.Create(ctx, wf); client.IgnoreAlreadyExists(err) != nil {	1,298✔
170	r.Recorder.Event(aw, corev1.EventTypeWarning, "CreationFailed", fmt.Sprintf("Failed to create workflow '%s': %v", wf.Name, err))	189✔
171	return errLogAndWrap(log, err, "failed to create argo workflow")	189✔
172	}	189✔
173	r.Recorder.Event(aw, corev1.EventTypeNormal, "Created", fmt.Sprintf("Created workflow '%s'", wf.Name))	920✔
174		920✔
175	aw.Status.Phase = arcv1alpha1.WorkflowPending	920✔
176	if err := r.Status().Update(ctx, aw); err != nil {	923✔
177	return errLogAndWrap(log, err, "failed to update status")	3✔
178	}	3✔
179	return nil	917✔
180	}
181
182	func (r ArtifactWorkflowReconciler) hydrateArgoWorkflow(aw arcv1alpha1.ArtifactWorkflow, srcSecret corev1.Secret, dstSecret corev1.Secret) *wfv1alpha1.Workflow {	1,109✔
183	srcVolume := corev1.Volume{	1,109✔
184	Name: "src-secret-vol",	1,109✔
185	VolumeSource: corev1.VolumeSource{	1,109✔
186	EmptyDir: &corev1.EmptyDirVolumeSource{},	1,109✔
187	},	1,109✔
188	}	1,109✔
189	if srcSecret.Name != "" {	1,804✔
190	srcVolume.VolumeSource = corev1.VolumeSource{	695✔
191	Secret: &corev1.SecretVolumeSource{	695✔
192	SecretName: srcSecret.Name,	695✔
193	},	695✔
194	}	695✔
195	}	695✔
196
197	dstVolume := corev1.Volume{	1,109✔
198	Name: "dst-secret-vol",	1,109✔
199	VolumeSource: corev1.VolumeSource{	1,109✔
200	EmptyDir: &corev1.EmptyDirVolumeSource{},	1,109✔
201	},	1,109✔
202	}	1,109✔
203	if dstSecret.Name != "" {	1,804✔
204	dstVolume.VolumeSource = corev1.VolumeSource{	695✔
205	Secret: &corev1.SecretVolumeSource{	695✔
206	SecretName: dstSecret.Name,	695✔
207	},	695✔
208	}	695✔
209	}	695✔
210
211	parameters := []wfv1alpha1.Parameter{}	1,109✔
212	for _, p := range aw.Spec.Parameters {	6,759✔
213	parameters = append(parameters, wfv1alpha1.Parameter{	5,650✔
214	Name: p.Name,	5,650✔
215	Value: (*wfv1alpha1.AnyString)(&p.Value),	5,650✔
216	})	5,650✔
217	}	5,650✔
218
219	wf := &wfv1alpha1.Workflow{	1,109✔
220	ObjectMeta: workflowObjectMeta(aw),	1,109✔
221	Spec: wfv1alpha1.WorkflowSpec{	1,109✔
222	WorkflowTemplateRef: &wfv1alpha1.WorkflowTemplateRef{	1,109✔
223	Name: aw.Spec.WorkflowTemplateRef.Name,	1,109✔
224	ClusterScope: aw.Spec.WorkflowTemplateRef.ClusterScope,	1,109✔
225	},	1,109✔
226	Volumes: []corev1.Volume{	1,109✔
227	srcVolume,	1,109✔
228	dstVolume,	1,109✔
229	},	1,109✔
230	Arguments: wfv1alpha1.Arguments{	1,109✔
231	Parameters: parameters,	1,109✔
232	},	1,109✔
233	},	1,109✔
234	}	1,109✔
235		1,109✔
236	return wf	1,109✔
237	}
238
239	func (r ArtifactWorkflowReconciler) checkArgoWorkflow(ctx context.Context, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) error {	920✔
240	wf := wfv1alpha1.Workflow{}	920✔
241	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Name), &wf); err != nil {	920✔
242	return errLogAndWrap(log, err, "failed to get workflow")	×
243	}	×
244	if aw.Status.Phase == arcv1alpha1.WorkflowPhase(wf.Status.Phase) {	925✔
245	return nil // nothing updated	5✔
246	}	5✔
247	aw.Status.Phase = arcv1alpha1.WorkflowPhase(wf.Status.Phase)	915✔
248		915✔
249	switch aw.Status.Phase {	915✔
250	case arcv1alpha1.WorkflowSucceeded:	2✔
251	aw.Status.CompletionTime = metav1.Now()	2✔
252	case arcv1alpha1.WorkflowError, arcv1alpha1.WorkflowFailed:	2✔
253	// If workflow has errored or failed, fetch logs and update status message	2✔
254	switch aw.Status.Phase {	2✔
255	case arcv1alpha1.WorkflowFailed:	1✔
256	r.generateWorkflowStatusMessage(ctx, wf, log, aw)	1✔
257	case arcv1alpha1.WorkflowError:	1✔
258	// TODO: Properly show why the workflow errored	1✔
259	aw.Status.Message = wf.Status.Message	1✔
260	}
261	}
262
263	if err := r.Status().Update(ctx, aw); err != nil {	917✔
264	return errLogAndWrap(log, err, "failed to update status")	2✔
265	}	2✔
266
267	return nil	913✔
268	}
269
270	func (r ArtifactWorkflowReconciler) generateWorkflowStatusMessage(ctx context.Context, wf wfv1alpha1.Workflow, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) {	1✔
271	failedNodes := []struct {	1✔
272	Name string	1✔
273	Pod string	1✔
274	Message string	1✔
275	}{}	1✔
276	for _, node := range wf.Status.Nodes {	4✔
277	if node.Phase == wfv1alpha1.NodeFailed && node.Type == wfv1alpha1.NodeTypePod {	5✔
278	nr := struct {	2✔
279	Name string	2✔
280	Pod string	2✔
281	Message string	2✔
282	}{	2✔
283	Name: node.DisplayName,	2✔
284	Pod: generatePodNameFromNodeStatus(node),	2✔
285	Message: node.Message,	2✔
286	}	2✔
287	failedNodes = append(failedNodes, nr)	2✔
288	}	2✔
289	}
290
291	for _, nr := range failedNodes {	3✔
292	logs, err := r.fetchPodLogs(ctx, aw.Namespace, nr.Pod)	2✔
293	if err != nil {	2✔
294	log.V(1).Info("failed to fetch pod logs", "pod", nr.Pod, "error", err)	×
295	continue	×
296	}
297	aw.Status.Message += fmt.Sprintf("Step '%s' failed:\n%s\nLogs:\n%s\n\n", nr.Name, nr.Message, logs)	2✔
298	}
299	}
300
301	func (r *ArtifactWorkflowReconciler) fetchPodLogs(ctx context.Context, namespace, podName string) (string, error) {	2✔
302	podLogOptions := corev1.PodLogOptions{	2✔
303	Container: "main", // Assuming the main container	2✔
304	Follow: false,	2✔
305	TailLines: sprint.ToPointer(int64(30)), // Fetch last 30 lines	2✔
306	}	2✔
307	req := r.ClientSet.CoreV1().Pods(namespace).GetLogs(podName, &podLogOptions)	2✔
308	podLogs, err := req.Stream(ctx)	2✔
309	if err != nil {	2✔
310	return "", err	×
311	}	×
312	defer sprint.PanicOnErrorFunc(podLogs.Close) // Close the stream when done	2✔
313		2✔
314	buf := new(bytes.Buffer)	2✔
315	_, err = io.Copy(buf, podLogs)	2✔
316	if err != nil {	2✔
317	return "", err	×
318	}	×
319	return buf.String(), nil	2✔
320	}
321
322	// SetupWithManager sets up the controller with the Manager.
323	func (r *ArtifactWorkflowReconciler) SetupWithManager(mgr ctrl.Manager) error {	1✔
324	return ctrl.NewControllerManagedBy(mgr).	1✔
325	For(&arcv1alpha1.ArtifactWorkflow{}).	1✔
326	Owns(&wfv1alpha1.Workflow{}).	1✔
327	Complete(r)	1✔
328	}	1✔

opendefensecloud / artifact-conduit / 19822120502

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