From a2583ad934346a20496a2f763901070e9f533bc3 Mon Sep 17 00:00:00 2001 From: Scot Schuchert-Wells Date: Wed, 11 Nov 2026 15:01:36 +0000 Subject: [PATCH 1/2] fix: create instances for the Parallel policy A Workload with instanceManagementPolicy Parallel was accepted by the API but never got instances: the stateful control only emitted create, delete and replace actions for OrderedReady, so every other value silently did nothing. Key changes: - implement Parallel: create and delete every pending instance at once without waiting on readiness, and replace outdated instances one at a time once nothing is being created or is waiting - restrict instanceManagementPolicy to OrderedReady and Parallel at admission so unknown values are rejected instead of ignored - return an error for an unsupported policy rather than doing nothing - add unit tests for Parallel scale up, scale down and template rollout --- api/v1alpha/workload_types.go | 8 +- ...ute.datumapis.com_workloaddeployments.yaml | 8 ++ .../compute.datumapis.com_workloads.yaml | 8 ++ docs/api/workloaddeployments.md | 10 +- docs/api/workloads.md | 10 +- .../stateful/stateful_control.go | 19 ++- .../stateful/stateful_control_test.go | 115 ++++++++++++++++++ 7 files changed, 172 insertions(+), 6 deletions(-) diff --git a/api/v1alpha/workload_types.go b/api/v1alpha/workload_types.go index 5444e3c0..12cab489 100644 --- a/api/v1alpha/workload_types.go +++ b/api/v1alpha/workload_types.go @@ -229,7 +229,13 @@ type HorizontalScaleSettings struct { // Controls how instances are managed during scale up and down, as well as // during maintenance events. // + // OrderedReady creates and deletes instances one at a time, waiting for + // each to become ready before moving to the next. Parallel creates and + // deletes instances all at once without waiting on readiness. Both replace + // instances on a template change one at a time. + // // +kubebuilder:validation:Required + // +kubebuilder:validation:Enum=OrderedReady;Parallel // +kubebuilder:default=OrderedReady InstanceManagementPolicy InstanceManagementPolicyType `json:"instanceManagementPolicy,omitempty"` } @@ -238,7 +244,7 @@ type InstanceManagementPolicyType string const ( OrderedReadyInstanceManagementPolicyType InstanceManagementPolicyType = "OrderedReady" - // ParallelInstanceManagementPolicyType InstanceManagementPolicyType = "Parallel" + ParallelInstanceManagementPolicyType InstanceManagementPolicyType = "Parallel" ) type MetricSpec struct { diff --git a/config/base/crd/bases/compute.datumapis.com_workloaddeployments.yaml b/config/base/crd/bases/compute.datumapis.com_workloaddeployments.yaml index 73ade506..f0ae078a 100644 --- a/config/base/crd/bases/compute.datumapis.com_workloaddeployments.yaml +++ b/config/base/crd/bases/compute.datumapis.com_workloaddeployments.yaml @@ -106,6 +106,14 @@ spec: description: |- Controls how instances are managed during scale up and down, as well as during maintenance events. + + OrderedReady creates and deletes instances one at a time, waiting for + each to become ready before moving to the next. Parallel creates and + deletes instances all at once without waiting on readiness. Both replace + instances on a template change one at a time. + enum: + - OrderedReady + - Parallel type: string maxReplicas: description: The maximum number of replicas. diff --git a/config/base/crd/bases/compute.datumapis.com_workloads.yaml b/config/base/crd/bases/compute.datumapis.com_workloads.yaml index d64fbb4f..a7a8ae24 100644 --- a/config/base/crd/bases/compute.datumapis.com_workloads.yaml +++ b/config/base/crd/bases/compute.datumapis.com_workloads.yaml @@ -163,6 +163,14 @@ spec: description: |- Controls how instances are managed during scale up and down, as well as during maintenance events. + + OrderedReady creates and deletes instances one at a time, waiting for + each to become ready before moving to the next. Parallel creates and + deletes instances all at once without waiting on readiness. Both replace + instances on a template change one at a time. + enum: + - OrderedReady + - Parallel type: string maxReplicas: description: The maximum number of replicas. diff --git a/docs/api/workloaddeployments.md b/docs/api/workloaddeployments.md index 365579e9..8f61c6cc 100644 --- a/docs/api/workloaddeployments.md +++ b/docs/api/workloaddeployments.md @@ -192,11 +192,17 @@ Scale settings such as minimum and maximum replica counts. instanceManagementPolicy - string + enum Controls how instances are managed during scale up and down, as well as -during maintenance events.
+during maintenance events. + +OrderedReady creates and deletes instances one at a time, waiting for +each to become ready before moving to the next. Parallel creates and +deletes instances all at once without waiting on readiness. Both replace +instances on a template change one at a time.

+ Enum: OrderedReady, Parallel
Default: OrderedReady
true diff --git a/docs/api/workloads.md b/docs/api/workloads.md index 6ea07fbf..0c4a046b 100644 --- a/docs/api/workloads.md +++ b/docs/api/workloads.md @@ -188,11 +188,17 @@ Scale settings such as minimum and maximum replica counts. instanceManagementPolicy - string + enum Controls how instances are managed during scale up and down, as well as -during maintenance events.
+during maintenance events. + +OrderedReady creates and deletes instances one at a time, waiting for +each to become ready before moving to the next. Parallel creates and +deletes instances all at once without waiting on readiness. Both replace +instances on a template change one at a time.

+ Enum: OrderedReady, Parallel
Default: OrderedReady
true diff --git a/internal/controller/instancecontrol/stateful/stateful_control.go b/internal/controller/instancecontrol/stateful/stateful_control.go index f61f7421..550fb837 100644 --- a/internal/controller/instancecontrol/stateful/stateful_control.go +++ b/internal/controller/instancecontrol/stateful/stateful_control.go @@ -211,7 +211,7 @@ func (c *statefulControl) GetActions( actions := make([]instancecontrol.Action, 0, len(createActions)+len(waitActions)+len(recreateActions)+len(deleteActions)+len(patchLabelActions)+len(patchTemplateHashActions)) switch deployment.Spec.ScaleSettings.InstanceManagementPolicy { - case v1alpha.OrderedReadyInstanceManagementPolicyType: + case v1alpha.OrderedReadyInstanceManagementPolicyType, "": // Add create and wait actions, and sort by ordinal. This allows us to wait // for instances to be processed in the correct order. @@ -233,6 +233,23 @@ func (c *statefulControl) GetActions( } } + case v1alpha.ParallelInstanceManagementPolicyType: + slices.SortFunc(createActions, ascendingOrdinal) + slices.SortFunc(waitActions, ascendingOrdinal) + + actions = append(actions, createActions...) + actions = append(actions, deleteActions...) + actions = append(actions, waitActions...) + + for i := range recreateActions { + if i > 0 || len(createActions) > 0 || len(waitActions) > 0 { + recreateActions[i].SkipExecution() + } + } + actions = append(actions, recreateActions...) + + default: + return nil, fmt.Errorf("unsupported instance management policy %q", deployment.Spec.ScaleSettings.InstanceManagementPolicy) } actions = append(actions, patchLabelActions...) diff --git a/internal/controller/instancecontrol/stateful/stateful_control_test.go b/internal/controller/instancecontrol/stateful/stateful_control_test.go index 98106543..c8738f0a 100644 --- a/internal/controller/instancecontrol/stateful/stateful_control_test.go +++ b/internal/controller/instancecontrol/stateful/stateful_control_test.go @@ -814,3 +814,118 @@ func getInstanceTemplate(name string, ordinal int) *v1alpha.Instance { return instance } + +func getParallelWorkloadDeployment(name string, minReplicas int32) *v1alpha.WorkloadDeployment { + deployment := getWorkloadDeployment(name, minReplicas) + deployment.Spec.ScaleSettings.InstanceManagementPolicy = v1alpha.ParallelInstanceManagementPolicyType + return deployment +} + +func TestParallel_FreshDeploymentCreatesAllInstances(t *testing.T) { + control := NewWithOptions(Options{}) + deployment := getParallelWorkloadDeployment("test-deploy", 3) + + actions, err := control.GetActions(context.Background(), scheme, deployment, 3, nil) + + require.NoError(t, err) + require.Len(t, actions, 3) + for i, action := range actions { + assert.Equal(t, fmt.Sprintf("test-deploy-%d", i), action.Object.GetName()) + assert.Equal(t, instancecontrol.ActionTypeCreate, action.ActionType()) + assert.False(t, action.IsSkipped()) + } +} + +func TestParallel_ScaleUpDoesNotWaitOnNotReadyInstance(t *testing.T) { + control := NewWithOptions(Options{}) + deployment := getParallelWorkloadDeployment("test-deploy", 3) + + notReady := getInstanceForDeployment(deployment, 0) + apimeta.SetStatusCondition(¬Ready.Status.Conditions, metav1.Condition{ + Type: v1alpha.InstanceReady, + Status: metav1.ConditionFalse, + }) + + actions, err := control.GetActions(context.Background(), scheme, deployment, 3, []v1alpha.Instance{*notReady}) + + require.NoError(t, err) + created := map[string]bool{} + for _, action := range actions { + if action.ActionType() == instancecontrol.ActionTypeCreate { + assert.False(t, action.IsSkipped(), action.Object.GetName()) + created[action.Object.GetName()] = true + } + } + assert.Equal(t, map[string]bool{"test-deploy-1": true, "test-deploy-2": true}, created) +} + +func TestParallel_ScaleDownDeletesAllExtraInstances(t *testing.T) { + control := NewWithOptions(Options{}) + deployment := getParallelWorkloadDeployment("test-deploy", 1) + + current := []v1alpha.Instance{ + *getInstanceForDeployment(deployment, 0), + *getInstanceForDeployment(deployment, 1), + *getInstanceForDeployment(deployment, 2), + } + + actions, err := control.GetActions(context.Background(), scheme, deployment, 1, current) + + require.NoError(t, err) + require.Len(t, actions, 2) + for _, action := range actions { + assert.Equal(t, instancecontrol.ActionTypeDelete, action.ActionType()) + assert.False(t, action.IsSkipped(), action.Object.GetName()) + } +} + +func TestParallel_TemplateChangeReplacesOneInstanceAtATime(t *testing.T) { + control := NewWithOptions(Options{}) + deployment := getParallelWorkloadDeployment("test-deploy", 2) + + current := []v1alpha.Instance{ + *getInstanceForDeployment(deployment, 0), + *getInstanceForDeployment(deployment, 1), + } + deployment.Spec.Template.Spec.Runtime.Sandbox.Containers[0].Image = "test-image-update" + + actions, err := control.GetActions(context.Background(), scheme, deployment, 2, current) + + require.NoError(t, err) + require.Len(t, actions, 2) + assert.Equal(t, "test-deploy-1", actions[0].Object.GetName()) + assert.Equal(t, instancecontrol.ActionTypeDelete, actions[0].ActionType()) + assert.False(t, actions[0].IsSkipped()) + assert.Equal(t, "test-deploy-0", actions[1].Object.GetName()) + assert.True(t, actions[1].IsSkipped()) +} + +func TestParallel_TemplateChangeHeldWhileInstanceIsReplaced(t *testing.T) { + control := NewWithOptions(Options{}) + deployment := getParallelWorkloadDeployment("test-deploy", 2) + + current := []v1alpha.Instance{*getInstanceForDeployment(deployment, 0)} + deployment.Spec.Template.Spec.Runtime.Sandbox.Containers[0].Image = "test-image-update" + + actions, err := control.GetActions(context.Background(), scheme, deployment, 2, current) + + require.NoError(t, err) + for _, action := range actions { + switch action.ActionType() { + case instancecontrol.ActionTypeCreate: + assert.False(t, action.IsSkipped(), action.Object.GetName()) + case instancecontrol.ActionTypeDelete: + assert.True(t, action.IsSkipped(), action.Object.GetName()) + } + } +} + +func TestUnsupportedInstanceManagementPolicyReturnsError(t *testing.T) { + control := NewWithOptions(Options{}) + deployment := getWorkloadDeployment("test-deploy", 1) + deployment.Spec.ScaleSettings.InstanceManagementPolicy = "Bogus" + + _, err := control.GetActions(context.Background(), scheme, deployment, 1, nil) + + assert.ErrorContains(t, err, "Bogus") +} From c88b3935573131a6cab58b9d83506123c283002f Mon Sep 17 00:00:00 2001 From: Scot Schuchert-Wells Date: Wed, 11 Nov 2026 15:04:24 +0000 Subject: [PATCH 2/2] refactor: split rollout ordering out of GetActions Adding the Parallel case pushed GetActions over the linter's complexity limit, so the per-policy ordering moves into its own function. Key changes: - move the policy switch into orderRolloutActions - use a constant for the repeated test image and drop an unused test helper parameter --- .../stateful/stateful_control.go | 24 ++++++++++++++----- .../stateful/stateful_control_test.go | 22 +++++++++-------- 2 files changed, 30 insertions(+), 16 deletions(-) diff --git a/internal/controller/instancecontrol/stateful/stateful_control.go b/internal/controller/instancecontrol/stateful/stateful_control.go index 550fb837..b446709a 100644 --- a/internal/controller/instancecontrol/stateful/stateful_control.go +++ b/internal/controller/instancecontrol/stateful/stateful_control.go @@ -208,9 +208,24 @@ func (c *statefulControl) GetActions( slices.SortFunc(recreateActions, descendingOrdinal) slices.SortFunc(deleteActions, descendingOrdinal) - actions := make([]instancecontrol.Action, 0, len(createActions)+len(waitActions)+len(recreateActions)+len(deleteActions)+len(patchLabelActions)+len(patchTemplateHashActions)) + actions, err := orderRolloutActions(deployment.Spec.ScaleSettings.InstanceManagementPolicy, createActions, waitActions, recreateActions, deleteActions) + if err != nil { + return nil, err + } + + actions = append(actions, patchLabelActions...) + actions = append(actions, patchTemplateHashActions...) + + return actions, nil +} + +func orderRolloutActions( + policy v1alpha.InstanceManagementPolicyType, + createActions, waitActions, recreateActions, deleteActions []instancecontrol.Action, +) ([]instancecontrol.Action, error) { + actions := make([]instancecontrol.Action, 0, len(createActions)+len(waitActions)+len(recreateActions)+len(deleteActions)) - switch deployment.Spec.ScaleSettings.InstanceManagementPolicy { + switch policy { case v1alpha.OrderedReadyInstanceManagementPolicyType, "": // Add create and wait actions, and sort by ordinal. This allows us to wait @@ -249,12 +264,9 @@ func (c *statefulControl) GetActions( actions = append(actions, recreateActions...) default: - return nil, fmt.Errorf("unsupported instance management policy %q", deployment.Spec.ScaleSettings.InstanceManagementPolicy) + return nil, fmt.Errorf("unsupported instance management policy %q", policy) } - actions = append(actions, patchLabelActions...) - actions = append(actions, patchTemplateHashActions...) - return actions, nil } diff --git a/internal/controller/instancecontrol/stateful/stateful_control_test.go b/internal/controller/instancecontrol/stateful/stateful_control_test.go index c8738f0a..09586f45 100644 --- a/internal/controller/instancecontrol/stateful/stateful_control_test.go +++ b/internal/controller/instancecontrol/stateful/stateful_control_test.go @@ -20,6 +20,8 @@ import ( "go.datum.net/compute/internal/controller/instancecontrol" ) +const testImageUpdate = "test-image-update" + var ( scheme = runtime.NewScheme() ) @@ -101,7 +103,7 @@ func TestUpdateWithAllReadyInstances(t *testing.T) { currentInstances = append(currentInstances, *getInstanceForDeployment(deployment, 0)) currentInstances = append(currentInstances, *getInstanceForDeployment(deployment, 1)) - deployment.Spec.Template.Spec.Runtime.Sandbox.Containers[0].Image = "test-image-update" + deployment.Spec.Template.Spec.Runtime.Sandbox.Containers[0].Image = testImageUpdate actions, err := control.GetActions(ctx, scheme, deployment, deployment.Spec.ScaleSettings.MinReplicas, currentInstances) @@ -815,15 +817,15 @@ func getInstanceTemplate(name string, ordinal int) *v1alpha.Instance { return instance } -func getParallelWorkloadDeployment(name string, minReplicas int32) *v1alpha.WorkloadDeployment { - deployment := getWorkloadDeployment(name, minReplicas) +func getParallelWorkloadDeployment(minReplicas int32) *v1alpha.WorkloadDeployment { + deployment := getWorkloadDeployment("test-deploy", minReplicas) deployment.Spec.ScaleSettings.InstanceManagementPolicy = v1alpha.ParallelInstanceManagementPolicyType return deployment } func TestParallel_FreshDeploymentCreatesAllInstances(t *testing.T) { control := NewWithOptions(Options{}) - deployment := getParallelWorkloadDeployment("test-deploy", 3) + deployment := getParallelWorkloadDeployment(3) actions, err := control.GetActions(context.Background(), scheme, deployment, 3, nil) @@ -838,7 +840,7 @@ func TestParallel_FreshDeploymentCreatesAllInstances(t *testing.T) { func TestParallel_ScaleUpDoesNotWaitOnNotReadyInstance(t *testing.T) { control := NewWithOptions(Options{}) - deployment := getParallelWorkloadDeployment("test-deploy", 3) + deployment := getParallelWorkloadDeployment(3) notReady := getInstanceForDeployment(deployment, 0) apimeta.SetStatusCondition(¬Ready.Status.Conditions, metav1.Condition{ @@ -861,7 +863,7 @@ func TestParallel_ScaleUpDoesNotWaitOnNotReadyInstance(t *testing.T) { func TestParallel_ScaleDownDeletesAllExtraInstances(t *testing.T) { control := NewWithOptions(Options{}) - deployment := getParallelWorkloadDeployment("test-deploy", 1) + deployment := getParallelWorkloadDeployment(1) current := []v1alpha.Instance{ *getInstanceForDeployment(deployment, 0), @@ -881,13 +883,13 @@ func TestParallel_ScaleDownDeletesAllExtraInstances(t *testing.T) { func TestParallel_TemplateChangeReplacesOneInstanceAtATime(t *testing.T) { control := NewWithOptions(Options{}) - deployment := getParallelWorkloadDeployment("test-deploy", 2) + deployment := getParallelWorkloadDeployment(2) current := []v1alpha.Instance{ *getInstanceForDeployment(deployment, 0), *getInstanceForDeployment(deployment, 1), } - deployment.Spec.Template.Spec.Runtime.Sandbox.Containers[0].Image = "test-image-update" + deployment.Spec.Template.Spec.Runtime.Sandbox.Containers[0].Image = testImageUpdate actions, err := control.GetActions(context.Background(), scheme, deployment, 2, current) @@ -902,10 +904,10 @@ func TestParallel_TemplateChangeReplacesOneInstanceAtATime(t *testing.T) { func TestParallel_TemplateChangeHeldWhileInstanceIsReplaced(t *testing.T) { control := NewWithOptions(Options{}) - deployment := getParallelWorkloadDeployment("test-deploy", 2) + deployment := getParallelWorkloadDeployment(2) current := []v1alpha.Instance{*getInstanceForDeployment(deployment, 0)} - deployment.Spec.Template.Spec.Runtime.Sandbox.Containers[0].Image = "test-image-update" + deployment.Spec.Template.Spec.Runtime.Sandbox.Containers[0].Image = testImageUpdate actions, err := control.GetActions(context.Background(), scheme, deployment, 2, current)