Add emulation forward compatibility into api enablement and RemoveDeletedKinds.

Signed-off-by: Siyuan Zhang <sizhang@google.com>

cmd/kube-apiserver/app/options/options_test.go

@@ -128,6 +128,7 @@ func TestAddFlags(t *testing.T) {
 		"--service-cluster-ip-range=192.168.128.0/17",
 		"--lease-reuse-duration-seconds=100",
 		"--emulated-version=test=1.31",
+		"--emulation-forward-compatible=true",
 	}
 	fs.Parse(args)
 	utilruntime.Must(componentGlobalsRegistry.Set())
@@ -147,6 +148,7 @@ func TestAddFlags(t *testing.T) {
 				MaxRequestBodyBytes:          int64(3 * 1024 * 1024),
 				ComponentGlobalsRegistry:     componentGlobalsRegistry,
 				ComponentName:                basecompatibility.DefaultKubeComponent,
+				EmulationForwardCompatible:   true,
 			},
 			Admission: &kubeoptions.AdmissionOptions{
 				GenericAdmission: &apiserveroptions.AdmissionOptions{

pkg/controlplane/apiserver/apis.go

@@ -89,7 +89,7 @@ func (s *Server) InstallAPIs(restStorageProviders ...RESTStorageProvider) error
 	nonLegacy := []*genericapiserver.APIGroupInfo{}
 
 	// used later in the loop to filter the served resource by those that have expired.
-	resourceExpirationEvaluator, err := genericapiserver.NewResourceExpirationEvaluator(s.GenericAPIServer.EffectiveVersion.EmulationVersion())
+	resourceExpirationEvaluator, err := genericapiserver.NewResourceExpirationEvaluator(s.GenericAPIServer.EffectiveVersion.EmulationVersion(), s.GenericAPIServer.EmulationForwardCompatible)
 	if err != nil {
 		return err
 	}
@@ -111,6 +111,7 @@ func (s *Server) InstallAPIs(restStorageProviders ...RESTStorageProvider) error
 		// We do this here so that we don't accidentally serve versions without resources or openapi information that for kinds we don't serve.
 		// This is a spot above the construction of individual storage handlers so that no sig accidentally forgets to check.
 		resourceExpirationEvaluator.RemoveDeletedKinds(groupName, apiGroupInfo.Scheme, apiGroupInfo.VersionedResourcesStorageMap)
+		resourceExpirationEvaluator.RemoveUnIntroducedKinds(groupName, apiGroupInfo.Scheme, apiGroupInfo.VersionedResourcesStorageMap)
 		if len(apiGroupInfo.VersionedResourcesStorageMap) == 0 {
 			klog.V(1).Infof("Removing API group %v because it is time to stop serving it because it has no versions per APILifecycle.", groupName)
 			continue

staging/src/k8s.io/apimachinery/pkg/runtime/interfaces.go

@@ -259,6 +259,7 @@ type ObjectDefaulter interface {
 
 type ObjectVersioner interface {
 	ConvertToVersion(in Object, gv GroupVersioner) (out Object, err error)
+	PrioritizedVersionsForGroup(group string) []schema.GroupVersion
 }
 
 // ObjectConvertor converts an object to a different version.

staging/src/k8s.io/apiserver/pkg/server/config.go

@@ -154,6 +154,10 @@ type Config struct {
 	// EffectiveVersion determines which apis and features are available
 	// based on when the api/feature lifecyle.
 	EffectiveVersion basecompatibility.EffectiveVersion
+	// EmulationForwardCompatible indicates APIs introduced after the emulation version are installed.
+	// If true, APIs that have higher priority than the APIs of the same group resource enabled at the emulation version will be installed.
+	// This is useful if a controller has switched to use newer APIs in the binary version, and we want it still functional in an older emulation version.
+	EmulationForwardCompatible bool
 	// FeatureGate is a way to plumb feature gate through if you have them.
 	FeatureGate featuregate.FeatureGate
 	// AuditBackend is where audit events are sent to.
@@ -839,8 +843,9 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
 		StorageReadinessHook:  NewStorageReadinessHook(c.StorageInitializationTimeout),
 		StorageVersionManager: c.StorageVersionManager,
 
-		EffectiveVersion: c.EffectiveVersion,
+		EffectiveVersion:           c.EffectiveVersion,
-		FeatureGate:      c.FeatureGate,
+		EmulationForwardCompatible: c.EmulationForwardCompatible,
+		FeatureGate:                c.FeatureGate,
 
 		muxAndDiscoveryCompleteSignals: map[string]<-chan struct{}{},
 	}

staging/src/k8s.io/apiserver/pkg/server/deleted_kinds.go

@@ -19,6 +19,7 @@ package server
 import (
 	"fmt"
 	"os"
+	"regexp"
 	"strconv"
 	"strings"
 
@@ -31,10 +32,13 @@ import (
 	"k8s.io/klog/v2"
 )
 
+var alphaPattern = regexp.MustCompile(`^v\d+alpha\d+$`)
+
 // resourceExpirationEvaluator holds info for deciding if a particular rest.Storage needs to excluded from the API
 type resourceExpirationEvaluator struct {
-	currentVersion *apimachineryversion.Version
+	currentVersion             *apimachineryversion.Version
-	isAlpha        bool
+	emulationForwardCompatible bool
+	isAlpha                    bool
 	// Special flag checking for the existence of alpha.0
 	// alpha.0 is a special case where everything merged to master is auto propagated to the release-1.n branch
 	isAlphaZero bool
@@ -53,17 +57,21 @@ type ResourceExpirationEvaluator interface {
 	// RemoveDeletedKinds inspects the storage map and modifies it in place by removing storage for kinds that have been deleted.
 	// versionedResourcesStorageMap mirrors the field on APIGroupInfo, it's a map from version to resource to the storage.
 	RemoveDeletedKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage)
+	// RemoveUnIntroducedKinds inspects the storage map and modifies it in place by removing storage for kinds that are introduced after the current version.
+	// versionedResourcesStorageMap mirrors the field on APIGroupInfo, it's a map from version to resource to the storage.
+	RemoveUnIntroducedKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage)
 	// ShouldServeForVersion returns true if a particular version cut off is after the current version
 	ShouldServeForVersion(majorRemoved, minorRemoved int) bool
 }
 
-func NewResourceExpirationEvaluator(currentVersion *apimachineryversion.Version) (ResourceExpirationEvaluator, error) {
+func NewResourceExpirationEvaluator(currentVersion *apimachineryversion.Version, emulationForwardCompatible bool) (ResourceExpirationEvaluator, error) {
 	if currentVersion == nil {
 		return nil, fmt.Errorf("empty NewResourceExpirationEvaluator currentVersion")
 	}
 	klog.V(1).Infof("NewResourceExpirationEvaluator with currentVersion: %s.", currentVersion)
 	ret := &resourceExpirationEvaluator{
 		strictRemovedHandlingInAlpha: false,
+		emulationForwardCompatible:   emulationForwardCompatible,
 	}
 	// Only keeps the major and minor versions from input version.
 	ret.currentVersion = apimachineryversion.MajorMinor(currentVersion.Major(), currentVersion.Minor())
@@ -89,7 +97,7 @@ func NewResourceExpirationEvaluator(currentVersion *apimachineryversion.Version)
 	return ret, nil
 }
 
-func (e *resourceExpirationEvaluator) shouldServe(gv schema.GroupVersion, versioner runtime.ObjectVersioner, resourceServingInfo rest.Storage) bool {
+func (e *resourceExpirationEvaluator) isNotRemoved(gv schema.GroupVersion, versioner runtime.ObjectVersioner, resourceServingInfo rest.Storage) bool {
 	internalPtr := resourceServingInfo.New()
 
 	target := gv
@@ -104,15 +112,6 @@ func (e *resourceExpirationEvaluator) shouldServe(gv schema.GroupVersion, versio
 		return false
 	}
 
-	introduced, ok := versionedPtr.(introducedInterface)
-	if ok {
-		majorIntroduced, minorIntroduced := introduced.APILifecycleIntroduced()
-		verIntroduced := apimachineryversion.MajorMinor(uint(majorIntroduced), uint(minorIntroduced))
-		if e.currentVersion.LessThan(verIntroduced) {
-			return false
-		}
-	}
-
 	removed, ok := versionedPtr.(removedInterface)
 	if !ok {
 		return true
@@ -153,7 +152,7 @@ type introducedInterface interface {
 	APILifecycleIntroduced() (major, minor int)
 }
 
-// removeDeletedKinds inspects the storage map and modifies it in place by removing storage for kinds that have been deleted.
+// RemoveDeletedKinds inspects the storage map and modifies it in place by removing storage for kinds that have been deleted.
 // versionedResourcesStorageMap mirrors the field on APIGroupInfo, it's a map from version to resource to the storage.
 func (e *resourceExpirationEvaluator) RemoveDeletedKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage) {
 	versionsToRemove := sets.NewString()
@@ -161,7 +160,7 @@ func (e *resourceExpirationEvaluator) RemoveDeletedKinds(groupName string, versi
 		versionToResource := versionedResourcesStorageMap[apiVersion]
 		resourcesToRemove := sets.NewString()
 		for resourceName, resourceServingInfo := range versionToResource {
-			if !e.shouldServe(schema.GroupVersion{Group: groupName, Version: apiVersion}, versioner, resourceServingInfo) {
+			if !e.isNotRemoved(schema.GroupVersion{Group: groupName, Version: apiVersion}, versioner, resourceServingInfo) {
 				resourcesToRemove.Insert(resourceName)
 			}
 		}
@@ -189,6 +188,82 @@ func (e *resourceExpirationEvaluator) RemoveDeletedKinds(groupName string, versi
 	}
 }
 
+// RemoveUnIntroducedKinds inspects the storage map and modifies it in place by removing storage for kinds that are introduced after the current version.
+// versionedResourcesStorageMap mirrors the field on APIGroupInfo, it's a map from version to resource to the storage.
+func (e *resourceExpirationEvaluator) RemoveUnIntroducedKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage) {
+	versionsToRemove := sets.NewString()
+	prioritizedVersions := versioner.PrioritizedVersionsForGroup(groupName)
+	enabledResources := sets.NewString()
+
+	// iterate from the end to the front, so that we remove the older versions first.
+	for i := len(prioritizedVersions) - 1; i >= 0; i-- {
+		apiVersion := prioritizedVersions[i].Version
+		versionToResource := versionedResourcesStorageMap[apiVersion]
+		resourcesToRemove := sets.NewString()
+		for resourceName, resourceServingInfo := range versionToResource {
+			// if an earlier version of the resource has been enabled, the same resource with higher priority
+			// should also be enabled if emulationForwardCompatible.
+			if e.emulationForwardCompatible && enabledResources.Has(resourceName) {
+				continue
+			}
+			verIntroduced := versionIntroduced(schema.GroupVersion{Group: groupName, Version: apiVersion}, versioner, resourceServingInfo)
+			if e.currentVersion.LessThan(verIntroduced) {
+				resourcesToRemove.Insert(resourceName)
+			} else {
+				// emulation forward compatibility is not applicable to alpha apis.
+				if !alphaPattern.MatchString(apiVersion) {
+					enabledResources.Insert(resourceName)
+				}
+			}
+		}
+
+		for resourceName := range versionedResourcesStorageMap[apiVersion] {
+			if !shouldRemoveResourceAndSubresources(resourcesToRemove, resourceName) {
+				continue
+			}
+
+			klog.V(1).Infof("Removing resource %v.%v.%v because it is introduced after the current version %s per APILifecycle.", resourceName, apiVersion, groupName, e.currentVersion.String())
+			storage := versionToResource[resourceName]
+			storage.Destroy()
+			delete(versionToResource, resourceName)
+		}
+		versionedResourcesStorageMap[apiVersion] = versionToResource
+
+		if len(versionedResourcesStorageMap[apiVersion]) == 0 {
+			versionsToRemove.Insert(apiVersion)
+		}
+	}
+
+	for _, apiVersion := range versionsToRemove.List() {
+		klog.V(1).Infof("Removing version %v.%v because it is introduced after the current version %s and because it has no resources per APILifecycle.", apiVersion, groupName, e.currentVersion.String())
+		delete(versionedResourcesStorageMap, apiVersion)
+	}
+}
+
+func versionIntroduced(gv schema.GroupVersion, versioner runtime.ObjectVersioner, resourceServingInfo rest.Storage) *apimachineryversion.Version {
+	defaultVer := apimachineryversion.MajorMinor(0, 0)
+	internalPtr := resourceServingInfo.New()
+
+	target := gv
+	// honor storage that overrides group version (used for things like scale subresources)
+	if versionProvider, ok := resourceServingInfo.(rest.GroupVersionKindProvider); ok {
+		target = versionProvider.GroupVersionKind(target).GroupVersion()
+	}
+
+	versionedPtr, err := versioner.ConvertToVersion(internalPtr, target)
+	if err != nil {
+		utilruntime.HandleError(err)
+		return defaultVer
+	}
+
+	introduced, ok := versionedPtr.(introducedInterface)
+	if ok {
+		majorIntroduced, minorIntroduced := introduced.APILifecycleIntroduced()
+		return apimachineryversion.MajorMinor(uint(majorIntroduced), uint(minorIntroduced))
+	}
+	return defaultVer
+}
+
 func shouldRemoveResourceAndSubresources(resourcesToRemove sets.String, resourceName string) bool {
 	for _, resourceToRemove := range resourcesToRemove.List() {
 		if resourceName == resourceToRemove {

staging/src/k8s.io/apiserver/pkg/server/deleted_kinds_test.go

@@ -65,7 +65,7 @@ func Test_newResourceExpirationEvaluator(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			actual, actualErr := NewResourceExpirationEvaluator(apimachineryversion.MustParse(tt.currentVersion))
+			actual, actualErr := NewResourceExpirationEvaluator(apimachineryversion.MustParse(tt.currentVersion), false)
 
 			checkErr(t, actualErr, tt.expectedErr)
 			if actualErr != nil {
@@ -80,29 +80,6 @@ func Test_newResourceExpirationEvaluator(t *testing.T) {
 	}
 }
 
-func storageRemovedIn(major, minor int) *removedInStorage {
-	return &removedInStorage{major: major, minor: minor}
-}
-
-func storageNeverRemoved() *removedInStorage {
-	return &removedInStorage{neverRemoved: true}
-}
-
-type removedInStorage struct {
-	major, minor int
-	neverRemoved bool
-}
-
-func (r *removedInStorage) New() runtime.Object {
-	if r.neverRemoved {
-		return &defaultObj{}
-	}
-	return &removedInObj{major: r.major, minor: r.minor}
-}
-
-func (r *removedInStorage) Destroy() {
-}
-
 type defaultObj struct {
 }
 
@@ -127,24 +104,6 @@ func (r *removedInObj) APILifecycleRemoved() (major, minor int) {
 	return r.major, r.minor
 }
 
-func storageIntroducedIn(major, minor int) *introducedInStorage {
-	return &introducedInStorage{major: major, minor: minor}
-}
-
-type introducedInStorage struct {
-	major, minor int
-}
-
-func (r *introducedInStorage) New() runtime.Object {
-	if r.major == 0 && r.minor == 0 {
-		return &defaultObj{}
-	}
-	return &IntroducedInObj{major: r.major, minor: r.minor}
-}
-
-func (r *introducedInStorage) Destroy() {
-}
-
 type IntroducedInObj struct {
 	major, minor int
 }
@@ -159,7 +118,61 @@ func (r *IntroducedInObj) APILifecycleIntroduced() (major, minor int) {
 	return r.major, r.minor
 }
 
-func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
+type introducedAndRemovedInObj struct {
+	majorIntroduced, minorIntroduced int
+	majorRemoved, minorRemoved       int
+}
+
+func (r *introducedAndRemovedInObj) GetObjectKind() schema.ObjectKind {
+	panic("don't do this")
+}
+func (r *introducedAndRemovedInObj) DeepCopyObject() runtime.Object {
+	panic("don't do this either")
+}
+func (r *introducedAndRemovedInObj) APILifecycleIntroduced() (major, minor int) {
+	return r.majorIntroduced, r.minorIntroduced
+}
+func (r *introducedAndRemovedInObj) APILifecycleRemoved() (major, minor int) {
+	return r.majorRemoved, r.minorRemoved
+}
+
+func storageRemovedIn(major, minor int) *introducedAndRemovedInStorage {
+	return &introducedAndRemovedInStorage{majorRemoved: major, minorRemoved: minor}
+}
+
+func storageNeverRemoved() *introducedAndRemovedInStorage {
+	return &introducedAndRemovedInStorage{}
+}
+
+func storageIntroducedIn(major, minor int) *introducedAndRemovedInStorage {
+	return &introducedAndRemovedInStorage{majorIntroduced: major, minorIntroduced: minor}
+}
+
+func storageIntroducedAndRemovedIn(majorIntroduced, minorIntroduced, majorRemoved, minorRemoved int) *introducedAndRemovedInStorage {
+	return &introducedAndRemovedInStorage{majorIntroduced: majorIntroduced, minorIntroduced: minorIntroduced, majorRemoved: majorRemoved, minorRemoved: minorRemoved}
+}
+
+type introducedAndRemovedInStorage struct {
+	majorIntroduced, minorIntroduced int
+	majorRemoved, minorRemoved       int
+}
+
+func (r *introducedAndRemovedInStorage) New() runtime.Object {
+	if r.majorIntroduced == 0 && r.minorIntroduced == 0 && r.majorRemoved == 0 && r.minorRemoved == 0 {
+		return &defaultObj{}
+	}
+	if r.majorIntroduced == 0 && r.minorIntroduced == 0 {
+		return &removedInObj{major: r.majorRemoved, minor: r.minorRemoved}
+	}
+	if r.majorRemoved == 0 && r.minorRemoved == 0 {
+		return &IntroducedInObj{major: r.majorIntroduced, minor: r.minorIntroduced}
+	}
+	return &introducedAndRemovedInObj{majorIntroduced: r.majorIntroduced, minorIntroduced: r.minorIntroduced, majorRemoved: r.majorRemoved, minorRemoved: r.minorRemoved}
+}
+
+func (r *introducedAndRemovedInStorage) Destroy() {}
+
+func Test_resourceExpirationEvaluator_isNotRemoved(t *testing.T) {
 	tests := []struct {
 		name                        string
 		resourceExpirationEvaluator resourceExpirationEvaluator
@@ -247,45 +260,13 @@ func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
 			restStorage: storageNeverRemoved(),
 			expected:    true,
 		},
-		{
-			name: "introduced-in-curr",
-			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentVersion: apimachineryversion.MajorMinor(1, 20),
-			},
-			restStorage: storageIntroducedIn(1, 20),
-			expected:    true,
-		},
-		{
-			name: "introduced-in-prev-major",
-			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentVersion: apimachineryversion.MajorMinor(1, 20),
-			},
-			restStorage: storageIntroducedIn(1, 19),
-			expected:    true,
-		},
-		{
-			name: "introduced-in-future",
-			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentVersion: apimachineryversion.MajorMinor(1, 20),
-			},
-			restStorage: storageIntroducedIn(1, 21),
-			expected:    false,
-		},
-		{
-			name: "missing-introduced",
-			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentVersion: apimachineryversion.MajorMinor(1, 20),
-			},
-			restStorage: storageIntroducedIn(0, 0),
-			expected:    true,
-		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			gv := schema.GroupVersion{Group: "mygroup", Version: "myversion"}
-			convertor := &dummyConvertor{}
+			convertor := &dummyConvertor{prioritizedVersions: []schema.GroupVersion{gv}}
-			if actual := tt.resourceExpirationEvaluator.shouldServe(gv, convertor, tt.restStorage); actual != tt.expected {
+			if actual := tt.resourceExpirationEvaluator.isNotRemoved(gv, convertor, tt.restStorage); actual != tt.expected {
-				t.Errorf("shouldServe() = %v, want %v", actual, tt.expected)
+				t.Errorf("isRemoved() = %v, want %v", actual, tt.expected)
 			}
 			if !reflect.DeepEqual(convertor.called, gv) {
 				t.Errorf("expected converter to be called with %#v, got %#v", gv, convertor.called)
@@ -295,7 +276,8 @@ func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
 }
 
 type dummyConvertor struct {
-	called runtime.GroupVersioner
+	called              runtime.GroupVersioner
+	prioritizedVersions []schema.GroupVersion
 }
 
 func (d *dummyConvertor) ConvertToVersion(in runtime.Object, gv runtime.GroupVersioner) (runtime.Object, error) {
@@ -303,6 +285,10 @@ func (d *dummyConvertor) ConvertToVersion(in runtime.Object, gv runtime.GroupVer
 	return in, nil
 }
 
+func (d *dummyConvertor) PrioritizedVersionsForGroup(group string) []schema.GroupVersion {
+	return d.prioritizedVersions
+}
+
 func checkErr(t *testing.T, actual error, expected string) {
 	t.Helper()
 	switch {
@@ -317,6 +303,7 @@ func checkErr(t *testing.T, actual error, expected string) {
 }
 
 func Test_removeDeletedKinds(t *testing.T) {
+	groupName := "group.name"
 	tests := []struct {
 		name                         string
 		resourceExpirationEvaluator  resourceExpirationEvaluator
@@ -381,8 +368,234 @@ func Test_removeDeletedKinds(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			convertor := &dummyConvertor{}
+			convertor := &dummyConvertor{prioritizedVersions: []schema.GroupVersion{
-			tt.resourceExpirationEvaluator.RemoveDeletedKinds("group.name", convertor, tt.versionedResourcesStorageMap)
+				{Group: groupName, Version: "v2"}, {Group: groupName, Version: "v1"}}}
+			tt.resourceExpirationEvaluator.RemoveDeletedKinds(groupName, convertor, tt.versionedResourcesStorageMap)
+			if !reflect.DeepEqual(tt.expectedStorage, tt.versionedResourcesStorageMap) {
+				t.Fatal(dump.Pretty(tt.versionedResourcesStorageMap))
+			}
+		})
+	}
+}
+
+func Test_removeUnIntroducedKinds(t *testing.T) {
+	groupName := "group.name"
+	resource1 := "resource1"
+	resource2 := "resource2"
+	tests := []struct {
+		name                         string
+		resourceExpirationEvaluator  resourceExpirationEvaluator
+		versionedResourcesStorageMap map[string]map[string]rest.Storage
+		expectedStorage              map[string]map[string]rest.Storage
+	}{
+		{
+			name: "remove-future-version",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2alpha1": {
+					resource1: storageIntroducedAndRemovedIn(1, 20, 1, 21),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2alpha1": {
+					resource1: storageIntroducedAndRemovedIn(1, 20, 1, 21),
+				},
+			},
+		},
+		{
+			name: "missing-introduced-version",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageRemovedIn(1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageRemovedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-ga-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 19),
+				emulationForwardCompatible: true,
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2alpha1": {
+					resource1: storageIntroducedAndRemovedIn(1, 20, 1, 21),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-alpha-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 20),
+				emulationForwardCompatible: true,
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2alpha1": {
+					resource1: storageIntroducedAndRemovedIn(1, 20, 1, 21),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2alpha1": {
+					resource1: storageIntroducedAndRemovedIn(1, 20, 1, 21),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-beta1-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 21),
+				emulationForwardCompatible: true,
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-new-resource",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 22),
+				emulationForwardCompatible: true,
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			convertor := &dummyConvertor{prioritizedVersions: []schema.GroupVersion{
+				{Group: groupName, Version: "v2"}, {Group: groupName, Version: "v1"},
+				{Group: groupName, Version: "v2beta2"}, {Group: groupName, Version: "v2beta1"},
+				{Group: groupName, Version: "v2alpha1"}}}
+			tt.resourceExpirationEvaluator.RemoveUnIntroducedKinds(groupName, convertor, tt.versionedResourcesStorageMap)
 			if !reflect.DeepEqual(tt.expectedStorage, tt.versionedResourcesStorageMap) {
 				t.Fatal(dump.Pretty(tt.versionedResourcesStorageMap))
 			}

staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go

@@ -245,6 +245,11 @@ type GenericAPIServer struct {
 	// EffectiveVersion determines which apis and features are available
 	// based on when the api/feature lifecyle.
 	EffectiveVersion basecompatibility.EffectiveVersion
+	// EmulationForwardCompatible indicates APIs introduced after the emulation version are installed.
+	// If true, APIs that have higher priority than the APIs of the same group resource enabled at the emulation version will be installed.
+	// This is useful if a controller has switched to use newer APIs in the binary version, and we want it still functional in an older emulation version.
+	EmulationForwardCompatible bool
+
 	// FeatureGate is a way to plumb feature gate through if you have them.
 	FeatureGate featuregate.FeatureGate
 

staging/src/k8s.io/apiserver/pkg/server/options/api_enablement.go

@@ -95,6 +95,14 @@ func (s *APIEnablementOptions) ApplyTo(c *server.Config, defaultResourceConfig *
 	}
 
 	mergedResourceConfig, err := resourceconfig.MergeAPIResourceConfigs(defaultResourceConfig, s.RuntimeConfig, registry)
+	if err != nil {
+		return err
+	}
+	// apply emulation forward compatibility to the api enablement if applicable.
+	if c.EmulationForwardCompatible {
+		mergedResourceConfig, err = resourceconfig.EmulationForwardCompatibleResourceConfig(mergedResourceConfig, s.RuntimeConfig, registry)
+	}
+
 	c.MergedResourceConfig = mergedResourceConfig
 
 	return err

staging/src/k8s.io/apiserver/pkg/server/options/server_run_options.go

@@ -98,6 +98,10 @@ type ServerRunOptions struct {
 	ComponentGlobalsRegistry basecompatibility.ComponentGlobalsRegistry
 	// ComponentName is name under which the server's global variabled are registered in the ComponentGlobalsRegistry.
 	ComponentName string
+	// EmulationForwardCompatible indicates APIs introduced after the emulation version are installed.
+	// If true, APIs that have higher priority than the APIs of the same group resource enabled at the emulation version will be installed.
+	// This is useful if a controller has switched to use newer APIs in the binary version, and we want it still functional in an older emulation version.
+	EmulationForwardCompatible bool
 }
 
 func NewServerRunOptions() *ServerRunOptions {
@@ -152,6 +156,7 @@ func (s *ServerRunOptions) ApplyTo(c *server.Config) error {
 	c.ShutdownWatchTerminationGracePeriod = s.ShutdownWatchTerminationGracePeriod
 	c.EffectiveVersion = s.ComponentGlobalsRegistry.EffectiveVersionFor(s.ComponentName)
 	c.FeatureGate = s.ComponentGlobalsRegistry.FeatureGateFor(s.ComponentName)
+	c.EmulationForwardCompatible = s.EmulationForwardCompatible
 
 	return nil
 }
@@ -231,6 +236,12 @@ func (s *ServerRunOptions) Validate() []error {
 	if errs := s.ComponentGlobalsRegistry.Validate(); len(errs) != 0 {
 		errors = append(errors, errs...)
 	}
+	if s.EmulationForwardCompatible {
+		effectiveVersion := s.ComponentGlobalsRegistry.EffectiveVersionFor(s.ComponentName)
+		if effectiveVersion.BinaryVersion().WithPatch(0).EqualTo(effectiveVersion.EmulationVersion()) {
+			errors = append(errors, fmt.Errorf("ServerRunOptions.EmulationForwardCompatible cannot be set to true if the emulation version is the same as the binary version"))
+		}
+	}
 	return errors
 }
 
@@ -376,6 +387,9 @@ func (s *ServerRunOptions) AddUniversalFlags(fs *pflag.FlagSet) {
 		"for active watch request(s) to drain during the graceful server shutdown window.")
 
 	s.ComponentGlobalsRegistry.AddFlags(fs)
+	fs.BoolVar(&s.EmulationForwardCompatible, "emulation-forward-compatible", s.EmulationForwardCompatible, ""+
+		"If true APIs that have higher priority than the APIs enabled at the emulation version of the same group resource will be installed. "+
+		"Can only be set to true if the emulation version is lower than the binary version.")
 }
 
 // Complete fills missing fields with defaults.

staging/src/k8s.io/apiserver/pkg/server/options/server_run_options_test.go

@@ -31,18 +31,12 @@ import (
 )
 
 func TestServerRunOptionsValidate(t *testing.T) {
-	defaultComponentGlobalsRegistry := basecompatibility.NewComponentGlobalsRegistry()
-	testRegistry := basecompatibility.NewComponentGlobalsRegistry()
-	featureGate := utilfeature.DefaultFeatureGate.DeepCopy()
-	effectiveVersion := basecompatibility.NewEffectiveVersionFromString("1.35", "1.32", "1.32")
-	effectiveVersion.SetEmulationVersion(version.MajorMinor(1, 31))
 	testComponent := "test"
-	utilruntime.Must(testRegistry.Register(testComponent, effectiveVersion, featureGate))
-
 	testCases := []struct {
-		name        string
+		name             string
-		testOptions *ServerRunOptions
+		testOptions      *ServerRunOptions
-		expectErr   string
+		emulationVersion string
+		expectErr        string
 	}{
 		{
 			name: "Test when MaxRequestsInFlight is negative value",
@@ -55,7 +49,8 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
-				ComponentGlobalsRegistry:    defaultComponentGlobalsRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
 			},
 			expectErr: "--max-requests-inflight can not be negative value",
 		},
@@ -70,7 +65,8 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
-				ComponentGlobalsRegistry:    defaultComponentGlobalsRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
 			},
 			expectErr: "--max-mutating-requests-inflight can not be negative value",
 		},
@@ -85,7 +81,8 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
-				ComponentGlobalsRegistry:    defaultComponentGlobalsRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
 			},
 			expectErr: "--request-timeout can not be negative value",
 		},
@@ -100,7 +97,8 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           -1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
-				ComponentGlobalsRegistry:    defaultComponentGlobalsRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
 			},
 			expectErr: "--min-request-timeout can not be negative value",
 		},
@@ -115,7 +113,8 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       -10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
-				ComponentGlobalsRegistry:    defaultComponentGlobalsRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
 			},
 			expectErr: "ServerRunOptions.JSONPatchMaxCopyBytes can not be negative value",
 		},
@@ -130,7 +129,8 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         -10 * 1024 * 1024,
-				ComponentGlobalsRegistry:    defaultComponentGlobalsRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
 			},
 			expectErr: "ServerRunOptions.MaxRequestBodyBytes can not be negative value",
 		},
@@ -146,7 +146,8 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				LivezGracePeriod:            -time.Second,
-				ComponentGlobalsRegistry:    defaultComponentGlobalsRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
 			},
 			expectErr: "--livez-grace-period can not be a negative value",
 		},
@@ -162,7 +163,8 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				ShutdownDelayDuration:       -time.Second,
-				ComponentGlobalsRegistry:    defaultComponentGlobalsRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
 			},
 			expectErr: "--shutdown-delay-duration can not be negative value",
 		},
@@ -178,7 +180,8 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
-				ComponentGlobalsRegistry:    defaultComponentGlobalsRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
 			},
 			expectErr: "--strict-transport-security-directives invalid, allowed values: max-age=expireTime, includeSubDomains, preload. see https://tools.ietf.org/html/rfc6797#section-6.1 for more information",
 		},
@@ -195,9 +198,46 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				ComponentName:               testComponent,
-				ComponentGlobalsRegistry:    testRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
 			},
-			expectErr: "emulation version 1.31 is not between [1.32, 1.35.0]",
+			emulationVersion: "1.31",
+			expectErr:        "emulation version 1.31 is not between [1.32, 1.35.0]",
+		},
+		{
+			name: "Test EmulationForwardCompatible cannot be true if not in emulation mode",
+			testOptions: &ServerRunOptions{
+				AdvertiseAddress:            netutils.ParseIPSloppy("192.168.10.10"),
+				CorsAllowedOriginList:       []string{"^10.10.10.100$", "^10.10.10.200$"},
+				HSTSDirectives:              []string{"max-age=31536000", "includeSubDomains", "preload"},
+				MaxRequestsInFlight:         400,
+				MaxMutatingRequestsInFlight: 200,
+				RequestTimeout:              time.Duration(2) * time.Minute,
+				MinRequestTimeout:           1800,
+				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
+				MaxRequestBodyBytes:         10 * 1024 * 1024,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
+				EmulationForwardCompatible:  true,
+			},
+			expectErr: "ServerRunOptions.EmulationForwardCompatible cannot be set to true if the emulation version is the same as the binary version",
+		},
+		{
+			name: "Test EmulationForwardCompatible can be true if in emulation mode",
+			testOptions: &ServerRunOptions{
+				AdvertiseAddress:            netutils.ParseIPSloppy("192.168.10.10"),
+				CorsAllowedOriginList:       []string{"^10.10.10.100$", "^10.10.10.200$"},
+				HSTSDirectives:              []string{"max-age=31536000", "includeSubDomains", "preload"},
+				MaxRequestsInFlight:         400,
+				MaxMutatingRequestsInFlight: 200,
+				RequestTimeout:              time.Duration(2) * time.Minute,
+				MinRequestTimeout:           1800,
+				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
+				MaxRequestBodyBytes:         10 * 1024 * 1024,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
+				EmulationForwardCompatible:  true,
+			},
+			emulationVersion: "1.34",
 		},
 		{
 			name: "Test when ServerRunOptions is valid",
@@ -211,13 +251,18 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
-				ComponentGlobalsRegistry:    defaultComponentGlobalsRegistry,
+				ComponentGlobalsRegistry:    newTestRegistry(testComponent),
+				ComponentName:               testComponent,
 			},
 		},
 	}
 
 	for _, testcase := range testCases {
 		t.Run(testcase.name, func(t *testing.T) {
+			if testcase.emulationVersion != "" {
+				effectiveVersion := testcase.testOptions.ComponentGlobalsRegistry.EffectiveVersionFor(testcase.testOptions.ComponentName)
+				effectiveVersion.(basecompatibility.MutableEffectiveVersion).SetEmulationVersion(version.MustParse(testcase.emulationVersion))
+			}
 			errs := testcase.testOptions.Validate()
 			if len(testcase.expectErr) != 0 && !strings.Contains(utilerrors.NewAggregate(errs).Error(), testcase.expectErr) {
 				t.Errorf("got err: %v, expected err: %s", errs, testcase.expectErr)
@@ -230,6 +275,14 @@ func TestServerRunOptionsValidate(t *testing.T) {
 	}
 }
 
+func newTestRegistry(componentName string) basecompatibility.ComponentGlobalsRegistry {
+	registry := basecompatibility.NewComponentGlobalsRegistry()
+	featureGate := utilfeature.DefaultFeatureGate.DeepCopy()
+	effectiveVersion := basecompatibility.NewEffectiveVersionFromString("1.35", "1.32", "1.32")
+	utilruntime.Must(registry.Register(componentName, effectiveVersion, featureGate))
+	return registry
+}
+
 func TestValidateCorsAllowedOriginList(t *testing.T) {
 	tests := []struct {
 		regexp           [][]string

staging/src/k8s.io/apiserver/pkg/server/resourceconfig/helpers.go

@@ -36,6 +36,8 @@ type GroupVersionRegistry interface {
 	IsVersionRegistered(v schema.GroupVersion) bool
 	// PrioritizedVersionsAllGroups returns all registered group versions.
 	PrioritizedVersionsAllGroups() []schema.GroupVersion
+	// PrioritizedVersionsForGroup returns versions for a single group in priority order
+	PrioritizedVersionsForGroup(group string) []schema.GroupVersion
 }
 
 // MergeResourceEncodingConfigs merges the given defaultResourceConfig with specific GroupVersionResource overrides.
@@ -100,7 +102,17 @@ func MergeAPIResourceConfigs(
 			}
 		}
 	}
+	if err := applyVersionAndResourcePreferences(resourceConfig, overrides, registry); err != nil {
+		return nil, err
+	}
+	return resourceConfig, nil
+}
 
+func applyVersionAndResourcePreferences(
+	resourceConfig *serverstore.ResourceConfig,
+	overrides cliflag.ConfigurationMap,
+	registry GroupVersionRegistry,
+) error {
 	type versionEnablementPreference struct {
 		key          string
 		enabled      bool
@@ -130,7 +142,7 @@ func MergeAPIResourceConfigs(
 		groupVersionString := tokens[0] + "/" + tokens[1]
 		groupVersion, err := schema.ParseGroupVersion(groupVersionString)
 		if err != nil {
-			return nil, fmt.Errorf("invalid key %s", key)
+			return fmt.Errorf("invalid key %s", key)
 		}
 
 		// Exclude group not registered into the registry.
@@ -140,11 +152,11 @@ func MergeAPIResourceConfigs(
 
 		// Verify that the groupVersion is registered into registry.
 		if !registry.IsVersionRegistered(groupVersion) {
-			return nil, fmt.Errorf("group version %s that has not been registered", groupVersion.String())
+			return fmt.Errorf("group version %s that has not been registered", groupVersion.String())
 		}
 		enabled, err := getRuntimeConfigValue(overrides, key, false)
 		if err != nil {
-			return nil, err
+			return err
 		}
 
 		switch len(tokens) {
@@ -156,7 +168,7 @@ func MergeAPIResourceConfigs(
 			})
 		case 3:
 			if strings.ToLower(tokens[2]) != tokens[2] {
-				return nil, fmt.Errorf("invalid key %v: group/version/resource and resource is always lowercase plural, not %q", key, tokens[2])
+				return fmt.Errorf("invalid key %v: group/version/resource and resource is always lowercase plural, not %q", key, tokens[2])
 			}
 			resourcePreferences = append(resourcePreferences, resourceEnablementPreference{
 				key:                  key,
@@ -187,8 +199,7 @@ func MergeAPIResourceConfigs(
 			resourceConfig.DisableResources(resourcePreference.groupVersionResource)
 		}
 	}
-
+	return nil
-	return resourceConfig, nil
 }
 
 func getRuntimeConfigValue(overrides cliflag.ConfigurationMap, apiKey string, defaultValue bool) (bool, error) {
@@ -227,3 +238,51 @@ func ParseGroups(resourceConfig cliflag.ConfigurationMap) ([]string, error) {
 
 	return groups, nil
 }
+
+// EmulationForwardCompatibleResourceConfig creates a new ResourceConfig that besides all the enabled resources in resourceConfig,
+// enables all higher priority versions of enabled resources, excluding alpha versions.
+// This is useful for ensuring forward compatibility when a new version of an API is introduced.
+func EmulationForwardCompatibleResourceConfig(
+	resourceConfig *serverstore.ResourceConfig,
+	resourceConfigOverrides cliflag.ConfigurationMap,
+	registry GroupVersionRegistry,
+) (*serverstore.ResourceConfig, error) {
+	ret := serverstore.NewResourceConfig()
+	for gv, enabled := range resourceConfig.GroupVersionConfigs {
+		ret.GroupVersionConfigs[gv] = enabled
+		if !enabled {
+			continue
+		}
+		// EmulationForwardCompatibility is not applicable to alpha apis.
+		if alphaPattern.MatchString(gv.Version) {
+			continue
+		}
+		for _, pgv := range registry.PrioritizedVersionsForGroup(gv.Group) {
+			if pgv.Version == gv.Version {
+				break
+			}
+			ret.EnableVersions(pgv)
+		}
+	}
+	for gvr, enabled := range resourceConfig.ResourceConfigs {
+		ret.ResourceConfigs[gvr] = enabled
+		if !enabled {
+			continue
+		}
+		// EmulationForwardCompatibility is not applicable to alpha apis.
+		if alphaPattern.MatchString(gvr.Version) {
+			continue
+		}
+		for _, pgv := range registry.PrioritizedVersionsForGroup(gvr.Group) {
+			if pgv.Version == gvr.Version {
+				break
+			}
+			ret.EnableResources(pgv.WithResource(gvr.Resource))
+		}
+	}
+	// need to reapply the version preferences if there is an override of a higher priority version.
+	if err := applyVersionAndResourcePreferences(ret, resourceConfigOverrides, registry); err != nil {
+		return nil, err
+	}
+	return ret, nil
+}

staging/src/k8s.io/apiserver/pkg/server/resourceconfig/helpers_test.go

@@ -27,6 +27,7 @@ import (
 	extensionsapiv1beta1 "k8s.io/api/extensions/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	runtimetesting "k8s.io/apimachinery/pkg/runtime/testing"
 	serverstore "k8s.io/apiserver/pkg/server/storage"
 )
 
@@ -547,6 +548,186 @@ func TestParseRuntimeConfig(t *testing.T) {
 	}
 }
 
+func TestEmulationForwardCompatibleResourceConfig(t *testing.T) {
+	scheme := newFakeScheme(t)
+	addTestGVs(t, scheme)
+	testGroup := "test"
+	v1 := schema.GroupVersion{Group: testGroup, Version: "v1"}
+	v2alpha1 := schema.GroupVersion{Group: "test", Version: "v2alpha1"}
+	v2beta1 := schema.GroupVersion{Group: testGroup, Version: "v2beta1"}
+	v2beta2 := schema.GroupVersion{Group: testGroup, Version: "v2beta2"}
+	v2 := schema.GroupVersion{Group: testGroup, Version: "v2"}
+
+	testCases := []struct {
+		name                    string
+		resourceConfig          func() *serverstore.ResourceConfig
+		resourceConfigOverrides map[string]string
+		expectedAPIConfig       func() *serverstore.ResourceConfig
+		err                     bool
+	}{
+		{
+			name: "emulation-forward-compatible-enabled-no-higher-priority",
+			resourceConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableVersions(v2)
+				return config
+			},
+			resourceConfigOverrides: map[string]string{},
+			expectedAPIConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableVersions(v2)
+				return config
+			},
+			err: false,
+		},
+		{
+			name: "emulation-forward-compatible-alpha-version",
+			resourceConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableVersions(v2alpha1)
+				return config
+			},
+			resourceConfigOverrides: map[string]string{},
+			expectedAPIConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableVersions(v2alpha1)
+				return config
+			},
+			err: false,
+		},
+		{
+			name: "emulation-forward-compatible-enabled-higher-priority-ga",
+			resourceConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableVersions(v1)
+				return config
+			},
+			resourceConfigOverrides: map[string]string{},
+			expectedAPIConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableVersions(v1, v2)
+				return config
+			},
+			err: false,
+		},
+		{
+			name: "emulation-forward-compatible-enabled-higher-priority",
+			resourceConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableVersions(v2beta1)
+				return config
+			},
+			resourceConfigOverrides: map[string]string{},
+			expectedAPIConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableVersions(v2beta1, v2beta2, v1, v2)
+				return config
+			},
+			err: false,
+		},
+		{
+			name: "emulation-forward-compatible-enabled-higher-priority-with-override",
+			resourceConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableVersions(v2beta1)
+				return config
+			},
+			resourceConfigOverrides: map[string]string{
+				"test/v2beta2": "false",
+			},
+			expectedAPIConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableVersions(v2beta1, v1, v2)
+				config.DisableVersions(v2beta2)
+				return config
+			},
+			err: false,
+		},
+		{
+			name: "emulation-forward-compatible-enabled-resource-no-higher-priority",
+			resourceConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableResources(v2.WithResource("testtype1"))
+				return config
+			},
+			resourceConfigOverrides: map[string]string{},
+			expectedAPIConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableResources(v2.WithResource("testtype1"))
+				return config
+			},
+			err: false,
+		},
+		{
+			name: "emulation-forward-compatible-alpha-resource",
+			resourceConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableResources(v2alpha1.WithResource("testtype1"))
+				return config
+			},
+			resourceConfigOverrides: map[string]string{},
+			expectedAPIConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableResources(v2alpha1.WithResource("testtype1"))
+				return config
+			},
+			err: false,
+		},
+		{
+			name: "emulation-forward-compatible-enabled-resource-higher-priority",
+			resourceConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableResources(v2beta1.WithResource("testtype1"))
+				return config
+			},
+			resourceConfigOverrides: map[string]string{},
+			expectedAPIConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableResources(v2beta1.WithResource("testtype1"), v2beta2.WithResource("testtype1"), v1.WithResource("testtype1"), v2.WithResource("testtype1"))
+				return config
+			},
+			err: false,
+		},
+		{
+			name: "emulation-forward-compatible-enabled-resource-higher-priority-with-override",
+			resourceConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableResources(v2beta1.WithResource("testtype1"))
+				return config
+			},
+			resourceConfigOverrides: map[string]string{
+				"test/v2beta2/testtype1": "false",
+			},
+			expectedAPIConfig: func() *serverstore.ResourceConfig {
+				config := serverstore.NewResourceConfig()
+				config.EnableResources(v2beta1.WithResource("testtype1"), v1.WithResource("testtype1"), v2.WithResource("testtype1"))
+				config.DisableResources(v2beta2.WithResource("testtype1"))
+				return config
+			},
+			err: false,
+		},
+	}
+
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+			actualAPIConfig, err := EmulationForwardCompatibleResourceConfig(test.resourceConfig(), test.resourceConfigOverrides, scheme)
+			if err == nil && test.err {
+				t.Fatalf("expected error")
+			} else if err != nil && !test.err {
+				t.Fatalf("unexpected error: %s, for test: %v", err, test)
+			}
+			if err != nil {
+				return
+			}
+
+			expectedConfig := test.expectedAPIConfig()
+			if !reflect.DeepEqual(actualAPIConfig, expectedConfig) {
+				t.Fatalf("unexpected apiResourceConfig. Actual: %v\n expected: %v", actualAPIConfig, expectedConfig)
+			}
+		})
+	}
+}
+
 func newFakeAPIResourceConfigSource() *serverstore.ResourceConfig {
 	ret := serverstore.NewResourceConfig()
 	// NOTE: GroupVersions listed here will be enabled by default. Don't put alpha versions in the list.
@@ -601,3 +782,20 @@ func newFakeScheme(t *testing.T) *runtime.Scheme {
 
 	return ret
 }
+
+func addTestGVs(t *testing.T, s *runtime.Scheme) {
+	v1 := schema.GroupVersion{Group: "test", Version: "v1"}
+	v2alpha1 := schema.GroupVersion{Group: "test", Version: "v2alpha1"}
+	v2beta1 := schema.GroupVersion{Group: "test", Version: "v2beta1"}
+	v2beta2 := schema.GroupVersion{Group: "test", Version: "v2beta2"}
+	v2 := schema.GroupVersion{Group: "test", Version: "v2"}
+
+	s.AddKnownTypes(v1, &runtimetesting.TestType1{})
+	s.AddKnownTypes(v2alpha1, &runtimetesting.TestType1{})
+	s.AddKnownTypes(v2beta1, &runtimetesting.TestType1{})
+	s.AddKnownTypes(v2beta2, &runtimetesting.TestType1{}, &runtimetesting.TestType2{})
+	s.AddKnownTypes(v2, &runtimetesting.TestType1{}, &runtimetesting.TestType2{})
+
+	require.NoError(t, runtimetesting.RegisterConversions(s))
+	require.NoError(t, s.SetVersionPriority(v2, v1, v2beta2, v2beta1))
+}

test/integration/apiserver/apiserver_test.go

@@ -3340,9 +3340,9 @@ func TestAllowedEmulationVersions(t *testing.T) {
 }
 
 func TestEnableEmulationVersion(t *testing.T) {
-	featuregatetesting.SetFeatureGateEmulationVersionDuringTest(t, utilfeature.DefaultFeatureGate, version.MustParse("1.32"))
+	featuregatetesting.SetFeatureGateEmulationVersionDuringTest(t, utilfeature.DefaultFeatureGate, version.MustParse("1.33"))
 	server := kubeapiservertesting.StartTestServerOrDie(t,
-		&kubeapiservertesting.TestServerInstanceOptions{BinaryVersion: "1.32"},
+		&kubeapiservertesting.TestServerInstanceOptions{BinaryVersion: "1.33"},
 		[]string{"--emulated-version=kube=1.31", "--runtime-config=api/beta=true"}, framework.SharedEtcd())
 	defer server.TearDownFn()
 
@@ -3368,17 +3368,79 @@ func TestEnableEmulationVersion(t *testing.T) {
 			expectedStatusCode: 200,
 		},
 		{
-			path:               "/apis/flowcontrol.apiserver.k8s.io/v1beta1/flowschemas", // introduced at 1.20, removed at 1.26
+			path:               "/apis/flowcontrol.apiserver.k8s.io/v1beta3/flowschemas", // introduced at 1.26, removed at 1.32
-			expectedStatusCode: 404,
+			expectedStatusCode: 200,
 		},
 		{
-			path:               "/apis/flowcontrol.apiserver.k8s.io/v1beta2/flowschemas", // introduced at 1.23, removed at 1.29
+			path:               "/apis/networking.k8s.io/v1beta1/servicecidrs", // introduced at 1.31, removed at 1.34
+			expectedStatusCode: 200,
+		},
+		{
+			path:               "/apis/networking.k8s.io/v1/servicecidrs", // introduced at 1.33
 			expectedStatusCode: 404,
 		},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.path, func(t *testing.T) {
+			req, err := http.NewRequest("GET", server.ClientConfig.Host+tc.path, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			resp, err := rt.RoundTrip(req)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if resp.StatusCode != tc.expectedStatusCode {
+				t.Errorf("expect status code: %d, got : %d\n", tc.expectedStatusCode, resp.StatusCode)
+			}
+			defer func() {
+				_ = resp.Body.Close()
+			}()
+		})
+	}
+}
+
+func TestEnableEmulationVersionForwardCompatible(t *testing.T) {
+	featuregatetesting.SetFeatureGateEmulationVersionDuringTest(t, utilfeature.DefaultFeatureGate, version.MustParse("1.33"))
+	server := kubeapiservertesting.StartTestServerOrDie(t,
+		&kubeapiservertesting.TestServerInstanceOptions{BinaryVersion: "1.33"},
+		[]string{"--emulated-version=kube=1.31", "--emulation-forward-compatible=true"}, framework.SharedEtcd())
+	defer server.TearDownFn()
+
+	rt, err := restclient.TransportFor(server.ClientConfig)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tcs := []struct {
+		path               string
+		expectedStatusCode int
+	}{
+		{
+			path:               "/",
+			expectedStatusCode: 200,
+		},
+		{
+			path:               "/apis/apps/v1/deployments",
+			expectedStatusCode: 200,
+		},
+		{
+			path:               "/apis/flowcontrol.apiserver.k8s.io/v1/flowschemas",
+			expectedStatusCode: 200,
+		},
 		{
 			path:               "/apis/flowcontrol.apiserver.k8s.io/v1beta3/flowschemas", // introduced at 1.26, removed at 1.32
 			expectedStatusCode: 200,
 		},
+		{
+			path:               "/apis/networking.k8s.io/v1beta1/servicecidrs", // introduced at 1.31, removed at 1.34
+			expectedStatusCode: 200,
+		},
+		{
+			path:               "/apis/networking.k8s.io/v1/servicecidrs", // introduced at 1.33
+			expectedStatusCode: 200,
+		},
 	}
 
 	for _, tc := range tcs {
@@ -3429,14 +3491,6 @@ func TestDisableEmulationVersion(t *testing.T) {
 			path:               "/apis/flowcontrol.apiserver.k8s.io/v1/flowschemas",
 			expectedStatusCode: 200,
 		},
-		{
-			path:               "/apis/flowcontrol.apiserver.k8s.io/v1beta1/flowschemas", // introduced at 1.20, removed at 1.26
-			expectedStatusCode: 404,
-		},
-		{
-			path:               "/apis/flowcontrol.apiserver.k8s.io/v1beta2/flowschemas", // introduced at 1.23, removed at 1.29
-			expectedStatusCode: 404,
-		},
 		{
 			path:               "/apis/flowcontrol.apiserver.k8s.io/v1beta3/flowschemas", // introduced at 1.26, removed at 1.32
 			expectedStatusCode: 404,