From 1c7001fb0a669c9285a525fd9f9860646115a727 Mon Sep 17 00:00:00 2001 From: Sebastien Estienne Date: Sat, 11 Apr 2026 23:15:23 -0700 Subject: [PATCH 1/7] feat(rbgsa-controller): gate scale-down during partition-based rollout StatefulSet/LWS delete pods from the highest ordinals on scale-down, which are the already-updated pods (ordinal >= partition). This destroys rollout progress and forces the rollout to restart from zero. Gate scale-down at the RBGSA controller level for partition-based workloads (StatefulSet, LWS, RoleInstanceSet) while a rolling update is in progress. Scale-up and Deployment-backed roles are unaffected. Changes: - Add Conditions field and ScaleDownDeferred condition to RBGSA status - Add rollout detection via RoleStatus.UpdatedReplicas < Replicas - Add Watches() on RBG for event-driven rollout completion detection - Emit Warning event on first deferral only (rate-limited) - Use SSA for all status writes (consistent field manager ownership) - DeepCopy after Get to prevent informer cache corruption Closes: sgl-project/rbg#269 --- .../rolebasedgroupscalingadapter_types.go | 10 + .../v1alpha2/zz_generated.deepcopy.go | 7 + .../rolebasedgroupscalingadapterstatus.go | 25 +- ...-k8s.io_rolebasedgroupscalingadapters.yaml | 53 ++ internal/controller/workloads/event.go | 3 + ...rolebasedgroupscalingadapter_controller.go | 87 +- ...lebasedgroupscalingadapter_rollout_test.go | 756 ++++++++++++++++++ 7 files changed, 931 insertions(+), 10 deletions(-) create mode 100644 internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go diff --git a/api/workloads/v1alpha2/rolebasedgroupscalingadapter_types.go b/api/workloads/v1alpha2/rolebasedgroupscalingadapter_types.go index 1581b6565..6c71f2df3 100644 --- a/api/workloads/v1alpha2/rolebasedgroupscalingadapter_types.go +++ b/api/workloads/v1alpha2/rolebasedgroupscalingadapter_types.go @@ -30,6 +30,12 @@ type RoleBasedGroupScalingAdapterSpec struct { ScaleTargetRef *AdapterScaleTargetRef `json:"scaleTargetRef"` } +// AdapterConditionScaleDownDeferred indicates that a scale-down request is being +// held because a partition-based rolling update is in progress for the target role. +// This condition is only present when gating is active; it is removed when scaling +// proceeds normally. +const AdapterConditionScaleDownDeferred = "ScaleDownDeferred" + // RoleBasedGroupScalingAdapterStatus shows the current state of a RoleBasedGroupScalingAdapter. type RoleBasedGroupScalingAdapterStatus struct { // Phase indicates the current phase of the RoleBasedGroupScalingAdapter. @@ -46,6 +52,10 @@ type RoleBasedGroupScalingAdapterStatus struct { // LastScaleTime is the last time the RoleBasedGroupScalingAdapter scaled the number of pods, LastScaleTime *metav1.Time `json:"lastScaleTime,omitempty"` + + // Conditions represent the latest available observations of the adapter's state. + // +optional + Conditions []metav1.Condition `json:"conditions,omitempty"` } type AdapterScaleTargetRef struct { diff --git a/api/workloads/v1alpha2/zz_generated.deepcopy.go b/api/workloads/v1alpha2/zz_generated.deepcopy.go index 2552124b9..c613cdd8d 100644 --- a/api/workloads/v1alpha2/zz_generated.deepcopy.go +++ b/api/workloads/v1alpha2/zz_generated.deepcopy.go @@ -607,6 +607,13 @@ func (in *RoleBasedGroupScalingAdapterStatus) DeepCopyInto(out *RoleBasedGroupSc in, out := &in.LastScaleTime, &out.LastScaleTime *out = (*in).DeepCopy() } + if in.Conditions != nil { + in, out := &in.Conditions, &out.Conditions + *out = make([]metav1.Condition, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RoleBasedGroupScalingAdapterStatus. diff --git a/client-go/applyconfiguration/workloads/v1alpha2/rolebasedgroupscalingadapterstatus.go b/client-go/applyconfiguration/workloads/v1alpha2/rolebasedgroupscalingadapterstatus.go index f56519109..b2e4a7756 100644 --- a/client-go/applyconfiguration/workloads/v1alpha2/rolebasedgroupscalingadapterstatus.go +++ b/client-go/applyconfiguration/workloads/v1alpha2/rolebasedgroupscalingadapterstatus.go @@ -19,17 +19,19 @@ package v1alpha2 import ( v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + metav1 "k8s.io/client-go/applyconfigurations/meta/v1" constants "sigs.k8s.io/rbgs/api/workloads/constants" ) // RoleBasedGroupScalingAdapterStatusApplyConfiguration represents a declarative configuration of the RoleBasedGroupScalingAdapterStatus type for use // with apply. type RoleBasedGroupScalingAdapterStatusApplyConfiguration struct { - Phase *constants.AdapterPhase `json:"phase,omitempty"` - Replicas *int32 `json:"replicas,omitempty"` - Selector *string `json:"selector,omitempty"` - ReadyReplicas *int32 `json:"readyReplicas,omitempty"` - LastScaleTime *v1.Time `json:"lastScaleTime,omitempty"` + Phase *constants.AdapterPhase `json:"phase,omitempty"` + Replicas *int32 `json:"replicas,omitempty"` + Selector *string `json:"selector,omitempty"` + ReadyReplicas *int32 `json:"readyReplicas,omitempty"` + LastScaleTime *v1.Time `json:"lastScaleTime,omitempty"` + Conditions []metav1.ConditionApplyConfiguration `json:"conditions,omitempty"` } // RoleBasedGroupScalingAdapterStatusApplyConfiguration constructs a declarative configuration of the RoleBasedGroupScalingAdapterStatus type for use with @@ -77,3 +79,16 @@ func (b *RoleBasedGroupScalingAdapterStatusApplyConfiguration) WithLastScaleTime b.LastScaleTime = &value return b } + +// WithConditions adds the given value to the Conditions field in the declarative configuration +// and returns the receiver, so that objects can be build by chaining "With" function invocations. +// If called multiple times, values provided by each call will be appended to the Conditions field. +func (b *RoleBasedGroupScalingAdapterStatusApplyConfiguration) WithConditions(values ...*metav1.ConditionApplyConfiguration) *RoleBasedGroupScalingAdapterStatusApplyConfiguration { + for i := range values { + if values[i] == nil { + panic("nil value passed to WithConditions") + } + b.Conditions = append(b.Conditions, *values[i]) + } + return b +} diff --git a/config/crd/bases/workloads.x-k8s.io_rolebasedgroupscalingadapters.yaml b/config/crd/bases/workloads.x-k8s.io_rolebasedgroupscalingadapters.yaml index eb9a17000..15421df6d 100644 --- a/config/crd/bases/workloads.x-k8s.io_rolebasedgroupscalingadapters.yaml +++ b/config/crd/bases/workloads.x-k8s.io_rolebasedgroupscalingadapters.yaml @@ -175,6 +175,59 @@ spec: description: RoleBasedGroupScalingAdapterStatus shows the current state of a RoleBasedGroupScalingAdapter. properties: + conditions: + description: Conditions represent the latest available observations + of the adapter's state. + items: + description: Condition contains details for one aspect of the current + state of this API Resource. + properties: + lastTransitionTime: + description: |- + lastTransitionTime is the last time the condition transitioned from one status to another. + This should be when the underlying condition changed. + format: date-time + type: string + message: + description: |- + message is a human readable message indicating details about the transition. + This may be an empty string. + maxLength: 32768 + type: string + observedGeneration: + description: |- + observedGeneration represents the .metadata.generation that the condition was set based upon. + For instance, if .metadata.generation is currently 12, but the .status.conditions[x]. + format: int64 + minimum: 0 + type: integer + reason: + description: reason contains a programmatic identifier indicating + the reason for the condition's last transition. + maxLength: 1024 + minLength: 1 + pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$ + type: string + status: + description: status of the condition, one of True, False, Unknown. + enum: + - "True" + - "False" + - Unknown + type: string + type: + description: type of condition in CamelCase or in foo.example.com/CamelCase. + maxLength: 316 + pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ + type: string + required: + - lastTransitionTime + - message + - reason + - status + - type + type: object + type: array lastScaleTime: description: LastScaleTime is the last time the RoleBasedGroupScalingAdapter scaled the number of pods, diff --git a/internal/controller/workloads/event.go b/internal/controller/workloads/event.go index f455dce29..dd6faee45 100644 --- a/internal/controller/workloads/event.go +++ b/internal/controller/workloads/event.go @@ -46,4 +46,7 @@ const ( FailedScale = "FailedScale" FailedGetRBGRole = "FailedGetRBGRole" FailedGetRBGScalingAdapter = "FailedGetRBGScalingAdapter" + // ScaleDownDeferred is emitted when a scale-down request is held because + // a partition-based rolling update is in progress for the target role. + ScaleDownDeferred = "ScaleDownDeferred" ) diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go b/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go index bea516a04..724171340 100644 --- a/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go @@ -24,6 +24,7 @@ import ( "github.com/pkg/errors" corev1 "k8s.io/api/core/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" @@ -72,13 +73,15 @@ func NewRoleBasedGroupScalingAdapterReconciler(mgr ctrl.Manager) *RoleBasedGroup // +kubebuilder:rbac:groups=workloads.x-k8s.io,resources=rolebasedgroupscalingadapters/status,verbs=get;update;patch // +kubebuilder:rbac:groups=workloads.x-k8s.io,resources=rolebasedgroupscalingadapters/finalizers,verbs=update func (r *RoleBasedGroupScalingAdapterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { - // Fetch the RoleBasedGroupScalingAdapter instance - rbgScalingAdapter := &workloadsv1alpha2.RoleBasedGroupScalingAdapter{} + // Fetch the RoleBasedGroupScalingAdapter instance. + // DeepCopy to avoid mutating the informer cache when modifying conditions. + cached := &workloadsv1alpha2.RoleBasedGroupScalingAdapter{} if err := r.client.Get( - ctx, types.NamespacedName{Name: req.Name, Namespace: req.Namespace}, rbgScalingAdapter, + ctx, types.NamespacedName{Name: req.Name, Namespace: req.Namespace}, cached, ); err != nil { return ctrl.Result{}, client.IgnoreNotFound(err) } + rbgScalingAdapter := cached.DeepCopy() logger := log.FromContext(ctx).WithValues("rbg-scaling-adapter", klog.KObj(rbgScalingAdapter)) ctx = ctrl.LoggerInto(ctx, logger) @@ -134,7 +137,9 @@ func (r *RoleBasedGroupScalingAdapterReconciler) Reconcile(ctx context.Context, logger.Error(err, "Failed to update status", "rbgScalingAdapterName", rbgScalingAdapterName) } } - // TODO: currently reconcile unbound adapter by a default reconcile interval, need to implement a rbg event-driven manager + // The Watches() on RBG enables event-driven reconciliation for bound adapters. + // This RequeueAfter is retained for unbound adapters whose target RBG does not + // yet exist (no watch events generated for non-existent objects). return ctrl.Result{RequeueAfter: 10 * time.Second}, nil } @@ -213,10 +218,55 @@ func (r *RoleBasedGroupScalingAdapterReconciler) Reconcile(ctx context.Context, desiredReplicas, currentReplicas := rbgScalingAdapter.Spec.Replicas, targetRole.Replicas if desiredReplicas == nil || currentReplicas == nil || *rbgScalingAdapter.Spec.Replicas == *targetRole.Replicas { - // nothing to do + if err := r.clearScaleDownDeferredCondition(ctx, rbgScalingAdapter); err != nil { + return ctrl.Result{}, err + } return ctrl.Result{}, nil } + // Gate scale-down for partition-based workloads during rolling update. + // StatefulSet/LWS delete pods from highest ordinals on scale-down, which are + // the already-updated pods (ordinal >= partition). Deferring scale-down until + // the rollout completes prevents rollout progress destruction. + if *desiredReplicas < *currentReplicas && workloadsv1alpha2.IsStatefulRole(targetRole) { + roleStatus, found := rbg.GetRoleStatus(targetRoleName) + if found && roleStatus.UpdatedReplicas < roleStatus.Replicas { + msg := fmt.Sprintf( + "Scale-down to %d replicas deferred (current: %d): partition-based rolling update in progress for role %s (updated %d/%d)", + *desiredReplicas, *currentReplicas, targetRoleName, + roleStatus.UpdatedReplicas, roleStatus.Replicas) + + logger.Info("Deferring scale-down during partition-based rollout", + "desired", *desiredReplicas, "current", *currentReplicas, + "updatedReplicas", roleStatus.UpdatedReplicas, "replicas", roleStatus.Replicas) + + // Only emit the event on first deferral, not on every requeue. + isFirstDeferral := apimeta.FindStatusCondition(rbgScalingAdapter.Status.Conditions, + workloadsv1alpha2.AdapterConditionScaleDownDeferred) == nil + + apimeta.SetStatusCondition(&rbgScalingAdapter.Status.Conditions, metav1.Condition{ + Type: workloadsv1alpha2.AdapterConditionScaleDownDeferred, + Status: metav1.ConditionTrue, + Reason: "RollingUpdateInProgress", + ObservedGeneration: rbgScalingAdapter.Generation, + Message: msg, + }) + if err := r.patchAdapterStatus(ctx, rbgScalingAdapter); err != nil { + return ctrl.Result{}, err + } + + if isFirstDeferral { + r.recorder.Eventf(rbgScalingAdapter, corev1.EventTypeWarning, ScaleDownDeferred, msg) + } + return ctrl.Result{RequeueAfter: 30 * time.Second}, nil + } + } + + // Clear condition in memory; the SSA status patch below persists the change + // in a single API call together with the replica/lastScaleTime update. + apimeta.RemoveStatusCondition(&rbgScalingAdapter.Status.Conditions, + workloadsv1alpha2.AdapterConditionScaleDownDeferred) + logger.Info("Start scaling", "desired replicas", *desiredReplicas, "current replicas", *currentReplicas) // scale role @@ -249,6 +299,30 @@ func (r *RoleBasedGroupScalingAdapterReconciler) Reconcile(ctx context.Context, return ctrl.Result{}, nil } +// patchAdapterStatus persists the RBGSA status using SSA, consistent with all +// other status writes in this controller. +func (r *RoleBasedGroupScalingAdapterReconciler) patchAdapterStatus( + ctx context.Context, rbgScalingAdapter *workloadsv1alpha2.RoleBasedGroupScalingAdapter, +) error { + applyConfig := ToRoleBasedGroupScalingAdapterApplyConfiguration(rbgScalingAdapter). + WithStatus(ToRoleBasedGroupScalingAdapterStatusApplyConfiguration(rbgScalingAdapter.Status, false)) + return utils.PatchObjectApplyConfiguration(ctx, r.client, applyConfig, utils.PatchStatus) +} + +// clearScaleDownDeferredCondition removes the ScaleDownDeferred condition if present +// and persists the change via SSA status patch. +func (r *RoleBasedGroupScalingAdapterReconciler) clearScaleDownDeferredCondition( + ctx context.Context, rbgScalingAdapter *workloadsv1alpha2.RoleBasedGroupScalingAdapter, +) error { + if apimeta.FindStatusCondition(rbgScalingAdapter.Status.Conditions, + workloadsv1alpha2.AdapterConditionScaleDownDeferred) == nil { + return nil + } + apimeta.RemoveStatusCondition(&rbgScalingAdapter.Status.Conditions, + workloadsv1alpha2.AdapterConditionScaleDownDeferred) + return r.patchAdapterStatus(ctx, rbgScalingAdapter) +} + func (r *RoleBasedGroupScalingAdapterReconciler) UpdateAdapterOwnerReference( ctx context.Context, rbgScalingAdapter *workloadsv1alpha2.RoleBasedGroupScalingAdapter, @@ -307,6 +381,9 @@ func ToRoleBasedGroupScalingAdapterStatusApplyConfiguration(status workloadsv1al if scale { statusApplyConfig = statusApplyConfig.WithLastScaleTime(metav1.Now()) } + // Always include conditions so SSA clears conditions that were previously set. + statusApplyConfig = statusApplyConfig.WithConditions( + ToConditionApplyConfigurations(status.Conditions)...) return statusApplyConfig } diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go b/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go new file mode 100644 index 000000000..0ca37a8da --- /dev/null +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go @@ -0,0 +1,756 @@ +/* +Copyright 2026 The RBG Authors. + +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 workloads + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/record" + "k8s.io/utils/ptr" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/log/zap" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + "sigs.k8s.io/rbgs/api/workloads/constants" + workloadsv1alpha2 "sigs.k8s.io/rbgs/api/workloads/v1alpha2" + "sigs.k8s.io/rbgs/test/wrappers" + wrappersv2 "sigs.k8s.io/rbgs/test/wrappers/v1alpha2" +) + +// newTestScheme creates a scheme with all types needed for RBGSA tests. +func newTestScheme() *runtime.Scheme { + scheme := runtime.NewScheme() + _ = workloadsv1alpha2.AddToScheme(scheme) + _ = corev1.AddToScheme(scheme) + _ = appsv1.AddToScheme(scheme) + return scheme +} + +// buildBoundRBGSA creates a bound RBGSA with the given desired and current replicas. +func buildBoundRBGSA(desiredReplicas int32) *workloadsv1alpha2.RoleBasedGroupScalingAdapter { + return &workloadsv1alpha2.RoleBasedGroupScalingAdapter{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-adapter", + Namespace: "default", + Generation: 1, + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: "workloads.x-k8s.io/v1alpha2", + Kind: "RoleBasedGroup", + Name: "test-rbg", + UID: "rbg-test-uid", + BlockOwnerDeletion: ptr.To(true), + }, + }, + }, + Spec: workloadsv1alpha2.RoleBasedGroupScalingAdapterSpec{ + Replicas: ptr.To(desiredReplicas), + ScaleTargetRef: &workloadsv1alpha2.AdapterScaleTargetRef{ + Name: "test-rbg", + Role: "worker", + }, + }, + Status: workloadsv1alpha2.RoleBasedGroupScalingAdapterStatus{ + Phase: constants.AdapterPhaseBound, + Replicas: ptr.To(int32(10)), + }, + } +} + +// buildRBGWithRoleAndStatus creates an RBG with a single role of the given workload type +// and optional role status. +func buildRBGWithRoleAndStatus( + workloadAPIVersion, workloadKind string, + roleReplicas int32, + roleStatuses []workloadsv1alpha2.RoleStatus, +) *workloadsv1alpha2.RoleBasedGroup { + rbg := wrappersv2.BuildBasicRoleBasedGroup("test-rbg", "default"). + WithRoles([]workloadsv1alpha2.RoleSpec{ + wrappersv2.BuildStandaloneRole("worker"). + WithWorkload(workloadAPIVersion, workloadKind). + WithReplicas(roleReplicas). + Obj(), + }).Obj() + if len(roleStatuses) > 0 { + rbg.Status.RoleStatuses = roleStatuses + } + return rbg +} + +// buildSTS creates a StatefulSet for the given RBG role. +func buildSTS(replicas int32) *appsv1.StatefulSet { + return &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-rbg-worker", + Namespace: "default", + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: "workloads.x-k8s.io/v1alpha2", + Kind: "RoleBasedGroup", + Name: "test-rbg", + UID: "rbg-test-uid", + BlockOwnerDeletion: ptr.To(true), + }, + }, + }, + Spec: appsv1.StatefulSetSpec{ + Replicas: ptr.To(replicas), + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + constants.GroupNameLabelKey: "test-rbg", + constants.RoleNameLabelKey: "worker", + }, + }, + Template: wrappers.BuildBasicPodTemplateSpec().Obj(), + }, + } +} + +// reconcileRBGSA runs a single reconciliation cycle for the RBGSA controller. +func reconcileRBGSA(t *testing.T, c client.Client, recorder *record.FakeRecorder) (ctrl.Result, error) { + t.Helper() + reconciler := &RoleBasedGroupScalingAdapterReconciler{ + client: c, + recorder: recorder, + } + req := reconcile.Request{ + NamespacedName: types.NamespacedName{ + Name: "test-adapter", + Namespace: "default", + }, + } + logger := zap.New().WithValues("env", "unit-test") + ctx := ctrl.LoggerInto(context.TODO(), logger) + return reconciler.Reconcile(ctx, req) +} + +// getUpdatedRBGSA fetches the RBGSA after reconciliation. +func getUpdatedRBGSA(t *testing.T, c client.Client) *workloadsv1alpha2.RoleBasedGroupScalingAdapter { + t.Helper() + adapter := &workloadsv1alpha2.RoleBasedGroupScalingAdapter{} + err := c.Get(context.TODO(), types.NamespacedName{Name: "test-adapter", Namespace: "default"}, adapter) + require.NoError(t, err) + return adapter +} + +// getUpdatedRBG fetches the RBG after reconciliation. +func getUpdatedRBG(t *testing.T, c client.Client) *workloadsv1alpha2.RoleBasedGroup { + t.Helper() + rbg := &workloadsv1alpha2.RoleBasedGroup{} + err := c.Get(context.TODO(), types.NamespacedName{Name: "test-rbg", Namespace: "default"}, rbg) + require.NoError(t, err) + return rbg +} + +func TestScaleDownDeferredDuringRollout(t *testing.T) { + scheme := newTestScheme() + + tests := []struct { + name string + // Setup + rbgsa *workloadsv1alpha2.RoleBasedGroupScalingAdapter + workloadAPIVersion string + workloadKind string + roleReplicas int32 + roleStatuses []workloadsv1alpha2.RoleStatus + // Expectations + expectDeferred bool + expectScaleProceeds bool + expectRequeueAfter time.Duration + expectConditionSet bool + expectWarningEvent bool + expectRBGReplicasSet *int32 // if scale proceeds, the RBG role should have this replicas value + }{ + { + name: "scale-down during rollout on StatefulSet is deferred", + rbgsa: buildBoundRBGSA(6), // HPA wants 6, current is 10 + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }, + expectDeferred: true, + expectRequeueAfter: 30 * time.Second, + expectConditionSet: true, + expectWarningEvent: true, + }, + { + name: "scale-down during rollout on LeaderWorkerSet is deferred", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "leaderworkerset.x-k8s.io/v1", + workloadKind: "LeaderWorkerSet", + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }, + expectDeferred: true, + expectRequeueAfter: 30 * time.Second, + expectConditionSet: true, + expectWarningEvent: true, + }, + { + name: "scale-down during rollout on Deployment proceeds (not partition-based)", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "apps/v1", + workloadKind: "Deployment", + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }, + expectDeferred: false, + expectScaleProceeds: true, + expectRBGReplicasSet: ptr.To(int32(6)), + }, + { + name: "scale-down during rollout on RoleInstanceSet (stateful default) is deferred", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "workloads.x-k8s.io/v1alpha2", + workloadKind: "RoleInstanceSet", + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }, + expectDeferred: true, + expectRequeueAfter: 30 * time.Second, + expectConditionSet: true, + expectWarningEvent: true, + }, + { + name: "scale-down when no rollout in progress proceeds normally", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 10, UpdatedReplicas: 10}, + }, + expectDeferred: false, + expectScaleProceeds: true, + expectRBGReplicasSet: ptr.To(int32(6)), + }, + { + name: "scale-up during rollout on StatefulSet proceeds normally", + rbgsa: buildBoundRBGSA(15), // HPA wants 15, current is 10 + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }, + expectDeferred: false, + expectScaleProceeds: true, + expectRBGReplicasSet: ptr.To(int32(15)), + }, + { + name: "scale-down with role status not found proceeds (don't block on missing status)", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + roleReplicas: 10, + roleStatuses: nil, // no role statuses + expectDeferred: false, + expectScaleProceeds: true, + expectRBGReplicasSet: ptr.To(int32(6)), + }, + { + name: "scale-down during initial creation (UpdatedReplicas=0) is deferred", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 0, UpdatedReplicas: 0}, + }, + expectDeferred: true, + expectRequeueAfter: 30 * time.Second, + expectConditionSet: true, + expectWarningEvent: true, + }, + { + name: "scale-down with maxSurge active (Replicas=12 includes surge) is deferred", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 12, ReadyReplicas: 10, UpdatedReplicas: 8}, + }, + expectDeferred: true, + expectRequeueAfter: 30 * time.Second, + expectConditionSet: true, + expectWarningEvent: true, + }, + { + name: "scale-down to zero during rollout is deferred", + rbgsa: buildBoundRBGSA(0), + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }, + expectDeferred: true, + expectRequeueAfter: 30 * time.Second, + expectConditionSet: true, + expectWarningEvent: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + rbg := buildRBGWithRoleAndStatus(tt.workloadAPIVersion, tt.workloadKind, tt.roleReplicas, tt.roleStatuses) + sts := buildSTS(tt.roleReplicas) + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(tt.rbgsa, rbg, sts). + WithStatusSubresource(tt.rbgsa). + Build() + recorder := record.NewFakeRecorder(100) + + result, err := reconcileRBGSA(t, fakeClient, recorder) + require.NoError(t, err) + + if tt.expectDeferred { + // Verify requeue + assert.Equal(t, tt.expectRequeueAfter, result.RequeueAfter, + "expected RequeueAfter=%v, got %v", tt.expectRequeueAfter, result.RequeueAfter) + + // Verify RBG replicas NOT changed + updatedRBG := getUpdatedRBG(t, fakeClient) + role, roleErr := updatedRBG.GetRole("worker") + require.NoError(t, roleErr) + assert.Equal(t, tt.roleReplicas, *role.Replicas, + "RBG role replicas should be unchanged when scale-down is deferred") + } + + if tt.expectScaleProceeds && tt.expectRBGReplicasSet != nil { + // Verify RBG replicas changed + updatedRBG := getUpdatedRBG(t, fakeClient) + role, roleErr := updatedRBG.GetRole("worker") + require.NoError(t, roleErr) + assert.Equal(t, *tt.expectRBGReplicasSet, *role.Replicas, + "RBG role replicas should be updated to desired value") + + // Verify no requeue (immediate completion) + assert.Zero(t, result.RequeueAfter, "should not requeue when scale proceeds") + } + + if tt.expectConditionSet { + // Verify ScaleDownDeferred condition is set + updatedAdapter := getUpdatedRBGSA(t, fakeClient) + cond := apimeta.FindStatusCondition(updatedAdapter.Status.Conditions, + workloadsv1alpha2.AdapterConditionScaleDownDeferred) + require.NotNil(t, cond, "ScaleDownDeferred condition should be set") + assert.Equal(t, metav1.ConditionTrue, cond.Status) + assert.Equal(t, "RollingUpdateInProgress", cond.Reason) + assert.Equal(t, updatedAdapter.Generation, cond.ObservedGeneration, + "ObservedGeneration should match adapter's Generation") + } + + if tt.expectWarningEvent { + // Verify Warning event emitted + select { + case evt := <-recorder.Events: + assert.Contains(t, evt, "Warning") + assert.Contains(t, evt, ScaleDownDeferred) + default: + t.Error("expected a Warning ScaleDownDeferred event but none was emitted") + } + } + }) + } +} + +func TestScaleDownDeferredConditionLifecycle(t *testing.T) { + scheme := newTestScheme() + + t.Run("condition removed after rollout completes and scale-down proceeds", func(t *testing.T) { + // Setup: RBGSA has an existing ScaleDownDeferred condition (from prior deferral) + rbgsa := buildBoundRBGSA(6) + rbgsa.Status.Conditions = []metav1.Condition{ + { + Type: workloadsv1alpha2.AdapterConditionScaleDownDeferred, + Status: metav1.ConditionTrue, + Reason: "RollingUpdateInProgress", + Message: "scale-down deferred", + LastTransitionTime: metav1.Now(), + ObservedGeneration: 1, + }, + } + + // RBG with rollout COMPLETED (UpdatedReplicas == Replicas) + rbg := buildRBGWithRoleAndStatus("apps/v1", "StatefulSet", 10, + []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 10, UpdatedReplicas: 10}, + }) + sts := buildSTS(10) + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(rbgsa, rbg, sts). + WithStatusSubresource(rbgsa). + Build() + recorder := record.NewFakeRecorder(100) + + result, err := reconcileRBGSA(t, fakeClient, recorder) + require.NoError(t, err) + + // Scale should proceed (not deferred) + assert.Zero(t, result.RequeueAfter, "should not requeue when scale proceeds") + + // RBG replicas should be updated + updatedRBG := getUpdatedRBG(t, fakeClient) + role, roleErr := updatedRBG.GetRole("worker") + require.NoError(t, roleErr) + assert.Equal(t, int32(6), *role.Replicas) + }) + + t.Run("HPA adjusts UP during deferral proceeds with scale-up", func(t *testing.T) { + // Setup: RBGSA was deferred at 6, now HPA adjusts to 12 (scale-up) + rbgsa := buildBoundRBGSA(12) // HPA now wants 12 + rbgsa.Status.Conditions = []metav1.Condition{ + { + Type: workloadsv1alpha2.AdapterConditionScaleDownDeferred, + Status: metav1.ConditionTrue, + Reason: "RollingUpdateInProgress", + Message: "previously deferred", + LastTransitionTime: metav1.Now(), + ObservedGeneration: 1, + }, + } + + rbg := buildRBGWithRoleAndStatus("apps/v1", "StatefulSet", 10, + []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }) + sts := buildSTS(10) + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(rbgsa, rbg, sts). + WithStatusSubresource(rbgsa). + Build() + recorder := record.NewFakeRecorder(100) + + result, err := reconcileRBGSA(t, fakeClient, recorder) + require.NoError(t, err) + + // Scale-up should proceed (not deferred) + assert.Zero(t, result.RequeueAfter) + + // RBG replicas should be updated to 12 + updatedRBG := getUpdatedRBG(t, fakeClient) + role, roleErr := updatedRBG.GetRole("worker") + require.NoError(t, roleErr) + assert.Equal(t, int32(12), *role.Replicas) + }) + + t.Run("repeated requeue during deferral does not re-emit event", func(t *testing.T) { + // Setup: RBGSA already has the condition set (simulating a previous requeue) + rbgsa := buildBoundRBGSA(6) + rbgsa.Status.Conditions = []metav1.Condition{ + { + Type: workloadsv1alpha2.AdapterConditionScaleDownDeferred, + Status: metav1.ConditionTrue, + Reason: "RollingUpdateInProgress", + Message: "previously deferred", + LastTransitionTime: metav1.Now(), + ObservedGeneration: 1, + }, + } + + rbg := buildRBGWithRoleAndStatus("apps/v1", "StatefulSet", 10, + []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }) + sts := buildSTS(10) + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(rbgsa, rbg, sts). + WithStatusSubresource(rbgsa). + Build() + recorder := record.NewFakeRecorder(100) + + result, err := reconcileRBGSA(t, fakeClient, recorder) + require.NoError(t, err) + + // Should still be deferred + assert.Equal(t, 30*time.Second, result.RequeueAfter) + + // Should NOT emit a Warning event (condition already existed) + select { + case evt := <-recorder.Events: + t.Errorf("expected no event on repeated requeue, but got: %s", evt) + default: + // No event — correct + } + }) + + t.Run("HPA adjusts to mid-value during deferral stays deferred", func(t *testing.T) { + // RBGSA was deferred at 6, HPA now adjusts to 8 (still < current 10) + rbgsa := buildBoundRBGSA(8) // HPA now wants 8 + rbgsa.Status.Conditions = []metav1.Condition{ + { + Type: workloadsv1alpha2.AdapterConditionScaleDownDeferred, + Status: metav1.ConditionTrue, + Reason: "RollingUpdateInProgress", + Message: "previously deferred", + LastTransitionTime: metav1.Now(), + ObservedGeneration: 1, + }, + } + + rbg := buildRBGWithRoleAndStatus("apps/v1", "StatefulSet", 10, + []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }) + sts := buildSTS(10) + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(rbgsa, rbg, sts). + WithStatusSubresource(rbgsa). + Build() + recorder := record.NewFakeRecorder(100) + + result, err := reconcileRBGSA(t, fakeClient, recorder) + require.NoError(t, err) + + // Still deferred (8 < 10) + assert.Equal(t, 30*time.Second, result.RequeueAfter) + + // RBG replicas should NOT change + updatedRBG := getUpdatedRBG(t, fakeClient) + role, roleErr := updatedRBG.GetRole("worker") + require.NoError(t, roleErr) + assert.Equal(t, int32(10), *role.Replicas, + "RBG replicas should stay at 10 when mid-value scale-down is still deferred") + + // Condition should still be set + updatedAdapter := getUpdatedRBGSA(t, fakeClient) + cond := apimeta.FindStatusCondition(updatedAdapter.Status.Conditions, + workloadsv1alpha2.AdapterConditionScaleDownDeferred) + require.NotNil(t, cond, "ScaleDownDeferred condition should still be set") + assert.Equal(t, metav1.ConditionTrue, cond.Status) + }) +} + +func TestRBGStatusChangedPredicate(t *testing.T) { + tests := []struct { + name string + oldStatus workloadsv1alpha2.RoleBasedGroupStatus + newStatus workloadsv1alpha2.RoleBasedGroupStatus + expect bool + }{ + { + name: "role status changed triggers predicate", + oldStatus: workloadsv1alpha2.RoleBasedGroupStatus{ + RoleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, UpdatedReplicas: 3}, + }, + }, + newStatus: workloadsv1alpha2.RoleBasedGroupStatus{ + RoleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, UpdatedReplicas: 10}, + }, + }, + expect: true, + }, + { + name: "spec-only change does not trigger predicate", + oldStatus: workloadsv1alpha2.RoleBasedGroupStatus{ + RoleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, UpdatedReplicas: 10}, + }, + }, + newStatus: workloadsv1alpha2.RoleBasedGroupStatus{ + RoleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, UpdatedReplicas: 10}, + }, + }, + expect: false, + }, + { + name: "empty to populated status triggers predicate", + oldStatus: workloadsv1alpha2.RoleBasedGroupStatus{}, + newStatus: workloadsv1alpha2.RoleBasedGroupStatus{ + RoleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, UpdatedReplicas: 3}, + }, + }, + expect: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + pred := RBGRoleStatusPredicate() + + oldRBG := &workloadsv1alpha2.RoleBasedGroup{ + ObjectMeta: metav1.ObjectMeta{Name: "test-rbg", Namespace: "default"}, + Status: tt.oldStatus, + } + newRBG := &workloadsv1alpha2.RoleBasedGroup{ + ObjectMeta: metav1.ObjectMeta{Name: "test-rbg", Namespace: "default"}, + Status: tt.newStatus, + } + + updateEvent := event.UpdateEvent{ + ObjectOld: oldRBG, + ObjectNew: newRBG, + } + assert.Equal(t, tt.expect, pred.UpdateFunc(updateEvent)) + }) + } +} + +func TestMapRBGToScalingAdaptersRollout(t *testing.T) { + scheme := newTestScheme() + + t.Run("returns matching RBGSA for RBG", func(t *testing.T) { + rbg := wrappersv2.BuildBasicRoleBasedGroup("my-rbg", "default").Obj() + + matchingAdapter := &workloadsv1alpha2.RoleBasedGroupScalingAdapter{ + ObjectMeta: metav1.ObjectMeta{ + Name: "adapter-1", + Namespace: "default", + Labels: map[string]string{constants.GroupNameLabelKey: "my-rbg"}, + }, + Spec: workloadsv1alpha2.RoleBasedGroupScalingAdapterSpec{ + ScaleTargetRef: &workloadsv1alpha2.AdapterScaleTargetRef{ + Name: "my-rbg", + Role: "worker", + }, + }, + } + nonMatchingAdapter := &workloadsv1alpha2.RoleBasedGroupScalingAdapter{ + ObjectMeta: metav1.ObjectMeta{ + Name: "adapter-2", + Namespace: "default", + Labels: map[string]string{constants.GroupNameLabelKey: "other-rbg"}, + }, + Spec: workloadsv1alpha2.RoleBasedGroupScalingAdapterSpec{ + ScaleTargetRef: &workloadsv1alpha2.AdapterScaleTargetRef{ + Name: "other-rbg", + Role: "worker", + }, + }, + } + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(rbg, matchingAdapter, nonMatchingAdapter). + Build() + + reconciler := &RoleBasedGroupScalingAdapterReconciler{ + client: fakeClient, + } + + requests := reconciler.mapRBGToScalingAdapters(context.TODO(), rbg) + require.Len(t, requests, 1) + assert.Equal(t, "adapter-1", requests[0].Name) + assert.Equal(t, "default", requests[0].Namespace) + }) + + t.Run("returns empty when no RBGSA matches", func(t *testing.T) { + rbg := wrappersv2.BuildBasicRoleBasedGroup("my-rbg", "default").Obj() + + nonMatchingAdapter := &workloadsv1alpha2.RoleBasedGroupScalingAdapter{ + ObjectMeta: metav1.ObjectMeta{ + Name: "adapter-2", + Namespace: "default", + Labels: map[string]string{constants.GroupNameLabelKey: "other-rbg"}, + }, + Spec: workloadsv1alpha2.RoleBasedGroupScalingAdapterSpec{ + ScaleTargetRef: &workloadsv1alpha2.AdapterScaleTargetRef{ + Name: "other-rbg", + Role: "worker", + }, + }, + } + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(rbg, nonMatchingAdapter). + Build() + + reconciler := &RoleBasedGroupScalingAdapterReconciler{ + client: fakeClient, + } + + requests := reconciler.mapRBGToScalingAdapters(context.TODO(), rbg) + assert.Empty(t, requests) + }) + + t.Run("returns multiple RBGSA for multi-role RBG", func(t *testing.T) { + rbg := wrappersv2.BuildBasicRoleBasedGroup("my-rbg", "default").Obj() + + adapter1 := &workloadsv1alpha2.RoleBasedGroupScalingAdapter{ + ObjectMeta: metav1.ObjectMeta{ + Name: "adapter-worker", + Namespace: "default", + Labels: map[string]string{constants.GroupNameLabelKey: "my-rbg"}, + }, + Spec: workloadsv1alpha2.RoleBasedGroupScalingAdapterSpec{ + ScaleTargetRef: &workloadsv1alpha2.AdapterScaleTargetRef{ + Name: "my-rbg", + Role: "worker", + }, + }, + } + adapter2 := &workloadsv1alpha2.RoleBasedGroupScalingAdapter{ + ObjectMeta: metav1.ObjectMeta{ + Name: "adapter-server", + Namespace: "default", + Labels: map[string]string{constants.GroupNameLabelKey: "my-rbg"}, + }, + Spec: workloadsv1alpha2.RoleBasedGroupScalingAdapterSpec{ + ScaleTargetRef: &workloadsv1alpha2.AdapterScaleTargetRef{ + Name: "my-rbg", + Role: "server", + }, + }, + } + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(rbg, adapter1, adapter2). + Build() + + reconciler := &RoleBasedGroupScalingAdapterReconciler{ + client: fakeClient, + } + + requests := reconciler.mapRBGToScalingAdapters(context.TODO(), rbg) + assert.Len(t, requests, 2) + }) +} From 8c6baab5686bbf61dcdf7cae1deccf1f538bf590 Mon Sep 17 00:00:00 2001 From: Sebastien Estienne Date: Sat, 11 Apr 2026 23:24:23 -0700 Subject: [PATCH 2/7] feat(rbgsa-controller): add ScaleDownPolicy for opt-in rollout gating Add ScaleDownPolicy field to ScalingAdapter spec with two values: - DeferDuringRollout (default): defers scale-down during partition-based rollout to prevent rollout progress destruction - Unrestricted: allows scale-down at any time (opt-out) The default is DeferDuringRollout when the field is unset, making the gating behavior the safe default without requiring explicit opt-in. --- .../v1alpha2/rolebasedgroup_types.go | 26 +++++++++ .../v1alpha2/zz_generated.deepcopy.go | 10 ++-- .../workloads/v1alpha2/scalingadapter.go | 24 ++++---- .../workloads.x-k8s.io_rolebasedgroups.yaml | 15 ++--- ...workloads.x-k8s.io_rolebasedgroupsets.yaml | 15 ++--- doc/reference/api.md | 9 +-- ...rolebasedgroupscalingadapter_controller.go | 13 ++++- ...lebasedgroupscalingadapter_rollout_test.go | 58 +++++++++++++++---- 8 files changed, 120 insertions(+), 50 deletions(-) diff --git a/api/workloads/v1alpha2/rolebasedgroup_types.go b/api/workloads/v1alpha2/rolebasedgroup_types.go index 81efb9659..248033c1a 100644 --- a/api/workloads/v1alpha2/rolebasedgroup_types.go +++ b/api/workloads/v1alpha2/rolebasedgroup_types.go @@ -337,6 +337,23 @@ type InstanceComponent struct { Template corev1.PodTemplateSpec `json:"template"` } +// ScaleDownPolicyType defines how scale-down interacts with rolling updates. +type ScaleDownPolicyType string + +const ( + // ScaleDownPolicyDeferDuringRollout defers HPA-driven scale-down for + // partition-based workloads while a rolling update is in progress. + // This prevents rollout progress destruction caused by StatefulSet/LWS + // deleting already-updated pods from the highest ordinals. + ScaleDownPolicyDeferDuringRollout ScaleDownPolicyType = "DeferDuringRollout" + + // ScaleDownPolicyUnrestricted allows scale-down at any time, even during + // an active rollout. Use with caution: for partition-based workloads, + // scale-down during rollout will delete already-updated pods and reset + // rollout progress. + ScaleDownPolicyUnrestricted ScaleDownPolicyType = "Unrestricted" +) + type ScalingAdapter struct { // Enable indicates whether the ScalingAdapter is enabled for the Role. // +optional @@ -348,6 +365,15 @@ type ScalingAdapter struct { // (group-name, role-name) take precedence and cannot be overridden. // +optional Labels map[string]string `json:"labels,omitempty"` + + // ScaleDownPolicy controls scale-down behavior during rolling updates. + // "DeferDuringRollout" defers scale-down for partition-based workloads + // while a rolling update is in progress. + // "Unrestricted" allows scale-down at any time. + // Defaults to "Unrestricted" when unset. + // +optional + // +kubebuilder:validation:Enum=DeferDuringRollout;Unrestricted + ScaleDownPolicy *ScaleDownPolicyType `json:"scaleDownPolicy,omitempty"` } // RoleBasedGroupStatus defines the observed state of RoleBasedGroup. diff --git a/api/workloads/v1alpha2/zz_generated.deepcopy.go b/api/workloads/v1alpha2/zz_generated.deepcopy.go index c613cdd8d..e38419291 100644 --- a/api/workloads/v1alpha2/zz_generated.deepcopy.go +++ b/api/workloads/v1alpha2/zz_generated.deepcopy.go @@ -1466,12 +1466,10 @@ func (in *RolloutStrategy) DeepCopy() *RolloutStrategy { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ScalingAdapter) DeepCopyInto(out *ScalingAdapter) { *out = *in - if in.Labels != nil { - in, out := &in.Labels, &out.Labels - *out = make(map[string]string, len(*in)) - for key, val := range *in { - (*out)[key] = val - } + if in.ScaleDownPolicy != nil { + in, out := &in.ScaleDownPolicy, &out.ScaleDownPolicy + *out = new(ScaleDownPolicyType) + **out = **in } } diff --git a/client-go/applyconfiguration/workloads/v1alpha2/scalingadapter.go b/client-go/applyconfiguration/workloads/v1alpha2/scalingadapter.go index 9eab8f719..5cfe5c536 100644 --- a/client-go/applyconfiguration/workloads/v1alpha2/scalingadapter.go +++ b/client-go/applyconfiguration/workloads/v1alpha2/scalingadapter.go @@ -17,11 +17,15 @@ limitations under the License. package v1alpha2 +import ( + workloadsv1alpha2 "sigs.k8s.io/rbgs/api/workloads/v1alpha2" +) + // ScalingAdapterApplyConfiguration represents a declarative configuration of the ScalingAdapter type for use // with apply. type ScalingAdapterApplyConfiguration struct { - Enable *bool `json:"enable,omitempty"` - Labels map[string]string `json:"labels,omitempty"` + Enable *bool `json:"enable,omitempty"` + ScaleDownPolicy *workloadsv1alpha2.ScaleDownPolicyType `json:"scaleDownPolicy,omitempty"` } // ScalingAdapterApplyConfiguration constructs a declarative configuration of the ScalingAdapter type for use with @@ -38,16 +42,10 @@ func (b *ScalingAdapterApplyConfiguration) WithEnable(value bool) *ScalingAdapte return b } -// WithLabels puts the entries into the Labels field in the declarative configuration -// and returns the receiver, so that objects can be build by chaining "With" function invocations. -// If called multiple times, the entries provided by each call will be put on the Labels field, -// overwriting an existing map entries in Labels field with the same key. -func (b *ScalingAdapterApplyConfiguration) WithLabels(entries map[string]string) *ScalingAdapterApplyConfiguration { - if b.Labels == nil && len(entries) > 0 { - b.Labels = make(map[string]string, len(entries)) - } - for k, v := range entries { - b.Labels[k] = v - } +// WithScaleDownPolicy sets the ScaleDownPolicy field in the declarative configuration to the given value +// and returns the receiver, so that objects can be built by chaining "With" function invocations. +// If called multiple times, the ScaleDownPolicy field is set to the value of the last call. +func (b *ScalingAdapterApplyConfiguration) WithScaleDownPolicy(value workloadsv1alpha2.ScaleDownPolicyType) *ScalingAdapterApplyConfiguration { + b.ScaleDownPolicy = &value return b } diff --git a/config/crd/bases/workloads.x-k8s.io_rolebasedgroups.yaml b/config/crd/bases/workloads.x-k8s.io_rolebasedgroups.yaml index af08445ab..6ffe7c00b 100644 --- a/config/crd/bases/workloads.x-k8s.io_rolebasedgroups.yaml +++ b/config/crd/bases/workloads.x-k8s.io_rolebasedgroups.yaml @@ -32891,14 +32891,15 @@ spec: description: Enable indicates whether the ScalingAdapter is enabled for the Role. type: boolean - labels: - additionalProperties: - type: string + scaleDownPolicy: description: |- - Labels are additional labels to apply to the auto-created - RoleBasedGroupScalingAdapter resource. Controller-managed labels - (group-name, role-name) take precedence and cannot be overridden. - type: object + ScaleDownPolicy controls scale-down behavior during rolling updates. + "DeferDuringRollout" defers scale-down for partition-based workloads + while a rolling update is in progress. + enum: + - DeferDuringRollout + - Unrestricted + type: string type: object servicePorts: items: diff --git a/config/crd/bases/workloads.x-k8s.io_rolebasedgroupsets.yaml b/config/crd/bases/workloads.x-k8s.io_rolebasedgroupsets.yaml index 72e23ef53..f08532a82 100644 --- a/config/crd/bases/workloads.x-k8s.io_rolebasedgroupsets.yaml +++ b/config/crd/bases/workloads.x-k8s.io_rolebasedgroupsets.yaml @@ -33706,14 +33706,15 @@ spec: description: Enable indicates whether the ScalingAdapter is enabled for the Role. type: boolean - labels: - additionalProperties: - type: string + scaleDownPolicy: description: |- - Labels are additional labels to apply to the auto-created - RoleBasedGroupScalingAdapter resource. Controller-managed labels - (group-name, role-name) take precedence and cannot be overridden. - type: object + ScaleDownPolicy controls scale-down behavior during rolling updates. + "DeferDuringRollout" defers scale-down for partition-based workloads + while a rolling update is in progress. + enum: + - DeferDuringRollout + - Unrestricted + type: string type: object servicePorts: items: diff --git a/doc/reference/api.md b/doc/reference/api.md index f0aa29356..246afe4c6 100644 --- a/doc/reference/api.md +++ b/doc/reference/api.md @@ -85,10 +85,11 @@ #### ScalingAdapter - Field | Description ---------|-------------------------------------------------------------------------------------------------------------- - enable | bool — whether the scaling adapter is enabled for the role (default=false) - labels | map[string]string — additional labels for the auto-created RBGSA; controller labels (group-name, role-name) take precedence (optional) + Field | Description +-----------------|-------------------------------------------------------------------------------------------------------------- + enable | bool — whether the scaling adapter is enabled for the role (default=false) + labels | map[string]string — additional labels for the auto-created RBGSA; controller labels (group-name, role-name) take precedence (optional) + scaleDownPolicy | *ScaleDownPolicyType — scale-down behavior during rollout (enum: DeferDuringRollout, Unrestricted; default=Unrestricted) ## RoleBasedGroupStatus diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go b/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go index 724171340..1f7067d79 100644 --- a/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go @@ -228,7 +228,9 @@ func (r *RoleBasedGroupScalingAdapterReconciler) Reconcile(ctx context.Context, // StatefulSet/LWS delete pods from highest ordinals on scale-down, which are // the already-updated pods (ordinal >= partition). Deferring scale-down until // the rollout completes prevents rollout progress destruction. - if *desiredReplicas < *currentReplicas && workloadsv1alpha2.IsStatefulRole(targetRole) { + if *desiredReplicas < *currentReplicas && + workloadsv1alpha2.IsStatefulRole(targetRole) && + shouldDeferScaleDown(targetRole) { roleStatus, found := rbg.GetRoleStatus(targetRoleName) if found && roleStatus.UpdatedReplicas < roleStatus.Replicas { msg := fmt.Sprintf( @@ -309,6 +311,15 @@ func (r *RoleBasedGroupScalingAdapterReconciler) patchAdapterStatus( return utils.PatchObjectApplyConfiguration(ctx, r.client, applyConfig, utils.PatchStatus) } +// shouldDeferScaleDown returns true if the role's ScaleDownPolicy is +// DeferDuringRollout (or unset, which defaults to DeferDuringRollout). +func shouldDeferScaleDown(role *workloadsv1alpha2.RoleSpec) bool { + if role.ScalingAdapter == nil || role.ScalingAdapter.ScaleDownPolicy == nil { + return true // default: defer during rollout + } + return *role.ScalingAdapter.ScaleDownPolicy == workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout +} + // clearScaleDownDeferredCondition removes the ScaleDownDeferred condition if present // and persists the change via SSA status patch. func (r *RoleBasedGroupScalingAdapterReconciler) clearScaleDownDeferredCondition( diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go b/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go index 0ca37a8da..1219ce431 100644 --- a/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go @@ -84,19 +84,23 @@ func buildBoundRBGSA(desiredReplicas int32) *workloadsv1alpha2.RoleBasedGroupSca } // buildRBGWithRoleAndStatus creates an RBG with a single role of the given workload type -// and optional role status. +// and optional role status. scaleDownPolicy can be nil (default: DeferDuringRollout). func buildRBGWithRoleAndStatus( workloadAPIVersion, workloadKind string, roleReplicas int32, roleStatuses []workloadsv1alpha2.RoleStatus, + scaleDownPolicy *workloadsv1alpha2.ScaleDownPolicyType, ) *workloadsv1alpha2.RoleBasedGroup { + role := wrappersv2.BuildStandaloneRole("worker"). + WithWorkload(workloadAPIVersion, workloadKind). + WithReplicas(roleReplicas). + WithScalingAdapter(true) + roleSpec := role.Obj() + if scaleDownPolicy != nil { + roleSpec.ScalingAdapter.ScaleDownPolicy = scaleDownPolicy + } rbg := wrappersv2.BuildBasicRoleBasedGroup("test-rbg", "default"). - WithRoles([]workloadsv1alpha2.RoleSpec{ - wrappersv2.BuildStandaloneRole("worker"). - WithWorkload(workloadAPIVersion, workloadKind). - WithReplicas(roleReplicas). - Obj(), - }).Obj() + WithRoles([]workloadsv1alpha2.RoleSpec{roleSpec}).Obj() if len(roleStatuses) > 0 { rbg.Status.RoleStatuses = roleStatuses } @@ -177,6 +181,7 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { rbgsa *workloadsv1alpha2.RoleBasedGroupScalingAdapter workloadAPIVersion string workloadKind string + scaleDownPolicy *workloadsv1alpha2.ScaleDownPolicyType roleReplicas int32 roleStatuses []workloadsv1alpha2.RoleStatus // Expectations @@ -321,11 +326,40 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { expectConditionSet: true, expectWarningEvent: true, }, + { + name: "scale-down during rollout with Unrestricted policy proceeds", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + scaleDownPolicy: ptr.To(workloadsv1alpha2.ScaleDownPolicyUnrestricted), + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }, + expectDeferred: false, + expectScaleProceeds: true, + expectRBGReplicasSet: ptr.To(int32(6)), + }, + { + name: "scale-down during rollout with explicit DeferDuringRollout policy is deferred", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + scaleDownPolicy: ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout), + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }, + expectDeferred: true, + expectRequeueAfter: 30 * time.Second, + expectConditionSet: true, + expectWarningEvent: true, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - rbg := buildRBGWithRoleAndStatus(tt.workloadAPIVersion, tt.workloadKind, tt.roleReplicas, tt.roleStatuses) + rbg := buildRBGWithRoleAndStatus(tt.workloadAPIVersion, tt.workloadKind, tt.roleReplicas, tt.roleStatuses, tt.scaleDownPolicy) sts := buildSTS(tt.roleReplicas) fakeClient := fake.NewClientBuilder(). @@ -410,7 +444,7 @@ func TestScaleDownDeferredConditionLifecycle(t *testing.T) { rbg := buildRBGWithRoleAndStatus("apps/v1", "StatefulSet", 10, []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 10, UpdatedReplicas: 10}, - }) + }, nil) sts := buildSTS(10) fakeClient := fake.NewClientBuilder(). @@ -450,7 +484,7 @@ func TestScaleDownDeferredConditionLifecycle(t *testing.T) { rbg := buildRBGWithRoleAndStatus("apps/v1", "StatefulSet", 10, []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, - }) + }, nil) sts := buildSTS(10) fakeClient := fake.NewClientBuilder(). @@ -490,7 +524,7 @@ func TestScaleDownDeferredConditionLifecycle(t *testing.T) { rbg := buildRBGWithRoleAndStatus("apps/v1", "StatefulSet", 10, []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, - }) + }, nil) sts := buildSTS(10) fakeClient := fake.NewClientBuilder(). @@ -532,7 +566,7 @@ func TestScaleDownDeferredConditionLifecycle(t *testing.T) { rbg := buildRBGWithRoleAndStatus("apps/v1", "StatefulSet", 10, []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, - }) + }, nil) sts := buildSTS(10) fakeClient := fake.NewClientBuilder(). From c2f9ddec52081b53ca6d1238f61c57bbcb94bee8 Mon Sep 17 00:00:00 2001 From: Sebastien Estienne Date: Sat, 11 Apr 2026 23:31:30 -0700 Subject: [PATCH 3/7] docs: add scaling adapter documentation and fix default ScaleDownPolicy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Change default ScaleDownPolicy from DeferDuringRollout to Unrestricted for backward compatibility — existing deployments are unaffected unless users explicitly opt in with scaleDownPolicy: DeferDuringRollout. Documentation: - Expand doc/features/autoscaler.md with full scaling adapter guide, ScaleDownPolicy explanation, and the partition ordinal problem - Add scaleDownPolicy field to doc/reference/api.md - Add ScaleDownDeferred condition to doc/reference/variables.md - Add autoscaling example link to doc/TOC.md --- doc/TOC.md | 2 + doc/features/autoscaler.md | 105 +++++++++++++++++- doc/reference/variables.md | 8 ++ ...rolebasedgroupscalingadapter_controller.go | 5 +- ...lebasedgroupscalingadapter_rollout_test.go | 32 +++++- 5 files changed, 143 insertions(+), 9 deletions(-) diff --git a/doc/TOC.md b/doc/TOC.md index 4611b2943..509438bb7 100644 --- a/doc/TOC.md +++ b/doc/TOC.md @@ -42,6 +42,8 @@ - [Rolling Update](../examples/basics/rolling-update.yaml) - Failure Handling - [Restart Policy](../examples/basics/restart-policy.yaml) + - Autoscaling + - [PD-Disagg with Scaling Adapter](../examples/pd-disagg/sglang/sglang-pd.yaml) - Scheduling - [Gang Scheduling](../examples/basics/scheduler-plugins-gang.yaml) - Monitoring diff --git a/doc/features/autoscaler.md b/doc/features/autoscaler.md index 8d3bf044d..220d03931 100644 --- a/doc/features/autoscaler.md +++ b/doc/features/autoscaler.md @@ -1,7 +1,110 @@ # Autoscaling -RBG scaling provides a scalingAdapter that supports HPA, KEDA, or KPA to adjust the number of replicas for Roles. +RBG provides a `ScalingAdapter` that supports HPA, KEDA, or KPA to adjust the number of replicas for Roles. The RBGSA exposes `status.readyReplicas` mirrored from the parent RBG's `status.roleStatuses[].readyReplicas`, allowing consumers to read readiness directly without a cross-resource lookup. ![autoscaler](../img/autoscaler.jpg) + +## Enabling the Scaling Adapter + +To enable autoscaling for a role, set `scalingAdapter.enable: true` in the role spec. RBG will automatically create a `RoleBasedGroupScalingAdapter` (RBGSA) resource that exposes a scale subresource for the HPA to target. + +```yaml +apiVersion: workloads.x-k8s.io/v1alpha2 +kind: RoleBasedGroup +metadata: + name: my-inference +spec: + roles: + - name: worker + replicas: 4 + scalingAdapter: + enable: true + workload: + apiVersion: apps/v1 + kind: StatefulSet + template: + # ... +``` + +Once created, an HPA can target the RBGSA: + +```yaml +apiVersion: autoscaling/v2 +kind: HorizontalPodAutoscaler +metadata: + name: my-inference-worker +spec: + scaleTargetRef: + apiVersion: workloads.x-k8s.io/v1alpha2 + kind: RoleBasedGroupScalingAdapter + name: my-inference-worker + minReplicas: 2 + maxReplicas: 20 + metrics: + - type: Resource + resource: + name: cpu + target: + type: Utilization + averageUtilization: 70 +``` + +### ScalingAdapter Configuration + +| Field | Type | Default | Description | +|-----------------|---------------------|----------------|-----------------------------------------------------------------------------| +| enable | bool | false | Whether the scaling adapter is enabled for the role. | +| scaleDownPolicy | ScaleDownPolicyType | Unrestricted | Controls scale-down behavior during rolling updates. | + +### ScaleDownPolicy Values + +| Value | Description | +|--------------------|-----------------------------------------------------------------------------------------------------------------| +| Unrestricted | Allows scale-down at any time, even during an active rollout. This is the default behavior. | +| DeferDuringRollout | Defers scale-down for partition-based workloads (StatefulSet, LWS) while a rolling update is in progress. | + +## Scale-Down During Rolling Updates + +### The Problem + +For partition-based workloads (StatefulSet, LeaderWorkerSet), rolling updates proceed from the highest ordinals down — pods with ordinal >= partition run the new template. However, StatefulSet also deletes from the highest ordinals on scale-down. This means **scale-down during a rollout kills the already-updated pods first**, resetting rollout progress to zero. + +``` +Before scale-down: replicas=10, partition=7 + Pods: [0 1 2 3 4 5 6 | 7 8 9] + ── old template ── ─ new ─ + +HPA scales down to 6: + StatefulSet deletes pods 9, 8, 7 (the updated ones) + +After scale-down: replicas=6, partition=6 + Pods: [0 1 2 3 4 5] + ── all old template ── + Rollout progress: 3 → 0 updated pods +``` + +### The Solution: DeferDuringRollout + +Setting `scaleDownPolicy: DeferDuringRollout` gates scale-down at the RBGSA controller level while a partition-based rolling update is in progress: + +```yaml +scalingAdapter: + enable: true + scaleDownPolicy: DeferDuringRollout +``` + +When gating is active: +- The RBGSA sets a `ScaleDownDeferred` condition (visible via `kubectl get rbgsa -o yaml`) +- A Warning event is emitted on the RBGSA resource +- The controller requeues every 30 seconds to re-check rollout status +- Once the rollout completes (`UpdatedReplicas == Replicas`), the pending scale-down proceeds + +Scale-up operations are never gated — new pods land on the new template automatically (their ordinal >= partition). + +**Note**: This policy only affects partition-based workloads (StatefulSet, LeaderWorkerSet, RoleInstanceSet). Deployment-backed roles are unaffected because Kubernetes Deployments handle concurrent scaling and rollout natively via ReplicaSet proportioning. + +## Example YAMLs + +- [PD-Disagg with Scaling Adapter](../../examples/pd-disagg/sglang/sglang-pd.yaml) diff --git a/doc/reference/variables.md b/doc/reference/variables.md index 7b91be491..deab768f0 100644 --- a/doc/reference/variables.md +++ b/doc/reference/variables.md @@ -19,6 +19,14 @@ rolebasedgroup.workloads.x-k8s.io/exclusive-topology | Declares the topology domain (e.g. kubernetes.io/hostname) for exclusive scheduling. rolebasedgroup.workloads.x-k8s.io/disable-exclusive-topology | Can be set to "true" on a Role template to skip exclusive-topology affinity injection for that role. +## Status Conditions + +### RoleBasedGroupScalingAdapter + + Condition | Description +--------------------|---------------------------------------------------------------------------------------------------------------------- + ScaleDownDeferred | Set to True when a scale-down request is deferred because a partition-based rolling update is in progress. Removed when scaling proceeds. + ## Env Variables Key | Description diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go b/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go index 1f7067d79..1e5d7f22f 100644 --- a/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go @@ -312,10 +312,11 @@ func (r *RoleBasedGroupScalingAdapterReconciler) patchAdapterStatus( } // shouldDeferScaleDown returns true if the role's ScaleDownPolicy is -// DeferDuringRollout (or unset, which defaults to DeferDuringRollout). +// explicitly set to DeferDuringRollout. When unset, defaults to Unrestricted +// (same behavior as before this feature was introduced). func shouldDeferScaleDown(role *workloadsv1alpha2.RoleSpec) bool { if role.ScalingAdapter == nil || role.ScalingAdapter.ScaleDownPolicy == nil { - return true // default: defer during rollout + return false // default: unrestricted (backward compatible) } return *role.ScalingAdapter.ScaleDownPolicy == workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout } diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go b/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go index 1219ce431..1fe9011a2 100644 --- a/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go @@ -193,10 +193,11 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { expectRBGReplicasSet *int32 // if scale proceeds, the RBG role should have this replicas value }{ { - name: "scale-down during rollout on StatefulSet is deferred", + name: "scale-down during rollout on StatefulSet with DeferDuringRollout is deferred", rbgsa: buildBoundRBGSA(6), // HPA wants 6, current is 10 workloadAPIVersion: "apps/v1", workloadKind: "StatefulSet", + scaleDownPolicy: ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout), roleReplicas: 10, roleStatuses: []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, @@ -207,10 +208,11 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { expectWarningEvent: true, }, { - name: "scale-down during rollout on LeaderWorkerSet is deferred", + name: "scale-down during rollout on LeaderWorkerSet with DeferDuringRollout is deferred", rbgsa: buildBoundRBGSA(6), workloadAPIVersion: "leaderworkerset.x-k8s.io/v1", workloadKind: "LeaderWorkerSet", + scaleDownPolicy: ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout), roleReplicas: 10, roleStatuses: []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, @@ -221,10 +223,11 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { expectWarningEvent: true, }, { - name: "scale-down during rollout on Deployment proceeds (not partition-based)", + name: "scale-down during rollout on Deployment with DeferDuringRollout proceeds (not partition-based)", rbgsa: buildBoundRBGSA(6), workloadAPIVersion: "apps/v1", workloadKind: "Deployment", + scaleDownPolicy: ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout), roleReplicas: 10, roleStatuses: []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, @@ -234,10 +237,11 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { expectRBGReplicasSet: ptr.To(int32(6)), }, { - name: "scale-down during rollout on RoleInstanceSet (stateful default) is deferred", + name: "scale-down during rollout on RoleInstanceSet with DeferDuringRollout is deferred", rbgsa: buildBoundRBGSA(6), workloadAPIVersion: "workloads.x-k8s.io/v1alpha2", workloadKind: "RoleInstanceSet", + scaleDownPolicy: ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout), roleReplicas: 10, roleStatuses: []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, @@ -247,6 +251,19 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { expectConditionSet: true, expectWarningEvent: true, }, + { + name: "scale-down during rollout with default policy (nil) proceeds (backward compatible)", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + roleReplicas: 10, + roleStatuses: []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, + }, + expectDeferred: false, + expectScaleProceeds: true, + expectRBGReplicasSet: ptr.To(int32(6)), + }, { name: "scale-down when no rollout in progress proceeds normally", rbgsa: buildBoundRBGSA(6), @@ -289,6 +306,7 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { rbgsa: buildBoundRBGSA(6), workloadAPIVersion: "apps/v1", workloadKind: "StatefulSet", + scaleDownPolicy: ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout), roleReplicas: 10, roleStatuses: []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 0, UpdatedReplicas: 0}, @@ -303,6 +321,7 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { rbgsa: buildBoundRBGSA(6), workloadAPIVersion: "apps/v1", workloadKind: "StatefulSet", + scaleDownPolicy: ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout), roleReplicas: 10, roleStatuses: []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 12, ReadyReplicas: 10, UpdatedReplicas: 8}, @@ -317,6 +336,7 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { rbgsa: buildBoundRBGSA(0), workloadAPIVersion: "apps/v1", workloadKind: "StatefulSet", + scaleDownPolicy: ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout), roleReplicas: 10, roleStatuses: []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, @@ -524,7 +544,7 @@ func TestScaleDownDeferredConditionLifecycle(t *testing.T) { rbg := buildRBGWithRoleAndStatus("apps/v1", "StatefulSet", 10, []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, - }, nil) + }, ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout)) sts := buildSTS(10) fakeClient := fake.NewClientBuilder(). @@ -566,7 +586,7 @@ func TestScaleDownDeferredConditionLifecycle(t *testing.T) { rbg := buildRBGWithRoleAndStatus("apps/v1", "StatefulSet", 10, []workloadsv1alpha2.RoleStatus{ {Name: "worker", Replicas: 10, ReadyReplicas: 7, UpdatedReplicas: 3}, - }, nil) + }, ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout)) sts := buildSTS(10) fakeClient := fake.NewClientBuilder(). From 4728fcbdeaa7a256e5b9821ff659e238eca3ac04 Mon Sep 17 00:00:00 2001 From: Sebastien Estienne Date: Sat, 11 Apr 2026 23:34:23 -0700 Subject: [PATCH 4/7] refactor(rbgsa-controller): extract methods to fix cyclomatic complexity Extract deferScaleDownIfNeeded() and scaleRole() from the Reconcile method to reduce cyclomatic complexity from 31 to under 30. Also fix gofmt import ordering. --- .../v1alpha2/zz_generated.deepcopy.go | 7 + .../workloads/v1alpha2/scalingadapter.go | 15 ++ .../workloads.x-k8s.io_rolebasedgroups.yaml | 8 + ...workloads.x-k8s.io_rolebasedgroupsets.yaml | 8 + doc/features/autoscaler.md | 2 +- ...rolebasedgroupscalingadapter_controller.go | 167 +++++++++++------- ...lebasedgroupscalingadapter_rollout_test.go | 12 +- 7 files changed, 147 insertions(+), 72 deletions(-) diff --git a/api/workloads/v1alpha2/zz_generated.deepcopy.go b/api/workloads/v1alpha2/zz_generated.deepcopy.go index e38419291..33f257f2b 100644 --- a/api/workloads/v1alpha2/zz_generated.deepcopy.go +++ b/api/workloads/v1alpha2/zz_generated.deepcopy.go @@ -1466,6 +1466,13 @@ func (in *RolloutStrategy) DeepCopy() *RolloutStrategy { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ScalingAdapter) DeepCopyInto(out *ScalingAdapter) { *out = *in + if in.Labels != nil { + in, out := &in.Labels, &out.Labels + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } if in.ScaleDownPolicy != nil { in, out := &in.ScaleDownPolicy, &out.ScaleDownPolicy *out = new(ScaleDownPolicyType) diff --git a/client-go/applyconfiguration/workloads/v1alpha2/scalingadapter.go b/client-go/applyconfiguration/workloads/v1alpha2/scalingadapter.go index 5cfe5c536..277089e7b 100644 --- a/client-go/applyconfiguration/workloads/v1alpha2/scalingadapter.go +++ b/client-go/applyconfiguration/workloads/v1alpha2/scalingadapter.go @@ -25,6 +25,7 @@ import ( // with apply. type ScalingAdapterApplyConfiguration struct { Enable *bool `json:"enable,omitempty"` + Labels map[string]string `json:"labels,omitempty"` ScaleDownPolicy *workloadsv1alpha2.ScaleDownPolicyType `json:"scaleDownPolicy,omitempty"` } @@ -42,6 +43,20 @@ func (b *ScalingAdapterApplyConfiguration) WithEnable(value bool) *ScalingAdapte return b } +// WithLabels puts the entries into the Labels field in the declarative configuration +// and returns the receiver, so that objects can be build by chaining "With" function invocations. +// If called multiple times, the entries provided by each call will be put on the Labels field, +// overwriting an existing map entries in Labels field with the same key. +func (b *ScalingAdapterApplyConfiguration) WithLabels(entries map[string]string) *ScalingAdapterApplyConfiguration { + if b.Labels == nil && len(entries) > 0 { + b.Labels = make(map[string]string, len(entries)) + } + for k, v := range entries { + b.Labels[k] = v + } + return b +} + // WithScaleDownPolicy sets the ScaleDownPolicy field in the declarative configuration to the given value // and returns the receiver, so that objects can be built by chaining "With" function invocations. // If called multiple times, the ScaleDownPolicy field is set to the value of the last call. diff --git a/config/crd/bases/workloads.x-k8s.io_rolebasedgroups.yaml b/config/crd/bases/workloads.x-k8s.io_rolebasedgroups.yaml index 6ffe7c00b..a90e7c01a 100644 --- a/config/crd/bases/workloads.x-k8s.io_rolebasedgroups.yaml +++ b/config/crd/bases/workloads.x-k8s.io_rolebasedgroups.yaml @@ -32891,6 +32891,14 @@ spec: description: Enable indicates whether the ScalingAdapter is enabled for the Role. type: boolean + labels: + additionalProperties: + type: string + description: |- + Labels are additional labels to apply to the auto-created + RoleBasedGroupScalingAdapter resource. Controller-managed labels + (group-name, role-name) take precedence and cannot be overridden. + type: object scaleDownPolicy: description: |- ScaleDownPolicy controls scale-down behavior during rolling updates. diff --git a/config/crd/bases/workloads.x-k8s.io_rolebasedgroupsets.yaml b/config/crd/bases/workloads.x-k8s.io_rolebasedgroupsets.yaml index f08532a82..fc4e0873d 100644 --- a/config/crd/bases/workloads.x-k8s.io_rolebasedgroupsets.yaml +++ b/config/crd/bases/workloads.x-k8s.io_rolebasedgroupsets.yaml @@ -33706,6 +33706,14 @@ spec: description: Enable indicates whether the ScalingAdapter is enabled for the Role. type: boolean + labels: + additionalProperties: + type: string + description: |- + Labels are additional labels to apply to the auto-created + RoleBasedGroupScalingAdapter resource. Controller-managed labels + (group-name, role-name) take precedence and cannot be overridden. + type: object scaleDownPolicy: description: |- ScaleDownPolicy controls scale-down behavior during rolling updates. diff --git a/doc/features/autoscaler.md b/doc/features/autoscaler.md index 220d03931..f06ae9556 100644 --- a/doc/features/autoscaler.md +++ b/doc/features/autoscaler.md @@ -71,7 +71,7 @@ spec: For partition-based workloads (StatefulSet, LeaderWorkerSet), rolling updates proceed from the highest ordinals down — pods with ordinal >= partition run the new template. However, StatefulSet also deletes from the highest ordinals on scale-down. This means **scale-down during a rollout kills the already-updated pods first**, resetting rollout progress to zero. -``` +```text Before scale-down: replicas=10, partition=7 Pods: [0 1 2 3 4 5 6 | 7 8 9] ── old template ── ─ new ─ diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go b/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go index 1e5d7f22f..7270bfcd8 100644 --- a/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go @@ -24,8 +24,8 @@ import ( "github.com/pkg/errors" corev1 "k8s.io/api/core/v1" - apimeta "k8s.io/apimachinery/pkg/api/meta" apierrors "k8s.io/apimachinery/pkg/api/errors" + apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" @@ -228,40 +228,11 @@ func (r *RoleBasedGroupScalingAdapterReconciler) Reconcile(ctx context.Context, // StatefulSet/LWS delete pods from highest ordinals on scale-down, which are // the already-updated pods (ordinal >= partition). Deferring scale-down until // the rollout completes prevents rollout progress destruction. - if *desiredReplicas < *currentReplicas && - workloadsv1alpha2.IsStatefulRole(targetRole) && - shouldDeferScaleDown(targetRole) { - roleStatus, found := rbg.GetRoleStatus(targetRoleName) - if found && roleStatus.UpdatedReplicas < roleStatus.Replicas { - msg := fmt.Sprintf( - "Scale-down to %d replicas deferred (current: %d): partition-based rolling update in progress for role %s (updated %d/%d)", - *desiredReplicas, *currentReplicas, targetRoleName, - roleStatus.UpdatedReplicas, roleStatus.Replicas) - - logger.Info("Deferring scale-down during partition-based rollout", - "desired", *desiredReplicas, "current", *currentReplicas, - "updatedReplicas", roleStatus.UpdatedReplicas, "replicas", roleStatus.Replicas) - - // Only emit the event on first deferral, not on every requeue. - isFirstDeferral := apimeta.FindStatusCondition(rbgScalingAdapter.Status.Conditions, - workloadsv1alpha2.AdapterConditionScaleDownDeferred) == nil - - apimeta.SetStatusCondition(&rbgScalingAdapter.Status.Conditions, metav1.Condition{ - Type: workloadsv1alpha2.AdapterConditionScaleDownDeferred, - Status: metav1.ConditionTrue, - Reason: "RollingUpdateInProgress", - ObservedGeneration: rbgScalingAdapter.Generation, - Message: msg, - }) - if err := r.patchAdapterStatus(ctx, rbgScalingAdapter); err != nil { - return ctrl.Result{}, err - } - - if isFirstDeferral { - r.recorder.Eventf(rbgScalingAdapter, corev1.EventTypeWarning, ScaleDownDeferred, msg) - } - return ctrl.Result{RequeueAfter: 30 * time.Second}, nil - } + if deferred, result, err := r.deferScaleDownIfNeeded( + ctx, rbgScalingAdapter, rbg, targetRole, targetRoleName, + *desiredReplicas, *currentReplicas, + ); deferred || err != nil { + return result, err } // Clear condition in memory; the SSA status patch below persists the change @@ -269,36 +240,7 @@ func (r *RoleBasedGroupScalingAdapterReconciler) Reconcile(ctx context.Context, apimeta.RemoveStatusCondition(&rbgScalingAdapter.Status.Conditions, workloadsv1alpha2.AdapterConditionScaleDownDeferred) - logger.Info("Start scaling", "desired replicas", *desiredReplicas, "current replicas", *currentReplicas) - - // scale role - if err := r.updateRoleReplicas(ctx, rbg, targetRoleName, desiredReplicas); err != nil { - r.recorder.Eventf( - rbgScalingAdapter, corev1.EventTypeWarning, FailedScale, - "Failed to scale target role [%s] of rbg [%s] from %v to %v replicas: %v", - targetRoleName, rbgName, *currentReplicas, *desiredReplicas, err, - ) - return ctrl.Result{}, err - } - rbgScalingAdapterApplyConfig := ToRoleBasedGroupScalingAdapterApplyConfiguration(rbgScalingAdapter). - WithStatus( - ToRoleBasedGroupScalingAdapterStatusApplyConfiguration(rbgScalingAdapter.Status, true).WithReplicas(*desiredReplicas), - ) - if err := utils.PatchObjectApplyConfiguration( - ctx, r.client, rbgScalingAdapterApplyConfig, utils.PatchStatus, - ); err != nil { - logger.Error(err, "Failed to update status", "rbgScalingAdapterName", rbgScalingAdapterName) - return ctrl.Result{}, err - } - - logger.Info("Scale successfully", "old replicas", *currentReplicas, "new replicas", *desiredReplicas) - r.recorder.Eventf( - rbgScalingAdapter, corev1.EventTypeNormal, SuccessfulScale, - "Succeed to scale target role [%s] of rbg [%s] from %v to %v replicas", - targetRoleName, rbgName, *currentReplicas, *desiredReplicas, - ) - - return ctrl.Result{}, nil + return r.scaleRole(ctx, rbgScalingAdapter, rbg, targetRoleName, *desiredReplicas, *currentReplicas) } // patchAdapterStatus persists the RBGSA status using SSA, consistent with all @@ -335,6 +277,101 @@ func (r *RoleBasedGroupScalingAdapterReconciler) clearScaleDownDeferredCondition return r.patchAdapterStatus(ctx, rbgScalingAdapter) } +// deferScaleDownIfNeeded checks whether a scale-down should be deferred because +// a partition-based rolling update is in progress. Returns (true, result, nil) +// if the scale-down was deferred, (false, {}, nil) if it should proceed. +func (r *RoleBasedGroupScalingAdapterReconciler) deferScaleDownIfNeeded( + ctx context.Context, + rbgScalingAdapter *workloadsv1alpha2.RoleBasedGroupScalingAdapter, + rbg *workloadsv1alpha2.RoleBasedGroup, + targetRole *workloadsv1alpha2.RoleSpec, + targetRoleName string, + desiredReplicas, currentReplicas int32, +) (bool, ctrl.Result, error) { + if desiredReplicas >= currentReplicas || + !workloadsv1alpha2.IsStatefulRole(targetRole) || + !shouldDeferScaleDown(targetRole) { + return false, ctrl.Result{}, nil + } + + roleStatus, found := rbg.GetRoleStatus(targetRoleName) + if !found || roleStatus.UpdatedReplicas >= roleStatus.Replicas { + return false, ctrl.Result{}, nil + } + + logger := log.FromContext(ctx) + msg := fmt.Sprintf( + "Scale-down to %d replicas deferred (current: %d): partition-based rolling update in progress for role %s (updated %d/%d)", + desiredReplicas, currentReplicas, targetRoleName, + roleStatus.UpdatedReplicas, roleStatus.Replicas) + + logger.Info("Deferring scale-down during partition-based rollout", + "desired", desiredReplicas, "current", currentReplicas, + "updatedReplicas", roleStatus.UpdatedReplicas, "replicas", roleStatus.Replicas) + + // Only emit the event on first deferral, not on every requeue. + isFirstDeferral := apimeta.FindStatusCondition(rbgScalingAdapter.Status.Conditions, + workloadsv1alpha2.AdapterConditionScaleDownDeferred) == nil + + apimeta.SetStatusCondition(&rbgScalingAdapter.Status.Conditions, metav1.Condition{ + Type: workloadsv1alpha2.AdapterConditionScaleDownDeferred, + Status: metav1.ConditionTrue, + Reason: "RollingUpdateInProgress", + ObservedGeneration: rbgScalingAdapter.Generation, + Message: msg, + }) + if err := r.patchAdapterStatus(ctx, rbgScalingAdapter); err != nil { + return true, ctrl.Result{}, err + } + + if isFirstDeferral { + r.recorder.Eventf(rbgScalingAdapter, corev1.EventTypeWarning, ScaleDownDeferred, msg) + } + return true, ctrl.Result{RequeueAfter: 30 * time.Second}, nil +} + +// scaleRole performs the actual replica update on the RBG and patches the RBGSA status. +func (r *RoleBasedGroupScalingAdapterReconciler) scaleRole( + ctx context.Context, + rbgScalingAdapter *workloadsv1alpha2.RoleBasedGroupScalingAdapter, + rbg *workloadsv1alpha2.RoleBasedGroup, + targetRoleName string, + desiredReplicas, currentReplicas int32, +) (ctrl.Result, error) { + logger := log.FromContext(ctx) + rbgName := rbg.Name + + logger.Info("Start scaling", "desired replicas", desiredReplicas, "current replicas", currentReplicas) + + if err := r.updateRoleReplicas(ctx, rbg, targetRoleName, &desiredReplicas); err != nil { + r.recorder.Eventf( + rbgScalingAdapter, corev1.EventTypeWarning, FailedScale, + "Failed to scale target role [%s] of rbg [%s] from %v to %v replicas: %v", + targetRoleName, rbgName, currentReplicas, desiredReplicas, err, + ) + return ctrl.Result{}, err + } + rbgScalingAdapterApplyConfig := ToRoleBasedGroupScalingAdapterApplyConfiguration(rbgScalingAdapter). + WithStatus( + ToRoleBasedGroupScalingAdapterStatusApplyConfiguration(rbgScalingAdapter.Status, true).WithReplicas(desiredReplicas), + ) + if err := utils.PatchObjectApplyConfiguration( + ctx, r.client, rbgScalingAdapterApplyConfig, utils.PatchStatus, + ); err != nil { + logger.Error(err, "Failed to update status for %s", rbgScalingAdapter.Name) + return ctrl.Result{}, err + } + + logger.Info("Scale successfully", "old replicas", currentReplicas, "new replicas", desiredReplicas) + r.recorder.Eventf( + rbgScalingAdapter, corev1.EventTypeNormal, SuccessfulScale, + "Succeed to scale target role [%s] of rbg [%s] from %v to %v replicas", + targetRoleName, rbgName, currentReplicas, desiredReplicas, + ) + + return ctrl.Result{}, nil +} + func (r *RoleBasedGroupScalingAdapterReconciler) UpdateAdapterOwnerReference( ctx context.Context, rbgScalingAdapter *workloadsv1alpha2.RoleBasedGroupScalingAdapter, diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go b/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go index 1fe9011a2..631231a9c 100644 --- a/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go @@ -291,12 +291,12 @@ func TestScaleDownDeferredDuringRollout(t *testing.T) { expectRBGReplicasSet: ptr.To(int32(15)), }, { - name: "scale-down with role status not found proceeds (don't block on missing status)", - rbgsa: buildBoundRBGSA(6), - workloadAPIVersion: "apps/v1", - workloadKind: "StatefulSet", - roleReplicas: 10, - roleStatuses: nil, // no role statuses + name: "scale-down with role status not found proceeds (don't block on missing status)", + rbgsa: buildBoundRBGSA(6), + workloadAPIVersion: "apps/v1", + workloadKind: "StatefulSet", + roleReplicas: 10, + roleStatuses: nil, // no role statuses expectDeferred: false, expectScaleProceeds: true, expectRBGReplicasSet: ptr.To(int32(6)), From 68d2ffa085c014d05f0542604fd8953fa5233cf2 Mon Sep 17 00:00:00 2001 From: Sebastien Estienne Date: Sat, 11 Apr 2026 23:47:38 -0700 Subject: [PATCH 5/7] ci: remove unused kubebuilder v2.3.1 download from unit test job The kubebuilder binary is not used by `make test` (which runs plain `go test`). The v2.3.1 download fails on arm64 runners since that release predates arm64 support, causing spurious CI failures. Envtest tests use `setup-envtest` (installed via Makefile) instead. --- .github/workflows/project-check.yml | 8 -------- 1 file changed, 8 deletions(-) diff --git a/.github/workflows/project-check.yml b/.github/workflows/project-check.yml index e40ad926d..c0fe62d22 100644 --- a/.github/workflows/project-check.yml +++ b/.github/workflows/project-check.yml @@ -105,14 +105,6 @@ jobs: path: ${{ env.GOPATH }}/src/sigs.k8s.io/rbgs fetch-depth: 0 - - name: Test Preparation - run: | - arch=$(go env GOARCH) - os=$(go env GOOS) - curl -L https://github.com/kubernetes-sigs/kubebuilder/releases/download/v2.3.1/kubebuilder_2.3.1_${os}_${arch}.tar.gz | tar -xz -C /tmp/ - sudo mv /tmp/kubebuilder_2.3.1_${os}_${arch} /usr/local/kubebuilder - export PATH=$PATH:/usr/local/kubebuilder/bin - - name: Run Unit Tests run: | make test From 9e754d41bf10aed9a4be8f8d25d62b77ca2d5d06 Mon Sep 17 00:00:00 2001 From: Sebastien Estienne Date: Sun, 12 Apr 2026 00:09:34 -0700 Subject: [PATCH 6/7] test(envtest): add integration tests for rollout-aware RBGSA scaling Add 6 envtest tests covering the full controller interaction: StatefulSet with DeferDuringRollout policy: - Scale-down deferred during rollout (condition set, replicas unchanged) - Scale-down proceeds after rollout completes (via RBG status watch) - Scale-up during rollout proceeds normally StatefulSet with Unrestricted policy: - Scale-down during rollout proceeds (no gating) Deployment with DeferDuringRollout policy: - Scale-down during rollout proceeds (Deployments are not partition-based) Existing basic_test.go test continues to pass. In envtest, STS status is manually patched to simulate rollout state since no real StatefulSet controller runs. For the "rollout completes" test, RBG status is patched directly because the RBG controller cannot maintain accurate RoleStatuses without a real STS controller to track ObservedGeneration. --- .../rbg/scaling_adapter_rollout_test.go | 350 ++++++++++++++++++ 1 file changed, 350 insertions(+) create mode 100644 test/envtest/testcase/rbg/scaling_adapter_rollout_test.go diff --git a/test/envtest/testcase/rbg/scaling_adapter_rollout_test.go b/test/envtest/testcase/rbg/scaling_adapter_rollout_test.go new file mode 100644 index 000000000..5e1133ecd --- /dev/null +++ b/test/envtest/testcase/rbg/scaling_adapter_rollout_test.go @@ -0,0 +1,350 @@ +/* +Copyright 2026 The RBG Authors. + +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 rbg + +import ( + "fmt" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + appsv1 "k8s.io/api/apps/v1" + autoscalingv1 "k8s.io/api/autoscaling/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + workloadsv1alpha2 "sigs.k8s.io/rbgs/api/workloads/v1alpha2" + "sigs.k8s.io/rbgs/pkg/scale" + "sigs.k8s.io/rbgs/test/envtest/testutil" + wrappersv2 "sigs.k8s.io/rbgs/test/wrappers/v1alpha2" +) + +var _ = Describe("ScalingAdapter Rollout-Aware Scaling", func() { + const ( + timeout = time.Second * 30 + interval = time.Millisecond * 250 + ) + + var testNs string + + BeforeEach(func() { + testNs = fmt.Sprintf("test-sa-rollout-%d", time.Now().UnixNano()) + testutil.CreateNamespace(testNs) + }) + + AfterEach(func() { + testutil.DeleteNamespace(testNs) + }) + + // patchSTSStatus patches a StatefulSet's status to simulate rollout state. + // In envtest, no STS controller runs, so we must manually set status fields. + // This must be called repeatedly because the RBG controller patches the STS + // spec on each reconcile, incrementing Generation and invalidating our + // ObservedGeneration. + patchSTSStatus := func(ns, stsName string, replicas, readyReplicas, updatedReplicas int32) { + Eventually(func() error { + sts := &appsv1.StatefulSet{} + if err := testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: stsName, Namespace: ns}, sts); err != nil { + return err + } + patch := client.MergeFrom(sts.DeepCopy()) + sts.Status.Replicas = replicas + sts.Status.ReadyReplicas = readyReplicas + sts.Status.UpdatedReplicas = updatedReplicas + sts.Status.ObservedGeneration = sts.Generation + return testutil.K8sClient.Status().Patch(testutil.Ctx, sts, patch) + }, timeout, interval).Should(Succeed(), "should patch STS %s status", stsName) + } + + // scaleRBGSA scales the RBGSA via the scale subresource (simulating HPA). + scaleRBGSA := func(ns, rbgsaName string, replicas int32) { + rbgsa := &workloadsv1alpha2.RoleBasedGroupScalingAdapter{} + Eventually(func() error { + return testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: rbgsaName, Namespace: ns}, rbgsa) + }, timeout, interval).Should(Succeed(), "RBGSA %s should exist", rbgsaName) + + // Wait for adapter to be bound + Eventually(func() bool { + if err := testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: rbgsaName, Namespace: ns}, rbgsa); err != nil { + return false + } + return rbgsa.Status.Phase == "Bound" + }, timeout, interval).Should(BeTrue(), "RBGSA should be in Bound phase") + + scaleObj := &autoscalingv1.Scale{} + Expect(testutil.K8sClient.SubResource("scale").Get(testutil.Ctx, rbgsa, scaleObj)). + Should(Succeed()) + scaleObj.Spec.Replicas = replicas + Expect(testutil.K8sClient.SubResource("scale").Update(testutil.Ctx, rbgsa, + client.WithSubResourceBody(scaleObj))).Should(Succeed()) + } + + // getRBGRoleReplicas returns the current replicas for a role in the RBG spec. + getRBGRoleReplicas := func(ns, rbgName, roleName string) func() int32 { + return func() int32 { + rbg := &workloadsv1alpha2.RoleBasedGroup{} + if err := testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: rbgName, Namespace: ns}, rbg); err != nil { + return -1 + } + role, err := rbg.GetRole(roleName) + if err != nil { + return -1 + } + if role.Replicas == nil { + return -1 + } + return *role.Replicas + } + } + + // hasScaleDownDeferredCondition checks if the RBGSA has the ScaleDownDeferred condition. + hasScaleDownDeferredCondition := func(ns, rbgsaName string) func() bool { + return func() bool { + rbgsa := &workloadsv1alpha2.RoleBasedGroupScalingAdapter{} + if err := testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: rbgsaName, Namespace: ns}, rbgsa); err != nil { + return false + } + cond := apimeta.FindStatusCondition(rbgsa.Status.Conditions, + workloadsv1alpha2.AdapterConditionScaleDownDeferred) + return cond != nil && cond.Status == metav1.ConditionTrue + } + } + + Context("StatefulSet with DeferDuringRollout policy", func() { + var ( + rbgName, stsName, rbgsaName string + ) + + BeforeEach(func() { + rbgName = "test-defer-sts" + stsName = rbgName + "-worker" + rbgsaName = scale.GenerateScalingAdapterName(rbgName, "worker") + + role := wrappersv2.BuildStandaloneRole("worker"). + WithWorkload("apps/v1", "StatefulSet"). + WithReplicas(3). + WithScalingAdapter(true). + Obj() + role.ScalingAdapter.ScaleDownPolicy = ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout) + + rbg := wrappersv2.BuildBasicRoleBasedGroup(rbgName, testNs). + WithRoles([]workloadsv1alpha2.RoleSpec{role}).Obj() + + Expect(testutil.K8sClient.Create(testutil.Ctx, rbg)).Should(Succeed()) + + // Wait for STS to be created by the RBG controller + Eventually(func() error { + return testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: stsName, Namespace: testNs}, + &appsv1.StatefulSet{}) + }, timeout, interval).Should(Succeed()) + }) + + It("should defer scale-down during rollout", func() { + // Simulate rollout in progress + patchSTSStatus(testNs, stsName, 3, 3, 1) + + // Wait for RBG status to reflect the rollout + Eventually(func() int32 { + rbg := &workloadsv1alpha2.RoleBasedGroup{} + if err := testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: rbgName, Namespace: testNs}, rbg); err != nil { + return -1 + } + status, found := rbg.GetRoleStatus("worker") + if !found { + return -1 + } + return status.UpdatedReplicas + }, timeout, interval).Should(Equal(int32(1))) + + // Scale down via RBGSA (simulating HPA) + scaleRBGSA(testNs, rbgsaName, 1) + + // Verify: RBG replicas should NOT change (scale-down deferred) + Consistently(getRBGRoleReplicas(testNs, rbgName, "worker"), + 5*time.Second, interval).Should(Equal(int32(3)), + "RBG role replicas should stay at 3 while rollout is in progress") + + // Verify: ScaleDownDeferred condition should be set + Eventually(hasScaleDownDeferredCondition(testNs, rbgsaName), + timeout, interval).Should(BeTrue(), + "RBGSA should have ScaleDownDeferred condition") + }) + + It("should proceed with scale-down after rollout completes", func() { + // Simulate rollout in progress + patchSTSStatus(testNs, stsName, 3, 3, 1) + + // Wait for RBG status to reflect rollout + Eventually(func() int32 { + rbg := &workloadsv1alpha2.RoleBasedGroup{} + if err := testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: rbgName, Namespace: testNs}, rbg); err != nil { + return -1 + } + status, found := rbg.GetRoleStatus("worker") + if !found { + return -1 + } + return status.UpdatedReplicas + }, timeout, interval).Should(Equal(int32(1))) + + // Scale down (will be deferred) + scaleRBGSA(testNs, rbgsaName, 1) + + // Confirm it's deferred + Eventually(hasScaleDownDeferredCondition(testNs, rbgsaName), + timeout, interval).Should(BeTrue()) + + // Simulate rollout completion by patching the RBG status directly. + // In envtest, no real STS controller runs to maintain ObservedGeneration, + // so the RBG controller can't construct RoleStatuses from STS status. + // Patching RBG status directly triggers the RBGSA controller's RBG watch. + Eventually(func() error { + rbg := &workloadsv1alpha2.RoleBasedGroup{} + if err := testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: rbgName, Namespace: testNs}, rbg); err != nil { + return err + } + patch := client.MergeFrom(rbg.DeepCopy()) + rbg.Status.RoleStatuses = []workloadsv1alpha2.RoleStatus{ + {Name: "worker", Replicas: 3, ReadyReplicas: 3, UpdatedReplicas: 3}, + } + return testutil.K8sClient.Status().Patch(testutil.Ctx, rbg, patch) + }, timeout, interval).Should(Succeed()) + + // Verify: RBG replicas should eventually change to 1. + // The RBGSA controller's RBG watch fires on RoleStatuses change, + // then the gating check passes (UpdatedReplicas == Replicas). + Eventually(getRBGRoleReplicas(testNs, rbgName, "worker"), + timeout, interval).Should(Equal(int32(1)), + "RBG role replicas should change to 1 after rollout completes") + }) + + It("should allow scale-up during rollout", func() { + // Simulate rollout in progress + patchSTSStatus(testNs, stsName, 3, 3, 1) + + // Wait for RBG status to reflect rollout + Eventually(func() int32 { + rbg := &workloadsv1alpha2.RoleBasedGroup{} + if err := testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: rbgName, Namespace: testNs}, rbg); err != nil { + return -1 + } + status, found := rbg.GetRoleStatus("worker") + if !found { + return -1 + } + return status.UpdatedReplicas + }, timeout, interval).Should(Equal(int32(1))) + + // Scale UP (should proceed immediately, not gated) + scaleRBGSA(testNs, rbgsaName, 5) + + Eventually(getRBGRoleReplicas(testNs, rbgName, "worker"), + timeout, interval).Should(Equal(int32(5)), + "Scale-up should proceed even during rollout") + }) + }) + + Context("StatefulSet with Unrestricted policy", func() { + It("should allow scale-down during rollout", func() { + rbgName := "test-unrestricted" + stsName := rbgName + "-worker" + rbgsaName := scale.GenerateScalingAdapterName(rbgName, "worker") + + role := wrappersv2.BuildStandaloneRole("worker"). + WithWorkload("apps/v1", "StatefulSet"). + WithReplicas(3). + WithScalingAdapter(true). + Obj() + role.ScalingAdapter.ScaleDownPolicy = ptr.To(workloadsv1alpha2.ScaleDownPolicyUnrestricted) + + rbg := wrappersv2.BuildBasicRoleBasedGroup(rbgName, testNs). + WithRoles([]workloadsv1alpha2.RoleSpec{role}).Obj() + + Expect(testutil.K8sClient.Create(testutil.Ctx, rbg)).Should(Succeed()) + + // Wait for STS and simulate rollout in progress + Eventually(func() error { + return testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: stsName, Namespace: testNs}, + &appsv1.StatefulSet{}) + }, timeout, interval).Should(Succeed()) + + patchSTSStatus(testNs, stsName, 3, 3, 1) + + // Scale down (should proceed — Unrestricted policy) + scaleRBGSA(testNs, rbgsaName, 1) + + Eventually(getRBGRoleReplicas(testNs, rbgName, "worker"), + timeout, interval).Should(Equal(int32(1)), + "Scale-down should proceed with Unrestricted policy") + }) + }) + + Context("Deployment with DeferDuringRollout policy", func() { + It("should allow scale-down during rollout (Deployments are not partition-based)", func() { + rbgName := "test-deploy-defer" + rbgsaName := scale.GenerateScalingAdapterName(rbgName, "worker") + + role := wrappersv2.BuildStandaloneRole("worker"). + WithWorkload("apps/v1", "Deployment"). + WithReplicas(3). + WithScalingAdapter(true). + Obj() + role.ScalingAdapter.ScaleDownPolicy = ptr.To(workloadsv1alpha2.ScaleDownPolicyDeferDuringRollout) + + rbg := wrappersv2.BuildBasicRoleBasedGroup(rbgName, testNs). + WithRoles([]workloadsv1alpha2.RoleSpec{role}).Obj() + + Expect(testutil.K8sClient.Create(testutil.Ctx, rbg)).Should(Succeed()) + + // Wait for Deployment to be created + deployName := rbgName + "-worker" + deploy := &appsv1.Deployment{} + Eventually(func() error { + return testutil.K8sClient.Get(testutil.Ctx, + types.NamespacedName{Name: deployName, Namespace: testNs}, deploy) + }, timeout, interval).Should(Succeed()) + + // Simulate rollout in progress on Deployment + patch := client.MergeFrom(deploy.DeepCopy()) + deploy.Status.Replicas = 3 + deploy.Status.ReadyReplicas = 3 + deploy.Status.UpdatedReplicas = 1 + deploy.Status.ObservedGeneration = deploy.Generation + Expect(testutil.K8sClient.Status().Patch(testutil.Ctx, deploy, patch)).Should(Succeed()) + + // Scale down (should proceed — Deployments are not partition-based) + scaleRBGSA(testNs, rbgsaName, 1) + + Eventually(getRBGRoleReplicas(testNs, rbgName, "worker"), + timeout, interval).Should(Equal(int32(1)), + "Scale-down should proceed for Deployment even with DeferDuringRollout policy") + }) + }) +}) From 546656825b6fca0b894d46d5a4311a1aec83d35c Mon Sep 17 00:00:00 2001 From: Sebastien Estienne Date: Sat, 25 Apr 2026 14:44:10 -0700 Subject: [PATCH 7/7] feat(rbgsa-controller): wake on per-role spec.replicas changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extend RBGRoleSpecOrStatusPredicate (renamed from RBGRoleStatusPredicate) to also fire when any role's spec.replicas value changes between the old and new RBG. Without this, the RBGSA controller's writeback path only wakes on status changes, so an external mutator that stomps spec.roles[].replicas (e.g. helm upgrade dropping the field, kubectl edit, non-SSA scripts) waits for the next status-change event before the apiserver-stored spec is corrected. Pairs with the in-memory override added in the RBG controller (PR #306): the override prevents the workload from being sized to the stale value on the next reconcile, and this predicate change closes the latency gap by waking the writeback immediately on the stomp instead of when status next changes. The new rolesReplicasDiffer helper matches roles by name; added, removed, and renamed roles all count as a difference. We deliberately do not fire on every spec change — other fields (image, env, command) churn frequently and the adapter reconcile is irrelevant to them. --- ...rolebasedgroupscalingadapter_controller.go | 48 ++++++- ...asedgroupscalingadapter_controller_test.go | 121 +++++++++++++----- ...lebasedgroupscalingadapter_rollout_test.go | 2 +- 3 files changed, 132 insertions(+), 39 deletions(-) diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go b/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go index 7270bfcd8..2cbf41feb 100644 --- a/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_controller.go @@ -444,7 +444,7 @@ func (r *RoleBasedGroupScalingAdapterReconciler) SetupWithManager(mgr ctrl.Manag Watches( &workloadsv1alpha2.RoleBasedGroup{}, handler.EnqueueRequestsFromMapFunc(r.mapRBGToScalingAdapters), - builder.WithPredicates(RBGRoleStatusPredicate()), + builder.WithPredicates(RBGRoleSpecOrStatusPredicate()), ). Named("workloads-rolebasedgroup-scalingadapter"). Complete(r) @@ -526,7 +526,12 @@ func RBGScalingAdapterPredicate() predicate.Funcs { } } -func RBGRoleStatusPredicate() predicate.Funcs { +// RBGRoleSpecOrStatusPredicate enqueues an RBG-related event when either +// status.roleStatuses changes (so we react to workload-level scaling +// completing or readyReplicas updating) or any role's spec.replicas +// changes (so we can correct an external mutator — e.g. a `helm upgrade` +// that drops the field — without waiting for status to change). +func RBGRoleSpecOrStatusPredicate() predicate.Funcs { return predicate.Funcs{ CreateFunc: func(e event.CreateEvent) bool { return false @@ -534,10 +539,13 @@ func RBGRoleStatusPredicate() predicate.Funcs { UpdateFunc: func(e event.UpdateEvent) bool { oldRBG, ok1 := e.ObjectOld.(*workloadsv1alpha2.RoleBasedGroup) newRBG, ok2 := e.ObjectNew.(*workloadsv1alpha2.RoleBasedGroup) - if ok1 && ok2 { - return !reflect.DeepEqual(oldRBG.Status.RoleStatuses, newRBG.Status.RoleStatuses) + if !ok1 || !ok2 { + return false } - return false + if !reflect.DeepEqual(oldRBG.Status.RoleStatuses, newRBG.Status.RoleStatuses) { + return true + } + return rolesReplicasDiffer(oldRBG.Spec.Roles, newRBG.Spec.Roles) }, DeleteFunc: func(e event.DeleteEvent) bool { return false @@ -548,6 +556,36 @@ func RBGRoleStatusPredicate() predicate.Funcs { } } +// rolesReplicasDiffer reports whether any role (matched by name) has a +// different spec.replicas value between old and new. Roles only present +// on one side count as a difference, so adding/removing/renaming a role +// also fires the watch. +func rolesReplicasDiffer(oldRoles, newRoles []workloadsv1alpha2.RoleSpec) bool { + if len(oldRoles) != len(newRoles) { + return true + } + oldByName := make(map[string]*int32, len(oldRoles)) + for i := range oldRoles { + oldByName[oldRoles[i].Name] = oldRoles[i].Replicas + } + for i := range newRoles { + oldRep, ok := oldByName[newRoles[i].Name] + if !ok { + return true + } + newRep := newRoles[i].Replicas + switch { + case oldRep == nil && newRep == nil: + continue + case oldRep == nil || newRep == nil: + return true + case *oldRep != *newRep: + return true + } + } + return false +} + func (r *RoleBasedGroupScalingAdapterReconciler) GetTargetRbgFromAdapter( ctx context.Context, rbgScalingAdapter *workloadsv1alpha2.RoleBasedGroupScalingAdapter, ) (*workloadsv1alpha2.RoleBasedGroup, error) { diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_controller_test.go b/internal/controller/workloads/rolebasedgroupscalingadapter_controller_test.go index ebd34b07b..bc978dce1 100644 --- a/internal/controller/workloads/rolebasedgroupscalingadapter_controller_test.go +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_controller_test.go @@ -878,8 +878,19 @@ func TestMapRBGToScalingAdapters(t *testing.T) { } // Gaps 9-11: RBG watch predicate -func TestRBGRoleStatusPredicate(t *testing.T) { - pred := RBGRoleStatusPredicate() +func TestRBGRoleSpecOrStatusPredicate(t *testing.T) { + pred := RBGRoleSpecOrStatusPredicate() + + rbgWithStatus := func(roleStatuses []workloadsv1alpha2.RoleStatus) *workloadsv1alpha2.RoleBasedGroup { + return &workloadsv1alpha2.RoleBasedGroup{ + Status: workloadsv1alpha2.RoleBasedGroupStatus{RoleStatuses: roleStatuses}, + } + } + rbgWithRoles := func(roles ...workloadsv1alpha2.RoleSpec) *workloadsv1alpha2.RoleBasedGroup { + return &workloadsv1alpha2.RoleBasedGroup{ + Spec: workloadsv1alpha2.RoleBasedGroupSpec{Roles: roles}, + } + } t.Run("UpdateFunc", func(t *testing.T) { tests := []struct { @@ -888,53 +899,97 @@ func TestRBGRoleStatusPredicate(t *testing.T) { expected bool }{ { - name: "returns true when RoleStatuses change", + name: "RoleStatuses change fires", event: event.UpdateEvent{ - ObjectOld: &workloadsv1alpha2.RoleBasedGroup{ - Status: workloadsv1alpha2.RoleBasedGroupStatus{ - RoleStatuses: []workloadsv1alpha2.RoleStatus{ - {Name: "role-a", ReadyReplicas: 2}, - }, - }, - }, - ObjectNew: &workloadsv1alpha2.RoleBasedGroup{ - Status: workloadsv1alpha2.RoleBasedGroupStatus{ - RoleStatuses: []workloadsv1alpha2.RoleStatus{ - {Name: "role-a", ReadyReplicas: 3}, - }, - }, - }, + ObjectOld: rbgWithStatus([]workloadsv1alpha2.RoleStatus{{Name: "a", ReadyReplicas: 2}}), + ObjectNew: rbgWithStatus([]workloadsv1alpha2.RoleStatus{{Name: "a", ReadyReplicas: 3}}), }, expected: true, }, { - name: "returns false when RoleStatuses unchanged", + name: "no change does not fire", event: event.UpdateEvent{ - ObjectOld: &workloadsv1alpha2.RoleBasedGroup{ - Status: workloadsv1alpha2.RoleBasedGroupStatus{ - RoleStatuses: []workloadsv1alpha2.RoleStatus{ - {Name: "role-a", ReadyReplicas: 2}, - }, - }, - }, - ObjectNew: &workloadsv1alpha2.RoleBasedGroup{ - Status: workloadsv1alpha2.RoleBasedGroupStatus{ - RoleStatuses: []workloadsv1alpha2.RoleStatus{ - {Name: "role-a", ReadyReplicas: 2}, - }, - }, - }, + ObjectOld: rbgWithStatus([]workloadsv1alpha2.RoleStatus{{Name: "a", ReadyReplicas: 2}}), + ObjectNew: rbgWithStatus([]workloadsv1alpha2.RoleStatus{{Name: "a", ReadyReplicas: 2}}), }, expected: false, }, { - name: "returns false for non-RBG types", + name: "non-RBG types do not fire", event: event.UpdateEvent{ ObjectOld: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "old"}}, ObjectNew: &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "new"}}, }, expected: false, }, + { + name: "spec.replicas change fires (helm-style stomp)", + event: event.UpdateEvent{ + ObjectOld: rbgWithRoles(workloadsv1alpha2.RoleSpec{Name: "a", Replicas: ptr.To[int32](10)}), + ObjectNew: rbgWithRoles(workloadsv1alpha2.RoleSpec{Name: "a", Replicas: ptr.To[int32](1)}), + }, + expected: true, + }, + { + name: "spec.replicas nil to set fires", + event: event.UpdateEvent{ + ObjectOld: rbgWithRoles(workloadsv1alpha2.RoleSpec{Name: "a"}), + ObjectNew: rbgWithRoles(workloadsv1alpha2.RoleSpec{Name: "a", Replicas: ptr.To[int32](1)}), + }, + expected: true, + }, + { + name: "spec.replicas set to nil fires", + event: event.UpdateEvent{ + ObjectOld: rbgWithRoles(workloadsv1alpha2.RoleSpec{Name: "a", Replicas: ptr.To[int32](5)}), + ObjectNew: rbgWithRoles(workloadsv1alpha2.RoleSpec{Name: "a"}), + }, + expected: true, + }, + { + name: "spec.replicas equal, other fields differ does not fire", + event: event.UpdateEvent{ + ObjectOld: rbgWithRoles(workloadsv1alpha2.RoleSpec{ + Name: "a", Replicas: ptr.To[int32](3), + Annotations: map[string]string{"k": "v1"}, + }), + ObjectNew: rbgWithRoles(workloadsv1alpha2.RoleSpec{ + Name: "a", Replicas: ptr.To[int32](3), + Annotations: map[string]string{"k": "v2"}, + }), + }, + expected: false, + }, + { + name: "added role fires", + event: event.UpdateEvent{ + ObjectOld: rbgWithRoles(workloadsv1alpha2.RoleSpec{Name: "a", Replicas: ptr.To[int32](1)}), + ObjectNew: rbgWithRoles( + workloadsv1alpha2.RoleSpec{Name: "a", Replicas: ptr.To[int32](1)}, + workloadsv1alpha2.RoleSpec{Name: "b", Replicas: ptr.To[int32](2)}, + ), + }, + expected: true, + }, + { + name: "removed role fires", + event: event.UpdateEvent{ + ObjectOld: rbgWithRoles( + workloadsv1alpha2.RoleSpec{Name: "a", Replicas: ptr.To[int32](1)}, + workloadsv1alpha2.RoleSpec{Name: "b", Replicas: ptr.To[int32](2)}, + ), + ObjectNew: rbgWithRoles(workloadsv1alpha2.RoleSpec{Name: "a", Replicas: ptr.To[int32](1)}), + }, + expected: true, + }, + { + name: "renamed role fires (different role names with same count)", + event: event.UpdateEvent{ + ObjectOld: rbgWithRoles(workloadsv1alpha2.RoleSpec{Name: "old", Replicas: ptr.To[int32](1)}), + ObjectNew: rbgWithRoles(workloadsv1alpha2.RoleSpec{Name: "new", Replicas: ptr.To[int32](1)}), + }, + expected: true, + }, } for _, tt := range tests { diff --git a/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go b/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go index 631231a9c..3a61dfa04 100644 --- a/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go +++ b/internal/controller/workloads/rolebasedgroupscalingadapter_rollout_test.go @@ -667,7 +667,7 @@ func TestRBGStatusChangedPredicate(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - pred := RBGRoleStatusPredicate() + pred := RBGRoleSpecOrStatusPredicate() oldRBG := &workloadsv1alpha2.RoleBasedGroup{ ObjectMeta: metav1.ObjectMeta{Name: "test-rbg", Namespace: "default"},