From 6a5d6352565b224945d32c2cc7b439a5e5fc941d Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 16:29:47 +0200 Subject: [PATCH 01/13] feat(controller): label artifact workflows with their artifact type If the artifact type is not a valid Kubernetes label value, the label is skipped rather than truncated: the workflow stays creatable and reports as artifact_type="unknown". A label value inherited from the cloned object is cleared in that case too, so a stale type never survives onto the new ArtifactWorkflow. --- api/arc/v1alpha1/labels.go | 12 +++++ pkg/controller/helpers.go | 25 +++++++-- pkg/controller/helpers_test.go | 81 +++++++++++++++++++++++++++++- pkg/controller/order_controller.go | 13 ++--- 4 files changed, 121 insertions(+), 10 deletions(-) create mode 100644 api/arc/v1alpha1/labels.go 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/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/order_controller.go b/pkg/controller/order_controller.go index 8d4e5677..8eaa650b 100644 --- a/pkg/controller/order_controller.go +++ b/pkg/controller/order_controller.go @@ -87,9 +87,10 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl // 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 +142,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) @@ -308,7 +309,7 @@ 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) @@ -325,7 +326,7 @@ 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) @@ -530,7 +531,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, From 4c6581d9f9ac87f773fcebc4e57f419a8d0cf618 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 16:40:43 +0200 Subject: [PATCH 02/13] feat(metrics): define ARC domain metrics --- go.mod | 2 +- pkg/metrics/metrics.go | 113 ++++++++++++++++++++++++++++++++++++ pkg/metrics/metrics_test.go | 72 +++++++++++++++++++++++ pkg/metrics/suite_test.go | 16 +++++ 4 files changed, 202 insertions(+), 1 deletion(-) create mode 100644 pkg/metrics/metrics.go create mode 100644 pkg/metrics/metrics_test.go create mode 100644 pkg/metrics/suite_test.go diff --git a/go.mod b/go.mod index e6250aa5..9ef55a89 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ 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/robfig/cron/v3 v3.0.1 github.com/spf13/pflag v1.0.10 go.opendefense.cloud/kit v0.3.4 @@ -76,7 +77,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 diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go new file mode 100644 index 00000000..2bf1fae9 --- /dev/null +++ b/pkg/metrics/metrics.go @@ -0,0 +1,113 @@ +// 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 ( + "runtime" + "runtime/debug" + + "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"}) + + buildInfo = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "arc_build_info", + Help: "Build information of the running controller manager. Always 1.", + }, []string{"version", "revision", "go_version"}) +) + +func init() { + ctrlmetrics.Registry.MustRegister(completions, duration, reconcileErrors, buildInfo) +} + +// 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) +} + +// 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() +} + +// SetBuildInfo publishes build information read from the embedded build data. +func SetBuildInfo() { + version, revision := "unknown", "unknown" + + if info, ok := debug.ReadBuildInfo(); ok { + if info.Main.Version != "" { + version = info.Main.Version + } + + for _, setting := range info.Settings { + if setting.Key == "vcs.revision" { + revision = setting.Value + } + } + } + + buildInfo.WithLabelValues(version, revision, runtime.Version()).Set(1) +} diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/metrics_test.go new file mode 100644 index 00000000..13bbea08 --- /dev/null +++ b/pkg/metrics/metrics_test.go @@ -0,0 +1,72 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus/testutil" + + arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +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 record durations in the histogram", func() { + ObserveDuration("oci", ResultSucceeded, 90) + + Expect(testutil.CollectAndCount(duration)).To(BeNumerically(">", 0)) + }) + + It("should expose build info as a constant one", func() { + SetBuildInfo() + Expect(testutil.CollectAndCount(buildInfo)).To(Equal(1)) + }) +}) 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") +} From 0ba7a3ea24d98276bdf3014b12aaba2b7e1e8ed9 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 16:47:12 +0200 Subject: [PATCH 03/13] feat(metrics): derive an aggregate phase for orders --- pkg/metrics/rollup.go | 53 ++++++++++++++++++++++++++ pkg/metrics/rollup_test.go | 76 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 129 insertions(+) create mode 100644 pkg/metrics/rollup.go create mode 100644 pkg/metrics/rollup_test.go diff --git a/pkg/metrics/rollup.go b/pkg/metrics/rollup.go new file mode 100644 index 00000000..2f048650 --- /dev/null +++ b/pkg/metrics/rollup.go @@ -0,0 +1,53 @@ +// 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. A failure anywhere makes the whole Order failed. A +// deliberate stop is reported rather than folded into Pending. Otherwise +// in progress work wins over completed work, so an Order reads Succeeded only +// once every workflow has succeeded. +// +// An Order containing a cron artifact never reaches a terminal phase, because +// cron workflows are rescheduled indefinitely. Running therefore means "has +// work in flight", not "is unhealthy". +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 stopped: + return arcv1alpha1.WorkflowStopped + case inProgress: + return arcv1alpha1.WorkflowRunning + 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..1b65b231 --- /dev/null +++ b/pkg/metrics/rollup_test.go @@ -0,0 +1,76 @@ +// 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 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)) + }) +}) From a5f5552d4bf025d6aef4e4d1305f967bbcf4aef9 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 16:54:09 +0200 Subject: [PATCH 04/13] feat(metrics): report order and workflow state from the manager cache The cron freshness gauges aggregate to one series per namespace and artifact type, taking the oldest timestamp in the group, so a healthy workflow cannot mask a stalled sibling under the same label tuple. --- pkg/metrics/collector.go | 223 +++++++++++++++++++++++++++ pkg/metrics/collector_test.go | 279 ++++++++++++++++++++++++++++++++++ pkg/metrics/leader.go | 45 ++++++ 3 files changed, 547 insertions(+) create mode 100644 pkg/metrics/collector.go create mode 100644 pkg/metrics/collector_test.go create mode 100644 pkg/metrics/leader.go diff --git a/pkg/metrics/collector.go b/pkg/metrics/collector.go new file mode 100644 index 00000000..8069223a --- /dev/null +++ b/pkg/metrics/collector.go @@ -0,0 +1,223 @@ +// 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" + + // phaseUnknown replaces the empty phase of a freshly created workflow, so + // no series carries an empty label value. + phaseUnknown = "Unknown" +) + +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 := modeOf(workflow) + + counts[key{workflow.Namespace, artifactType, mode, phaseLabel(workflow.Status.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 +} + +func modeOf(workflow *arcv1alpha1.ArtifactWorkflow) string { + if workflow.Spec.Cron != nil { + return modeCron + } + + return modeSingle +} + +func phaseLabel(phase arcv1alpha1.WorkflowPhase) string { + if phase == arcv1alpha1.WorkflowUnknown { + return phaseUnknown + } + + return string(phase) +} diff --git a/pkg/metrics/collector_test.go b/pkg/metrics/collector_test.go new file mode 100644 index 00000000..e64892e4 --- /dev/null +++ b/pkg/metrics/collector_test.go @@ -0,0 +1,279 @@ +// Copyright 2025 BWI GmbH and Artifact Conduit contributors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "context" + "strings" + + "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/fake" + + arcv1alpha1 "go.opendefense.cloud/arc/api/arc/v1alpha1" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +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("LeaderRunnable", 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()) + + done := make(chan error, 1) + go func() { done <- collector.LeaderRunnable().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..2a0d8449 --- /dev/null +++ b/pkg/metrics/leader.go @@ -0,0 +1,45 @@ +// 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" +) + +// leaderGate flips the collector on for as long as this replica holds +// leadership. When leader election is disabled the manager starts leader +// election runnables immediately, so single replica installs report normally. +type leaderGate struct { + collector *Collector +} + +var ( + _ manager.Runnable = &leaderGate{} + _ manager.LeaderElectionRunnable = &leaderGate{} +) + +// LeaderRunnable returns a Runnable that must be added to the manager for the +// collector to report anything. +func (c *Collector) LeaderRunnable() manager.Runnable { + return &leaderGate{collector: c} +} + +// NeedLeaderElection implements manager.LeaderElectionRunnable. +func (g *leaderGate) NeedLeaderElection() bool { + return true +} + +// Start implements manager.Runnable. The manager starts caches and waits for +// them to sync before starting leader election runnables, so by the time this +// flips the collector on, the cache the collector reads is already synced. +func (g *leaderGate) Start(ctx context.Context) error { + g.collector.isLeader.Store(true) + defer g.collector.isLeader.Store(false) + + <-ctx.Done() + + return nil +} From 172448168211718f8a73369e1d90e9deb9157313 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 17:04:08 +0200 Subject: [PATCH 05/13] feat(metrics): record workflow completions and durations Durations come from the Argo workflow's own start and finish times rather than ARC status timestamps, which are observation times and are never set for failures. Also fixes the cron status path skipping its phase update: the short-circuiting || meant setStatusFromWorkflow was never called once any earlier field had already marked the status dirty. --- pkg/controller/artifactworkflow_controller.go | 7 +- .../artifactworkflow_controller_test.go | 85 +++++++++++++++++++ pkg/controller/metrics.go | 65 ++++++++++++++ pkg/controller/workflow_handler.go | 12 ++- pkg/metrics/metrics.go | 6 ++ 5 files changed, 169 insertions(+), 6 deletions(-) create mode 100644 pkg/controller/metrics.go diff --git a/pkg/controller/artifactworkflow_controller.go b/pkg/controller/artifactworkflow_controller.go index 165437be..e2d2116d 100644 --- a/pkg/controller/artifactworkflow_controller.go +++ b/pkg/controller/artifactworkflow_controller.go @@ -146,10 +146,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 +163,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) { diff --git a/pkg/controller/artifactworkflow_controller_test.go b/pkg/controller/artifactworkflow_controller_test.go index fa5bcf10..efac81ea 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" @@ -167,6 +169,38 @@ 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()) + + Eventually(func() float64 { + return testutil.ToFloat64(counter) - before + }).Should(Equal(1.0)) + }) + It("should track failed Workflow information of created ArtifactWorkflows", func() { awName := "track-failed-status" aw := &arcv1alpha1.ArtifactWorkflow{ @@ -504,3 +538,54 @@ 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 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/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/workflow_handler.go b/pkg/controller/workflow_handler.go index 82ee7b83..b0ed0de3 100644 --- a/pkg/controller/workflow_handler.go +++ b/pkg/controller/workflow_handler.go @@ -83,7 +83,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 +106,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 } @@ -207,7 +211,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 +223,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/metrics.go b/pkg/metrics/metrics.go index 2bf1fae9..df7cdf4a 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -86,6 +86,12 @@ 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. From 006d37f0de245b26c2d25c88ea3a1be79924fea2 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 17:19:22 +0200 Subject: [PATCH 06/13] feat(metrics): count reconcile errors by event reason --- pkg/controller/const.go | 24 ++++++++++ pkg/controller/order_controller.go | 61 ++++++++++++++++++------- pkg/controller/order_controller_test.go | 13 ++++++ 3 files changed, 81 insertions(+), 17 deletions(-) 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/order_controller.go b/pkg/controller/order_controller.go index 8eaa650b..37732ae0 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,7 +83,7 @@ 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 { @@ -168,7 +169,8 @@ 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) + r.Recorder.Eventf(order, nil, corev1.EventTypeWarning, ReasonComputationFailed, "Compute", "Failed to compute desired artifact workflow for artifact index %d: %v", i, err) + metrics.RecordReconcileError(ControllerOrder, ReasonComputationFailed) 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") @@ -222,7 +224,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 != "" { @@ -272,13 +276,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") } @@ -288,7 +296,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 { @@ -312,7 +321,9 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl 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") } @@ -329,7 +340,9 @@ func (r *OrderReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl 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") } @@ -425,25 +438,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") } @@ -451,7 +470,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 ( @@ -474,13 +495,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") } @@ -489,7 +512,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") } } @@ -497,7 +522,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") } } diff --git a/pkg/controller/order_controller_test.go b/pkg/controller/order_controller_test.go index 4854578d..7d7634e0 100644 --- a/pkg/controller/order_controller_test.go +++ b/pkg/controller/order_controller_test.go @@ -1169,3 +1169,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")) + }) +}) From 7c775b569c5be44aa9227ad893fd74da1c0b2cc5 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 17:32:10 +0200 Subject: [PATCH 07/13] feat(metrics): register the ARC metrics collector with the manager --- cmd/arc-controller-manager/main.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/cmd/arc-controller-manager/main.go b/cmd/arc-controller-manager/main.go index a081b2cd..9b347fd5 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,16 @@ func main() { } } + arcMetrics := arcmetrics.NewCollector(mgr.GetCache()) + ctrlmetrics.Registry.MustRegister(arcMetrics) + + if err := mgr.Add(arcMetrics.LeaderRunnable()); err != nil { + setupLog.Error(err, "unable to add metrics leader gate") + os.Exit(1) + } + + arcmetrics.SetBuildInfo() + if err := wfv1alpha1.AddToScheme(mgr.GetScheme()); err != nil { setupLog.Error(err, "failed to add Argo Workflows types to scheme") os.Exit(1) From 9c57110c6aa7664038a4f8bc861aa5bd1e5dbcb0 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 17:54:14 +0200 Subject: [PATCH 08/13] test(metrics): cover duration buckets, collector cache errors and completion counting --- go.mod | 2 +- .../artifactworkflow_controller_test.go | 46 +++++++++++++- pkg/controller/order_controller_test.go | 21 +++++++ pkg/metrics/collector_test.go | 60 +++++++++++++++++++ pkg/metrics/metrics.go | 6 ++ pkg/metrics/metrics_test.go | 51 +++++++++++++++- 6 files changed, 181 insertions(+), 5 deletions(-) diff --git a/go.mod b/go.mod index 9ef55a89..acbab382 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,7 @@ require ( 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 @@ -77,7 +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_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_test.go b/pkg/controller/artifactworkflow_controller_test.go index efac81ea..0e143111 100644 --- a/pkg/controller/artifactworkflow_controller_test.go +++ b/pkg/controller/artifactworkflow_controller_test.go @@ -196,9 +196,11 @@ var _ = Describe("ArtifactWorkflowController", func() { wf.Status.Phase = wfv1alpha1.WorkflowSucceeded Expect(k8sClient.Update(ctx, wf)).To(Succeed()) - Eventually(func() float64 { + delta := func() float64 { return testutil.ToFloat64(counter) - before - }).Should(Equal(1.0)) + } + Eventually(delta).Should(Equal(1.0)) + Consistently(delta).Should(Equal(1.0)) }) It("should track failed Workflow information of created ArtifactWorkflows", func() { @@ -577,6 +579,46 @@ var _ = Describe("newCompletion", func() { 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 diff --git a/pkg/controller/order_controller_test.go b/pkg/controller/order_controller_test.go index 7d7634e0..7512a932 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,12 @@ var _ = Describe("OrderController", func() { }) It("should fail when artifact type does not exist", func() { + // A missing ArtifactType/ClusterArtifactType surfaces here as + // ReasonComputationFailed, recorded once Reconcile wraps the + // error from computeDesiredAW, rather than ReasonInvalidArtifactType. + counter := metrics.ReconcileErrorsCounterForTest(ControllerOrder, ReasonComputationFailed) + before := testutil.ToFloat64(counter) + createEndpoints("src-nonexistent", "dst-nonexistent") // Create order referencing a non-existent artifact type @@ -937,6 +945,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 +960,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 +989,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{} diff --git a/pkg/metrics/collector_test.go b/pkg/metrics/collector_test.go index e64892e4..4a80e51c 100644 --- a/pkg/metrics/collector_test.go +++ b/pkg/metrics/collector_test.go @@ -5,12 +5,15 @@ 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" @@ -19,6 +22,37 @@ import ( . "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()) @@ -256,6 +290,32 @@ arc_artifactworkflow_last_success_timestamp_seconds{artifact_type="oci",namespac }) }) +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("LeaderRunnable", func() { It("should report only while running", func() { client := fake.NewClientBuilder().WithScheme(newScheme()).WithObjects( diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index df7cdf4a..4fc56497 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -99,6 +99,12 @@ 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) +} + // SetBuildInfo publishes build information read from the embedded build data. func SetBuildInfo() { version, revision := "unknown", "unknown" diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/metrics_test.go index 13bbea08..2ae90c5a 100644 --- a/pkg/metrics/metrics_test.go +++ b/pkg/metrics/metrics_test.go @@ -5,6 +5,8 @@ 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" @@ -12,6 +14,44 @@ import ( . "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{ @@ -59,10 +99,17 @@ var _ = Describe("Recording helpers", func() { Expect(after - before).To(Equal(1.0)) }) - It("should record durations in the histogram", func() { + 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) - Expect(testutil.CollectAndCount(duration)).To(BeNumerically(">", 0)) + 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") }) It("should expose build info as a constant one", func() { From 0e242063d9e5515fb9319831a9df20cd58cc0d66 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 18:25:09 +0200 Subject: [PATCH 09/13] refactor(metrics): drop arc_build_info The metric cannot carry a useful value in the shipped image. Nothing stamps a version through ldflags, and the Docker build context excludes .git -> no vcs.revision in the embedded build info. Every release would report version="(devel)", and "(devel)" passes the != "" check, so it even beats the honest "unknown" fallback. Better no metric than one that lies about which build is running. Comes back once the release build stamps version and revision properly. --- cmd/arc-controller-manager/main.go | 2 -- pkg/metrics/metrics.go | 29 +---------------------------- pkg/metrics/metrics_test.go | 5 ----- 3 files changed, 1 insertion(+), 35 deletions(-) diff --git a/cmd/arc-controller-manager/main.go b/cmd/arc-controller-manager/main.go index 9b347fd5..362cdbe9 100644 --- a/cmd/arc-controller-manager/main.go +++ b/cmd/arc-controller-manager/main.go @@ -178,8 +178,6 @@ func main() { os.Exit(1) } - arcmetrics.SetBuildInfo() - 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/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 4fc56497..c002e8ce 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -8,9 +8,6 @@ package metrics import ( - "runtime" - "runtime/debug" - "github.com/prometheus/client_golang/prometheus" ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" @@ -47,15 +44,10 @@ var ( 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"}) - - buildInfo = prometheus.NewGaugeVec(prometheus.GaugeOpts{ - Name: "arc_build_info", - Help: "Build information of the running controller manager. Always 1.", - }, []string{"version", "revision", "go_version"}) ) func init() { - ctrlmetrics.Registry.MustRegister(completions, duration, reconcileErrors, buildInfo) + ctrlmetrics.Registry.MustRegister(completions, duration, reconcileErrors) } // ResultFor maps a terminal ArtifactWorkflow phase onto a result label value. @@ -104,22 +96,3 @@ func RecordReconcileError(controller, reason string) { func ReconcileErrorsCounterForTest(controller, reason string) prometheus.Counter { return reconcileErrors.WithLabelValues(controller, reason) } - -// SetBuildInfo publishes build information read from the embedded build data. -func SetBuildInfo() { - version, revision := "unknown", "unknown" - - if info, ok := debug.ReadBuildInfo(); ok { - if info.Main.Version != "" { - version = info.Main.Version - } - - for _, setting := range info.Settings { - if setting.Key == "vcs.revision" { - revision = setting.Value - } - } - } - - buildInfo.WithLabelValues(version, revision, runtime.Version()).Set(1) -} diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/metrics_test.go index 2ae90c5a..a964a9d8 100644 --- a/pkg/metrics/metrics_test.go +++ b/pkg/metrics/metrics_test.go @@ -111,9 +111,4 @@ var _ = Describe("Recording helpers", func() { 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") }) - - It("should expose build info as a constant one", func() { - SetBuildInfo() - Expect(testutil.CollectAndCount(buildInfo)).To(Equal(1)) - }) }) From 00293952fcdc23cb686ec6698e6ab78c08708e98 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 18:25:18 +0200 Subject: [PATCH 10/13] fix(controller): count every reconcile failure exactly once Two problems with the reason label, both of them break aggregates over it. Double counting: every failure inside computeDesiredAW recorded its own reason and then the caller recorded again as ComputationFailed. One failed endpoint fetch counted under InvalidEndpoint and under ComputationFailed, so sum(rate(arc_reconcile_errors_total[5m])), the natural "how many reconcile failures" query, read 2x the truth for that whole class, with ComputationFailed shadowing every specific reason as a sibling series. The caller no longer counts. ComputationFailed is now reserved for the two paths in computeDesiredAW that have no Event of their own, the ClusterArtifactType fetch and the sha encoding, where it is the reason the only Event fires under anyway. Missing controller: const.go claims the reason constants keep the metric and the Event in agreement, which was true for the Order controller only. The ArtifactWorkflow controller still emitted warning Events with raw literals and recorded nothing, and ControllerArtifactWorkflow had zero references. All six sites now use the constants and record beside the Event: missing secrets, failed Argo workflow creation, failed deletion, for both the single and the cron handler. No literal warning reasons are left in pkg/controller. Events are untouched, only the metric calls changed. --- pkg/controller/artifactworkflow_controller.go | 9 +++++-- .../artifactworkflow_controller_test.go | 26 +++++++++++++++++++ pkg/controller/order_controller.go | 13 +++++++++- pkg/controller/order_controller_test.go | 8 +++--- pkg/controller/workflow_handler.go | 17 +++++++++--- 5 files changed, 63 insertions(+), 10 deletions(-) diff --git a/pkg/controller/artifactworkflow_controller.go b/pkg/controller/artifactworkflow_controller.go index e2d2116d..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 ( @@ -225,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) } } @@ -233,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 0e143111..cc09382f 100644 --- a/pkg/controller/artifactworkflow_controller_test.go +++ b/pkg/controller/artifactworkflow_controller_test.go @@ -124,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{ diff --git a/pkg/controller/order_controller.go b/pkg/controller/order_controller.go index 37732ae0..fc948d55 100644 --- a/pkg/controller/order_controller.go +++ b/pkg/controller/order_controller.go @@ -169,8 +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 { + // 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) - metrics.RecordReconcileError(ControllerOrder, ReasonComputationFailed) 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") @@ -482,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 @@ -549,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") } diff --git a/pkg/controller/order_controller_test.go b/pkg/controller/order_controller_test.go index 7512a932..c3f8780e 100644 --- a/pkg/controller/order_controller_test.go +++ b/pkg/controller/order_controller_test.go @@ -913,9 +913,11 @@ var _ = Describe("OrderController", func() { }) It("should fail when artifact type does not exist", func() { - // A missing ArtifactType/ClusterArtifactType surfaces here as - // ReasonComputationFailed, recorded once Reconcile wraps the - // error from computeDesiredAW, rather than ReasonInvalidArtifactType. + // 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) diff --git a/pkg/controller/workflow_handler.go b/pkg/controller/workflow_handler.go index b0ed0de3..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())) @@ -133,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)) @@ -155,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 { From a2ccb7f4ffd548fe01e0625e14a44a5686b82f26 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Wed, 26 Aug 2026 18:25:26 +0200 Subject: [PATCH 11/13] docs(metrics): correct the cron rollup and cache sync comments Two comments described behaviour the code does not have. rollup.go claimed an Order containing a cron artifact never reaches a terminal phase. It does: a cron ArtifactWorkflow reaches Succeeded between runs (artifactworkflow_controller_test.go covers it) and the Order mirrors that status straight through, so cron Orders read Succeeded between runs and stay Failed after a failed one until the next run says otherwise. leader.go claimed the cache is synced by the time the gate flips. Only half true. 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 the gate. The first scrape after election can therefore create an informer and wait for it, bounded by collectTimeout. Degrades to a failed scrape, not a hung one, so no code change, just an honest comment. --- pkg/metrics/leader.go | 10 +++++++--- pkg/metrics/rollup.go | 9 ++++++--- 2 files changed, 13 insertions(+), 6 deletions(-) diff --git a/pkg/metrics/leader.go b/pkg/metrics/leader.go index 2a0d8449..99d0434e 100644 --- a/pkg/metrics/leader.go +++ b/pkg/metrics/leader.go @@ -32,9 +32,13 @@ func (g *leaderGate) NeedLeaderElection() bool { return true } -// Start implements manager.Runnable. The manager starts caches and waits for -// them to sync before starting leader election runnables, so by the time this -// flips the collector on, the cache the collector reads is already synced. +// Start implements manager.Runnable. 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 (g *leaderGate) Start(ctx context.Context) error { g.collector.isLeader.Store(true) defer g.collector.isLeader.Store(false) diff --git a/pkg/metrics/rollup.go b/pkg/metrics/rollup.go index 2f048650..5ba48928 100644 --- a/pkg/metrics/rollup.go +++ b/pkg/metrics/rollup.go @@ -15,9 +15,12 @@ import ( // in progress work wins over completed work, so an Order reads Succeeded only // once every workflow has succeeded. // -// An Order containing a cron artifact never reaches a terminal phase, because -// cron workflows are rescheduled indefinitely. Running therefore means "has -// work in flight", not "is unhealthy". +// 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 From b7d4432d083edbc0b6f560feb5e01e3186c115d2 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Fri, 28 Aug 2026 11:57:28 +0200 Subject: [PATCH 12/13] fix(metrics): prefer in flight work over Stopped in the order rollup An Order holding one Stopped workflow and one Running workflow reported Stopped, which hid work that was still going. In flight now outranks a stop. The stop is not lost, it surfaces once nothing is in flight. --- pkg/metrics/rollup.go | 15 +++++++++------ pkg/metrics/rollup_test.go | 18 ++++++++++++++++++ 2 files changed, 27 insertions(+), 6 deletions(-) diff --git a/pkg/metrics/rollup.go b/pkg/metrics/rollup.go index 5ba48928..7dca5490 100644 --- a/pkg/metrics/rollup.go +++ b/pkg/metrics/rollup.go @@ -10,10 +10,13 @@ import ( // 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. A failure anywhere makes the whole Order failed. A -// deliberate stop is reported rather than folded into Pending. Otherwise -// in progress work wins over completed work, so an Order reads Succeeded only -// once every workflow has succeeded. +// 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 @@ -44,10 +47,10 @@ func OrderPhase(statuses map[string]arcv1alpha1.OrderArtifactWorkflowStatus) arc } switch { - case stopped: - return arcv1alpha1.WorkflowStopped case inProgress: return arcv1alpha1.WorkflowRunning + case stopped: + return arcv1alpha1.WorkflowStopped case succeeded == len(statuses): return arcv1alpha1.WorkflowSucceeded default: diff --git a/pkg/metrics/rollup_test.go b/pkg/metrics/rollup_test.go index 1b65b231..f90fd32c 100644 --- a/pkg/metrics/rollup_test.go +++ b/pkg/metrics/rollup_test.go @@ -50,6 +50,24 @@ var _ = Describe("OrderPhase", func() { ))).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, From 63f3ff6c3d5ceb4d96af8c0c843b0f2bfbbd0449 Mon Sep 17 00:00:00 2001 From: Chris Bargmann Date: Fri, 28 Aug 2026 15:10:48 +0200 Subject: [PATCH 13/13] refactor(metrics): drop the leader gate wrapper and two single caller helpers Collector implements Runnable and LeaderElectionRunnable directly, so the leaderGate struct and its constructor go away. modeOf and phaseLabel each had one caller and are inlined. --- cmd/arc-controller-manager/main.go | 2 +- pkg/metrics/collector.go | 35 +++++++++------------- pkg/metrics/collector_test.go | 6 ++-- pkg/metrics/leader.go | 47 +++++++++++++----------------- 4 files changed, 38 insertions(+), 52 deletions(-) diff --git a/cmd/arc-controller-manager/main.go b/cmd/arc-controller-manager/main.go index 362cdbe9..95b591b9 100644 --- a/cmd/arc-controller-manager/main.go +++ b/cmd/arc-controller-manager/main.go @@ -173,7 +173,7 @@ func main() { arcMetrics := arcmetrics.NewCollector(mgr.GetCache()) ctrlmetrics.Registry.MustRegister(arcMetrics) - if err := mgr.Add(arcMetrics.LeaderRunnable()); err != nil { + if err := mgr.Add(arcMetrics); err != nil { setupLog.Error(err, "unable to add metrics leader gate") os.Exit(1) } diff --git a/pkg/metrics/collector.go b/pkg/metrics/collector.go index 8069223a..c71345b9 100644 --- a/pkg/metrics/collector.go +++ b/pkg/metrics/collector.go @@ -22,10 +22,6 @@ const collectTimeout = 5 * time.Second const ( modeSingle = "single" modeCron = "cron" - - // phaseUnknown replaces the empty phase of a freshly created workflow, so - // no series carries an empty label value. - phaseUnknown = "Unknown" ) var ( @@ -142,9 +138,20 @@ func (c *Collector) collectWorkflows(ctx context.Context, ch chan<- prometheus.M for i := range workflows.Items { workflow := &workflows.Items[i] artifactType := ArtifactTypeOf(workflow) - mode := modeOf(workflow) - counts[key{workflow.Namespace, artifactType, mode, phaseLabel(workflow.Status.Phase)}]++ + 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 @@ -205,19 +212,3 @@ func ArtifactTypeOf(workflow *arcv1alpha1.ArtifactWorkflow) string { return UnknownArtifactType } - -func modeOf(workflow *arcv1alpha1.ArtifactWorkflow) string { - if workflow.Spec.Cron != nil { - return modeCron - } - - return modeSingle -} - -func phaseLabel(phase arcv1alpha1.WorkflowPhase) string { - if phase == arcv1alpha1.WorkflowUnknown { - return phaseUnknown - } - - return string(phase) -} diff --git a/pkg/metrics/collector_test.go b/pkg/metrics/collector_test.go index 4a80e51c..33c09bd8 100644 --- a/pkg/metrics/collector_test.go +++ b/pkg/metrics/collector_test.go @@ -316,7 +316,7 @@ var _ = Describe("Collector cache errors", func() { }) }) -var _ = Describe("LeaderRunnable", func() { +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), @@ -327,8 +327,10 @@ var _ = Describe("LeaderRunnable", func() { ctx, cancel := context.WithCancel(context.Background()) + Expect(collector.NeedLeaderElection()).To(BeTrue()) + done := make(chan error, 1) - go func() { done <- collector.LeaderRunnable().Start(ctx) }() + go func() { done <- collector.Start(ctx) }() Eventually(func() int { return testutil.CollectAndCount(collector) }).Should(BeNumerically(">", 0)) diff --git a/pkg/metrics/leader.go b/pkg/metrics/leader.go index 99d0434e..3f2aee52 100644 --- a/pkg/metrics/leader.go +++ b/pkg/metrics/leader.go @@ -9,39 +9,32 @@ import ( "sigs.k8s.io/controller-runtime/pkg/manager" ) -// leaderGate flips the collector on for as long as this replica holds -// leadership. When leader election is disabled the manager starts leader -// election runnables immediately, so single replica installs report normally. -type leaderGate struct { - collector *Collector -} - var ( - _ manager.Runnable = &leaderGate{} - _ manager.LeaderElectionRunnable = &leaderGate{} + _ manager.Runnable = &Collector{} + _ manager.LeaderElectionRunnable = &Collector{} ) -// LeaderRunnable returns a Runnable that must be added to the manager for the -// collector to report anything. -func (c *Collector) LeaderRunnable() manager.Runnable { - return &leaderGate{collector: c} -} - -// NeedLeaderElection implements manager.LeaderElectionRunnable. -func (g *leaderGate) NeedLeaderElection() bool { +// 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. 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 (g *leaderGate) Start(ctx context.Context) error { - g.collector.isLeader.Store(true) - defer g.collector.isLeader.Store(false) +// 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()