diff --git a/README.md b/README.md index d294f40..7d36618 100644 --- a/README.md +++ b/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 diff --git a/SPEC.md b/SPEC.md index fd91b45..73b82f7 100644 --- a/SPEC.md +++ b/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: diff --git a/deploy/kustomize/release/kustomization.yaml b/deploy/kustomize/release/kustomization.yaml index bb65a0a..6866f22 100644 --- a/deploy/kustomize/release/kustomization.yaml +++ b/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 diff --git a/pkg/webhook/webhook.go b/pkg/webhook/webhook.go index a9ba4ae..3ff9012 100644 --- a/pkg/webhook/webhook.go +++ b/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 { diff --git a/pkg/webhook/webhook_test.go b/pkg/webhook/webhook_test.go new file mode 100644 index 0000000..00c1534 --- /dev/null +++ b/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) + } + } +}