19642810445

Committed 24 Nov 2025 05:11PM UTC coverage: 59.251%. First build

Build # 19642810445

Build Type

Pull #70

github

Committed by

web-flow

Commit Message

Merge 5e509c893 into fadaf7570

Pull Request Pull Request #70: Make sure parameters from artifact workflow and artifact type can not…

Coverage Stats

33 of 41 new or added lines in 2 files covered. (80.49%)

522 of 881 relevant lines covered (59.25%)

1009.37 hits per line

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

85.25

/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"
        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
}

//+kubebuilder:rbac:groups=arc.bwi.de,resources=artifacttypes,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

// 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 {
                        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.Spec.Type), &artifactType); err != nil {
                return ctrl.Result{}, errLogAndWrap(log, err, "failed to retrieve artifact type")
        }

        srcSecret := corev1.Secret{}
        if aw.Spec.SrcSecretRef.Name != "" {
                if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.SrcSecretRef.Name), &srcSecret); err != nil {
                        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 {
                        return ctrl.Result{}, errLogAndWrap(log, err, "failed to fetch secret for destination")
                }
        }

        wf := r.hydrateArgoWorkflow(log, aw, &artifactType, &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 {
                return ctrl.Result{}, errLogAndWrap(log, err, "failed to create argo workflow")
        }

        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(log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow, artifactType *arcv1alpha1.ArtifactType, 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,
                        },
                }
        }

        parameterMap := map[string]string{}
        // Initially fill map with workflow parameters
        for _, p := range aw.Spec.Parameters {
                parameterMap[p.Name] = p.Value
        }
        // Spec parameters with higher precedence now may
        // overwrite some parameters of the workflow
        for _, p := range artifactType.Spec.Parameters {
                if _, exists := parameterMap[p.Name]; exists {
                        // Log when an ArtifactType parameter overrides an ArtifactWorkflow parameter
                        log.Info("ArtifactType parameter overriding ArtifactWorkflow parameter",
                                "artifactWorkflow", aw.Name,
                                "artifactType", artifactType.Name,
                                "parameter", p.Name,
                                "workflowValue", parameterMap[p.Name],
                                "typeValue", p.Value)
                }
                parameterMap[p.Name] = p.Value
        }
        parameters := []wfv1alpha1.Parameter{}
        for name, value := range parameterMap {
                parameters = append(parameters, wfv1alpha1.Parameter{
                        Name:  name,
                        Value: (*wfv1alpha1.AnyString)(&value),
                })
        }

        wf := &wfv1alpha1.Workflow{
                ObjectMeta: metav1.ObjectMeta{
                        Name:      aw.Name,
                        Namespace: aw.Namespace,
                },
                Spec: wfv1alpha1.WorkflowSpec{
                        WorkflowTemplateRef: &wfv1alpha1.WorkflowTemplateRef{
                                Name:         artifactType.Spec.WorkflowTemplateRef.Name,
                                ClusterScope: true,
                        },
                        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 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	ctrl "sigs.k8s.io/controller-runtime"
23	"sigs.k8s.io/controller-runtime/pkg/client"
24	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
25	)
26
27	const (
28	artifactWorkflowFinalizer = "arc.bwi.de/artifact-workflow-finalizer"
29	)
30
31	// ArtifactWorkflowReconciler reconciles a ArtifactWorkflow object
32	type ArtifactWorkflowReconciler struct {
33	client.Client
34	ClientSet kubernetes.Interface
35	Scheme *runtime.Scheme
36	}
37
38	//+kubebuilder:rbac:groups=arc.bwi.de,resources=artifacttypes,verbs=get;list;watch
39	//+kubebuilder:rbac:groups=arc.bwi.de,resources=artifactworkflows/status,verbs=get;update;patch
40	//+kubebuilder:rbac:groups=arc.bwi.de,resources=artifactworkflows/finalizers,verbs=update
41	//+kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch;create;update;patch;delete
42	//+kubebuilder:rbac:groups=argoproj.io,resources=workflows,verbs=get;list;watch;create;update;patch;delete
43
44	// Reconcile moves the current state of the cluster closer to the desired state
45	func (r *ArtifactWorkflowReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {	2,591✔
46	log := ctrl.LoggerFrom(ctx)	2,591✔
47		2,591✔
48	aw := &arcv1alpha1.ArtifactWorkflow{}	2,591✔
49	if err := r.Get(ctx, req.NamespacedName, aw); err != nil {	2,594✔
50	if apierrors.IsNotFound(err) {	6✔
51	// Object not found, return.	3✔
52	return ctrl.Result{}, nil	3✔
53	}	3✔
54	return ctrl.Result{}, errLogAndWrap(log, err, "failed to get object")	×
55	}
56
57	if !aw.DeletionTimestamp.IsZero() {	2,591✔
58	log.V(1).Info("ArtifactWorkflow is being deleted")	3✔
59	// Cleanup workflow, if exists	3✔
60	wf := wfv1alpha1.Workflow{	3✔
61	ObjectMeta: metav1.ObjectMeta{	3✔
62	Namespace: aw.Namespace,	3✔
63	Name: aw.Name,	3✔
64	},	3✔
65	}	3✔
66	if err := r.Delete(ctx, &wf); client.IgnoreNotFound(err) != nil {	3✔
67	return ctrl.Result{}, errLogAndWrap(log, err, "workflow deletion failed")	×
68	}	×
69	// Remove finalizer
70	if slices.Contains(aw.Finalizers, artifactWorkflowFinalizer) {	6✔
71	log.V(1).Info("Removing finalizer from ArtifactWorkflow")	3✔
72	aw.Finalizers = slices.DeleteFunc(aw.Finalizers, func(f string) bool {	6✔
73	return f == artifactWorkflowFinalizer	3✔
74	})	3✔
75	if err := r.Update(ctx, aw); err != nil {	3✔
76	return ctrl.Result{}, errLogAndWrap(log, err, "failed to remove finalizer")	×
77	}	×
78	}
79	return ctrl.Result{}, nil	3✔
80	}
81
82	// Add finalizer if not present and not deleting
83	if aw.DeletionTimestamp.IsZero() {	5,170✔
84	if !slices.Contains(aw.Finalizers, artifactWorkflowFinalizer) {	2,604✔
85	log.V(1).Info("Adding finalizer to ArtifactWorkflow")	19✔
86	aw.Finalizers = append(aw.Finalizers, artifactWorkflowFinalizer)	19✔
87	if err := r.Update(ctx, aw); err != nil {	19✔
88	return ctrl.Result{}, errLogAndWrap(log, err, "failed to add finalizer")	×
89	}	×
90	// Return without requeue; the Update event will trigger reconciliation again
91	return ctrl.Result{}, nil	19✔
92	}
93	}
94
95	if aw.Status.Phase == arcv1alpha1.WorkflowUnknown {	3,918✔
96	return r.createArgoWorkflow(ctx, log, aw)	1,352✔
97	}	1,352✔
98
99	if aw.Status.Phase.InProgress() {	2,426✔
100	return r.checkArgoWorkflow(ctx, log, aw)	1,212✔
101	}	1,212✔
102
103	return ctrl.Result{}, nil	2✔
104	}
105
106	func (r ArtifactWorkflowReconciler) createArgoWorkflow(ctx context.Context, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) (ctrl.Result, error) {	1,352✔
107	artifactType := arcv1alpha1.ArtifactType{}	1,352✔
108	if err := r.Get(ctx, namespacedName("", aw.Spec.Type), &artifactType); err != nil {	1,478✔
109	return ctrl.Result{}, errLogAndWrap(log, err, "failed to retrieve artifact type")	126✔
110	}	126✔
111
112	srcSecret := corev1.Secret{}	1,226✔
113	if aw.Spec.SrcSecretRef.Name != "" {	2,275✔
114	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.SrcSecretRef.Name), &srcSecret); err != nil {	1,049✔
115	return ctrl.Result{}, errLogAndWrap(log, err, "failed to fetch secret for source")	×
116	}	×
117	}
118
119	dstSecret := corev1.Secret{}	1,226✔
120	if aw.Spec.DstSecretRef.Name != "" {	2,275✔
121	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Spec.DstSecretRef.Name), &dstSecret); err != nil {	1,049✔
122	return ctrl.Result{}, errLogAndWrap(log, err, "failed to fetch secret for destination")	×
123	}	×
124	}
125
126	wf := r.hydrateArgoWorkflow(log, aw, &artifactType, &srcSecret, &dstSecret)	1,226✔
127		1,226✔
128	if err := controllerutil.SetControllerReference(aw, wf, r.Scheme); err != nil {	1,226✔
129	return ctrl.Result{}, errLogAndWrap(log, err, "failed to set controller reference")	×
130	}	×
131
132	if err := r.Create(ctx, wf); client.IgnoreAlreadyExists(err) != nil {	1,241✔
133	return ctrl.Result{}, errLogAndWrap(log, err, "failed to create argo workflow")	15✔
134	}	15✔
135
136	aw.Status.Phase = arcv1alpha1.WorkflowPending	1,211✔
137	if err := r.Status().Update(ctx, aw); err != nil {	1,213✔
138	return ctrl.Result{}, errLogAndWrap(log, err, "failed to update status")	2✔
139	}	2✔
140	return ctrl.Result{}, nil	1,209✔
141	}
142
143	func (r ArtifactWorkflowReconciler) hydrateArgoWorkflow(log logr.Logger, aw arcv1alpha1.ArtifactWorkflow, artifactType arcv1alpha1.ArtifactType, srcSecret corev1.Secret, dstSecret corev1.Secret) wfv1alpha1.Workflow {	1,226✔
144	srcVolume := corev1.Volume{	1,226✔
145	Name: "src-secret-vol",	1,226✔
146	VolumeSource: corev1.VolumeSource{	1,226✔
147	EmptyDir: &corev1.EmptyDirVolumeSource{},	1,226✔
148	},	1,226✔
149	}	1,226✔
150	if srcSecret.Name != "" {	2,275✔
151	srcVolume.VolumeSource = corev1.VolumeSource{	1,049✔
152	Secret: &corev1.SecretVolumeSource{	1,049✔
153	SecretName: srcSecret.Name,	1,049✔
154	},	1,049✔
155	}	1,049✔
156	}	1,049✔
157
158	dstVolume := corev1.Volume{	1,226✔
159	Name: "dst-secret-vol",	1,226✔
160	VolumeSource: corev1.VolumeSource{	1,226✔
161	EmptyDir: &corev1.EmptyDirVolumeSource{},	1,226✔
162	},	1,226✔
163	}	1,226✔
164	if dstSecret.Name != "" {	2,275✔
165	dstVolume.VolumeSource = corev1.VolumeSource{	1,049✔
166	Secret: &corev1.SecretVolumeSource{	1,049✔
167	SecretName: dstSecret.Name,	1,049✔
168	},	1,049✔
169	}	1,049✔
170	}	1,049✔
171
172	parameterMap := map[string]string{}	1,226✔
173	// Initially fill map with workflow parameters	1,226✔
174	for _, p := range aw.Spec.Parameters {	7,701✔
175	parameterMap[p.Name] = p.Value	6,475✔
176	}	6,475✔
177	// Spec parameters with higher precedence now may
178	// overwrite some parameters of the workflow
179	for _, p := range artifactType.Spec.Parameters {	2,452✔
180	if _, exists := parameterMap[p.Name]; exists {	1,226✔
NEW 181	// Log when an ArtifactType parameter overrides an ArtifactWorkflow parameter	×
NEW 182	log.Info("ArtifactType parameter overriding ArtifactWorkflow parameter",	×
NEW 183	"artifactWorkflow", aw.Name,	×
NEW 184	"artifactType", artifactType.Name,	×
NEW 185	"parameter", p.Name,	×
NEW 186	"workflowValue", parameterMap[p.Name],	×
NEW 187	"typeValue", p.Value)	×
NEW 188	}	×
189	parameterMap[p.Name] = p.Value	1,226✔
190	}
191	parameters := []wfv1alpha1.Parameter{}	1,226✔
192	for name, value := range parameterMap {	8,927✔
193	parameters = append(parameters, wfv1alpha1.Parameter{	7,701✔
194	Name: name,	7,701✔
195	Value: (*wfv1alpha1.AnyString)(&value),	7,701✔
196	})	7,701✔
197	}	7,701✔
198
199	wf := &wfv1alpha1.Workflow{	1,226✔
200	ObjectMeta: metav1.ObjectMeta{	1,226✔
201	Name: aw.Name,	1,226✔
202	Namespace: aw.Namespace,	1,226✔
203	},	1,226✔
204	Spec: wfv1alpha1.WorkflowSpec{	1,226✔
205	WorkflowTemplateRef: &wfv1alpha1.WorkflowTemplateRef{	1,226✔
206	Name: artifactType.Spec.WorkflowTemplateRef.Name,	1,226✔
207	ClusterScope: true,	1,226✔
208	},	1,226✔
209	Volumes: []corev1.Volume{	1,226✔
210	srcVolume,	1,226✔
211	dstVolume,	1,226✔
212	},	1,226✔
213	Arguments: wfv1alpha1.Arguments{	1,226✔
214	Parameters: parameters,	1,226✔
215	},	1,226✔
216	},	1,226✔
217	}	1,226✔
218		1,226✔
219	return wf	1,226✔
220	}
221
222	func (r ArtifactWorkflowReconciler) checkArgoWorkflow(ctx context.Context, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) (ctrl.Result, error) {	1,212✔
223	wf := wfv1alpha1.Workflow{}	1,212✔
224	if err := r.Get(ctx, namespacedName(aw.Namespace, aw.Name), &wf); err != nil {	1,212✔
225	return ctrl.Result{}, errLogAndWrap(log, err, "failed to get workflow")	×
226	}	×
227	if aw.Status.Phase == arcv1alpha1.WorkflowPhase(wf.Status.Phase) {	1,214✔
228	return ctrl.Result{}, nil // nothing updated	2✔
229	}	2✔
230	aw.Status.Phase = arcv1alpha1.WorkflowPhase(wf.Status.Phase)	1,210✔
231		1,210✔
232	// If workflow has errored or failed, fetch logs and update status message	1,210✔
233	if (aw.Status.Phase == arcv1alpha1.WorkflowError \|\| aw.Status.Phase == arcv1alpha1.WorkflowFailed) && aw.Status.Message == "" {	1,211✔
234	switch aw.Status.Phase {	1✔
235	case arcv1alpha1.WorkflowFailed:	1✔
236	r.generateWorkflowStatusMessage(ctx, wf, log, aw)	1✔
237	case arcv1alpha1.WorkflowError:	×
238	// TODO: Properly show why the workflow errored	×
239	aw.Status.Message = wf.Status.Message	×
240	}
241	}
242
243	if err := r.Status().Update(ctx, aw); err != nil {	1,211✔
244	return ctrl.Result{}, errLogAndWrap(log, err, "failed to update status")	1✔
245	}	1✔
246	return ctrl.Result{}, nil	1,209✔
247	}
248
249	func (r ArtifactWorkflowReconciler) generateWorkflowStatusMessage(ctx context.Context, wf wfv1alpha1.Workflow, log logr.Logger, aw arcv1alpha1.ArtifactWorkflow) {	1✔
250	failedNodes := []struct {	1✔
251	Name string	1✔
252	Pod string	1✔
253	Message string	1✔
254	}{}	1✔
255	for _, node := range wf.Status.Nodes {	4✔
256	if node.Phase == wfv1alpha1.NodeFailed && node.Type == wfv1alpha1.NodeTypePod {	5✔
257	nr := struct {	2✔
258	Name string	2✔
259	Pod string	2✔
260	Message string	2✔
261	}{	2✔
262	Name: node.DisplayName,	2✔
263	Pod: generatePodNameFromNodeStatus(node),	2✔
264	Message: node.Message,	2✔
265	}	2✔
266	failedNodes = append(failedNodes, nr)	2✔
267	}	2✔
268	}
269
270	for _, nr := range failedNodes {	3✔
271	logs, err := r.fetchPodLogs(ctx, aw.Namespace, nr.Pod)	2✔
272	if err != nil {	2✔
273	log.V(1).Info("failed to fetch pod logs", "pod", nr.Pod, "error", err)	×
274	continue	×
275	}
276	aw.Status.Message += fmt.Sprintf("Step '%s' failed:\n%s\nLogs:\n%s\n\n", nr.Name, nr.Message, logs)	2✔
277	}
278	}
279
280	func (r *ArtifactWorkflowReconciler) fetchPodLogs(ctx context.Context, namespace, podName string) (string, error) {	2✔
281	podLogOptions := corev1.PodLogOptions{	2✔
282	Container: "main", // Assuming the main container	2✔
283	Follow: false,	2✔
284	TailLines: sprint.ToPointer(int64(30)), // Fetch last 30 lines	2✔
285	}	2✔
286	req := r.ClientSet.CoreV1().Pods(namespace).GetLogs(podName, &podLogOptions)	2✔
287	podLogs, err := req.Stream(ctx)	2✔
288	if err != nil {	2✔
289	return "", err	×
290	}	×
291	defer sprint.PanicOnErrorFunc(podLogs.Close) // Close the stream when done	2✔
292		2✔
293	buf := new(bytes.Buffer)	2✔
294	_, err = io.Copy(buf, podLogs)	2✔
295	if err != nil {	2✔
296	return "", err	×
297	}	×
298	return buf.String(), nil	2✔
299	}
300
301	// SetupWithManager sets up the controller with the Manager.
302	func (r *ArtifactWorkflowReconciler) SetupWithManager(mgr ctrl.Manager) error {	1✔
303	return ctrl.NewControllerManagedBy(mgr).	1✔
304	For(&arcv1alpha1.ArtifactWorkflow{}).	1✔
305	Owns(&wfv1alpha1.Workflow{}).	1✔
306	Complete(r)	1✔
307	}	1✔

opendefensecloud / artifact-conduit / 19642810445

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