provide directly decodable versions for storageversion API

cmd/kube-apiserver/app/apiextensions.go

@@ -67,6 +67,7 @@ func createAPIExtensionsConfig(
 	// copy the etcd options so we don't mutate originals.
 	etcdOptions := *commandOptions.Etcd
 	etcdOptions.StorageConfig.Paging = utilfeature.DefaultFeatureGate.Enabled(features.APIListChunking)
+	// this is where the true decodable levels come from.
 	etcdOptions.StorageConfig.Codec = apiextensionsapiserver.Codecs.LegacyCodec(v1beta1.SchemeGroupVersion, v1.SchemeGroupVersion)
 	// prefer the more compact serialization (v1beta1) for storage until http://issue.k8s.io/82292 is resolved for objects whose v1 serialization is too big but whose v1beta1 serialization can be stored
 	etcdOptions.StorageConfig.EncodeVersioner = runtime.NewMultiGroupVersioner(v1beta1.SchemeGroupVersion, schema.GroupKind{Group: v1beta1.GroupName})

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

@@ -229,6 +229,32 @@ func (s *Scheme) KnownTypes(gv schema.GroupVersion) map[string]reflect.Type {
 	return types
 }
 
+// VersionsForGroupKind returns the versions that a particular GroupKind can be converted to within the given group.
+// A GroupKind might be converted to a different group. That information is available in EquivalentResourceMapper.
+func (s *Scheme) VersionsForGroupKind(gk schema.GroupKind) []schema.GroupVersion {
+	availableVersions := []schema.GroupVersion{}
+	for gvk := range s.gvkToType {
+		if gk != gvk.GroupKind() {
+			continue
+		}
+
+		availableVersions = append(availableVersions, gvk.GroupVersion())
+	}
+
+	// order the return for stability
+	ret := []schema.GroupVersion{}
+	for _, version := range s.PrioritizedVersionsForGroup(gk.Group) {
+		for _, availableVersion := range availableVersions {
+			if version != availableVersion {
+				continue
+			}
+			ret = append(ret, availableVersion)
+		}
+	}
+
+	return ret
+}
+
 // AllKnownTypes returns the all known types.
 func (s *Scheme) AllKnownTypes() map[schema.GroupVersionKind]reflect.Type {
 	return s.gvkToType

staging/src/k8s.io/apiserver/pkg/endpoints/groupversion.go

@@ -36,6 +36,13 @@ import (
 	openapiproto "k8s.io/kube-openapi/pkg/util/proto"
 )
 
+// ConvertabilityChecker indicates what versions a GroupKind is available in.
+type ConvertabilityChecker interface {
+	// VersionsForGroupKind indicates what versions are available to convert a group kind. This determines
+	// what our decoding abilities are.
+	VersionsForGroupKind(gk schema.GroupKind) []schema.GroupVersion
+}
+
 // APIGroupVersion is a helper for exposing rest.Storage objects as http.Handlers via go-restful
 // It handles URLs of the form:
 // /${storage_key}[/${object_name}]
@@ -70,6 +77,7 @@ type APIGroupVersion struct {
 	Typer                 runtime.ObjectTyper
 	Creater               runtime.ObjectCreater
 	Convertor             runtime.ObjectConvertor
+	ConvertabilityChecker ConvertabilityChecker
 	Defaulter             runtime.ObjectDefaulter
 	Linker                runtime.SelfLinker
 	UnsafeConvertor       runtime.ObjectConvertor

staging/src/k8s.io/apiserver/pkg/endpoints/installer.go

@@ -513,6 +513,10 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
 		if err != nil {
 			return nil, nil, err
 		}
+		decodableVersions := []schema.GroupVersion{}
+		if a.group.ConvertabilityChecker != nil {
+			decodableVersions = a.group.ConvertabilityChecker.VersionsForGroupKind(fqKindToRegister.GroupKind())
+		}
 		resourceInfo = &storageversion.ResourceInfo{
 			GroupResource: schema.GroupResource{
 				Group:    a.group.GroupVersion.Group,
@@ -523,6 +527,8 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
 			// DecodableVersions immediately because API installation must
 			// be completed first for us to know equivalent APIs
 			EquivalentResourceMapper: a.group.EquivalentResourceRegistry,
+
+			DirectlyDecodableVersions: decodableVersions,
 		}
 	}
 

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

@@ -553,6 +553,7 @@ func (s *GenericAPIServer) newAPIGroupVersion(apiGroupInfo *APIGroupInfo, groupV
 		Serializer:            apiGroupInfo.NegotiatedSerializer,
 		Creater:               apiGroupInfo.Scheme,
 		Convertor:             apiGroupInfo.Scheme,
+		ConvertabilityChecker: apiGroupInfo.Scheme,
 		UnsafeConvertor:       runtime.UnsafeObjectConvertor(apiGroupInfo.Scheme),
 		Defaulter:             apiGroupInfo.Scheme,
 		Typer:                 apiGroupInfo.Scheme,

staging/src/k8s.io/apiserver/pkg/storageversion/manager.go

@@ -40,6 +40,10 @@ type ResourceInfo struct {
 	// Used to calculate decodable versions. Can only be used after all
 	// equivalent versions are registered by InstallREST.
 	EquivalentResourceMapper runtime.EquivalentResourceRegistry
+
+	// DirectlyDecodableVersions is a list of versions that the converter for REST storage knows how to convert.  This
+	// contains items like apiextensions.k8s.io/v1beta1 even if we don't serve that version.
+	DirectlyDecodableVersions []schema.GroupVersion
 }
 
 // Manager records the resources whose StorageVersions need updates, and provides a method to update those StorageVersions.
@@ -133,13 +137,13 @@ func (s *defaultManager) UpdateStorageVersions(kubeAPIServerClientConfig *rest.C
 	// StorageVersion objects have CommonEncodingVersion (each with one server registered).
 	sortResourceInfosByGroupResource(resources)
 	for _, r := range dedupResourceInfos(resources) {
-		dv := decodableVersions(r.EquivalentResourceMapper, r.GroupResource)
+		decodableVersions := decodableVersions(r.DirectlyDecodableVersions, r.EquivalentResourceMapper, r.GroupResource)
 		gr := r.GroupResource
 		// Group must be a valid subdomain in DNS (RFC 1123)
 		if len(gr.Group) == 0 {
 			gr.Group = "core"
 		}
-		if err := updateStorageVersionFor(sc, serverID, gr, r.EncodingVersion, dv); err != nil {
+		if err := updateStorageVersionFor(sc, serverID, gr, r.EncodingVersion, decodableVersions); err != nil {
 			utilruntime.HandleError(fmt.Errorf("failed to update storage version for %v: %v", r.GroupResource, err))
 			s.recordStatusFailure(&r, err)
 			hasFailure = true
@@ -267,10 +271,23 @@ func (s *defaultManager) Completed() bool {
 	return s.completed.Load().(bool)
 }
 
-func decodableVersions(e runtime.EquivalentResourceRegistry, gr schema.GroupResource) []string {
+func decodableVersions(directlyDecodableVersions []schema.GroupVersion, e runtime.EquivalentResourceRegistry, gr schema.GroupResource) []string {
 	var versions []string
+	for _, decodableVersions := range directlyDecodableVersions {
+		versions = append(versions, decodableVersions.String())
+	}
+
 	decodingGVRs := e.EquivalentResourcesFor(gr.WithVersion(""), "")
 	for _, v := range decodingGVRs {
+		found := false
+		for _, existingVersion := range versions {
+			if existingVersion == v.GroupVersion().String() {
+				found = true
+			}
+		}
+		if found {
+			continue
+		}
 		versions = append(versions, v.GroupVersion().String())
 	}
 	return versions