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: 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 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/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 new file mode 100644 index 0000000..6527a08 --- /dev/null +++ b/pkg/controller/buckety/observability_test.go @@ -0,0 +1,228 @@ +package buckety + +import ( + "context" + "strings" + "testing" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/tools/record" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + bucketyv1 "github.com/Yolean/buckety-controller/pkg/api/v1alpha1" + "github.com/Yolean/buckety-controller/pkg/config" + "github.com/Yolean/buckety-controller/pkg/drivers/registry" +) + +// changingDriver is a provisioningDriver whose EnsureBuckety +// reports the given changes through OnChange, or answers with +// ParameterDrift while drift is set. +type changingDriver struct { + provisioningDriver + changes []registry.Change + drift bool + gotReq registry.EnsureRequest +} + +func (c *changingDriver) EnsureBuckety(_ context.Context, req registry.EnsureRequest) error { + c.gotReq = req + if c.drift { + return ®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) + } + } +} + +// 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 7725693..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 @@ -259,8 +263,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 +538,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/kadm/driver.go b/pkg/drivers/kadm/driver.go index c81adf5..f60329e 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,20 @@ 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 + // 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{ @@ -364,9 +391,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 +430,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 +439,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..f290308 100644 --- a/pkg/drivers/kadm/driver_test.go +++ b/pkg/drivers/kadm/driver_test.go @@ -1,10 +1,21 @@ 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" + crmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" + + "github.com/Yolean/buckety-controller/pkg/drivers/registry" ) // Validation methods never touch the broker clients, so a zero @@ -163,3 +174,215 @@ 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) + } +} + +// 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) + } + } +} 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) +} 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