Expose metrics for topics configuration changes

README.md

@@ -468,6 +468,84 @@ transition, so `kubectl describe buckety/<name>` (or
 `bucketyaccess/<name>`) shows the history even after a condition
 clears.
 
+### Metrics
+
+The controller serves Prometheus metrics on `--metrics-addr`
+(default `:8080`, container port `metrics`): controller-runtime's
+standard workqueue and reconcile metrics, plus these. Labels name
+resources, never parameter values, so there is one series per
+`Buckety` (per config key for the config counter).
+
+| Metric | Type | Labels | Meaning |
+| --- | --- | --- | --- |
+| `buckety_parameter_drift` | gauge | `namespace`, `buckety`, `backend` | `1` while the `Buckety` has `ParameterDrift=True` (any driver), `0` otherwise; removed when the `Buckety` is deleted. |
+| `buckety_kadm_partitions_added_total` | counter | `namespace`, `buckety`, `topic` | Partitions the `kadm` driver added to an existing topic because `parameters.partitions` asked for more than the topic had. |
+| `buckety_kadm_config_changes_total` | counter | `namespace`, `buckety`, `topic`, `key` | Topic config keys (Kafka names, e.g. `retention.ms`) the `kadm` driver set on an existing topic because the broker's value differed from `config.<key>`. |
+
+The `kadm` driver reconciles partition increases and `config.*`
+differences in place, on every reconcile and periodic re-check;
+that matters most for topics claimed with `adoption: Adopt`, whose
+first reconcile can change a production topic. The counters move
+only after the broker accepted the change and never for values
+that already match, so any increase means the controller modified
+a topic. Each change is also an Event on the `Buckety`, reason
+`PartitionsAdded` or `TopicConfigChanged`, with the message
+`<parameter>: <old> -> <new> on "<topic>"`, and an info log line
+`backend resource changed in place` carrying `parameter`, `old` and
+`new`. A config counter that keeps rising for the same key means
+the broker reports the value differently from how the spec writes
+it, so every re-check sets it again; write the value in the
+broker's form.
+
+Suggested alerts. The counters exist at `0` from the first
+reconcile of a topic, so `increase()` catches later changes. The
+first reconcile of a newly adopted `Buckety` (or the
+first after a controller restart) can create a series and
+increment it at once, and `increase()` needs an earlier sample to
+see that; the `unless ... offset 1h` term covers series that are
+new within the window and already above zero.
+
+```yaml
+apiVersion: monitoring.coreos.com/v1
+kind: PrometheusRule
+metadata:
+  name: buckety-controller
+spec:
+  groups:
+  - name: buckety-controller
+    rules:
+    - alert: BucketyKafkaPartitionsAdded
+      expr: |
+        increase(buckety_kadm_partitions_added_total[1h]) > 0
+        or
+        (buckety_kadm_partitions_added_total > 0 unless buckety_kadm_partitions_added_total offset 1h)
+      labels:
+        severity: info
+      annotations:
+        summary: buckety-controller added partitions to topic {{ $labels.topic }} ({{ $labels.namespace }}/{{ $labels.buckety }})
+    - alert: BucketyKafkaTopicConfigChanged
+      expr: |
+        increase(buckety_kadm_config_changes_total[1h]) > 0
+        or
+        (buckety_kadm_config_changes_total > 0 unless buckety_kadm_config_changes_total offset 1h)
+      labels:
+        severity: info
+      annotations:
+        summary: buckety-controller changed {{ $labels.key }} on topic {{ $labels.topic }} ({{ $labels.namespace }}/{{ $labels.buckety }})
+    - alert: BucketyParameterDrift
+      expr: buckety_parameter_drift == 1
+      for: 10m
+      labels:
+        severity: warning
+      annotations:
+        summary: Buckety {{ $labels.namespace }}/{{ $labels.buckety }} has ParameterDrift; see kubectl describe buckety
+```
+
+The counters are per controller process: they restart from zero
+with the Pod, and a deleted `Buckety`'s counter series stay until
+the next restart. The Events and the log line are the exact record
+of what changed and from what.
+
 ### Controller won't start
 
 Strict YAML decode catches typos:

deploy/kustomize/release/kustomization.yaml

@@ -7,7 +7,7 @@ kind: Kustomization
 resources:
 - ../base
 images:
-- digest: sha256:8c3c493780935b8084e2241722f60605464f2cf3d70f83a9c68c2edd43f222d6
+- digest: sha256:7c66ed62e57235805405323fce20fa1a1c35e01d6aa6045cebe88d9f0e6b4655
   name: ghcr.io/yolean/buckety-controller
   newName: ghcr.io/yolean/buckety-controller
-  newTag: 20261002T051036Z
+  newTag: 20261005T134152Z

go.mod

@@ -11,7 +11,10 @@ require (
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.24
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.103.3
 	github.com/aws/smithy-go v1.27.2
+	github.com/go-logr/logr v1.4.3
 	github.com/go-sql-driver/mysql v1.10.1
+	github.com/prometheus/client_golang v1.23.2
+	github.com/prometheus/client_model v0.6.2
 	github.com/twmb/franz-go v1.21.2
 	github.com/twmb/franz-go/pkg/kadm v1.18.0
 	github.com/twmb/franz-go/pkg/kmsg v1.13.1
@@ -57,7 +60,6 @@ require (
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.9.0 // indirect
 	github.com/go-jose/go-jose/v4 v4.1.4 // indirect
-	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-logr/zapr v1.3.0 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
@@ -78,8 +80,6 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.26 // indirect
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // 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.67.5 // indirect
 	github.com/prometheus/procfs v0.19.2 // indirect
 	github.com/spf13/pflag v1.0.10 // indirect

pkg/controller/buckety/metrics.go

@@ -0,0 +1,46 @@
+package buckety
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	crmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
+
+	bucketyv1 "github.com/Yolean/buckety-controller/pkg/api/v1alpha1"
+)
+
+// parameterDrift mirrors each Buckety's ParameterDrift condition:
+// 1 while it is True, 0 otherwise, series removed once the Buckety
+// is gone. Driver-independent: any driver's ErrParameterDrift sets
+// the condition. spec.backend is immutable, so the label set of a
+// Buckety never changes. Registered with controller-runtime's
+// registry, which is what --metrics-addr serves.
+var parameterDrift = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+	Name: "buckety_parameter_drift",
+	Help: "1 while the Buckety's ParameterDrift condition is True (backend state the driver cannot reconcile in place, e.g. a Kafka partition shrink request), 0 otherwise.",
+}, []string{"namespace", "buckety", "backend"})
+
+func init() {
+	crmetrics.Registry.MustRegister(parameterDrift)
+}
+
+// recordDrift sets the drift gauge from the conditions the
+// reconcile leaves on bky, or drops the series once the finalizer
+// is released and the Buckety is about to disappear.
+func recordDrift(bky *bucketyv1.Buckety) {
+	if !bky.DeletionTimestamp.IsZero() && !controllerutil.ContainsFinalizer(bky, bucketyv1.FinalizerCleanup) {
+		forgetDrift(bky.Namespace, bky.Name)
+		return
+	}
+	v := 0.0
+	if meta.IsStatusConditionTrue(bky.Status.Conditions, "ParameterDrift") {
+		v = 1
+	}
+	parameterDrift.WithLabelValues(bky.Namespace, bky.Name, bky.Spec.Backend).Set(v)
+}
+
+// forgetDrift removes the drift series of a Buckety that no longer
+// exists.
+func forgetDrift(namespace, name string) {
+	parameterDrift.DeletePartialMatch(prometheus.Labels{"namespace": namespace, "buckety": name})
+}

pkg/controller/buckety/observability_test.go

@@ -0,0 +1,228 @@
+package buckety
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/record"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	bucketyv1 "github.com/Yolean/buckety-controller/pkg/api/v1alpha1"
+	"github.com/Yolean/buckety-controller/pkg/config"
+	"github.com/Yolean/buckety-controller/pkg/drivers/registry"
+)
+
+// changingDriver is a provisioningDriver whose EnsureBuckety
+// reports the given changes through OnChange, or answers with
+// ParameterDrift while drift is set.
+type changingDriver struct {
+	provisioningDriver
+	changes []registry.Change
+	drift   bool
+	gotReq  registry.EnsureRequest
+}
+
+func (c *changingDriver) EnsureBuckety(_ context.Context, req registry.EnsureRequest) error {
+	c.gotReq = req
+	if c.drift {
+		return &registry.ErrParameterDrift{Reason: "partitions: current=6 requested=3"}
+	}
+	for _, ch := range c.changes {
+		if req.OnChange != nil {
+			req.OnChange(ch)
+		}
+	}
+	return nil
+}
+
+// observedBuckety is a Buckety past first reconcile (sticky fields
+// stamped) on backend "be", so Reconcile goes straight to Ensure.
+func observedBuckety(ns, name string) *bucketyv1.Buckety {
+	return &bucketyv1.Buckety{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name, Namespace: ns,
+			Finalizers: []string{bucketyv1.FinalizerCleanup},
+		},
+		Spec: bucketyv1.BucketySpec{Backend: "be", RetentionPolicy: bucketyv1.RetentionRetain},
+		Status: bucketyv1.BucketyStatus{
+			Backend: "be", Driver: "prov", BackendResourceName: ns + "." + name,
+		},
+	}
+}
+
+func newObservabilityReconciler(t *testing.T, drv registry.Driver, objs ...client.Object) (*Reconciler, client.Client, *record.FakeRecorder) {
+	t.Helper()
+	scheme := runtime.NewScheme()
+	if err := clientgoscheme.AddToScheme(scheme); err != nil {
+		t.Fatal(err)
+	}
+	if err := bucketyv1.AddToScheme(scheme); err != nil {
+		t.Fatal(err)
+	}
+	cl := fake.NewClientBuilder().WithScheme(scheme).
+		WithObjects(objs...).
+		WithStatusSubresource(&bucketyv1.Buckety{}, &bucketyv1.BucketyAccess{}).
+		Build()
+	rec := record.NewFakeRecorder(32)
+	r := &Reconciler{Client: cl, Scheme: scheme, Recorder: rec, Config: &config.Loaded{
+		Backends: map[string]config.Backend{"be": {Name: "be", Driver: drv}},
+	}}
+	return r, cl, rec
+}
+
+// drainEvents returns the events recorded so far.
+func drainEvents(rec *record.FakeRecorder) []string {
+	var out []string
+	for {
+		select {
+		case e := <-rec.Events:
+			out = append(out, e)
+		default:
+			return out
+		}
+	}
+}
+
+// Every change a driver reports becomes a Normal Event carrying the
+// old and new values, and the request identifies the Buckety so
+// drivers can label their own metrics.
+func TestDriverChangesBecomeEvents(t *testing.T) {
+	drv := &changingDriver{changes: []registry.Change{
+		{Reason: "PartitionsAdded", Parameter: "partitions", Old: "3", New: "6"},
+		{Reason: "TopicConfigChanged", Parameter: "config.retention.ms", Old: "604800000", New: "86400000"},
+		{Reason: "TopicConfigChanged", Parameter: "config.cleanup.policy", New: "compact"},
+	}}
+	r, _, rec := newObservabilityReconciler(t, drv, observedBuckety("t1", "orders"))
+	req := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: "t1", Name: "orders"}}
+	if _, err := r.Reconcile(context.Background(), req); err != nil {
+		t.Fatal(err)
+	}
+	if drv.gotReq.Namespace != "t1" || drv.gotReq.BucketyName != "orders" {
+		t.Errorf("EnsureRequest identity = %q/%q, want t1/orders", drv.gotReq.Namespace, drv.gotReq.BucketyName)
+	}
+	events := drainEvents(rec)
+	for _, want := range []string{
+		`Normal PartitionsAdded partitions: 3 -> 6 on "t1.orders"`,
+		`Normal TopicConfigChanged config.retention.ms: 604800000 -> 86400000 on "t1.orders"`,
+		`Normal TopicConfigChanged config.cleanup.policy: (unset) -> compact on "t1.orders"`,
+	} {
+		found := false
+		for _, e := range events {
+			if e == want {
+				found = true
+			}
+		}
+		if !found {
+			t.Errorf("missing event %q in %q", want, events)
+		}
+	}
+
+	// A no-op Ensure reports nothing and so records no change events.
+	drv.changes = nil
+	if _, err := r.Reconcile(context.Background(), req); err != nil {
+		t.Fatal(err)
+	}
+	for _, e := range drainEvents(rec) {
+		if strings.Contains(e, "PartitionsAdded") || strings.Contains(e, "TopicConfigChanged") {
+			t.Errorf("no-op reconcile recorded change event %q", e)
+		}
+	}
+}
+
+// driftSeries returns the drift gauge's value for ns/name, and
+// whether such a series exists at all.
+func driftSeries(t *testing.T, ns, name string) (float64, bool) {
+	t.Helper()
+	ch := make(chan prometheus.Metric, 64)
+	go func() { parameterDrift.Collect(ch); close(ch) }()
+	val, found := 0.0, false
+	for m := range ch {
+		var d dto.Metric
+		if err := m.Write(&d); err != nil {
+			t.Fatal(err)
+		}
+		labels := map[string]string{}
+		for _, l := range d.GetLabel() {
+			labels[l.GetName()] = l.GetValue()
+		}
+		if labels["namespace"] == ns && labels["buckety"] == name {
+			if labels["backend"] != "be" {
+				t.Errorf("drift series backend label = %q, want be", labels["backend"])
+			}
+			val, found = d.GetGauge().GetValue(), true
+		}
+	}
+	return val, found
+}
+
+// The drift gauge follows the ParameterDrift condition: 1 while the
+// driver answers ErrParameterDrift, 0 once Ensure succeeds again,
+// and gone with the Buckety.
+func TestParameterDriftGauge(t *testing.T) {
+	const ns = "drift-gauge"
+	ctx := context.Background()
+	drv := &changingDriver{drift: true}
+	r, cl, _ := newObservabilityReconciler(t, drv, observedBuckety(ns, "orders"))
+	key := types.NamespacedName{Namespace: ns, Name: "orders"}
+	req := reconcile.Request{NamespacedName: key}
+
+	if _, err := r.Reconcile(ctx, req); err != nil {
+		t.Fatal(err)
+	}
+	if v, ok := driftSeries(t, ns, "orders"); !ok || v != 1 {
+		t.Fatalf("drift gauge = %v (present %v), want 1", v, ok)
+	}
+	// Steady state while drifted.
+	if _, err := r.Reconcile(ctx, req); err != nil {
+		t.Fatal(err)
+	}
+	if v, _ := driftSeries(t, ns, "orders"); v != 1 {
+		t.Fatalf("drift gauge on repeat = %v, want 1", v)
+	}
+
+	// Resolved (spec adjusted to match the backend).
+	drv.drift = false
+	if _, err := r.Reconcile(ctx, req); err != nil {
+		t.Fatal(err)
+	}
+	if v, ok := driftSeries(t, ns, "orders"); !ok || v != 0 {
+		t.Fatalf("drift gauge after resolution = %v (present %v), want 0", v, ok)
+	}
+
+	// Drifted again, then deleted: the series goes with the Buckety.
+	drv.drift = true
+	if _, err := r.Reconcile(ctx, req); err != nil {
+		t.Fatal(err)
+	}
+	var bky bucketyv1.Buckety
+	if err := cl.Get(ctx, key, &bky); err != nil {
+		t.Fatal(err)
+	}
+	if err := cl.Delete(ctx, &bky); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := r.Reconcile(ctx, req); err != nil {
+		t.Fatal(err)
+	}
+	if v, ok := driftSeries(t, ns, "orders"); ok {
+		t.Errorf("drift series still present after finalizer release: %v", v)
+	}
+	// A late NotFound pass for a Buckety this process never saw
+	// drop its finalizer also clears any series.
+	parameterDrift.WithLabelValues(ns, "gone", "be").Set(1)
+	if _, err := r.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Namespace: ns, Name: "gone"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := driftSeries(t, ns, "gone"); ok {
+		t.Error("drift series survived a NotFound reconcile")
+	}
+}