Metrics for current effective kafka topic config

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.<key>`. |
+| `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

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

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.

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)
+		}
+	}
+}

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

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{

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)
+}