Warn on apply with removed keys, keep ignoring them in drivers

README.md

@@ -506,6 +506,17 @@ the value the broker acknowledged, with one info log line saying
 the broker does not report it; a change to the requested value is
 set again.
 
+Removing a `config.*` key from a `Buckety` does not revert it: the
+topic keeps the value last applied, and the controller stops
+managing it. What the broker would fall back to (its default, a
+broker-level setting) is not something the controller can predict.
+To make that visible, the admission webhook allows the apply with a
+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`).
+
 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

SPEC.md

@@ -449,6 +449,10 @@ Two categories of parameter keys:
    own validation catches typos; the driver SHOULD document which
    `config.*` keys are immutable on the backend side and surface
    `ParameterDrift` conditions where it can't reconcile in place.
+   A `config.*` key removed from `spec.parameters` is not reverted:
+   the backend keeps the value last applied, unmanaged, because
+   what it would fall back to is the backend's business. Admission
+   allows the update with a warning naming the removed keys.
 
 Naming conventions per driver:
 

deploy/kustomize/release/kustomization.yaml

@@ -7,7 +7,7 @@ kind: Kustomization
 resources:
 - ../base
 images:
-- digest: sha256:a1f60a76e70280d09d7bb863f31444b3036dfb6b490bee893e04b99790841911
+- digest: sha256:d3fa99320e940037574071795199b88c513a45b72803077f0a7012066e2999cd
   name: ghcr.io/yolean/buckety-controller
   newName: ghcr.io/yolean/buckety-controller
-  newTag: 20261011T045210Z
+  newTag: 20261011T075523Z

pkg/webhook/webhook.go

@@ -13,9 +13,12 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"sort"
+	"strings"
 
 	admissionv1 "k8s.io/api/admission/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
 	bucketyv1 "github.com/Yolean/buckety-controller/pkg/api/v1alpha1"
@@ -48,7 +51,7 @@ func (v *Validator) Handle(ctx context.Context, req admission.Request) admission
 	}
 }
 
-func (v *Validator) validateBuckety(_ context.Context, req admission.Request) admission.Response {
+func (v *Validator) validateBuckety(ctx context.Context, req admission.Request) admission.Response {
 	if req.Operation == admissionv1.Delete {
 		return admission.Allowed("")
 	}
@@ -96,6 +99,7 @@ func (v *Validator) validateBuckety(_ context.Context, req admission.Request) ad
 		return admission.Denied(fmt.Sprintf("spec.parameters: %v", err))
 	}
 
+	var warnings []string
 	if req.Operation == admissionv1.Update && len(req.OldObject.Raw) > 0 {
 		var old bucketyv1.Buckety
 		if err := json.Unmarshal(req.OldObject.Raw, &old); err != nil {
@@ -115,6 +119,14 @@ func (v *Validator) validateBuckety(_ context.Context, req admission.Request) ad
 		if err := backend.Driver.ValidateUpdateParameters(oldParams, params); err != nil {
 			return admission.Denied(fmt.Sprintf("spec.parameters: %v", err))
 		}
+		if removed := removedConfigKeys(oldParams, params); len(removed) > 0 {
+			msg := fmt.Sprintf("spec.parameters: %s removed; the backend keeps the value last applied, which buckety no longer manages (pass-through config.* keys are not reverted)", strings.Join(removed, ", "))
+			warnings = append(warnings, msg)
+			// kubectl diff and --dry-run=server are not applies.
+			if req.DryRun == nil || !*req.DryRun {
+				logf.FromContext(ctx).Info(msg, "namespace", bky.Namespace, "buckety", bky.Name, "removed", removed)
+			}
+		}
 	} else if err := backend.Driver.ValidateParameters(params); err != nil {
 		return admission.Denied(fmt.Sprintf("spec.parameters: %v", err))
 	}
@@ -124,7 +136,27 @@ func (v *Validator) validateBuckety(_ context.Context, req admission.Request) ad
 		return admission.Denied("spec.defaultAccess.credentialsSecretName is required when defaultAccess is set")
 	}
 
-	return admission.Allowed("")
+	return admission.Allowed("").WithWarnings(warnings...)
+}
+
+// removedConfigKeys returns the pass-through config.* keys (SPEC
+// "Parameters", category 2) that oldParams has and newParams does
+// not, sorted. Both are merged views, so a key that falls back to a
+// backend default is not removed. Drivers do not revert such a key:
+// an omitted parameter means "pass nothing", and what the backend
+// does with a deleted override (its default, a broker-level value,
+// whatever the server version decides) is not something the
+// controller can predict. The value last applied stays in place,
+// unmanaged, so the apply that drops it gets a warning instead.
+func removedConfigKeys(oldParams, newParams map[string]string) []string {
+	var removed []string
+	for k := range oldParams {
+		if _, ok := newParams[k]; !ok && strings.HasPrefix(k, "config.") {
+			removed = append(removed, k)
+		}
+	}
+	sort.Strings(removed)
+	return removed
 }
 
 func (v *Validator) validateAccess(_ context.Context, req admission.Request) admission.Response {

pkg/webhook/webhook_test.go

@@ -0,0 +1,102 @@
+package webhook
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	admissionv1 "k8s.io/api/admission/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
+
+	bucketyv1 "github.com/Yolean/buckety-controller/pkg/api/v1alpha1"
+	"github.com/Yolean/buckety-controller/pkg/config"
+	_ "github.com/Yolean/buckety-controller/pkg/drivers/kadm"
+	"github.com/Yolean/buckety-controller/pkg/drivers/registry"
+)
+
+// kadmValidator validates against a kadm backend whose defaults
+// include config.cleanup.policy. The driver never dials during
+// admission, so the seed broker need not exist.
+func kadmValidator(t *testing.T) *Validator {
+	t.Helper()
+	f, ok := registry.Lookup("kadm")
+	if !ok {
+		t.Fatal("kadm driver not registered")
+	}
+	d, err := f(json.RawMessage(`{"seedBrokers":["127.0.0.1:1"]}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return &Validator{Config: &config.Loaded{Backends: map[string]config.Backend{
+		"stream": {Name: "stream", Driver: d, Parameters: map[string]string{"config.cleanup.policy": "delete"}},
+	}}}
+}
+
+func topicUpdate(t *testing.T, v *Validator, oldParams, newParams map[string]string, dryRun bool) admission.Response {
+	t.Helper()
+	obj := func(p map[string]string) runtime.RawExtension {
+		raw, err := json.Marshal(&bucketyv1.Buckety{
+			TypeMeta:   metav1.TypeMeta{APIVersion: "buckety.yolean.se/v1alpha1", Kind: "Buckety"},
+			ObjectMeta: metav1.ObjectMeta{Name: "topic-ops", Namespace: "t1"},
+			Spec:       bucketyv1.BucketySpec{Backend: "stream", Parameters: p},
+			Status:     bucketyv1.BucketyStatus{BackendResourceName: "t1.ops"},
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return runtime.RawExtension{Raw: raw}
+	}
+	return v.Handle(context.Background(), admission.Request{AdmissionRequest: admissionv1.AdmissionRequest{
+		Kind:      metav1.GroupVersionKind{Kind: "Buckety"},
+		Operation: admissionv1.Update,
+		Object:    obj(newParams),
+		OldObject: obj(oldParams),
+		DryRun:    &dryRun,
+	}})
+}
+
+// An apply that drops a pass-through config.* key is allowed, with
+// a warning that the broker keeps the value last applied. Keys still
+// present, changed, or falling back to a backend default are not
+// removed.
+func TestRemovedConfigKeyWarns(t *testing.T) {
+	v := kadmValidator(t)
+	base := map[string]string{"partitions": "1", "config.max.message.bytes": "500006"}
+	with := func(kv ...string) map[string]string {
+		m := map[string]string{}
+		for k, val := range base {
+			m[k] = val
+		}
+		for i := 0; i < len(kv); i += 2 {
+			m[kv[i]] = kv[i+1]
+		}
+		return m
+	}
+
+	res := topicUpdate(t, v, with("config.retention.ms", "-1", "config.retention.bytes", "-1"), base, false)
+	if !res.Allowed {
+		t.Fatalf("removal denied: %+v", res.Result)
+	}
+	if len(res.Warnings) != 1 || !strings.Contains(res.Warnings[0], "config.retention.bytes, config.retention.ms removed") {
+		t.Errorf("warnings %q", res.Warnings)
+	}
+	if res := topicUpdate(t, v, with("config.retention.ms", "-1"), base, true); !res.Allowed || len(res.Warnings) != 1 {
+		t.Errorf("dry run: allowed=%v warnings %q", res.Allowed, res.Warnings)
+	}
+
+	for name, c := range map[string][2]map[string]string{
+		"unchanged":           {with("config.retention.ms", "-1"), with("config.retention.ms", "-1")},
+		"changed":             {with("config.retention.ms", "-1"), with("config.retention.ms", "1000")},
+		"added":               {base, with("config.retention.ms", "-1")},
+		"driver-known key":    {with("partitions", "3"), map[string]string{"config.max.message.bytes": "500006"}},
+		"back to the default": {with("config.cleanup.policy", "compact"), base},
+	} {
+		res := topicUpdate(t, v, c[0], c[1], false)
+		if !res.Allowed || len(res.Warnings) != 0 {
+			t.Errorf("%s: allowed=%v warnings %q", name, res.Allowed, res.Warnings)
+		}
+	}
+}