From a2506e0c08c373cef001009941461c09ccf69bc1 Mon Sep 17 00:00:00 2001 From: Staffan Olsson Date: Mon, 12 Oct 2026 06:59:41 +0200 Subject: [PATCH 1/2] kadm: export each topic's effective retention, compaction and size config A Buckety's parameters do not say what retention a topic has. The broker merges topic overrides with its own config and built-in defaults: a topic whose spec leaves retention.ms out keeps records forever on a Kafka with log.retention.hours=-1 and for 7 days on a default Redpanda, and a config.* key removed from the spec stays on the topic as an unmanaged override. Two gauges, set from the DescribeConfigs answer every reconcile already reads (no extra broker call, as fresh as the re-check cadence): buckety_kadm_topic_config{namespace,buckety,topic,key,source} buckety_kadm_topic_cleanup_policy{namespace,buckety,topic,policy,source} 1 for retention.ms, retention.bytes, delete.retention.ms, min/max.compaction.lag.ms, min.cleanable.dirty.ratio, segment.ms, segment.bytes, max.message.bytes and min.insync.replicas. source is where the broker says the value comes from: topic (an override), broker (broker or cluster config), default (built in) or unknown. An alter the broker acknowledged is exported at once as source=topic; a source or policy change replaces its series; a key the broker does not report has none. Configuration only: the topic's state (sizes, offsets, partitions, replicas) stays with Kafka exporters. Series go when their Buckety does. registry.ForgetBuckety is a new optional driver capability the reconciler calls on every backend once a Buckety is NotFound (its backend is no longer readable, and DeleteBuckety only runs for retentionPolicy=Delete). kadm's implementation also drops the change counters, which until now stayed until the next restart. e2e parameter-mutation asserts the requested retention.ms as a topic override before and after the patch, and an inherited retention.bytes; test/e2e/lib.sh gains controller_metric. Run locally against Redpanda v26.1.9: an unconfigured topic exported all ten keys but min.insync.replicas, each source=default (retention.ms 604800000), and no buckety_kadm_* series remained after the scenario namespace was deleted. Co-Authored-By: Claude Opus 5.5 --- README.md | 36 ++++- examples/kadm/parameter-mutation/assert.sh | 22 +++ pkg/controller/buckety/forget_test.go | 50 ++++++ pkg/controller/buckety/reconciler.go | 5 + pkg/drivers/kadm/driver.go | 11 ++ pkg/drivers/kadm/effective.go | 180 +++++++++++++++++++++ pkg/drivers/kadm/effective_test.go | 176 ++++++++++++++++++++ pkg/drivers/kadm/metrics.go | 20 ++- pkg/drivers/registry/registry.go | 12 ++ test/e2e/lib.sh | 18 +++ 10 files changed, 526 insertions(+), 4 deletions(-) create mode 100644 pkg/controller/buckety/forget_test.go create mode 100644 pkg/drivers/kadm/effective.go create mode 100644 pkg/drivers/kadm/effective_test.go diff --git a/README.md b/README.md index 7d36618..508cd37 100644 --- a/README.md +++ b/README.md @@ -481,6 +481,8 @@ resources, never parameter values, so there is one series per | `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.`. | +| `buckety_kadm_topic_config` | gauge | `namespace`, `buckety`, `topic`, `key`, `source` | Effective value of a topic config key as the broker reports it: `retention.ms`, `retention.bytes`, `delete.retention.ms`, `min.compaction.lag.ms`, `max.compaction.lag.ms`, `min.cleanable.dirty.ratio`, `segment.ms`, `segment.bytes`, `max.message.bytes`, `min.insync.replicas`. `source` is `topic` (an override on the topic), `broker` (broker or cluster config), `default` (built-in) or `unknown`. `-1` means unlimited for the retention keys. | +| `buckety_kadm_topic_cleanup_policy` | gauge | `namespace`, `buckety`, `topic`, `policy`, `source` | `1` for the topic's effective `cleanup.policy` (`delete`, `compact`, `compact,delete`), with `source` as above. | The `kadm` driver reconciles partition increases and `config.*` differences in place, on every reconcile and periodic re-check; @@ -515,7 +517,35 @@ warning (`kubectl apply` prints it) naming the removed keys, and the controller logs the same message once per such apply. To go back to the broker's default, delete the override on the broker (`kafka-configs --alter --delete-config`, `rpk topic alter-config ---delete`). +--delete`). `buckety_kadm_topic_config` with `source="topic"` shows +which overrides a topic still carries. + +The two `buckety_kadm_topic_*` gauges say what retention, +compaction and size limits a topic actually has. A broker merges +topic overrides with its own config and defaults, so the +`Buckety`'s parameters do not tell: a topic whose spec says nothing +about `retention.ms` keeps records for as long as the broker's +default says (forever on a Kafka with `log.retention.hours=-1`, +7 days on a default Redpanda), and a key removed from the spec +stays on the topic as an override. They are configuration only; +the topic's state (sizes, offsets, partitions, replicas, lag) +belongs to Kafka exporters such as kminion. Values come from the +DescribeConfigs call every reconcile already makes, so they are as +fresh as the re-check cadence, and an alter the broker acknowledged +shows at once with `source="topic"`. A key the broker does not +report (Redpanda 24.2 with `min.compaction.lag.ms`) has no series. +A topic's series appear with the first reconcile that finds it +existing: shortly after the pass that created it (seconds to tens +of seconds, while the client's metadata catches up), or, after a +controller start, the first pass of its `Buckety`, which for a +settled `Buckety` can be up to one `--periodic-recheck` later. For example: + +```promql +# topics of a Buckety name that will delete records by time +buckety_kadm_topic_config{buckety="orders-log",key="retention.ms"} != -1 +# every topic-level override, whether its Buckety still sets it or not +buckety_kadm_topic_config{source="topic"} +``` Suggested alerts. The counters exist at `0` from the first reconcile of a topic, so `increase()` catches later changes. The @@ -562,8 +592,8 @@ spec: ``` 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 +with the Pod. All `buckety_kadm_*` series of a `Buckety` are +removed once it is deleted. The Events and the log line are the exact record of what changed and from what. ### Controller won't start diff --git a/examples/kadm/parameter-mutation/assert.sh b/examples/kadm/parameter-mutation/assert.sh index 7cfc5b0..b14868a 100755 --- a/examples/kadm/parameter-mutation/assert.sh +++ b/examples/kadm/parameter-mutation/assert.sh @@ -6,6 +6,25 @@ wait_ready buckety/cfg-mut 120s secret_has_keys cfg-mut-topic bootstrap topic topic_name="$(secret_value cfg-mut-topic topic)" +# Effective config metrics: the requested retention.ms is a topic +# override; retention.bytes, which the spec leaves out, is exported +# with what the broker says it inherits from. +effective_retention() { + controller_metric buckety_kadm_topic_config "namespace=\"$E2E_NAMESPACE\"" 'buckety="cfg-mut"' "key=\"$1\"" +} +expect_effective() { + local key="$1" want="$2" got="" + for _ in $(seq 1 30); do + got="$(effective_retention "$key")" + [[ "$got" =~ $want ]] && return 0 + sleep 2 + done + fail "buckety_kadm_topic_config for $key: got '$got', want a line matching $want" +} +log "checking effective config metrics" +expect_effective retention.ms 'source="topic".*\} 6\.048e\+08$' +expect_effective retention.bytes 'source="(broker|default)".*\} ' + # Phase 1: patch to a new retention. The impl branch must # reconcile this to the broker; this assert checks the broker # state via rpk. @@ -44,6 +63,9 @@ for _ in $(seq 1 12); do done [[ "$broker_retention" == "3600000" ]] \ || fail "broker retention.ms=$broker_retention, expected 3600000" +expect_effective retention.ms 'source="topic".*\} 3\.6e\+06$' +[[ "$(effective_retention retention.ms | wc -l)" -eq 1 ]] \ + || fail "more than one retention.ms series: $(effective_retention retention.ms)" # Immutable-parameter rejection. Patching replicationFactor must # fail at the admission webhook. diff --git a/pkg/controller/buckety/forget_test.go b/pkg/controller/buckety/forget_test.go new file mode 100644 index 0000000..570ed17 --- /dev/null +++ b/pkg/controller/buckety/forget_test.go @@ -0,0 +1,50 @@ +package buckety + +import ( + "context" + "testing" + + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "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" +) + +// forgettingDriver records ForgetBuckety calls. +type forgettingDriver struct { + provisioningDriver + forgot []string +} + +func (f *forgettingDriver) ForgetBuckety(namespace, name string) { + f.forgot = append(f.forgot, namespace+"/"+name) +} + +// A Buckety that no longer exists is forgotten by every backend's +// driver: its backend can no longer be read from it, and drivers +// export per-Buckety series whatever its retentionPolicy was. +func TestGoneBucketyIsForgottenByDrivers(t *testing.T) { + scheme := runtime.NewScheme() + if err := clientgoscheme.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + if err := bucketyv1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + a, b := &forgettingDriver{}, &forgettingDriver{} + r := &Reconciler{Client: fake.NewClientBuilder().WithScheme(scheme).Build(), Scheme: scheme, Config: &config.Loaded{ + Backends: map[string]config.Backend{"a": {Name: "a", Driver: a}, "b": {Name: "b", Driver: b}, "c": {Name: "c", Driver: &provisioningDriver{}}}, + }} + if _, err := r.Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Namespace: "t1", Name: "gone"}}); err != nil { + t.Fatal(err) + } + for _, d := range []*forgettingDriver{a, b} { + if len(d.forgot) != 1 || d.forgot[0] != "t1/gone" { + t.Errorf("forgot %q, want [t1/gone]", d.forgot) + } + } +} diff --git a/pkg/controller/buckety/reconciler.go b/pkg/controller/buckety/reconciler.go index ef51e6a..56893c6 100644 --- a/pkg/controller/buckety/reconciler.go +++ b/pkg/controller/buckety/reconciler.go @@ -112,6 +112,11 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu if apierrors.IsNotFound(err) { forgetDrift(req.Namespace, req.Name) r.startup.Forget(req.NamespacedName) + // The Buckety's backend is no longer known; ask every + // driver to drop what it exports for it. + for _, b := range r.Config.Backends { + registry.ForgetBuckety(b.Driver, req.Namespace, req.Name) + } return ctrl.Result{}, nil } return ctrl.Result{}, err diff --git a/pkg/drivers/kadm/driver.go b/pkg/drivers/kadm/driver.go index 78d9d25..a9fd5b8 100644 --- a/pkg/drivers/kadm/driver.go +++ b/pkg/drivers/kadm/driver.go @@ -84,6 +84,10 @@ type Driver struct { // a restart such a key is altered once more. mu sync.Mutex accepted map[string]map[string]*acceptedConfig + + // effective tracks the effective config series exported per + // Buckety (effective.go). + effective effectiveExport } // acceptedConfig is a config value the broker acknowledged. @@ -160,6 +164,7 @@ func (d *Driver) EnsureBuckety(ctx context.Context, req registry.EnsureRequest) d.forgetAccepted(req.Name) return d.createTopic(ctx, req.Name, wantParts, wantRF, wantCfgs) } + d.recordEffective(req.Namespace, req.BucketyName, req.Name, existing) return d.alignTopic(ctx, req, existing, wantParts, wantRF, wantCfgs) } @@ -287,6 +292,9 @@ type topicView struct { partitions int32 rf int16 configs map[string]string + // sources says where the broker took each value in configs + // from: the topic's override, broker config or a default. + sources map[string]kmsg.ConfigSource } // topicExists asks the brokers directly. kadm's ListTopics (which @@ -351,6 +359,7 @@ func (d *Driver) describeTopic(ctx context.Context, name string) (*topicView, er return nil, fmt.Errorf("kadm: describe configs %q: %w", name, err) } view.configs = make(map[string]string) + view.sources = make(map[string]kmsg.ConfigSource) for _, rc := range cfgs { if rc.Err != nil { // Skipping would leave the config view empty, and the @@ -361,6 +370,7 @@ func (d *Driver) describeTopic(ctx context.Context, name string) (*topicView, er for _, c := range rc.Configs { if c.Value != nil { view.configs[c.Key] = *c.Value + view.sources[c.Key] = c.Source } } } @@ -474,6 +484,7 @@ func (d *Driver) alignTopic(ctx context.Context, req registry.EnsureRequest, exi // keys or none, so success means every key changed. sort.Slice(alters, func(i, j int) bool { return alters[i].Name < alters[j].Name }) d.rememberAccepted(name, alters) + d.recordAltered(req.Namespace, req.BucketyName, name, alters) for _, a := range alters { configChanges.WithLabelValues(req.Namespace, req.BucketyName, name, a.Name).Inc() report(req, registry.Change{ diff --git a/pkg/drivers/kadm/effective.go b/pkg/drivers/kadm/effective.go new file mode 100644 index 0000000..8171241 --- /dev/null +++ b/pkg/drivers/kadm/effective.go @@ -0,0 +1,180 @@ +package kadm + +import ( + "strconv" + "sync" + + "github.com/prometheus/client_golang/prometheus" + "github.com/twmb/franz-go/pkg/kadm" + "github.com/twmb/franz-go/pkg/kmsg" +) + +// effectiveKeys are the topic config keys exported as +// buckety_kadm_topic_config: retention, compaction and size limits, +// whose effective value depends on how the broker merges topic +// overrides with broker config and defaults, so the Buckety's own +// parameters cannot tell it. cleanup.policy is a string and has its +// own gauge. +var effectiveKeys = []string{ + "retention.ms", + "retention.bytes", + "delete.retention.ms", + "min.compaction.lag.ms", + "max.compaction.lag.ms", + "min.cleanable.dirty.ratio", + "segment.ms", + "segment.bytes", + "max.message.bytes", + "min.insync.replicas", +} + +const cleanupPolicyKey = "cleanup.policy" + +// sourceLabel names where the broker says a config value comes +// from: the topic's own override, any broker-level config, or the +// built-in default. +func sourceLabel(s kmsg.ConfigSource) string { + switch s { + case kmsg.ConfigSourceDynamicTopicConfig: + return "topic" + case kmsg.ConfigSourceDynamicBrokerConfig, + kmsg.ConfigSourceDynamicDefaultBrokerConfig, + kmsg.ConfigSourceStaticBrokerConfig: + return "broker" + case kmsg.ConfigSourceDefaultConfig: + return "default" + } + return "unknown" +} + +// effectiveExport remembers, per Buckety, the label values of the +// series last exported for it, so that a value whose source or +// policy changed replaces its old series instead of leaving it +// behind, and ForgetBuckety can drop them all. +type effectiveExport struct { + mu sync.Mutex + byBuckety map[bucketyID]*exported +} + +type bucketyID struct{ namespace, name string } + +type exported struct { + topic string + // source per effectiveKeys key; absent when not exported + source map[string]string + // policy and its source, empty when not exported + policy, policySource string +} + +// recordEffective exports the effective config in view for the +// Buckety's topic. A key the broker does not report (or reports +// as a non-number) has no series. +func (d *Driver) recordEffective(namespace, buckety, topic string, view *topicView) { + d.effective.mu.Lock() + defer d.effective.mu.Unlock() + e := d.effective.entry(namespace, buckety, topic) + for _, k := range effectiveKeys { + v, ok := view.configs[k] + var f float64 + if ok { + var err error + f, err = strconv.ParseFloat(v, 64) + ok = err == nil + } + if !ok { + if src, had := e.source[k]; had { + topicConfig.DeleteLabelValues(namespace, buckety, topic, k, src) + delete(e.source, k) + } + continue + } + e.setConfig(namespace, buckety, k, f, sourceLabel(view.sources[k])) + } + if p, ok := view.configs[cleanupPolicyKey]; ok { + e.setPolicy(namespace, buckety, p, sourceLabel(view.sources[cleanupPolicyKey])) + } else if e.policy != "" { + topicCleanupPolicy.DeleteLabelValues(namespace, buckety, topic, e.policy, e.policySource) + e.policy, e.policySource = "", "" + } +} + +// recordAltered exports the values of an alter the broker +// acknowledged: they are now topic overrides. +func (d *Driver) recordAltered(namespace, buckety, topic string, alters []kadm.AlterConfig) { + d.effective.mu.Lock() + defer d.effective.mu.Unlock() + e := d.effective.entry(namespace, buckety, topic) + for _, a := range alters { + if a.Value == nil { + continue + } + if a.Name == cleanupPolicyKey { + e.setPolicy(namespace, buckety, *a.Value, "topic") + continue + } + if f, err := strconv.ParseFloat(*a.Value, 64); err == nil && isEffectiveKey(a.Name) { + e.setConfig(namespace, buckety, a.Name, f, "topic") + } + } +} + +// ForgetBuckety drops every series this driver exports for the +// Buckety: the effective config gauges and the change counters. +// The reconciler calls it once the Buckety is gone (registry +// ForgetBuckety). +func (d *Driver) ForgetBuckety(namespace, name string) { + labels := prometheus.Labels{"namespace": namespace, "buckety": name} + topicConfig.DeletePartialMatch(labels) + topicCleanupPolicy.DeletePartialMatch(labels) + configChanges.DeletePartialMatch(labels) + partitionsAdded.DeletePartialMatch(labels) + d.effective.mu.Lock() + defer d.effective.mu.Unlock() + delete(d.effective.byBuckety, bucketyID{namespace, name}) +} + +// entry returns the Buckety's export record, starting over (and +// dropping its series) if its topic changed. Callers hold mu. +func (x *effectiveExport) entry(namespace, buckety, topic string) *exported { + if x.byBuckety == nil { + x.byBuckety = map[bucketyID]*exported{} + } + id := bucketyID{namespace, buckety} + e := x.byBuckety[id] + if e != nil && e.topic != topic { + labels := prometheus.Labels{"namespace": namespace, "buckety": buckety, "topic": e.topic} + topicConfig.DeletePartialMatch(labels) + topicCleanupPolicy.DeletePartialMatch(labels) + e = nil + } + if e == nil { + e = &exported{topic: topic, source: map[string]string{}} + x.byBuckety[id] = e + } + return e +} + +func isEffectiveKey(key string) bool { + for _, k := range effectiveKeys { + if k == key { + return true + } + } + return false +} + +func (e *exported) setConfig(namespace, buckety, key string, value float64, source string) { + if old, had := e.source[key]; had && old != source { + topicConfig.DeleteLabelValues(namespace, buckety, e.topic, key, old) + } + e.source[key] = source + topicConfig.WithLabelValues(namespace, buckety, e.topic, key, source).Set(value) +} + +func (e *exported) setPolicy(namespace, buckety, policy, source string) { + if e.policy != "" && (e.policy != policy || e.policySource != source) { + topicCleanupPolicy.DeleteLabelValues(namespace, buckety, e.topic, e.policy, e.policySource) + } + e.policy, e.policySource = policy, source + topicCleanupPolicy.WithLabelValues(namespace, buckety, e.topic, policy, source).Set(1) +} diff --git a/pkg/drivers/kadm/effective_test.go b/pkg/drivers/kadm/effective_test.go new file mode 100644 index 0000000..1c169bf --- /dev/null +++ b/pkg/drivers/kadm/effective_test.go @@ -0,0 +1,176 @@ +package kadm + +import ( + "context" + "fmt" + "sort" + "strings" + "testing" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/twmb/franz-go/pkg/kmsg" + + "github.com/Yolean/buckety-controller/pkg/drivers/registry" +) + +// series returns the series vec exports for Buckety ns/buckety, as +// sorted "label=value,... value" lines (labels in Prometheus' order, +// which is alphabetical) without the namespace and buckety labels. +func series(t *testing.T, vec prometheus.Collector, ns, buckety string) []string { + t.Helper() + ch := make(chan prometheus.Metric, 100) + vec.Collect(ch) + close(ch) + var out []string + for m := range ch { + var d dto.Metric + if err := m.Write(&d); err != nil { + t.Fatal(err) + } + var labels []string + match := 0 + for _, l := range d.GetLabel() { + switch { + case l.GetName() == "namespace" && l.GetValue() == ns, l.GetName() == "buckety" && l.GetValue() == buckety: + match++ + default: + labels = append(labels, l.GetName()+"="+l.GetValue()) + } + } + if match == 2 { + out = append(out, fmt.Sprintf("%s %g", strings.Join(labels, ","), d.GetGauge().GetValue())) + } + } + sort.Strings(out) + return out +} + +func viewOf(cfgs map[string]string, srcs map[string]kmsg.ConfigSource) *topicView { + return &topicView{partitions: 1, rf: 1, configs: cfgs, sources: srcs} +} + +func wantSeries(t *testing.T, what string, got []string, want ...string) { + t.Helper() + if strings.Join(got, "\n") != strings.Join(want, "\n") { + t.Errorf("%s:\n got %q\nwant %q", what, got, want) + } +} + +// The effective value and its source are exported for every listed +// key the broker reports; a source or policy change replaces the +// series, and a key that stops being reported loses it. +func TestEffectiveConfigFollowsTheBroker(t *testing.T) { + const ns, bky, topic = "eff1", "orders", "eff1.orders" + d := &Driver{} + + d.recordEffective(ns, bky, topic, viewOf( + map[string]string{"retention.ms": "-1", "retention.bytes": "-1", "segment.bytes": "1073741824", "cleanup.policy": "delete", "compression.type": "producer"}, + map[string]kmsg.ConfigSource{ + "retention.ms": kmsg.ConfigSourceStaticBrokerConfig, + "retention.bytes": kmsg.ConfigSourceDefaultConfig, + "segment.bytes": kmsg.ConfigSourceDynamicDefaultBrokerConfig, + "cleanup.policy": kmsg.ConfigSourceDefaultConfig, + })) + wantSeries(t, "first pass", series(t, topicConfig, ns, bky), + "key=retention.bytes,source=default,topic=eff1.orders -1", + "key=retention.ms,source=broker,topic=eff1.orders -1", + "key=segment.bytes,source=broker,topic=eff1.orders 1.073741824e+09", + ) + wantSeries(t, "first pass policy", series(t, topicCleanupPolicy, ns, bky), + "policy=delete,source=default,topic=eff1.orders 1") + + // An override appears out of band, the policy changes, and the + // broker stops reporting segment.bytes. + d.recordEffective(ns, bky, topic, viewOf( + map[string]string{"retention.ms": "604800000", "retention.bytes": "-1", "cleanup.policy": "compact,delete"}, + map[string]kmsg.ConfigSource{ + "retention.ms": kmsg.ConfigSourceDynamicTopicConfig, + "retention.bytes": kmsg.ConfigSourceDefaultConfig, + "cleanup.policy": kmsg.ConfigSourceDynamicTopicConfig, + })) + wantSeries(t, "second pass", series(t, topicConfig, ns, bky), + "key=retention.bytes,source=default,topic=eff1.orders -1", + "key=retention.ms,source=topic,topic=eff1.orders 6.048e+08", + ) + wantSeries(t, "second pass policy", series(t, topicCleanupPolicy, ns, bky), + "policy=compact,delete,source=topic,topic=eff1.orders 1") +} + +// An alter the broker acknowledged is exported at once as a topic +// override, replacing the inherited value's series. +func TestEffectiveConfigAfterAlter(t *testing.T) { + const ns, bky, topic = "eff2", "orders", "eff2.orders" + m := &fakeMutator{} + d := &Driver{mutator: m} + existing := viewOf( + map[string]string{"retention.ms": "-1", "cleanup.policy": "delete"}, + map[string]kmsg.ConfigSource{"retention.ms": kmsg.ConfigSourceStaticBrokerConfig, "cleanup.policy": kmsg.ConfigSourceDefaultConfig}) + d.recordEffective(ns, bky, topic, existing) + err := d.alignTopic(context.Background(), registry.EnsureRequest{Name: topic, Namespace: ns, BucketyName: bky}, existing, 0, 0, + map[string]*string{"retention.ms": strp("86400000"), "cleanup.policy": strp("compact")}) + if err != nil { + t.Fatal(err) + } + wantSeries(t, "after alter", series(t, topicConfig, ns, bky), + "key=retention.ms,source=topic,topic=eff2.orders 8.64e+07") + wantSeries(t, "after alter policy", series(t, topicCleanupPolicy, ns, bky), + "policy=compact,source=topic,topic=eff2.orders 1") + + // A rejected alter changes nothing. + m.alterErr = errInvalidConfigForTest + _ = d.alignTopic(context.Background(), registry.EnsureRequest{Name: topic, Namespace: ns, BucketyName: bky}, existing, 0, 0, + map[string]*string{"retention.ms": strp("1")}) + wantSeries(t, "after a rejected alter", series(t, topicConfig, ns, bky), + "key=retention.ms,source=topic,topic=eff2.orders 8.64e+07") +} + +// Once the Buckety is gone, every series the driver exports for it +// goes, change counters included; other Bucketies keep theirs. +func TestForgetBucketyDropsItsSeries(t *testing.T) { + const ns, topic = "eff3", "eff3.a" + m := &fakeMutator{} + d := &Driver{mutator: m} + view := func() *topicView { + return viewOf(map[string]string{"retention.ms": "-1", "cleanup.policy": "delete"}, + map[string]kmsg.ConfigSource{"retention.ms": kmsg.ConfigSourceDefaultConfig, "cleanup.policy": kmsg.ConfigSourceDefaultConfig}) + } + for _, b := range []string{"a", "b"} { + d.recordEffective(ns, b, topic+b, view()) + if err := d.alignTopic(context.Background(), registry.EnsureRequest{Name: topic + b, Namespace: ns, BucketyName: b}, view(), 0, 0, + map[string]*string{"retention.bytes": strp("1")}); err != nil { + t.Fatal(err) + } + } + + registry.ForgetBuckety(d, ns, "a") + for _, c := range []struct { + name string + vec prometheus.Collector + }{{"config", topicConfig}, {"policy", topicCleanupPolicy}, {"changes", configChanges}, {"partitions", partitionsAdded}} { + if got := series(t, c.vec, ns, "a"); len(got) != 0 { + t.Errorf("%s series left for a forgotten Buckety: %q", c.name, got) + } + if got := series(t, c.vec, ns, "b"); len(got) == 0 { + t.Errorf("%s series of another Buckety dropped", c.name) + } + } + if _, ok := d.effective.byBuckety[bucketyID{ns, "a"}]; ok { + t.Error("export record kept for a forgotten Buckety") + } +} + +func TestSourceLabel(t *testing.T) { + for s, want := range map[kmsg.ConfigSource]string{ + kmsg.ConfigSourceDynamicTopicConfig: "topic", + kmsg.ConfigSourceDynamicBrokerConfig: "broker", + kmsg.ConfigSourceDynamicDefaultBrokerConfig: "broker", + kmsg.ConfigSourceStaticBrokerConfig: "broker", + kmsg.ConfigSourceDefaultConfig: "default", + kmsg.ConfigSourceUnknown: "unknown", + } { + if got := sourceLabel(s); got != want { + t.Errorf("%v: %q, want %q", s, got, want) + } + } +} diff --git a/pkg/drivers/kadm/metrics.go b/pkg/drivers/kadm/metrics.go index 7707323..232f7fc 100644 --- a/pkg/drivers/kadm/metrics.go +++ b/pkg/drivers/kadm/metrics.go @@ -30,6 +30,24 @@ var ( }, []string{"namespace", "buckety", "topic", "key"}) ) +// Gauges for the topic configuration the broker reports as in +// effect (effective.go): configuration only, never the topic's +// state (sizes, offsets, partitions, replicas), which Kafka +// exporters already measure. Set from the DescribeConfigs answer +// every reconcile of the Buckety already reads, so they cost no +// extra broker call, and are as fresh as the re-check cadence. +var ( + topicConfig = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "buckety_kadm_topic_config", + Help: "Effective value of a topic config key as the broker reports it, whether set on the topic or inherited from the broker's config or built-in default (source: topic, broker, default, unknown); -1 means unlimited for retention.ms and retention.bytes. Configuration only, nothing about the topic's state.", + }, []string{"namespace", "buckety", "topic", "key", "source"}) + + topicCleanupPolicy = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "buckety_kadm_topic_cleanup_policy", + Help: "1 for the topic's effective cleanup.policy (policy: delete, compact or compact,delete as the broker reports it), with the same source label as buckety_kadm_topic_config.", + }, []string{"namespace", "buckety", "topic", "policy", "source"}) +) + func init() { - crmetrics.Registry.MustRegister(partitionsAdded, configChanges) + crmetrics.Registry.MustRegister(partitionsAdded, configChanges, topicConfig, topicCleanupPolicy) } diff --git a/pkg/drivers/registry/registry.go b/pkg/drivers/registry/registry.go index a73bada..a581aec 100644 --- a/pkg/drivers/registry/registry.go +++ b/pkg/drivers/registry/registry.go @@ -251,6 +251,18 @@ func TemplatedParameters(d Driver) []string { return nil } +// ForgetBuckety tells a driver with the optional capability that +// the Buckety namespace/name is gone, so it can drop what it keeps +// per Buckety, such as the metric series it exports. Drivers +// without it are skipped. The controller calls it on every backend +// once the object no longer exists, whatever its retentionPolicy: +// DeleteBuckety runs only for retentionPolicy=Delete. +func ForgetBuckety(d Driver, namespace, name string) { + if f, ok := d.(interface{ ForgetBuckety(namespace, name string) }); ok { + f.ForgetBuckety(namespace, name) + } +} + // ErrParameterDrift is the typed error EnsureBuckety returns when // it observes drift on the backend it cannot reconcile in place // (e.g. Kafka partition shrink). The controller maps this to a diff --git a/test/e2e/lib.sh b/test/e2e/lib.sh index 8feb049..82a43b7 100755 --- a/test/e2e/lib.sh +++ b/test/e2e/lib.sh @@ -423,6 +423,24 @@ controller_rollout_wait() { done } +# controller_metric ... +# Prints the controller's series of metric whose labels +# contain every (e.g. 'buckety="cfg-mut"'), one per +# line as exposed, scraped through the apiserver's pod proxy from +# the running controller Pod. No match prints nothing and succeeds, +# so callers can poll. +controller_metric() { + local name="$1"; shift + local pod + pod="$(kcg -n "$E2E_CONTROLLER_NS" get pods -l app.kubernetes.io/name=buckety-controller \ + --field-selector=status.phase=Running -o jsonpath='{.items[0].metadata.name}')" + kcg get --raw "/api/v1/namespaces/$E2E_CONTROLLER_NS/pods/$pod:8080/proxy/metrics" \ + | { grep "^$name{" || true; } | { local l m; while IFS= read -r l; do + for m in "$@"; do [[ "$l" == *"$m"* ]] || continue 2; done + printf '%s\n' "$l" + done; } +} + # controller_config_apply # Installs as the controller config Secret and restarts the # controller - the operation a platform performs on a config From d078e74f15be8d3c54e20f8a74c530ebf3bd2383 Mon Sep 17 00:00:00 2001 From: Staffan Olsson Date: Mon, 12 Oct 2026 07:02:04 +0200 Subject: [PATCH 2/2] release: pin the effective topic config metrics build scripts/bump-release.sh with the go.mod toolchain (go1.27.1). 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 6866f22..b271f25 100644 --- a/deploy/kustomize/release/kustomization.yaml +++ b/deploy/kustomize/release/kustomization.yaml @@ -7,7 +7,7 @@ kind: Kustomization resources: - ../base images: -- digest: sha256:d3fa99320e940037574071795199b88c513a45b72803077f0a7012066e2999cd +- digest: sha256:107d7d0ec71c5d7fcadc4206ff63c6fdf655daa61a26a5c6b7037ea8953cbe7f name: ghcr.io/yolean/buckety-controller newName: ghcr.io/yolean/buckety-controller - newTag: 20261011T075523Z + newTag: 20261012T050156Z