diff --git a/api/arc/v1alpha1/labels.go b/api/arc/v1alpha1/labels.go new file mode 100644 index 00000000..c30c63dd --- /dev/null +++ b/api/arc/v1alpha1/labels.go @@ -0,0 +1,12 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package v1alpha1 + +const ( + // LabelArtifactType records which ArtifactType an ArtifactWorkflow was + // derived from. The type is only present on the owning Order, so the + // controller stamps it here to make workflows selectable and observable + // by type. + LabelArtifactType = "arc.opendefense.cloud/artifact-type" +) diff --git a/cmd/arc-controller-manager/main.go b/cmd/arc-controller-manager/main.go index a081b2cd..95b591b9 100644 --- a/cmd/arc-controller-manager/main.go +++ b/cmd/arc-controller-manager/main.go @@ -21,11 +21,13 @@ import ( "sigs.k8s.io/controller-runtime/pkg/certwatcher" "sigs.k8s.io/controller-runtime/pkg/healthz" "sigs.k8s.io/controller-runtime/pkg/log/zap" + ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" "sigs.k8s.io/controller-runtime/pkg/metrics/filters" metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" "go.opendefense.cloud/arc/pkg/controller" + arcmetrics "go.opendefense.cloud/arc/pkg/metrics" _ "k8s.io/client-go/plugin/pkg/client/auth" ) @@ -168,6 +170,14 @@ func main() { } } + arcMetrics := arcmetrics.NewCollector(mgr.GetCache()) + ctrlmetrics.Registry.MustRegister(arcMetrics) + + if err := mgr.Add(arcMetrics); err != nil { + setupLog.Error(err, "unable to add metrics leader gate") + os.Exit(1) + } + if err := wfv1alpha1.AddToScheme(mgr.GetScheme()); err != nil { setupLog.Error(err, "failed to add Argo Workflows types to scheme") os.Exit(1) diff --git a/go.mod b/go.mod index e6250aa5..acbab382 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,8 @@ require ( github.com/jastBytes/sprint v0.0.3 github.com/onsi/ginkgo/v2 v2.32.1 github.com/onsi/gomega v1.42.1 + github.com/prometheus/client_golang v1.23.2 + github.com/prometheus/client_model v0.6.2 github.com/robfig/cron/v3 v3.0.1 github.com/spf13/pflag v1.0.10 go.opendefense.cloud/kit v0.3.4 @@ -76,8 +78,6 @@ require ( github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect - github.com/prometheus/client_golang v1.23.2 // indirect - github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.70.1 // indirect github.com/prometheus/procfs v0.21.0 // indirect github.com/spf13/cobra v1.10.2 // indirect diff --git a/pkg/controller/artifactworkflow_controller.go b/pkg/controller/artifactworkflow_controller.go index 165437be..732b2f74 100644 --- a/pkg/controller/artifactworkflow_controller.go +++ b/pkg/controller/artifactworkflow_controller.go @@ -28,6 +28,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/reconcile" arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + "go.opendefense.cloud/arc/pkg/metrics" ) const ( @@ -146,10 +147,11 @@ func (r *ArtifactWorkflowReconciler) Reconcile(ctx context.Context, req ctrl.Req return ctrlResult, nil } -func (r *ArtifactWorkflowReconciler) setStatusFromWorkflow(ctx context.Context, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow, wf *wfv1alpha1.Workflow) bool { +func (r *ArtifactWorkflowReconciler) setStatusFromWorkflow(ctx context.Context, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow, wf *wfv1alpha1.Workflow) (bool, *completion) { if aw.Status.Phase == arcv1alpha1.WorkflowPhase(wf.Status.Phase) { - return false // nothing updated + return false, nil // nothing updated } + aw.Status.Phase = arcv1alpha1.WorkflowPhase(wf.Status.Phase) switch aw.Status.Phase { @@ -162,7 +164,7 @@ func (r *ArtifactWorkflowReconciler) setStatusFromWorkflow(ctx context.Context, default: } - return true + return true, newCompletion(aw, wf) } func (r *ArtifactWorkflowReconciler) generateWorkflowStatusMessage(ctx context.Context, wf *wfv1alpha1.Workflow, log logr.Logger, aw *arcv1alpha1.ArtifactWorkflow) { @@ -224,7 +226,9 @@ func (r *ArtifactWorkflowReconciler) retrieveSecrets(ctx context.Context, aw *ar 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.Eventf(aw, nil, corev1.EventTypeWarning, "InvalidSecret", "FetchSecret", fmt.Sprintf("Failed to fetch source secret '%s': %v", aw.Spec.SrcSecretRef.Name, err)) + r.Recorder.Eventf(aw, nil, corev1.EventTypeWarning, ReasonInvalidSecret, "FetchSecret", fmt.Sprintf("Failed to fetch source secret '%s': %v", aw.Spec.SrcSecretRef.Name, err)) + metrics.RecordReconcileError(ControllerArtifactWorkflow, ReasonInvalidSecret) + return nil, nil, fmt.Errorf("failed to fetch secret for source: %w", err) } } @@ -232,7 +236,9 @@ func (r *ArtifactWorkflowReconciler) retrieveSecrets(ctx context.Context, aw *ar 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.Eventf(aw, nil, corev1.EventTypeWarning, "InvalidSecret", "FetchSecret", fmt.Sprintf("Failed to fetch destination secret '%s': %v", aw.Spec.DstSecretRef.Name, err)) + r.Recorder.Eventf(aw, nil, corev1.EventTypeWarning, ReasonInvalidSecret, "FetchSecret", fmt.Sprintf("Failed to fetch destination secret '%s': %v", aw.Spec.DstSecretRef.Name, err)) + metrics.RecordReconcileError(ControllerArtifactWorkflow, ReasonInvalidSecret) + return nil, nil, fmt.Errorf("failed to fetch secret for destination: %w", err) } } diff --git a/pkg/controller/artifactworkflow_controller_test.go b/pkg/controller/artifactworkflow_controller_test.go index fa5bcf10..cc09382f 100644 --- a/pkg/controller/artifactworkflow_controller_test.go +++ b/pkg/controller/artifactworkflow_controller_test.go @@ -7,6 +7,7 @@ import ( "fmt" wfv1alpha1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" + "github.com/prometheus/client_golang/prometheus/testutil" "go.opendefense.cloud/kit/envtest" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -14,6 +15,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + "go.opendefense.cloud/arc/pkg/metrics" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -122,6 +124,32 @@ var _ = Describe("ArtifactWorkflowController", func() { })) }) + It("should count a missing secret under the reason its Event carries", func() { + counter := metrics.ReconcileErrorsCounterForTest(ControllerArtifactWorkflow, ReasonInvalidSecret) + before := testutil.ToFloat64(counter) + + aw := &arcv1alpha1.ArtifactWorkflow{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: ns.Name, + Name: "missing-secret", + }, + Spec: arcv1alpha1.ArtifactWorkflowSpec{ + WorkflowTemplateRef: at.Spec.WorkflowTemplateRef, + SrcSecretRef: corev1.LocalObjectReference{Name: "does-not-exist"}, + }, + } + Expect(k8sClient.Create(ctx, aw)).To(Succeed()) + + Eventually(func() float64 { + return testutil.ToFloat64(counter) - before + }).Should(BeNumerically(">=", 1.0)) + + // The workflow must not be created from secrets that could not be read. + Consistently(func() error { + return k8sClient.Get(ctx, namespacedName(ns.Name, aw.Name), &wfv1alpha1.Workflow{}) + }).ShouldNot(Succeed()) + }) + It("should track Workflow status changes of created ArtifactWorkflows", func() { awName := "track-status" aw := &arcv1alpha1.ArtifactWorkflow{ @@ -167,6 +195,40 @@ var _ = Describe("ArtifactWorkflowController", func() { }).Should(Equal(int64(1))) }) + It("should count a completion once the workflow succeeds", func() { + counter := metrics.CompletionsCounterForTest(ns.Name, metrics.UnknownArtifactType, metrics.ResultSucceeded) + before := testutil.ToFloat64(counter) + + awName := "count-completion" + aw := &arcv1alpha1.ArtifactWorkflow{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: ns.Name, + Name: awName, + }, + Spec: arcv1alpha1.ArtifactWorkflowSpec{ + WorkflowTemplateRef: at.Spec.WorkflowTemplateRef, + Parameters: []arcv1alpha1.ArtifactWorkflowParameter{ + {Name: awName, Value: awName}, + }, + }, + } + Expect(k8sClient.Create(ctx, aw)).To(Succeed()) + + wf := &wfv1alpha1.Workflow{} + Eventually(func() error { + return k8sClient.Get(ctx, namespacedName(aw.Namespace, aw.Name), wf) + }).Should(Succeed()) + + wf.Status.Phase = wfv1alpha1.WorkflowSucceeded + Expect(k8sClient.Update(ctx, wf)).To(Succeed()) + + delta := func() float64 { + return testutil.ToFloat64(counter) - before + } + Eventually(delta).Should(Equal(1.0)) + Consistently(delta).Should(Equal(1.0)) + }) + It("should track failed Workflow information of created ArtifactWorkflows", func() { awName := "track-failed-status" aw := &arcv1alpha1.ArtifactWorkflow{ @@ -504,3 +566,94 @@ var _ = Describe("ArtifactWorkflowController", func() { }) }) }) + +var _ = Describe("newCompletion", func() { + It("should return nil for non terminal phases", func() { + aw := &arcv1alpha1.ArtifactWorkflow{} + aw.Status.Phase = arcv1alpha1.WorkflowRunning + + Expect(newCompletion(aw, &wfv1alpha1.Workflow{})).To(BeNil()) + }) + + It("should return nil for Stopped, which is an action not a result", func() { + aw := &arcv1alpha1.ArtifactWorkflow{} + aw.Status.Phase = arcv1alpha1.WorkflowStopped + + Expect(newCompletion(aw, &wfv1alpha1.Workflow{})).To(BeNil()) + }) + + It("should take the duration from the argo workflow", func() { + aw := &arcv1alpha1.ArtifactWorkflow{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Labels: map[string]string{arcv1alpha1.LabelArtifactType: "oci"}, + }, + } + aw.Status.Phase = arcv1alpha1.WorkflowSucceeded + + wf := &wfv1alpha1.Workflow{} + wf.Status.StartedAt = metav1.NewTime(metav1.Unix(1700000000, 0).Time) + wf.Status.FinishedAt = metav1.NewTime(metav1.Unix(1700000090, 0).Time) + + completion := newCompletion(aw, wf) + + Expect(completion).NotTo(BeNil()) + Expect(completion.namespace).To(Equal("team-a")) + Expect(completion.artifactType).To(Equal("oci")) + Expect(completion.result).To(Equal(metrics.ResultSucceeded)) + Expect(completion.hasDuration).To(BeTrue()) + Expect(completion.seconds).To(Equal(90.0)) + }) + + It("should record a zero length duration when start and finish share the same second", func() { + aw := &arcv1alpha1.ArtifactWorkflow{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Labels: map[string]string{arcv1alpha1.LabelArtifactType: "oci"}, + }, + } + aw.Status.Phase = arcv1alpha1.WorkflowSucceeded + + same := metav1.NewTime(metav1.Unix(1700000000, 0).Time) + wf := &wfv1alpha1.Workflow{} + wf.Status.StartedAt = same + wf.Status.FinishedAt = same + + completion := newCompletion(aw, wf) + + Expect(completion).NotTo(BeNil()) + Expect(completion.hasDuration).To(BeTrue()) + Expect(completion.seconds).To(Equal(0.0)) + }) + + It("should not record a duration when finish precedes start", func() { + aw := &arcv1alpha1.ArtifactWorkflow{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Labels: map[string]string{arcv1alpha1.LabelArtifactType: "oci"}, + }, + } + aw.Status.Phase = arcv1alpha1.WorkflowSucceeded + + wf := &wfv1alpha1.Workflow{} + wf.Status.StartedAt = metav1.NewTime(metav1.Unix(1700000090, 0).Time) + wf.Status.FinishedAt = metav1.NewTime(metav1.Unix(1700000000, 0).Time) + + completion := newCompletion(aw, wf) + + Expect(completion).NotTo(BeNil()) + Expect(completion.hasDuration).To(BeFalse()) + }) + + It("should record a failure without a duration when argo has no timestamps", func() { + aw := &arcv1alpha1.ArtifactWorkflow{ObjectMeta: metav1.ObjectMeta{Namespace: "team-a"}} + aw.Status.Phase = arcv1alpha1.WorkflowFailed + + completion := newCompletion(aw, &wfv1alpha1.Workflow{}) + + Expect(completion).NotTo(BeNil()) + Expect(completion.result).To(Equal(metrics.ResultFailed)) + Expect(completion.artifactType).To(Equal(metrics.UnknownArtifactType)) + Expect(completion.hasDuration).To(BeFalse()) + }) +}) diff --git a/pkg/controller/const.go b/pkg/controller/const.go index 06946e05..fb1aaacb 100644 --- a/pkg/controller/const.go +++ b/pkg/controller/const.go @@ -7,3 +7,27 @@ const ( AnnotationRequestedAt = "arc.opendefense.cloud/requested-at" AnnotationForceAt = "arc.opendefense.cloud/force-at" ) + +// Event reasons. These double as the reason label on arc_reconcile_errors_total, +// so the metric and the Kubernetes Event for the same failure always agree. +const ( + ReasonInvalid = "Invalid" + ReasonInvalidEndpoint = "InvalidEndpoint" + ReasonInvalidArtifactType = "InvalidArtifactType" + ReasonInvalidSecret = "InvalidSecret" + ReasonComputationFailed = "ComputationFailed" + ReasonHydrationFailed = "HydrationFailed" + ReasonCreationFailed = "CreationFailed" + ReasonDeletionFailed = "DeletionFailed" +) + +// Controller names used as the controller label on arc_reconcile_errors_total. +const ( + ControllerOrder = "order" + ControllerArtifactWorkflow = "artifactworkflow" +) + +// ReasonDeleting is the Event reason for the informational warning emitted while +// an order's deletion is in progress. It is not a failure, so it is not counted +// on arc_reconcile_errors_total. +const ReasonDeleting = "Deleting" diff --git a/pkg/controller/helpers.go b/pkg/controller/helpers.go index 8f1c9594..a21a3a16 100644 --- a/pkg/controller/helpers.go +++ b/pkg/controller/helpers.go @@ -6,6 +6,7 @@ package controller import ( "encoding/json" "fmt" + "maps" "strconv" "strings" "time" @@ -13,6 +14,7 @@ import ( wfv1alpha1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/validation" arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" ) @@ -37,12 +39,29 @@ func cloneObjectMeta(meta metav1.ObjectMeta, name string) metav1.ObjectMeta { return metav1.ObjectMeta{ Namespace: meta.Namespace, Name: name, - Labels: meta.Labels, + Labels: maps.Clone(meta.Labels), } } -func awObjectMeta(order *arcv1alpha1.Order, sha string) metav1.ObjectMeta { - return cloneObjectMeta(order.ObjectMeta, awName(order, sha)) +func awObjectMeta(order *arcv1alpha1.Order, sha, artifactType string) metav1.ObjectMeta { + meta := cloneObjectMeta(order.ObjectMeta, awName(order, sha)) + + // Nothing validates OrderArtifact.Type against the 63 character label value + // limit, and an invalid value would make every create of this + // ArtifactWorkflow fail permanently. Leaving the label off keeps the + // workflow creatable and reports it as artifact_type="unknown". Truncating + // would report a wrong type instead of a missing one. + if len(validation.IsValidLabelValue(artifactType)) > 0 { + delete(meta.Labels, arcv1alpha1.LabelArtifactType) + return meta + } + + if meta.Labels == nil { + meta.Labels = map[string]string{} + } + meta.Labels[arcv1alpha1.LabelArtifactType] = artifactType + + return meta } func workflowObjectMeta(aw *arcv1alpha1.ArtifactWorkflow) metav1.ObjectMeta { diff --git a/pkg/controller/helpers_test.go b/pkg/controller/helpers_test.go index aeb4654d..e109ed2e 100644 --- a/pkg/controller/helpers_test.go +++ b/pkg/controller/helpers_test.go @@ -4,6 +4,7 @@ package controller import ( + "strings" "time" wfv1alpha1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" @@ -112,10 +113,88 @@ var _ = Describe("Helper Functions", func() { }, } - result := awObjectMeta(order, "sha123") + result := awObjectMeta(order, "sha123", "oci") Expect(result.Namespace).To(Equal("test-ns")) Expect(result.Name).To(Equal("test-order-sha123")) Expect(result.Labels).To(HaveKeyWithValue("app", "test")) + Expect(result.Labels).To(HaveKeyWithValue(arcv1alpha1.LabelArtifactType, "oci")) + }) + + It("should stamp the artifact type label", func() { + order := &arcv1alpha1.Order{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Name: "nightly", + Labels: map[string]string{"owner": "platform"}, + }, + } + + meta := awObjectMeta(order, "abc123", "oci") + + Expect(meta.Namespace).To(Equal("team-a")) + Expect(meta.Name).To(Equal("nightly-abc123")) + Expect(meta.Labels).To(HaveKeyWithValue("owner", "platform")) + Expect(meta.Labels).To(HaveKeyWithValue(arcv1alpha1.LabelArtifactType, "oci")) + }) + + It("should not mutate the order's own labels", func() { + order := &arcv1alpha1.Order{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Name: "nightly", + Labels: map[string]string{"owner": "platform"}, + }, + } + + awObjectMeta(order, "abc123", "oci") + + Expect(order.Labels).To(HaveLen(1)) + Expect(order.Labels).NotTo(HaveKey(arcv1alpha1.LabelArtifactType)) + }) + + It("should skip the artifact type label when the value is not a valid label", func() { + order := &arcv1alpha1.Order{ + ObjectMeta: metav1.ObjectMeta{Namespace: "team-a", Name: "nightly"}, + } + + // 64 characters, one over the label value limit. The workflow still + // has to be creatable, so the label is left off rather than + // truncated to a type that does not exist. + meta := awObjectMeta(order, "abc123", strings.Repeat("a", 64)) + + Expect(meta.Name).To(Equal("nightly-abc123")) + Expect(meta.Labels).NotTo(HaveKey(arcv1alpha1.LabelArtifactType)) + }) + + It("should work when the order has no labels", func() { + order := &arcv1alpha1.Order{ + ObjectMeta: metav1.ObjectMeta{Namespace: "team-a", Name: "nightly"}, + } + + meta := awObjectMeta(order, "abc123", "helm") + + Expect(meta.Labels).To(HaveKeyWithValue(arcv1alpha1.LabelArtifactType, "helm")) + }) + + It("should drop inherited artifact type label when artifact type is invalid", func() { + longType := strings.Repeat("x", 64) + order := &arcv1alpha1.Order{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Name: "nightly", + Labels: map[string]string{ + "owner": "platform", + arcv1alpha1.LabelArtifactType: "inherited-type", + }, + }, + } + + meta := awObjectMeta(order, "abc123", longType) + + Expect(meta.Namespace).To(Equal("team-a")) + Expect(meta.Name).To(Equal("nightly-abc123")) + Expect(meta.Labels).To(HaveKeyWithValue("owner", "platform")) + Expect(meta.Labels).NotTo(HaveKey(arcv1alpha1.LabelArtifactType)) }) }) diff --git a/pkg/controller/metrics.go b/pkg/controller/metrics.go new file mode 100644 index 00000000..550f0337 --- /dev/null +++ b/pkg/controller/metrics.go @@ -0,0 +1,65 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package controller + +import ( + wfv1alpha1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" + + arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + "go.opendefense.cloud/arc/pkg/metrics" +) + +// completion carries what a finished ArtifactWorkflow contributes to the +// metrics, held between detecting the transition and the status write that +// commits it. Recording before that write would count twice whenever the write +// loses a conflict and the object is reconciled again from a stale cache. +type completion struct { + namespace string + artifactType string + result string + seconds float64 + hasDuration bool +} + +// newCompletion returns the metrics contribution of a workflow that has just +// reached a terminal phase, or nil if the phase is not a terminal result. +func newCompletion(aw *arcv1alpha1.ArtifactWorkflow, wf *wfv1alpha1.Workflow) *completion { + result, ok := metrics.ResultFor(aw.Status.Phase) + if !ok { + return nil + } + + c := &completion{ + namespace: aw.Namespace, + artifactType: metrics.ArtifactTypeOf(aw), + result: result, + } + + // Argo's own start and finish times are the honest measurement. ARC's + // status timestamps are observation times taken on the controller clock at + // reconcile, against a creation timestamp from the API server clock. + // + // elapsed can legitimately be zero: metav1.Time serialises at second + // granularity, so a workflow that starts and finishes inside the same + // second measures as zero and that is still a truthful observation. Only + // a negative elapsed, from clock skew or a FinishedAt that precedes + // StartedAt, is rejected. + if !wf.Status.StartedAt.IsZero() && !wf.Status.FinishedAt.IsZero() { + if elapsed := wf.Status.FinishedAt.Sub(wf.Status.StartedAt.Time); elapsed >= 0 { + c.seconds = elapsed.Seconds() + c.hasDuration = true + } + } + + return c +} + +// record publishes the completion. Call it only after the status write succeeded. +func (c *completion) record() { + metrics.RecordCompletion(c.namespace, c.artifactType, c.result) + + if c.hasDuration { + metrics.ObserveDuration(c.artifactType, c.result, c.seconds) + } +} diff --git a/pkg/controller/order_controller.go b/pkg/controller/order_controller.go index 8d4e5677..fc948d55 100644 --- a/pkg/controller/order_controller.go +++ b/pkg/controller/order_controller.go @@ -24,6 +24,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + "go.opendefense.cloud/arc/pkg/metrics" ) const ( @@ -82,14 +83,15 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl // Handle deletion: cleanup artifact workflows, then remove finalizer if !order.DeletionTimestamp.IsZero() { log.V(1).Info("Order is being deleted") - r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, "Deleting", "Delete", "Order is being deleted, cleaning up artifact workflows") + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonDeleting, "Delete", "Order is being deleted, cleaning up artifact workflows") // Cleanup all artifact workflows if len(order.Status.ArtifactWorkflows) > 0 { for sha := range order.Status.ArtifactWorkflows { - // Remove ArtifactWorkflow + // Remove ArtifactWorkflow. The artifact type label is not needed here since + // Delete matches targets by namespace and name, not labels. aw := &arcv1alpha1.ArtifactWorkflow{ - ObjectMeta: awObjectMeta(order, sha), + ObjectMeta: awObjectMeta(order, sha, ""), } _ = r.Delete(ctx, aw) // Ignore errors delete(order.Status.ArtifactWorkflows, sha) @@ -141,7 +143,7 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl for sha := range order.Status.ArtifactWorkflows { // Remove Secret and ArtifactWorkflow aw := &arcv1alpha1.ArtifactWorkflow{ - ObjectMeta: awObjectMeta(order, sha), + ObjectMeta: awObjectMeta(order, sha, ""), } _ = r.Delete(ctx, aw) // Ignore errors delete(order.Status.ArtifactWorkflows, sha) @@ -167,7 +169,10 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl for i, artifact := range order.Spec.Artifacts { daw, err := r.computeDesiredAW(ctx, log, order, &artifact, i) if err != nil { - r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, "ComputationFailed", "Compute", "Failed to compute desired artifact workflow for artifact index %d: %v", i, err) + // computeDesiredAW counts its own failures under the reason that + // describes them, so counting again here would report every one of + // them twice and shadow the specific reason with a generic sibling. + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonComputationFailed, "Compute", "Failed to compute desired artifact workflow for artifact index %d: %v", i, err) order.Status.Message = fmt.Sprintf("Failed to compute desired artifact workflow for artifact index %d: %v", i, err) if err := r.Status().Update(ctx, order); err != nil { return ctrlResult, errLogAndWrap(log, err, "failed to update status") @@ -221,7 +226,9 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl // Get ArtifactWorkflow object and check TTLs. artifactWorkflow := &arcv1alpha1.ArtifactWorkflow{} if err := r.Get(ctx, types.NamespacedName{Namespace: order.Namespace, Name: awName(order, sha)}, artifactWorkflow); err != nil && !apierrors.IsNotFound(err) { - r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, "Invalid", "Fetch", "Failed to fetch ArtifactWorkflow: %v", sha) + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonInvalid, "Fetch", "Failed to fetch ArtifactWorkflow: %v", sha) + metrics.RecordReconcileError(ControllerOrder, ReasonInvalid) + return ctrlResult, errLogAndWrap(log, err, "") } if artifactWorkflow.Name != "" { @@ -271,13 +278,17 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl daw := desiredAWs[sha] aw, err := r.hydrateArtifactWorkflow(&daw) if err != nil { - r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, "HydrationFailed", "Hydrate", "Failed to hydrate artifact workflow for artifact index %d: %v", daw.index, err) + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonHydrationFailed, "Hydrate", "Failed to hydrate artifact workflow for artifact index %d: %v", daw.index, err) + metrics.RecordReconcileError(ControllerOrder, ReasonHydrationFailed) + return ctrlResult, errLogAndWrap(log, err, "failed to hydrate artifact workflow") } // Set owner references if err := controllerutil.SetControllerReference(order, aw, r.Scheme); err != nil { - r.Recorder.Eventf(order, aw, corev1.EventTypeWarning, "HydrationFailed", "Hydrate", "Failed to set controller reference for artifact workflow: %v", err) + r.Recorder.Eventf(order, aw, corev1.EventTypeWarning, ReasonHydrationFailed, "Hydrate", "Failed to set controller reference for artifact workflow: %v", err) + metrics.RecordReconcileError(ControllerOrder, ReasonHydrationFailed) + return ctrlResult, errLogAndWrap(log, err, "failed to set controller reference") } @@ -287,7 +298,8 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl // Already created by a previous reconcile — that's fine continue } - r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, "CreationFailed", "Create", "Failed to create artifact workflow for artifact index %d: %v", daw.index, err) + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonCreationFailed, "Create", "Failed to create artifact workflow for artifact index %d: %v", daw.index, err) + metrics.RecordReconcileError(ControllerOrder, ReasonCreationFailed) return ctrlResult, errLogAndWrap(log, err, "failed to create artifact workflow") } else { @@ -308,10 +320,12 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl for _, sha := range deleteAWs { // Does not exist anymore, let's clean up! aw := &arcv1alpha1.ArtifactWorkflow{ - ObjectMeta: awObjectMeta(order, sha), + ObjectMeta: awObjectMeta(order, sha, ""), } if err := r.Delete(ctx, aw); client.IgnoreNotFound(err) != nil { - r.Recorder.Eventf(order, aw, corev1.EventTypeWarning, "DeletionFailed", "Delete", "Failed to delete obsolete artifact workflow '%s': %v", sha, err) + r.Recorder.Eventf(order, aw, corev1.EventTypeWarning, ReasonDeletionFailed, "Delete", "Failed to delete obsolete artifact workflow '%s': %v", sha, err) + metrics.RecordReconcileError(ControllerOrder, ReasonDeletionFailed) + return ctrlResult, errLogAndWrap(log, err, "failed to delete artifact workflow") } @@ -325,10 +339,12 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl for _, sha := range finishedAWs { // Finished, let's clean up! aw := &arcv1alpha1.ArtifactWorkflow{ - ObjectMeta: awObjectMeta(order, sha), + ObjectMeta: awObjectMeta(order, sha, ""), } if err := r.Delete(ctx, aw); client.IgnoreNotFound(err) != nil { - r.Recorder.Eventf(order, aw, corev1.EventTypeWarning, "DeletionFailed", "Delete", "Failed to delete finished artifact workflow '%s': %v", sha, err) + r.Recorder.Eventf(order, aw, corev1.EventTypeWarning, ReasonDeletionFailed, "Delete", "Failed to delete finished artifact workflow '%s': %v", sha, err) + metrics.RecordReconcileError(ControllerOrder, ReasonDeletionFailed) + return ctrlResult, errLogAndWrap(log, err, "failed to delete artifact workflow") } @@ -424,25 +440,31 @@ func (r *OrderReconciler) computeDesiredAW(ctx context.Context, log logr.Logger, srcEndpoint := &arcv1alpha1.Endpoint{} if err := r.Get(ctx, namespacedName(order.Namespace, srcRefName), srcEndpoint); err != nil { - r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, "InvalidEndpoint", "FetchEndpoint", "Failed to fetch source endpoint '%s': %v", srcRefName, err) + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonInvalidEndpoint, "FetchEndpoint", "Failed to fetch source endpoint '%s': %v", srcRefName, err) + metrics.RecordReconcileError(ControllerOrder, ReasonInvalidEndpoint) + return nil, errLogAndWrap(log, err, "failed to fetch endpoint for source") } dstEndpoint := &arcv1alpha1.Endpoint{} if err := r.Get(ctx, namespacedName(order.Namespace, dstRefName), dstEndpoint); err != nil { - r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, "InvalidEndpoint", "FetchEndpoint", "Failed to fetch destination endpoint '%s': %v", dstRefName, err) + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonInvalidEndpoint, "FetchEndpoint", "Failed to fetch destination endpoint '%s': %v", dstRefName, err) + metrics.RecordReconcileError(ControllerOrder, ReasonInvalidEndpoint) + return nil, errLogAndWrap(log, err, "failed to fetch endpoint for destination") } // Validate that the endpoint usage is correct if srcEndpoint.Spec.Usage != arcv1alpha1.EndpointUsagePullOnly && srcEndpoint.Spec.Usage != arcv1alpha1.EndpointUsageAll { err := fmt.Errorf("endpoint '%s' usage '%s' is not compatible with source usage", srcEndpoint.Name, srcEndpoint.Spec.Usage) - r.Recorder.Eventf(order, srcEndpoint, corev1.EventTypeWarning, "InvalidEndpoint", "ValidateEndpoint", "Source endpoint '%s' has incompatible usage '%s'", srcEndpoint.Name, srcEndpoint.Spec.Usage) + r.Recorder.Eventf(order, srcEndpoint, corev1.EventTypeWarning, ReasonInvalidEndpoint, "ValidateEndpoint", "Source endpoint '%s' has incompatible usage '%s'", srcEndpoint.Name, srcEndpoint.Spec.Usage) + metrics.RecordReconcileError(ControllerOrder, ReasonInvalidEndpoint) return nil, errLogAndWrap(log, err, "artifact validation failed") } if dstEndpoint.Spec.Usage != arcv1alpha1.EndpointUsagePushOnly && dstEndpoint.Spec.Usage != arcv1alpha1.EndpointUsageAll { err := fmt.Errorf("endpoint '%s' usage '%s' is not compatible with destination usage", dstEndpoint.Name, dstEndpoint.Spec.Usage) - r.Recorder.Eventf(order, dstEndpoint, corev1.EventTypeWarning, "InvalidEndpoint", "ValidateEndpoint", "Destination endpoint '%s' has incompatible usage '%s'", dstEndpoint.Name, dstEndpoint.Spec.Usage) + r.Recorder.Eventf(order, dstEndpoint, corev1.EventTypeWarning, ReasonInvalidEndpoint, "ValidateEndpoint", "Destination endpoint '%s' has incompatible usage '%s'", dstEndpoint.Name, dstEndpoint.Spec.Usage) + metrics.RecordReconcileError(ControllerOrder, ReasonInvalidEndpoint) return nil, errLogAndWrap(log, err, "artifact validation failed") } @@ -450,7 +472,9 @@ func (r *OrderReconciler) computeDesiredAW(ctx context.Context, log logr.Logger, // Validate against ArtifactType rules artifactType := &arcv1alpha1.ArtifactType{} if err := r.Get(ctx, namespacedName(order.Namespace, artifact.Type), artifactType); client.IgnoreNotFound(err) != nil { - r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, "InvalidArtifactType", "FetchArtifactType", "Failed to fetch ArtifactType '%s': %v", artifact.Type, err) + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonInvalidArtifactType, "FetchArtifactType", "Failed to fetch ArtifactType '%s': %v", artifact.Type, err) + metrics.RecordReconcileError(ControllerOrder, ReasonInvalidArtifactType) + return nil, errLogAndWrap(log, err, "failed to fetch referenced ArtifactType") } var ( @@ -460,6 +484,11 @@ func (r *OrderReconciler) computeDesiredAW(ctx context.Context, log logr.Logger, if artifactType.Name == "" { // was not found, let's check ClusterArtifactType clusterArtifactType := &arcv1alpha1.ClusterArtifactType{} if err := r.Get(ctx, namespacedName("", artifact.Type), clusterArtifactType); err != nil { + // No Event of its own is emitted here. The only Event this failure + // produces is the caller's ComputationFailed, so that is the reason + // the metric has to carry for the two to agree. + metrics.RecordReconcileError(ControllerOrder, ReasonComputationFailed) + return nil, errLogAndWrap(log, err, "failed to fetch ArtifactType or ClusterArtifactType") } artifactTypeSpec = &clusterArtifactType.Spec @@ -473,13 +502,15 @@ func (r *OrderReconciler) computeDesiredAW(ctx context.Context, log logr.Logger, if len(artifactTypeSpec.Rules.SrcTypes) > 0 && !slices.Contains(artifactTypeSpec.Rules.SrcTypes, srcEndpoint.Spec.Type) { err := fmt.Errorf("source endpoint type '%s' is not allowed by ArtifactType rules", srcEndpoint.Spec.Type) - r.Recorder.Eventf(order, artifactType, corev1.EventTypeWarning, "InvalidArtifactType", "ValidateArtifactType", "Source endpoint type '%s' is not allowed by ArtifactType '%s' rules", srcEndpoint.Spec.Type, artifact.Type) + r.Recorder.Eventf(order, artifactType, corev1.EventTypeWarning, ReasonInvalidArtifactType, "ValidateArtifactType", "Source endpoint type '%s' is not allowed by ArtifactType '%s' rules", srcEndpoint.Spec.Type, artifact.Type) + metrics.RecordReconcileError(ControllerOrder, ReasonInvalidArtifactType) return nil, errLogAndWrap(log, err, "artifact validation failed") } if len(artifactTypeSpec.Rules.DstTypes) > 0 && !slices.Contains(artifactTypeSpec.Rules.DstTypes, dstEndpoint.Spec.Type) { err := fmt.Errorf("destination endpoint type '%s' is not allowed by ArtifactType rules", dstEndpoint.Spec.Type) - r.Recorder.Eventf(order, artifactType, corev1.EventTypeWarning, "InvalidArtifactType", "ValidateArtifactType", "Destination endpoint type '%s' is not allowed by ArtifactType '%s' rules", dstEndpoint.Spec.Type, artifact.Type) + r.Recorder.Eventf(order, artifactType, corev1.EventTypeWarning, ReasonInvalidArtifactType, "ValidateArtifactType", "Destination endpoint type '%s' is not allowed by ArtifactType '%s' rules", dstEndpoint.Spec.Type, artifact.Type) + metrics.RecordReconcileError(ControllerOrder, ReasonInvalidArtifactType) return nil, errLogAndWrap(log, err, "artifact validation failed") } @@ -488,7 +519,9 @@ func (r *OrderReconciler) computeDesiredAW(ctx context.Context, log logr.Logger, srcSecret := &corev1.Secret{} if srcEndpoint.Spec.SecretRef.Name != "" { if err := r.Get(ctx, namespacedName(order.Namespace, srcEndpoint.Spec.SecretRef.Name), srcSecret); err != nil { - r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, "InvalidSecret", "FetchSecret", "Failed to fetch source secret '%s': %v", srcEndpoint.Spec.SecretRef.Name, err) + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonInvalidSecret, "FetchSecret", "Failed to fetch source secret '%s': %v", srcEndpoint.Spec.SecretRef.Name, err) + metrics.RecordReconcileError(ControllerOrder, ReasonInvalidSecret) + return nil, errLogAndWrap(log, err, "failed to fetch secret for source") } } @@ -496,7 +529,9 @@ func (r *OrderReconciler) computeDesiredAW(ctx context.Context, log logr.Logger, dstSecret := &corev1.Secret{} if dstEndpoint.Spec.SecretRef.Name != "" { if err := r.Get(ctx, namespacedName(order.Namespace, dstEndpoint.Spec.SecretRef.Name), dstSecret); err != nil { - r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, "InvalidSecret", "FetchSecret", "Failed to fetch destination secret '%s': %v", dstEndpoint.Spec.SecretRef.Name, err) + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonInvalidSecret, "FetchSecret", "Failed to fetch destination secret '%s': %v", dstEndpoint.Spec.SecretRef.Name, err) + metrics.RecordReconcileError(ControllerOrder, ReasonInvalidSecret) + return nil, errLogAndWrap(log, err, "failed to fetch secret for destination") } } @@ -521,6 +556,10 @@ func (r *OrderReconciler) computeDesiredAW(ctx context.Context, log logr.Logger, } if err := json.NewEncoder(h).Encode(data); err != nil { + // Same as the ClusterArtifactType fetch above: the caller's + // ComputationFailed is the only Event this failure produces. + metrics.RecordReconcileError(ControllerOrder, ReasonComputationFailed) + return nil, errLogAndWrap(log, err, "failed to marshal artifact workflow data") } @@ -530,7 +569,7 @@ func (r *OrderReconciler) computeDesiredAW(ctx context.Context, log logr.Logger, // Let's store it to compare it to the current status! return &desiredAW{ index: i, - objectMeta: awObjectMeta(order, sha), + objectMeta: awObjectMeta(order, sha, artifact.Type), artifact: artifact, typeSpec: artifactTypeSpec, srcEndpoint: srcEndpoint, diff --git a/pkg/controller/order_controller_test.go b/pkg/controller/order_controller_test.go index 4854578d..c3f8780e 100644 --- a/pkg/controller/order_controller_test.go +++ b/pkg/controller/order_controller_test.go @@ -10,6 +10,7 @@ import ( "time" wfv1alpha1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" + "github.com/prometheus/client_golang/prometheus/testutil" "go.opendefense.cloud/kit/envtest" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -17,6 +18,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + "go.opendefense.cloud/arc/pkg/metrics" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -911,6 +913,14 @@ var _ = Describe("OrderController", func() { }) It("should fail when artifact type does not exist", func() { + // A missing ArtifactType/ClusterArtifactType is the one failure in + // computeDesiredAW with no Event of its own, so ComputationFailed, + // the reason its only Event carries, is what it counts under. Every + // other failure there counts under its own reason and is not + // counted a second time by the caller. + counter := metrics.ReconcileErrorsCounterForTest(ControllerOrder, ReasonComputationFailed) + before := testutil.ToFloat64(counter) + createEndpoints("src-nonexistent", "dst-nonexistent") // Create order referencing a non-existent artifact type @@ -937,6 +947,11 @@ var _ = Describe("OrderController", func() { return order.Status.Message }).Should(ContainSubstring("failed to fetch ArtifactType or ClusterArtifactType")) + // Verify the reconcile error was counted under the matching reason + Eventually(func() float64 { + return testutil.ToFloat64(counter) - before + }).Should(BeNumerically(">=", 1.0)) + // Verify no artifact workflows were created Consistently(func() int { awList := &arcv1alpha1.ArtifactWorkflowList{} @@ -947,6 +962,9 @@ var _ = Describe("OrderController", func() { }) It("should fail when source endpoint does not exist", func() { + counter := metrics.ReconcileErrorsCounterForTest(ControllerOrder, ReasonInvalidEndpoint) + before := testutil.ToFloat64(counter) + createEndpoints("dst-only") // Create order referencing a non-existent source endpoint @@ -973,6 +991,11 @@ var _ = Describe("OrderController", func() { return order.Status.Message }).Should(ContainSubstring("failed to fetch endpoint for source")) + // Verify the reconcile error was counted under the matching reason + Eventually(func() float64 { + return testutil.ToFloat64(counter) - before + }).Should(BeNumerically(">=", 1.0)) + // Verify no artifact workflows were created Consistently(func() int { awList := &arcv1alpha1.ArtifactWorkflowList{} @@ -1169,3 +1192,16 @@ var _ = Describe("OrderController", func() { }) }) + +var _ = Describe("Reconcile error reasons", func() { + It("should use the same strings as the events", func() { + Expect(ReasonInvalidEndpoint).To(Equal("InvalidEndpoint")) + Expect(ReasonInvalidArtifactType).To(Equal("InvalidArtifactType")) + Expect(ReasonInvalidSecret).To(Equal("InvalidSecret")) + Expect(ReasonComputationFailed).To(Equal("ComputationFailed")) + Expect(ReasonHydrationFailed).To(Equal("HydrationFailed")) + Expect(ReasonCreationFailed).To(Equal("CreationFailed")) + Expect(ReasonDeletionFailed).To(Equal("DeletionFailed")) + Expect(ReasonInvalid).To(Equal("Invalid")) + }) +}) diff --git a/pkg/controller/workflow_handler.go b/pkg/controller/workflow_handler.go index 82ee7b83..cad1801f 100644 --- a/pkg/controller/workflow_handler.go +++ b/pkg/controller/workflow_handler.go @@ -15,6 +15,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + "go.opendefense.cloud/arc/pkg/metrics" ) type WorkflowHandler interface { @@ -43,7 +44,9 @@ func (h *SingleWorkflowHandler) DeleteArgoResources(ctx context.Context) error { }, } if err := h.Delete(ctx, &wf); client.IgnoreNotFound(err) != nil { - h.Recorder.Eventf(h.aw, nil, corev1.EventTypeWarning, "DeletionFailed", "Delete", fmt.Sprintf("Failed to delete associated workflow '%s': %v", h.aw.Name, err)) + h.Recorder.Eventf(h.aw, nil, corev1.EventTypeWarning, ReasonDeletionFailed, "Delete", fmt.Sprintf("Failed to delete associated workflow '%s': %v", h.aw.Name, err)) + metrics.RecordReconcileError(ControllerArtifactWorkflow, ReasonDeletionFailed) + return errLogAndWrap(h.log, err, "workflow deletion failed") } h.Recorder.Eventf(h.aw, nil, corev1.EventTypeNormal, "Deleted", "Delete", fmt.Sprintf("Deleted workflow '%s'", h.aw.Name)) @@ -64,7 +67,9 @@ func (h *SingleWorkflowHandler) CreateArgoResources(ctx context.Context) error { } if err := h.Create(ctx, wf); client.IgnoreAlreadyExists(err) != nil { - h.Recorder.Eventf(h.aw, nil, corev1.EventTypeWarning, "CreationFailed", "Create", fmt.Sprintf("Failed to create workflow '%s': %v", wf.GetName(), err)) + h.Recorder.Eventf(h.aw, nil, corev1.EventTypeWarning, ReasonCreationFailed, "Create", fmt.Sprintf("Failed to create workflow '%s': %v", wf.GetName(), err)) + metrics.RecordReconcileError(ControllerArtifactWorkflow, ReasonCreationFailed) + return errLogAndWrap(h.log, err, "failed to create argo workflow") } h.Recorder.Eventf(h.aw, nil, corev1.EventTypeNormal, "Created", "Create", fmt.Sprintf("Created workflow '%s'", wf.GetName())) @@ -83,7 +88,7 @@ func (h *SingleWorkflowHandler) CheckArgoResources(ctx context.Context) error { return errLogAndWrap(h.log, err, "failed to get workflow") } - updated := h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf) + updated, done := h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf) if wf.Status.Phase == wfv1alpha1.WorkflowSucceeded && h.aw.Status.Succeeded != 1 { h.aw.Status.Succeeded = 1 @@ -106,6 +111,10 @@ func (h *SingleWorkflowHandler) CheckArgoResources(ctx context.Context) error { return errLogAndWrap(h.log, err, "failed to update status") } + if done != nil { + done.record() + } + return nil } @@ -129,7 +138,9 @@ func (h *CronWorkflowHandler) DeleteArgoResources(ctx context.Context) error { }, } if err := h.Delete(ctx, &cwf); client.IgnoreNotFound(err) != nil { - h.Recorder.Eventf(h.aw, nil, corev1.EventTypeWarning, "DeletionFailed", "Delete", fmt.Sprintf("Failed to delete associated cron workflow '%s': %v", h.aw.Name, err)) + h.Recorder.Eventf(h.aw, nil, corev1.EventTypeWarning, ReasonDeletionFailed, "Delete", fmt.Sprintf("Failed to delete associated cron workflow '%s': %v", h.aw.Name, err)) + metrics.RecordReconcileError(ControllerArtifactWorkflow, ReasonDeletionFailed) + return errLogAndWrap(h.log, err, "cron workflow deletion failed") } h.Recorder.Eventf(h.aw, nil, corev1.EventTypeNormal, "Deleted", "Delete", fmt.Sprintf("Deleted cron workflow '%s'", h.aw.Name)) @@ -151,7 +162,9 @@ func (h *CronWorkflowHandler) CreateArgoResources(ctx context.Context) error { if err := h.Create(ctx, cwf); err != nil { if client.IgnoreAlreadyExists(err) != nil { - h.Recorder.Eventf(h.aw, nil, corev1.EventTypeWarning, "CreationFailed", "Create", fmt.Sprintf("Failed to create cron workflow '%s': %v", cwf.GetName(), err)) + h.Recorder.Eventf(h.aw, nil, corev1.EventTypeWarning, ReasonCreationFailed, "Create", fmt.Sprintf("Failed to create cron workflow '%s': %v", cwf.GetName(), err)) + metrics.RecordReconcileError(ControllerArtifactWorkflow, ReasonCreationFailed) + return errLogAndWrap(h.log, err, "failed to create argo cron workflow") } } else { @@ -207,7 +220,8 @@ func (h *CronWorkflowHandler) CheckArgoResources(ctx context.Context) error { h.aw.Status.Message = "" h.aw.Status.Phase = arcv1alpha1.WorkflowActive - updated = updated || h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf) + changed, _ := h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf) + updated = changed || updated } } @@ -218,7 +232,8 @@ func (h *CronWorkflowHandler) CheckArgoResources(ctx context.Context) error { return errLogAndWrap(h.log, err, "failed to fetch active workflow") } - updated = updated || h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf) + changed, _ := h.setStatusFromWorkflow(ctx, h.log, h.aw, &wf) + updated = changed || updated if wf.Status.Phase.Completed() { h.aw.Status.ActiveWorkflowRef.Name = "" diff --git a/pkg/metrics/collector.go b/pkg/metrics/collector.go new file mode 100644 index 00000000..c71345b9 --- /dev/null +++ b/pkg/metrics/collector.go @@ -0,0 +1,214 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "context" + "sync/atomic" + "time" + + "github.com/prometheus/client_golang/prometheus" + "sigs.k8s.io/controller-runtime/pkg/client" + + arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" +) + +// collectTimeout bounds a single scrape. Collect has no context of its own, and +// a lazily created or resyncing informer can block in List, so an unbounded +// read would pile up hung scrape goroutines instead of failing. +const collectTimeout = 5 * time.Second + +const ( + modeSingle = "single" + modeCron = "cron" +) + +var ( + ordersDesc = prometheus.NewDesc( + "arc_orders", + "Number of Orders currently in each aggregate phase. This is a current state count, not a cumulative total.", + []string{"namespace", "phase"}, nil, + ) + + workflowsDesc = prometheus.NewDesc( + "arc_artifactworkflows", + "Number of ArtifactWorkflows currently in each phase. This is a current state count, not a cumulative total.", + []string{"namespace", "artifact_type", "mode", "phase"}, nil, + ) + + lastScheduledDesc = prometheus.NewDesc( + "arc_artifactworkflow_last_scheduled_timestamp_seconds", + "Unix timestamp of the most recent scheduling of a cron ArtifactWorkflow. "+ + "Where several cron ArtifactWorkflows share a namespace and artifact type, the oldest of their timestamps is reported.", + []string{"namespace", "artifact_type"}, nil, + ) + + lastSuccessDesc = prometheus.NewDesc( + "arc_artifactworkflow_last_success_timestamp_seconds", + "Unix timestamp of the most recent successful run of a cron ArtifactWorkflow. "+ + "Where several cron ArtifactWorkflows share a namespace and artifact type, the oldest of their timestamps is reported.", + []string{"namespace", "artifact_type"}, nil, + ) +) + +// Collector reports current ARC state by reading the manager cache at scrape +// time. Reading on demand means deleted objects stop being reported without any +// bookkeeping, which is the failure mode that makes reconcile-loop gauges lie. +// +// It reports nothing unless this replica holds leadership. Every replica serves +// the metrics endpoint but only the leader reconciles, so ungated gauges would +// be multiplied by the replica count. +type Collector struct { + reader client.Reader + isLeader atomic.Bool +} + +var _ prometheus.Collector = &Collector{} + +// NewCollector returns a Collector reading through the given reader, which is +// normally the manager cache. +func NewCollector(reader client.Reader) *Collector { + return &Collector{reader: reader} +} + +// Describe implements prometheus.Collector. +func (c *Collector) Describe(ch chan<- *prometheus.Desc) { + ch <- ordersDesc + ch <- workflowsDesc + ch <- lastScheduledDesc + ch <- lastSuccessDesc +} + +// Collect implements prometheus.Collector. +func (c *Collector) Collect(ch chan<- prometheus.Metric) { + if !c.isLeader.Load() { + return + } + + ctx, cancel := context.WithTimeout(context.Background(), collectTimeout) + defer cancel() + + c.collectOrders(ctx, ch) + c.collectWorkflows(ctx, ch) +} + +func (c *Collector) collectOrders(ctx context.Context, ch chan<- prometheus.Metric) { + orders := &arcv1alpha1.OrderList{} + if err := c.reader.List(ctx, orders); err != nil { + // Report the failure rather than reporting zeros. A broken collector + // must not look like a healthy, idle system. + ch <- prometheus.NewInvalidMetric(ordersDesc, err) + + return + } + + type key struct{ namespace, phase string } + + counts := map[key]int{} + for i := range orders.Items { + order := &orders.Items[i] + counts[key{order.Namespace, string(OrderPhase(order.Status.ArtifactWorkflows))}]++ + } + + for k, count := range counts { + ch <- prometheus.MustNewConstMetric(ordersDesc, prometheus.GaugeValue, float64(count), k.namespace, k.phase) + } +} + +func (c *Collector) collectWorkflows(ctx context.Context, ch chan<- prometheus.Metric) { + workflows := &arcv1alpha1.ArtifactWorkflowList{} + if err := c.reader.List(ctx, workflows); err != nil { + ch <- prometheus.NewInvalidMetric(workflowsDesc, err) + + return + } + + type key struct{ namespace, artifactType, mode, phase string } + + counts := map[key]int{} + + // The cron gauges are labelled by namespace and artifact type only, which is + // not unique per object, so they have to be aggregated before they are + // emitted. Two metrics with the same name and labels make the registry + // reject the whole scrape. + scheduled := map[cronKey]int64{} + succeeded := map[cronKey]int64{} + + for i := range workflows.Items { + workflow := &workflows.Items[i] + artifactType := ArtifactTypeOf(workflow) + + mode := modeSingle + if workflow.Spec.Cron != nil { + mode = modeCron + } + + // A freshly created workflow has no phase yet. An explicit Unknown + // beats an empty label value. + phase := string(workflow.Status.Phase) + if phase == "" { + phase = "Unknown" + } + + counts[key{workflow.Namespace, artifactType, mode, phase}]++ + + if mode != modeCron { + continue + } + + cron := cronKey{workflow.Namespace, artifactType} + + if workflow.Status.LastScheduled != nil && !workflow.Status.LastScheduled.IsZero() { + keepOldest(scheduled, cron, workflow.Status.LastScheduled.Unix()) + } + + // Succeeded is the count Argo reports for the cron workflow, so a + // non zero value is what makes CompletionTime a success rather than a + // stop. + if workflow.Status.Succeeded > 0 && !workflow.Status.CompletionTime.IsZero() { + keepOldest(succeeded, cron, workflow.Status.CompletionTime.Unix()) + } + } + + for k, count := range counts { + ch <- prometheus.MustNewConstMetric(workflowsDesc, prometheus.GaugeValue, float64(count), + k.namespace, k.artifactType, k.mode, k.phase) + } + + emitCronTimestamps(ch, lastScheduledDesc, scheduled) + emitCronTimestamps(ch, lastSuccessDesc, succeeded) +} + +// cronKey is the label tuple of the cron timestamp gauges. It is coarser than +// one object, so several cron ArtifactWorkflows can share it. +type cronKey struct{ namespace, artifactType string } + +// keepOldest reduces a group of timestamps to its minimum. The oldest value is +// the one a staleness alert has to see: taking the newest would let a healthy +// workflow mask a stalled sibling in the same group, which is the failure these +// gauges exist to catch. +func keepOldest(timestamps map[cronKey]int64, key cronKey, seconds int64) { + if current, ok := timestamps[key]; ok && current <= seconds { + return + } + + timestamps[key] = seconds +} + +func emitCronTimestamps(ch chan<- prometheus.Metric, desc *prometheus.Desc, timestamps map[cronKey]int64) { + for k, seconds := range timestamps { + ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(seconds), k.namespace, k.artifactType) + } +} + +// ArtifactTypeOf reads the artifact type an ArtifactWorkflow was derived from. +// Workflows created before the label existed report UnknownArtifactType rather +// than being dropped from the metrics. +func ArtifactTypeOf(workflow *arcv1alpha1.ArtifactWorkflow) string { + if value, ok := workflow.Labels[arcv1alpha1.LabelArtifactType]; ok && value != "" { + return value + } + + return UnknownArtifactType +} diff --git a/pkg/metrics/collector_test.go b/pkg/metrics/collector_test.go new file mode 100644 index 00000000..33c09bd8 --- /dev/null +++ b/pkg/metrics/collector_test.go @@ -0,0 +1,341 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "context" + "errors" + "strings" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + dto "github.com/prometheus/client_model/go" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// errSentinel is the error a failingReader returns, so a spec can assert it +// surfaced rather than merely that some error occurred. +var errSentinel = errors.New("sentinel: cache read failed") + +// failingReader is a client.Reader stub that fails List for one object list +// type and succeeds (returning an empty list) for the other, so a spec can +// pin down that a given List call site is actually checked for errors. +type failingReader struct { + failOrders bool + failWorkflows bool +} + +func (r *failingReader) Get(_ context.Context, _ client.ObjectKey, _ client.Object, _ ...client.GetOption) error { + return nil +} + +func (r *failingReader) List(_ context.Context, list client.ObjectList, _ ...client.ListOption) error { + switch list.(type) { + case *arcv1alpha1.OrderList: + if r.failOrders { + return errSentinel + } + case *arcv1alpha1.ArtifactWorkflowList: + if r.failWorkflows { + return errSentinel + } + } + + return nil +} + +func newScheme() *runtime.Scheme { + scheme := runtime.NewScheme() + Expect(arcv1alpha1.AddToScheme(scheme)).To(Succeed()) + + return scheme +} + +func aw(namespace, name, artifactType string, cron bool, phase arcv1alpha1.WorkflowPhase) *arcv1alpha1.ArtifactWorkflow { + obj := &arcv1alpha1.ArtifactWorkflow{ + ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, + Status: arcv1alpha1.ArtifactWorkflowStatus{WorkflowStatus: arcv1alpha1.WorkflowStatus{Phase: phase}}, + } + if artifactType != "" { + obj.Labels = map[string]string{arcv1alpha1.LabelArtifactType: artifactType} + } + if cron { + obj.Spec.Cron = &arcv1alpha1.Cron{} + } + + return obj +} + +// cronWorkflow builds a cron ArtifactWorkflow that has already run, so both +// freshness timestamps are populated. +func cronWorkflow(namespace, name, artifactType string, scheduled, completed int64) *arcv1alpha1.ArtifactWorkflow { + obj := aw(namespace, name, artifactType, true, arcv1alpha1.WorkflowActive) + + lastScheduled := metav1.Unix(scheduled, 0) + obj.Status.LastScheduled = &lastScheduled + obj.Status.CompletionTime = metav1.Unix(completed, 0) + obj.Status.Succeeded = 1 + + return obj +} + +// seriesCount reports how many series a gathered metric family holds. A name +// that was not gathered at all counts as zero rather than failing. +func seriesCount(families []*dto.MetricFamily, name string) int { + for _, family := range families { + if family.GetName() == name { + return len(family.GetMetric()) + } + } + + return 0 +} + +var _ = Describe("Collector", func() { + It("should emit nothing while not the leader", func() { + client := fake.NewClientBuilder().WithScheme(newScheme()). + WithObjects(aw("team-a", "one", "oci", false, arcv1alpha1.WorkflowRunning)).Build() + + collector := NewCollector(client) + + Expect(testutil.CollectAndCount(collector)).To(Equal(0)) + }) + + It("should count workflows by namespace, type, mode and phase", func() { + client := fake.NewClientBuilder().WithScheme(newScheme()).WithObjects( + aw("team-a", "one", "oci", false, arcv1alpha1.WorkflowRunning), + aw("team-a", "two", "oci", false, arcv1alpha1.WorkflowRunning), + aw("team-a", "three", "helm", true, arcv1alpha1.WorkflowActive), + ).Build() + + collector := NewCollector(client) + collector.isLeader.Store(true) + + expected := ` +# HELP arc_artifactworkflows Number of ArtifactWorkflows currently in each phase. This is a current state count, not a cumulative total. +# TYPE arc_artifactworkflows gauge +arc_artifactworkflows{artifact_type="helm",mode="cron",namespace="team-a",phase="Active"} 1 +arc_artifactworkflows{artifact_type="oci",mode="single",namespace="team-a",phase="Running"} 2 +` + Expect(testutil.CollectAndCompare(collector, strings.NewReader(expected), "arc_artifactworkflows")).To(Succeed()) + }) + + It("should keep counts separate across namespaces", func() { + client := fake.NewClientBuilder().WithScheme(newScheme()).WithObjects( + aw("team-a", "one", "oci", false, arcv1alpha1.WorkflowRunning), + aw("team-b", "two", "oci", false, arcv1alpha1.WorkflowRunning), + ).Build() + + collector := NewCollector(client) + collector.isLeader.Store(true) + + expected := ` +# HELP arc_artifactworkflows Number of ArtifactWorkflows currently in each phase. This is a current state count, not a cumulative total. +# TYPE arc_artifactworkflows gauge +arc_artifactworkflows{artifact_type="oci",mode="single",namespace="team-a",phase="Running"} 1 +arc_artifactworkflows{artifact_type="oci",mode="single",namespace="team-b",phase="Running"} 1 +` + Expect(testutil.CollectAndCompare(collector, strings.NewReader(expected), "arc_artifactworkflows")).To(Succeed()) + }) + + It("should report workflows without the type label as unknown", func() { + client := fake.NewClientBuilder().WithScheme(newScheme()).WithObjects( + aw("team-a", "legacy", "", false, arcv1alpha1.WorkflowSucceeded), + ).Build() + + collector := NewCollector(client) + collector.isLeader.Store(true) + + expected := ` +# HELP arc_artifactworkflows Number of ArtifactWorkflows currently in each phase. This is a current state count, not a cumulative total. +# TYPE arc_artifactworkflows gauge +arc_artifactworkflows{artifact_type="unknown",mode="single",namespace="team-a",phase="Succeeded"} 1 +` + Expect(testutil.CollectAndCompare(collector, strings.NewReader(expected), "arc_artifactworkflows")).To(Succeed()) + }) + + It("should normalise an empty phase to Unknown", func() { + client := fake.NewClientBuilder().WithScheme(newScheme()).WithObjects( + aw("team-a", "fresh", "oci", false, arcv1alpha1.WorkflowUnknown), + ).Build() + + collector := NewCollector(client) + collector.isLeader.Store(true) + + expected := ` +# HELP arc_artifactworkflows Number of ArtifactWorkflows currently in each phase. This is a current state count, not a cumulative total. +# TYPE arc_artifactworkflows gauge +arc_artifactworkflows{artifact_type="oci",mode="single",namespace="team-a",phase="Unknown"} 1 +` + Expect(testutil.CollectAndCompare(collector, strings.NewReader(expected), "arc_artifactworkflows")).To(Succeed()) + }) + + It("should roll orders up to an aggregate phase", func() { + order := &arcv1alpha1.Order{ + ObjectMeta: metav1.ObjectMeta{Namespace: "team-a", Name: "nightly"}, + Status: arcv1alpha1.OrderStatus{ + ArtifactWorkflows: map[string]arcv1alpha1.OrderArtifactWorkflowStatus{ + "a": {WorkflowStatus: arcv1alpha1.WorkflowStatus{Phase: arcv1alpha1.WorkflowSucceeded}}, + "b": {WorkflowStatus: arcv1alpha1.WorkflowStatus{Phase: arcv1alpha1.WorkflowFailed}}, + }, + }, + } + + client := fake.NewClientBuilder().WithScheme(newScheme()).WithObjects(order).Build() + + collector := NewCollector(client) + collector.isLeader.Store(true) + + expected := ` +# HELP arc_orders Number of Orders currently in each aggregate phase. This is a current state count, not a cumulative total. +# TYPE arc_orders gauge +arc_orders{namespace="team-a",phase="Failed"} 1 +` + Expect(testutil.CollectAndCompare(collector, strings.NewReader(expected), "arc_orders")).To(Succeed()) + }) + + It("should publish cron freshness timestamps only for successful cron workflows", func() { + scheduled := metav1.NewTime(metav1.Unix(1700000000, 0).Time) + completed := metav1.NewTime(metav1.Unix(1700000600, 0).Time) + + cronAW := aw("team-a", "sync", "oci", true, arcv1alpha1.WorkflowActive) + cronAW.Status.LastScheduled = &scheduled + cronAW.Status.CompletionTime = completed + cronAW.Status.Succeeded = 3 + + neverRan := aw("team-a", "new", "oci", true, arcv1alpha1.WorkflowPending) + + client := fake.NewClientBuilder().WithScheme(newScheme()).WithObjects(cronAW, neverRan).Build() + + collector := NewCollector(client) + collector.isLeader.Store(true) + + expected := ` +# HELP arc_artifactworkflow_last_success_timestamp_seconds Unix timestamp of the most recent successful run of a cron ArtifactWorkflow. Where several cron ArtifactWorkflows share a namespace and artifact type, the oldest of their timestamps is reported. +# TYPE arc_artifactworkflow_last_success_timestamp_seconds gauge +arc_artifactworkflow_last_success_timestamp_seconds{artifact_type="oci",namespace="team-a"} 1.7000006e+09 +` + Expect(testutil.CollectAndCompare(collector, strings.NewReader(expected), + "arc_artifactworkflow_last_success_timestamp_seconds")).To(Succeed()) + }) + + It("should report the oldest timestamp when cron workflows share a namespace and type", func() { + older := cronWorkflow("team-a", "sync-a", "oci", 1700000000, 1700000600) + newer := cronWorkflow("team-a", "sync-b", "oci", 1700009000, 1700009600) + + // A workflow outside the group must keep its own series rather than be + // folded into the minimum. + other := cronWorkflow("team-b", "sync-c", "helm", 1700020000, 1700020600) + + client := fake.NewClientBuilder().WithScheme(newScheme()).WithObjects(older, newer, other).Build() + + collector := NewCollector(client) + collector.isLeader.Store(true) + + // One series per namespace and artifact type, carrying the older of the + // two team-a timestamps, so a staleness alert fires on the stalest + // workflow in the group. + expectedScheduled := ` +# HELP arc_artifactworkflow_last_scheduled_timestamp_seconds Unix timestamp of the most recent scheduling of a cron ArtifactWorkflow. Where several cron ArtifactWorkflows share a namespace and artifact type, the oldest of their timestamps is reported. +# TYPE arc_artifactworkflow_last_scheduled_timestamp_seconds gauge +arc_artifactworkflow_last_scheduled_timestamp_seconds{artifact_type="helm",namespace="team-b"} 1.70002e+09 +arc_artifactworkflow_last_scheduled_timestamp_seconds{artifact_type="oci",namespace="team-a"} 1.7e+09 +` + Expect(testutil.CollectAndCompare(collector, strings.NewReader(expectedScheduled), + "arc_artifactworkflow_last_scheduled_timestamp_seconds")).To(Succeed()) + + expectedSuccess := ` +# HELP arc_artifactworkflow_last_success_timestamp_seconds Unix timestamp of the most recent successful run of a cron ArtifactWorkflow. Where several cron ArtifactWorkflows share a namespace and artifact type, the oldest of their timestamps is reported. +# TYPE arc_artifactworkflow_last_success_timestamp_seconds gauge +arc_artifactworkflow_last_success_timestamp_seconds{artifact_type="helm",namespace="team-b"} 1.7000206e+09 +arc_artifactworkflow_last_success_timestamp_seconds{artifact_type="oci",namespace="team-a"} 1.7000006e+09 +` + Expect(testutil.CollectAndCompare(collector, strings.NewReader(expectedSuccess), + "arc_artifactworkflow_last_success_timestamp_seconds")).To(Succeed()) + }) + + It("should gather cleanly with cron workflows sharing a namespace and type", func() { + // Duplicate series are rejected by the registry for the whole response, + // not just for the offending metric, so this asserts on a full Gather + // rather than on one filtered metric name. + client := fake.NewClientBuilder().WithScheme(newScheme()).WithObjects( + cronWorkflow("team-a", "sync-a", "oci", 1700000000, 1700000600), + cronWorkflow("team-a", "sync-b", "oci", 1700009000, 1700009600), + ).Build() + + collector := NewCollector(client) + collector.isLeader.Store(true) + + registry := prometheus.NewPedanticRegistry() + Expect(registry.Register(collector)).To(Succeed()) + + families, err := registry.Gather() + Expect(err).NotTo(HaveOccurred()) + + for _, name := range []string{ + "arc_artifactworkflow_last_scheduled_timestamp_seconds", + "arc_artifactworkflow_last_success_timestamp_seconds", + } { + Expect(seriesCount(families, name)).To(Equal(1), "%s should hold one series per namespace and artifact type", name) + } + }) +}) + +var _ = Describe("Collector cache errors", func() { + It("should surface a scrape error when the Orders list fails", func() { + collector := NewCollector(&failingReader{failOrders: true}) + collector.isLeader.Store(true) + + registry := prometheus.NewPedanticRegistry() + Expect(registry.Register(collector)).To(Succeed()) + + _, err := registry.Gather() + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring(errSentinel.Error())) + }) + + It("should surface a scrape error when the ArtifactWorkflows list fails", func() { + collector := NewCollector(&failingReader{failWorkflows: true}) + collector.isLeader.Store(true) + + registry := prometheus.NewPedanticRegistry() + Expect(registry.Register(collector)).To(Succeed()) + + _, err := registry.Gather() + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring(errSentinel.Error())) + }) +}) + +var _ = Describe("Collector as a leader elected runnable", func() { + It("should report only while running", func() { + client := fake.NewClientBuilder().WithScheme(newScheme()).WithObjects( + aw("team-a", "one", "oci", false, arcv1alpha1.WorkflowRunning), + ).Build() + + collector := NewCollector(client) + Expect(testutil.CollectAndCount(collector)).To(Equal(0)) + + ctx, cancel := context.WithCancel(context.Background()) + + Expect(collector.NeedLeaderElection()).To(BeTrue()) + + done := make(chan error, 1) + go func() { done <- collector.Start(ctx) }() + + Eventually(func() int { return testutil.CollectAndCount(collector) }).Should(BeNumerically(">", 0)) + + cancel() + Expect(<-done).To(Succeed()) + Eventually(func() int { return testutil.CollectAndCount(collector) }).Should(Equal(0)) + }) +}) diff --git a/pkg/metrics/leader.go b/pkg/metrics/leader.go new file mode 100644 index 00000000..3f2aee52 --- /dev/null +++ b/pkg/metrics/leader.go @@ -0,0 +1,42 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "context" + + "sigs.k8s.io/controller-runtime/pkg/manager" +) + +var ( + _ manager.Runnable = &Collector{} + _ manager.LeaderElectionRunnable = &Collector{} +) + +// NeedLeaderElection implements manager.LeaderElectionRunnable. Every replica +// serves the metrics endpoint but only the leader reconciles, so ungated gauges +// would be multiplied by the replica count. +func (c *Collector) NeedLeaderElection() bool { + return true +} + +// Start implements manager.Runnable, and reports for as long as this replica +// holds leadership. The collector must be added to the manager or it reports +// nothing at all. When leader election is disabled the manager starts leader +// election runnables immediately, so single replica installs report normally. +// +// The manager syncs the caches it knows about before starting leader election +// runnables, but the Order and ArtifactWorkflow informers are created lazily by +// the controllers, which are leader election runnables started alongside this +// one. A scrape that arrives before a controller has asked for its informer +// therefore creates it and waits for it to sync, bounded by collectTimeout, so +// the worst case is a scrape that fails rather than one that hangs. +func (c *Collector) Start(ctx context.Context) error { + c.isLeader.Store(true) + defer c.isLeader.Store(false) + + <-ctx.Done() + + return nil +} diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go new file mode 100644 index 00000000..c002e8ce --- /dev/null +++ b/pkg/metrics/metrics.go @@ -0,0 +1,98 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +// Package metrics exposes ARC domain metrics on the controller manager's +// existing Prometheus endpoint. Current state is reported by a collector that +// reads the manager cache at scrape time; flow and failures are recorded from +// the reconcile path. +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" + + arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" +) + +// Result values for terminal ArtifactWorkflow phases. +const ( + ResultSucceeded = "succeeded" + ResultFailed = "failed" + ResultError = "error" +) + +// UnknownArtifactType is reported for ArtifactWorkflows created before the +// artifact type label existed. +const UnknownArtifactType = "unknown" + +var ( + completions = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "arc_artifactworkflow_completions_total", + Help: "Total single mode ArtifactWorkflows that reached a terminal phase, by result.", + }, []string{"namespace", "artifact_type", "result"}) + + // Buckets are sized for artifact transfers rather than web requests. The + // client_golang defaults top out at 10s, which would put every real run + // into the +Inf bucket. + duration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "arc_artifactworkflow_duration_seconds", + Help: "Argo execution time of single mode ArtifactWorkflows, by result.", + Buckets: []float64{10, 30, 60, 120, 300, 600, 1800, 3600, 7200}, + }, []string{"artifact_type", "result"}) + + reconcileErrors = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "arc_reconcile_errors_total", + Help: "Total classified reconcile failures. The reason label matches the Kubernetes Event reason for the same failure.", + }, []string{"controller", "reason"}) +) + +func init() { + ctrlmetrics.Registry.MustRegister(completions, duration, reconcileErrors) +} + +// ResultFor maps a terminal ArtifactWorkflow phase onto a result label value. +// The second return value is false for phases that are not terminal outcomes, +// including Stopped, which is an operator action rather than a result. +func ResultFor(phase arcv1alpha1.WorkflowPhase) (string, bool) { + switch phase { + case arcv1alpha1.WorkflowSucceeded: + return ResultSucceeded, true + case arcv1alpha1.WorkflowFailed: + return ResultFailed, true + case arcv1alpha1.WorkflowError: + return ResultError, true + default: + return "", false + } +} + +// RecordCompletion counts one ArtifactWorkflow reaching a terminal phase. +// Call it only after the status write that recorded the transition succeeded, +// so a conflicting write cannot be counted twice. +func RecordCompletion(namespace, artifactType, result string) { + completions.WithLabelValues(namespace, artifactType, result).Inc() +} + +// ObserveDuration records how long Argo took to run a workflow. +func ObserveDuration(artifactType, result string, seconds float64) { + duration.WithLabelValues(artifactType, result).Observe(seconds) +} + +// CompletionsCounterForTest exposes one completions series for assertions in +// controller tests. It is not part of the runtime API. +func CompletionsCounterForTest(namespace, artifactType, result string) prometheus.Counter { + return completions.WithLabelValues(namespace, artifactType, result) +} + +// RecordReconcileError counts one classified reconcile failure. The reason must +// be one of the Event reason constants in the controller package so the metric +// and the Kubernetes Event always agree. +func RecordReconcileError(controller, reason string) { + reconcileErrors.WithLabelValues(controller, reason).Inc() +} + +// ReconcileErrorsCounterForTest exposes one reconcile error series for +// assertions in controller tests. It is not part of the runtime API. +func ReconcileErrorsCounterForTest(controller, reason string) prometheus.Counter { + return reconcileErrors.WithLabelValues(controller, reason) +} diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/metrics_test.go new file mode 100644 index 00000000..a964a9d8 --- /dev/null +++ b/pkg/metrics/metrics_test.go @@ -0,0 +1,114 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus/testutil" + dto "github.com/prometheus/client_model/go" + ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" + + arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// cumulativeBucket reads the cumulative count of the given bucket upper bound +// for the artifact_type="oci",result="succeeded" series of +// arc_artifactworkflow_duration_seconds, straight from ctrlmetrics.Registry. +func cumulativeBucket(upperBound float64) float64 { + families, err := ctrlmetrics.Registry.Gather() + Expect(err).NotTo(HaveOccurred()) + + for _, mf := range families { + if mf.GetName() != "arc_artifactworkflow_duration_seconds" { + continue + } + + for _, m := range mf.GetMetric() { + if !hasLabel(m.GetLabel(), "artifact_type", "oci") || !hasLabel(m.GetLabel(), "result", ResultSucceeded) { + continue + } + + for _, b := range m.GetHistogram().GetBucket() { + if b.GetUpperBound() == upperBound { + return float64(b.GetCumulativeCount()) + } + } + } + } + + return 0 +} + +func hasLabel(labels []*dto.LabelPair, name, value string) bool { + for _, l := range labels { + if l.GetName() == name && l.GetValue() == value { + return true + } + } + + return false +} + +var _ = Describe("ResultFor", func() { + It("should map terminal phases to results", func() { + for phase, want := range map[arcv1alpha1.WorkflowPhase]string{ + arcv1alpha1.WorkflowSucceeded: ResultSucceeded, + arcv1alpha1.WorkflowFailed: ResultFailed, + arcv1alpha1.WorkflowError: ResultError, + } { + got, ok := ResultFor(phase) + Expect(ok).To(BeTrue(), "phase %q should be terminal", phase) + Expect(got).To(Equal(want)) + } + }) + + It("should reject non-terminal phases", func() { + for _, phase := range []arcv1alpha1.WorkflowPhase{ + arcv1alpha1.WorkflowUnknown, + arcv1alpha1.WorkflowPending, + arcv1alpha1.WorkflowRunning, + arcv1alpha1.WorkflowActive, + arcv1alpha1.WorkflowStopped, + } { + _, ok := ResultFor(phase) + Expect(ok).To(BeFalse(), "phase %q should not be terminal", phase) + } + }) +}) + +var _ = Describe("Recording helpers", func() { + It("should count completions per namespace, type and result", func() { + before := testutil.ToFloat64(completions.WithLabelValues("team-a", "oci", ResultSucceeded)) + + RecordCompletion("team-a", "oci", ResultSucceeded) + RecordCompletion("team-a", "oci", ResultSucceeded) + + after := testutil.ToFloat64(completions.WithLabelValues("team-a", "oci", ResultSucceeded)) + Expect(after - before).To(Equal(2.0)) + }) + + It("should count reconcile errors per controller and reason", func() { + before := testutil.ToFloat64(reconcileErrors.WithLabelValues("order", "InvalidSecret")) + + RecordReconcileError("order", "InvalidSecret") + + after := testutil.ToFloat64(reconcileErrors.WithLabelValues("order", "InvalidSecret")) + Expect(after - before).To(Equal(1.0)) + }) + + It("should place a 90 second observation in the 120 second bucket, not the 60 second bucket", func() { + before60 := cumulativeBucket(60) + before120 := cumulativeBucket(120) + + ObserveDuration("oci", ResultSucceeded, 90) + + after60 := cumulativeBucket(60) + after120 := cumulativeBucket(120) + + Expect(after120-before120).To(Equal(1.0), "a 90 second observation should land in the le=\"120\" bucket") + Expect(after60-before60).To(Equal(0.0), "a 90 second observation must not increment the le=\"60\" bucket") + }) +}) diff --git a/pkg/metrics/rollup.go b/pkg/metrics/rollup.go new file mode 100644 index 00000000..7dca5490 --- /dev/null +++ b/pkg/metrics/rollup.go @@ -0,0 +1,59 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" +) + +// OrderPhase derives an aggregate phase for an Order from the phases of the +// ArtifactWorkflows it owns. Orders have no phase of their own. +// +// Precedence is deliberate: Failed, then Running, then Stopped, then Succeeded, +// then Pending. A failure anywhere makes the whole Order failed. In flight work +// outranks a deliberate stop, because an Order with one stopped workflow and one +// still running is not settled, and reporting Stopped there would hide the work +// that is still going. The stop is not lost, it surfaces once nothing is in +// flight. A stop is still reported rather than folded into Pending, and an Order +// reads Succeeded only once every workflow has succeeded. +// +// An Order containing a cron artifact moves between phases for as long as it +// exists rather than settling in one. It reads Running while a run is in +// flight, Succeeded between runs, and stays Failed after a failed run until the +// next run reports otherwise. Running therefore means "has work in flight", not +// "is unhealthy", and Succeeded means "nothing outstanding right now", not +// "finished for good". +func OrderPhase(statuses map[string]arcv1alpha1.OrderArtifactWorkflowStatus) arcv1alpha1.WorkflowPhase { + if len(statuses) == 0 { + return arcv1alpha1.WorkflowPending + } + + var stopped, inProgress bool + + succeeded := 0 + + for _, status := range statuses { + switch { + case status.Phase == arcv1alpha1.WorkflowFailed, status.Phase == arcv1alpha1.WorkflowError: + return arcv1alpha1.WorkflowFailed + case status.Phase == arcv1alpha1.WorkflowStopped: + stopped = true + case status.Phase.InProgress(): + inProgress = true + case status.Phase == arcv1alpha1.WorkflowSucceeded: + succeeded++ + } + } + + switch { + case inProgress: + return arcv1alpha1.WorkflowRunning + case stopped: + return arcv1alpha1.WorkflowStopped + case succeeded == len(statuses): + return arcv1alpha1.WorkflowSucceeded + default: + return arcv1alpha1.WorkflowPending + } +} diff --git a/pkg/metrics/rollup_test.go b/pkg/metrics/rollup_test.go new file mode 100644 index 00000000..f90fd32c --- /dev/null +++ b/pkg/metrics/rollup_test.go @@ -0,0 +1,94 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func awStatuses(phases ...arcv1alpha1.WorkflowPhase) map[string]arcv1alpha1.OrderArtifactWorkflowStatus { + out := map[string]arcv1alpha1.OrderArtifactWorkflowStatus{} + for i, phase := range phases { + out[string(rune('a'+i))] = arcv1alpha1.OrderArtifactWorkflowStatus{ + WorkflowStatus: arcv1alpha1.WorkflowStatus{Phase: phase}, + ArtifactIndex: i, + } + } + + return out +} + +var _ = Describe("OrderPhase", func() { + It("should report Pending for an order with no workflows yet", func() { + Expect(OrderPhase(nil)).To(Equal(arcv1alpha1.WorkflowPending)) + Expect(OrderPhase(awStatuses())).To(Equal(arcv1alpha1.WorkflowPending)) + }) + + It("should report Failed when any workflow failed or errored", func() { + Expect(OrderPhase(awStatuses( + arcv1alpha1.WorkflowSucceeded, arcv1alpha1.WorkflowFailed, + ))).To(Equal(arcv1alpha1.WorkflowFailed)) + + Expect(OrderPhase(awStatuses( + arcv1alpha1.WorkflowRunning, arcv1alpha1.WorkflowError, + ))).To(Equal(arcv1alpha1.WorkflowFailed)) + }) + + It("should prefer Failed over Stopped", func() { + Expect(OrderPhase(awStatuses( + arcv1alpha1.WorkflowStopped, arcv1alpha1.WorkflowFailed, + ))).To(Equal(arcv1alpha1.WorkflowFailed)) + }) + + It("should report Stopped rather than laundering it into Pending", func() { + Expect(OrderPhase(awStatuses( + arcv1alpha1.WorkflowSucceeded, arcv1alpha1.WorkflowStopped, + ))).To(Equal(arcv1alpha1.WorkflowStopped)) + }) + + It("should prefer in flight work over Stopped", func() { + for _, phase := range []arcv1alpha1.WorkflowPhase{ + arcv1alpha1.WorkflowPending, + arcv1alpha1.WorkflowRunning, + arcv1alpha1.WorkflowActive, + } { + Expect(OrderPhase(awStatuses( + arcv1alpha1.WorkflowStopped, phase, + ))).To(Equal(arcv1alpha1.WorkflowRunning), "phase %q should outrank Stopped", phase) + } + }) + + It("should report Stopped once nothing is in flight any more", func() { + Expect(OrderPhase(awStatuses( + arcv1alpha1.WorkflowStopped, arcv1alpha1.WorkflowSucceeded, + ))).To(Equal(arcv1alpha1.WorkflowStopped)) + }) + + It("should report Running while any workflow is in progress", func() { + for _, phase := range []arcv1alpha1.WorkflowPhase{ + arcv1alpha1.WorkflowPending, + arcv1alpha1.WorkflowRunning, + arcv1alpha1.WorkflowActive, + } { + Expect(OrderPhase(awStatuses( + arcv1alpha1.WorkflowSucceeded, phase, + ))).To(Equal(arcv1alpha1.WorkflowRunning)) + } + }) + + It("should report Succeeded only when every workflow succeeded", func() { + Expect(OrderPhase(awStatuses( + arcv1alpha1.WorkflowSucceeded, arcv1alpha1.WorkflowSucceeded, + ))).To(Equal(arcv1alpha1.WorkflowSucceeded)) + }) + + It("should report Pending for workflows with no phase set", func() { + Expect(OrderPhase(awStatuses( + arcv1alpha1.WorkflowUnknown, + ))).To(Equal(arcv1alpha1.WorkflowPending)) + }) +}) diff --git a/pkg/metrics/suite_test.go b/pkg/metrics/suite_test.go new file mode 100644 index 00000000..53d22c62 --- /dev/null +++ b/pkg/metrics/suite_test.go @@ -0,0 +1,16 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestMetrics(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Metrics Suite") +}