diff --git a/docs/source/designs/aibrix-stormservice.rst b/docs/source/designs/aibrix-stormservice.rst index 466417a83..5f4cc783f 100644 --- a/docs/source/designs/aibrix-stormservice.rst +++ b/docs/source/designs/aibrix-stormservice.rst @@ -40,6 +40,8 @@ Stormservice supports two deployment modes: **Replica Mode** and **Pooled Mode** 1. These two modes are mutually exclusive. The mode is declared through the `stormservice.spec.mode` field, which accepts `Replica` or `Pooled`. 2. `spec.mode` is optional and is not defaulted. When it is omitted the mode is inferred for backward compatibility from `stormservice.spec.replicas`: replica mode when `replicas > 1`, otherwise pooled mode. 3. When `spec.mode` is set to `Pooled`, `spec.replicas` must stay at `1`; roles are scaled through `spec.template.spec.roles[].replicas`. + 4. A declared `spec.mode` drives the update path: `Replica` uses the rolling update path and `Pooled` uses the in-place update path, even when `spec.updateStrategy.type` holds the (possibly CRD-defaulted) `RollingUpdate` value. Declaring `mode: Replica` together with `updateStrategy.type: InPlaceUpdate` is rejected by the webhook. When `spec.mode` is omitted, `spec.updateStrategy.type` keeps selecting the update path as before. + 5. A declared `spec.mode` is also the source of truth for PodAutoscaler role-level scaling. The `autoscaling.aibrix.ai/storm-service-mode` annotation is deprecated and only honored when the target StormService does not declare `spec.mode`. Replica Mode diff --git a/docs/source/features/autoscaling/metric-based-autoscaling.rst b/docs/source/features/autoscaling/metric-based-autoscaling.rst index 18f11e0b9..427393261 100644 --- a/docs/source/features/autoscaling/metric-based-autoscaling.rst +++ b/docs/source/features/autoscaling/metric-based-autoscaling.rst @@ -243,9 +243,9 @@ Example: StormService Role-Level Autoscaling ------------------------------------ -For StormService in pooled mode (``replicas=1``), different roles (e.g., prefill and decode) can be autoscaled independently. This enables fine-grained control where each role scales based on its specific metrics. +For StormService in pooled mode (``spec.mode: Pooled``), different roles (e.g., prefill and decode) can be autoscaled independently. This enables fine-grained control where each role scales based on its specific metrics. -Use the ``subTargetSelector`` field to target a specific role within a StormService. Additionally, add the annotation `autoscaling.aibrix.ai/storm-service-mode: "pool"` to the PodAutoscaler object. This helps the AIBrix autoscaler better distinguish ``replicas=1`` scenarios. +Use the ``subTargetSelector`` field to target a specific role within a StormService, and declare ``spec.mode`` on the StormService (``Pooled`` to scale the targeted role, ``Replica`` to scale ``spec.replicas``). The autoscaler reads ``spec.mode`` to route role-level scaling; ``replicas=1`` alone cannot distinguish the two modes. The PodAutoscaler annotation ``autoscaling.aibrix.ai/storm-service-mode`` is deprecated and only honored as a compatibility fallback when the target StormService does not declare ``spec.mode``. **Key features:** diff --git a/pkg/controller/podautoscaler/workload_scale.go b/pkg/controller/podautoscaler/workload_scale.go index c022aaf88..c6b8fce7a 100644 --- a/pkg/controller/podautoscaler/workload_scale.go +++ b/pkg/controller/podautoscaler/workload_scale.go @@ -39,8 +39,34 @@ import ( "github.com/vllm-project/aibrix/pkg/controller/constants" ) +// AutoscalingStormServiceModeAnnotationKey is the PodAutoscaler annotation that +// historically selected how role-level scaling applies to a StormService target: +// "replica" scales StormService.spec.replicas, anything else scales the targeted +// role's replicas. +// +// Deprecated: declare StormService.spec.mode instead. The annotation is only +// honored as a compatibility fallback when the target StormService does not +// declare spec.mode, and may be removed once spec.mode is broadly adopted. const AutoscalingStormServiceModeAnnotationKey = "autoscaling.aibrix.ai/storm-service-mode" +// stormServiceScalingMode resolves the deployment mode used to route role-level +// scaling for a StormService target. A declared StormService.spec.mode is the +// source of truth. When spec.mode is unset, the deprecated +// autoscaling.aibrix.ai/storm-service-mode annotation on the PodAutoscaler is +// honored as a compatibility fallback ("replica" selects replica mode), and any +// other value keeps the legacy pooled default. spec.replicas is intentionally +// not used for inference here: replicas == 1 cannot distinguish a pooled +// StormService from a scaled-down replica-mode one. +func stormServiceScalingMode(pa *autoscalingv1alpha1.PodAutoscaler, ss *orchestrationv1alpha1.StormService) orchestrationv1alpha1.StormServiceMode { + if ss.Spec.Mode != "" { + return ss.Spec.Mode + } + if pa.Annotations[AutoscalingStormServiceModeAnnotationKey] == "replica" { + return orchestrationv1alpha1.StormServiceReplicaMode + } + return orchestrationv1alpha1.StormServicePooledMode +} + // WorkloadScale provides scaling operations for different workload types. // It provides the mechanism to get/set replica counts on workload resources, // while AutoScaler provides the intelligence to compute desired replica counts. @@ -152,9 +178,11 @@ func (s *workloadScale) getCurrentReplicasForRole(ctx context.Context, pa *autos return 0, err } - // replica mode, return the replicas directly - // we can not easily use `*ss.Spec.Replicas > 1` as condition since 1 could be pool or replica both case under autoscaling scenarios - if pa.Annotations[AutoscalingStormServiceModeAnnotationKey] == "replica" { + // Replica mode scales the whole StormService, so report spec.replicas directly. + if stormServiceScalingMode(pa, ss) == orchestrationv1alpha1.StormServiceReplicaMode { + if ss.Spec.Replicas == nil { + return 0, nil + } return *ss.Spec.Replicas, nil } @@ -246,9 +274,9 @@ func (s *workloadScale) setDesiredReplicasForRole(ctx context.Context, pa *autos return err } upd := cur.DeepCopy() - // TODO: tricky part. it's hard to know replica=1 is pooling or replica mode, we can use autoscaling to limit it. - // we can extract the method and fallback to ss object annotation to check as well. - if pa.Annotations[AutoscalingStormServiceModeAnnotationKey] == "replica" { + // Replica mode scales the whole StormService through spec.replicas; pooled mode + // scales the targeted role. See stormServiceScalingMode for the resolution order. + if stormServiceScalingMode(pa, cur) == orchestrationv1alpha1.StormServiceReplicaMode { upd.Spec.Replicas = ptr.To(replicas) return s.client.Patch(ctx, upd, client.MergeFrom(cur)) } diff --git a/pkg/controller/podautoscaler/workload_scale_test.go b/pkg/controller/podautoscaler/workload_scale_test.go index 71d79b496..1522d711e 100644 --- a/pkg/controller/podautoscaler/workload_scale_test.go +++ b/pkg/controller/podautoscaler/workload_scale_test.go @@ -30,10 +30,41 @@ import ( v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" ) +func TestStormServiceScalingMode(t *testing.T) { + tests := map[string]struct { + specMode orchestrationv1alpha1.StormServiceMode + annotation string + want orchestrationv1alpha1.StormServiceMode + }{ + "declared replica mode wins": {specMode: orchestrationv1alpha1.StormServiceReplicaMode, want: orchestrationv1alpha1.StormServiceReplicaMode}, + "declared pooled mode wins over annotation": {specMode: orchestrationv1alpha1.StormServicePooledMode, annotation: "replica", want: orchestrationv1alpha1.StormServicePooledMode}, + "declared replica mode wins over annotation": {specMode: orchestrationv1alpha1.StormServiceReplicaMode, annotation: "pool", want: orchestrationv1alpha1.StormServiceReplicaMode}, + "no mode falls back to replica annotation": {annotation: "replica", want: orchestrationv1alpha1.StormServiceReplicaMode}, + "no mode with pool annotation defaults to pool": {annotation: "pool", want: orchestrationv1alpha1.StormServicePooledMode}, + "no mode and no annotation defaults to pool": {want: orchestrationv1alpha1.StormServicePooledMode}, + "no mode with unknown annotation stays pooled": {annotation: "bogus", want: orchestrationv1alpha1.StormServicePooledMode}, + "declared pooled mode with no annotation pooled": {specMode: orchestrationv1alpha1.StormServicePooledMode, want: orchestrationv1alpha1.StormServicePooledMode}, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + pa := &autoscalingv1alpha1.PodAutoscaler{} + if tc.annotation != "" { + pa.Annotations = map[string]string{AutoscalingStormServiceModeAnnotationKey: tc.annotation} + } + ss := &orchestrationv1alpha1.StormService{ + Spec: orchestrationv1alpha1.StormServiceSpec{Mode: tc.specMode}, + } + assert.Equal(t, tc.want, stormServiceScalingMode(pa, ss)) + }) + } +} + func TestGetCurrentReplicasFromScale(t *testing.T) { expectedReplicas := int32(2) @@ -166,6 +197,74 @@ func TestGetCurrentReplicasFromScale(t *testing.T) { }, scale: scaleStormService, }, + { + // spec.mode replaces the annotation as the mode signal. + name: "storm_service_with_declared_replica_mode", + pa: &autoscalingv1alpha1.PodAutoscaler{ + ObjectMeta: v1.ObjectMeta{ + Namespace: "default", + }, + Spec: autoscalingv1alpha1.PodAutoscalerSpec{ + SubTargetSelector: &autoscalingv1alpha1.SubTargetSelector{}, + ScaleTargetRef: corev1.ObjectReference{ + Kind: "StormService", + Name: "test-storm", + }, + }, + }, + ss: &orchestrationv1alpha1.StormService{ + ObjectMeta: v1.ObjectMeta{ + Name: "test-storm", + Namespace: "default", + }, + Spec: orchestrationv1alpha1.StormServiceSpec{ + Mode: orchestrationv1alpha1.StormServiceReplicaMode, + Replicas: &expectedReplicas, + }, + }, + scale: scaleStormService, + }, + { + // A declared pooled mode wins over a stale "replica" annotation: + // the role status is reported, not spec.replicas. + name: "storm_service_declared_pooled_mode_wins_over_annotation", + pa: &autoscalingv1alpha1.PodAutoscaler{ + ObjectMeta: v1.ObjectMeta{ + Namespace: "default", + Annotations: map[string]string{ + AutoscalingStormServiceModeAnnotationKey: "replica", + }, + }, + Spec: autoscalingv1alpha1.PodAutoscalerSpec{ + SubTargetSelector: &autoscalingv1alpha1.SubTargetSelector{ + RoleName: "prefill", + }, + ScaleTargetRef: corev1.ObjectReference{ + Kind: "StormService", + Name: "test-storm", + }, + }, + }, + ss: &orchestrationv1alpha1.StormService{ + ObjectMeta: v1.ObjectMeta{ + Name: "test-storm", + Namespace: "default", + }, + Spec: orchestrationv1alpha1.StormServiceSpec{ + Mode: orchestrationv1alpha1.StormServicePooledMode, + Replicas: ptr.To(int32(5)), // must not be reported in pooled mode + }, + Status: orchestrationv1alpha1.StormServiceStatus{ + RoleStatuses: []orchestrationv1alpha1.RoleStatus{ + { + Name: "prefill", + Replicas: expectedReplicas, + }, + }, + }, + }, + scale: scaleStormService, + }, } for _, tt := range table { @@ -455,6 +554,92 @@ func TestSetDesiredReplicas(t *testing.T) { assert.Equal(t, expectedReplicas, *ss.Spec.Template.Spec.Roles[0].Replicas) }, }, + { + // spec.mode replaces the annotation as the mode signal: a declared + // replica mode updates spec.replicas without any annotation set. + name: "storm_service_with_declared_replica_mode", + pa: &autoscalingv1alpha1.PodAutoscaler{ + ObjectMeta: v1.ObjectMeta{ + Namespace: "default", + }, + Spec: autoscalingv1alpha1.PodAutoscalerSpec{ + SubTargetSelector: &autoscalingv1alpha1.SubTargetSelector{ + RoleName: "", + }, + ScaleTargetRef: corev1.ObjectReference{ + Kind: "StormService", + Name: "test-storm", + }, + }, + }, + deployment: &appsv1.Deployment{}, + ss: &orchestrationv1alpha1.StormService{ + ObjectMeta: v1.ObjectMeta{ + Name: "test-storm", + Namespace: "default", + }, + Spec: orchestrationv1alpha1.StormServiceSpec{ + Mode: orchestrationv1alpha1.StormServiceReplicaMode, + Replicas: ¤tReplicas, + }, + }, + assertReplicas: func(t *testing.T, fakeClient client.Client) { + ss := &orchestrationv1alpha1.StormService{} + err := fakeClient.Get(context.TODO(), client.ObjectKey{Namespace: "default", Name: "test-storm"}, ss) + assert.NoError(t, err) + assert.Equal(t, expectedReplicas, *ss.Spec.Replicas) + }, + }, + { + // A declared pooled mode wins over a stale "replica" annotation: the + // targeted role scales while spec.replicas stays untouched. + name: "storm_service_declared_pooled_mode_wins_over_annotation", + pa: &autoscalingv1alpha1.PodAutoscaler{ + ObjectMeta: v1.ObjectMeta{ + Namespace: "default", + Annotations: map[string]string{ + AutoscalingStormServiceModeAnnotationKey: "replica", + }, + }, + Spec: autoscalingv1alpha1.PodAutoscalerSpec{ + SubTargetSelector: &autoscalingv1alpha1.SubTargetSelector{ + RoleName: "prefill", + }, + ScaleTargetRef: corev1.ObjectReference{ + Kind: "StormService", + Name: "test-storm", + }, + }, + }, + deployment: &appsv1.Deployment{}, + ss: &orchestrationv1alpha1.StormService{ + ObjectMeta: v1.ObjectMeta{ + Name: "test-storm", + Namespace: "default", + }, + Spec: orchestrationv1alpha1.StormServiceSpec{ + Mode: orchestrationv1alpha1.StormServicePooledMode, + Replicas: ptr.To(int32(1)), + Template: orchestrationv1alpha1.RoleSetTemplateSpec{ + Spec: &orchestrationv1alpha1.RoleSetSpec{ + Roles: []orchestrationv1alpha1.RoleSpec{ + { + Name: "prefill", + Replicas: ¤tReplicas, + }, + }, + }, + }, + }, + }, + assertReplicas: func(t *testing.T, fakeClient client.Client) { + ss := &orchestrationv1alpha1.StormService{} + err := fakeClient.Get(context.TODO(), client.ObjectKey{Namespace: "default", Name: "test-storm"}, ss) + assert.NoError(t, err) + assert.Equal(t, expectedReplicas, *ss.Spec.Template.Spec.Roles[0].Replicas) + assert.Equal(t, int32(1), *ss.Spec.Replicas) + }, + }, } for _, tt := range table { diff --git a/pkg/controller/stormservice/sync.go b/pkg/controller/stormservice/sync.go index 731ed3daa..c0c51f7ae 100644 --- a/pkg/controller/stormservice/sync.go +++ b/pkg/controller/stormservice/sync.go @@ -272,17 +272,16 @@ func (r *StormServiceReconciler) rollout(ctx context.Context, stormService, curr if len(updated) == int(expectReplica) { return nil } - switch stormService.Spec.UpdateStrategy.Type { - case "": - // By default use RollingUpdate strategy - fallthrough - case orchestrationv1alpha1.RollingUpdateStormServiceStrategyType: - return r.rollingUpdate(allRoleSets, stormService, current, currentCR, updateCR) - case orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType: + // The update path follows the declared spec.mode when it is set and falls back to + // the legacy updateStrategy.type selection otherwise, see EffectiveUpdateStrategyType. + strategyType, err := EffectiveUpdateStrategyType(stormService) + if err != nil { + return err + } + if strategyType == orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType { return r.inPlaceUpdate(allRoleSets, stormService, current, currentCR, updateCR) - default: - return fmt.Errorf("unexpected stormService strategy type: %s", stormService.Spec.UpdateStrategy.Type) } + return r.rollingUpdate(allRoleSets, stormService, current, currentCR, updateCR) } // rollingUpdate: rolling update logic for replica mode diff --git a/pkg/controller/stormservice/sync_test.go b/pkg/controller/stormservice/sync_test.go index 1b0fef016..f04e87393 100644 --- a/pkg/controller/stormservice/sync_test.go +++ b/pkg/controller/stormservice/sync_test.go @@ -21,10 +21,13 @@ import ( "reflect" "testing" + apps "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + intstrutil "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/client-go/tools/record" + "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -279,3 +282,171 @@ func TestSyncHeadlessService(t *testing.T) { }) } } + +// newPooledStormServiceWithSurge returns a pooled StormService shaped the way the CRD +// persists it: mode: Pooled with updateStrategy.type defaulted to RollingUpdate and an +// explicit maxSurge. Before IsRollingUpdate was routed through EffectiveUpdateStrategyType, +// this combination handed scaling() a non-zero surge budget. +func newPooledStormServiceWithSurge(maxSurge int32) *orchestrationv1alpha1.StormService { + surge := intstrutil.FromInt32(maxSurge) + return &orchestrationv1alpha1.StormService{ + ObjectMeta: metav1.ObjectMeta{ + Name: "pooled-storm", + Namespace: "default", + UID: "pooled-storm-uid", + }, + Spec: orchestrationv1alpha1.StormServiceSpec{ + Replicas: ptr.To(int32(1)), + Mode: orchestrationv1alpha1.StormServicePooledMode, + UpdateStrategy: orchestrationv1alpha1.StormServiceUpdateStrategy{ + // The CRD defaults type to RollingUpdate whenever the updateStrategy + // block is present, so this is what a pooled object with maxSurge set + // actually looks like in the API server. + Type: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + MaxSurge: &surge, + }, + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"app": "pooled-storm"}, + }, + Template: orchestrationv1alpha1.RoleSetTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{"app": "pooled-storm"}, + }, + Spec: &orchestrationv1alpha1.RoleSetSpec{ + Roles: []orchestrationv1alpha1.RoleSpec{ + { + Name: "engine", + Replicas: ptr.To(int32(1)), + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + {Name: "main", Image: "engine:v1"}, + }, + }, + }, + }, + }, + }, + }, + }, + } +} + +// pooledRoleSet returns a RoleSet owned by newPooledStormServiceWithSurge's object at the +// given revision. Terminating RoleSets carry a DeletionTimestamp plus a finalizer so the +// fake client keeps them around, mirroring a RoleSet that is still tearing down. +func pooledRoleSet(name, revision string, terminating bool) *orchestrationv1alpha1.RoleSet { + rs := &orchestrationv1alpha1.RoleSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + Labels: map[string]string{ + "app": "pooled-storm", + constants.StormServiceNameLabelKey: "pooled-storm", + constants.StormServiceRevisionLabelKey: revision, + }, + }, + } + if terminating { + now := metav1.Now() + rs.DeletionTimestamp = &now + rs.Finalizers = []string{"orchestration.aibrix.ai/test-teardown"} + } + return rs +} + +// TestScalingPooledModeNeverCreatesSecondRoleSet covers the review scenario for +// mode: Pooled + CRD-defaulted RollingUpdate + maxSurge: 2: the surge budget must +// evaluate to 0 so scaling() never brings a second RoleSet into existence next to +// the one RoleSet a pooled StormService owns. +func TestScalingPooledModeNeverCreatesSecondRoleSet(t *testing.T) { + scheme := runtime.NewScheme() + _ = corev1.AddToScheme(scheme) + _ = orchestrationv1alpha1.AddToScheme(scheme) + + const revision = "pooled-storm-rev1" + + tests := []struct { + name string + existingRoleSets []*orchestrationv1alpha1.RoleSet + wantScaling bool + wantRoleSets int + }{ + { + name: "steady state keeps the single roleset", + existingRoleSets: []*orchestrationv1alpha1.RoleSet{pooledRoleSet("pooled-storm-roleset-a", revision, false)}, + wantScaling: false, + wantRoleSets: 1, + }, + { + name: "scale out from zero creates exactly one roleset", + existingRoleSets: nil, + wantScaling: true, + wantRoleSets: 1, + }, + { + // The regression case: with the surge budget read from the raw + // updateStrategy.type, scaling() created a replacement RoleSet while the + // old one was still terminating, so two RoleSets existed at once. + name: "no replacement is surged while the old roleset terminates", + existingRoleSets: []*orchestrationv1alpha1.RoleSet{pooledRoleSet("pooled-storm-roleset-a", revision, true)}, + wantScaling: false, + wantRoleSets: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var objs []client.Object + for _, rs := range tt.existingRoleSets { + objs = append(objs, rs) + } + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objs...). + Build() + + r := &StormServiceReconciler{ + Client: fakeClient, + EventRecorder: &record.FakeRecorder{}, + } + + stormService := newPooledStormServiceWithSurge(2) + cr := &apps.ControllerRevision{ + ObjectMeta: metav1.ObjectMeta{Name: revision, Namespace: "default"}, + Revision: 1, + } + + scaling, err := r.scaling(context.TODO(), stormService, stormService, cr, cr) + if err != nil { + t.Fatalf("scaling() error = %v", err) + } + if scaling != tt.wantScaling { + t.Errorf("scaling() = %v, want %v", scaling, tt.wantScaling) + } + + roleSetList := &orchestrationv1alpha1.RoleSetList{} + if err := fakeClient.List(context.TODO(), roleSetList); err != nil { + t.Fatalf("failed to list roleSets: %v", err) + } + if len(roleSetList.Items) != tt.wantRoleSets { + names := make([]string, 0, len(roleSetList.Items)) + for _, rs := range roleSetList.Items { + names = append(names, rs.Name) + } + t.Fatalf("expected %d roleSet(s), got %d: %v", tt.wantRoleSets, len(roleSetList.Items), names) + } + // A pooled StormService must never own more than one RoleSet, no matter + // how often scaling() runs; re-run to make sure the budget stays zero. + if _, err := r.scaling(context.TODO(), stormService, stormService, cr, cr); err != nil { + t.Fatalf("second scaling() error = %v", err) + } + if err := fakeClient.List(context.TODO(), roleSetList); err != nil { + t.Fatalf("failed to list roleSets: %v", err) + } + if len(roleSetList.Items) > 1 { + t.Fatalf("pooled stormservice ended up with %d roleSets, a second RoleSet must never be created", len(roleSetList.Items)) + } + }) + } +} diff --git a/pkg/controller/stormservice/utils.go b/pkg/controller/stormservice/utils.go index 72f339750..a90849fb3 100644 --- a/pkg/controller/stormservice/utils.go +++ b/pkg/controller/stormservice/utils.go @@ -17,6 +17,7 @@ limitations under the License. package stormservice import ( + "fmt" "sort" ctrlutil "github.com/vllm-project/aibrix/pkg/controller/util" @@ -88,9 +89,72 @@ func MaxSurge(stormService *orchestrationv1alpha1.StormService) int32 { return maxSurge } -// IsRollingUpdate returns true if the strategy type is a rolling update. +// IsRollingUpdate returns true if the effective strategy type is a rolling update. +// +// It intentionally resolves the strategy through EffectiveUpdateStrategyType instead of +// reading spec.updateStrategy.type directly, so a declared spec.mode wins over the +// (possibly CRD-defaulted) strategy type. In Pooled mode this returns false, which makes +// MaxSurge, MaxUnavailable and MinAvailable evaluate to 0 and keeps scaling() from ever +// creating RoleSets beyond spec.replicas for a pooled StormService. func IsRollingUpdate(stormService *orchestrationv1alpha1.StormService) bool { - return stormService.Spec.UpdateStrategy.Type == "" || stormService.Spec.UpdateStrategy.Type == orchestrationv1alpha1.RollingUpdateStormServiceStrategyType + effectiveType, err := EffectiveUpdateStrategyType(stormService) + if err != nil { + // An unresolvable mode/strategy combination must not surge; rollout() surfaces the error. + return false + } + return effectiveType == orchestrationv1alpha1.RollingUpdateStormServiceStrategyType +} + +// EffectiveUpdateStrategyType returns the update strategy type that drives the rollout path. +// +// A declared spec.mode is the source of truth: Replica mode replaces RoleSets through the +// RollingUpdate path and Pooled mode updates its single RoleSet through the InPlaceUpdate +// path. The webhook rejects a declared InPlaceUpdate strategy combined with Replica mode. +// The reverse conflict (Pooled with RollingUpdate) cannot be rejected at admission because +// the CRD defaults spec.updateStrategy.type to RollingUpdate whenever the updateStrategy +// block is present, so a RollingUpdate value is indistinguishable from the default; the +// controller resolves that conflict in favor of the declared mode and logs the override +// at verbose level only, because it recurs on every reconcile. +// +// When spec.mode is unset, the legacy updateStrategy.type selection is kept so existing +// manifests behave as before. ResolvedMode is intentionally not consulted for the inferred +// case: its replicas-based inference does not always match the declared strategy type +// (for example replicas: 1 with RollingUpdate infers Pooled but must keep rolling). +func EffectiveUpdateStrategyType(stormService *orchestrationv1alpha1.StormService) (orchestrationv1alpha1.StormServiceUpdateStrategyType, error) { + declaredType := stormService.Spec.UpdateStrategy.Type + if stormService.Spec.Mode != "" { + switch mode := stormService.Spec.ResolvedMode(); mode { + case orchestrationv1alpha1.StormServiceReplicaMode: + if declaredType == orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType { + // The webhook rejects Replica mode with a declared InPlaceUpdate, so this + // branch is only reachable for objects that bypassed admission; keep a + // verbose-only trace for those. + klog.V(4).Infof("stormservice %s/%s declares mode %s, overriding updateStrategy.type %s with %s", + stormService.Namespace, stormService.Name, mode, declaredType, orchestrationv1alpha1.RollingUpdateStormServiceStrategyType) + } + return orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, nil + case orchestrationv1alpha1.StormServicePooledMode: + if declaredType == orchestrationv1alpha1.RollingUpdateStormServiceStrategyType { + // RollingUpdate is the CRD default whenever the updateStrategy block is + // present, so this override is steady state for pooled objects and fires + // on every reconcile; log at verbose level to avoid flooding operator logs. + klog.V(4).Infof("stormservice %s/%s declares mode %s, overriding updateStrategy.type %s with %s", + stormService.Namespace, stormService.Name, mode, declaredType, orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType) + } + return orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, nil + default: + return "", fmt.Errorf("unexpected stormService mode: %s", mode) + } + } + switch declaredType { + case "": + // By default use RollingUpdate strategy + return orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, nil + case orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType: + return declaredType, nil + default: + return "", fmt.Errorf("unexpected stormService strategy type: %s", declaredType) + } } // ResolveFenceposts resolves both maxSurge and maxUnavailable. This needs to happen in one diff --git a/pkg/controller/stormservice/utils_test.go b/pkg/controller/stormservice/utils_test.go index 0ecd2c467..bd7c258e7 100644 --- a/pkg/controller/stormservice/utils_test.go +++ b/pkg/controller/stormservice/utils_test.go @@ -27,6 +27,7 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" intstrutil "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/utils/ptr" ) func newStormService(replicas int32, surge, unavail string) orchestrationv1alpha1.StormService { @@ -115,6 +116,132 @@ func TestIsRollingUpdate(t *testing.T) { ss.Spec.UpdateStrategy.Type = orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType assert.False(t, IsRollingUpdate(&ss)) + + // A declared mode wins over the (possibly CRD-defaulted) strategy type. + ss.Spec.UpdateStrategy.Type = orchestrationv1alpha1.RollingUpdateStormServiceStrategyType + ss.Spec.Mode = orchestrationv1alpha1.StormServicePooledMode + assert.False(t, IsRollingUpdate(&ss), "pooled mode must not roll even with a defaulted RollingUpdate type") + + ss.Spec.UpdateStrategy.Type = orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType + ss.Spec.Mode = orchestrationv1alpha1.StormServiceReplicaMode + assert.True(t, IsRollingUpdate(&ss), "replica mode rolls even when a bypassed-webhook object declares InPlaceUpdate") + + // An unresolvable mode must not surge. + ss.Spec.Mode = orchestrationv1alpha1.StormServiceMode("Bogus") + assert.False(t, IsRollingUpdate(&ss)) +} + +// TestPooledModeZeroesSurgeAndUnavailable guards the pooled single-RoleSet contract: +// with mode: Pooled and the CRD-defaulted RollingUpdate type, surge and unavailable +// budgets consumed by scaling() must all evaluate to 0 regardless of maxSurge. +func TestPooledModeZeroesSurgeAndUnavailable(t *testing.T) { + ss := newStormService(10, "25%", "20%") // surge=3, unavail=2 when rolling + ss.Spec.Mode = orchestrationv1alpha1.StormServicePooledMode + assert.Equal(t, int32(0), MaxSurge(&ss)) + assert.Equal(t, int32(0), MaxUnavailable(ss)) + assert.Equal(t, int32(0), MinAvailable(&ss)) + + // maxSurge as a plain integer, the shape the reviewer's scenario uses. + surge := intstrutil.FromInt32(2) + pooled := orchestrationv1alpha1.StormService{ + Spec: orchestrationv1alpha1.StormServiceSpec{ + Replicas: ptr.To(int32(1)), + Mode: orchestrationv1alpha1.StormServicePooledMode, + UpdateStrategy: orchestrationv1alpha1.StormServiceUpdateStrategy{ + Type: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + MaxSurge: &surge, + }, + }, + } + assert.Equal(t, int32(0), MaxSurge(&pooled)) + assert.Equal(t, int32(0), MaxUnavailable(pooled)) + assert.Equal(t, int32(0), MinAvailable(&pooled)) +} + +func TestEffectiveUpdateStrategyType(t *testing.T) { + tests := map[string]struct { + mode orchestrationv1alpha1.StormServiceMode + strategyType orchestrationv1alpha1.StormServiceUpdateStrategyType + replicas *int32 + want orchestrationv1alpha1.StormServiceUpdateStrategyType + wantErr bool + }{ + // Undeclared mode keeps the legacy strategy-driven selection, regardless + // of what ResolvedMode would infer from spec.replicas. + "no mode, empty type defaults to rolling": { + want: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + }, + "no mode, rolling type stays rolling even when pooled is inferred": { + strategyType: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + replicas: ptr.To(int32(1)), + want: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + }, + "no mode, in-place type stays in-place even when replica is inferred": { + strategyType: orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, + replicas: ptr.To(int32(3)), + want: orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, + }, + "no mode, unknown type errors": { + strategyType: orchestrationv1alpha1.StormServiceUpdateStrategyType("Bogus"), + wantErr: true, + }, + // A declared mode is the source of truth for the update path. + "replica mode with empty type rolls": { + mode: orchestrationv1alpha1.StormServiceReplicaMode, + want: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + }, + "replica mode with rolling type rolls": { + mode: orchestrationv1alpha1.StormServiceReplicaMode, + strategyType: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + want: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + }, + "replica mode overrides in-place type": { + // Rejected at admission; kept deterministic for objects that bypassed the webhook. + mode: orchestrationv1alpha1.StormServiceReplicaMode, + strategyType: orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, + want: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + }, + "pooled mode with empty type updates in place": { + mode: orchestrationv1alpha1.StormServicePooledMode, + want: orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, + }, + "pooled mode overrides rolling type": { + // RollingUpdate may come from CRD defaulting, so the declared mode wins. + mode: orchestrationv1alpha1.StormServicePooledMode, + strategyType: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + want: orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, + }, + "pooled mode with in-place type updates in place": { + mode: orchestrationv1alpha1.StormServicePooledMode, + strategyType: orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, + want: orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, + }, + "unknown mode errors": { + mode: orchestrationv1alpha1.StormServiceMode("Bogus"), + wantErr: true, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + ss := &orchestrationv1alpha1.StormService{ + Spec: orchestrationv1alpha1.StormServiceSpec{ + Mode: tc.mode, + Replicas: tc.replicas, + UpdateStrategy: orchestrationv1alpha1.StormServiceUpdateStrategy{ + Type: tc.strategyType, + }, + }, + } + got, err := EffectiveUpdateStrategyType(ss) + if tc.wantErr { + assert.Error(t, err) + return + } + assert.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } } func TestSortRoleSetByRevision(t *testing.T) { diff --git a/pkg/webhook/stormservice_webhook.go b/pkg/webhook/stormservice_webhook.go index 829ae7a85..023632a66 100644 --- a/pkg/webhook/stormservice_webhook.go +++ b/pkg/webhook/stormservice_webhook.go @@ -220,15 +220,31 @@ func (r *StormServiceCustomDefaulter) ValidateUpdate(_ context.Context, _, newOb return nil, nil } -// validateStormServiceMode rejects mode/replicas combinations that cannot be satisfied. +// validateStormServiceMode rejects declared mode combinations that cannot be satisfied. +// // Pooled mode runs a single RoleSet and scales roles through spec.template.spec.roles[], -// so a replica count above one is ambiguous. Only an explicitly declared spec.mode is -// checked, so objects that rely on the inferred mode keep scaling spec.replicas freely. +// so a replica count above one is ambiguous. Replica mode derives its update path from +// the mode (rolling replacement of RoleSets), so combining it with a declared +// spec.updateStrategy.type of InPlaceUpdate is contradictory; InPlaceUpdate can only be +// user-written because the CRD defaults the type to RollingUpdate. The reverse +// combination (Pooled with RollingUpdate) is not rejected: a RollingUpdate value is +// indistinguishable from the CRD default, so the controller resolves it in favor of the +// declared mode instead (see EffectiveUpdateStrategyType in the stormservice controller). +// +// Only an explicitly declared spec.mode is checked, so objects that rely on the inferred +// mode keep scaling spec.replicas and choosing updateStrategy.type freely. func validateStormServiceMode(stormService *orchestrationv1alpha1.StormService) error { - if stormService.Spec.Mode == orchestrationv1alpha1.StormServicePooledMode && - stormService.Spec.Replicas != nil && *stormService.Spec.Replicas > 1 { - return fmt.Errorf("StormService in %s mode must not set spec.replicas > 1 (got %d); scale roles through spec.template.spec.roles[].replicas instead", - orchestrationv1alpha1.StormServicePooledMode, *stormService.Spec.Replicas) + switch stormService.Spec.Mode { + case orchestrationv1alpha1.StormServicePooledMode: + if stormService.Spec.Replicas != nil && *stormService.Spec.Replicas > 1 { + return fmt.Errorf("StormService in %s mode must not set spec.replicas > 1 (got %d); scale roles through spec.template.spec.roles[].replicas instead", + orchestrationv1alpha1.StormServicePooledMode, *stormService.Spec.Replicas) + } + case orchestrationv1alpha1.StormServiceReplicaMode: + if stormService.Spec.UpdateStrategy.Type == orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType { + return fmt.Errorf("StormService in %s mode must not set spec.updateStrategy.type %s; use %s, or leave spec.mode unset to keep in-place updates for inferred replica mode", + orchestrationv1alpha1.StormServiceReplicaMode, orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, orchestrationv1alpha1.RollingUpdateStormServiceStrategyType) + } } return nil } diff --git a/pkg/webhook/stormservice_webhook_test.go b/pkg/webhook/stormservice_webhook_test.go index b7dae0fcf..1f20e4832 100644 --- a/pkg/webhook/stormservice_webhook_test.go +++ b/pkg/webhook/stormservice_webhook_test.go @@ -77,6 +77,68 @@ func TestStormServiceValidateCreate_ModeReplicas(t *testing.T) { } } +func TestStormServiceValidateCreate_ModeUpdateStrategy(t *testing.T) { + validator := &StormServiceCustomDefaulter{} + + tests := map[string]struct { + mode orchestrationv1alpha1.StormServiceMode + strategyType orchestrationv1alpha1.StormServiceUpdateStrategyType + expectError bool + }{ + // InPlaceUpdate can only be user-written (the CRD defaults the type to + // RollingUpdate), so it is a provable contradiction with Replica mode. + "replica with in-place strategy is rejected": {mode: orchestrationv1alpha1.StormServiceReplicaMode, strategyType: orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, expectError: true}, + "replica with rolling strategy is allowed": {mode: orchestrationv1alpha1.StormServiceReplicaMode, strategyType: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, expectError: false}, + "replica with empty strategy is allowed": {mode: orchestrationv1alpha1.StormServiceReplicaMode, strategyType: "", expectError: false}, + // A RollingUpdate type may come from CRD defaulting rather than the user, + // so pooled mode does not reject it; the controller resolves it to in-place. + "pooled with rolling strategy is allowed": {mode: orchestrationv1alpha1.StormServicePooledMode, strategyType: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, expectError: false}, + "pooled with in-place strategy is allowed": {mode: orchestrationv1alpha1.StormServicePooledMode, strategyType: orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, expectError: false}, + // Undeclared mode keeps choosing the strategy freely. + "no mode with in-place strategy is allowed": {mode: "", strategyType: orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType, expectError: false}, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + ss := &orchestrationv1alpha1.StormService{ + Spec: orchestrationv1alpha1.StormServiceSpec{ + Mode: tc.mode, + UpdateStrategy: orchestrationv1alpha1.StormServiceUpdateStrategy{ + Type: tc.strategyType, + }, + Template: orchestrationv1alpha1.RoleSetTemplateSpec{ + Spec: &orchestrationv1alpha1.RoleSetSpec{}, + }, + }, + } + _, err := validator.ValidateCreate(context.Background(), ss) + if tc.expectError { + require.Error(t, err) + } else { + require.NoError(t, err) + } + }) + } +} + +func TestStormServiceValidateUpdate_ModeUpdateStrategy(t *testing.T) { + validator := &StormServiceCustomDefaulter{} + + oldSS := &orchestrationv1alpha1.StormService{ + Spec: orchestrationv1alpha1.StormServiceSpec{ + Mode: orchestrationv1alpha1.StormServiceReplicaMode, + UpdateStrategy: orchestrationv1alpha1.StormServiceUpdateStrategy{ + Type: orchestrationv1alpha1.RollingUpdateStormServiceStrategyType, + }, + }, + } + newSS := oldSS.DeepCopy() + newSS.Spec.UpdateStrategy.Type = orchestrationv1alpha1.InPlaceUpdateStormServiceStrategyType + + _, err := validator.ValidateUpdate(context.Background(), oldSS, newSS) + require.Error(t, err) +} + func TestStormServiceValidateUpdate_ModeReplicas(t *testing.T) { validator := &StormServiceCustomDefaulter{} diff --git a/samples/autoscaling/stormservice-pool.yaml b/samples/autoscaling/stormservice-pool.yaml index f57fc5f31..6ac31a4e1 100644 --- a/samples/autoscaling/stormservice-pool.yaml +++ b/samples/autoscaling/stormservice-pool.yaml @@ -10,8 +10,6 @@ kind: PodAutoscaler metadata: name: ss-pool-prefill namespace: default - annotations: - autoscaling.aibrix.ai/storm-service-mode: "pool" spec: scaleTargetRef: apiVersion: orchestration.aibrix.ai/v1alpha1 @@ -41,8 +39,6 @@ kind: PodAutoscaler metadata: name: ss-pool-decode namespace: default - annotations: - autoscaling.aibrix.ai/storm-service-mode: "pool" spec: scaleTargetRef: apiVersion: orchestration.aibrix.ai/v1alpha1 diff --git a/samples/autoscaling/stormservice-replica.yaml b/samples/autoscaling/stormservice-replica.yaml index 80e0ccaf9..beae1854c 100644 --- a/samples/autoscaling/stormservice-replica.yaml +++ b/samples/autoscaling/stormservice-replica.yaml @@ -10,8 +10,6 @@ kind: PodAutoscaler metadata: name: ss-replica-hpa namespace: default - annotations: - autoscaling.aibrix.ai/storm-service-mode: "replica" spec: scaleTargetRef: apiVersion: orchestration.aibrix.ai/v1alpha1 diff --git a/test/integration/webhook/stormservice_webhook_test.go b/test/integration/webhook/stormservice_webhook_test.go index 56a2a2e0c..0e04abdce 100644 --- a/test/integration/webhook/stormservice_webhook_test.go +++ b/test/integration/webhook/stormservice_webhook_test.go @@ -264,6 +264,60 @@ var _ = ginkgo.Describe("stormservice default webhook", func() { }, failed: false, }), + + // Declared Replica mode drives the rolling update path, so a declared + // InPlaceUpdate strategy is a contradiction and is rejected. + ginkgo.Entry("rejects explicit Replica mode with InPlaceUpdate strategy", &testValidatingCase{ + stormservice: func() *orchestrationapi.StormService { + return wrapper.MakeStormService("replica-inplace-conflict"). + Namespace(ns.Name). + WithDefaultConfiguration(). + Mode(orchestrationapi.StormServiceReplicaMode). + Replicas(ptr.To(int32(3))). + UpdateStrategyType(orchestrationapi.InPlaceUpdateStormServiceStrategyType). + Obj() + }, + failed: true, + }), + + ginkgo.Entry("accepts explicit Replica mode with RollingUpdate strategy", &testValidatingCase{ + stormservice: func() *orchestrationapi.StormService { + return wrapper.MakeStormService("replica-rolling"). + Namespace(ns.Name). + WithDefaultConfiguration(). + Mode(orchestrationapi.StormServiceReplicaMode). + Replicas(ptr.To(int32(3))). + Obj() + }, + failed: false, + }), + + // A RollingUpdate type may come from CRD defaulting, so an explicit Pooled + // mode does not reject it; the controller resolves the path from the mode. + ginkgo.Entry("accepts explicit Pooled mode with defaulted RollingUpdate strategy", &testValidatingCase{ + stormservice: func() *orchestrationapi.StormService { + return wrapper.MakeStormService("pooled-defaulted-rolling"). + Namespace(ns.Name). + WithDefaultConfiguration(). + Mode(orchestrationapi.StormServicePooledMode). + Obj() + }, + failed: false, + }), + + // Undeclared mode keeps choosing InPlaceUpdate freely (legacy in-place + // updates for inferred replica mode stay valid). + ginkgo.Entry("accepts InPlaceUpdate with replicas > 1 when mode is not declared", &testValidatingCase{ + stormservice: func() *orchestrationapi.StormService { + return wrapper.MakeStormService("undeclared-mode-inplace"). + Namespace(ns.Name). + WithDefaultConfiguration(). + Replicas(ptr.To(int32(3))). + UpdateStrategyType(orchestrationapi.InPlaceUpdateStormServiceStrategyType). + Obj() + }, + failed: false, + }), ) ginkgo.It("rejects scaling an explicit Pooled StormService above one replica", func() {