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/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 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