Fix custom resource handler in-memory version

staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler.go

@@ -477,6 +477,7 @@ func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResource
 
 		storages[v.Name] = customresource.NewStorage(
 			schema.GroupResource{Group: crd.Spec.Group, Resource: crd.Status.AcceptedNames.Plural},
+			schema.GroupVersionKind{Group: crd.Spec.Group, Version: v.Name, Kind: crd.Status.AcceptedNames.Kind},
 			schema.GroupVersionKind{Group: crd.Spec.Group, Version: v.Name, Kind: crd.Status.AcceptedNames.ListKind},
 			customresource.NewStrategy(
 				typer,
@@ -525,6 +526,9 @@ func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResource
 			Resource: schema.GroupVersionResource{Group: crd.Spec.Group, Version: v.Name, Resource: crd.Status.AcceptedNames.Plural},
 			Kind:     kind,
 
+			// a handler for a specific group-version of a custom resource uses that version as the in-memory representation
+			HubGroupVersion: kind.GroupVersion(),
+
 			MetaGroupVersion: metav1.SchemeGroupVersion,
 
 			TableConvertor: storages[v.Name].CustomResource,

staging/src/k8s.io/apiextensions-apiserver/pkg/registry/customresource/etcd.go

@@ -40,8 +40,8 @@ type CustomResourceStorage struct {
 	Scale          *ScaleREST
 }
 
-func NewStorage(resource schema.GroupResource, listKind schema.GroupVersionKind, strategy customResourceStrategy, optsGetter generic.RESTOptionsGetter, categories []string, tableConvertor rest.TableConvertor) CustomResourceStorage {
-	customResourceREST, customResourceStatusREST := newREST(resource, listKind, strategy, optsGetter, categories, tableConvertor)
+func NewStorage(resource schema.GroupResource, kind, listKind schema.GroupVersionKind, strategy customResourceStrategy, optsGetter generic.RESTOptionsGetter, categories []string, tableConvertor rest.TableConvertor) CustomResourceStorage {
+	customResourceREST, customResourceStatusREST := newREST(resource, kind, listKind, strategy, optsGetter, categories, tableConvertor)
 
 	s := CustomResourceStorage{
 		CustomResource: customResourceREST,
@@ -75,9 +75,14 @@ type REST struct {
 }
 
 // newREST returns a RESTStorage object that will work against API services.
-func newREST(resource schema.GroupResource, listKind schema.GroupVersionKind, strategy customResourceStrategy, optsGetter generic.RESTOptionsGetter, categories []string, tableConvertor rest.TableConvertor) (*REST, *StatusREST) {
+func newREST(resource schema.GroupResource, kind, listKind schema.GroupVersionKind, strategy customResourceStrategy, optsGetter generic.RESTOptionsGetter, categories []string, tableConvertor rest.TableConvertor) (*REST, *StatusREST) {
 	store := &genericregistry.Store{
-		NewFunc: func() runtime.Object { return &unstructured.Unstructured{} },
+		NewFunc: func() runtime.Object {
+			// set the expected group/version/kind in the new object as a signal to the versioning decoder
+			ret := &unstructured.Unstructured{}
+			ret.SetGroupVersionKind(kind)
+			return ret
+		},
 		NewListFunc: func() runtime.Object {
 			// lists are never stored, only manufactured, so stomp in the right kind
 			ret := &unstructured.UnstructuredList{}

staging/src/k8s.io/apiextensions-apiserver/pkg/registry/customresource/etcd_test.go

@@ -91,6 +91,7 @@ func newStorage(t *testing.T) (customresource.CustomResourceStorage, *etcdtestin
 
 	storage := customresource.NewStorage(
 		schema.GroupResource{Group: "mygroup.example.com", Resource: "noxus"},
+		kind,
 		schema.GroupVersionKind{Group: "mygroup.example.com", Version: "v1beta1", Kind: "NoxuItemList"},
 		customresource.NewStrategy(
 			typer,

staging/src/k8s.io/apiextensions-apiserver/test/integration/BUILD

@@ -45,6 +45,7 @@ go_test(
         "//vendor/github.com/coreos/etcd/clientv3:go_default_library",
         "//vendor/github.com/coreos/etcd/pkg/transport:go_default_library",
         "//vendor/github.com/ghodss/yaml:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/github.com/stretchr/testify/require:go_default_library",
     ],
 )

staging/src/k8s.io/apiextensions-apiserver/test/integration/subresources_test.go

@@ -51,6 +51,10 @@ func NewNoxuSubresourcesCRD(scope apiextensionsv1beta1.ResourceScope) *apiextens
 				ShortNames: []string{"foo", "bar", "abc", "def"},
 				ListKind:   "NoxuItemList",
 			},
+			Versions: []apiextensionsv1beta1.CustomResourceDefinitionVersion{
+				{Name: "v1beta1", Served: true, Storage: false},
+				{Name: "v1", Served: true, Storage: true},
+			},
 			Scope: scope,
 			Subresources: &apiextensionsv1beta1.CustomResourceSubresources{
 				Status: &apiextensionsv1beta1.CustomResourceSubresourceStatus{},

staging/src/k8s.io/apiextensions-apiserver/test/integration/versioning_test.go

@@ -17,14 +17,116 @@ limitations under the License.
 package integration
 
 import (
+	"fmt"
+	"net/http"
 	"reflect"
 	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	"k8s.io/apiextensions-apiserver/test/integration/fixtures"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+func TestInternalVersionIsHandlerVersion(t *testing.T) {
+	tearDown, apiExtensionClient, dynamicClient, err := fixtures.StartDefaultServerWithClients(t)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tearDown()
+
+	noxuDefinition := fixtures.NewMultipleVersionNoxuCRD(apiextensionsv1beta1.NamespaceScoped)
+
+	assert.Equal(t, "v1beta1", noxuDefinition.Spec.Versions[0].Name)
+	assert.Equal(t, "v1beta2", noxuDefinition.Spec.Versions[1].Name)
+	assert.True(t, noxuDefinition.Spec.Versions[1].Storage)
+
+	noxuDefinition, err = fixtures.CreateNewCustomResourceDefinition(noxuDefinition, apiExtensionClient, dynamicClient)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ns := "not-the-default"
+
+	noxuNamespacedResourceClientV1beta1 := newNamespacedCustomResourceVersionedClient(ns, dynamicClient, noxuDefinition, "v1beta1") // use the non-storage version v1beta1
+
+	t.Logf("Creating foo")
+	noxuInstanceToCreate := fixtures.NewNoxuInstance(ns, "foo")
+	_, err = noxuNamespacedResourceClientV1beta1.Create(noxuInstanceToCreate, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// update validation via update because the cache priming in CreateNewCustomResourceDefinition will fail otherwise
+	t.Logf("Updating CRD to validate apiVersion")
+	noxuDefinition, err = updateCustomResourceDefinitionWithRetry(apiExtensionClient, noxuDefinition.Name, func(crd *apiextensionsv1beta1.CustomResourceDefinition) {
+		crd.Spec.Validation = &apiextensionsv1beta1.CustomResourceValidation{
+			OpenAPIV3Schema: &apiextensionsv1beta1.JSONSchemaProps{
+				Properties: map[string]apiextensionsv1beta1.JSONSchemaProps{
+					"apiVersion": {
+						Pattern: "^mygroup.example.com/v1beta1$", // this means we can only patch via the v1beta1 handler version
+					},
+				},
+				Required: []string{"apiVersion"},
+			},
+		}
+	})
+	assert.NoError(t, err)
+
+	time.Sleep(time.Second)
+
+	// patches via handler version v1beta1 should succeed (validation allows that API version)
+	{
+		t.Logf("patch of handler version v1beta1 (non-storage version) should succeed")
+		i := 0
+		err = wait.PollImmediate(time.Millisecond*100, wait.ForeverTestTimeout, func() (bool, error) {
+			patch := []byte(fmt.Sprintf(`{"i": %d}`, i))
+			i++
+
+			_, err := noxuNamespacedResourceClientV1beta1.Patch("foo", types.MergePatchType, patch, metav1.UpdateOptions{})
+			if err != nil {
+				// work around "grpc: the client connection is closing" error
+				// TODO: fix the grpc error
+				if err, ok := err.(*errors.StatusError); ok && err.Status().Code == http.StatusInternalServerError {
+					return false, nil
+				}
+				return false, err
+			}
+			return true, nil
+		})
+		assert.NoError(t, err)
+	}
+
+	// patches via handler version matching storage version should fail (validation does not allow that API version)
+	{
+		t.Logf("patch of handler version v1beta2 (storage version) should fail")
+		i := 0
+		noxuNamespacedResourceClientV1beta2 := newNamespacedCustomResourceVersionedClient(ns, dynamicClient, noxuDefinition, "v1beta2") // use the storage version v1beta2
+		err = wait.PollImmediate(time.Millisecond*100, wait.ForeverTestTimeout, func() (bool, error) {
+			patch := []byte(fmt.Sprintf(`{"i": %d}`, i))
+			i++
+
+			_, err := noxuNamespacedResourceClientV1beta2.Patch("foo", types.MergePatchType, patch, metav1.UpdateOptions{})
+			assert.NotNil(t, err)
+
+			// work around "grpc: the client connection is closing" error
+			// TODO: fix the grpc error
+			if err, ok := err.(*errors.StatusError); ok && err.Status().Code == http.StatusInternalServerError {
+				return false, nil
+			}
+
+			assert.Contains(t, err.Error(), "apiVersion")
+			return true, nil
+		})
+		assert.NoError(t, err)
+	}
+}
+
 func TestVersionedNamspacedScopedCRD(t *testing.T) {
 	tearDown, apiExtensionClient, dynamicClient, err := fixtures.StartDefaultServerWithClients(t)
 	if err != nil {

staging/src/k8s.io/apimachinery/pkg/runtime/serializer/versioning/versioning.go

@@ -18,6 +18,7 @@ package versioning
 
 import (
 	"io"
+	"reflect"
 
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -90,7 +91,16 @@ func (c *codec) Decode(data []byte, defaultGVK *schema.GroupVersionKind, into ru
 		into = versioned.Last()
 	}
 
-	obj, gvk, err := c.decoder.Decode(data, defaultGVK, into)
+	// If the into object is unstructured and expresses an opinion about its group/version,
+	// create a new instance of the type so we always exercise the conversion path (skips short-circuiting on `into == obj`)
+	decodeInto := into
+	if into != nil {
+		if _, ok := into.(runtime.Unstructured); ok && !into.GetObjectKind().GroupVersionKind().GroupVersion().Empty() {
+			decodeInto = reflect.New(reflect.TypeOf(into).Elem()).Interface().(runtime.Object)
+		}
+	}
+
+	obj, gvk, err := c.decoder.Decode(data, defaultGVK, decodeInto)
 	if err != nil {
 		return nil, gvk, err
 	}

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/create.go

@@ -77,7 +77,7 @@ func createHandler(r rest.NamedCreater, scope RequestScope, admit admission.Inte
 			scope.err(err, w, req)
 			return
 		}
-		decoder := scope.Serializer.DecoderToVersion(s.Serializer, schema.GroupVersion{Group: gv.Group, Version: runtime.APIVersionInternal})
+		decoder := scope.Serializer.DecoderToVersion(s.Serializer, scope.HubGroupVersion)
 
 		body, err := readBody(req)
 		if err != nil {

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/patch.go

@@ -118,9 +118,10 @@ func PatchResource(r rest.Patcher, scope RequestScope, admit admission.Interface
 			return
 		}
 		gv := scope.Kind.GroupVersion()
+
 		codec := runtime.NewCodec(
 			scope.Serializer.EncoderForVersion(s.Serializer, gv),
-			scope.Serializer.DecoderToVersion(s.Serializer, schema.GroupVersion{Group: gv.Group, Version: runtime.APIVersionInternal}),
+			scope.Serializer.DecoderToVersion(s.Serializer, scope.HubGroupVersion),
 		)
 
 		userInfo, _ := request.UserFrom(ctx)
@@ -163,6 +164,8 @@ func PatchResource(r rest.Patcher, scope RequestScope, admit admission.Interface
 			kind:            scope.Kind,
 			resource:        scope.Resource,
 
+			hubGroupVersion: scope.HubGroupVersion,
+
 			createValidation: rest.AdmissionToValidateObjectFunc(admit, staticAdmissionAttributes),
 			updateValidation: rest.AdmissionToValidateObjectUpdateFunc(admit, staticAdmissionAttributes),
 			admissionCheck:   admissionCheck,
@@ -218,6 +221,8 @@ type patcher struct {
 	resource        schema.GroupVersionResource
 	kind            schema.GroupVersionKind
 
+	hubGroupVersion schema.GroupVersion
+
 	// Validation functions
 	createValidation rest.ValidateObjectFunc
 	updateValidation rest.ValidateObjectUpdateFunc
@@ -242,11 +247,6 @@ type patcher struct {
 	mechanism         patchMechanism
 }
 
-func (p *patcher) toUnversioned(versionedObj runtime.Object) (runtime.Object, error) {
-	gvk := p.kind.GroupKind().WithVersion(runtime.APIVersionInternal)
-	return p.unsafeConvertor.ConvertToVersion(versionedObj, gvk.GroupVersion())
-}
-
 type patchMechanism interface {
 	applyPatchToCurrentObject(currentObject runtime.Object) (runtime.Object, error)
 }
@@ -320,13 +320,8 @@ func (p *smpPatcher) applyPatchToCurrentObject(currentObject runtime.Object) (ru
 	if err := strategicPatchObject(p.defaulter, currentVersionedObject, p.patchJS, versionedObjToUpdate, p.schemaReferenceObj); err != nil {
 		return nil, err
 	}
-	// Convert the object back to unversioned (aka internal version).
-	unversionedObjToUpdate, err := p.toUnversioned(versionedObjToUpdate)
-	if err != nil {
-		return nil, err
-	}
-
-	return unversionedObjToUpdate, nil
+	// Convert the object back to the hub version
+	return p.unsafeConvertor.ConvertToVersion(versionedObjToUpdate, p.hubGroupVersion)
 }
 
 // strategicPatchObject applies a strategic merge patch of <patchJS> to

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/rest.go

@@ -65,6 +65,9 @@ type RequestScope struct {
 	Subresource string
 
 	MetaGroupVersion schema.GroupVersion
+
+	// HubGroupVersion indicates what version objects read from etcd or incoming requests should be converted to for in-memory handling.
+	HubGroupVersion schema.GroupVersion
 }
 
 func (scope *RequestScope) err(err error, w http.ResponseWriter, req *http.Request) {

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/rest_test.go

@@ -367,6 +367,7 @@ func (tc *patchTestCase) Run(t *testing.T) {
 	kind := examplev1.SchemeGroupVersion.WithKind("Pod")
 	resource := examplev1.SchemeGroupVersion.WithResource("pods")
 	schemaReferenceObj := &examplev1.Pod{}
+	hubVersion := example.SchemeGroupVersion
 
 	for _, patchType := range []types.PatchType{types.JSONPatchType, types.MergePatchType, types.StrategicMergePatchType} {
 		// This needs to be reset on each iteration.
@@ -439,6 +440,8 @@ func (tc *patchTestCase) Run(t *testing.T) {
 			kind:            kind,
 			resource:        resource,
 
+			hubGroupVersion: hubVersion,
+
 			createValidation: rest.ValidateAllObjectFunc,
 			updateValidation: admissionValidation,
 			admissionCheck:   admissionMutation,

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/update.go

@@ -89,8 +89,9 @@ func UpdateResource(r rest.Updater, scope RequestScope, admit admission.Interfac
 		}
 		defaultGVK := scope.Kind
 		original := r.New()
+
 		trace.Step("About to convert to expected version")
-		decoder := scope.Serializer.DecoderToVersion(s.Serializer, schema.GroupVersion{Group: defaultGVK.Group, Version: runtime.APIVersionInternal})
+		decoder := scope.Serializer.DecoderToVersion(s.Serializer, scope.HubGroupVersion)
 		obj, gvk, err := decoder.Decode(body, &defaultGVK, original)
 		if err != nil {
 			err = transformDecodeError(scope.Typer, err, original, gvk, body)

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

@@ -506,6 +506,8 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
 		Subresource: subresource,
 		Kind:        fqKindToRegister,
 
+		HubGroupVersion: schema.GroupVersion{Group: fqKindToRegister.Group, Version: runtime.APIVersionInternal},
+
 		MetaGroupVersion: metav1.SchemeGroupVersion,
 	}
 	if a.group.MetaGroupVersion != nil {