From 6c63f55790ff87e5f5bd70162210d07101cb330e Mon Sep 17 00:00:00 2001 From: Yolean gle-qa Date: Mon, 5 Oct 2026 13:19:23 +0000 Subject: [PATCH 1/6] Report in-place backend changes from EnsureBuckety as Events Drivers may change an existing backend resource whenever the spec licenses it (kadm adds partitions and alters topic configs on every reconcile and recheck). That is intended, but until now it left no trace: an adopted topic whose spec disagrees with the broker gets modified silently. EnsureRequest gains an optional OnChange callback that drivers call once per acknowledged in-place change, plus the Buckety's identity so drivers can label their own metrics. The reconciler turns each change into a Normal Event (old -> new) and a structured log line. Events are not transition-gated: a change that repeats every recheck should show up repeatedly. Drivers that never call OnChange are unaffected. Co-Authored-By: Claude Opus 5.5 --- pkg/controller/buckety/observability_test.go | 137 +++++++++++++++++++ pkg/controller/buckety/reconciler.go | 35 ++++- pkg/drivers/registry/registry.go | 28 ++++ 3 files changed, 198 insertions(+), 2 deletions(-) create mode 100644 pkg/controller/buckety/observability_test.go diff --git a/pkg/controller/buckety/observability_test.go b/pkg/controller/buckety/observability_test.go new file mode 100644 index 0000000..eacf939 --- /dev/null +++ b/pkg/controller/buckety/observability_test.go @@ -0,0 +1,137 @@ +package buckety + +import ( + "context" + "strings" + "testing" + + 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 ®istry.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) + } + } +} diff --git a/pkg/controller/buckety/reconciler.go b/pkg/controller/buckety/reconciler.go index 7725693..ffae7fb 100644 --- a/pkg/controller/buckety/reconciler.go +++ b/pkg/controller/buckety/reconciler.go @@ -259,8 +259,13 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu // Reconcile the backend resource itself. status.Set(&bky.Status.Conditions, "Reconciling", metav1.ConditionTrue, "Ensuring", "calling driver.EnsureBuckety", bky.Generation) if err := backend.Driver.EnsureBuckety(ctx, registry.EnsureRequest{ - Name: bky.Status.BackendResourceName, - Parameters: effective, + Name: bky.Status.BackendResourceName, + Parameters: effective, + Namespace: bky.Namespace, + BucketyName: bky.Name, + OnChange: func(c registry.Change) { + r.reportChange(ctx, &bky, c) + }, }); err != nil { if registry.IsParameterDrift(err) { status.Event(r.Recorder, &bky, baseBky.Status.Conditions, "ParameterDrift", metav1.ConditionTrue, "Unreconcilable", @@ -529,6 +534,32 @@ func (r *Reconciler) reconcileImplicitAccess(ctx context.Context, bky *bucketyv1 return nil } +// reportChange logs and records as an Event one in-place change a +// driver made to the backend resource (registry.EnsureRequest +// OnChange). Every call is a change the backend acknowledged, so +// unlike condition events these are not gated on a transition: an +// alter that repeats on every recheck (a broker normalising a value +// differently from the spec) shows up as a repeating Event, which +// is the point. +func (r *Reconciler) reportChange(ctx context.Context, bky *bucketyv1.Buckety, c registry.Change) { + old := c.Old + if old == "" { + old = "(unset)" + } + log.FromContext(ctx).Info("backend resource changed in place", + "buckety", types.NamespacedName{Namespace: bky.Namespace, Name: bky.Name}, + "backend", bky.Spec.Backend, + "backendResource", bky.Status.BackendResourceName, + "reason", c.Reason, + "parameter", c.Parameter, + "old", c.Old, + "new", c.New) + if r.Recorder != nil { + r.Recorder.Event(bky, corev1.EventTypeNormal, c.Reason, + fmt.Sprintf("%s: %s -> %s on %q", c.Parameter, old, c.New, bky.Status.BackendResourceName)) + } +} + func (r *Reconciler) surfaceBackendUnavailable(ctx context.Context, bky, base *bucketyv1.Buckety, reason, msg string) (ctrl.Result, error) { status.Event(r.Recorder, bky, base.Status.Conditions, "BackendUnavailable", metav1.ConditionTrue, reason, corev1.EventTypeWarning, "BackendUnavailable", msg) diff --git a/pkg/drivers/registry/registry.go b/pkg/drivers/registry/registry.go index f334138..c54955d 100644 --- a/pkg/drivers/registry/registry.go +++ b/pkg/drivers/registry/registry.go @@ -118,6 +118,34 @@ type EnsureRequest struct { // merged under spec.parameters, declared templated keys // resolved). Parameters map[string]string + // Namespace and BucketyName identify the Buckety this request + // reconciles. Drivers use them only to label what they report + // (metrics); the backend resource is always Name. + Namespace string + BucketyName string + // OnChange, when non-nil, is called once per in-place + // modification EnsureBuckety made to a backend resource that + // already existed, after the backend acknowledged it. It is not + // called when the resource is created, nor for a value that + // already matched. The reconciler turns each call into an Event + // and a log line, so changes the spec licenses (kadm partition + // increases, topic config alters) stay visible even though they + // need no human decision. + OnChange func(Change) +} + +// Change describes one in-place modification reported through +// EnsureRequest.OnChange. +type Change struct { + // Reason is a CamelCase Event reason from a small fixed set per + // driver (kadm: PartitionsAdded, TopicConfigChanged). + Reason string + // Parameter is the spec.parameters key whose value was applied + // (partitions, config.retention.ms, ...). + Parameter string + // Old is the backend's value before the change, "" when the + // backend reported none. New is the value now applied. + Old, New string } // DeleteRequest carries what DeleteBuckety needs to tear down the From 72ee0108ea82a24f5510d4fa6441137f3ebb728e Mon Sep 17 00:00:00 2001 From: Yolean gle-qa Date: Mon, 5 Oct 2026 13:20:59 +0000 Subject: [PATCH 2/6] kadm: count and report partition increases and topic config alters alignTopic adds partitions when the Buckety asks for more than the topic has, and sets every config.* key whose broker value differs, on every reconcile and recheck. That behaviour stays. What changes is that each such modification is now observable, so operators can alert when the controller touches an adopted topic: - buckety_kadm_partitions_added_total{namespace,buckety,topic}, by the number of partitions added - buckety_kadm_config_changes_total{namespace,buckety,topic,key}, once per key set - a registry.Change per modification (PartitionsAdded, TopicConfigChanged) for the reconciler's Events and logs All of these move only after the broker acknowledged the change. kadm reports per-topic CreatePartitions rejections in the response rather than the error, and alignTopic has always carried on past them; it still does (no reconcile behaviour change), but a rejection is logged instead of counted. Labels are identities only, never values. The broker mutations go through a two-method interface satisfied by *kadm.Client so unit tests can drive alignTopic without a broker. Co-Authored-By: Claude Opus 5.5 --- go.mod | 6 +- pkg/drivers/kadm/driver.go | 65 ++++++++++- pkg/drivers/kadm/driver_test.go | 192 ++++++++++++++++++++++++++++++++ pkg/drivers/kadm/metrics.go | 35 ++++++ 4 files changed, 290 insertions(+), 8 deletions(-) create mode 100644 pkg/drivers/kadm/metrics.go diff --git a/go.mod b/go.mod index 1e38991..68e9741 100644 --- a/go.mod +++ b/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 diff --git a/pkg/drivers/kadm/driver.go b/pkg/drivers/kadm/driver.go index c81adf5..21eeb20 100644 --- a/pkg/drivers/kadm/driver.go +++ b/pkg/drivers/kadm/driver.go @@ -13,10 +13,12 @@ import ( "errors" "fmt" "regexp" + "sort" "strconv" "strings" "github.com/Yolean/buckety-controller/pkg/drivers/registry" + "github.com/go-logr/logr" "github.com/twmb/franz-go/pkg/kadm" "github.com/twmb/franz-go/pkg/kerr" "github.com/twmb/franz-go/pkg/kgo" @@ -53,10 +55,12 @@ func factory(raw json.RawMessage) (registry.Driver, error) { if err != nil { return nil, fmt.Errorf("kadm: build kgo client: %w", err) } + ac := kadm.NewClient(kc) return &Driver{ cfg: cfg, kclient: kc, - aclient: kadm.NewClient(kc), + aclient: ac, + mutator: ac, }, nil } @@ -65,6 +69,16 @@ type Driver struct { cfg *Config kclient *kgo.Client aclient *kadm.Client + // mutator is aclient, narrowed to the calls alignTopic changes + // an existing topic with, so unit tests can observe them + // without a broker. + mutator topicMutator +} + +// topicMutator is the subset of *kadm.Client that alignTopic uses. +type topicMutator interface { + UpdatePartitions(ctx context.Context, set int, topics ...string) (kadm.CreatePartitionsResponses, error) + AlterTopicConfigs(ctx context.Context, configs []kadm.AlterConfig, topics ...string) (kadm.AlterConfigsResponses, error) } func (d *Driver) Name() string { return DriverName } @@ -127,7 +141,7 @@ func (d *Driver) EnsureBuckety(ctx context.Context, req registry.EnsureRequest) if existing == nil { return d.createTopic(ctx, req.Name, wantParts, wantRF, wantCfgs) } - return d.alignTopic(ctx, req.Name, existing, wantParts, wantRF, wantCfgs) + return d.alignTopic(ctx, req, existing, wantParts, wantRF, wantCfgs) } // DeleteBuckety removes the topic. Idempotent on NotFound. @@ -356,7 +370,12 @@ func (d *Driver) createTopic(ctx context.Context, name string, parts int32, rf i return nil } -func (d *Driver) alignTopic(ctx context.Context, name string, existing *topicView, wantParts int32, wantRF int16, wantCfgs map[string]*string) error { +// alignTopic brings an existing topic to the requested shape. Every +// change it makes is counted (metrics.go) and reported through +// req.OnChange once the broker has acknowledged it, so partition +// increases and config alters on adopted topics are observable. +func (d *Driver) alignTopic(ctx context.Context, req registry.EnsureRequest, existing *topicView, wantParts int32, wantRF int16, wantCfgs map[string]*string) error { + name := req.Name if wantParts > 0 && existing.partitions != wantParts { if wantParts < existing.partitions { return ®istry.ErrParameterDrift{ @@ -364,9 +383,26 @@ func (d *Driver) alignTopic(ctx context.Context, name string, existing *topicVie } } // Add partitions to reach the requested count. - if _, err := d.aclient.UpdatePartitions(ctx, int(wantParts), name); err != nil { + resp, err := d.mutator.UpdatePartitions(ctx, int(wantParts), name) + if err != nil { return fmt.Errorf("kadm: update partitions %q: %w", name, err) } + // kadm reports per-topic rejections (authorization, broker + // policy) in resp, not err. This loop has always carried on + // regardless; it still does, but a rejected increase is not + // a change, so it is logged instead of counted. + if r, ok := resp[name]; ok && r.Err != nil { + logr.FromContextOrDiscard(ctx).Error(r.Err, "kadm: partition increase rejected by broker", + "topic", name, "current", existing.partitions, "requested", wantParts, "brokerMessage", r.ErrMessage) + } else { + partitionsAdded.WithLabelValues(req.Namespace, req.BucketyName, name).Add(float64(wantParts - existing.partitions)) + report(req, registry.Change{ + Reason: ReasonPartitionsAdded, + Parameter: "partitions", + Old: strconv.Itoa(int(existing.partitions)), + New: strconv.Itoa(int(wantParts)), + }) + } } if wantRF > 0 && existing.rf != 0 && existing.rf != wantRF { return ®istry.ErrParameterDrift{ @@ -386,7 +422,7 @@ func (d *Driver) alignTopic(ctx context.Context, name string, existing *topicVie } } if len(alters) > 0 { - resp, err := d.aclient.AlterTopicConfigs(ctx, alters, name) + resp, err := d.mutator.AlterTopicConfigs(ctx, alters, name) if err != nil { return fmt.Errorf("kadm: alter configs %q: %w", name, err) } @@ -395,10 +431,29 @@ func (d *Driver) alignTopic(ctx context.Context, name string, existing *topicVie return fmt.Errorf("kadm: alter configs %q: %w", name, r.Err) } } + // One incremental alter per topic: the broker applies all + // keys or none, so success means every key changed. + sort.Slice(alters, func(i, j int) bool { return alters[i].Name < alters[j].Name }) + for _, a := range alters { + configChanges.WithLabelValues(req.Namespace, req.BucketyName, name, a.Name).Inc() + report(req, registry.Change{ + Reason: ReasonTopicConfigChanged, + Parameter: "config." + a.Name, + Old: existing.configs[a.Name], + New: *a.Value, + }) + } } return nil } +// report calls req.OnChange when the caller asked for changes. +func report(req registry.EnsureRequest, c registry.Change) { + if req.OnChange != nil { + req.OnChange(c) + } +} + // translateParameters splits spec.parameters into (partitions, // replicationFactor, config-keys). config.* keys are passed // through with the "config." prefix stripped (Kafka's wire-level diff --git a/pkg/drivers/kadm/driver_test.go b/pkg/drivers/kadm/driver_test.go index c34ec2e..4ebb1bb 100644 --- a/pkg/drivers/kadm/driver_test.go +++ b/pkg/drivers/kadm/driver_test.go @@ -1,10 +1,20 @@ package kadm import ( + "context" "encoding/json" + "errors" + "fmt" "os" "strings" "testing" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/twmb/franz-go/pkg/kadm" + "github.com/twmb/franz-go/pkg/kerr" + + "github.com/Yolean/buckety-controller/pkg/drivers/registry" ) // Validation methods never touch the broker clients, so a zero @@ -163,3 +173,185 @@ func TestParametersSchemaInSync(t *testing.T) { } } } + +// fakeMutator records the broker mutations alignTopic issues and +// answers with the configured errors. +type fakeMutator struct { + updates []int + alters [][]kadm.AlterConfig + updateErr error // request-level + topicErr error // per-topic, in the response + alterErr error // per-resource, in the response +} + +func (f *fakeMutator) UpdatePartitions(_ context.Context, set int, topics ...string) (kadm.CreatePartitionsResponses, error) { + if f.updateErr != nil { + return nil, f.updateErr + } + f.updates = append(f.updates, set) + rs := kadm.CreatePartitionsResponses{} + for _, t := range topics { + rs[t] = kadm.CreatePartitionsResponse{Topic: t, Err: f.topicErr} + } + return rs, nil +} + +func (f *fakeMutator) AlterTopicConfigs(_ context.Context, cfgs []kadm.AlterConfig, topics ...string) (kadm.AlterConfigsResponses, error) { + f.alters = append(f.alters, cfgs) + var rs kadm.AlterConfigsResponses + for _, t := range topics { + rs = append(rs, kadm.AlterConfigsResponse{Name: t, Err: f.alterErr}) + } + return rs, nil +} + +func counterValue(t *testing.T, c prometheus.Counter) float64 { + t.Helper() + var m dto.Metric + if err := c.Write(&m); err != nil { + t.Fatal(err) + } + return m.GetCounter().GetValue() +} + +func strp(s string) *string { return &s } + +// alignCase runs alignTopic for Buckety ns/orders on topic and +// returns the changes it reported. +func alignCase(t *testing.T, m *fakeMutator, ns, topic string, existing *topicView, wantParts int32, wantRF int16, wantCfgs map[string]*string) ([]registry.Change, error) { + t.Helper() + var got []registry.Change + drv := &Driver{mutator: m} + err := drv.alignTopic(context.Background(), registry.EnsureRequest{ + Name: topic, Namespace: ns, BucketyName: "orders", + OnChange: func(c registry.Change) { got = append(got, c) }, + }, existing, wantParts, wantRF, wantCfgs) + return got, err +} + +// Partition increases and config alters move the counters by what +// was actually changed and report one Change each; a topic already +// in shape moves nothing and reports nothing. +func TestAlignTopicCountsChanges(t *testing.T) { + const ns, topic = "align-counts", "align-counts.orders" + parts := partitionsAdded.WithLabelValues(ns, "orders", topic) + retention := configChanges.WithLabelValues(ns, "orders", topic, "retention.ms") + cleanup := configChanges.WithLabelValues(ns, "orders", topic, "cleanup.policy") + segment := configChanges.WithLabelValues(ns, "orders", topic, "segment.bytes") + + m := &fakeMutator{} + existing := &topicView{partitions: 3, rf: 3, configs: map[string]string{ + "retention.ms": "604800000", + "cleanup.policy": "delete", + "segment.bytes": "1073741824", + }} + changes, err := alignCase(t, m, ns, topic, existing, 6, 3, map[string]*string{ + "retention.ms": strp("86400000"), + "cleanup.policy": strp("delete"), // equal: not altered, not counted + "segment.bytes": strp("536870912"), + }) + if err != nil { + t.Fatal(err) + } + if len(m.updates) != 1 || m.updates[0] != 6 { + t.Errorf("UpdatePartitions calls %v, want [6]", m.updates) + } + if got := counterValue(t, parts); got != 3 { + t.Errorf("partitions added = %v, want 3", got) + } + if got := counterValue(t, retention); got != 1 { + t.Errorf("retention.ms changes = %v, want 1", got) + } + if got := counterValue(t, segment); got != 1 { + t.Errorf("segment.bytes changes = %v, want 1", got) + } + if got := counterValue(t, cleanup); got != 0 { + t.Errorf("cleanup.policy (equal) changes = %v, want 0", got) + } + want := []registry.Change{ + {Reason: ReasonPartitionsAdded, Parameter: "partitions", Old: "3", New: "6"}, + {Reason: ReasonTopicConfigChanged, Parameter: "config.retention.ms", Old: "604800000", New: "86400000"}, + {Reason: ReasonTopicConfigChanged, Parameter: "config.segment.bytes", Old: "1073741824", New: "536870912"}, + } + if fmt.Sprint(changes) != fmt.Sprint(want) { + t.Errorf("changes\n got %v\nwant %v", changes, want) + } + + // Next reconcile sees the topic in shape: no calls, no counts. + inShape := &topicView{partitions: 6, rf: 3, configs: map[string]string{ + "retention.ms": "86400000", + "cleanup.policy": "delete", + "segment.bytes": "536870912", + }} + changes, err = alignCase(t, m, ns, topic, inShape, 6, 3, map[string]*string{ + "retention.ms": strp("86400000"), + "cleanup.policy": strp("delete"), + "segment.bytes": strp("536870912"), + }) + if err != nil { + t.Fatal(err) + } + if len(changes) != 0 || len(m.updates) != 1 || len(m.alters) != 1 { + t.Errorf("no-op align: changes=%v updates=%v alters=%d", changes, m.updates, len(m.alters)) + } + if counterValue(t, parts) != 3 || counterValue(t, retention) != 1 || counterValue(t, segment) != 1 { + t.Error("no-op align moved a counter") + } + + // A key the broker does not report counts as a change from unset. + changes, err = alignCase(t, m, ns, topic, &topicView{partitions: 6, rf: 3, configs: map[string]string{}}, 0, 0, + map[string]*string{"cleanup.policy": strp("compact")}) + if err != nil { + t.Fatal(err) + } + if got := counterValue(t, cleanup); got != 1 { + t.Errorf("cleanup.policy changes = %v, want 1", got) + } + if len(changes) != 1 || changes[0].Old != "" || changes[0].New != "compact" { + t.Errorf("changes %v, want one from unset to compact", changes) + } +} + +// Nothing is counted or reported unless the broker accepted the +// change, and a shrink request stays ParameterDrift. +func TestAlignTopicCountsOnlyAcknowledgedChanges(t *testing.T) { + const ns, topic = "align-failures", "align-failures.orders" + parts := partitionsAdded.WithLabelValues(ns, "orders", topic) + retention := configChanges.WithLabelValues(ns, "orders", topic, "retention.ms") + existing := &topicView{partitions: 3, rf: 3, configs: map[string]string{"retention.ms": "1"}} + want := map[string]*string{"retention.ms": strp("2")} + + cases := []struct { + name string + m *fakeMutator + parts int32 + cfgs map[string]*string + wantErr string + }{ + {"request fails", &fakeMutator{updateErr: errors.New("broker down")}, 6, nil, "update partitions"}, + {"topic rejected", &fakeMutator{topicErr: kerr.PolicyViolation}, 6, nil, ""}, + {"alter rejected", &fakeMutator{alterErr: kerr.InvalidConfig}, 0, want, "alter configs"}, + {"shrink", &fakeMutator{}, 2, want, "cannot shrink"}, + } + for _, c := range cases { + changes, err := alignCase(t, c.m, ns, topic, existing, c.parts, 3, c.cfgs) + if c.wantErr == "" && err != nil { + t.Errorf("%s: %v", c.name, err) + } + if c.wantErr != "" && (err == nil || !strings.Contains(err.Error(), c.wantErr)) { + t.Errorf("%s: err %v, want %q", c.name, err, c.wantErr) + } + if len(changes) != 0 { + t.Errorf("%s: reported %v", c.name, changes) + } + } + if _, err := alignCase(t, &fakeMutator{}, ns, topic, existing, 2, 3, nil); !registry.IsParameterDrift(err) { + t.Errorf("shrink: %v, want ParameterDrift", err) + } + if got := counterValue(t, parts); got != 0 { + t.Errorf("partitions added = %v, want 0", got) + } + if got := counterValue(t, retention); got != 0 { + t.Errorf("retention.ms changes = %v, want 0", got) + } +} diff --git a/pkg/drivers/kadm/metrics.go b/pkg/drivers/kadm/metrics.go new file mode 100644 index 0000000..7707323 --- /dev/null +++ b/pkg/drivers/kadm/metrics.go @@ -0,0 +1,35 @@ +package kadm + +import ( + "github.com/prometheus/client_golang/prometheus" + crmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" +) + +// Event reasons for registry.Change values this driver reports. +const ( + ReasonPartitionsAdded = "PartitionsAdded" + ReasonTopicConfigChanged = "TopicConfigChanged" +) + +// Counters for the changes alignTopic makes to existing topics. +// Both move only after the broker acknowledged the change, never +// for values that already matched. Labels are identities, never +// values: one series per Buckety (and per managed config key), so +// cardinality follows the number of Bucketies, not their history. +// Registered with controller-runtime's registry, which is what +// --metrics-addr serves. +var ( + partitionsAdded = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "buckety_kadm_partitions_added_total", + Help: "Partitions the kadm driver added to existing topics because the Buckety requested more than the topic had.", + }, []string{"namespace", "buckety", "topic"}) + + configChanges = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "buckety_kadm_config_changes_total", + Help: "Topic config keys the kadm driver set on existing topics because the broker's value differed from the Buckety's config.* parameter.", + }, []string{"namespace", "buckety", "topic", "key"}) +) + +func init() { + crmetrics.Registry.MustRegister(partitionsAdded, configChanges) +} From 6f941c9f01210d5d70acda11c0d449debbdae5b7 Mon Sep 17 00:00:00 2001 From: Yolean gle-qa Date: Mon, 5 Oct 2026 13:22:12 +0000 Subject: [PATCH 3/6] Export buckety_parameter_drift gauge from the ParameterDrift condition A partition shrink request (or any driver's ErrParameterDrift) already sets ParameterDrift=True and Ready=False, but only kubectl shows it. Operators adopting existing topics need to alert on it. buckety_parameter_drift{namespace,buckety,backend} is set at the end of every reconcile from the condition the pass leaves behind, so it is driver-independent and agrees with status whichever branch returned: 1 while True, 0 otherwise. The series is dropped when the finalizer is released and on a NotFound reconcile, so deleted Bucketies do not keep stale series. spec.backend is immutable, so a Buckety's label set is fixed. Reconcile decisions are unchanged. Co-Authored-By: Claude Opus 5.5 --- pkg/controller/buckety/metrics.go | 46 ++++++++++ pkg/controller/buckety/observability_test.go | 91 ++++++++++++++++++++ pkg/controller/buckety/reconciler.go | 4 + 3 files changed, 141 insertions(+) create mode 100644 pkg/controller/buckety/metrics.go diff --git a/pkg/controller/buckety/metrics.go b/pkg/controller/buckety/metrics.go new file mode 100644 index 0000000..9e14de5 --- /dev/null +++ b/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}) +} diff --git a/pkg/controller/buckety/observability_test.go b/pkg/controller/buckety/observability_test.go index eacf939..6527a08 100644 --- a/pkg/controller/buckety/observability_test.go +++ b/pkg/controller/buckety/observability_test.go @@ -5,6 +5,8 @@ import ( "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" @@ -135,3 +137,92 @@ func TestDriverChangesBecomeEvents(t *testing.T) { } } } + +// 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") + } +} diff --git a/pkg/controller/buckety/reconciler.go b/pkg/controller/buckety/reconciler.go index ffae7fb..6c6ae88 100644 --- a/pkg/controller/buckety/reconciler.go +++ b/pkg/controller/buckety/reconciler.go @@ -103,10 +103,14 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu var bky bucketyv1.Buckety if err := r.Get(ctx, req.NamespacedName, &bky); err != nil { if apierrors.IsNotFound(err) { + forgetDrift(req.Namespace, req.Name) return ctrl.Result{}, nil } return ctrl.Result{}, err } + // The drift gauge mirrors whatever ParameterDrift condition this + // pass leaves on the Buckety, whichever branch returns. + defer recordDrift(&bky) // Resolve the backend up front; nearly every branch needs it. // The driver behind a backend name is part of the sticky From 0e0036c20b0225ef333969e61d60ea90616b5787 Mon Sep 17 00:00:00 2001 From: Yolean gle-qa Date: Mon, 5 Oct 2026 13:23:26 +0000 Subject: [PATCH 4/6] kadm: export the change counters at 0 before the first change increase() and rate() need an earlier sample to see an increment. A counter series created by its own first increment has none, so an alert on increase(...[1h]) > 0 would never fire for the first partition increase or config alter a topic gets after a controller start. Touching the series for every aligned topic and managed config key makes later changes visible. Cardinality is unchanged: these are the same series the first change would create. Co-Authored-By: Claude Opus 5.5 --- pkg/drivers/kadm/driver.go | 8 ++++++++ pkg/drivers/kadm/driver_test.go | 31 +++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/pkg/drivers/kadm/driver.go b/pkg/drivers/kadm/driver.go index 21eeb20..f60329e 100644 --- a/pkg/drivers/kadm/driver.go +++ b/pkg/drivers/kadm/driver.go @@ -376,6 +376,14 @@ func (d *Driver) createTopic(ctx context.Context, name string, parts int32, rf i // increases and config alters on adopted topics are observable. func (d *Driver) alignTopic(ctx context.Context, req registry.EnsureRequest, existing *topicView, wantParts int32, wantRF int16, wantCfgs map[string]*string) error { name := req.Name + // Export the counters at 0 for every topic and managed key this + // driver aligns, so that rate()/increase() see a later change as + // an increase. A series born at its first increment has no + // earlier sample, and increase() reads that change as nothing. + partitionsAdded.WithLabelValues(req.Namespace, req.BucketyName, name) + for k := range wantCfgs { + configChanges.WithLabelValues(req.Namespace, req.BucketyName, name, k) + } if wantParts > 0 && existing.partitions != wantParts { if wantParts < existing.partitions { return ®istry.ErrParameterDrift{ diff --git a/pkg/drivers/kadm/driver_test.go b/pkg/drivers/kadm/driver_test.go index 4ebb1bb..f290308 100644 --- a/pkg/drivers/kadm/driver_test.go +++ b/pkg/drivers/kadm/driver_test.go @@ -13,6 +13,7 @@ import ( dto "github.com/prometheus/client_model/go" "github.com/twmb/franz-go/pkg/kadm" "github.com/twmb/franz-go/pkg/kerr" + crmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" "github.com/Yolean/buckety-controller/pkg/drivers/registry" ) @@ -355,3 +356,33 @@ func TestAlignTopicCountsOnlyAcknowledgedChanges(t *testing.T) { t.Errorf("retention.ms changes = %v, want 0", got) } } + +// An in-shape topic still exports its counters, at 0, so a change +// on a later reconcile reads as an increase. +func TestAlignTopicExportsZeroSeries(t *testing.T) { + const ns, topic = "align-zero", "align-zero.orders" + if _, err := alignCase(t, &fakeMutator{}, ns, topic, + &topicView{partitions: 3, rf: 3, configs: map[string]string{"retention.ms": "1"}}, 3, 3, + map[string]*string{"retention.ms": strp("1")}); err != nil { + t.Fatal(err) + } + families, err := crmetrics.Registry.Gather() + if err != nil { + t.Fatal(err) + } + seen := map[string]float64{} + for _, f := range families { + for _, m := range f.GetMetric() { + for _, l := range m.GetLabel() { + if l.GetName() == "namespace" && l.GetValue() == ns { + seen[f.GetName()] = m.GetCounter().GetValue() + } + } + } + } + for _, name := range []string{"buckety_kadm_partitions_added_total", "buckety_kadm_config_changes_total"} { + if v, ok := seen[name]; !ok || v != 0 { + t.Errorf("%s for %s: value %v present %v, want 0 present", name, ns, v, ok) + } + } +} From f3d04416c9ff6e2964f941ed11cfbd1a35d6ff6f Mon Sep 17 00:00:00 2001 From: Yolean gle-qa Date: Mon, 5 Oct 2026 13:25:11 +0000 Subject: [PATCH 5/6] README: document metrics, change Events and suggested alerts Describes the drift gauge and the two kadm change counters under Troubleshooting, next to the conditions and Events they complement, with a PrometheusRule to start from. The counter alerts are not a bare increase(...[1h]) > 0: the first reconcile of an adopted Buckety can create a counter series and increment it in the same pass, and increase() needs an earlier sample to see that, so the first change to a newly adopted topic would never fire. The "> 0 unless ... offset 1h" term covers series that are new within the window. Checked with promqltest: an adopted topic's first change alerts, a later change on a zero series alerts, an unchanged nonzero series does not. Co-Authored-By: Claude Opus 5.5 --- README.md | 78 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/README.md b/README.md index 17bfe90..40bc894 100644 --- a/README.md +++ b/README.md @@ -468,6 +468,84 @@ transition, so `kubectl describe buckety/` (or `bucketyaccess/`) 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.`. | + +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 +`: -> on ""`, 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: From 2a8199e3e6d394d9a9999e3a3c205088a48e858f Mon Sep 17 00:00:00 2001 From: Yolean gle-qa Date: Mon, 5 Oct 2026 13:42:12 +0000 Subject: [PATCH 6/6] release: pin the change metrics build scripts/bump-release.sh at the head of the metrics changes, with the go.mod toolchain (go1.27.1). The same host reproduced main's pin, 20261002T051036Z@sha256:8c3c4937, byte for byte before this bump. Co-Authored-By: Claude Opus 5.5 --- deploy/kustomize/release/kustomization.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/deploy/kustomize/release/kustomization.yaml b/deploy/kustomize/release/kustomization.yaml index 2b7dce0..49fcb4f 100644 --- a/deploy/kustomize/release/kustomization.yaml +++ b/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