[CORE-149029] Fix preferred affinity restart loop caused by atomic list merge semantics

CORE-149029-design-notes.md

@@ -0,0 +1,137 @@
+# CORE-149029 — Broker restart loop with admission controllers
+
+Design/investigation notes. Branch `CORE-149029-fix`. Fix committed as `c78708b3`.
+
+## Problem
+
+Kafka broker pods (e.g. `pipeline-kafka-202`) crash-loop in a delete/recreate
+cycle every ~8s during rolling upgrade. The already-merged preferred-affinity
+fix (`ignorePreferredAffinities`) was deployed, but the loop persisted.
+
+Evidence: `tmp/logs.txt` (koperator log), `tmp/202.yml` (crash-looping pod,
+ScaleOps VPA applied), `tmp/202-post.yml` (healthy pod, VPA not applied).
+
+## Root cause
+
+Two independent admission-controller mutations can trigger koperator's rolling
+upgrade; the affinity fix only covered one.
+
+1. **Affinities (already fixed):** koperator-emitted preferred affinity lists
+   are atomic (no `patchMergeKey`); a webhook appending to them looked like a
+   value change and got reverted. Only relevant when koperator itself emits
+   preferred terms (e.g. `oneBrokerPerNode: false`). Not the trigger on
+   pipeline-kafka.
+
+2. **Resources (the actual trigger):** ScaleOps VPA rewrites the live pod's
+   container `requests` after admission (kafka cpu `1`→`392m`; fluent-bit cpu
+   `100m`→`11m`, mem `256Mi`→`71420084`). koperator's last-applied annotation
+   and freshly-generated desired pod both keep the CR values.
+
+### Why the diff reverts it
+
+`patch.DefaultPatchMaker.Calculate` (k8s-objectmatcher v1.8.0) uses
+`CreateThreeWayMergePatch(original, modified, current)`. Per apimachinery
+`strategicpatch/patch.go:2070-2078`:
+
+```
+deltaMap     = diffMaps(current,  modified, {IgnoreDeletions: true})   // changes + additions
+deletionsMap = diffMaps(original, modified, {IgnoreChangesAndAdditions: true})
+patch        = merge(deletionsMap, deltaMap)
+```
+
+The change-detection half (`deltaMap`) diffs **current (live pod) vs modified
+(desired)**. `IgnoreDeletions:true` preserves webhook *additions* (annotations,
+labels, extra `podAffinity`), which is why those don't loop — but a **value
+change** to a field koperator declares (`requests.cpu`) is reverted → non-empty
+patch → rolling upgrade → pod deleted → ScaleOps re-mutates → loop.
+
+### Why the last-applied annotation doesn't help
+
+`original` (last-applied) is only consulted for the *deletions* half. The
+changes half never looks at it. So the annotation protects against foreign
+*additions* and governs *deletions*, but cannot stop koperator reverting a
+value change to a field it still declares. Same rule as `kubectl apply` vs
+HPA/VPA: an autoscaler can own a field only if you omit it from your applied
+config.
+
+## Fix — intent-aware two-way merge
+
+Trigger a rolling upgrade **iff koperator's own desired spec changed since it
+last applied it**: `diff(original, modified)`, ignoring the live pod entirely.
+
+- External mutation, CR unchanged → `original == modified` → no restart; live
+  value (VPA tuning, webhook affinity) preserved.
+- CR edit (resources or soft affinity) → `original != modified` → one restart,
+  new pod gets the CR value, then settles (ScaleOps re-tunes; next reconcile
+  `original == modified` again).
+
+The patch is only a boolean change-signal (+ logging); koperator recreates the
+pod rather than patching it in place, so two-way `diff(original, modified)` is
+exactly the right question.
+
+This **subsumes** the affinity fix and is strictly better: the old strip
+approach also *swallowed intentional soft-affinity edits made via the CR*; the
+intent diff propagates them.
+
+### Implementation (commit c78708b3)
+
+- `pkg/resources/kafka/util.go`: new `podSpecIntentChanged(currentPod,
+  desiredPod)` — `GetOriginalConfiguration` (last-applied) vs marshaled desired
+  via `strategicpatch.CreateTwoWayMergePatch`, plus a `$setElementOrder`
+  re-check. Removed `ignorePreferredAffinities()`/`deletePreferredAffinities()`.
+- `pkg/resources/kafka/kafka.go`: `handleRollingUpgrade` switch uses
+  `!intentChanged` in place of `patchResult.IsEmpty()`. The
+  `isPodTainted(currentPod)` case is unchanged and still ordered *first*.
+- `pkg/resources/kafka/util_test.go`: `TestPodSpecIntentChanged` (5 rows) and
+  `TestParkedBrokerRestartsIndependentOfIntent`.
+
+Verified: `go build ./...`, `go vet`, `gofmt -l`, full package tests — all
+clean/green.
+
+## Shredder park flow is unaffected
+
+`shredder.ethos.adobe.net/upgrade-status=parked` restart works via
+`isPodTainted(currentPod)` → `Spec.TaintedBrokersSelector` (a label lookup on
+the *live* pod), which is independent of the diff and ordered before the intent
+check. Covered by `TestParkedBrokerRestartsIndependentOfIntent`. (Config
+requirement: `TaintedBrokersSelector` must actually select that label.)
+
+## Why this matches built-in controllers
+
+Every built-in workload controller decides from recorded intent, never from a
+live-pod spec diff — which is why they don't fight admission webhooks:
+
+| Controller     | Intent signal                        | Compared against          |
+|----------------|--------------------------------------|---------------------------|
+| ReplicaSet     | replica count                        | number of owned pods      |
+| Deployment     | `pod-template-hash` (from template)  | each RS's stored template |
+| StatefulSet    | `controller-revision-hash` (template)| pod's revision label      |
+| koperator (new)| last-applied annotation              | freshly generated desired |
+
+koperator is a hand-rolled StatefulSet for Kafka; the bug was that it diffed the
+live pod instead of recorded intent. The fix is koperator's analog of the
+StatefulSet revision-hash check.
+
+## Security tradeoff (accepted)
+
+The old live-pod diff *incidentally* reverted out-of-band tampering to
+live-mutable fields (e.g. a `kubectl patch` image swap). The intent diff no
+longer does — consistent with RS/Deployment/StatefulSet, none of which revert
+live-pod drift. This was never a reliable control; the real defenses are RBAC
+least-privilege on `pods` write/exec, admission policy on UPDATE (registry
+allowlist + image signature verification), digest pinning, and audit/detection.
+If operator-level tamper detection is wanted, it should be an explicit
+detect/alert feature, not coupled to the reconcile diff (that coupling is what
+caused the loop).
+
+## Status / follow-ups
+
+- Committed `c78708b3` on `CORE-149029-fix`. **Not pushed** — `origin`
+  (`git@github-public:adobe/koperator.git`) needs the `github-public` SSH
+  identity, which is not loaded in the non-interactive session; run the push
+  interactively.
+- `make lint` blocked by an **environmental** toolchain mismatch (pinned
+  golangci-lint 2.12.2 built with go1.25.3 rejects the go1.26.0 target) — fails
+  repo-wide, unrelated to this diff. Run in CI or with a go1.26-built linter.
+- Toleration merge (`kafka.go:947-958`) likely redundant now; candidate for a
+  follow-up removal + verification.

go.mod

@@ -97,7 +97,7 @@ require (
 	github.com/jcmturner/gofork v1.7.6 // indirect
 	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
-	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/json-iterator/go v1.1.12
 	github.com/klauspost/compress v1.18.6 // indirect
 	github.com/mattn/go-colorable v0.1.15 // indirect
 	github.com/mattn/go-isatty v0.0.22 // indirect

pkg/resources/kafka/kafka.go

@@ -956,27 +956,27 @@ func (r *Reconciler) handleRollingUpgrade(log logr.Logger, desiredPod, currentPo
 		}
 		desiredPod.Spec.Tolerations = uniqueTolerations
 	}
-	// Check if the resource actually updated or if labels match TaintedBrokersSelector
-	patchResult, err := patch.DefaultPatchMaker.Calculate(currentPod, desiredPod)
+	// Decide whether a rolling upgrade is needed by comparing koperator's own
+	// desired spec against what it last applied (the last-applied annotation on
+	// the current pod) — NOT against the live pod. This keeps mutations made by
+	// admission controllers (autoscalers, webhooks) from triggering restarts,
+	// while intentional CR changes always do. Also check TaintedBrokersSelector.
+	intentChanged, patchBytes, err := podSpecIntentChanged(currentPod, desiredPod)
 	switch {
 	case err != nil:
-		log.Error(err, "could not match objects", "kind", desiredType)
+		log.Error(err, "could not compare desired pod against last-applied configuration", "kind", desiredType)
 	case r.isPodTainted(log, currentPod):
 		log.Info("pod has tainted labels, deleting it", "pod", currentPod)
-	case patchResult.IsEmpty():
+	case !intentChanged:
 		if !k8sutil.IsPodContainsTerminatedContainer(currentPod) &&
 			r.KafkaCluster.Status.BrokersState[currentPod.Labels[banzaiv1beta1.BrokerIdLabelKey]].ConfigurationState == banzaiv1beta1.ConfigInSync &&
 			!k8sutil.IsPodContainsEvictedContainer(currentPod) &&
 			!k8sutil.IsPodContainsShutdownContainer(currentPod) {
-			log.V(1).Info("resource is in sync")
+			log.V(1).Info("desired pod spec is in sync with last-applied configuration")
 			return nil
 		}
 	default:
-		log.V(1).Info("kafka pod resource diffs",
-			"patch", string(patchResult.Patch),
-			"current", string(patchResult.Current),
-			"modified", string(patchResult.Modified),
-			"original", string(patchResult.Original))
+		log.V(1).Info("kafka pod spec changed", "patch", string(patchBytes))
 	}
 
 	if err := patch.DefaultAnnotator.SetLastAppliedAnnotation(desiredPod); err != nil {

pkg/resources/kafka/util.go

@@ -20,7 +20,13 @@ import (
 	"fmt"
 	"sort"
 
+	"emperror.dev/errors"
 	"github.com/google/uuid"
+	json "github.com/json-iterator/go"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/strategicpatch"
+
+	"github.com/banzaicloud/k8s-objectmatcher/patch"
 
 	"github.com/banzaicloud/koperator/api/v1beta1"
 )
@@ -73,3 +79,61 @@ func generateRandomClusterID() string {
 	randomUUID := uuid.New()
 	return base64.URLEncoding.EncodeToString(randomUUID[:])
 }
+
+// podSpecIntentChanged reports whether koperator's own desired pod spec has
+// changed since it last applied it. It diffs the last-applied-configuration
+// annotation recorded on the current pod (original) against the freshly
+// generated desired pod (modified). The live pod (current) is deliberately
+// NOT part of the comparison.
+//
+// This is the mechanism that lets koperator coexist with any admission
+// controller (autoscalers, mutating webhooks, the node lifecycle controller):
+// a field is reconciled only when koperator's own CR-derived value differs
+// from what koperator last applied. Mutations made to the running pod by other
+// actors never enter the decision, so they can't trigger a rolling restart,
+// while intentional changes made through the KafkaCluster CR always do. It
+// therefore also handles preferred affinities generically — including the case
+// where the operator intentionally edits a soft affinity in the CR, which the
+// previous approach (stripping preferred affinities from the diff) silently
+// swallowed.
+//
+// The returned patch is the two-way strategic merge patch from original to
+// modified; it is only used as a change signal (and for logging), since a
+// rolling upgrade recreates the pod rather than patching it in place.
+func podSpecIntentChanged(currentPod, desiredPod *corev1.Pod) (bool, []byte, error) {
+	original, err := patch.DefaultAnnotator.GetOriginalConfiguration(currentPod)
+	if err != nil {
+		return false, nil, errors.WrapIf(err, "could not read last-applied configuration from current pod")
+	}
+
+	modified, err := json.ConfigCompatibleWithStandardLibrary.Marshal(desiredPod)
+	if err != nil {
+		return false, nil, errors.WrapIf(err, "could not marshal desired pod")
+	}
+	// Mirror how the last-applied annotation is produced so absent fields on one
+	// side don't masquerade as diffs.
+	if modified, _, err = patch.DeleteNullInJson(modified); err != nil {
+		return false, nil, errors.WrapIf(err, "could not clean desired pod json")
+	}
+
+	patchBytes, err := strategicpatch.CreateTwoWayMergePatch(original, modified, corev1.Pod{})
+	if err != nil {
+		return false, nil, errors.WrapIf(err, "could not create two-way merge patch")
+	}
+	if string(patchBytes) == "{}" {
+		return false, patchBytes, nil
+	}
+
+	// A $setElementOrder directive can make the patch non-empty without any
+	// actual change; confirm by applying it and re-diffing (mirrors
+	// k8s-objectmatcher's PatchMaker.Calculate).
+	patched, err := strategicpatch.StrategicMergePatch(original, patchBytes, corev1.Pod{})
+	if err != nil {
+		return false, nil, errors.WrapIf(err, "could not apply patch to last-applied configuration")
+	}
+	patchBytes, err = strategicpatch.CreateTwoWayMergePatch(original, patched, corev1.Pod{})
+	if err != nil {
+		return false, nil, errors.WrapIf(err, "could not recompute two-way merge patch")
+	}
+	return string(patchBytes) != "{}", patchBytes, nil
+}