• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

opendefensecloud / artifact-conduit / 20776998573

07 Jan 2026 09:36AM UTC coverage: 83.673% (-1.4%) from 85.034%
20776998573

Pull #165

github

web-flow
Merge 234bb52de into 91d2cf8b0
Pull Request #165: Feature/helm workflow example use trivy

738 of 882 relevant lines covered (83.67%)

936.22 hits per line

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

80.73
/pkg/controller/workflow_handler.go
1
// Copyright 2025 BWI GmbH and Artifact Conduit contributors
2
// SPDX-License-Identifier: Apache-2.0
3

4
package controller
5

6
import (
7
        "context"
8
        "fmt"
9

10
        wfv1alpha1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
11
        "github.com/go-logr/logr"
12
        "github.com/jastBytes/sprint"
13
        arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1"
14
        corev1 "k8s.io/api/core/v1"
15
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
16
        "sigs.k8s.io/controller-runtime/pkg/client"
17
        "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
18
)
19

20
type WorkflowHandler interface {
21
        DeleteArgoResources(ctx context.Context) error
22
        CreateArgoResources(ctx context.Context) error
23
        CheckArgoResources(ctx context.Context) error
24
}
25

26
var _ WorkflowHandler = &SingleWorkflowHandler{}
27

28
type SingleWorkflowHandler struct {
29
        *ArtifactWorkflowReconciler
30
        log logr.Logger
31
        aw  *arcv1alpha1.ArtifactWorkflow
32
}
33

34
func NewSingleWorkflowHandler(r *ArtifactWorkflowReconciler, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow) *SingleWorkflowHandler {
2,100✔
35
        return &SingleWorkflowHandler{r, log, aw}
2,100✔
36
}
2,100✔
37

38
func (h *SingleWorkflowHandler) DeleteArgoResources(ctx context.Context) error {
6✔
39
        wf := wfv1alpha1.Workflow{
6✔
40
                ObjectMeta: metav1.ObjectMeta{
6✔
41
                        Namespace: h.aw.Namespace,
6✔
42
                        Name:      h.aw.Name,
6✔
43
                },
6✔
44
        }
6✔
45
        if err := h.Delete(ctx, &wf); client.IgnoreNotFound(err) != nil {
6✔
46
                h.Recorder.Event(h.aw, corev1.EventTypeWarning, "DeletionFailed", fmt.Sprintf("Failed to delete associated workflow '%s': %v", h.aw.Name, err))
×
47
                return errLogAndWrap(h.log, err, "workflow deletion failed")
×
48
        }
×
49
        h.Recorder.Event(h.aw, corev1.EventTypeNormal, "Deleted", fmt.Sprintf("Deleted workflow '%s'", h.aw.Name))
6✔
50
        return nil
6✔
51
}
52

53
func (h *SingleWorkflowHandler) CreateArgoResources(ctx context.Context) error {
1,138✔
54
        srcSecret, dstSecret, err := h.retrieveSecrets(ctx, h.aw)
1,138✔
55
        if err != nil {
1,138✔
56
                return errLogAndWrap(h.log, err, "failed to fetch secrets for artifact workflow")
×
57
        }
×
58

59
        wf := hydrateArgoWorkflow(h.aw, srcSecret, dstSecret)
1,138✔
60

1,138✔
61
        if err := controllerutil.SetControllerReference(h.aw, wf, h.Scheme); err != nil {
1,138✔
62
                return errLogAndWrap(h.log, err, "failed to set controller reference")
×
63
        }
×
64

65
        if err := h.Create(ctx, wf); client.IgnoreAlreadyExists(err) != nil {
1,347✔
66
                h.Recorder.Event(h.aw, corev1.EventTypeWarning, "CreationFailed", fmt.Sprintf("Failed to create workflow '%s': %v", wf.GetName(), err))
209✔
67
                return errLogAndWrap(h.log, err, "failed to create argo workflow")
209✔
68
        }
209✔
69
        h.Recorder.Event(h.aw, corev1.EventTypeNormal, "Created", fmt.Sprintf("Created workflow '%s'", wf.GetName()))
929✔
70

929✔
71
        h.aw.Status.Phase = arcv1alpha1.WorkflowPending
929✔
72
        if err := h.Status().Update(ctx, h.aw); err != nil {
936✔
73
                return errLogAndWrap(h.log, err, "failed to update status")
7✔
74
        }
7✔
75
        return nil
922✔
76
}
77

78
func (h *SingleWorkflowHandler) CheckArgoResources(ctx context.Context) error {
925✔
79
        wf := wfv1alpha1.Workflow{}
925✔
80
        if err := h.Get(ctx, namespacedName(h.aw.Namespace, h.aw.Name), &wf); err != nil {
925✔
81
                return errLogAndWrap(h.log, err, "failed to get workflow")
×
82
        }
×
83

84
        if updated := h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf); !updated {
928✔
85
                return nil // nothing updated
3✔
86
        }
3✔
87

88
        if err := h.Status().Update(ctx, h.aw); err != nil {
922✔
89
                return errLogAndWrap(h.log, err, "failed to update status")
×
90
        }
×
91

92
        return nil
922✔
93
}
94

95
var _ WorkflowHandler = &CronWorkflowHandler{}
96

97
type CronWorkflowHandler struct {
98
        *ArtifactWorkflowReconciler
99
        log logr.Logger
100
        aw  *arcv1alpha1.ArtifactWorkflow
101
}
102

103
func NewCronWorkflowHandler(r *ArtifactWorkflowReconciler, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow) *CronWorkflowHandler {
16✔
104
        return &CronWorkflowHandler{r, log, aw}
16✔
105
}
16✔
106

107
func (h *CronWorkflowHandler) DeleteArgoResources(ctx context.Context) error {
×
108
        cwf := wfv1alpha1.CronWorkflow{
×
109
                ObjectMeta: metav1.ObjectMeta{
×
110
                        Namespace: h.aw.Namespace,
×
111
                        Name:      h.aw.Name,
×
112
                },
×
113
        }
×
114
        if err := h.Delete(ctx, &cwf); client.IgnoreNotFound(err) != nil {
×
115
                h.Recorder.Event(h.aw, corev1.EventTypeWarning, "DeletionFailed", fmt.Sprintf("Failed to delete associated cron workflow '%s': %v", h.aw.Name, err))
×
116
                return errLogAndWrap(h.log, err, "cron workflow deletion failed")
×
117
        }
×
118
        h.Recorder.Event(h.aw, corev1.EventTypeNormal, "Deleted", fmt.Sprintf("Deleted cron workflow '%s'", h.aw.Name))
×
119
        return nil
×
120
}
121

122
func (h *CronWorkflowHandler) CreateArgoResources(ctx context.Context) error {
1✔
123
        srcSecret, dstSecret, err := h.retrieveSecrets(ctx, h.aw)
1✔
124
        if err != nil {
1✔
125
                return errLogAndWrap(h.log, err, "failed to fetch secrets for artifact workflow")
×
126
        }
×
127

128
        cwf := hydrateArgoCronWorkflow(h.aw, srcSecret, dstSecret)
1✔
129

1✔
130
        if err := controllerutil.SetControllerReference(h.aw, cwf, h.Scheme); err != nil {
1✔
131
                return errLogAndWrap(h.log, err, "failed to set controller reference")
×
132
        }
×
133

134
        if err := h.Create(ctx, cwf); err != nil {
1✔
135
                if client.IgnoreAlreadyExists(err) != nil {
×
136
                        h.Recorder.Event(h.aw, corev1.EventTypeWarning, "CreationFailed", fmt.Sprintf("Failed to create cron workflow '%s': %v", cwf.GetName(), err))
×
137
                        return errLogAndWrap(h.log, err, "failed to create argo cron workflow")
×
138
                }
×
139
        } else {
1✔
140
                h.Recorder.Event(h.aw, corev1.EventTypeNormal, "Created", fmt.Sprintf("Created cron workflow '%s'", cwf.GetName()))
1✔
141
        }
1✔
142

143
        h.aw.Status.Phase = arcv1alpha1.WorkflowPending
1✔
144
        if err := h.Status().Update(ctx, h.aw); err != nil {
1✔
145
                return errLogAndWrap(h.log, err, "failed to update status")
×
146
        }
×
147
        return nil
1✔
148
}
149

150
func (h *CronWorkflowHandler) CheckArgoResources(ctx context.Context) error {
14✔
151
        cwf := wfv1alpha1.CronWorkflow{}
14✔
152
        if err := h.Get(ctx, namespacedName(h.aw.Namespace, h.aw.Name), &cwf); err != nil {
14✔
153
                return errLogAndWrap(h.log, err, "failed to get cron workflow")
×
154
        }
×
155

156
        updated := false
14✔
157

14✔
158
        if !h.aw.Status.LastScheduled.Equal(cwf.Status.LastScheduledTime) {
15✔
159
                h.aw.Status.LastScheduled = cwf.Status.LastScheduledTime
1✔
160
                updated = true
1✔
161
        }
1✔
162
        if h.aw.Status.Failed != cwf.Status.Failed {
15✔
163
                h.aw.Status.Failed = cwf.Status.Failed
1✔
164
                updated = true
1✔
165
        }
1✔
166
        if h.aw.Status.Succeeded != cwf.Status.Succeeded {
15✔
167
                h.aw.Status.Succeeded = cwf.Status.Succeeded
1✔
168
                updated = true
1✔
169
        }
1✔
170

171
        // If the active workflow is not the same as the current one, update the reference
172
        if len(cwf.Status.Active) > 0 {
26✔
173
                // Should only contain a single element at most (expected to be in the same namespace!)
12✔
174
                ref := cwf.Status.Active[len(cwf.Status.Active)-1]
12✔
175

12✔
176
                if h.aw.Status.ActiveWorkflowRef.Name != ref.Name {
21✔
177
                        h.log.V(1).Info("Updating reference for cron workflow", "cronWorkflow", cwf.Name, "activeWorkflow", ref.Name)
9✔
178

9✔
179
                        // Get the active workflow
9✔
180
                        wf := wfv1alpha1.Workflow{}
9✔
181
                        if err := h.Get(ctx, namespacedName(h.aw.Namespace, ref.Name), &wf); err != nil {
9✔
182
                                return errLogAndWrap(h.log, err, "failed to fetch active workflow")
×
183
                        }
×
184

185
                        h.aw.Status.ActiveWorkflowRef = corev1.LocalObjectReference{
9✔
186
                                Name: wf.Name,
9✔
187
                        }
9✔
188
                        h.aw.Status.Message = ""
9✔
189
                        h.aw.Status.Phase = arcv1alpha1.WorkflowActive
9✔
190

9✔
191
                        updated = updated || h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf)
9✔
192
                }
193
        }
194

195
        // If there is an active workflow, check its status
196
        if h.aw.Status.ActiveWorkflowRef.Name != "" {
26✔
197
                wf := wfv1alpha1.Workflow{}
12✔
198
                if err := h.Get(ctx, namespacedName(h.aw.Namespace, h.aw.Status.ActiveWorkflowRef.Name), &wf); err != nil {
12✔
199
                        return errLogAndWrap(h.log, err, "failed to fetch active workflow")
×
200
                }
×
201

202
                updated = updated || h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf)
12✔
203

12✔
204
                if wf.Status.Phase.Completed() {
21✔
205
                        h.aw.Status.ActiveWorkflowRef.Name = ""
9✔
206
                        updated = true
9✔
207
                }
9✔
208
        }
209

210
        if !updated {
17✔
211
                return nil
3✔
212
        }
3✔
213

214
        h.log.V(1).Info("Updating status from active workflow", "cronWorkflow", cwf.Name)
11✔
215

11✔
216
        if err := h.Status().Update(ctx, h.aw); err != nil {
11✔
217
                return errLogAndWrap(h.log, err, "failed to update status")
×
218
        }
×
219

220
        return nil
11✔
221
}
222

223
func hydrateArgoWorkflowSpec(aw *arcv1alpha1.ArtifactWorkflow, srcSecret *corev1.Secret, dstSecret *corev1.Secret) wfv1alpha1.WorkflowSpec {
1,139✔
224
        srcVolume := corev1.Volume{
1,139✔
225
                Name: "src-secret-vol",
1,139✔
226
                VolumeSource: corev1.VolumeSource{
1,139✔
227
                        EmptyDir: &corev1.EmptyDirVolumeSource{},
1,139✔
228
                },
1,139✔
229
        }
1,139✔
230
        if srcSecret.Name != "" {
1,908✔
231
                srcVolume.VolumeSource = corev1.VolumeSource{
769✔
232
                        Secret: &corev1.SecretVolumeSource{
769✔
233
                                SecretName: srcSecret.Name,
769✔
234
                        },
769✔
235
                }
769✔
236
        }
769✔
237

238
        dstVolume := corev1.Volume{
1,139✔
239
                Name: "dst-secret-vol",
1,139✔
240
                VolumeSource: corev1.VolumeSource{
1,139✔
241
                        EmptyDir: &corev1.EmptyDirVolumeSource{},
1,139✔
242
                },
1,139✔
243
        }
1,139✔
244
        if dstSecret.Name != "" {
1,908✔
245
                dstVolume.VolumeSource = corev1.VolumeSource{
769✔
246
                        Secret: &corev1.SecretVolumeSource{
769✔
247
                                SecretName: dstSecret.Name,
769✔
248
                        },
769✔
249
                }
769✔
250
        }
769✔
251

252
        parameters := []wfv1alpha1.Parameter{}
1,139✔
253
        for _, p := range aw.Spec.Parameters {
8,026✔
254
                parameters = append(parameters, wfv1alpha1.Parameter{
6,887✔
255
                        Name:  p.Name,
6,887✔
256
                        Value: (*wfv1alpha1.AnyString)(&p.Value),
6,887✔
257
                })
6,887✔
258
        }
6,887✔
259

260
        return wfv1alpha1.WorkflowSpec{
1,139✔
261
                WorkflowTemplateRef: &wfv1alpha1.WorkflowTemplateRef{
1,139✔
262
                        Name:         aw.Spec.WorkflowTemplateRef.Name,
1,139✔
263
                        ClusterScope: aw.Spec.WorkflowTemplateRef.ClusterScope,
1,139✔
264
                },
1,139✔
265
                Volumes: []corev1.Volume{
1,139✔
266
                        srcVolume,
1,139✔
267
                        dstVolume,
1,139✔
268
                },
1,139✔
269
                Arguments: wfv1alpha1.Arguments{
1,139✔
270
                        Parameters: parameters,
1,139✔
271
                },
1,139✔
272
        }
1,139✔
273
}
274

275
func hydrateArgoWorkflow(aw *arcv1alpha1.ArtifactWorkflow, srcSecret *corev1.Secret, dstSecret *corev1.Secret) *wfv1alpha1.Workflow {
1,138✔
276
        return &wfv1alpha1.Workflow{
1,138✔
277
                ObjectMeta: workflowObjectMeta(aw),
1,138✔
278
                Spec:       hydrateArgoWorkflowSpec(aw, srcSecret, dstSecret),
1,138✔
279
        }
1,138✔
280
}
1,138✔
281

282
func hydrateArgoCronWorkflow(aw *arcv1alpha1.ArtifactWorkflow, srcSecret *corev1.Secret, dstSecret *corev1.Secret) *wfv1alpha1.CronWorkflow {
1✔
283
        om := workflowObjectMeta(aw)
1✔
284
        wf := &wfv1alpha1.CronWorkflow{
1✔
285
                ObjectMeta: om,
1✔
286
                Spec: wfv1alpha1.CronWorkflowSpec{
1✔
287
                        WorkflowSpec:               hydrateArgoWorkflowSpec(aw, srcSecret, dstSecret),
1✔
288
                        Schedules:                  aw.Spec.Cron.Schedules,
1✔
289
                        ConcurrencyPolicy:          wfv1alpha1.ReplaceConcurrent,
1✔
290
                        StartingDeadlineSeconds:    aw.Spec.Cron.StartingDeadlineSeconds,
1✔
291
                        Timezone:                   aw.Spec.Cron.Timezone,
1✔
292
                        When:                       aw.Spec.Cron.When,
1✔
293
                        SuccessfulJobsHistoryLimit: sprint.ToPointer(int32(1)),
1✔
294
                        FailedJobsHistoryLimit:     sprint.ToPointer(int32(1)),
1✔
295
                        WorkflowMetadata:           &om,
1✔
296
                },
1✔
297
        }
1✔
298

1✔
299
        return wf
1✔
300
}
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc