Merge pull request #112744 from pwschuurman/statefulset-slice-impl

Add implementation of KEP-3335, StatefulSetSlice

pkg/apis/apps/types.go

@@ -143,6 +143,21 @@ type StatefulSetPersistentVolumeClaimRetentionPolicy struct {
 	WhenScaled PersistentVolumeClaimRetentionPolicyType
 }
 
+// StatefulSetOrdinals describes the policy used for replica ordinal assignment
+// in this StatefulSet.
+type StatefulSetOrdinals struct {
+	// start is the number representing the first replica's index. It may be used
+	// to number replicas from an alternate index (eg: 1-indexed) over the default
+	// 0-indexed names, or to orchestrate progressive movement of replicas from
+	// one StatefulSet to another.
+	// If set, replica indices will be in the range:
+	//   [.spec.ordinals.start, .spec.ordinals.start + .spec.replicas).
+	// If unset, defaults to 0. Replica indices will be in the range:
+	//   [0, .spec.replicas).
+	// +optional
+	Start int32
+}
+
 // A StatefulSetSpec is the specification of a StatefulSet.
 type StatefulSetSpec struct {
 	// Replicas is the desired number of replicas of the given Template.
@@ -162,7 +177,9 @@ type StatefulSetSpec struct {
 	// Template is the object that describes the pod that will be created if
 	// insufficient replicas are detected. Each pod stamped out by the StatefulSet
 	// will fulfill this Template, but have a unique identity from the rest
-	// of the StatefulSet.
+	// of the StatefulSet. Each pod will be named with the format
+	// <statefulsetname>-<podindex>. For example, a pod in a StatefulSet named
+	// "web" with index number "3" would be named "web-3".
 	Template api.PodTemplateSpec
 
 	// VolumeClaimTemplates is a list of claims that pods are allowed to reference.
@@ -215,6 +232,14 @@ type StatefulSetSpec struct {
 	// StatefulSetAutoDeletePVC feature gate to be enabled, which is alpha.
 	// +optional
 	PersistentVolumeClaimRetentionPolicy *StatefulSetPersistentVolumeClaimRetentionPolicy
+
+	// ordinals controls the numbering of replica indices in a StatefulSet. The
+	// default ordinals behavior assigns a "0" index to the first replica and
+	// increments the index by one for each additional replica requested. Using
+	// the ordinals field requires the StatefulSetStartOrdinal feature gate to be
+	// enabled, which is alpha.
+	// +optional
+	Ordinals *StatefulSetOrdinals
 }
 
 // StatefulSetStatus represents the current state of a StatefulSet.

pkg/apis/apps/v1/zz_generated.conversion.go

@@ -262,6 +262,16 @@ func RegisterConversions(s *runtime.Scheme) error {
 	}); err != nil {
 		return err
 	}
+	if err := s.AddGeneratedConversionFunc((*v1.StatefulSetOrdinals)(nil), (*apps.StatefulSetOrdinals)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_v1_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(a.(*v1.StatefulSetOrdinals), b.(*apps.StatefulSetOrdinals), scope)
+	}); err != nil {
+		return err
+	}
+	if err := s.AddGeneratedConversionFunc((*apps.StatefulSetOrdinals)(nil), (*v1.StatefulSetOrdinals)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_apps_StatefulSetOrdinals_To_v1_StatefulSetOrdinals(a.(*apps.StatefulSetOrdinals), b.(*v1.StatefulSetOrdinals), scope)
+	}); err != nil {
+		return err
+	}
 	if err := s.AddGeneratedConversionFunc((*v1.StatefulSetPersistentVolumeClaimRetentionPolicy)(nil), (*apps.StatefulSetPersistentVolumeClaimRetentionPolicy)(nil), func(a, b interface{}, scope conversion.Scope) error {
 		return Convert_v1_StatefulSetPersistentVolumeClaimRetentionPolicy_To_apps_StatefulSetPersistentVolumeClaimRetentionPolicy(a.(*v1.StatefulSetPersistentVolumeClaimRetentionPolicy), b.(*apps.StatefulSetPersistentVolumeClaimRetentionPolicy), scope)
 	}); err != nil {
@@ -1167,6 +1177,26 @@ func Convert_apps_StatefulSetList_To_v1_StatefulSetList(in *apps.StatefulSetList
 	return autoConvert_apps_StatefulSetList_To_v1_StatefulSetList(in, out, s)
 }
 
+func autoConvert_v1_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(in *v1.StatefulSetOrdinals, out *apps.StatefulSetOrdinals, s conversion.Scope) error {
+	out.Start = in.Start
+	return nil
+}
+
+// Convert_v1_StatefulSetOrdinals_To_apps_StatefulSetOrdinals is an autogenerated conversion function.
+func Convert_v1_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(in *v1.StatefulSetOrdinals, out *apps.StatefulSetOrdinals, s conversion.Scope) error {
+	return autoConvert_v1_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(in, out, s)
+}
+
+func autoConvert_apps_StatefulSetOrdinals_To_v1_StatefulSetOrdinals(in *apps.StatefulSetOrdinals, out *v1.StatefulSetOrdinals, s conversion.Scope) error {
+	out.Start = in.Start
+	return nil
+}
+
+// Convert_apps_StatefulSetOrdinals_To_v1_StatefulSetOrdinals is an autogenerated conversion function.
+func Convert_apps_StatefulSetOrdinals_To_v1_StatefulSetOrdinals(in *apps.StatefulSetOrdinals, out *v1.StatefulSetOrdinals, s conversion.Scope) error {
+	return autoConvert_apps_StatefulSetOrdinals_To_v1_StatefulSetOrdinals(in, out, s)
+}
+
 func autoConvert_v1_StatefulSetPersistentVolumeClaimRetentionPolicy_To_apps_StatefulSetPersistentVolumeClaimRetentionPolicy(in *v1.StatefulSetPersistentVolumeClaimRetentionPolicy, out *apps.StatefulSetPersistentVolumeClaimRetentionPolicy, s conversion.Scope) error {
 	out.WhenDeleted = apps.PersistentVolumeClaimRetentionPolicyType(in.WhenDeleted)
 	out.WhenScaled = apps.PersistentVolumeClaimRetentionPolicyType(in.WhenScaled)
@@ -1206,6 +1236,7 @@ func autoConvert_v1_StatefulSetSpec_To_apps_StatefulSetSpec(in *v1.StatefulSetSp
 	out.RevisionHistoryLimit = (*int32)(unsafe.Pointer(in.RevisionHistoryLimit))
 	out.MinReadySeconds = in.MinReadySeconds
 	out.PersistentVolumeClaimRetentionPolicy = (*apps.StatefulSetPersistentVolumeClaimRetentionPolicy)(unsafe.Pointer(in.PersistentVolumeClaimRetentionPolicy))
+	out.Ordinals = (*apps.StatefulSetOrdinals)(unsafe.Pointer(in.Ordinals))
 	return nil
 }
 
@@ -1226,6 +1257,7 @@ func autoConvert_apps_StatefulSetSpec_To_v1_StatefulSetSpec(in *apps.StatefulSet
 	out.RevisionHistoryLimit = (*int32)(unsafe.Pointer(in.RevisionHistoryLimit))
 	out.MinReadySeconds = in.MinReadySeconds
 	out.PersistentVolumeClaimRetentionPolicy = (*v1.StatefulSetPersistentVolumeClaimRetentionPolicy)(unsafe.Pointer(in.PersistentVolumeClaimRetentionPolicy))
+	out.Ordinals = (*v1.StatefulSetOrdinals)(unsafe.Pointer(in.Ordinals))
 	return nil
 }
 

pkg/apis/apps/v1beta1/zz_generated.conversion.go

@@ -213,6 +213,16 @@ func RegisterConversions(s *runtime.Scheme) error {
 	}); err != nil {
 		return err
 	}
+	if err := s.AddGeneratedConversionFunc((*v1beta1.StatefulSetOrdinals)(nil), (*apps.StatefulSetOrdinals)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_v1beta1_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(a.(*v1beta1.StatefulSetOrdinals), b.(*apps.StatefulSetOrdinals), scope)
+	}); err != nil {
+		return err
+	}
+	if err := s.AddGeneratedConversionFunc((*apps.StatefulSetOrdinals)(nil), (*v1beta1.StatefulSetOrdinals)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_apps_StatefulSetOrdinals_To_v1beta1_StatefulSetOrdinals(a.(*apps.StatefulSetOrdinals), b.(*v1beta1.StatefulSetOrdinals), scope)
+	}); err != nil {
+		return err
+	}
 	if err := s.AddGeneratedConversionFunc((*v1beta1.StatefulSetPersistentVolumeClaimRetentionPolicy)(nil), (*apps.StatefulSetPersistentVolumeClaimRetentionPolicy)(nil), func(a, b interface{}, scope conversion.Scope) error {
 		return Convert_v1beta1_StatefulSetPersistentVolumeClaimRetentionPolicy_To_apps_StatefulSetPersistentVolumeClaimRetentionPolicy(a.(*v1beta1.StatefulSetPersistentVolumeClaimRetentionPolicy), b.(*apps.StatefulSetPersistentVolumeClaimRetentionPolicy), scope)
 	}); err != nil {
@@ -831,6 +841,26 @@ func Convert_apps_StatefulSetList_To_v1beta1_StatefulSetList(in *apps.StatefulSe
 	return autoConvert_apps_StatefulSetList_To_v1beta1_StatefulSetList(in, out, s)
 }
 
+func autoConvert_v1beta1_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(in *v1beta1.StatefulSetOrdinals, out *apps.StatefulSetOrdinals, s conversion.Scope) error {
+	out.Start = in.Start
+	return nil
+}
+
+// Convert_v1beta1_StatefulSetOrdinals_To_apps_StatefulSetOrdinals is an autogenerated conversion function.
+func Convert_v1beta1_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(in *v1beta1.StatefulSetOrdinals, out *apps.StatefulSetOrdinals, s conversion.Scope) error {
+	return autoConvert_v1beta1_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(in, out, s)
+}
+
+func autoConvert_apps_StatefulSetOrdinals_To_v1beta1_StatefulSetOrdinals(in *apps.StatefulSetOrdinals, out *v1beta1.StatefulSetOrdinals, s conversion.Scope) error {
+	out.Start = in.Start
+	return nil
+}
+
+// Convert_apps_StatefulSetOrdinals_To_v1beta1_StatefulSetOrdinals is an autogenerated conversion function.
+func Convert_apps_StatefulSetOrdinals_To_v1beta1_StatefulSetOrdinals(in *apps.StatefulSetOrdinals, out *v1beta1.StatefulSetOrdinals, s conversion.Scope) error {
+	return autoConvert_apps_StatefulSetOrdinals_To_v1beta1_StatefulSetOrdinals(in, out, s)
+}
+
 func autoConvert_v1beta1_StatefulSetPersistentVolumeClaimRetentionPolicy_To_apps_StatefulSetPersistentVolumeClaimRetentionPolicy(in *v1beta1.StatefulSetPersistentVolumeClaimRetentionPolicy, out *apps.StatefulSetPersistentVolumeClaimRetentionPolicy, s conversion.Scope) error {
 	out.WhenDeleted = apps.PersistentVolumeClaimRetentionPolicyType(in.WhenDeleted)
 	out.WhenScaled = apps.PersistentVolumeClaimRetentionPolicyType(in.WhenScaled)
@@ -870,6 +900,7 @@ func autoConvert_v1beta1_StatefulSetSpec_To_apps_StatefulSetSpec(in *v1beta1.Sta
 	out.RevisionHistoryLimit = (*int32)(unsafe.Pointer(in.RevisionHistoryLimit))
 	out.MinReadySeconds = in.MinReadySeconds
 	out.PersistentVolumeClaimRetentionPolicy = (*apps.StatefulSetPersistentVolumeClaimRetentionPolicy)(unsafe.Pointer(in.PersistentVolumeClaimRetentionPolicy))
+	out.Ordinals = (*apps.StatefulSetOrdinals)(unsafe.Pointer(in.Ordinals))
 	return nil
 }
 
@@ -890,6 +921,7 @@ func autoConvert_apps_StatefulSetSpec_To_v1beta1_StatefulSetSpec(in *apps.Statef
 	out.RevisionHistoryLimit = (*int32)(unsafe.Pointer(in.RevisionHistoryLimit))
 	out.MinReadySeconds = in.MinReadySeconds
 	out.PersistentVolumeClaimRetentionPolicy = (*v1beta1.StatefulSetPersistentVolumeClaimRetentionPolicy)(unsafe.Pointer(in.PersistentVolumeClaimRetentionPolicy))
+	out.Ordinals = (*v1beta1.StatefulSetOrdinals)(unsafe.Pointer(in.Ordinals))
 	return nil
 }
 

pkg/apis/apps/v1beta2/zz_generated.conversion.go

@@ -283,6 +283,16 @@ func RegisterConversions(s *runtime.Scheme) error {
 	}); err != nil {
 		return err
 	}
+	if err := s.AddGeneratedConversionFunc((*v1beta2.StatefulSetOrdinals)(nil), (*apps.StatefulSetOrdinals)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_v1beta2_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(a.(*v1beta2.StatefulSetOrdinals), b.(*apps.StatefulSetOrdinals), scope)
+	}); err != nil {
+		return err
+	}
+	if err := s.AddGeneratedConversionFunc((*apps.StatefulSetOrdinals)(nil), (*v1beta2.StatefulSetOrdinals)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_apps_StatefulSetOrdinals_To_v1beta2_StatefulSetOrdinals(a.(*apps.StatefulSetOrdinals), b.(*v1beta2.StatefulSetOrdinals), scope)
+	}); err != nil {
+		return err
+	}
 	if err := s.AddGeneratedConversionFunc((*v1beta2.StatefulSetPersistentVolumeClaimRetentionPolicy)(nil), (*apps.StatefulSetPersistentVolumeClaimRetentionPolicy)(nil), func(a, b interface{}, scope conversion.Scope) error {
 		return Convert_v1beta2_StatefulSetPersistentVolumeClaimRetentionPolicy_To_apps_StatefulSetPersistentVolumeClaimRetentionPolicy(a.(*v1beta2.StatefulSetPersistentVolumeClaimRetentionPolicy), b.(*apps.StatefulSetPersistentVolumeClaimRetentionPolicy), scope)
 	}); err != nil {
@@ -1263,6 +1273,26 @@ func Convert_apps_StatefulSetList_To_v1beta2_StatefulSetList(in *apps.StatefulSe
 	return autoConvert_apps_StatefulSetList_To_v1beta2_StatefulSetList(in, out, s)
 }
 
+func autoConvert_v1beta2_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(in *v1beta2.StatefulSetOrdinals, out *apps.StatefulSetOrdinals, s conversion.Scope) error {
+	out.Start = in.Start
+	return nil
+}
+
+// Convert_v1beta2_StatefulSetOrdinals_To_apps_StatefulSetOrdinals is an autogenerated conversion function.
+func Convert_v1beta2_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(in *v1beta2.StatefulSetOrdinals, out *apps.StatefulSetOrdinals, s conversion.Scope) error {
+	return autoConvert_v1beta2_StatefulSetOrdinals_To_apps_StatefulSetOrdinals(in, out, s)
+}
+
+func autoConvert_apps_StatefulSetOrdinals_To_v1beta2_StatefulSetOrdinals(in *apps.StatefulSetOrdinals, out *v1beta2.StatefulSetOrdinals, s conversion.Scope) error {
+	out.Start = in.Start
+	return nil
+}
+
+// Convert_apps_StatefulSetOrdinals_To_v1beta2_StatefulSetOrdinals is an autogenerated conversion function.
+func Convert_apps_StatefulSetOrdinals_To_v1beta2_StatefulSetOrdinals(in *apps.StatefulSetOrdinals, out *v1beta2.StatefulSetOrdinals, s conversion.Scope) error {
+	return autoConvert_apps_StatefulSetOrdinals_To_v1beta2_StatefulSetOrdinals(in, out, s)
+}
+
 func autoConvert_v1beta2_StatefulSetPersistentVolumeClaimRetentionPolicy_To_apps_StatefulSetPersistentVolumeClaimRetentionPolicy(in *v1beta2.StatefulSetPersistentVolumeClaimRetentionPolicy, out *apps.StatefulSetPersistentVolumeClaimRetentionPolicy, s conversion.Scope) error {
 	out.WhenDeleted = apps.PersistentVolumeClaimRetentionPolicyType(in.WhenDeleted)
 	out.WhenScaled = apps.PersistentVolumeClaimRetentionPolicyType(in.WhenScaled)
@@ -1302,6 +1332,7 @@ func autoConvert_v1beta2_StatefulSetSpec_To_apps_StatefulSetSpec(in *v1beta2.Sta
 	out.RevisionHistoryLimit = (*int32)(unsafe.Pointer(in.RevisionHistoryLimit))
 	out.MinReadySeconds = in.MinReadySeconds
 	out.PersistentVolumeClaimRetentionPolicy = (*apps.StatefulSetPersistentVolumeClaimRetentionPolicy)(unsafe.Pointer(in.PersistentVolumeClaimRetentionPolicy))
+	out.Ordinals = (*apps.StatefulSetOrdinals)(unsafe.Pointer(in.Ordinals))
 	return nil
 }
 
@@ -1322,6 +1353,7 @@ func autoConvert_apps_StatefulSetSpec_To_v1beta2_StatefulSetSpec(in *apps.Statef
 	out.RevisionHistoryLimit = (*int32)(unsafe.Pointer(in.RevisionHistoryLimit))
 	out.MinReadySeconds = in.MinReadySeconds
 	out.PersistentVolumeClaimRetentionPolicy = (*v1beta2.StatefulSetPersistentVolumeClaimRetentionPolicy)(unsafe.Pointer(in.PersistentVolumeClaimRetentionPolicy))
+	out.Ordinals = (*v1beta2.StatefulSetOrdinals)(unsafe.Pointer(in.Ordinals))
 	return nil
 }
 

pkg/apis/apps/validation/validation.go

@@ -28,9 +28,11 @@ import (
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/validation/field"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/apis/apps"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	apivalidation "k8s.io/kubernetes/pkg/apis/core/validation"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 // ValidateStatefulSetName can be used to check whether the given StatefulSet name is valid.
@@ -128,6 +130,12 @@ func ValidateStatefulSetSpec(spec *apps.StatefulSetSpec, fldPath *field.Path, op
 
 	allErrs = append(allErrs, apivalidation.ValidateNonnegativeField(int64(spec.Replicas), fldPath.Child("replicas"))...)
 	allErrs = append(allErrs, apivalidation.ValidateNonnegativeField(int64(spec.MinReadySeconds), fldPath.Child("minReadySeconds"))...)
+	if utilfeature.DefaultFeatureGate.Enabled(features.StatefulSetStartOrdinal) {
+		if spec.Ordinals != nil {
+			replicaStartOrdinal := spec.Ordinals.Start
+			allErrs = append(allErrs, apivalidation.ValidateNonnegativeField(int64(replicaStartOrdinal), fldPath.Child("ordinals.start"))...)
+		}
+	}
 
 	if spec.Selector == nil {
 		allErrs = append(allErrs, field.Required(fldPath.Child("selector"), ""))
@@ -177,10 +185,17 @@ func ValidateStatefulSetUpdate(statefulSet, oldStatefulSet *apps.StatefulSet, op
 	newStatefulSetClone.Spec.Template = oldStatefulSet.Spec.Template               // +k8s:verify-mutation:reason=clone
 	newStatefulSetClone.Spec.UpdateStrategy = oldStatefulSet.Spec.UpdateStrategy   // +k8s:verify-mutation:reason=clone
 	newStatefulSetClone.Spec.MinReadySeconds = oldStatefulSet.Spec.MinReadySeconds // +k8s:verify-mutation:reason=clone
+	if utilfeature.DefaultFeatureGate.Enabled(features.StatefulSetStartOrdinal) {
+		newStatefulSetClone.Spec.Ordinals = oldStatefulSet.Spec.Ordinals // +k8s:verify-mutation:reason=clone
+	}
 
 	newStatefulSetClone.Spec.PersistentVolumeClaimRetentionPolicy = oldStatefulSet.Spec.PersistentVolumeClaimRetentionPolicy // +k8s:verify-mutation:reason=clone
 	if !apiequality.Semantic.DeepEqual(newStatefulSetClone.Spec, oldStatefulSet.Spec) {
-		allErrs = append(allErrs, field.Forbidden(field.NewPath("spec"), "updates to statefulset spec for fields other than 'replicas', 'template', 'updateStrategy', 'persistentVolumeClaimRetentionPolicy' and 'minReadySeconds' are forbidden"))
+		if utilfeature.DefaultFeatureGate.Enabled(features.StatefulSetStartOrdinal) {
+			allErrs = append(allErrs, field.Forbidden(field.NewPath("spec"), "updates to statefulset spec for fields other than 'replicas', 'ordinals', 'template', 'updateStrategy', 'persistentVolumeClaimRetentionPolicy' and 'minReadySeconds' are forbidden"))
+		} else {
+			allErrs = append(allErrs, field.Forbidden(field.NewPath("spec"), "updates to statefulset spec for fields other than 'replicas', 'template', 'updateStrategy', 'persistentVolumeClaimRetentionPolicy' and 'minReadySeconds' are forbidden"))
+		}
 	}
 
 	return allErrs

pkg/apis/apps/validation/validation_test.go

@@ -86,6 +86,8 @@ func TestValidateStatefulSet(t *testing.T) {
 
 	const enableStatefulSetAutoDeletePVC = "[enable StatefulSetAutoDeletePVC]"
 
+	const enableStatefulSetStartOrdinal = "[enable StatefulSetStartOrdinal]"
+
 	type testCase struct {
 		name string
 		set  apps.StatefulSet
@@ -193,6 +195,20 @@ func TestValidateStatefulSet(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "ordinals.start positive value " + enableStatefulSetStartOrdinal,
+			set: apps.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{Name: "abc-123", Namespace: metav1.NamespaceDefault},
+				Spec: apps.StatefulSetSpec{
+					PodManagementPolicy: apps.ParallelPodManagement,
+					UpdateStrategy:      apps.StatefulSetUpdateStrategy{Type: apps.RollingUpdateStatefulSetStrategyType},
+					Selector:            &metav1.LabelSelector{MatchLabels: validLabels},
+					Template:            validPodTemplate.Template,
+					Replicas:            3,
+					Ordinals:            &apps.StatefulSetOrdinals{Start: 2},
+				},
+			},
+		},
 	}
 
 	errorCases := []testCase{
@@ -635,6 +651,23 @@ func TestValidateStatefulSet(t *testing.T) {
 				field.Invalid(field.NewPath("spec", "updateStrategy", "rollingUpdate", "maxUnavailable"), nil, ""),
 			},
 		},
+		{
+			name: "invalid ordinals.start " + enableStatefulSetStartOrdinal,
+			set: apps.StatefulSet{
+				ObjectMeta: metav1.ObjectMeta{Name: "abc-123", Namespace: metav1.NamespaceDefault},
+				Spec: apps.StatefulSetSpec{
+					PodManagementPolicy: apps.ParallelPodManagement,
+					UpdateStrategy:      apps.StatefulSetUpdateStrategy{Type: apps.RollingUpdateStatefulSetStrategyType},
+					Selector:            &metav1.LabelSelector{MatchLabels: validLabels},
+					Template:            validPodTemplate.Template,
+					Replicas:            3,
+					Ordinals:            &apps.StatefulSetOrdinals{Start: -2},
+				},
+			},
+			errs: field.ErrorList{
+				field.Invalid(field.NewPath("spec", "ordinals.start"), nil, ""),
+			},
+		},
 	}
 
 	cmpOpts := []cmp.Option{cmpopts.IgnoreFields(field.Error{}, "BadValue", "Detail"), cmpopts.SortSlices(func(a, b *field.Error) bool { return a.Error() < b.Error() })}
@@ -651,6 +684,9 @@ func TestValidateStatefulSet(t *testing.T) {
 			if strings.Contains(name, enableStatefulSetAutoDeletePVC) {
 				defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.StatefulSetAutoDeletePVC, true)()
 			}
+			if strings.Contains(name, enableStatefulSetStartOrdinal) {
+				defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.StatefulSetStartOrdinal, true)()
+			}
 
 			errs := ValidateStatefulSet(&testCase.set, pod.GetValidationOptionsFromPodTemplate(&testCase.set.Spec.Template, nil))
 			wantErrs := testCase.errs

pkg/apis/apps/zz_generated.deepcopy.go

@@ -718,6 +718,22 @@ func (in *StatefulSetList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *StatefulSetOrdinals) DeepCopyInto(out *StatefulSetOrdinals) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new StatefulSetOrdinals.
+func (in *StatefulSetOrdinals) DeepCopy() *StatefulSetOrdinals {
+	if in == nil {
+		return nil
+	}
+	out := new(StatefulSetOrdinals)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *StatefulSetPersistentVolumeClaimRetentionPolicy) DeepCopyInto(out *StatefulSetPersistentVolumeClaimRetentionPolicy) {
 	*out = *in
@@ -761,6 +777,11 @@ func (in *StatefulSetSpec) DeepCopyInto(out *StatefulSetSpec) {
 		*out = new(StatefulSetPersistentVolumeClaimRetentionPolicy)
 		**out = **in
 	}
+	if in.Ordinals != nil {
+		in, out := &in.Ordinals, &out.Ordinals
+		*out = new(StatefulSetOrdinals)
+		**out = **in
+	}
 	return
 }
 

pkg/controller/statefulset/stateful_set_control.go

@@ -303,52 +303,53 @@ func (ssc *defaultStatefulSetControl) updateStatefulSet(
 	*status.CollisionCount = collisionCount
 
 	replicaCount := int(*set.Spec.Replicas)
-	// slice that will contain all Pods such that 0 <= getOrdinal(pod) < set.Spec.Replicas
+	// slice that will contain all Pods such that getStartOrdinal(set) <= getOrdinal(pod) <= getEndOrdinal(set)
 	replicas := make([]*v1.Pod, replicaCount)
-	// slice that will contain all Pods such that set.Spec.Replicas <= getOrdinal(pod)
+	// slice that will contain all Pods such that getOrdinal(pod) < getStartOrdinal(set) OR getOrdinal(pod) > getEndOrdinal(set)
 	condemned := make([]*v1.Pod, 0, len(pods))
 	unhealthy := 0
 	var firstUnhealthyPod *v1.Pod
 
 	// First we partition pods into two lists valid replicas and condemned Pods
-	for i := range pods {
+	for _, pod := range pods {
 		status.Replicas++
 
 		// count the number of running and ready replicas
-		if isRunningAndReady(pods[i]) {
+		if isRunningAndReady(pod) {
 			status.ReadyReplicas++
 			// count the number of running and available replicas
-			if isRunningAndAvailable(pods[i], set.Spec.MinReadySeconds) {
+			if isRunningAndAvailable(pod, set.Spec.MinReadySeconds) {
 				status.AvailableReplicas++
 			}
 
 		}
 
 		// count the number of current and update replicas
-		if isCreated(pods[i]) && !isTerminating(pods[i]) {
-			if getPodRevision(pods[i]) == currentRevision.Name {
+		if isCreated(pod) && !isTerminating(pod) {
+			if getPodRevision(pod) == currentRevision.Name {
 				status.CurrentReplicas++
 			}
-			if getPodRevision(pods[i]) == updateRevision.Name {
+			if getPodRevision(pod) == updateRevision.Name {
 				status.UpdatedReplicas++
 			}
 		}
 
-		if ord := getOrdinal(pods[i]); 0 <= ord && ord < replicaCount {
+		if podInOrdinalRange(pod, set) {
 			// if the ordinal of the pod is within the range of the current number of replicas,
 			// insert it at the indirection of its ordinal
-			replicas[ord] = pods[i]
-		} else if ord >= replicaCount {
-			// if the ordinal is greater than the number of replicas add it to the condemned list
-			condemned = append(condemned, pods[i])
+			replicas[getOrdinal(pod)-getStartOrdinal(set)] = pod
+		} else if getOrdinal(pod) >= 0 {
+			// if the ordinal is valid, but not within the range add it to the condemned list
+			condemned = append(condemned, pod)
 		}
 		// If the ordinal could not be parsed (ord < 0), ignore the Pod.
 	}
 
 	// for any empty indices in the sequence [0,set.Spec.Replicas) create a new Pod at the correct revision
-	for ord := 0; ord < replicaCount; ord++ {
-		if replicas[ord] == nil {
-			replicas[ord] = newVersionedStatefulSetPod(
+	for ord := getStartOrdinal(set); ord <= getEndOrdinal(set); ord++ {
+		replicaIdx := ord - getStartOrdinal(set)
+		if replicas[replicaIdx] == nil {
+			replicas[replicaIdx] = newVersionedStatefulSetPod(
 				currentSet,
 				updateSet,
 				currentRevision.Name,
@@ -409,12 +410,13 @@ func (ssc *defaultStatefulSetControl) updateStatefulSet(
 				status.UpdatedReplicas--
 			}
 			status.Replicas--
+			replicaOrd := i + getStartOrdinal(set)
 			replicas[i] = newVersionedStatefulSetPod(
 				currentSet,
 				updateSet,
 				currentRevision.Name,
 				updateRevision.Name,
-				i)
+				replicaOrd)
 		}
 		// If we find a Pod that has not been created we create the Pod
 		if !isCreated(replicas[i]) {
@@ -501,7 +503,7 @@ func (ssc *defaultStatefulSetControl) updateStatefulSet(
 
 	// At this point, all of the current Replicas are Running, Ready and Available, we can consider termination.
 	// We will wait for all predecessors to be Running and Ready prior to attempting a deletion.
-	// We will terminate Pods in a monotonically decreasing order over [len(pods),set.Spec.Replicas).
+	// We will terminate Pods in a monotonically decreasing order.
 	// Note that we do not resurrect Pods in this interval. Also note that scaling will take precedence over
 	// updates.
 	for target := len(condemned) - 1; target >= 0; target-- {
@@ -614,7 +616,7 @@ func updateStatefulSetAfterInvariantEstablished(
 		}
 	}
 
-	// Collect all targets in the range between the 0 and Spec.Replicas. Count any targets in that range
+	// Collect all targets in the range between getStartOrdinal(set) and getEndOrdinal(set). Count any targets in that range
 	// that are unhealthy i.e. terminated or not running and ready as unavailable). Select the
 	// (MaxUnavailable - Unavailable) Pods, in order with respect to their ordinal for termination. Delete
 	// those pods and count the successful deletions. Update the status with the correct number of deletions.

pkg/controller/statefulset/stateful_set_control_test.go

@@ -559,6 +559,74 @@ func PodRecreateDeleteFailure(t *testing.T, set *apps.StatefulSet, invariants in
 	}
 }
 
+func emptyInvariants(set *apps.StatefulSet, om *fakeObjectManager) error {
+	return nil
+}
+
+func TestStatefulSetControlWithStartOrdinal(t *testing.T) {
+	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.StatefulSetStartOrdinal, true)()
+
+	simpleSetFn := func() *apps.StatefulSet {
+		statefulSet := newStatefulSet(3)
+		statefulSet.Spec.Ordinals = &apps.StatefulSetOrdinals{Start: int32(2)}
+		return statefulSet
+	}
+
+	testCases := []struct {
+		fn  func(*testing.T, *apps.StatefulSet, invariantFunc)
+		obj func() *apps.StatefulSet
+	}{
+		{CreatesPodsWithStartOrdinal, simpleSetFn},
+	}
+
+	for _, testCase := range testCases {
+		testObj := testCase.obj
+		testFn := testCase.fn
+
+		set := testObj()
+		testFn(t, set, emptyInvariants)
+	}
+}
+
+func CreatesPodsWithStartOrdinal(t *testing.T, set *apps.StatefulSet, invariants invariantFunc) {
+	client := fake.NewSimpleClientset(set)
+	om, _, ssc := setupController(client)
+
+	if err := scaleUpStatefulSetControl(set, ssc, om, invariants); err != nil {
+		t.Errorf("Failed to turn up StatefulSet : %s", err)
+	}
+	var err error
+	set, err = om.setsLister.StatefulSets(set.Namespace).Get(set.Name)
+	if err != nil {
+		t.Fatalf("Error getting updated StatefulSet: %v", err)
+	}
+	if set.Status.Replicas != 3 {
+		t.Error("Failed to scale statefulset to 3 replicas")
+	}
+	if set.Status.ReadyReplicas != 3 {
+		t.Error("Failed to set ReadyReplicas correctly")
+	}
+	if set.Status.UpdatedReplicas != 3 {
+		t.Error("Failed to set UpdatedReplicas correctly")
+	}
+	selector, err := metav1.LabelSelectorAsSelector(set.Spec.Selector)
+	if err != nil {
+		t.Error(err)
+	}
+	pods, err := om.podsLister.Pods(set.Namespace).List(selector)
+	if err != nil {
+		t.Error(err)
+	}
+	sort.Sort(ascendingOrdinal(pods))
+	for i, pod := range pods {
+		expectedOrdinal := 2 + i
+		actualPodOrdinal := getOrdinal(pod)
+		if actualPodOrdinal != expectedOrdinal {
+			t.Errorf("Expected pod ordinal %d. Got %d", expectedOrdinal, actualPodOrdinal)
+		}
+	}
+}
+
 func TestStatefulSetControlScaleDownDeleteError(t *testing.T) {
 	runTestOverPVCRetentionPolicies(
 		t, "", func(t *testing.T, policy *apps.StatefulSetPersistentVolumeClaimRetentionPolicy) {
@@ -2299,6 +2367,16 @@ func (om *fakeObjectManager) SetDeleteStatefulPodError(err error, after int) {
 	om.deletePodTracker.after = after
 }
 
+func findPodByOrdinal(pods []*v1.Pod, ordinal int) *v1.Pod {
+	for _, pod := range pods {
+		if getOrdinal(pod) == ordinal {
+			return pod.DeepCopy()
+		}
+	}
+
+	return nil
+}
+
 func (om *fakeObjectManager) setPodPending(set *apps.StatefulSet, ordinal int) ([]*v1.Pod, error) {
 	selector, err := metav1.LabelSelectorAsSelector(set.Spec.Selector)
 	if err != nil {
@@ -2308,11 +2386,10 @@ func (om *fakeObjectManager) setPodPending(set *apps.StatefulSet, ordinal int) (
 	if err != nil {
 		return nil, err
 	}
-	if 0 > ordinal || ordinal >= len(pods) {
-		return nil, fmt.Errorf("ordinal %d out of range [0,%d)", ordinal, len(pods))
+	pod := findPodByOrdinal(pods, ordinal)
+	if pod == nil {
+		return nil, fmt.Errorf("setPodPending: pod ordinal %d not found", ordinal)
 	}
-	sort.Sort(ascendingOrdinal(pods))
-	pod := pods[ordinal].DeepCopy()
 	pod.Status.Phase = v1.PodPending
 	fakeResourceVersion(pod)
 	om.podsIndexer.Update(pod)
@@ -2328,11 +2405,10 @@ func (om *fakeObjectManager) setPodRunning(set *apps.StatefulSet, ordinal int) (
 	if err != nil {
 		return nil, err
 	}
-	if 0 > ordinal || ordinal >= len(pods) {
-		return nil, fmt.Errorf("ordinal %d out of range [0,%d)", ordinal, len(pods))
+	pod := findPodByOrdinal(pods, ordinal)
+	if pod == nil {
+		return nil, fmt.Errorf("setPodRunning: pod ordinal %d not found", ordinal)
 	}
-	sort.Sort(ascendingOrdinal(pods))
-	pod := pods[ordinal].DeepCopy()
 	pod.Status.Phase = v1.PodRunning
 	fakeResourceVersion(pod)
 	om.podsIndexer.Update(pod)
@@ -2348,11 +2424,10 @@ func (om *fakeObjectManager) setPodReady(set *apps.StatefulSet, ordinal int) ([]
 	if err != nil {
 		return nil, err
 	}
-	if 0 > ordinal || ordinal >= len(pods) {
-		return nil, fmt.Errorf("ordinal %d out of range [0,%d)", ordinal, len(pods))
+	pod := findPodByOrdinal(pods, ordinal)
+	if pod == nil {
+		return nil, fmt.Errorf("setPodReady: pod ordinal %d not found", ordinal)
 	}
-	sort.Sort(ascendingOrdinal(pods))
-	pod := pods[ordinal].DeepCopy()
 	condition := v1.PodCondition{Type: v1.PodReady, Status: v1.ConditionTrue}
 	podutil.UpdatePodCondition(&pod.Status, &condition)
 	fakeResourceVersion(pod)
@@ -2369,11 +2444,10 @@ func (om *fakeObjectManager) setPodAvailable(set *apps.StatefulSet, ordinal int,
 	if err != nil {
 		return nil, err
 	}
-	if 0 > ordinal || ordinal >= len(pods) {
-		return nil, fmt.Errorf("ordinal %d out of range [0,%d)", ordinal, len(pods))
+	pod := findPodByOrdinal(pods, ordinal)
+	if pod == nil {
+		return nil, fmt.Errorf("setPodAvailable: pod ordinal %d not found", ordinal)
 	}
-	sort.Sort(ascendingOrdinal(pods))
-	pod := pods[ordinal].DeepCopy()
 	condition := v1.PodCondition{Type: v1.PodReady, Status: v1.ConditionTrue, LastTransitionTime: metav1.Time{Time: lastTransitionTime}}
 	_, existingCondition := podutil.GetPodCondition(&pod.Status, condition.Type)
 	if existingCondition != nil {
@@ -2467,28 +2541,28 @@ func assertMonotonicInvariants(set *apps.StatefulSet, om *fakeObjectManager) err
 		return err
 	}
 	sort.Sort(ascendingOrdinal(pods))
-	for ord := 0; ord < len(pods); ord++ {
-		if ord > 0 && isRunningAndReady(pods[ord]) && !isRunningAndReady(pods[ord-1]) {
-			return fmt.Errorf("successor %s is Running and Ready while %s is not", pods[ord].Name, pods[ord-1].Name)
+	for idx := 0; idx < len(pods); idx++ {
+		if idx > 0 && isRunningAndReady(pods[idx]) && !isRunningAndReady(pods[idx-1]) {
+			return fmt.Errorf("successor %s is Running and Ready while %s is not", pods[idx].Name, pods[idx-1].Name)
 		}
 
-		if getOrdinal(pods[ord]) != ord {
-			return fmt.Errorf("pods %s deployed in the wrong order %d", pods[ord].Name, ord)
+		if ord := idx + getStartOrdinal(set); getOrdinal(pods[idx]) != ord {
+			return fmt.Errorf("pods %s deployed in the wrong order %d", pods[idx].Name, ord)
 		}
 
-		if !storageMatches(set, pods[ord]) {
-			return fmt.Errorf("pods %s does not match the storage specification of StatefulSet %s ", pods[ord].Name, set.Name)
+		if !storageMatches(set, pods[idx]) {
+			return fmt.Errorf("pods %s does not match the storage specification of StatefulSet %s ", pods[idx].Name, set.Name)
 		}
 
-		for _, claim := range getPersistentVolumeClaims(set, pods[ord]) {
+		for _, claim := range getPersistentVolumeClaims(set, pods[idx]) {
 			claim, _ := om.claimsLister.PersistentVolumeClaims(set.Namespace).Get(claim.Name)
-			if err := checkClaimInvarients(set, pods[ord], claim, ord); err != nil {
+			if err := checkClaimInvarients(set, pods[idx], claim); err != nil {
 				return err
 			}
 		}
 
-		if !identityMatches(set, pods[ord]) {
-			return fmt.Errorf("pods %s does not match the identity specification of StatefulSet %s ", pods[ord].Name, set.Name)
+		if !identityMatches(set, pods[idx]) {
+			return fmt.Errorf("pods %s does not match the identity specification of StatefulSet %s ", pods[idx].Name, set.Name)
 		}
 	}
 	return nil
@@ -2504,24 +2578,24 @@ func assertBurstInvariants(set *apps.StatefulSet, om *fakeObjectManager) error {
 		return err
 	}
 	sort.Sort(ascendingOrdinal(pods))
-	for ord := 0; ord < len(pods); ord++ {
-		if !storageMatches(set, pods[ord]) {
-			return fmt.Errorf("pods %s does not match the storage specification of StatefulSet %s ", pods[ord].Name, set.Name)
+	for _, pod := range pods {
+		if !storageMatches(set, pod) {
+			return fmt.Errorf("pods %s does not match the storage specification of StatefulSet %s ", pod.Name, set.Name)
 		}
 
-		for _, claim := range getPersistentVolumeClaims(set, pods[ord]) {
+		for _, claim := range getPersistentVolumeClaims(set, pod) {
 			claim, err := om.claimsLister.PersistentVolumeClaims(set.Namespace).Get(claim.Name)
 			if err != nil {
 				return err
 			}
-			if err := checkClaimInvarients(set, pods[ord], claim, ord); err != nil {
+			if err := checkClaimInvarients(set, pod, claim); err != nil {
 				return err
 			}
 		}
 
-		if !identityMatches(set, pods[ord]) {
+		if !identityMatches(set, pod) {
 			return fmt.Errorf("pods %s does not match the identity specification of StatefulSet %s ",
-				pods[ord].Name,
+				pod.Name,
 				set.Name)
 		}
 	}
@@ -2538,24 +2612,24 @@ func assertUpdateInvariants(set *apps.StatefulSet, om *fakeObjectManager) error
 		return err
 	}
 	sort.Sort(ascendingOrdinal(pods))
-	for ord := 0; ord < len(pods); ord++ {
+	for _, pod := range pods {
 
-		if !storageMatches(set, pods[ord]) {
-			return fmt.Errorf("pod %s does not match the storage specification of StatefulSet %s ", pods[ord].Name, set.Name)
+		if !storageMatches(set, pod) {
+			return fmt.Errorf("pod %s does not match the storage specification of StatefulSet %s ", pod.Name, set.Name)
 		}
 
-		for _, claim := range getPersistentVolumeClaims(set, pods[ord]) {
+		for _, claim := range getPersistentVolumeClaims(set, pod) {
 			claim, err := om.claimsLister.PersistentVolumeClaims(set.Namespace).Get(claim.Name)
 			if err != nil {
 				return err
 			}
-			if err := checkClaimInvarients(set, pods[ord], claim, ord); err != nil {
+			if err := checkClaimInvarients(set, pod, claim); err != nil {
 				return err
 			}
 		}
 
-		if !identityMatches(set, pods[ord]) {
-			return fmt.Errorf("pod %s does not match the identity specification of StatefulSet %s ", pods[ord].Name, set.Name)
+		if !identityMatches(set, pod) {
+			return fmt.Errorf("pod %s does not match the identity specification of StatefulSet %s ", pod.Name, set.Name)
 		}
 	}
 	if set.Spec.UpdateStrategy.Type == apps.OnDeleteStatefulSetStrategyType {
@@ -2576,7 +2650,7 @@ func assertUpdateInvariants(set *apps.StatefulSet, om *fakeObjectManager) error
 	return nil
 }
 
-func checkClaimInvarients(set *apps.StatefulSet, pod *v1.Pod, claim *v1.PersistentVolumeClaim, ordinal int) error {
+func checkClaimInvarients(set *apps.StatefulSet, pod *v1.Pod, claim *v1.PersistentVolumeClaim) error {
 	policy := apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
 		WhenScaled:  apps.RetainPersistentVolumeClaimRetentionPolicyType,
 		WhenDeleted: apps.RetainPersistentVolumeClaimRetentionPolicyType,
@@ -2618,14 +2692,14 @@ func checkClaimInvarients(set *apps.StatefulSet, pod *v1.Pod, claim *v1.Persiste
 		if hasOwnerRef(claim, set) {
 			return fmt.Errorf("claim %s has unexpected owner ref on %s for scaledown only", claim.Name, set.Name)
 		}
-		if ordinal >= int(*set.Spec.Replicas) && !hasOwnerRef(claim, pod) {
+		if !podInOrdinalRange(pod, set) && !hasOwnerRef(claim, pod) {
 			return fmt.Errorf("claim %s does not have owner ref on condemned pod %s for scaledown delete", claim.Name, pod.Name)
 		}
-		if ordinal < int(*set.Spec.Replicas) && hasOwnerRef(claim, pod) {
+		if podInOrdinalRange(pod, set) && hasOwnerRef(claim, pod) {
 			return fmt.Errorf("claim %s has unexpected owner ref on condemned pod %s for scaledown delete", claim.Name, pod.Name)
 		}
 	case policy.WhenScaled == delete && policy.WhenDeleted == delete:
-		if ordinal >= int(*set.Spec.Replicas) {
+		if !podInOrdinalRange(pod, set) {
 			if !hasOwnerRef(claim, pod) || hasOwnerRef(claim, set) {
 				return fmt.Errorf("condemned claim %s has bad owner refs: %v", claim.Name, claim.GetOwnerReferences())
 			}
@@ -2666,9 +2740,9 @@ func scaleUpStatefulSetControl(set *apps.StatefulSet,
 		sort.Sort(ascendingOrdinal(pods))
 
 		// ensure all pods are valid (have a phase)
-		for ord, pod := range pods {
+		for _, pod := range pods {
 			if pod.Status.Phase == "" {
-				if pods, err = om.setPodPending(set, ord); err != nil {
+				if pods, err = om.setPodPending(set, getOrdinal(pod)); err != nil {
 					return err
 				}
 				break
@@ -2677,15 +2751,15 @@ func scaleUpStatefulSetControl(set *apps.StatefulSet,
 
 		// select one of the pods and move it forward in status
 		if len(pods) > 0 {
-			ord := int(rand.Int63n(int64(len(pods))))
-			pod := pods[ord]
+			idx := int(rand.Int63n(int64(len(pods))))
+			pod := pods[idx]
 			switch pod.Status.Phase {
 			case v1.PodPending:
-				if pods, err = om.setPodRunning(set, ord); err != nil {
+				if pods, err = om.setPodRunning(set, getOrdinal(pod)); err != nil {
 					return err
 				}
 			case v1.PodRunning:
-				if pods, err = om.setPodReady(set, ord); err != nil {
+				if pods, err = om.setPodReady(set, getOrdinal(pod)); err != nil {
 					return err
 				}
 			default:
@@ -2720,7 +2794,7 @@ func scaleDownStatefulSetControl(set *apps.StatefulSet, ssc StatefulSetControlIn
 			return err
 		}
 		sort.Sort(ascendingOrdinal(pods))
-		if ordinal := len(pods) - 1; ordinal >= 0 {
+		if idx := len(pods) - 1; idx >= 0 {
 			if _, err := ssc.UpdateStatefulSet(context.TODO(), set, pods); err != nil {
 				return err
 			}
@@ -2728,7 +2802,7 @@ func scaleDownStatefulSetControl(set *apps.StatefulSet, ssc StatefulSetControlIn
 			if err != nil {
 				return err
 			}
-			if pods, err = om.addTerminatingPod(set, ordinal); err != nil {
+			if pods, err = om.addTerminatingPod(set, getOrdinal(pods[idx])); err != nil {
 				return err
 			}
 			if _, err = ssc.UpdateStatefulSet(context.TODO(), set, pods); err != nil {
@@ -2860,9 +2934,9 @@ func updateStatefulSetControl(set *apps.StatefulSet,
 		}
 		sort.Sort(ascendingOrdinal(pods))
 		initialized := false
-		for ord, pod := range pods {
+		for _, pod := range pods {
 			if pod.Status.Phase == "" {
-				if pods, err = om.setPodPending(set, ord); err != nil {
+				if pods, err = om.setPodPending(set, getOrdinal(pod)); err != nil {
 					return err
 				}
 				break
@@ -2873,15 +2947,15 @@ func updateStatefulSetControl(set *apps.StatefulSet,
 		}
 
 		if len(pods) > 0 {
-			ord := int(rand.Int63n(int64(len(pods))))
-			pod := pods[ord]
+			idx := int(rand.Int63n(int64(len(pods))))
+			pod := pods[idx]
 			switch pod.Status.Phase {
 			case v1.PodPending:
-				if pods, err = om.setPodRunning(set, ord); err != nil {
+				if pods, err = om.setPodRunning(set, getOrdinal(pod)); err != nil {
 					return err
 				}
 			case v1.PodRunning:
-				if pods, err = om.setPodReady(set, ord); err != nil {
+				if pods, err = om.setPodReady(set, getOrdinal(pod)); err != nil {
 					return err
 				}
 			default:

pkg/controller/statefulset/stateful_set_utils.go

@@ -29,11 +29,13 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/klog/v2"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
 	"k8s.io/kubernetes/pkg/controller"
 	"k8s.io/kubernetes/pkg/controller/history"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 var patchCodec = scheme.Codecs.LegacyCodec(apps.SchemeGroupVersion)
@@ -85,6 +87,29 @@ func getOrdinal(pod *v1.Pod) int {
 	return ordinal
 }
 
+// getStartOrdinal gets the first possible ordinal (inclusive).
+// Returns spec.ordinals.start if spec.ordinals is set, otherwise returns 0.
+func getStartOrdinal(set *apps.StatefulSet) int {
+	if utilfeature.DefaultFeatureGate.Enabled(features.StatefulSetStartOrdinal) {
+		if set.Spec.Ordinals != nil {
+			return int(set.Spec.Ordinals.Start)
+		}
+	}
+	return 0
+}
+
+// getEndOrdinal gets the last possible ordinal (inclusive).
+func getEndOrdinal(set *apps.StatefulSet) int {
+	return getStartOrdinal(set) + int(*set.Spec.Replicas) - 1
+}
+
+// podInOrdinalRange returns true if the pod ordinal is within the allowed
+// range of ordinals that this StatefulSet is set to control.
+func podInOrdinalRange(pod *v1.Pod, set *apps.StatefulSet) bool {
+	ordinal := getOrdinal(pod)
+	return ordinal >= getStartOrdinal(set) && ordinal <= getEndOrdinal(set)
+}
+
 // getPodName gets the name of set's child Pod with an ordinal index of ordinal
 func getPodName(set *apps.StatefulSet, ordinal int) string {
 	return fmt.Sprintf("%s-%d", set.Name, ordinal)
@@ -170,12 +195,12 @@ func claimOwnerMatchesSetAndPod(claim *v1.PersistentVolumeClaim, set *apps.State
 		if hasOwnerRef(claim, set) {
 			return false
 		}
-		podScaledDown := getOrdinal(pod) >= int(*set.Spec.Replicas)
+		podScaledDown := !podInOrdinalRange(pod, set)
 		if podScaledDown != hasOwnerRef(claim, pod) {
 			return false
 		}
 	case policy.WhenScaled == delete && policy.WhenDeleted == delete:
-		podScaledDown := getOrdinal(pod) >= int(*set.Spec.Replicas)
+		podScaledDown := !podInOrdinalRange(pod, set)
 		// If a pod is scaled down, there should be no set ref and a pod ref;
 		// if the pod is not scaled down it's the other way around.
 		if podScaledDown == hasOwnerRef(claim, set) {
@@ -226,7 +251,7 @@ func updateClaimOwnerRefForSetAndPod(claim *v1.PersistentVolumeClaim, set *apps.
 		needsUpdate = removeOwnerRef(claim, pod) || needsUpdate
 	case policy.WhenScaled == delete && policy.WhenDeleted == retain:
 		needsUpdate = removeOwnerRef(claim, set) || needsUpdate
-		podScaledDown := getOrdinal(pod) >= int(*set.Spec.Replicas)
+		podScaledDown := !podInOrdinalRange(pod, set)
 		if podScaledDown {
 			needsUpdate = setOwnerRef(claim, pod, &podMeta) || needsUpdate
 		}
@@ -234,7 +259,7 @@ func updateClaimOwnerRefForSetAndPod(claim *v1.PersistentVolumeClaim, set *apps.
 			needsUpdate = removeOwnerRef(claim, pod) || needsUpdate
 		}
 	case policy.WhenScaled == delete && policy.WhenDeleted == delete:
-		podScaledDown := getOrdinal(pod) >= int(*set.Spec.Replicas)
+		podScaledDown := !podInOrdinalRange(pod, set)
 		if podScaledDown {
 			needsUpdate = removeOwnerRef(claim, set) || needsUpdate
 			needsUpdate = setOwnerRef(claim, pod, &podMeta) || needsUpdate
@@ -440,8 +465,8 @@ func newStatefulSetPod(set *apps.StatefulSet, ordinal int) *v1.Pod {
 // returned error is nil, the returned Pod is valid.
 func newVersionedStatefulSetPod(currentSet, updateSet *apps.StatefulSet, currentRevision, updateRevision string, ordinal int) *v1.Pod {
 	if currentSet.Spec.UpdateStrategy.Type == apps.RollingUpdateStatefulSetStrategyType &&
-		(currentSet.Spec.UpdateStrategy.RollingUpdate == nil && ordinal < int(currentSet.Status.CurrentReplicas)) ||
-		(currentSet.Spec.UpdateStrategy.RollingUpdate != nil && ordinal < int(*currentSet.Spec.UpdateStrategy.RollingUpdate.Partition)) {
+		(currentSet.Spec.UpdateStrategy.RollingUpdate == nil && ordinal < (getStartOrdinal(currentSet)+int(currentSet.Status.CurrentReplicas))) ||
+		(currentSet.Spec.UpdateStrategy.RollingUpdate != nil && ordinal < (getStartOrdinal(currentSet)+int(*currentSet.Spec.UpdateStrategy.RollingUpdate.Partition))) {
 		pod := newStatefulSetPod(currentSet, ordinal)
 		setPodRevision(pod, currentRevision)
 		return pod

pkg/features/kube_features.go

@@ -776,6 +776,12 @@ const (
 	// StatefulSetMinReadySeconds allows minReadySeconds to be respected by StatefulSet controller
 	StatefulSetMinReadySeconds featuregate.Feature = "StatefulSetMinReadySeconds"
 
+	// owner: @psch
+	// alpha: v1.26
+	//
+	// Enables a StatefulSet to start from an arbitrary non zero ordinal
+	StatefulSetStartOrdinal featuregate.Feature = "StatefulSetStartOrdinal"
+
 	// owner: @robscott
 	// kep: https://kep.k8s.io/2433
 	// alpha: v1.21
@@ -1087,6 +1093,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 
 	StatefulSetMinReadySeconds: {Default: true, PreRelease: featuregate.GA, LockToDefault: true}, // remove in 1.27
 
+	StatefulSetStartOrdinal: {Default: false, PreRelease: featuregate.Alpha},
+
 	TopologyAwareHints: {Default: true, PreRelease: featuregate.Beta},
 
 	TopologyManager: {Default: true, PreRelease: featuregate.Beta},

pkg/generated/openapi/zz_generated.openapi.go

@@ -100,6 +100,7 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
 		"k8s.io/api/apps/v1.StatefulSet":                                                                  schema_k8sio_api_apps_v1_StatefulSet(ref),
 		"k8s.io/api/apps/v1.StatefulSetCondition":                                                         schema_k8sio_api_apps_v1_StatefulSetCondition(ref),
 		"k8s.io/api/apps/v1.StatefulSetList":                                                              schema_k8sio_api_apps_v1_StatefulSetList(ref),
+		"k8s.io/api/apps/v1.StatefulSetOrdinals":                                                          schema_k8sio_api_apps_v1_StatefulSetOrdinals(ref),
 		"k8s.io/api/apps/v1.StatefulSetPersistentVolumeClaimRetentionPolicy":                              schema_k8sio_api_apps_v1_StatefulSetPersistentVolumeClaimRetentionPolicy(ref),
 		"k8s.io/api/apps/v1.StatefulSetSpec":                                                              schema_k8sio_api_apps_v1_StatefulSetSpec(ref),
 		"k8s.io/api/apps/v1.StatefulSetStatus":                                                            schema_k8sio_api_apps_v1_StatefulSetStatus(ref),
@@ -122,6 +123,7 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
 		"k8s.io/api/apps/v1beta1.StatefulSet":                                                             schema_k8sio_api_apps_v1beta1_StatefulSet(ref),
 		"k8s.io/api/apps/v1beta1.StatefulSetCondition":                                                    schema_k8sio_api_apps_v1beta1_StatefulSetCondition(ref),
 		"k8s.io/api/apps/v1beta1.StatefulSetList":                                                         schema_k8sio_api_apps_v1beta1_StatefulSetList(ref),
+		"k8s.io/api/apps/v1beta1.StatefulSetOrdinals":                                                     schema_k8sio_api_apps_v1beta1_StatefulSetOrdinals(ref),
 		"k8s.io/api/apps/v1beta1.StatefulSetPersistentVolumeClaimRetentionPolicy":                         schema_k8sio_api_apps_v1beta1_StatefulSetPersistentVolumeClaimRetentionPolicy(ref),
 		"k8s.io/api/apps/v1beta1.StatefulSetSpec":                                                         schema_k8sio_api_apps_v1beta1_StatefulSetSpec(ref),
 		"k8s.io/api/apps/v1beta1.StatefulSetStatus":                                                       schema_k8sio_api_apps_v1beta1_StatefulSetStatus(ref),
@@ -154,6 +156,7 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
 		"k8s.io/api/apps/v1beta2.StatefulSet":                                                             schema_k8sio_api_apps_v1beta2_StatefulSet(ref),
 		"k8s.io/api/apps/v1beta2.StatefulSetCondition":                                                    schema_k8sio_api_apps_v1beta2_StatefulSetCondition(ref),
 		"k8s.io/api/apps/v1beta2.StatefulSetList":                                                         schema_k8sio_api_apps_v1beta2_StatefulSetList(ref),
+		"k8s.io/api/apps/v1beta2.StatefulSetOrdinals":                                                     schema_k8sio_api_apps_v1beta2_StatefulSetOrdinals(ref),
 		"k8s.io/api/apps/v1beta2.StatefulSetPersistentVolumeClaimRetentionPolicy":                         schema_k8sio_api_apps_v1beta2_StatefulSetPersistentVolumeClaimRetentionPolicy(ref),
 		"k8s.io/api/apps/v1beta2.StatefulSetSpec":                                                         schema_k8sio_api_apps_v1beta2_StatefulSetSpec(ref),
 		"k8s.io/api/apps/v1beta2.StatefulSetStatus":                                                       schema_k8sio_api_apps_v1beta2_StatefulSetStatus(ref),
@@ -4857,6 +4860,27 @@ func schema_k8sio_api_apps_v1_StatefulSetList(ref common.ReferenceCallback) comm
 	}
 }
 
+func schema_k8sio_api_apps_v1_StatefulSetOrdinals(ref common.ReferenceCallback) common.OpenAPIDefinition {
+	return common.OpenAPIDefinition{
+		Schema: spec.Schema{
+			SchemaProps: spec.SchemaProps{
+				Description: "StatefulSetOrdinals describes the policy used for replica ordinal assignment in this StatefulSet.",
+				Type:        []string{"object"},
+				Properties: map[string]spec.Schema{
+					"start": {
+						SchemaProps: spec.SchemaProps{
+							Description: "start is the number representing the first replica's index. It may be used to number replicas from an alternate index (eg: 1-indexed) over the default 0-indexed names, or to orchestrate progressive movement of replicas from one StatefulSet to another. If set, replica indices will be in the range:\n  [.spec.ordinals.start, .spec.ordinals.start + .spec.replicas).\nIf unset, defaults to 0. Replica indices will be in the range:\n  [0, .spec.replicas).",
+							Default:     0,
+							Type:        []string{"integer"},
+							Format:      "int32",
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
 func schema_k8sio_api_apps_v1_StatefulSetPersistentVolumeClaimRetentionPolicy(ref common.ReferenceCallback) common.OpenAPIDefinition {
 	return common.OpenAPIDefinition{
 		Schema: spec.Schema{
@@ -4906,7 +4930,7 @@ func schema_k8sio_api_apps_v1_StatefulSetSpec(ref common.ReferenceCallback) comm
 					},
 					"template": {
 						SchemaProps: spec.SchemaProps{
-							Description: "template is the object that describes the pod that will be created if insufficient replicas are detected. Each pod stamped out by the StatefulSet will fulfill this Template, but have a unique identity from the rest of the StatefulSet.",
+							Description: "template is the object that describes the pod that will be created if insufficient replicas are detected. Each pod stamped out by the StatefulSet will fulfill this Template, but have a unique identity from the rest of the StatefulSet. Each pod will be named with the format <statefulsetname>-<podindex>. For example, a pod in a StatefulSet named \"web\" with index number \"3\" would be named \"web-3\".",
 							Default:     map[string]interface{}{},
 							Ref:         ref("k8s.io/api/core/v1.PodTemplateSpec"),
 						},
@@ -4967,12 +4991,18 @@ func schema_k8sio_api_apps_v1_StatefulSetSpec(ref common.ReferenceCallback) comm
 							Ref:         ref("k8s.io/api/apps/v1.StatefulSetPersistentVolumeClaimRetentionPolicy"),
 						},
 					},
+					"ordinals": {
+						SchemaProps: spec.SchemaProps{
+							Description: "ordinals controls the numbering of replica indices in a StatefulSet. The default ordinals behavior assigns a \"0\" index to the first replica and increments the index by one for each additional replica requested. Using the ordinals field requires the StatefulSetStartOrdinal feature gate to be enabled, which is alpha.",
+							Ref:         ref("k8s.io/api/apps/v1.StatefulSetOrdinals"),
+						},
+					},
 				},
 				Required: []string{"selector", "template", "serviceName"},
 			},
 		},
 		Dependencies: []string{
-			"k8s.io/api/apps/v1.StatefulSetPersistentVolumeClaimRetentionPolicy", "k8s.io/api/apps/v1.StatefulSetUpdateStrategy", "k8s.io/api/core/v1.PersistentVolumeClaim", "k8s.io/api/core/v1.PodTemplateSpec", "k8s.io/apimachinery/pkg/apis/meta/v1.LabelSelector"},
+			"k8s.io/api/apps/v1.StatefulSetOrdinals", "k8s.io/api/apps/v1.StatefulSetPersistentVolumeClaimRetentionPolicy", "k8s.io/api/apps/v1.StatefulSetUpdateStrategy", "k8s.io/api/core/v1.PersistentVolumeClaim", "k8s.io/api/core/v1.PodTemplateSpec", "k8s.io/apimachinery/pkg/apis/meta/v1.LabelSelector"},
 	}
 }
 
@@ -5965,6 +5995,27 @@ func schema_k8sio_api_apps_v1beta1_StatefulSetList(ref common.ReferenceCallback)
 	}
 }
 
+func schema_k8sio_api_apps_v1beta1_StatefulSetOrdinals(ref common.ReferenceCallback) common.OpenAPIDefinition {
+	return common.OpenAPIDefinition{
+		Schema: spec.Schema{
+			SchemaProps: spec.SchemaProps{
+				Description: "StatefulSetOrdinals describes the policy used for replica ordinal assignment in this StatefulSet.",
+				Type:        []string{"object"},
+				Properties: map[string]spec.Schema{
+					"start": {
+						SchemaProps: spec.SchemaProps{
+							Description: "start is the number representing the first replica's index. It may be used to number replicas from an alternate index (eg: 1-indexed) over the default 0-indexed names, or to orchestrate progressive movement of replicas from one StatefulSet to another. If set, replica indices will be in the range:\n  [.spec.ordinals.start, .spec.ordinals.start + .spec.replicas).\nIf unset, defaults to 0. Replica indices will be in the range:\n  [0, .spec.replicas).",
+							Default:     0,
+							Type:        []string{"integer"},
+							Format:      "int32",
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
 func schema_k8sio_api_apps_v1beta1_StatefulSetPersistentVolumeClaimRetentionPolicy(ref common.ReferenceCallback) common.OpenAPIDefinition {
 	return common.OpenAPIDefinition{
 		Schema: spec.Schema{
@@ -6014,7 +6065,7 @@ func schema_k8sio_api_apps_v1beta1_StatefulSetSpec(ref common.ReferenceCallback)
 					},
 					"template": {
 						SchemaProps: spec.SchemaProps{
-							Description: "template is the object that describes the pod that will be created if insufficient replicas are detected. Each pod stamped out by the StatefulSet will fulfill this Template, but have a unique identity from the rest of the StatefulSet.",
+							Description: "template is the object that describes the pod that will be created if insufficient replicas are detected. Each pod stamped out by the StatefulSet will fulfill this Template, but have a unique identity from the rest of the StatefulSet. Each pod will be named with the format <statefulsetname>-<podindex>. For example, a pod in a StatefulSet named \"web\" with index number \"3\" would be named \"web-3\".",
 							Default:     map[string]interface{}{},
 							Ref:         ref("k8s.io/api/core/v1.PodTemplateSpec"),
 						},
@@ -6075,12 +6126,18 @@ func schema_k8sio_api_apps_v1beta1_StatefulSetSpec(ref common.ReferenceCallback)
 							Ref:         ref("k8s.io/api/apps/v1beta1.StatefulSetPersistentVolumeClaimRetentionPolicy"),
 						},
 					},
+					"ordinals": {
+						SchemaProps: spec.SchemaProps{
+							Description: "ordinals controls the numbering of replica indices in a StatefulSet. The default ordinals behavior assigns a \"0\" index to the first replica and increments the index by one for each additional replica requested. Using the ordinals field requires the StatefulSetStartOrdinal feature gate to be enabled, which is alpha.",
+							Ref:         ref("k8s.io/api/apps/v1beta1.StatefulSetOrdinals"),
+						},
+					},
 				},
 				Required: []string{"template", "serviceName"},
 			},
 		},
 		Dependencies: []string{
-			"k8s.io/api/apps/v1beta1.StatefulSetPersistentVolumeClaimRetentionPolicy", "k8s.io/api/apps/v1beta1.StatefulSetUpdateStrategy", "k8s.io/api/core/v1.PersistentVolumeClaim", "k8s.io/api/core/v1.PodTemplateSpec", "k8s.io/apimachinery/pkg/apis/meta/v1.LabelSelector"},
+			"k8s.io/api/apps/v1beta1.StatefulSetOrdinals", "k8s.io/api/apps/v1beta1.StatefulSetPersistentVolumeClaimRetentionPolicy", "k8s.io/api/apps/v1beta1.StatefulSetUpdateStrategy", "k8s.io/api/core/v1.PersistentVolumeClaim", "k8s.io/api/core/v1.PodTemplateSpec", "k8s.io/apimachinery/pkg/apis/meta/v1.LabelSelector"},
 	}
 }
 
@@ -7622,6 +7679,27 @@ func schema_k8sio_api_apps_v1beta2_StatefulSetList(ref common.ReferenceCallback)
 	}
 }
 
+func schema_k8sio_api_apps_v1beta2_StatefulSetOrdinals(ref common.ReferenceCallback) common.OpenAPIDefinition {
+	return common.OpenAPIDefinition{
+		Schema: spec.Schema{
+			SchemaProps: spec.SchemaProps{
+				Description: "StatefulSetOrdinals describes the policy used for replica ordinal assignment in this StatefulSet.",
+				Type:        []string{"object"},
+				Properties: map[string]spec.Schema{
+					"start": {
+						SchemaProps: spec.SchemaProps{
+							Description: "start is the number representing the first replica's index. It may be used to number replicas from an alternate index (eg: 1-indexed) over the default 0-indexed names, or to orchestrate progressive movement of replicas from one StatefulSet to another. If set, replica indices will be in the range:\n  [.spec.ordinals.start, .spec.ordinals.start + .spec.replicas).\nIf unset, defaults to 0. Replica indices will be in the range:\n  [0, .spec.replicas).",
+							Default:     0,
+							Type:        []string{"integer"},
+							Format:      "int32",
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
 func schema_k8sio_api_apps_v1beta2_StatefulSetPersistentVolumeClaimRetentionPolicy(ref common.ReferenceCallback) common.OpenAPIDefinition {
 	return common.OpenAPIDefinition{
 		Schema: spec.Schema{
@@ -7671,7 +7749,7 @@ func schema_k8sio_api_apps_v1beta2_StatefulSetSpec(ref common.ReferenceCallback)
 					},
 					"template": {
 						SchemaProps: spec.SchemaProps{
-							Description: "template is the object that describes the pod that will be created if insufficient replicas are detected. Each pod stamped out by the StatefulSet will fulfill this Template, but have a unique identity from the rest of the StatefulSet.",
+							Description: "template is the object that describes the pod that will be created if insufficient replicas are detected. Each pod stamped out by the StatefulSet will fulfill this Template, but have a unique identity from the rest of the StatefulSet. Each pod will be named with the format <statefulsetname>-<podindex>. For example, a pod in a StatefulSet named \"web\" with index number \"3\" would be named \"web-3\".",
 							Default:     map[string]interface{}{},
 							Ref:         ref("k8s.io/api/core/v1.PodTemplateSpec"),
 						},
@@ -7732,12 +7810,18 @@ func schema_k8sio_api_apps_v1beta2_StatefulSetSpec(ref common.ReferenceCallback)
 							Ref:         ref("k8s.io/api/apps/v1beta2.StatefulSetPersistentVolumeClaimRetentionPolicy"),
 						},
 					},
+					"ordinals": {
+						SchemaProps: spec.SchemaProps{
+							Description: "ordinals controls the numbering of replica indices in a StatefulSet. The default ordinals behavior assigns a \"0\" index to the first replica and increments the index by one for each additional replica requested. Using the ordinals field requires the StatefulSetStartOrdinal feature gate to be enabled, which is alpha.",
+							Ref:         ref("k8s.io/api/apps/v1beta2.StatefulSetOrdinals"),
+						},
+					},
 				},
 				Required: []string{"selector", "template", "serviceName"},
 			},
 		},
 		Dependencies: []string{
-			"k8s.io/api/apps/v1beta2.StatefulSetPersistentVolumeClaimRetentionPolicy", "k8s.io/api/apps/v1beta2.StatefulSetUpdateStrategy", "k8s.io/api/core/v1.PersistentVolumeClaim", "k8s.io/api/core/v1.PodTemplateSpec", "k8s.io/apimachinery/pkg/apis/meta/v1.LabelSelector"},
+			"k8s.io/api/apps/v1beta2.StatefulSetOrdinals", "k8s.io/api/apps/v1beta2.StatefulSetPersistentVolumeClaimRetentionPolicy", "k8s.io/api/apps/v1beta2.StatefulSetUpdateStrategy", "k8s.io/api/core/v1.PersistentVolumeClaim", "k8s.io/api/core/v1.PodTemplateSpec", "k8s.io/apimachinery/pkg/apis/meta/v1.LabelSelector"},
 	}
 }
 

pkg/registry/apps/statefulset/strategy.go

@@ -128,6 +128,12 @@ func dropStatefulSetDisabledFields(newSS *apps.StatefulSet, oldSS *apps.Stateful
 			newSS.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable = nil
 		}
 	}
+	if !utilfeature.DefaultFeatureGate.Enabled(features.StatefulSetStartOrdinal) {
+		if oldSS == nil || oldSS.Spec.Ordinals == nil {
+			// Reset Spec.Ordinals to the default value (nil).
+			newSS.Spec.Ordinals = nil
+		}
+	}
 }
 
 // Validate validates a new StatefulSet.

pkg/registry/apps/statefulset/strategy_test.go

@@ -328,14 +328,29 @@ func getMaxUnavailable(maxUnavailable int) *int {
 	return &maxUnavailable
 }
 
+func createOrdinalsWithStart(start int) *apps.StatefulSetOrdinals {
+	return &apps.StatefulSetOrdinals{
+		Start: int32(start),
+	}
+}
+
+func makeStatefulSetWithStatefulSetOrdinals(ordinals *apps.StatefulSetOrdinals) *apps.StatefulSet {
+	return &apps.StatefulSet{
+		Spec: apps.StatefulSetSpec{
+			Ordinals: ordinals,
+		},
+	}
+}
+
 // TestDropStatefulSetDisabledFields tests if the drop functionality is working fine or not
 func TestDropStatefulSetDisabledFields(t *testing.T) {
 	testCases := []struct {
-		name                 string
-		enableMaxUnavailable bool
-		ss                   *apps.StatefulSet
-		oldSS                *apps.StatefulSet
-		expectedSS           *apps.StatefulSet
+		name                          string
+		enableMaxUnavailable          bool
+		enableStatefulSetStartOrdinal bool
+		ss                            *apps.StatefulSet
+		oldSS                         *apps.StatefulSet
+		expectedSS                    *apps.StatefulSet
 	}{
 		{
 			name:       "set minReadySeconds, no update",
@@ -388,11 +403,39 @@ func TestDropStatefulSetDisabledFields(t *testing.T) {
 			ss:                   makeStatefulSetWithMaxUnavailable(getMaxUnavailable(1)),
 			oldSS:                makeStatefulSetWithMaxUnavailable(getMaxUnavailable(3)),
 			expectedSS:           makeStatefulSetWithMaxUnavailable(getMaxUnavailable(1)),
+		}, {
+			name:                          "StatefulSetStartOrdinal disabled, ordinals in use in new only",
+			enableStatefulSetStartOrdinal: false,
+			ss:                            makeStatefulSetWithStatefulSetOrdinals(createOrdinalsWithStart(2)),
+			oldSS:                         nil,
+			expectedSS:                    makeStatefulSetWithStatefulSetOrdinals(nil),
+		},
+		{
+			name:                          "StatefulSetStartOrdinal disabled, ordinals in use in both old and new",
+			enableStatefulSetStartOrdinal: false,
+			ss:                            makeStatefulSetWithStatefulSetOrdinals(createOrdinalsWithStart(2)),
+			oldSS:                         makeStatefulSetWithStatefulSetOrdinals(createOrdinalsWithStart(1)),
+			expectedSS:                    makeStatefulSetWithStatefulSetOrdinals(createOrdinalsWithStart(2)),
+		},
+		{
+			name:                          "StatefulSetStartOrdinal enabled, ordinals in use in new only",
+			enableStatefulSetStartOrdinal: true,
+			ss:                            makeStatefulSetWithStatefulSetOrdinals(createOrdinalsWithStart(2)),
+			oldSS:                         nil,
+			expectedSS:                    makeStatefulSetWithStatefulSetOrdinals(createOrdinalsWithStart(2)),
+		},
+		{
+			name:                          "StatefulSetStartOrdinal enabled, ordinals in use in both old and new",
+			enableStatefulSetStartOrdinal: true,
+			ss:                            makeStatefulSetWithStatefulSetOrdinals(createOrdinalsWithStart(2)),
+			oldSS:                         makeStatefulSetWithStatefulSetOrdinals(createOrdinalsWithStart(1)),
+			expectedSS:                    makeStatefulSetWithStatefulSetOrdinals(createOrdinalsWithStart(2)),
 		},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.MaxUnavailableStatefulSet, tc.enableMaxUnavailable)()
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.StatefulSetStartOrdinal, tc.enableStatefulSetStartOrdinal)()
 			old := tc.oldSS.DeepCopy()
 
 			dropStatefulSetDisabledFields(tc.ss, tc.oldSS)