diff --git a/.gitignore b/.gitignore index 64ead7802..ddcfb22ca 100644 --- a/.gitignore +++ b/.gitignore @@ -48,3 +48,4 @@ manager_image_patch.yaml **/go.work **/go.work.sum .claude/worktrees/** +.worktrees/ diff --git a/api/v1beta1/common_types.go b/api/v1beta1/common_types.go index 38cdd1d97..46dccfe82 100644 --- a/api/v1beta1/common_types.go +++ b/api/v1beta1/common_types.go @@ -129,6 +129,16 @@ func (r CruiseControlState) IsDownscaleStalled() bool { return r == GracefulDownscaleCompletedWithError || r == GracefulDownscalePaused } +// IsDownscaleRunning returns true only for GracefulDownscaleRunning - the task is actively executing in +// Cruise Control right now. Unlike IsDownscale(), this deliberately excludes Required (no CruiseControlOperation +// exists yet) and Scheduled (the CruiseControlOperation exists but has not been submitted to Cruise Control +// yet, e.g. because it is itself waiting behind this same roll gate) - states where no in-flight CC-side task +// can be disrupted by a capacity.json roll, so gating on the broader IsDownscale() there stalls a concurrent +// add_broker forever without ever protecting a running task (see #301 and isBrokerDeletionInProgress). +func (r CruiseControlState) IsDownscaleRunning() bool { + return r == GracefulDownscaleRunning +} + // IsRunningState returns true if CruiseControlState indicates // that the CC operation is scheduled and in-progress func (r CruiseControlState) IsRunningState() bool { diff --git a/config/samples/kraft/simplekafkacluster_kraft.yaml b/config/samples/kraft/simplekafkacluster_kraft.yaml index 5bc56fe33..c69a1ff4c 100644 --- a/config/samples/kraft/simplekafkacluster_kraft.yaml +++ b/config/samples/kraft/simplekafkacluster_kraft.yaml @@ -305,18 +305,8 @@ spec: { "min.insync.replicas": 3 } - capacityConfig: |- - { - "brokerCapacities":[ - { - "brokerId": "-1", - "capacity": { - "DISK": {"/kafka-logs-broker/kafka": "10240"}, - "CPU": {"num.cores": "1"}, - "NW_IN": "900000", - "NW_OUT": "900000" - }, - "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in cores, network throughput is in KB." - } - ] - } \ No newline at end of file + # NOTE: intentionally NO capacityConfig here. A "brokerId": "-1" universal-default entry makes + # GenerateCapacityConfig return early, so Cruise Control's capacity.json stays constant across scaling + # and the #301 capacity-roll path (and its fix) is never exercised. Omitting it forces per-broker + # capacity generation, which is exactly what the KRaft broker-scaling e2e regression-tests. Do not add + # a universal default back or the regression coverage silently disappears. \ No newline at end of file diff --git a/controllers/cruisecontroloperation_controller.go b/controllers/cruisecontroloperation_controller.go index be7d7011e..6c877b973 100644 --- a/controllers/cruisecontroloperation_controller.go +++ b/controllers/cruisecontroloperation_controller.go @@ -25,6 +25,8 @@ import ( "emperror.dev/errors" "github.com/go-logr/logr" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" apiErrors "k8s.io/apimachinery/pkg/api/errors" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -39,6 +41,7 @@ import ( apiutil "github.com/banzaicloud/koperator/api/util" banzaiv1alpha1 "github.com/banzaicloud/koperator/api/v1alpha1" banzaiv1beta1 "github.com/banzaicloud/koperator/api/v1beta1" + "github.com/banzaicloud/koperator/pkg/resources/cruisecontrol" "github.com/banzaicloud/koperator/pkg/scale" "github.com/banzaicloud/koperator/pkg/util" ) @@ -51,6 +54,13 @@ const ( ccOperationRetryExecution = "ccOperationRetryExecution" ccOperationInProgress = "ccOperationInProgress" defaultCruiseControlStatusOperationMaxDuration = time.Duration(5) * time.Minute + // stalledCCDeploymentRequeueIntervalSeconds backs off the roll gate's requeue when the CC Deployment + // rollout is wedged (ProgressDeadlineExceeded), so the error surfaced for it is not re-emitted every + // default interval for the (potentially hours-long) life of the wedge. + stalledCCDeploymentRequeueIntervalSeconds = 60 + // progressDeadlineExceededReason is the well-known reason the Deployment controller sets on the + // Progressing condition when a rollout exceeds spec.progressDeadlineSeconds. + progressDeadlineExceededReason = "ProgressDeadlineExceeded" ) var ( @@ -76,6 +86,12 @@ type CruiseControlOperationReconciler struct { // +kubebuilder:rbac:groups=kafka.banzaicloud.io,resources=cruisecontroloperations,verbs=get;list;watch;create;update;patch;delete;deletecollection // +kubebuilder:rbac:groups=kafka.banzaicloud.io,resources=cruisecontroloperations/status,verbs=get;update;patch // +kubebuilder:rbac:groups=kafka.banzaicloud.io,resources=cruisecontroloperations/finalizers,verbs=create;update;patch;delete +// The roll gate (requeueIfCCDeploymentNotRolledOut) reads the Cruise Control Deployment and ConfigMap; declare +// those reads locally so the controller's permissions stay correct even if the aggregated ClusterRole is ever +// split per-controller. These are a subset of what the KafkaCluster controller already grants, so regenerating +// the aggregated role produces no diff. +// +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch +// +kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch //nolint:gocyclo func (r *CruiseControlOperationReconciler) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.Result, error) { @@ -224,6 +240,13 @@ func (r *CruiseControlOperationReconciler) Reconcile(ctx context.Context, reques return requeueAfter(defaultRequeueIntervalInSeconds) } + // Defer any Cruise Control operation (except stop_execution) until the CC Deployment is safe to submit to + // - settled and carrying the current capacity.json - see requeueIfCCDeploymentNotRolledOut (#301). + if result, handled, err := r.requeueIfCCDeploymentNotRolledOut(ctx, log, kafkaCluster, + ccOperationExecution.CurrentTaskOperation(), ccOperationExecution.CurrentTaskParameters()); handled { + return result, err + } + log.Info("executing Cruise Control task", "operation", ccOperationExecution.CurrentTaskOperation(), "parameters", ccOperationExecution.CurrentTaskParameters()) // Executing operation cruseControlTaskResult, err := r.executeOperation(ctx, ccOperationExecution) @@ -270,6 +293,199 @@ func (r *CruiseControlOperationReconciler) addFinalizer(ctx context.Context, cur return nil } +// requeueIfCCDeploymentNotRolledOut defers a Cruise Control operation until the CC Deployment is safe to +// submit to. Any broker-affecting op (add_broker, remove_broker, rebalance, remove_disks) submitted while the +// CC Deployment is mid-rollout can be wiped: during a RollingUpdate two CC pods briefly run behind one +// Service, so the fresh pod loses the in-memory task and resets its metric-sampling window, stalling the op +// (see #301). A roll is triggered by any change hashed into the CC pod template (capacity.json, +// cruisecontrol.properties, clusterConfigs.json, log4j.properties), so this is not specific to broker scaling. +// stop_execution is never gated - it is how an operator aborts a stuck task. +// +// "Settled right now" is not enough on its own: a ConfigMap change may not have been rolled into the pod +// template yet. KafkaClusterReconciler marks a new broker GracefulUpscaleRequired (which makes the task +// controller create the add_broker op) BEFORE, and independently of, the CC reconciler that regenerates the +// ConfigMap and patches the Deployment - so an op can be picked up while the Deployment is still settled on +// the OLD config, before the roll is even triggered. We therefore also require the running pod template's +// four hash annotations - stamped by cruisecontrol.GeneratePodAnnotations under CapacityConfigHashAnnotationKey, +// ConfigHashAnnotationKey, ClusterConfigHashAnnotationKey and LogConfigHashAnnotationKey - to each match the +// corresponding entry in the current ConfigMap for EVERY gated op (a rebalance/remove_disks would likewise be +// wiped by a roll triggered by any of capacity.json, cruisecontrol.properties, clusterConfigs.json or +// log4j.properties changing and landing just after submission - not just capacity). add_broker additionally +// requires capacity.json to contain the target broker(s), so it runs against a CC that has loaded their exact +// capacity (no dependency on capacity estimation). This is an implicit contract with the CC reconciler: both +// sides compute the same hash for each entry; keep them in sync (see GeneratePodAnnotations / TestCapacityConfigHash). +// +// Every check fails open when the evidence to gate on is absent (no Deployment, no ConfigMap, or a given +// hashed entry not koperator-managed) so it can never deadlock an operation. The returned bool reports +// whether the caller should return the (result, error) as-is. +func (r *CruiseControlOperationReconciler) requeueIfCCDeploymentNotRolledOut(ctx context.Context, log logr.Logger, + kafkaCluster *banzaiv1beta1.KafkaCluster, op banzaiv1alpha1.CruiseControlTaskOperation, parameters map[string]string) (ctrl.Result, bool, error) { + // stop_execution aborts an in-flight task and must never be deferred. + if op == banzaiv1alpha1.OperationStopExecution { + return ctrl.Result{}, false, nil + } + + // Read the Deployment and ConfigMap through the non-cached API reader: freshness is part of this gate's + // correctness contract. A lagging informer cache could show the OLD ConfigMap/Deployment annotations as + // matching, let an op be submitted, and then have the real update land and roll Cruise Control - killing + // the just-submitted task, the exact #301 race this gate prevents. This is a low-frequency path (only + // selected non-stop_execution ops reach it), so the direct read is affordable. Fall back to the cached + // client for manually constructed reconcilers (e.g. unit tests) that leave DirectClient unset. + reader := r.DirectClient + if reader == nil { + reader = r.Client + } + + deployment := &appsv1.Deployment{} + deploymentKey := client.ObjectKey{ + Name: cruisecontrol.DeploymentName(kafkaCluster), + Namespace: kafkaCluster.Namespace, + } + if err := reader.Get(ctx, deploymentKey, deployment); err != nil { + if apiErrors.IsNotFound(err) { + // No Cruise Control Deployment: no rollout in progress to race with, so do not gate. + return ctrl.Result{}, false, nil + } + result, wErr := requeueWithError(log, "could not determine Cruise Control Deployment rollout state", err) + return result, true, wErr + } + + // No op may be submitted to a CC that is mid-rollout - a restart wipes the in-flight task. + if isDeploymentRolling(deployment) { + requeueInterval := defaultRequeueIntervalInSeconds + if deploymentRolloutTimedOut(deployment) { + // The old CC pod keeps the Service up (so CC still reports ready) while the new pod never becomes + // available - the operation would otherwise defer forever with only a routine log. Surface it, and + // back off so a long-lived wedge does not re-emit this error every default interval. + log.Error(errors.New("Cruise Control Deployment rollout exceeded its progress deadline"), + "deferring Cruise Control operation on a wedged rollout - inspect the Cruise Control pod (image/crashloop); the operation stays deferred until the roll completes", + "operation", op, "deployment", deployment.Name, + "generation", deployment.Generation, "observedGeneration", deployment.Status.ObservedGeneration, + "replicas", deployment.Status.Replicas, "updatedReplicas", deployment.Status.UpdatedReplicas) + requeueInterval = stalledCCDeploymentRequeueIntervalSeconds + } else { + log.Info("requeue: Cruise Control Deployment is mid-rollout; deferring operation to avoid racing a CC restart", + "operation", op, "deployment", deployment.Name, + "generation", deployment.Generation, "observedGeneration", deployment.Status.ObservedGeneration, + "replicas", deployment.Status.Replicas, "updatedReplicas", deployment.Status.UpdatedReplicas) + } + result, _ := requeueAfter(requeueInterval) + return result, true, nil + } + + // Pre-roll protection: the Deployment is settled, but a ConfigMap change may not have been rolled into the + // pod template yet. This applies to EVERY gated op - a rebalance/remove_disks submitted now would also be + // wiped by a roll (triggered by any of the four hashed entries below) that lands just after (see doc + // comment). Each entry is checked independently: an absent annotation means that particular entry is not + // koperator-managed on this Deployment (e.g. a custom capacity annotation, or a Deployment that predates + // one of these annotations), so there is nothing to wait for on that axis specifically. + configMap := &corev1.ConfigMap{} + configMapKey := client.ObjectKey{ + Name: cruisecontrol.ConfigMapName(kafkaCluster), + Namespace: kafkaCluster.Namespace, + } + if err := reader.Get(ctx, configMapKey, configMap); err != nil { + if apiErrors.IsNotFound(err) { + // No Cruise Control ConfigMap to compare against yet; do not gate. + return ctrl.Result{}, false, nil + } + result, wErr := requeueWithError(log, "could not read Cruise Control ConfigMap", err) + return result, true, wErr + } + + hashedConfigMapEntries := []struct { + annotationKey string + content string + }{ + {cruisecontrol.ConfigHashAnnotationKey, configMap.Data[cruisecontrol.PropertiesConfigMapKey]}, + {cruisecontrol.ClusterConfigHashAnnotationKey, configMap.Data[cruisecontrol.ClusterConfigsConfigMapKey]}, + {cruisecontrol.LogConfigHashAnnotationKey, configMap.Data[cruisecontrol.Log4jConfigMapKey]}, + {cruisecontrol.CapacityConfigHashAnnotationKey, configMap.Data[cruisecontrol.CapacityConfigMapKey]}, + } + for _, entry := range hashedConfigMapEntries { + deployedHash, ok := deployment.Spec.Template.Annotations[entry.annotationKey] + if !ok { + continue + } + if deployedHash != cruisecontrol.ConfigHash(entry.content) { + log.Info("requeue: Cruise Control pod template does not yet carry the current ConfigMap; deferring operation until the roll settles", + "operation", op, "annotation", entry.annotationKey, + "deployedHash", deployedHash, "expectedHash", cruisecontrol.ConfigHash(entry.content), + "generation", deployment.Generation, "observedGeneration", deployment.Status.ObservedGeneration) + result, _ := requeueAfter(defaultRequeueIntervalInSeconds) + return result, true, nil + } + } + + // Only add_broker additionally needs the target broker's capacity present before it runs. + if op == banzaiv1alpha1.OperationAddBroker { + brokerIDs := splitNonEmpty(parameters[scale.ParamBrokerID]) + hasAll, err := cruisecontrol.CapacityConfigContainsBrokers(configMap.Data[cruisecontrol.CapacityConfigMapKey], brokerIDs) + if err != nil { + result, wErr := requeueWithError(log, "could not inspect Cruise Control capacity config for the brokers being added", err) + return result, true, wErr + } + if !hasAll { + log.Info("requeue: capacity.json does not yet contain the broker(s) being added; deferring add_broker until their capacity is generated", + "operation", op, "brokerIDs", brokerIDs, "configMap", configMapKey.Name, "deployment", deployment.Name) + result, _ := requeueAfter(defaultRequeueIntervalInSeconds) + return result, true, nil + } + } + + return ctrl.Result{}, false, nil +} + +// splitNonEmpty splits a comma-separated parameter value, returning nil (rather than a single empty element) +// for an empty string so an absent parameter yields no broker ids. +func splitNonEmpty(s string) []string { + if s == "" { + return nil + } + return strings.Split(s, ",") +} + +// isDeploymentRolling reports whether a Deployment is actively in the middle of a rollout: a new pod +// template has been applied but not yet observed by the Deployment controller, its pods have surged above +// the desired count, or not all running replicas are the latest revision yet. It deliberately keys off +// positive evidence of an in-progress rollout rather than "fully settled" so that a Deployment whose status +// has never been populated (observedGeneration == 0, e.g. under envtest where no Deployment controller +// runs) reads as NOT rolling and the check does not block. Initial CruiseControl availability is enforced +// separately by CruiseControlStatus.IsReady; this gate only guards against submitting a broker operation +// while an already-running CC is being re-rolled (e.g. by a capacity.json change). +func isDeploymentRolling(deployment *appsv1.Deployment) bool { + specReplicas := int32(1) + if deployment.Spec.Replicas != nil { + specReplicas = *deployment.Spec.Replicas + } + s := deployment.Status + switch { + case s.ObservedGeneration != 0 && deployment.Generation > s.ObservedGeneration: + // A new pod template was applied but the Deployment controller has not observed it yet. + return true + case s.Replicas > specReplicas: + // RollingUpdate surge: an old-revision pod is still running alongside the new one. + return true + case s.UpdatedReplicas < s.Replicas: + // Not all running pods are the latest revision yet. + return true + default: + return false + } +} + +// deploymentRolloutTimedOut reports whether the Deployment controller has given up on the current rollout +// (Progressing=False with reason ProgressDeadlineExceeded), i.e. the new pod never became available - a +// wedged rollout the operator would otherwise defer a broker operation behind indefinitely. +func deploymentRolloutTimedOut(deployment *appsv1.Deployment) bool { + for _, c := range deployment.Status.Conditions { + if c.Type == appsv1.DeploymentProgressing && c.Reason == progressDeadlineExceededReason { + return true + } + } + return false +} + func (r *CruiseControlOperationReconciler) executeOperation(ctx context.Context, ccOperationExecution *banzaiv1alpha1.CruiseControlOperation) (*scale.Result, error) { var cruseControlTaskResult *scale.Result var err error diff --git a/controllers/cruisecontroloperation_controller_test.go b/controllers/cruisecontroloperation_controller_test.go index 2cbb84ddf..9934e7cf2 100644 --- a/controllers/cruisecontroloperation_controller_test.go +++ b/controllers/cruisecontroloperation_controller_test.go @@ -23,15 +23,22 @@ import ( "github.com/go-logr/logr" "github.com/stretchr/testify/assert" "go.uber.org/mock/gomock" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apiErrors "k8s.io/apimachinery/pkg/api/errors" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + k8sscheme "k8s.io/client-go/kubernetes/scheme" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" "github.com/banzaicloud/koperator/api/v1alpha1" "github.com/banzaicloud/koperator/api/v1beta1" mocks "github.com/banzaicloud/koperator/controllers/tests/mocks" + "github.com/banzaicloud/koperator/pkg/resources/cruisecontrol" "github.com/banzaicloud/koperator/pkg/scale" + "github.com/banzaicloud/koperator/pkg/util" ) func createCCRetryExecutionOperation(createTime time.Time, id string, operation v1alpha1.CruiseControlTaskOperation) *v1alpha1.CruiseControlOperation { @@ -54,6 +61,385 @@ func createCCRetryExecutionOperation(createTime time.Time, id string, operation } } +func TestIsDeploymentRolling(t *testing.T) { + dep := func(generation, observedGeneration int64, specReplicas, replicas, updated int32) *appsv1.Deployment { + return &appsv1.Deployment{ + ObjectMeta: v1.ObjectMeta{Generation: generation}, + Spec: appsv1.DeploymentSpec{Replicas: util.Int32Pointer(specReplicas)}, + Status: appsv1.DeploymentStatus{ + ObservedGeneration: observedGeneration, + Replicas: replicas, + UpdatedReplicas: updated, + }, + } + } + + tests := []struct { + name string + d *appsv1.Deployment + want bool + }{ + {"settled single replica is not rolling", dep(3, 3, 1, 1, 1), false}, + {"new pod template not yet observed is rolling", dep(4, 3, 1, 1, 1), true}, + {"surge: an old-revision pod still present is rolling", dep(3, 3, 1, 2, 1), true}, + {"not all running replicas updated yet is rolling", dep(3, 3, 1, 1, 0), true}, + // envtest / no Deployment controller: status never populated (observedGeneration == 0). Must read + // as NOT rolling so the gate does not block where nothing rolls the Deployment. + {"unpopulated status (observedGeneration 0) is not rolling", dep(1, 0, 1, 0, 0), false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, isDeploymentRolling(tt.d)) + }) + } + + t.Run("nil spec.replicas defaults to 1; settled is not rolling", func(t *testing.T) { + d := &appsv1.Deployment{ + ObjectMeta: v1.ObjectMeta{Generation: 1}, + Status: appsv1.DeploymentStatus{ObservedGeneration: 1, Replicas: 1, UpdatedReplicas: 1}, + } + assert.False(t, isDeploymentRolling(d)) + }) +} + +// TestRequeueIfCCDeploymentNotRolledOut directly exercises the broker-op gate's branching: it must requeue +// (defer) on a mismatched capacity hash and on an add_broker whose target capacity is not yet in +// capacity.json, execute once both hold, and fail open when the Deployment/ConfigMap is absent or capacity +// is not koperator-managed - the deferral behavior the e2e (which only asserts the eventual matched state) +// cannot prove. +// +//nolint:funlen +func TestRequeueIfCCDeploymentNotRolledOut(t *testing.T) { + const namespace = "kafka" + kc := &v1beta1.KafkaCluster{ObjectMeta: v1.ObjectMeta{Name: "kafka", Namespace: namespace}} + depName := cruisecontrol.DeploymentName(kc) + cmName := cruisecontrol.ConfigMapName(kc) + addParams := map[string]string{scale.ParamBrokerID: "103"} + + capacityWith103 := `{"brokerCapacities":[{"brokerId":"100"},{"brokerId":"103"}]}` + capacityNo103 := `{"brokerCapacities":[{"brokerId":"100"}]}` + + // settledDeployment is a fully rolled-out CC Deployment carrying the given capacity-hash annotation (nil + // annotations => capacity is not koperator-managed). + settledDeployment := func(annotations map[string]string) *appsv1.Deployment { + return &appsv1.Deployment{ + ObjectMeta: v1.ObjectMeta{Name: depName, Namespace: namespace, Generation: 1}, + Spec: appsv1.DeploymentSpec{ + Replicas: util.Int32Pointer(1), + Template: corev1.PodTemplateSpec{ObjectMeta: v1.ObjectMeta{Annotations: annotations}}, + }, + Status: appsv1.DeploymentStatus{ObservedGeneration: 1, Replicas: 1, UpdatedReplicas: 1}, + } + } + hashAnnotation := func(capacityJSON string) map[string]string { + return map[string]string{cruisecontrol.CapacityConfigHashAnnotationKey: cruisecontrol.CapacityConfigHash(capacityJSON)} + } + rollingDeployment := func() *appsv1.Deployment { + d := settledDeployment(hashAnnotation(capacityWith103)) + d.Generation = 2 // new pod template not yet observed => rolling + return d + } + configMap := func(capacityJSON string) *corev1.ConfigMap { + return &corev1.ConfigMap{ + ObjectMeta: v1.ObjectMeta{Name: cmName, Namespace: namespace}, + Data: map[string]string{cruisecontrol.CapacityConfigMapKey: capacityJSON}, + } + } + configMapWithProperties := func(capacityJSON, ccProperties string) *corev1.ConfigMap { + cm := configMap(capacityJSON) + cm.Data[cruisecontrol.PropertiesConfigMapKey] = ccProperties + return cm + } + + tests := []struct { + name string + op v1alpha1.CruiseControlTaskOperation + params map[string]string + objects []client.Object + wantHandled bool // true => the caller returns immediately (deferred / gated) + wantRequeue bool // true => the returned result asks for a requeue + }{ + { + name: "stop_execution is never gated, even mid-rollout", + op: v1alpha1.OperationStopExecution, + objects: []client.Object{rollingDeployment()}, + }, + { + name: "missing CC Deployment fails open", + op: v1alpha1.OperationAddBroker, + params: addParams, + objects: nil, + }, + { + name: "rebalance is deferred while the Deployment is rolling", + op: v1alpha1.OperationRebalance, + objects: []client.Object{rollingDeployment()}, + wantHandled: true, + wantRequeue: true, + }, + { + name: "remove_disks is deferred while the Deployment is rolling", + op: v1alpha1.OperationRemoveDisks, + objects: []client.Object{rollingDeployment()}, + wantHandled: true, + wantRequeue: true, + }, + { + // No capacity annotation => capacity is not koperator-managed, nothing to wait for. + name: "rebalance on a settled Deployment with no capacity annotation fails open", + op: v1alpha1.OperationRebalance, + objects: []client.Object{settledDeployment(nil)}, + }, + { + // Pre-roll protection now covers non-broker ops too: a settled Deployment whose capacity hash + // still lags the ConfigMap defers rebalance/remove_disks, not just broker add/remove. + name: "rebalance defers while the pod template capacity hash lags the ConfigMap", + op: v1alpha1.OperationRebalance, + objects: []client.Object{settledDeployment(map[string]string{cruisecontrol.CapacityConfigHashAnnotationKey: "stale"}), configMap(capacityWith103)}, + wantHandled: true, + wantRequeue: true, + }, + { + name: "rebalance executes once the pod template carries the current capacity hash", + op: v1alpha1.OperationRebalance, + objects: []client.Object{settledDeployment(hashAnnotation(capacityWith103)), configMap(capacityWith103)}, + }, + { + // Capacity hash matches, but cruisecontrol.properties changed and has not been rolled into the + // pod template yet: a rebalance submitted now would still be wiped by the roll that follows, so + // the gate must cover this hash too, not just capacity.json. + name: "rebalance defers while the pod template cruisecontrol.properties hash lags the ConfigMap", + op: v1alpha1.OperationRebalance, + objects: []client.Object{ + settledDeployment(util.MergeAnnotations(hashAnnotation(capacityWith103), map[string]string{cruisecontrol.ConfigHashAnnotationKey: "stale"})), + configMapWithProperties(capacityWith103, "some.property=value"), + }, + wantHandled: true, + wantRequeue: true, + }, + { + // Same gap, exercised via clusterConfigs.json / log4j.properties (ClusterConfigHashAnnotationKey / + // LogConfigHashAnnotationKey) so all three non-capacity hashes are covered, not just one. + name: "rebalance defers while the pod template clusterConfigs.json hash lags the ConfigMap", + op: v1alpha1.OperationRebalance, + objects: []client.Object{ + settledDeployment(util.MergeAnnotations(hashAnnotation(capacityWith103), map[string]string{cruisecontrol.ClusterConfigHashAnnotationKey: "stale"})), + configMap(capacityWith103), + }, + wantHandled: true, + wantRequeue: true, + }, + { + name: "rebalance executes once all four hashes match the ConfigMap", + op: v1alpha1.OperationRebalance, + objects: []client.Object{ + settledDeployment(util.MergeAnnotations(hashAnnotation(capacityWith103), map[string]string{ + cruisecontrol.ConfigHashAnnotationKey: cruisecontrol.ConfigHash("some.property=value"), + cruisecontrol.ClusterConfigHashAnnotationKey: cruisecontrol.ConfigHash(""), + cruisecontrol.LogConfigHashAnnotationKey: cruisecontrol.ConfigHash(""), + })), + configMapWithProperties(capacityWith103, "some.property=value"), + }, + }, + { + name: "Deployment mid-rollout requeues", + op: v1alpha1.OperationAddBroker, + params: addParams, + objects: []client.Object{rollingDeployment(), configMap(capacityWith103)}, + wantHandled: true, + wantRequeue: true, + }, + { + name: "settled without capacity annotation fails open (capacity not koperator-managed)", + op: v1alpha1.OperationAddBroker, + params: addParams, + objects: []client.Object{settledDeployment(nil), configMap(capacityWith103)}, + }, + { + name: "settled, matching hash, missing ConfigMap fails open", + op: v1alpha1.OperationAddBroker, + params: addParams, + objects: []client.Object{settledDeployment(hashAnnotation(capacityWith103))}, + }, + { + name: "capacity hash mismatch requeues", + op: v1alpha1.OperationAddBroker, + params: addParams, + objects: []client.Object{settledDeployment(map[string]string{cruisecontrol.CapacityConfigHashAnnotationKey: "stale"}), configMap(capacityWith103)}, + wantHandled: true, + wantRequeue: true, + }, + { + name: "add_broker requeues while target broker capacity is absent", + op: v1alpha1.OperationAddBroker, + params: addParams, + objects: []client.Object{settledDeployment(hashAnnotation(capacityNo103)), configMap(capacityNo103)}, + wantHandled: true, + wantRequeue: true, + }, + { + name: "add_broker executes once capacity is rolled and contains the target broker", + op: v1alpha1.OperationAddBroker, + params: addParams, + objects: []client.Object{settledDeployment(hashAnnotation(capacityWith103)), configMap(capacityWith103)}, + }, + { + name: "remove_broker executes on a settled, hash-matched CC (no broker-presence check)", + op: v1alpha1.OperationRemoveBroker, + params: map[string]string{scale.ParamBrokerID: "103"}, + objects: []client.Object{settledDeployment(hashAnnotation(capacityWith103)), configMap(capacityWith103)}, + }, + } + + scheme := runtime.NewScheme() + assert.NoError(t, k8sscheme.AddToScheme(scheme)) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // DirectClient deliberately left unset here so the whole table also exercises the nil-DirectClient + // fallback to the cached Client (see the reader-selection in requeueIfCCDeploymentNotRolledOut). + fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tt.objects...).Build() + r := &CruiseControlOperationReconciler{Client: fakeClient} + + result, handled, err := r.requeueIfCCDeploymentNotRolledOut(context.Background(), logr.Discard(), kc, tt.op, tt.params) + assert.NoError(t, err) + assert.Equal(t, tt.wantHandled, handled, "handled") + assert.Equal(t, tt.wantRequeue, result.RequeueAfter > 0, "requeue") + }) + } + + // The gate's correctness depends on reading the freshest Deployment/ConfigMap: a lagging informer cache + // could show the OLD state as matching and let an op be submitted just before a roll wipes it (the #301 + // race). It must therefore read through DirectClient (the non-cached API reader) when one is set. We prove + // this by making the cached Client and DirectClient disagree: a stale cache showing a settled, hash-matched + // CC (which would execute) versus a fresh direct view showing a mid-rollout CC (which must defer). The gate + // must follow the fresh view and defer. + t.Run("reads through DirectClient when set (fresh mid-rollout view defers over a stale settled cache)", func(t *testing.T) { + staleCache := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(settledDeployment(hashAnnotation(capacityWith103)), configMap(capacityWith103)).Build() + freshDirect := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(rollingDeployment(), configMap(capacityWith103)).Build() + r := &CruiseControlOperationReconciler{Client: staleCache, DirectClient: freshDirect} + + result, handled, err := r.requeueIfCCDeploymentNotRolledOut(context.Background(), logr.Discard(), kc, v1alpha1.OperationAddBroker, addParams) + assert.NoError(t, err) + assert.True(t, handled, "gate must defer using DirectClient's fresh mid-rollout view, not the stale settled cache") + assert.True(t, result.RequeueAfter > 0, "requeue") + }) + + // Symmetric proof it is really DirectClient (not Client) being consulted: a stale cache showing a + // mid-rollout CC (which would defer) versus a fresh direct view showing a settled, hash-matched CC that + // contains the target broker (which must execute). The gate must follow the fresh view and execute. + t.Run("reads through DirectClient when set (fresh settled view executes over a stale rolling cache)", func(t *testing.T) { + staleCache := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(rollingDeployment(), configMap(capacityWith103)).Build() + freshDirect := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(settledDeployment(hashAnnotation(capacityWith103)), configMap(capacityWith103)).Build() + r := &CruiseControlOperationReconciler{Client: staleCache, DirectClient: freshDirect} + + result, handled, err := r.requeueIfCCDeploymentNotRolledOut(context.Background(), logr.Discard(), kc, v1alpha1.OperationAddBroker, addParams) + assert.NoError(t, err) + assert.False(t, handled, "gate must execute using DirectClient's fresh settled view, not the stale rolling cache") + assert.Zero(t, result.RequeueAfter, "no requeue") + }) + + // A wedged rollout (mid-rollout AND Progressing=False/ProgressDeadlineExceeded) is the behavior most novel + // to this change: the op must still be deferred (the new CC pod never became available, so submitting would + // race a restart), but with the longer backoff so the surfaced error is not re-emitted every default + // interval for the life of the wedge. Assert the exact stalled interval, which the boolean table cannot. + t.Run("wedged rollout defers with the stalled backoff interval", func(t *testing.T) { + wedged := rollingDeployment() + wedged.Status.Conditions = []appsv1.DeploymentCondition{ + {Type: appsv1.DeploymentProgressing, Status: corev1.ConditionFalse, Reason: progressDeadlineExceededReason}, + } + fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(wedged, configMap(capacityWith103)).Build() + r := &CruiseControlOperationReconciler{Client: fakeClient} + + result, handled, err := r.requeueIfCCDeploymentNotRolledOut(context.Background(), logr.Discard(), kc, v1alpha1.OperationAddBroker, addParams) + assert.NoError(t, err) + assert.True(t, handled, "wedged rollout must defer the operation") + assert.Equal(t, time.Duration(stalledCCDeploymentRequeueIntervalSeconds)*time.Second, result.RequeueAfter, + "wedged rollout must back off with the stalled interval, not the default") + }) + + // A failing (non-NotFound) Deployment read must surface an error and defer, not fail open - none of the + // table cases (all assert NoError) cover the requeueWithError branch. Use an interceptor that errors on the + // Deployment Get. + t.Run("errored Deployment read surfaces the error and defers", func(t *testing.T) { + failing := interceptor.Funcs{ + Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*appsv1.Deployment); ok { + return apiErrors.NewServiceUnavailable("apiserver down") + } + return c.Get(ctx, key, obj, opts...) + }, + } + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(settledDeployment(hashAnnotation(capacityWith103)), configMap(capacityWith103)). + WithInterceptorFuncs(failing).Build() + r := &CruiseControlOperationReconciler{Client: fakeClient} + + _, handled, err := r.requeueIfCCDeploymentNotRolledOut(context.Background(), logr.Discard(), kc, v1alpha1.OperationAddBroker, addParams) + assert.Error(t, err, "a non-NotFound Deployment read error must be surfaced, not swallowed") + assert.True(t, handled, "an errored read must defer, not fail open") + // requeueWithError requeues via the non-nil error (RequeueAfter stays 0), so do not assert on it here. + }) + + // The sibling of the Deployment-read error: a failing (non-NotFound) ConfigMap read must also surface the + // error and defer, not fail open (the Deployment is read first, so it must succeed for the ConfigMap read + // to be reached). + t.Run("errored ConfigMap read surfaces the error and defers", func(t *testing.T) { + failing := interceptor.Funcs{ + Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*corev1.ConfigMap); ok { + return apiErrors.NewServiceUnavailable("apiserver down") + } + return c.Get(ctx, key, obj, opts...) + }, + } + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(settledDeployment(hashAnnotation(capacityWith103)), configMap(capacityWith103)). + WithInterceptorFuncs(failing).Build() + r := &CruiseControlOperationReconciler{Client: fakeClient} + + _, handled, err := r.requeueIfCCDeploymentNotRolledOut(context.Background(), logr.Discard(), kc, v1alpha1.OperationAddBroker, addParams) + assert.Error(t, err, "a non-NotFound ConfigMap read error must be surfaced, not swallowed") + assert.True(t, handled, "an errored read must defer, not fail open") + }) + + // An add_broker whose capacity.json is present and hash-matched but unparseable must surface the + // CapacityConfigContainsBrokers error and defer, not fail open - exercising that error branch through the + // gate (TestCapacityConfigContainsBrokers only covers the helper in isolation). + t.Run("add_broker with unparseable capacity.json surfaces the error and defers", func(t *testing.T) { + malformed := "{not json" + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(settledDeployment(hashAnnotation(malformed)), configMap(malformed)).Build() + r := &CruiseControlOperationReconciler{Client: fakeClient} + + _, handled, err := r.requeueIfCCDeploymentNotRolledOut(context.Background(), logr.Discard(), kc, v1alpha1.OperationAddBroker, addParams) + assert.Error(t, err, "an unparseable capacity.json must surface the inspection error, not fail open") + assert.True(t, handled, "an errored capacity inspection must defer, not fail open") + }) +} + +func TestDeploymentRolloutTimedOut(t *testing.T) { + timedOut := &appsv1.Deployment{Status: appsv1.DeploymentStatus{Conditions: []appsv1.DeploymentCondition{ + {Type: appsv1.DeploymentProgressing, Status: corev1.ConditionFalse, Reason: progressDeadlineExceededReason}, + }}} + progressing := &appsv1.Deployment{Status: appsv1.DeploymentStatus{Conditions: []appsv1.DeploymentCondition{ + {Type: appsv1.DeploymentProgressing, Status: corev1.ConditionTrue, Reason: "NewReplicaSetAvailable"}, + }}} + assert.True(t, deploymentRolloutTimedOut(timedOut)) + assert.False(t, deploymentRolloutTimedOut(progressing)) + assert.False(t, deploymentRolloutTimedOut(&appsv1.Deployment{})) +} + +func TestSplitNonEmpty(t *testing.T) { + assert.Nil(t, splitNonEmpty("")) + assert.Equal(t, []string{"103"}, splitNonEmpty("103")) + assert.Equal(t, []string{"100", "101"}, splitNonEmpty("100,101")) +} + func TestSortOperations(t *testing.T) { timeNow := time.Now() testCases := []struct { diff --git a/pkg/resources/cruisecontrol/configmap.go b/pkg/resources/cruisecontrol/configmap.go index b121f70dc..f0277139e 100644 --- a/pkg/resources/cruisecontrol/configmap.go +++ b/pkg/resources/cruisecontrol/configmap.go @@ -48,6 +48,17 @@ const ( storageConfigNWINDefaultValue = "125000" storageConfigNWOUTDefaultValue = "125000" defaultDoc = "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB." + + // PropertiesConfigMapKey, ClusterConfigsConfigMapKey, Log4jConfigMapKey and CapacityConfigMapKey are the + // keys under which the four hashed entries are stored in the Cruise Control ConfigMap. Exported so all three + // sites that must agree on them - the write side (configMap), the pod-template stamp side + // (GeneratePodAnnotations) and the operation controller's roll gate - share one source of truth instead of + // duplicating the literals, where a rename would silently break a lookup (make the gate hash an empty + // string, never match the deployed annotation, and defer operations forever - the #301 stall class). + PropertiesConfigMapKey = "cruisecontrol.properties" + ClusterConfigsConfigMapKey = "clusterConfigs.json" + Log4jConfigMapKey = "log4j.properties" + CapacityConfigMapKey = "capacity.json" ) func (r *Reconciler) configMap(clientPass string, capacityConfig string, log logr.Logger) runtime.Object { @@ -100,10 +111,10 @@ func (r *Reconciler) configMap(clientPass string, capacityConfig string, log log r.KafkaCluster, ), Data: map[string]string{ - "cruisecontrol.properties": ccConfig.String(), - "capacity.json": capacityConfig, - "clusterConfigs.json": r.KafkaCluster.Spec.CruiseControlConfig.ClusterConfig, - "log4j.properties": r.KafkaCluster.Spec.CruiseControlConfig.GetCCLog4jConfig(), + PropertiesConfigMapKey: ccConfig.String(), + CapacityConfigMapKey: capacityConfig, + ClusterConfigsConfigMapKey: r.KafkaCluster.Spec.CruiseControlConfig.ClusterConfig, + Log4jConfigMapKey: r.KafkaCluster.Spec.CruiseControlConfig.GetCCLog4jConfig(), }, } return configMap @@ -169,17 +180,11 @@ func GenerateCapacityConfig(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger, return "", errors.Wrap(err, "could not unmarshal the user-provided broker capacity config") } for _, brokerCapacity := range capacityConfig.Capacities { - brokerCapacityMap, ok := brokerCapacity.(map[string]interface{}) - if !ok { - continue - } - brokerId, ok, err := unstructured.NestedString(brokerCapacityMap, v1beta1.BrokerIdLabelKey) + brokerId, err := brokerIDFromCapacityEntry(brokerCapacity) if err != nil { - return "", errors.WrapIfWithDetails(err, - "could not retrieve broker Id from broker capacity configuration", - "capacity configuration", brokerCapacityMap) + return "", err } - if !ok { + if brokerId == "" { continue } // If the -1 default exists we don't have to do anything else here since all brokers will have values. @@ -190,12 +195,25 @@ func GenerateCapacityConfig(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger, userConfigBrokerIds = append(userConfigBrokerIds, brokerId) } } - // During cluster downscale the CR does not contain data for brokers being downscaled which is - // required to generate the proper capacity json for CC so we are reusing the old one. - // We can only remove brokers from capacity config when they were removed (pods deleted) from CC as well. + // During a scaling operation the CR does not carry capacity data for every broker Cruise Control still + // knows about (a broker dropped from the spec but not yet deleted from CC), so we reuse the already + // deployed capacity.json instead of regenerating a fallback entry for it - regenerating changes + // capacity.json, which rolls the CC Deployment mid-scaling (see #301 and isBrokerRemovalPending). We + // only ADD entries for brokers that have newly joined (present in the spec/status but missing from the + // deployed config) so a mixed add+remove edit still writes the new broker's capacity before add_broker. + // A departing broker keeps its deployed entry until its pod is gone. if config != nil { - if data, ok := config.Data["capacity.json"]; ok { - return data, err + if data, ok := config.Data[CapacityConfigMapKey]; ok { + // While a downscale is actively running there is a remove_broker task in flight on the CC pod; + // appending a concurrently-added broker here would change capacity.json, roll the CC Deployment, + // and kill that in-flight task - the exact #301 failure class. Reuse the deployed config verbatim + // in that window. The added broker's capacity is written by the next full regeneration once the + // downscale completes, and its add_broker is deferred until then by the operation controller's roll + // gate (requeueIfCCDeploymentNotRolledOut), so nothing is lost. + if isBrokerDeletionInProgress(kafkaCluster.Status.BrokersState) { + return data, nil + } + return mergeCapacityConfig(kafkaCluster, log, data, capacityConfig.Capacities, userConfigBrokerIds) } } @@ -215,6 +233,149 @@ func GenerateCapacityConfig(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger, return string(result), err } +// mergeCapacityConfig returns the already-deployed capacity.json augmented with entries for brokers that +// have joined the cluster since it was written (present in the spec/status but absent from the deployed +// config). A newly added broker takes its user-provided capacity when the CR defines one, otherwise a +// generated entry, so add_broker always has capacity data - even in a mixed add+remove edit. Entries +// already in the deployed config - including brokers being removed (dropped from the spec but still in the +// status) - are preserved verbatim so their capacity is not rewritten to the fallback default; rewriting it +// would change capacity.json and roll Cruise Control mid-scaling (see #301). +// +// When no broker needs to be added the deployed config is returned byte-for-byte unchanged so no spurious +// roll happens - this preserves the pre-existing "reuse the deployed capacity.json during downscale" +// behaviour exactly, while additionally covering a mixed add+remove edit. A side effect of preserving +// deployed entries verbatim is that a capacity change to a broker that stays in the cluster (e.g. a disk +// resize) is not written until the reuse window closes and full regeneration resumes; this is invoked only +// while a broker is departing (see isBrokerRemovalPending / isBrokerDeletionInProgress in Reconcile). +// +// The reuse window is bounded only for a healthy downscale: it closes once the departing broker's pod is gone +// and it drops out of the status. If a removal wedges - the broker stays in the status but out of the spec +// indefinitely (a stuck/paused downscale that is never resolved) - isBrokerRemovalPending stays true and this +// suppression of staying-brokers' capacity changes is unbounded until an operator resolves the removal. That +// is an accepted trade-off: a wedged downscale already needs manual investigation, and rolling CC to write an +// unrelated capacity change while a removal is stuck would not help. +func mergeCapacityConfig(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger, deployedCapacityConfig string, userCapacities []interface{}, userConfigBrokerIds []string) (string, error) { + var deployed JBODInvariantCapacityConfig + if err := json.Unmarshal([]byte(deployedCapacityConfig), &deployed); err != nil { + // The deployed capacity.json is unexpectedly unparseable; keep reusing it verbatim rather than + // regenerating from scratch, which could roll Cruise Control mid-scaling. + log.Error(err, "could not parse deployed cruise control capacity config, reusing it verbatim") + return deployedCapacityConfig, nil + } + + // Broker ids already present in the deployed config. Their entries win (kept verbatim) so we neither + // generate nor re-add them, which avoids rolling Cruise Control for brokers that already have capacity. + deployedBrokerIds := make(map[string]struct{}, len(deployed.Capacities)) + for _, brokerCapacity := range deployed.Capacities { + brokerId, err := brokerIDFromCapacityEntry(brokerCapacity) + if err != nil { + return "", err + } + if brokerId == "-1" { + // A "-1" universal-default entry already covers every broker, so nothing needs appending and + // appending a redundant per-broker entry would only change capacity.json and roll CC. Reuse + // verbatim. (Unreachable via the normal flow - a user "-1" makes GenerateCapacityConfig return + // early before merge - but guard against a deployed config that ever carries one.) + return deployedCapacityConfig, nil + } + if brokerId != "" { + deployedBrokerIds[brokerId] = struct{}{} + } + } + + // User-provided capacities for brokers that are not yet in the deployed config are the explicit + // capacity for a newly added broker and must be carried over verbatim (finding: a mixed add+remove + // edit must not drop the new broker's user-provided capacity). + var newBrokerCapacities []interface{} + for _, userCapacity := range userCapacities { + brokerId, err := brokerIDFromCapacityEntry(userCapacity) + if err != nil { + return "", err + } + if brokerId == "" { + continue + } + if _, ok := deployedBrokerIds[brokerId]; !ok { + newBrokerCapacities = append(newBrokerCapacities, userCapacity) + } + } + + // Generate entries for the remaining newly joined brokers (missing from the deployed config and not + // covered by a user-provided capacity), matching the non-reuse path's generation. + coveredBrokerIds := append([]string(nil), userConfigBrokerIds...) + for brokerId := range deployedBrokerIds { + coveredBrokerIds = append(coveredBrokerIds, brokerId) + } + generatedBrokerCapacities, err := appendGeneratedBrokerCapacities(kafkaCluster, log, coveredBrokerIds) + if err != nil { + return "", err + } + newBrokerCapacities = append(newBrokerCapacities, generatedBrokerCapacities...) + + // No broker has joined since the config was deployed: reuse it verbatim so Cruise Control is not rolled. + if len(newBrokerCapacities) == 0 { + return deployedCapacityConfig, nil + } + + deployed.Capacities = append(deployed.Capacities, newBrokerCapacities...) + result, err := json.MarshalIndent(deployed, "", " ") + if err != nil { + return "", errors.WrapIf(err, "could not marshal merged cruise control capacity config") + } + log.V(1).Info("merged newly added brokers into the deployed capacity config", "capacity config", string(result)) + return string(result), nil +} + +// CapacityConfigContainsBrokers reports whether the given capacity.json defines a capacity entry for every +// broker id in brokerIDs. A "-1" universal-default entry counts as covering every broker. It lets a caller +// confirm a newly added broker's capacity has actually been written before acting on it. +func CapacityConfigContainsBrokers(capacityConfigJSON string, brokerIDs []string) (bool, error) { + var capacityConfig JBODInvariantCapacityConfig + if err := json.Unmarshal([]byte(capacityConfigJSON), &capacityConfig); err != nil { + return false, errors.Wrap(err, "could not unmarshal capacity config") + } + present := make(map[string]struct{}, len(capacityConfig.Capacities)) + for _, entry := range capacityConfig.Capacities { + brokerID, err := brokerIDFromCapacityEntry(entry) + if err != nil { + return false, err + } + if brokerID == "-1" { + // Universal default: every broker is covered. + return true, nil + } + if brokerID != "" { + present[brokerID] = struct{}{} + } + } + for _, id := range brokerIDs { + if _, ok := present[id]; !ok { + return false, nil + } + } + return true, nil +} + +// brokerIDFromCapacityEntry extracts the "brokerId" field from a capacity.json entry. It returns an empty +// string (no error) when the entry is not a JSON object or has no broker id, matching how the rest of the +// capacity handling tolerates heterogeneous JBOD/non-JBOD entries. +func brokerIDFromCapacityEntry(entry interface{}) (string, error) { + entryMap, ok := entry.(map[string]interface{}) + if !ok { + return "", nil + } + brokerId, ok, err := unstructured.NestedString(entryMap, v1beta1.BrokerIdLabelKey) + if err != nil { + return "", errors.WrapIfWithDetails(err, + "could not retrieve broker Id from broker capacity configuration", + "capacity configuration", entryMap) + } + if !ok { + return "", nil + } + return brokerId, nil +} + func appendGeneratedBrokerCapacities(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger, userConfigBrokerIds []string) ([]interface{}, error) { var brokerCapacities []interface{} diff --git a/pkg/resources/cruisecontrol/configmap_test.go b/pkg/resources/cruisecontrol/configmap_test.go index 89ec08f27..49e1e4384 100644 --- a/pkg/resources/cruisecontrol/configmap_test.go +++ b/pkg/resources/cruisecontrol/configmap_test.go @@ -17,7 +17,9 @@ package cruisecontrol import ( "encoding/json" + "fmt" "reflect" + "sort" "testing" "github.com/go-logr/logr" @@ -1018,3 +1020,359 @@ func TestGenerateCapacityConfigWithUserProvidedInput(t *testing.T) { }) } } + +// TestGenerateCapacityConfigReuseAndMerge covers the config-reuse path taken during a scaling operation +// (config != nil): a departing broker's deployed entry must be preserved verbatim (so capacity.json does +// not change and roll Cruise Control mid-removal, see #301), while a newly added broker must get a freshly +// generated entry so add_broker has capacity data even in a mixed add+remove edit. +// +//nolint:funlen +func TestGenerateCapacityConfigReuseAndMerge(t *testing.T) { + tenGiQuantity, _ := resource.ParseQuantity("10Gi") + + // The already-deployed capacity.json. Broker 100 (the one being removed) carries deliberately unusual + // values so we can prove it is preserved verbatim rather than rewritten to the fallback default. + deployedCapacityConfig := `{ + "brokerCapacities": [ + {"brokerId": "0", "capacity": {"DISK": {"/kafka-logs-broker/kafka": "10240"}, "CPU": "100", "NW_IN": "125000", "NW_OUT": "125000"}, "doc": "d"}, + {"brokerId": "1", "capacity": {"DISK": {"/kafka-logs-broker/kafka": "10240"}, "CPU": "100", "NW_IN": "125000", "NW_OUT": "125000"}, "doc": "d"}, + {"brokerId": "2", "capacity": {"DISK": {"/kafka-logs-broker/kafka": "10240"}, "CPU": "100", "NW_IN": "125000", "NW_OUT": "125000"}, "doc": "d"}, + {"brokerId": "100", "capacity": {"DISK": {"/kafka-logs-broker/kafka": "99999"}, "CPU": "777", "NW_IN": "111", "NW_OUT": "222"}, "doc": "departing-verbatim"} + ] +}` + deployedConfigMap := &v1.ConfigMap{Data: map[string]string{CapacityConfigMapKey: deployedCapacityConfig}} + + brokerConfigGroups := map[string]v1beta1.BrokerConfig{ + "broker": { + StorageConfigs: []v1beta1.StorageConfig{ + { + MountPath: "/kafka-logs-broker", + PvcSpec: &v1.PersistentVolumeClaimSpec{ + Resources: v1.VolumeResourceRequirements{ + Requests: v1.ResourceList{v1.ResourceStorage: tenGiQuantity}, + }, + }, + }, + }, + }, + } + + brokers := func(ids ...int32) []v1beta1.Broker { + out := make([]v1beta1.Broker, 0, len(ids)) + for _, id := range ids { + out = append(out, v1beta1.Broker{Id: id, BrokerConfigGroup: "broker"}) + } + return out + } + statusState := func(ids ...string) map[string]v1beta1.BrokerState { + out := map[string]v1beta1.BrokerState{} + for _, id := range ids { + out[id] = v1beta1.BrokerState{} + } + return out + } + + t.Run("pure removal returns the deployed config verbatim (no roll)", func(t *testing.T) { + // Broker 100 dropped from the spec but still in the status: nothing new to add, so the deployed + // capacity.json must be returned byte-for-byte unchanged. + kc := &v1beta1.KafkaCluster{ + Spec: v1beta1.KafkaClusterSpec{ + BrokerConfigGroups: brokerConfigGroups, + Brokers: brokers(0, 1, 2), + }, + Status: v1beta1.KafkaClusterStatus{BrokersState: statusState("0", "1", "2", "100")}, + } + + actual, err := GenerateCapacityConfig(kc, logr.Discard(), deployedConfigMap) + if err != nil { + t.Fatal(err, "unexpected error") + } + if actual != deployedCapacityConfig { + t.Errorf("expected the deployed capacity.json to be reused verbatim.\nExpected:\n%s\nGot:\n%s", deployedCapacityConfig, actual) + } + }) + + t.Run("mixed add+remove preserves the departing broker and adds the new one", func(t *testing.T) { + // Broker 100 removed and broker 101 added in one edit; both are still/already in the status. + kc := &v1beta1.KafkaCluster{ + Spec: v1beta1.KafkaClusterSpec{ + BrokerConfigGroups: brokerConfigGroups, + Brokers: brokers(0, 1, 2, 101), + }, + Status: v1beta1.KafkaClusterStatus{BrokersState: statusState("0", "1", "2", "100", "101")}, + } + + actual, err := GenerateCapacityConfig(kc, logr.Discard(), deployedConfigMap) + if err != nil { + t.Fatal(err, "unexpected error") + } + + var merged CapacityConfig + if err := json.Unmarshal([]byte(actual), &merged); err != nil { + t.Fatal(err, "could not unmarshal merged json") + } + + byID := map[string]BrokerCapacity{} + for _, bc := range merged.BrokerCapacities { + byID[bc.BrokerID] = bc + } + + gotIDs := make([]string, 0, len(byID)) + for id := range byID { + gotIDs = append(gotIDs, id) + } + sort.Strings(gotIDs) + if !reflect.DeepEqual(gotIDs, []string{"0", "1", "100", "101", "2"}) { + t.Errorf("unexpected broker ids in merged config: %v", gotIDs) + } + + // Departing broker 100 must be preserved verbatim (its unusual deployed values, not the fallback). + departing := byID["100"] + if departing.Capacity.CPU != "777" || departing.Capacity.DISK["/kafka-logs-broker/kafka"] != "99999" { + t.Errorf("departing broker 100 capacity was not preserved verbatim: %+v", departing) + } + + // Newly added broker 101 must have a generated entry so add_broker has capacity data. + added, ok := byID["101"] + if !ok { + t.Fatal("newly added broker 101 is missing from the merged capacity config") + } + if len(added.Capacity.DISK) == 0 { + t.Errorf("newly added broker 101 has no generated disk capacity: %+v", added) + } + }) + + for _, state := range []v1beta1.CruiseControlState{v1beta1.GracefulDownscaleRequired, v1beta1.GracefulDownscaleScheduled} { + state := state + t.Run(fmt.Sprintf("mixed add+remove with downscale %s still adds the new broker (no deadlock)", state), func(t *testing.T) { + // Broker 100's downscale has not started executing in Cruise Control yet (Required: no + // CruiseControlOperation created; Scheduled: one exists but has not been submitted to CC) - there + // is no in-flight CC-side task for a capacity.json roll to disrupt. Reusing the deployed config + // verbatim here (as isBrokerDeletionInProgress used to do for every IsDownscale() state) would + // never write broker 101's capacity, so add_broker's roll gate + // (CapacityConfigContainsBrokers) would defer it forever; and because the task controller + // prioritizes add_broker over remove_broker, broker 100's downscale would never advance past this + // state either - a permanent deadlock. The merge must proceed instead. + kc := &v1beta1.KafkaCluster{ + Spec: v1beta1.KafkaClusterSpec{ + BrokerConfigGroups: brokerConfigGroups, + Brokers: brokers(0, 1, 2, 101), + }, + Status: v1beta1.KafkaClusterStatus{BrokersState: map[string]v1beta1.BrokerState{ + "0": {}, "1": {}, "2": {}, + "100": {GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: state}}, + "101": {}, + }}, + } + + actual, err := GenerateCapacityConfig(kc, logr.Discard(), deployedConfigMap) + if err != nil { + t.Fatal(err, "unexpected error") + } + + var merged CapacityConfig + if err := json.Unmarshal([]byte(actual), &merged); err != nil { + t.Fatal(err, "could not unmarshal merged json") + } + byID := map[string]BrokerCapacity{} + for _, bc := range merged.BrokerCapacities { + byID[bc.BrokerID] = bc + } + + added, ok := byID["101"] + if !ok { + t.Fatalf("newly added broker 101 is missing from the capacity config while broker 100 is %s - this deadlocks add_broker", state) + } + if len(added.Capacity.DISK) == 0 { + t.Errorf("newly added broker 101 has no generated disk capacity: %+v", added) + } + + // Departing broker 100 must still be preserved verbatim. + if got := byID["100"].Capacity.CPU; got != "777" { + t.Errorf("departing broker 100 was not preserved verbatim: CPU=%q", got) + } + }) + } + + t.Run("mixed add+remove keeps the new broker's user-provided capacity (not generated, not dropped)", func(t *testing.T) { + // The CR provides explicit per-broker capacity (no "-1" universal default) for the newly added + // broker 101. During the removal-pending window the merge must carry that user-provided entry over + // verbatim rather than dropping it (and rather than generating a different one). + kc := &v1beta1.KafkaCluster{ + Spec: v1beta1.KafkaClusterSpec{ + BrokerConfigGroups: brokerConfigGroups, + Brokers: brokers(0, 1, 2, 101), + CruiseControlConfig: v1beta1.CruiseControlConfig{ + CapacityConfig: `{ + "brokerCapacities": [ + {"brokerId": "101", "capacity": {"DISK": {"/kafka-logs-broker/kafka": "424242"}, "CPU": "314", "NW_IN": "271", "NW_OUT": "161"}, "doc": "user-provided-101"} + ] +}`, + }, + }, + Status: v1beta1.KafkaClusterStatus{BrokersState: statusState("0", "1", "2", "100", "101")}, + } + + actual, err := GenerateCapacityConfig(kc, logr.Discard(), deployedConfigMap) + if err != nil { + t.Fatal(err, "unexpected error") + } + + var merged CapacityConfig + if err := json.Unmarshal([]byte(actual), &merged); err != nil { + t.Fatal(err, "could not unmarshal merged json") + } + byID := map[string]BrokerCapacity{} + for _, bc := range merged.BrokerCapacities { + byID[bc.BrokerID] = bc + } + + // Broker 100 (departing) preserved verbatim, broker 101 present with the user-provided values. + if got := byID["100"].Capacity.CPU; got != "777" { + t.Errorf("departing broker 100 was not preserved verbatim: CPU=%q", got) + } + added, ok := byID["101"] + if !ok { + t.Fatal("newly added user-configured broker 101 is missing from the merged capacity config") + } + if added.Capacity.CPU != "314" || added.Capacity.DISK["/kafka-logs-broker/kafka"] != "424242" { + t.Errorf("broker 101 did not keep its user-provided capacity: %+v", added) + } + }) + + t.Run("unparseable deployed capacity.json is reused verbatim (fail-safe, no roll)", func(t *testing.T) { + // A corrupt deployed capacity.json must not crash or regenerate from scratch (which would roll CC + // mid-scaling); mergeCapacityConfig logs and returns it unchanged. + malformed := `{ this is not valid capacity json` + cm := &v1.ConfigMap{Data: map[string]string{CapacityConfigMapKey: malformed}} + kc := &v1beta1.KafkaCluster{ + Spec: v1beta1.KafkaClusterSpec{ + BrokerConfigGroups: brokerConfigGroups, + Brokers: brokers(0, 1, 2), + }, + Status: v1beta1.KafkaClusterStatus{BrokersState: statusState("0", "1", "2", "100")}, + } + + actual, err := GenerateCapacityConfig(kc, logr.Discard(), cm) + if err != nil { + t.Fatal(err, "unexpected error") + } + if actual != malformed { + t.Errorf("expected the unparseable capacity.json to be reused verbatim.\nExpected:\n%s\nGot:\n%s", malformed, actual) + } + }) + + t.Run("active downscale reuses verbatim and does NOT append a concurrently added broker (would roll CC)", func(t *testing.T) { + // Broker 100 is actively downscaling (a remove_broker task is in flight); broker 103 was added + // concurrently and is already in the status. Appending 103 would change capacity.json and roll CC, + // killing the in-flight removal (the #301 failure class), so the deployed config must be reused + // verbatim - 103's capacity is written by the next full regeneration once the downscale completes. + kc := &v1beta1.KafkaCluster{ + Spec: v1beta1.KafkaClusterSpec{ + BrokerConfigGroups: brokerConfigGroups, + Brokers: brokers(0, 1, 2, 103), + }, + Status: v1beta1.KafkaClusterStatus{BrokersState: map[string]v1beta1.BrokerState{ + "0": {}, "1": {}, "2": {}, + "100": {GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: v1beta1.GracefulDownscaleRunning}}, + "103": {}, + }}, + } + + actual, err := GenerateCapacityConfig(kc, logr.Discard(), deployedConfigMap) + if err != nil { + t.Fatal(err, "unexpected error") + } + if actual != deployedCapacityConfig { + t.Errorf("expected verbatim reuse during an active downscale (no 103 appended).\nExpected:\n%s\nGot:\n%s", deployedCapacityConfig, actual) + } + }) + + t.Run("deployed config with a -1 universal default is reused verbatim (no redundant append/roll)", func(t *testing.T) { + // A "-1" default covers every broker, so a newly added broker needs no per-broker entry; appending one + // would only change capacity.json and roll CC. (Latent - not reachable via the normal flow.) + deployedWithMinus1 := `{ + "brokerCapacities": [ + {"brokerId": "-1", "capacity": {"DISK": {"/kafka-logs-broker/kafka": "10240"}, "CPU": "100", "NW_IN": "125000", "NW_OUT": "125000"}, "doc": "default"} + ] +}` + cm := &v1.ConfigMap{Data: map[string]string{CapacityConfigMapKey: deployedWithMinus1}} + kc := &v1beta1.KafkaCluster{ + Spec: v1beta1.KafkaClusterSpec{ + BrokerConfigGroups: brokerConfigGroups, + Brokers: brokers(0, 1, 2, 103), + }, + // No downscale in progress, so the merge path runs and must hit the -1 guard. + Status: v1beta1.KafkaClusterStatus{BrokersState: statusState("0", "1", "2", "103")}, + } + + actual, err := GenerateCapacityConfig(kc, logr.Discard(), cm) + if err != nil { + t.Fatal(err, "unexpected error") + } + if actual != deployedWithMinus1 { + t.Errorf("expected verbatim reuse when a -1 universal default is present.\nExpected:\n%s\nGot:\n%s", deployedWithMinus1, actual) + } + }) +} + +func TestCapacityConfigContainsBrokers(t *testing.T) { + perBroker := `{"brokerCapacities":[ + {"brokerId":"0","capacity":{"DISK":{"/k":"1"},"CPU":"1","NW_IN":"1","NW_OUT":"1"},"doc":"d"}, + {"brokerId":"103","capacity":{"DISK":{"/k":"1"},"CPU":"1","NW_IN":"1","NW_OUT":"1"},"doc":"d"} + ]}` + universal := `{"brokerCapacities":[ + {"brokerId":"-1","capacity":{"DISK":{"/k":"1"},"CPU":"1","NW_IN":"1","NW_OUT":"1"},"doc":"d"} + ]}` + + tests := []struct { + name string + capacity string + brokerIDs []string + expected bool + expectError bool + }{ + {name: "all present", capacity: perBroker, brokerIDs: []string{"0", "103"}, expected: true}, + {name: "one missing", capacity: perBroker, brokerIDs: []string{"103", "104"}, expected: false}, + {name: "no brokers requested", capacity: perBroker, brokerIDs: nil, expected: true}, + {name: "universal default covers any broker", capacity: universal, brokerIDs: []string{"999"}, expected: true}, + {name: "invalid json", capacity: "{not json", brokerIDs: []string{"0"}, expectError: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got, err := CapacityConfigContainsBrokers(test.capacity, test.brokerIDs) + if test.expectError { + if err == nil { + t.Fatal("expected an error, got nil") + } + return + } + if err != nil { + t.Fatal(err, "unexpected error") + } + if got != test.expected { + t.Errorf("expected %v, got %v", test.expected, got) + } + }) + } +} + +func TestCapacityConfigHash(t *testing.T) { + // Deterministic and sensitive to content; must equal what GeneratePodAnnotations stamps into the pod + // template for the same capacity.json. + a := CapacityConfigHash(`{"brokerCapacities":[]}`) + if CapacityConfigHash(`{"brokerCapacities":[]}`) != a { + t.Error("hash is not deterministic") + } + if CapacityConfigHash(`{"brokerCapacities":[{"brokerId":"0"}]}`) == a { + t.Error("hash did not change for different content") + } + + capacityJSON := `{"brokerCapacities":[{"brokerId":"0"}]}` + viaAnnotations := GeneratePodAnnotations(nil, map[string]string{CapacityConfigMapKey: capacityJSON}) + if viaAnnotations[CapacityConfigHashAnnotationKey] != CapacityConfigHash(capacityJSON) { + t.Errorf("CapacityConfigHash (%s) does not match GeneratePodAnnotations (%s)", + CapacityConfigHash(capacityJSON), viaAnnotations[CapacityConfigHashAnnotationKey]) + } +} diff --git a/pkg/resources/cruisecontrol/cruisecontrol.go b/pkg/resources/cruisecontrol/cruisecontrol.go index a361e98c8..6455c3f25 100644 --- a/pkg/resources/cruisecontrol/cruisecontrol.go +++ b/pkg/resources/cruisecontrol/cruisecontrol.go @@ -18,6 +18,7 @@ package cruisecontrol import ( "context" "fmt" + "strconv" "emperror.dev/errors" "github.com/go-logr/logr" @@ -66,6 +67,19 @@ func ccLabelSelector(kafkaCluster string) map[string]string { } } +// DeploymentName returns the name of the Cruise Control Deployment reconciled for kafkaCluster. Callers +// outside this package (e.g. controllers) must use this instead of re-deriving the name so the convention +// stays in one place. +func DeploymentName(kafkaCluster *v1beta1.KafkaCluster) string { + return fmt.Sprintf(deploymentNameTemplate, kafkaCluster.Name) +} + +// ConfigMapName returns the name of the Cruise Control ConfigMap reconciled for kafkaCluster (the one that +// holds capacity.json). See DeploymentName for why this is centralized. +func ConfigMapName(kafkaCluster *v1beta1.KafkaCluster) string { + return fmt.Sprintf(configAndVolumeNameTemplate, kafkaCluster.Name) +} + // New creates a new reconciler for CC func New(client client.Client, cluster *v1beta1.KafkaCluster, kafkaClientProvider kafkaclient.Provider) *Reconciler { return &Reconciler{ @@ -112,7 +126,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error { } var config *corev1.ConfigMap - if isBrokerDeletionInProgress(r.KafkaCluster.Status.BrokersState) { + if isBrokerDeletionInProgress(r.KafkaCluster.Status.BrokersState) || isBrokerRemovalPending(r.KafkaCluster) { key := types.NamespacedName{ Name: fmt.Sprintf(configAndVolumeNameTemplate, r.KafkaCluster.Name), Namespace: r.KafkaCluster.Namespace, @@ -193,9 +207,55 @@ func (r *Reconciler) getClientSecret() (*corev1.Secret, error) { return clientSecret, nil } +// isBrokerDeletionInProgress reports whether a broker in the status is actively being downscaled by Cruise +// Control right now (GracefulDownscaleRunning). It deliberately checks IsDownscaleRunning(), not the broader +// IsDownscale(): Required/Scheduled downscale states have no in-flight CC-side task to protect from a +// capacity.json roll, and treating them as "in progress" here can deadlock a concurrent add - the config +// generator would refuse to add the new broker's capacity forever, while the task controller's add-before- +// remove priority (cruisecontroltask_controller.go) never lets the downscale advance past Required/Scheduled +// to unblock it (see #301). +// +// KNOWN LIMITATION (accepted; documented here as the record, no separate tracking issue filed): +// BrokersState.CruiseControlState is a lagging mirror - +// CruiseControlTaskReconciler copies the CruiseControlOperation's task state into it asynchronously, and the +// KafkaCluster is read here via the cached client. So there is a brief window where remove_broker is already +// running on Cruise Control but this still reads GracefulDownscaleScheduled, i.e. returns false. If a broker +// that is NOT yet in the deployed capacity.json is added concurrently during that window, GenerateCapacityConfig +// takes the merge path, rewrites capacity.json, and rolls CC - killing the in-flight remove_broker (the #301 +// class, for a narrower trigger). This is intentionally not guarded here because it is narrow (needs a second, +// independent broker add landing inside the seconds-wide Scheduled->Running mirror-lag; pure removals and +// single-apply mixed edits are safe because the added broker is already in capacity.json before the removal +// runs) and self-heals (the killed remove_broker is retried under ErrorPolicyRetry). The robust fix is to key +// off the live CruiseControlOperation task state (via GracefulActionState.CruiseControlOperationReference, read +// through the non-cached DirectClient) instead of this mirrored enum - a deliberate design change left as a +// documented follow-up. func isBrokerDeletionInProgress(brokerState map[string]v1beta1.BrokerState) bool { for _, state := range brokerState { - if state.GracefulActionState.CruiseControlState.IsDownscale() { + if state.GracefulActionState.CruiseControlState.IsDownscaleRunning() { + return true + } + } + return false +} + +// isBrokerRemovalPending reports whether a broker that is still present in the status has already been +// dropped from the spec - i.e. a removal that has not yet been marked as a Cruise Control downscale. +// +// During this window the operator must keep reusing the already-deployed capacity.json instead of +// regenerating a fallback entry for the departing broker. Regenerating it changes capacity.json, which +// (because capacity.json is hashed into the Cruise Control pod template) rolls the Cruise Control +// Deployment and resets CC's metric-sampling window - keeping CC un-ready exactly when +// reconcileKafkaPodDelete needs CC ready (via BrokersWithState) to mark the downscale. That chicken-and-egg +// otherwise prevents the remove_broker operation from ever being created and stalls the removal (see #301). +// Once the pod is gone and the broker is dropped from the status too, capacity.json shrinks with a single +// harmless roll and no operation in flight. +func isBrokerRemovalPending(kafkaCluster *v1beta1.KafkaCluster) bool { + specBrokerIDs := make(map[string]struct{}, len(kafkaCluster.Spec.Brokers)) + for i := range kafkaCluster.Spec.Brokers { + specBrokerIDs[strconv.Itoa(int(kafkaCluster.Spec.Brokers[i].Id))] = struct{}{} + } + for brokerID := range kafkaCluster.Status.BrokersState { + if _, ok := specBrokerIDs[brokerID]; !ok { return true } } diff --git a/pkg/resources/cruisecontrol/cruisecontrol_test.go b/pkg/resources/cruisecontrol/cruisecontrol_test.go new file mode 100644 index 000000000..1015e6d9d --- /dev/null +++ b/pkg/resources/cruisecontrol/cruisecontrol_test.go @@ -0,0 +1,105 @@ +// Copyright 2026 Adobe. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package cruisecontrol + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/banzaicloud/koperator/api/v1beta1" +) + +func TestIsBrokerRemovalPending(t *testing.T) { + cluster := func(specIDs []int32, statusIDs []string) *v1beta1.KafkaCluster { + kc := &v1beta1.KafkaCluster{} + for _, id := range specIDs { + kc.Spec.Brokers = append(kc.Spec.Brokers, v1beta1.Broker{Id: id}) + } + kc.Status.BrokersState = map[string]v1beta1.BrokerState{} + for _, id := range statusIDs { + kc.Status.BrokersState[id] = v1beta1.BrokerState{} + } + return kc + } + + tests := []struct { + testName string + cluster *v1beta1.KafkaCluster + expected bool + }{ + { + testName: "steady state: every status broker is in the spec", + cluster: cluster([]int32{0, 1, 2}, []string{"0", "1", "2"}), + expected: false, + }, + { + testName: "removal pending: a status broker was dropped from the spec", + cluster: cluster([]int32{0, 1, 2}, []string{"0", "1", "2", "103"}), + expected: true, + }, + { + testName: "upscale in progress: a new spec broker not yet in status is NOT a removal", + cluster: cluster([]int32{0, 1, 2, 103}, []string{"0", "1", "2"}), + expected: false, + }, + { + testName: "empty status", + cluster: cluster([]int32{0, 1, 2}, nil), + expected: false, + }, + } + + for _, test := range tests { + t.Run(test.testName, func(t *testing.T) { + require.Equal(t, test.expected, isBrokerRemovalPending(test.cluster)) + }) + } +} + +// TestIsBrokerDeletionInProgress guards against reintroducing the #301 deadlock: isBrokerDeletionInProgress +// must only report true for a downscale that Cruise Control is actively executing right now +// (GracefulDownscaleRunning), not for Required/Scheduled (no in-flight CC-side task to protect) or Succeeded +// (already done). Treating Required/Scheduled as "in progress" makes the capacity generator refuse to add a +// concurrently-added broker's capacity forever, which in turn stalls its add_broker op and - because the task +// controller processes add_broker before remove_broker - prevents the downscale from ever advancing past +// Required/Scheduled either. +func TestIsBrokerDeletionInProgress(t *testing.T) { + state := func(ccState v1beta1.CruiseControlState) v1beta1.BrokerState { + return v1beta1.BrokerState{GracefulActionState: v1beta1.GracefulActionState{CruiseControlState: ccState}} + } + + tests := []struct { + testName string + ccState v1beta1.CruiseControlState + expected bool + }{ + {testName: "no downscale state", ccState: "", expected: false}, + {testName: "downscale required: no CruiseControlOperation exists yet", ccState: v1beta1.GracefulDownscaleRequired, expected: false}, + {testName: "downscale scheduled: CruiseControlOperation created but not yet submitted to CC", ccState: v1beta1.GracefulDownscaleScheduled, expected: false}, + {testName: "downscale running: actively executing in CC", ccState: v1beta1.GracefulDownscaleRunning, expected: true}, + {testName: "downscale succeeded", ccState: v1beta1.GracefulDownscaleSucceeded, expected: false}, + {testName: "downscale completed with error", ccState: v1beta1.GracefulDownscaleCompletedWithError, expected: false}, + {testName: "downscale paused", ccState: v1beta1.GracefulDownscalePaused, expected: false}, + {testName: "upscale running is not a downscale", ccState: v1beta1.GracefulUpscaleRunning, expected: false}, + } + + for _, test := range tests { + t.Run(test.testName, func(t *testing.T) { + brokerState := map[string]v1beta1.BrokerState{"100": state(test.ccState)} + require.Equal(t, test.expected, isBrokerDeletionInProgress(brokerState)) + }) + } +} diff --git a/pkg/resources/cruisecontrol/deployment.go b/pkg/resources/cruisecontrol/deployment.go index 3021fd122..54d34e702 100644 --- a/pkg/resources/cruisecontrol/deployment.go +++ b/pkg/resources/cruisecontrol/deployment.go @@ -184,24 +184,54 @@ fi`}, } } -func GeneratePodAnnotations(cruiseControlAnnotations, cruiseControlConfig map[string]string) map[string]string { - hashedCruiseControlConfigJson := sha256.Sum256([]byte(cruiseControlConfig["cruisecontrol.properties"])) - hashedCruiseControlClusterConfigJson := sha256.Sum256([]byte(cruiseControlConfig["clusterConfigs.json"])) - hashedCruiseControlLogConfigJson := sha256.Sum256([]byte(cruiseControlConfig["log4j.properties"])) +// CapacityConfigHashAnnotationKey is the CC pod-template annotation that carries the hash of the deployed +// capacity.json. Its presence marks a Deployment whose capacity is koperator-managed (see GeneratePodAnnotations), +// and comparing it against ConfigHash of the current ConfigMap tells whether the running CC pod has +// already been rolled with that capacity. The operation controller's roll gate +// (CruiseControlOperationReconciler.requeueIfCCDeploymentNotRolledOut) reads this to decide whether a broker +// op is safe to submit - changing how it is computed/stamped must keep that gate in sync. +const CapacityConfigHashAnnotationKey = "cruiseControlCapacity.json" + +// ConfigHashAnnotationKey, ClusterConfigHashAnnotationKey and LogConfigHashAnnotationKey are the CC +// pod-template annotations that carry the hash of, respectively, cruisecontrol.properties, +// clusterConfigs.json and log4j.properties (see GeneratePodAnnotations). Like +// CapacityConfigHashAnnotationKey, the operation controller's roll gate +// (CruiseControlOperationReconciler.requeueIfCCDeploymentNotRolledOut) compares these against the current +// ConfigMap to decide whether a broker op is safe to submit - a change to any one of them rolls the CC +// Deployment same as a capacity.json change, so the gate must cover all four, not just capacity. +const ( + ConfigHashAnnotationKey = "cruiseControlConfig.json" + ClusterConfigHashAnnotationKey = "cruiseControlClusterConfig.json" + LogConfigHashAnnotationKey = "cruiseControlLogConfig.json" +) + +// ConfigHash returns the hex-encoded sha256 of a ConfigMap entry's content, matching the value +// GeneratePodAnnotations stamps into the CC pod template for each of the Config/ClusterConfig/LogConfig/ +// Capacity hash annotation keys. +func ConfigHash(content string) string { + sum := sha256.Sum256([]byte(content)) + return hex.EncodeToString(sum[:]) +} + +// CapacityConfigHash returns the hex-encoded sha256 of a capacity.json. Kept as a distinctly named alias of +// ConfigHash for callers/tests that reference capacity.json specifically. +func CapacityConfigHash(capacityConfigJSON string) string { + return ConfigHash(capacityConfigJSON) +} +func GeneratePodAnnotations(cruiseControlAnnotations, cruiseControlConfig map[string]string) map[string]string { annotations := []map[string]string{ cruiseControlAnnotations, { - "cruiseControlConfig.json": hex.EncodeToString(hashedCruiseControlConfigJson[:]), - "cruiseControlClusterConfig.json": hex.EncodeToString(hashedCruiseControlClusterConfigJson[:]), - "cruiseControlLogConfig.json": hex.EncodeToString(hashedCruiseControlLogConfigJson[:]), + ConfigHashAnnotationKey: ConfigHash(cruiseControlConfig[PropertiesConfigMapKey]), + ClusterConfigHashAnnotationKey: ConfigHash(cruiseControlConfig[ClusterConfigsConfigMapKey]), + LogConfigHashAnnotationKey: ConfigHash(cruiseControlConfig[Log4jConfigMapKey]), }, } if value, ok := cruiseControlAnnotations[capacityConfigAnnotation]; !ok || value == string(staticCapacityConfig) { - hashedCruiseControlCapacityJson := sha256.Sum256([]byte(cruiseControlConfig["capacity.json"])) annotations = append(annotations, - map[string]string{"cruiseControlCapacity.json": hex.EncodeToString(hashedCruiseControlCapacityJson[:])}) + map[string]string{CapacityConfigHashAnnotationKey: CapacityConfigHash(cruiseControlConfig[CapacityConfigMapKey])}) } return util.MergeAnnotations(annotations...) diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go index 132859684..d05d8fabf 100644 --- a/pkg/resources/kafka/kafka.go +++ b/pkg/resources/kafka/kafka.go @@ -489,7 +489,24 @@ func (r *Reconciler) Reconcile(log logr.Logger) error { // reconcile flow. The services must be deleted at the end of the reconcile flow after the new services // were created and broker configurations reflecting the new services otherwise the Kafka brokers // won't be reachable by koperator. - if r.KafkaCluster.Spec.HeadlessServiceEnabled { + // + // A broker pod can still be running with a live pod outside spec.Brokers while its removal is pending + // (deletion is gated on Cruise Control finishing the data migration off it - see isBrokerRemovalPending). + // That pod is not touched by the per-broker loop above, so it never gets a replacement address on the + // other addressing scheme; its only reachable address is whichever Service this reconcile is about to + // delete. Deleting it out from under a broker that is still alive and still needed would orphan that + // broker from the network entirely, stalling the very Cruise Control operation the removal is waiting on + // (see #316). Rather than reasoning about whether this particular deletion happens to be safe this time, + // skip both service-topology transitions outright while any such pod exists, log it clearly, and let the + // next reconcile (triggered once that pod is actually gone) retry - koperator would rather stay wedged + // here than risk cutting off a broker it still needs. + if pendingBrokerID, stillPending := firstRunningBrokerOutsideSpec(runningBrokers, r.KafkaCluster.Spec.Brokers); stillPending { + log.Info("deferring headless/non-headless service reconciliation: a broker outside spec.Brokers still has a running pod", + "component", componentName, + "clusterName", r.KafkaCluster.Name, + "clusterNamespace", r.KafkaCluster.Namespace, + "pendingBrokerId", pendingBrokerID) + } else if r.KafkaCluster.Spec.HeadlessServiceEnabled { log.V(1).Info("deleting non-headless services for all of the brokers") if err := r.deleteNonHeadlessServices(ctx); err != nil { @@ -1682,6 +1699,25 @@ func getServiceFromExternalListener(client client.Client, cluster *banzaiv1beta1 return foundLBService, nil } +// firstRunningBrokerOutsideSpec reports whether any broker with a running pod (runningBrokers, keyed by +// BrokerIdLabelKey value) is absent from desiredBrokers, returning its ID for logging. Such a broker is +// outside the per-broker reconcile loop entirely (see reorderBrokers) - e.g. one dropped from spec.Brokers +// while its removal is still pending Cruise Control's data migration - so it never receives a replacement +// address on the other Service-addressing scheme. Iteration order over the map is nondeterministic, but +// finding any single one is enough for callers that only need to know whether it is safe to proceed. +func firstRunningBrokerOutsideSpec(runningBrokers map[string]struct{}, desiredBrokers []banzaiv1beta1.Broker) (string, bool) { + inSpec := make(map[string]struct{}, len(desiredBrokers)) + for _, b := range desiredBrokers { + inSpec[strconv.Itoa(int(b.Id))] = struct{}{} + } + for id := range runningBrokers { + if _, ok := inSpec[id]; !ok { + return id, true + } + } + return "", false +} + // reorderBrokers returns the KafkaCluster brokers list reordered for reconciliation such that: // - the controller broker is reconciled last // - prioritize missing broker pods where downscale operation has not been finished yet to give bigger chance to be scheduled and downscale operation to be continued diff --git a/pkg/resources/kafka/kafka_test.go b/pkg/resources/kafka/kafka_test.go index 2c2b34dad..61f9ca609 100644 --- a/pkg/resources/kafka/kafka_test.go +++ b/pkg/resources/kafka/kafka_test.go @@ -591,6 +591,52 @@ func TestReorderBrokers(t *testing.T) { } } +func TestFirstRunningBrokerOutsideSpec(t *testing.T) { + testCases := []struct { + testName string + runningBrokers map[string]struct{} + desiredBrokers []v1beta1.Broker + wantFound bool + wantID string + }{ + { + testName: "no running brokers at all", + runningBrokers: map[string]struct{}{}, + desiredBrokers: []v1beta1.Broker{{Id: 0}, {Id: 1}}, + wantFound: false, + }, + { + testName: "every running broker is still in spec", + runningBrokers: map[string]struct{}{"0": {}, "1": {}, "2": {}}, + desiredBrokers: []v1beta1.Broker{{Id: 0}, {Id: 1}, {Id: 2}}, + wantFound: false, + }, + { + testName: "a broker dropped from spec still has a running pod", + // Mirrors #316: broker 3 removed from spec.Brokers while its pod (pending Cruise + // Control's data migration) is still Running. + runningBrokers: map[string]struct{}{"0": {}, "1": {}, "2": {}, "3": {}}, + desiredBrokers: []v1beta1.Broker{{Id: 0}, {Id: 1}, {Id: 2}}, + wantFound: true, + wantID: "3", + }, + { + testName: "a desired broker has no running pod yet (scale-up in progress)", + runningBrokers: map[string]struct{}{"0": {}}, + desiredBrokers: []v1beta1.Broker{{Id: 0}, {Id: 1}}, + wantFound: false, + }, + } + + for _, tt := range testCases { + t.Run(tt.testName, func(t *testing.T) { + id, found := firstRunningBrokerOutsideSpec(tt.runningBrokers, tt.desiredBrokers) + assert.Equal(t, tt.wantFound, found) + assert.Equal(t, tt.wantID, id) + }) + } +} + func TestGetServerPasswordKeysAndUsers(t *testing.T) { //nolint funlen t.Parallel() testCases := []struct { diff --git a/tests/e2e/const.go b/tests/e2e/const.go index 158cc8e64..f31d3e152 100644 --- a/tests/e2e/const.go +++ b/tests/e2e/const.go @@ -87,7 +87,8 @@ const ( contourName = "contour" // Kubernetes resource kinds. - podsResource = "pods" + podsResource = "pods" + configMapsResource = "configmaps" // Common string values and CLI flags/keys. falseString = "false" diff --git a/tests/e2e/kafkacluster_brokers.go b/tests/e2e/kafkacluster_brokers.go new file mode 100644 index 000000000..5ff1e05ee --- /dev/null +++ b/tests/e2e/kafkacluster_brokers.go @@ -0,0 +1,84 @@ +// Copyright 2026 Adobe. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build e2e + +package e2e + +import ( + "encoding/json" + + "github.com/gruntwork-io/terratest/modules/k8s" + + "github.com/banzaicloud/koperator/api/v1beta1" +) + +// patchKafkaClusterBrokers fetches the running KafkaCluster CR, applies mutate to its current spec.brokers, +// and patches only that field back - leaving every other field (including the surviving brokers' own +// per-broker config) untouched. Scaling tests previously triggered an add/remove by applying a second, +// independently maintained sample manifest for the desired broker count; besides the add/remove itself, that +// manifest can silently carry unrelated drift (e.g. simplekafkacluster.yaml's headlessServiceEnabled: false, +// which raced the operator's headless Service teardown against an in-flight broker removal - see #316). +// Patching only spec.brokers cannot introduce that kind of unrelated drift. +func patchKafkaClusterBrokers(kubectlOptions k8s.KubectlOptions, clusterName string, mutate func([]v1beta1.Broker) []v1beta1.Broker) error { + raw, err := runKubectlSilent(kubectlOptions, "get", "kafkacluster", clusterName, "-o", "json") + if err != nil { + return err + } + + var current struct { + Spec struct { + Brokers []v1beta1.Broker `json:"brokers"` + } `json:"spec"` + } + if err := json.Unmarshal([]byte(raw), ¤t); err != nil { + return err + } + + mergePatch, err := json.Marshal(map[string]any{ + "spec": map[string]any{ + "brokers": mutate(current.Spec.Brokers), + }, + }) + if err != nil { + return err + } + + _, err = runKubectlSilent(kubectlOptions, "patch", "kafkacluster", clusterName, "--type=merge", "-p", string(mergePatch)) + return err +} + +// removeKafkaClusterBrokers patches spec.brokers to exclude the given IDs. See patchKafkaClusterBrokers. +func removeKafkaClusterBrokers(kubectlOptions k8s.KubectlOptions, clusterName string, brokerIDsToRemove ...int32) error { + remove := make(map[int32]bool, len(brokerIDsToRemove)) + for _, id := range brokerIDsToRemove { + remove[id] = true + } + return patchKafkaClusterBrokers(kubectlOptions, clusterName, func(brokers []v1beta1.Broker) []v1beta1.Broker { + remaining := make([]v1beta1.Broker, 0, len(brokers)) + for _, broker := range brokers { + if !remove[broker.Id] { + remaining = append(remaining, broker) + } + } + return remaining + }) +} + +// addKafkaClusterBroker patches spec.brokers to append newBroker. See patchKafkaClusterBrokers. +func addKafkaClusterBroker(kubectlOptions k8s.KubectlOptions, clusterName string, newBroker v1beta1.Broker) error { + return patchKafkaClusterBrokers(kubectlOptions, clusterName, func(brokers []v1beta1.Broker) []v1beta1.Broker { + return append(brokers, newBroker) + }) +} diff --git a/tests/e2e/koperator_suite_test.go b/tests/e2e/koperator_suite_test.go index bf4e4d5f8..eae139116 100644 --- a/tests/e2e/koperator_suite_test.go +++ b/tests/e2e/koperator_suite_test.go @@ -93,6 +93,7 @@ var _ = ginkgo.When("Testing e2e test altogether", ginkgo.Ordered, func() { testInstallKafkaCluster("../../config/samples/kraft/simplekafkacluster_kraft.yaml") testProduceConsumeInternal() testJmxExporter() + testKRaftBrokerScaling() testUninstallKafkaCluster() testUninstall() snapshotClusterAndCompare(snapshottedInfo) diff --git a/tests/e2e/test_broker_removal.go b/tests/e2e/test_broker_removal.go index 9ca5ff8df..ee78ed99e 100644 --- a/tests/e2e/test_broker_removal.go +++ b/tests/e2e/test_broker_removal.go @@ -74,9 +74,10 @@ func testBatchedBrokerRemoval() bool { }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) }) - ginkgo.It("Applying 3-broker manifest to trigger removal of brokers 3 and 4", func() { - ginkgo.By("Patching KafkaCluster to remove brokers 3 and 4") - applyK8sResourceManifest(kubectlOptions, "../../config/samples/simplekafkacluster.yaml") + ginkgo.It("Removing brokers 3 and 4 from the running KafkaCluster", func() { + ginkgo.By("Fetching the running KafkaCluster and patching out brokers 3 and 4") + err := removeKafkaClusterBrokers(kubectlOptions, kafkaClusterName, 3, 4) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) }) ginkgo.It("Waiting for exactly one remove_broker CruiseControlOperation to be created", func() { @@ -112,44 +113,53 @@ func testBatchedBrokerRemoval() bool { // hasExactlyOneRemoveBrokerOperation returns true if there is exactly one CruiseControlOperation // of type remove_broker in the namespace. func hasExactlyOneRemoveBrokerOperation(kubectlOptions k8s.KubectlOptions) (bool, error) { - ops, err := getK8sResources(kubectlOptions, - []string{"cruisecontroloperation"}, - "", - "", - "-o", "jsonpath={range .items[*]}{.status.currentTask.operation}{'\\n'}{end}", - ) - if err != nil { - return false, err - } - - count := 0 - for _, op := range ops { - if op == string(v1alpha1.OperationRemoveBroker) { - count++ - } - } - return count == 1, nil + return hasExactlyNBrokerOperations(kubectlOptions, v1alpha1.OperationRemoveBroker, 1) } -// hasNoInFlightCruiseControlOperation returns true when no CruiseControlOperation in the namespace -// has a currently-running task (Active or InExecution). Completed / CompletedWithError tasks and -// operations without a currentTask count as idle. It is used to gate mutation tests (broker/disk -// removal) on a quiescent Cruise Control so a new operation does not race an in-flight one, such as -// the rebalance CruiseControl runs right after a fresh cluster install. +// hasNoInFlightCruiseControlOperation returns true when no scaling/mutation CruiseControlOperation in the +// namespace has a task that is running or queued. "status" operations (Cruise Control health reads) are +// ignored - they are not mutations and are created without an error policy, so a stale/failed one must not +// wedge this gate. A mutation operation is idle only when koperator considers it finished (matching +// CruiseControlOperation.IsFinished): its currentTask is Completed, or it is CompletedWithError under the +// "ignore" error policy. Everything else with a currentTask is busy - Active / InExecution (running), an +// empty state (queued for first execution), and CompletedWithError under the default "retry" policy (queued +// for retry). Operations without a currentTask are idle. It is used to gate mutation tests (broker/disk +// removal, KRaft scaling) on a quiescent Cruise Control so a new operation does not race an in-flight, +// about-to-run, or about-to-retry one, such as the rebalance CruiseControl runs right after a fresh install. func hasNoInFlightCruiseControlOperation(kubectlOptions k8s.KubectlOptions) (bool, error) { - states, err := getK8sResources(kubectlOptions, + // One line per operation: "//". The first two + // render empty when there is no currentTask, so an operation with a set operation but empty state is a + // queued (not-yet-run) task. + lines, err := getK8sResources(kubectlOptions, []string{"cruisecontroloperation"}, "", "", - "-o", "jsonpath={range .items[*]}{.status.currentTask.state}{'\\n'}{end}", + "-o", "jsonpath={range .items[*]}{.status.currentTask.operation}/{.status.currentTask.state}/{.spec.errorPolicy}{'\\n'}{end}", ) if err != nil { return false, err } - for _, state := range states { - if state == string(v1beta1.CruiseControlTaskActive) || state == string(v1beta1.CruiseControlTaskInExecution) { - return false, nil + for _, line := range lines { + operation, rest, _ := strings.Cut(line, "/") + state, errorPolicy, _ := strings.Cut(rest, "/") + if operation == "" { + // No currentTask yet - nothing running or queued for this operation. + continue } + if operation == string(v1alpha1.OperationStatus) { + // "status" operations are Cruise Control health reads, not scaling/mutation tasks, and the + // operation controller creates them without an error policy - a failed/stale one would otherwise + // wedge this gate. They never conflict with a new scaling operation, so ignore them. + continue + } + taskState := v1beta1.CruiseControlUserTaskState(state) + finished := taskState == v1beta1.CruiseControlTaskCompleted || + (taskState == v1beta1.CruiseControlTaskCompletedWithError && errorPolicy == string(v1alpha1.ErrorPolicyIgnore)) + if finished { + continue + } + // Running, queued for first execution, or CompletedWithError awaiting retry - Cruise Control is busy. + return false, nil } return true, nil } diff --git a/tests/e2e/test_kraft_broker_scaling.go b/tests/e2e/test_kraft_broker_scaling.go new file mode 100644 index 000000000..0a19c4e9b --- /dev/null +++ b/tests/e2e/test_kraft_broker_scaling.go @@ -0,0 +1,302 @@ +// Copyright 2026 Adobe. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build e2e + +package e2e + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "sort" + "strings" + + "github.com/gruntwork-io/terratest/modules/k8s" + ginkgo "github.com/onsi/ginkgo/v2" + gomega "github.com/onsi/gomega" + + "github.com/banzaicloud/koperator/api/v1alpha1" + "github.com/banzaicloud/koperator/api/v1beta1" +) + +// testKRaftBrokerScaling upscales a KRaft cluster from 3 broker-only nodes to 4 (add broker 103) and then +// downscales it back to 3, asserting Cruise Control drives each direction with exactly one +// add_broker / remove_broker operation, the broker-only node set tracks the spec by exact id, and the 3 +// controller-only nodes (ids 0,1,2) are never touched (controller-only nodes are not CC brokers). The +// assertions check exact broker/controller ids - not just pod counts - so a manifest that silently swaps +// ids (rather than adding a single broker) would fail instead of passing green. +// +// This is the regression test for #301, and it deliberately exercises BOTH directions: +// - UPSCALE is the case a "stop hashing capacity.json" fix would have silently broken: the capacity roll +// is what loads the new broker's exact capacity into CC, so the fix must keep rolling on add and +// sequence the add_broker op after the roll settles. +// - DOWNSCALE is the case that stalled on master: the pre-downscale capacity roll kept CC un-ready so the +// remove_broker op was never created. +func testKRaftBrokerScaling() bool { + return ginkgo.When("KRaft broker scaling: upscale then downscale, controllers untouched", func() { + var kubectlOptions k8s.KubectlOptions + var err error + + ginkgo.It("Acquiring K8s config and context", func() { + kubectlOptions, err = kubectlOptionsForCurrentContext() + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + kubectlOptions.Namespace = koperatorLocalHelmDescriptor.Namespace + }) + + ginkgo.It("Waiting for Cruise Control to be ready and settled before upscale", func() { + ginkgo.By("Ensuring the KafkaCluster is running") + gomega.Expect(waitForKafkaClusterWithPodStatusCheck(kubectlOptions, kafkaClusterName, kafkaClusterResourceReadinessTimeout)).NotTo(gomega.HaveOccurred()) + ginkgo.By("Waiting until no Cruise Control operation is in flight (initial rebalance finished)") + gomega.Eventually(context.Background(), func() (bool, error) { + return hasNoInFlightCruiseControlOperation(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + ginkgo.By("Waiting until the Cruise Control Deployment is fully rolled out (single Ready replica)") + gomega.Eventually(context.Background(), func() (bool, error) { + return isCruiseControlDeploymentRolledOut(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + }) + + ginkgo.It("Asserting the cluster starts with broker-only nodes 100,101,102 and controllers 0,1,2", func() { + ids, err := brokerPodIDs(kubectlOptions) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + gomega.Expect(ids).To(gomega.ConsistOf("100", "101", "102"), "unexpected broker-only ids before scaling") + ids, err = controllerPodIDs(kubectlOptions) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + gomega.Expect(ids).To(gomega.ConsistOf("0", "1", "2"), "unexpected controller-only ids before scaling") + }) + + // --- Upscale: add broker 103 --- + + ginkgo.It("Adding broker 103 to the running KafkaCluster", func() { + ginkgo.By("Fetching the running KafkaCluster and adding broker 103") + // Matches the "broker" config group simplekafkacluster_kraft.yaml assigns brokers 100-102. + err := addKafkaClusterBroker(kubectlOptions, kafkaClusterName, v1beta1.Broker{Id: 103, BrokerConfigGroup: "broker"}) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + }) + + ginkgo.It("Waiting for exactly one add_broker CruiseControlOperation", func() { + gomega.Eventually(context.Background(), func() (bool, error) { + return hasExactlyNBrokerOperations(kubectlOptions, v1alpha1.OperationAddBroker, 1) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + }) + + ginkgo.It("Waiting for broker 103 to join (broker-only nodes 100,101,102,103) and the cluster to be healthy", func() { + gomega.Eventually(context.Background(), func() ([]string, error) { + return brokerPodIDs(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.ConsistOf("100", "101", "102", "103")) + gomega.Expect(waitForKafkaClusterWithPodStatusCheck(kubectlOptions, kafkaClusterName, kafkaClusterResourceReadinessTimeout)).NotTo(gomega.HaveOccurred()) + }) + + ginkgo.It("Asserting controllers are untouched after upscale (still exactly controllers 0,1,2)", func() { + ids, err := controllerPodIDs(kubectlOptions) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + gomega.Expect(ids).To(gomega.ConsistOf("0", "1", "2"), "controller-only nodes must not be affected by an upscale") + }) + + ginkgo.It("Asserting Cruise Control's capacity.json gained a real entry for the added broker 103", func() { + // add_broker runs with AllowCapacityEstimation=true, so a healthy add alone does not prove the + // operator wrote broker 103's capacity - CC could have estimated it. Assert the per-broker entry + // is actually present so a regression that stops generating/rolling capacity.json is caught. + gomega.Eventually(context.Background(), func() ([]string, error) { + return cruiseControlCapacityBrokerIDs(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.ContainElement("103"), + "broker 103's capacity must be written to capacity.json, not left to CC estimation") + }) + + ginkgo.It("Asserting the running CC pod's template carries the current capacity.json hash", func() { + // The per-broker entry existing in the ConfigMap is not enough: the capacity must also have been + // hashed into the pod template and rolled out. Assert the hash on the *running* CC pod (not just + // the Deployment template) matches the current capacity.json - so a Ready CC pod started from that + // capacity. It does not by itself prove the CC process re-read the file, but combined with a Ready + // pod it is strong evidence the roll landed rather than add_broker relying on capacity estimation. + gomega.Eventually(context.Background(), func() (bool, error) { + return runningCruiseControlPodHasCurrentCapacityHash(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue(), + "the running CC pod's template must carry the hash of the current capacity.json") + }) + + // --- Downscale: remove broker 103 --- + + ginkgo.It("Waiting for Cruise Control to be settled again before downscale", func() { + gomega.Eventually(context.Background(), func() (bool, error) { + return hasNoInFlightCruiseControlOperation(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + gomega.Eventually(context.Background(), func() (bool, error) { + return isCruiseControlDeploymentRolledOut(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + }) + + ginkgo.It("Removing broker 103 from the running KafkaCluster", func() { + ginkgo.By("Fetching the running KafkaCluster and patching out broker 103") + err := removeKafkaClusterBrokers(kubectlOptions, kafkaClusterName, 103) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + }) + + ginkgo.It("Waiting for exactly one remove_broker CruiseControlOperation", func() { + gomega.Eventually(context.Background(), func() (bool, error) { + return hasExactlyNBrokerOperations(kubectlOptions, v1alpha1.OperationRemoveBroker, 1) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + }) + + ginkgo.It("Waiting for broker 103 to be removed (broker-only nodes back to 100,101,102) and the cluster to be healthy", func() { + gomega.Eventually(context.Background(), func() ([]string, error) { + return brokerPodIDs(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.ConsistOf("100", "101", "102")) + gomega.Expect(waitForKafkaClusterWithPodStatusCheck(kubectlOptions, kafkaClusterName, kafkaClusterResourceReadinessTimeout)).NotTo(gomega.HaveOccurred()) + }) + + ginkgo.It("Asserting controllers are untouched after downscale (still exactly controllers 0,1,2)", func() { + ids, err := controllerPodIDs(kubectlOptions) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + gomega.Expect(ids).To(gomega.ConsistOf("0", "1", "2"), "controller-only nodes must not be affected by a downscale") + }) + + ginkgo.It("Asserting Cruise Control's capacity.json dropped the removed broker 103", func() { + // Once broker 103's pod is gone and it leaves the status, the merge stops preserving its entry + // and capacity.json shrinks back - the departing broker's capacity must not linger indefinitely. + gomega.Eventually(context.Background(), func() ([]string, error) { + return cruiseControlCapacityBrokerIDs(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).ShouldNot(gomega.ContainElement("103"), + "broker 103's capacity must be removed from capacity.json once it is fully downscaled") + }) + }) +} + +// hasExactlyNBrokerOperations returns true when exactly n CruiseControlOperations whose current task is the +// given operation type exist in the namespace. +func hasExactlyNBrokerOperations(kubectlOptions k8s.KubectlOptions, operation v1alpha1.CruiseControlTaskOperation, n int) (bool, error) { + ops, err := getK8sResources(kubectlOptions, + []string{"cruisecontroloperation"}, + "", + "", + "-o", "jsonpath={range .items[*]}{.status.currentTask.operation}{'\\n'}{end}", + ) + if err != nil { + return false, err + } + count := 0 + for _, op := range ops { + if op == string(operation) { + count++ + } + } + return count == n, nil +} + +// brokerPodIDs returns the sorted broker ids (brokerId label) of the Running broker-only pods +// (isControllerNode=false) in the namespace. +func brokerPodIDs(kubectlOptions k8s.KubectlOptions) ([]string, error) { + return podBrokerIDs(kubectlOptions, kafkaLabelSelectorBrokers) +} + +// controllerPodIDs returns the sorted broker ids (brokerId label) of the Running controller-only pods +// (isControllerNode=true) in the namespace. +func controllerPodIDs(kubectlOptions k8s.KubectlOptions) ([]string, error) { + return podBrokerIDs(kubectlOptions, kafkaLabelSelectorControllers) +} + +// podBrokerIDs returns the sorted brokerId label values of the Running pods matching roleSelector. Asserting +// on the exact id set (rather than just a pod count) is what lets the scaling test catch a manifest that +// swaps ids instead of adding/removing a single node. +func podBrokerIDs(kubectlOptions k8s.KubectlOptions, roleSelector string) ([]string, error) { + ids, err := getK8sResources(kubectlOptions, + []string{podsResource}, + v1beta1.KafkaCRLabelKey+"="+kafkaClusterName+","+roleSelector, + "", + "--field-selector=status.phase=Running", + "-o", "jsonpath={range .items[*]}{.metadata.labels.brokerId}{'\\n'}{end}", + ) + if err != nil { + return nil, err + } + sort.Strings(ids) + return ids, nil +} + +// cruiseControlCapacityBrokerIDs returns the sorted broker ids present in Cruise Control's capacity.json +// (read from the CC ConfigMap). Asserting a scaled broker appears/disappears here proves the operator +// actually generated and rolled its per-broker capacity, independently of add_broker succeeding via Cruise +// Control's capacity estimation (AllowCapacityEstimation) - so a regression that stops writing per-broker +// capacity is caught rather than masked by estimation. +func cruiseControlCapacityBrokerIDs(kubectlOptions k8s.KubectlOptions) ([]string, error) { + lines, err := getK8sResources(kubectlOptions, + []string{configMapsResource}, + v1beta1.KafkaCRLabelKey+"="+kafkaClusterName+",app=cruisecontrol", + "", + "-o", "jsonpath={range .items[*]}{.data.capacity\\.json}{end}", + ) + if err != nil { + return nil, err + } + raw := strings.TrimSpace(strings.Join(lines, "\n")) + if raw == "" { + return nil, nil + } + var parsed struct { + BrokerCapacities []struct { + BrokerID string `json:"brokerId"` + } `json:"brokerCapacities"` + } + if err := json.Unmarshal([]byte(raw), &parsed); err != nil { + return nil, err + } + ids := make([]string, 0, len(parsed.BrokerCapacities)) + for _, bc := range parsed.BrokerCapacities { + ids = append(ids, bc.BrokerID) + } + sort.Strings(ids) + return ids, nil +} + +// runningCruiseControlPodHasCurrentCapacityHash reports whether the Running CC pod's template carries the hash +// of the CC ConfigMap's current capacity.json (the annotation koperator stamps in GeneratePodAnnotations). +// Reading the pod (not the Deployment template) shows what the currently-running CC actually started from, so +// with a Ready pod it is evidence the capacity roll landed rather than add_broker relying on capacity +// estimation. The hash is computed the same way the operator does (hex(sha256(capacity.json))). During a +// rollout there may briefly be two CC pods with different hashes; requiring an exact single-value match makes +// this true only once the roll has settled to the current-capacity pod. +func runningCruiseControlPodHasCurrentCapacityHash(kubectlOptions k8s.KubectlOptions) (bool, error) { + ccSelector := v1beta1.KafkaCRLabelKey + "=" + kafkaClusterName + ",app=cruisecontrol" + + cmLines, err := getK8sResources(kubectlOptions, + []string{configMapsResource}, + ccSelector, + "", + "-o", "jsonpath={range .items[*]}{.data.capacity\\.json}{end}", + ) + if err != nil { + return false, err + } + // Do not trim the capacity.json - it is hashed byte-for-byte by the operator. + capacityJSON := strings.Join(cmLines, "\n") + if capacityJSON == "" { + return false, nil + } + + podLines, err := getK8sResources(kubectlOptions, + []string{podsResource}, + ccSelector, + "", + "--field-selector=status.phase=Running", + "-o", "jsonpath={range .items[*]}{.metadata.annotations.cruiseControlCapacity\\.json}{'\\n'}{end}", + ) + if err != nil { + return false, err + } + runningHash := strings.TrimSpace(strings.Join(podLines, "\n")) + sum := sha256.Sum256([]byte(capacityJSON)) + return runningHash != "" && runningHash == hex.EncodeToString(sum[:]), nil +}