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

opendefensecloud / artifact-conduit / 25523720375

07 May 2026 09:42PM UTC coverage: 84.483% (+0.1%) from 84.375%
25523720375

Pull #346

github

web-flow
Merge 8c7fe6cda into b1f18c553
Pull Request #346: chore(deps): update dependency go to v1.26.3

784 of 928 relevant lines covered (84.48%)

542.03 hits per line

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

83.18
/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/v4/pkg/apis/workflow/v1alpha1"
11
        "github.com/go-logr/logr"
12
        corev1 "k8s.io/api/core/v1"
13
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
14
        "sigs.k8s.io/controller-runtime/pkg/client"
15
        "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
16

17
        arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1"
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 {
1,600✔
35
        return &SingleWorkflowHandler{r, log, aw}
1,600✔
36
}
1,600✔
37

38
func (h *SingleWorkflowHandler) DeleteArgoResources(ctx context.Context) error {
10✔
39
        wf := wfv1alpha1.Workflow{
10✔
40
                ObjectMeta: metav1.ObjectMeta{
10✔
41
                        Namespace: h.aw.Namespace,
10✔
42
                        Name:      h.aw.Name,
10✔
43
                },
10✔
44
        }
10✔
45
        if err := h.Delete(ctx, &wf); client.IgnoreNotFound(err) != nil {
10✔
46
                h.Recorder.Eventf(h.aw, nil, corev1.EventTypeWarning, "DeletionFailed", "Delete", 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.Eventf(h.aw, nil, corev1.EventTypeNormal, "Deleted", "Delete", fmt.Sprintf("Deleted workflow '%s'", h.aw.Name))
10✔
50

10✔
51
        return nil
10✔
52
}
53

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

60
        wf := hydrateArgoWorkflow(h.aw, srcSecret, dstSecret)
893✔
61

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

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

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

77
        return nil
659✔
78
}
79

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

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

90
        if err := h.Status().Update(ctx, h.aw); err != nil {
663✔
91
                return errLogAndWrap(h.log, err, "failed to update status")
4✔
92
        }
4✔
93

94
        return nil
655✔
95
}
96

97
var _ WorkflowHandler = &CronWorkflowHandler{}
98

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

105
func NewCronWorkflowHandler(r *ArtifactWorkflowReconciler, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow) *CronWorkflowHandler {
24✔
106
        return &CronWorkflowHandler{r, log, aw}
24✔
107
}
24✔
108

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

×
122
        return nil
×
123
}
124

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

131
        cwf := hydrateArgoCronWorkflow(h.aw, srcSecret, dstSecret)
2✔
132

2✔
133
        if err := controllerutil.SetControllerReference(h.aw, cwf, h.Scheme); err != nil {
2✔
134
                return errLogAndWrap(h.log, err, "failed to set controller reference")
×
135
        }
×
136

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

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

151
        return nil
2✔
152
}
153

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

160
        updated := false
20✔
161

20✔
162
        if !h.aw.Status.LastScheduled.Equal(cwf.Status.LastScheduledTime) {
23✔
163
                h.aw.Status.LastScheduled = cwf.Status.LastScheduledTime
3✔
164
                updated = true
3✔
165
        }
3✔
166
        if h.aw.Status.Failed != cwf.Status.Failed {
21✔
167
                h.aw.Status.Failed = cwf.Status.Failed
1✔
168
                updated = true
1✔
169
        }
1✔
170
        if h.aw.Status.Succeeded != cwf.Status.Succeeded {
21✔
171
                h.aw.Status.Succeeded = cwf.Status.Succeeded
1✔
172
                updated = true
1✔
173
        }
1✔
174

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

17✔
180
                if h.aw.Status.ActiveWorkflowRef.Name != ref.Name {
28✔
181
                        h.log.V(1).Info("Updating reference for cron workflow", "cronWorkflow", cwf.Name, "activeWorkflow", ref.Name)
11✔
182

11✔
183
                        // Get the active workflow
11✔
184
                        wf := wfv1alpha1.Workflow{}
11✔
185
                        if err := h.Get(ctx, namespacedName(h.aw.Namespace, ref.Name), &wf); err != nil {
12✔
186
                                return errLogAndWrap(h.log, err, "failed to fetch active workflow")
1✔
187
                        }
1✔
188

189
                        h.aw.Status.ActiveWorkflowRef = corev1.LocalObjectReference{
10✔
190
                                Name: wf.Name,
10✔
191
                        }
10✔
192
                        h.aw.Status.Message = ""
10✔
193
                        h.aw.Status.Phase = arcv1alpha1.WorkflowActive
10✔
194

10✔
195
                        updated = updated || h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf)
10✔
196
                }
197
        }
198

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

206
                updated = updated || h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf)
16✔
207

16✔
208
                if wf.Status.Phase.Completed() {
25✔
209
                        h.aw.Status.ActiveWorkflowRef.Name = ""
9✔
210
                        updated = true
9✔
211
                }
9✔
212
        }
213

214
        if !updated {
24✔
215
                return nil
5✔
216
        }
5✔
217

218
        h.log.V(1).Info("Updating status from active workflow", "cronWorkflow", cwf.Name)
14✔
219

14✔
220
        if err := h.Status().Update(ctx, h.aw); err != nil {
15✔
221
                return errLogAndWrap(h.log, err, "failed to update status")
1✔
222
        }
1✔
223

224
        return nil
13✔
225
}
226

227
func hydrateArgoWorkflowSpec(aw *arcv1alpha1.ArtifactWorkflow, srcSecret *corev1.Secret, dstSecret *corev1.Secret) wfv1alpha1.WorkflowSpec {
895✔
228
        srcVolume := corev1.Volume{
895✔
229
                Name: "src-secret-vol",
895✔
230
                VolumeSource: corev1.VolumeSource{
895✔
231
                        EmptyDir: &corev1.EmptyDirVolumeSource{},
895✔
232
                },
895✔
233
        }
895✔
234
        if srcSecret.Name != "" {
1,441✔
235
                srcVolume.VolumeSource = corev1.VolumeSource{
546✔
236
                        Secret: &corev1.SecretVolumeSource{
546✔
237
                                SecretName: srcSecret.Name,
546✔
238
                        },
546✔
239
                }
546✔
240
        }
546✔
241

242
        dstVolume := corev1.Volume{
895✔
243
                Name: "dst-secret-vol",
895✔
244
                VolumeSource: corev1.VolumeSource{
895✔
245
                        EmptyDir: &corev1.EmptyDirVolumeSource{},
895✔
246
                },
895✔
247
        }
895✔
248
        if dstSecret.Name != "" {
1,441✔
249
                dstVolume.VolumeSource = corev1.VolumeSource{
546✔
250
                        Secret: &corev1.SecretVolumeSource{
546✔
251
                                SecretName: dstSecret.Name,
546✔
252
                        },
546✔
253
                }
546✔
254
        }
546✔
255

256
        parameters := []wfv1alpha1.Parameter{}
895✔
257
        for _, p := range aw.Spec.Parameters {
5,608✔
258
                parameters = append(parameters, wfv1alpha1.Parameter{
4,713✔
259
                        Name:  p.Name,
4,713✔
260
                        Value: (*wfv1alpha1.AnyString)(&p.Value),
4,713✔
261
                })
4,713✔
262
        }
4,713✔
263

264
        return wfv1alpha1.WorkflowSpec{
895✔
265
                WorkflowTemplateRef: &wfv1alpha1.WorkflowTemplateRef{
895✔
266
                        Name:         aw.Spec.WorkflowTemplateRef.Name,
895✔
267
                        ClusterScope: aw.Spec.WorkflowTemplateRef.ClusterScope,
895✔
268
                },
895✔
269
                Volumes: []corev1.Volume{
895✔
270
                        srcVolume,
895✔
271
                        dstVolume,
895✔
272
                },
895✔
273
                Arguments: wfv1alpha1.Arguments{
895✔
274
                        Parameters: parameters,
895✔
275
                },
895✔
276
        }
895✔
277
}
278

279
func hydrateArgoWorkflow(aw *arcv1alpha1.ArtifactWorkflow, srcSecret *corev1.Secret, dstSecret *corev1.Secret) *wfv1alpha1.Workflow {
893✔
280
        return &wfv1alpha1.Workflow{
893✔
281
                ObjectMeta: workflowObjectMeta(aw),
893✔
282
                Spec:       hydrateArgoWorkflowSpec(aw, srcSecret, dstSecret),
893✔
283
        }
893✔
284
}
893✔
285

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

2✔
303
        return wf
2✔
304
}
2✔
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