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

opendefensecloud / artifact-conduit / 25494830649

07 May 2026 12:08PM UTC coverage: 84.806% (+0.1%) from 84.698%
25494830649

Pull #334

github

web-flow
Merge 106247195 into 6882bb228
Pull Request #334: fix(deps): update k8s.io/kube-openapi digest to 3334182

787 of 928 relevant lines covered (84.81%)

375.83 hits per line

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

83.64
/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
        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,128✔
35
        return &SingleWorkflowHandler{r, log, aw}
1,128✔
36
}
1,128✔
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 {
647✔
55
        srcSecret, dstSecret, err := h.retrieveSecrets(ctx, h.aw)
647✔
56
        if err != nil {
647✔
57
                return errLogAndWrap(h.log, err, "failed to fetch secrets for artifact workflow")
×
58
        }
×
59

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

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

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

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

77
        return nil
432✔
78
}
79

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

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

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

94
        return nil
429✔
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 {
23✔
106
        return &CronWorkflowHandler{r, log, aw}
23✔
107
}
23✔
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 {
3✔
126
        srcSecret, dstSecret, err := h.retrieveSecrets(ctx, h.aw)
3✔
127
        if err != nil {
3✔
128
                return errLogAndWrap(h.log, err, "failed to fetch secrets for artifact workflow")
×
129
        }
×
130

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

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

137
        if err := h.Create(ctx, cwf); err != nil {
4✔
138
                if client.IgnoreAlreadyExists(err) != nil {
1✔
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
3✔
147
        if err := h.Status().Update(ctx, h.aw); err != nil {
4✔
148
                return errLogAndWrap(h.log, err, "failed to update status")
1✔
149
        }
1✔
150

151
        return nil
2✔
152
}
153

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

160
        updated := false
18✔
161

18✔
162
        if !h.aw.Status.LastScheduled.Equal(cwf.Status.LastScheduledTime) {
20✔
163
                h.aw.Status.LastScheduled = cwf.Status.LastScheduledTime
2✔
164
                updated = true
2✔
165
        }
2✔
166
        if h.aw.Status.Failed != cwf.Status.Failed {
19✔
167
                h.aw.Status.Failed = cwf.Status.Failed
1✔
168
                updated = true
1✔
169
        }
1✔
170
        if h.aw.Status.Succeeded != cwf.Status.Succeeded {
19✔
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 {
33✔
177
                // Should only contain a single element at most (expected to be in the same namespace!)
15✔
178
                ref := cwf.Status.Active[len(cwf.Status.Active)-1]
15✔
179

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

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

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

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

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

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

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

214
        if !updated {
21✔
215
                return nil
4✔
216
        }
4✔
217

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

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

224
        return nil
13✔
225
}
226

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

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

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

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

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

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

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