CRD versioning with no-op converter

pkg/master/controller/crdregistration/crdregistration_controller.go

@@ -77,9 +77,11 @@ func NewAutoRegistrationController(crdinformer crdinformers.CustomResourceDefini
 			cast := obj.(*apiextensions.CustomResourceDefinition)
 			c.enqueueCRD(cast)
 		},
-		UpdateFunc: func(_, obj interface{}) {
-			cast := obj.(*apiextensions.CustomResourceDefinition)
-			c.enqueueCRD(cast)
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			// Enqueue both old and new object to make sure we remove and add appropriate API services.
+			// The working queue will resolve any duplicates and only changes will stay in the queue.
+			c.enqueueCRD(oldObj.(*apiextensions.CustomResourceDefinition))
+			c.enqueueCRD(newObj.(*apiextensions.CustomResourceDefinition))
 		},
 		DeleteFunc: func(obj interface{}) {
 			cast, ok := obj.(*apiextensions.CustomResourceDefinition)
@@ -120,11 +122,13 @@ func (c *crdRegistrationController) Run(threadiness int, stopCh <-chan struct{})
 		utilruntime.HandleError(err)
 	} else {
 		for _, crd := range crds {
-			if err := c.syncHandler(schema.GroupVersion{Group: crd.Spec.Group, Version: crd.Spec.Version}); err != nil {
+			for _, version := range crd.Spec.Versions {
+				if err := c.syncHandler(schema.GroupVersion{Group: crd.Spec.Group, Version: version.Name}); err != nil {
 					utilruntime.HandleError(err)
 				}
 			}
 		}
+	}
 	close(c.syncedInitialSet)
 
 	// start up your worker threads based on threadiness.  Some controllers have multiple kinds of workers
@@ -182,11 +186,12 @@ func (c *crdRegistrationController) processNextWorkItem() bool {
 }
 
 func (c *crdRegistrationController) enqueueCRD(crd *apiextensions.CustomResourceDefinition) {
-	c.queue.Add(schema.GroupVersion{Group: crd.Spec.Group, Version: crd.Spec.Version})
+	for _, version := range crd.Spec.Versions {
+		c.queue.Add(schema.GroupVersion{Group: crd.Spec.Group, Version: version.Name})
+	}
 }
 
 func (c *crdRegistrationController) handleVersionUpdate(groupVersion schema.GroupVersion) error {
-	found := false
 	apiServiceName := groupVersion.Version + "." + groupVersion.Group
 
 	// check all CRDs.  There shouldn't that many, but if we have problems later we can index them
@@ -195,15 +200,12 @@ func (c *crdRegistrationController) handleVersionUpdate(groupVersion schema.Grou
 		return err
 	}
 	for _, crd := range crds {
-		if crd.Spec.Version == groupVersion.Version && crd.Spec.Group == groupVersion.Group {
-			found = true
-			break
+		if crd.Spec.Group != groupVersion.Group {
+			continue
 		}
-	}
-
-	if !found {
-		c.apiServiceRegistration.RemoveAPIServiceToSync(apiServiceName)
-		return nil
+		for _, version := range crd.Spec.Versions {
+			if version.Name != groupVersion.Version || !version.Served {
+				continue
 			}
 
 			c.apiServiceRegistration.AddAPIServiceToSync(&apiregistration.APIService{
@@ -212,9 +214,13 @@ func (c *crdRegistrationController) handleVersionUpdate(groupVersion schema.Grou
 					Group:                groupVersion.Group,
 					Version:              groupVersion.Version,
 					GroupPriorityMinimum: 1000, // CRDs should have relatively low priority
-			VersionPriority:      100,  // CRDs should have relatively low priority
+					VersionPriority:      100,  // CRDs will be sorted by kube-like versions like any other APIService with the same VersionPriority
 				},
 			})
+			return nil
+		}
+	}
 
+	c.apiServiceRegistration.RemoveAPIServiceToSync(apiServiceName)
 	return nil
 }

pkg/master/controller/crdregistration/crdregistration_controller_test.go

@@ -43,7 +43,15 @@ func TestHandleVersionUpdate(t *testing.T) {
 				{
 					Spec: apiextensions.CustomResourceDefinitionSpec{
 						Group: "group.com",
-						Version: "v1",
+						// Version field is deprecated and crd registration won't rely on it at all.
+						// defaulting route will fill up Versions field if user only provided version field.
+						Versions: []apiextensions.CustomResourceDefinitionVersion{
+							{
+								Name:    "v1",
+								Served:  true,
+								Storage: true,
+							},
+						},
 					},
 				},
 			},
@@ -67,7 +75,13 @@ func TestHandleVersionUpdate(t *testing.T) {
 				{
 					Spec: apiextensions.CustomResourceDefinitionSpec{
 						Group: "group.com",
-						Version: "v1",
+						Versions: []apiextensions.CustomResourceDefinitionVersion{
+							{
+								Name:    "v1",
+								Served:  true,
+								Storage: true,
+							},
+						},
 					},
 				},
 			},
@@ -98,7 +112,6 @@ func TestHandleVersionUpdate(t *testing.T) {
 			t.Errorf("%s expected %v, got %v", test.name, test.expectedRemoved, registration.removed)
 		}
 	}
-
 }
 
 type fakeAPIServiceRegistration struct {

staging/src/k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/fuzzer/fuzzer.go

@@ -42,6 +42,28 @@ func Funcs(codecs runtimeserializer.CodecFactory) []interface{} {
 			if len(obj.Names.ListKind) == 0 && len(obj.Names.Kind) > 0 {
 				obj.Names.ListKind = obj.Names.Kind + "List"
 			}
+			if len(obj.Versions) == 0 && len(obj.Version) != 0 {
+				obj.Versions = []apiextensions.CustomResourceDefinitionVersion{
+					{
+						Name:    obj.Version,
+						Served:  true,
+						Storage: true,
+					},
+				}
+			} else if len(obj.Versions) != 0 {
+				obj.Version = obj.Versions[0].Name
+			}
+		},
+		func(obj *apiextensions.CustomResourceDefinition, c fuzz.Continue) {
+			c.FuzzNoCustom(obj)
+
+			if len(obj.Status.StoredVersions) == 0 {
+				for _, v := range obj.Spec.Versions {
+					if v.Storage && !apiextensions.IsStoredVersion(obj, v.Name) {
+						obj.Status.StoredVersions = append(obj.Status.StoredVersions, v.Name)
+					}
+				}
+			}
 		},
 		func(obj *apiextensions.JSONSchemaProps, c fuzz.Continue) {
 			// we cannot use c.FuzzNoCustom because of the interface{} fields. So let's loop with reflection.

staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/converter.go

@@ -0,0 +1,67 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package conversion
+
+import (
+	"fmt"
+
+	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// NewCRDConverter returns a new CRD converter based on the conversion settings in crd object.
+func NewCRDConverter(crd *apiextensions.CustomResourceDefinition) (safe, unsafe runtime.ObjectConvertor) {
+	validVersions := map[schema.GroupVersion]bool{}
+	for _, version := range crd.Spec.Versions {
+		validVersions[schema.GroupVersion{Group: crd.Spec.Group, Version: version.Name}] = true
+	}
+
+	// The only converter right now is nopConverter. More converters will be returned based on the
+	// CRD object when they introduced.
+	unsafe = &nopConverter{
+		clusterScoped: crd.Spec.Scope == apiextensions.ClusterScoped,
+		validVersions: validVersions,
+	}
+	return &safeConverterWrapper{unsafe}, unsafe
+}
+
+// safeConverterWrapper is a wrapper over an unsafe object converter that makes copy of the input and then delegate to the unsafe converter.
+type safeConverterWrapper struct {
+	unsafe runtime.ObjectConvertor
+}
+
+var _ runtime.ObjectConvertor = &nopConverter{}
+
+// ConvertFieldLabel delegate the call to the unsafe converter.
+func (c *safeConverterWrapper) ConvertFieldLabel(version, kind, label, value string) (string, string, error) {
+	return c.unsafe.ConvertFieldLabel(version, kind, label, value)
+}
+
+// Convert makes a copy of in object and then delegate the call to the unsafe converter.
+func (c *safeConverterWrapper) Convert(in, out, context interface{}) error {
+	inObject, ok := in.(runtime.Object)
+	if !ok {
+		return fmt.Errorf("input type %T in not valid for object conversion", in)
+	}
+	return c.unsafe.Convert(inObject.DeepCopyObject(), out, context)
+}
+
+// ConvertToVersion makes a copy of in object and then delegate the call to the unsafe converter.
+func (c *safeConverterWrapper) ConvertToVersion(in runtime.Object, target runtime.GroupVersioner) (runtime.Object, error) {
+	return c.unsafe.ConvertToVersion(in.DeepCopyObject(), target)
+}

staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/nop_converter.go

@@ -0,0 +1,100 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package conversion
+
+import (
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// nopConverter is a converter that only sets the apiVersion fields, but does not real conversion. It supports fields selectors.
+type nopConverter struct {
+	clusterScoped bool
+	validVersions map[schema.GroupVersion]bool
+}
+
+var _ runtime.ObjectConvertor = &nopConverter{}
+
+func (c *nopConverter) ConvertFieldLabel(version, kind, label, value string) (string, string, error) {
+	// We currently only support metadata.namespace and metadata.name.
+	switch {
+	case label == "metadata.name":
+		return label, value, nil
+	case !c.clusterScoped && label == "metadata.namespace":
+		return label, value, nil
+	default:
+		return "", "", fmt.Errorf("field label not supported: %s", label)
+	}
+}
+
+func (c *nopConverter) Convert(in, out, context interface{}) error {
+	unstructIn, ok := in.(*unstructured.Unstructured)
+	if !ok {
+		return fmt.Errorf("input type %T in not valid for unstructured conversion", in)
+	}
+
+	unstructOut, ok := out.(*unstructured.Unstructured)
+	if !ok {
+		return fmt.Errorf("output type %T in not valid for unstructured conversion", out)
+	}
+
+	outGVK := unstructOut.GroupVersionKind()
+	if !c.validVersions[outGVK.GroupVersion()] {
+		return fmt.Errorf("request to convert CRD from an invalid group/version: %s", outGVK.String())
+	}
+	inGVK := unstructIn.GroupVersionKind()
+	if !c.validVersions[inGVK.GroupVersion()] {
+		return fmt.Errorf("request to convert CRD to an invalid group/version: %s", inGVK.String())
+	}
+
+	unstructOut.SetUnstructuredContent(unstructIn.UnstructuredContent())
+	_, err := c.ConvertToVersion(unstructOut, outGVK.GroupVersion())
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *nopConverter) convertToVersion(in runtime.Object, target runtime.GroupVersioner) error {
+	kind := in.GetObjectKind().GroupVersionKind()
+	gvk, ok := target.KindForGroupVersionKinds([]schema.GroupVersionKind{kind})
+	if !ok {
+		// TODO: should this be a typed error?
+		return fmt.Errorf("%v is unstructured and is not suitable for converting to %q", kind, target)
+	}
+	if !c.validVersions[gvk.GroupVersion()] {
+		return fmt.Errorf("request to convert CRD to an invalid group/version: %s", gvk.String())
+	}
+	in.GetObjectKind().SetGroupVersionKind(gvk)
+	return nil
+}
+
+// ConvertToVersion converts in object to the given gvk in place and returns the same `in` object.
+func (c *nopConverter) ConvertToVersion(in runtime.Object, target runtime.GroupVersioner) (runtime.Object, error) {
+	var err error
+	// Run the converter on the list items instead of list itself
+	if list, ok := in.(*unstructured.UnstructuredList); ok {
+		err = list.EachListItem(func(item runtime.Object) error {
+			return c.convertToVersion(item, target)
+		})
+	}
+	err = c.convertToVersion(in, target)
+	return in, err
+}

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

@@ -18,6 +18,7 @@ package apiserver
 
 import (
 	"fmt"
+	"sort"
 	"time"
 
 	"github.com/golang/glog"
@@ -28,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/apiserver/pkg/endpoints/discovery"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
@@ -75,6 +77,7 @@ func (c *DiscoveryController) sync(version schema.GroupVersion) error {
 
 	apiVersionsForDiscovery := []metav1.GroupVersionForDiscovery{}
 	apiResourcesForDiscovery := []metav1.APIResource{}
+	versionsForDiscoveryMap := map[metav1.GroupVersion]bool{}
 
 	crds, err := c.crdLister.List(labels.Everything())
 	if err != nil {
@@ -90,13 +93,29 @@ func (c *DiscoveryController) sync(version schema.GroupVersion) error {
 		if crd.Spec.Group != version.Group {
 			continue
 		}
-		foundGroup = true
-		apiVersionsForDiscovery = append(apiVersionsForDiscovery, metav1.GroupVersionForDiscovery{
-			GroupVersion: crd.Spec.Group + "/" + crd.Spec.Version,
-			Version:      crd.Spec.Version,
-		})
 
-		if crd.Spec.Version != version.Version {
+		foundThisVersion := false
+		for _, v := range crd.Spec.Versions {
+			if !v.Served {
+				continue
+			}
+			// If there is any Served version, that means the group should show up in discovery
+			foundGroup = true
+
+			gv := metav1.GroupVersion{Group: crd.Spec.Group, Version: v.Name}
+			if !versionsForDiscoveryMap[gv] {
+				versionsForDiscoveryMap[gv] = true
+				apiVersionsForDiscovery = append(apiVersionsForDiscovery, metav1.GroupVersionForDiscovery{
+					GroupVersion: crd.Spec.Group + "/" + v.Name,
+					Version:      v.Name,
+				})
+			}
+			if v.Name == version.Version {
+				foundThisVersion = true
+			}
+		}
+
+		if !foundThisVersion {
 			continue
 		}
 		foundVersion = true
@@ -144,10 +163,13 @@ func (c *DiscoveryController) sync(version schema.GroupVersion) error {
 		return nil
 	}
 
+	sortGroupDiscoveryByKubeAwareVersion(apiVersionsForDiscovery)
+
 	apiGroup := metav1.APIGroup{
 		Name:     version.Group,
 		Versions: apiVersionsForDiscovery,
-		// the preferred versions for a group is arbitrary since there cannot be duplicate resources
+		// the preferred versions for a group is the first item in
+		// apiVersionsForDiscovery after it put in the right ordered
 		PreferredVersion: apiVersionsForDiscovery[0],
 	}
 	c.groupHandler.setDiscovery(version.Group, discovery.NewAPIGroupHandler(Codecs, apiGroup))
@@ -163,6 +185,12 @@ func (c *DiscoveryController) sync(version schema.GroupVersion) error {
 	return nil
 }
 
+func sortGroupDiscoveryByKubeAwareVersion(gd []metav1.GroupVersionForDiscovery) {
+	sort.Slice(gd, func(i, j int) bool {
+		return version.CompareKubeAwareVersionStrings(gd[i].Version, gd[j].Version) > 0
+	})
+}
+
 func (c *DiscoveryController) Run(stopCh <-chan struct{}) {
 	defer utilruntime.HandleCrash()
 	defer c.queue.ShutDown()
@@ -207,7 +235,9 @@ func (c *DiscoveryController) processNextWorkItem() bool {
 }
 
 func (c *DiscoveryController) enqueue(obj *apiextensions.CustomResourceDefinition) {
-	c.queue.Add(schema.GroupVersion{Group: obj.Spec.Group, Version: obj.Spec.Version})
+	for _, v := range obj.Spec.Versions {
+		c.queue.Add(schema.GroupVersion{Group: obj.Spec.Group, Version: v.Name})
+	}
 }
 
 func (c *DiscoveryController) addCustomResourceDefinition(obj interface{}) {
@@ -216,10 +246,14 @@ func (c *DiscoveryController) addCustomResourceDefinition(obj interface{}) {
 	c.enqueue(castObj)
 }
 
-func (c *DiscoveryController) updateCustomResourceDefinition(obj, _ interface{}) {
-	castObj := obj.(*apiextensions.CustomResourceDefinition)
-	glog.V(4).Infof("Updating customresourcedefinition %s", castObj.Name)
-	c.enqueue(castObj)
+func (c *DiscoveryController) updateCustomResourceDefinition(oldObj, newObj interface{}) {
+	castNewObj := newObj.(*apiextensions.CustomResourceDefinition)
+	castOldObj := oldObj.(*apiextensions.CustomResourceDefinition)
+	glog.V(4).Infof("Updating customresourcedefinition %s", castOldObj.Name)
+	// Enqueue both old and new object to make sure we remove and add appropriate Versions.
+	// The working queue will resolve any duplicates and only changes will stay in the queue.
+	c.enqueue(castNewObj)
+	c.enqueue(castOldObj)
 }
 
 func (c *DiscoveryController) deleteCustomResourceDefinition(obj interface{}) {

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

@@ -58,6 +58,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
+	"k8s.io/apiextensions-apiserver/pkg/apiserver/conversion"
 	apiservervalidation "k8s.io/apiextensions-apiserver/pkg/apiserver/validation"
 	informers "k8s.io/apiextensions-apiserver/pkg/client/informers/internalversion/apiextensions/internalversion"
 	listers "k8s.io/apiextensions-apiserver/pkg/client/listers/apiextensions/internalversion"
@@ -94,11 +95,20 @@ type crdInfo struct {
 	spec          *apiextensions.CustomResourceDefinitionSpec
 	acceptedNames *apiextensions.CustomResourceDefinitionNames
 
-	storage customresource.CustomResourceStorage
+	// Storage per version
+	storages map[string]customresource.CustomResourceStorage
 
-	requestScope       handlers.RequestScope
-	scaleRequestScope  handlers.RequestScope
-	statusRequestScope handlers.RequestScope
+	// Request scope per version
+	requestScopes map[string]handlers.RequestScope
+
+	// Scale scope per version
+	scaleRequestScopes map[string]handlers.RequestScope
+
+	// Status scope per version
+	statusRequestScopes map[string]handlers.RequestScope
+
+	// storageVersion is the CRD version used when storing the object in etcd.
+	storageVersion string
 }
 
 // crdStorageMap goes from customresourcedefinition to its storage
@@ -120,7 +130,6 @@ func NewCustomResourceDefinitionHandler(
 		restOptionsGetter:       restOptionsGetter,
 		admission:               admission,
 	}
-
 	crdInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		UpdateFunc: ret.updateCustomResourceDefinition,
 		DeleteFunc: func(obj interface{}) {
@@ -168,7 +177,7 @@ func (r *crdHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	if crd.Spec.Version != requestInfo.APIVersion {
+	if !apiextensions.HasServedCRDVersion(crd, requestInfo.APIVersion) {
 		r.delegate.ServeHTTP(w, req)
 		return
 	}
@@ -214,8 +223,8 @@ func (r *crdHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 }
 
 func (r *crdHandler) serveResource(w http.ResponseWriter, req *http.Request, requestInfo *apirequest.RequestInfo, crdInfo *crdInfo, terminating bool, supportedTypes []string) http.HandlerFunc {
-	requestScope := crdInfo.requestScope
-	storage := crdInfo.storage.CustomResource
+	requestScope := crdInfo.requestScopes[requestInfo.APIVersion]
+	storage := crdInfo.storages[requestInfo.APIVersion].CustomResource
 	minRequestTimeout := 1 * time.Minute
 
 	switch requestInfo.Verb {
@@ -250,8 +259,8 @@ func (r *crdHandler) serveResource(w http.ResponseWriter, req *http.Request, req
 }
 
 func (r *crdHandler) serveStatus(w http.ResponseWriter, req *http.Request, requestInfo *apirequest.RequestInfo, crdInfo *crdInfo, terminating bool, supportedTypes []string) http.HandlerFunc {
-	requestScope := crdInfo.statusRequestScope
-	storage := crdInfo.storage.Status
+	requestScope := crdInfo.statusRequestScopes[requestInfo.APIVersion]
+	storage := crdInfo.storages[requestInfo.APIVersion].Status
 
 	switch requestInfo.Verb {
 	case "get":
@@ -267,8 +276,8 @@ func (r *crdHandler) serveStatus(w http.ResponseWriter, req *http.Request, reque
 }
 
 func (r *crdHandler) serveScale(w http.ResponseWriter, req *http.Request, requestInfo *apirequest.RequestInfo, crdInfo *crdInfo, terminating bool, supportedTypes []string) http.HandlerFunc {
-	requestScope := crdInfo.scaleRequestScope
-	storage := crdInfo.storage.Scale
+	requestScope := crdInfo.scaleRequestScopes[requestInfo.APIVersion]
+	storage := crdInfo.storages[requestInfo.APIVersion].Scale
 
 	switch requestInfo.Verb {
 	case "get":
@@ -306,8 +315,10 @@ func (r *crdHandler) updateCustomResourceDefinition(oldObj, newObj interface{})
 	// as it is used without locking elsewhere.
 	storageMap2 := storageMap.clone()
 	if oldInfo, ok := storageMap2[types.UID(oldCRD.UID)]; ok {
+		for _, storage := range oldInfo.storages {
 			// destroy only the main storage. Those for the subresources share cacher and etcd clients.
-		oldInfo.storage.CustomResource.DestroyFunc()
+			storage.CustomResource.DestroyFunc()
+		}
 		delete(storageMap2, types.UID(oldCRD.UID))
 	}
 
@@ -338,9 +349,11 @@ func (r *crdHandler) removeDeadStorage() {
 			}
 		}
 		if !found {
-			glog.V(4).Infof("Removing dead CRD storage for %v", s.requestScope.Resource)
+			for version, storage := range s.storages {
+				glog.V(4).Infof("Removing dead CRD storage for %v", s.requestScopes[version].Resource)
 				// destroy only the main storage. Those for the subresources share cacher and etcd clients.
-			s.storage.CustomResource.DestroyFunc()
+				storage.CustomResource.DestroyFunc()
+			}
 			delete(storageMap2, uid)
 		}
 	}
@@ -354,7 +367,7 @@ func (r *crdHandler) GetCustomResourceListerCollectionDeleter(crd *apiextensions
 	if err != nil {
 		return nil, err
 	}
-	return info.storage.CustomResource, nil
+	return info.storages[info.storageVersion].CustomResource, nil
 }
 
 func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResourceDefinition) (*crdInfo, error) {
@@ -371,10 +384,23 @@ func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResource
 		return ret, nil
 	}
 
+	storageVersion, err := apiextensions.GetCRDStorageVersion(crd)
+	if err != nil {
+		return nil, err
+	}
+
+	// Scope/Storages per version.
+	requestScopes := map[string]handlers.RequestScope{}
+	storages := map[string]customresource.CustomResourceStorage{}
+	statusScopes := map[string]handlers.RequestScope{}
+	scaleScopes := map[string]handlers.RequestScope{}
+
+	for _, v := range crd.Spec.Versions {
+		safeConverter, unsafeConverter := conversion.NewCRDConverter(crd)
 		// In addition to Unstructured objects (Custom Resources), we also may sometimes need to
 		// decode unversioned Options objects, so we delegate to parameterScheme for such types.
 		parameterScheme := runtime.NewScheme()
-	parameterScheme.AddUnversionedTypes(schema.GroupVersion{Group: crd.Spec.Group, Version: crd.Spec.Version},
+		parameterScheme.AddUnversionedTypes(schema.GroupVersion{Group: crd.Spec.Group, Version: v.Name},
 			&metav1.ListOptions{},
 			&metav1.ExportOptions{},
 			&metav1.GetOptions{},
@@ -382,11 +408,8 @@ func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResource
 		)
 		parameterCodec := runtime.NewParameterCodec(parameterScheme)
 
-	kind := schema.GroupVersionKind{Group: crd.Spec.Group, Version: crd.Spec.Version, Kind: crd.Status.AcceptedNames.Kind}
-	typer := UnstructuredObjectTyper{
-		Delegate:          parameterScheme,
-		UnstructuredTyper: discovery.NewUnstructuredObjectTyper(),
-	}
+		kind := schema.GroupVersionKind{Group: crd.Spec.Group, Version: v.Name, Kind: crd.Status.AcceptedNames.Kind}
+		typer := newUnstructuredObjectTyper(parameterScheme)
 		creator := unstructuredCreator{}
 
 		validator, _, err := apiservervalidation.NewSchemaValidator(crd.Spec.Validation)
@@ -422,9 +445,9 @@ func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResource
 			glog.V(2).Infof("The CRD for %v has an invalid printer specification, falling back to default printing: %v", kind, err)
 		}
 
-	customResourceStorage := customresource.NewStorage(
+		storages[v.Name] = customresource.NewStorage(
 			schema.GroupResource{Group: crd.Spec.Group, Resource: crd.Status.AcceptedNames.Plural},
-		schema.GroupVersionKind{Group: crd.Spec.Group, Version: crd.Spec.Version, Kind: crd.Status.AcceptedNames.ListKind},
+			schema.GroupVersionKind{Group: crd.Spec.Group, Version: v.Name, Kind: crd.Status.AcceptedNames.ListKind},
 			customresource.NewStrategy(
 				typer,
 				crd.Spec.Scope == apiextensions.NamespaceScoped,
@@ -434,7 +457,12 @@ func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResource
 				statusSpec,
 				scaleSpec,
 			),
-		r.restOptionsGetter,
+			crdConversionRESTOptionsGetter{
+				RESTOptionsGetter: r.restOptionsGetter,
+				converter:         safeConverter,
+				decoderVersion:    schema.GroupVersion{Group: crd.Spec.Group, Version: v.Name},
+				encoderVersion:    schema.GroupVersion{Group: crd.Spec.Group, Version: storageVersion},
+			},
 			crd.Status.AcceptedNames.Categories,
 			table,
 		)
@@ -442,70 +470,73 @@ func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResource
 		selfLinkPrefix := ""
 		switch crd.Spec.Scope {
 		case apiextensions.ClusterScoped:
-		selfLinkPrefix = "/" + path.Join("apis", crd.Spec.Group, crd.Spec.Version) + "/" + crd.Status.AcceptedNames.Plural + "/"
+			selfLinkPrefix = "/" + path.Join("apis", crd.Spec.Group, v.Name) + "/" + crd.Status.AcceptedNames.Plural + "/"
 		case apiextensions.NamespaceScoped:
-		selfLinkPrefix = "/" + path.Join("apis", crd.Spec.Group, crd.Spec.Version, "namespaces") + "/"
+			selfLinkPrefix = "/" + path.Join("apis", crd.Spec.Group, v.Name, "namespaces") + "/"
 		}
 
 		clusterScoped := crd.Spec.Scope == apiextensions.ClusterScoped
 
-	requestScope := handlers.RequestScope{
+		requestScopes[v.Name] = handlers.RequestScope{
 			Namer: handlers.ContextBasedNaming{
 				SelfLinker:         meta.NewAccessor(),
 				ClusterScoped:      clusterScoped,
 				SelfLinkPathPrefix: selfLinkPrefix,
 			},
-
-		Serializer:     unstructuredNegotiatedSerializer{typer: typer, creator: creator},
+			Serializer:     unstructuredNegotiatedSerializer{typer: typer, creator: creator, converter: safeConverter},
 			ParameterCodec: parameterCodec,
 
 			Creater:         creator,
-		Convertor: crdObjectConverter{
-			UnstructuredObjectConverter: unstructured.UnstructuredObjectConverter{},
-			clusterScoped:               clusterScoped,
-		},
+			Convertor:       safeConverter,
 			Defaulter:       unstructuredDefaulter{parameterScheme},
 			Typer:           typer,
-		UnsafeConvertor: unstructured.UnstructuredObjectConverter{},
+			UnsafeConvertor: unsafeConverter,
 
-		Resource: schema.GroupVersionResource{Group: crd.Spec.Group, Version: crd.Spec.Version, Resource: crd.Status.AcceptedNames.Plural},
+			Resource: schema.GroupVersionResource{Group: crd.Spec.Group, Version: v.Name, Resource: crd.Status.AcceptedNames.Plural},
 			Kind:     kind,
 
 			MetaGroupVersion: metav1.SchemeGroupVersion,
 
-		TableConvertor: customResourceStorage.CustomResource,
-	}
-
-	ret := &crdInfo{
-		spec:          &crd.Spec,
-		acceptedNames: &crd.Status.AcceptedNames,
-
-		storage:            customResourceStorage,
-		requestScope:       requestScope,
-		scaleRequestScope:  requestScope, // shallow copy
-		statusRequestScope: requestScope, // shallow copy
+			TableConvertor: storages[v.Name].CustomResource,
 		}
 
 		// override scaleSpec subresource values
+		// shallow copy
+		scaleScope := requestScopes[v.Name]
 		scaleConverter := scale.NewScaleConverter()
-	ret.scaleRequestScope.Subresource = "scale"
-	ret.scaleRequestScope.Serializer = serializer.NewCodecFactory(scaleConverter.Scheme())
-	ret.scaleRequestScope.Kind = autoscalingv1.SchemeGroupVersion.WithKind("Scale")
-	ret.scaleRequestScope.Namer = handlers.ContextBasedNaming{
+		scaleScope.Subresource = "scale"
+		scaleScope.Serializer = serializer.NewCodecFactory(scaleConverter.Scheme())
+		scaleScope.Kind = autoscalingv1.SchemeGroupVersion.WithKind("Scale")
+		scaleScope.Namer = handlers.ContextBasedNaming{
 			SelfLinker:         meta.NewAccessor(),
 			ClusterScoped:      clusterScoped,
 			SelfLinkPathPrefix: selfLinkPrefix,
 			SelfLinkPathSuffix: "/scale",
 		}
+		scaleScopes[v.Name] = scaleScope
 
 		// override status subresource values
-	ret.statusRequestScope.Subresource = "status"
-	ret.statusRequestScope.Namer = handlers.ContextBasedNaming{
+		// shallow copy
+		statusScope := requestScopes[v.Name]
+		statusScope.Subresource = "status"
+		statusScope.Namer = handlers.ContextBasedNaming{
 			SelfLinker:         meta.NewAccessor(),
 			ClusterScoped:      clusterScoped,
 			SelfLinkPathPrefix: selfLinkPrefix,
 			SelfLinkPathSuffix: "/status",
 		}
+		statusScopes[v.Name] = statusScope
+	}
+
+	ret := &crdInfo{
+		spec:                &crd.Spec,
+		acceptedNames:       &crd.Status.AcceptedNames,
+		storages:            storages,
+		requestScopes:       requestScopes,
+		scaleRequestScopes:  scaleScopes,
+		statusRequestScopes: statusScopes,
+		storageVersion:      storageVersion,
+	}
 
 	// Copy because we cannot write to storageMap without a race
 	// as it is used without locking elsewhere.
@@ -517,27 +548,10 @@ func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResource
 	return ret, nil
 }
 
-// crdObjectConverter is a converter that supports field selectors for CRDs.
-type crdObjectConverter struct {
-	unstructured.UnstructuredObjectConverter
-	clusterScoped bool
-}
-
-func (c crdObjectConverter) ConvertFieldLabel(version, kind, label, value string) (string, string, error) {
-	// We currently only support metadata.namespace and metadata.name.
-	switch {
-	case label == "metadata.name":
-		return label, value, nil
-	case !c.clusterScoped && label == "metadata.namespace":
-		return label, value, nil
-	default:
-		return "", "", fmt.Errorf("field label not supported: %s", label)
-	}
-}
-
 type unstructuredNegotiatedSerializer struct {
 	typer     runtime.ObjectTyper
 	creator   runtime.ObjectCreater
+	converter runtime.ObjectConvertor
 }
 
 func (s unstructuredNegotiatedSerializer) SupportedMediaTypes() []runtime.SerializerInfo {
@@ -562,7 +576,7 @@ func (s unstructuredNegotiatedSerializer) SupportedMediaTypes() []runtime.Serial
 }
 
 func (s unstructuredNegotiatedSerializer) EncoderForVersion(encoder runtime.Encoder, gv runtime.GroupVersioner) runtime.Encoder {
-	return versioning.NewDefaultingCodecForScheme(Scheme, encoder, nil, gv, nil)
+	return versioning.NewCodec(encoder, nil, s.converter, Scheme, Scheme, Scheme, gv, nil)
 }
 
 func (s unstructuredNegotiatedSerializer) DecoderToVersion(decoder runtime.Decoder, gv runtime.GroupVersioner) runtime.Decoder {
@@ -574,6 +588,13 @@ type UnstructuredObjectTyper struct {
 	UnstructuredTyper runtime.ObjectTyper
 }
 
+func newUnstructuredObjectTyper(Delegate runtime.ObjectTyper) UnstructuredObjectTyper {
+	return UnstructuredObjectTyper{
+		Delegate:          Delegate,
+		UnstructuredTyper: discovery.NewUnstructuredObjectTyper(),
+	}
+}
+
 func (t UnstructuredObjectTyper) ObjectKinds(obj runtime.Object) ([]schema.GroupVersionKind, bool, error) {
 	// Delegate for things other than Unstructured.
 	if _, ok := obj.(runtime.Unstructured); !ok {
@@ -640,3 +661,20 @@ func (in crdStorageMap) clone() crdStorageMap {
 	}
 	return out
 }
+
+// crdConversionRESTOptionsGetter overrides the codec with one using the
+// provided custom converter and custom encoder and decoder version.
+type crdConversionRESTOptionsGetter struct {
+	generic.RESTOptionsGetter
+	converter      runtime.ObjectConvertor
+	encoderVersion schema.GroupVersion
+	decoderVersion schema.GroupVersion
+}
+
+func (t crdConversionRESTOptionsGetter) GetRESTOptions(resource schema.GroupResource) (generic.RESTOptions, error) {
+	ret, err := t.RESTOptionsGetter.GetRESTOptions(resource)
+	if err == nil {
+		ret.StorageConfig.Codec = versioning.NewCodec(ret.StorageConfig.Codec, ret.StorageConfig.Codec, t.converter, &unstructuredCreator{}, discovery.NewUnstructuredObjectTyper(), &unstructuredDefaulter{delegate: Scheme}, t.encoderVersion, t.decoderVersion)
+	}
+	return ret, err
+}

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

@@ -19,7 +19,8 @@ package apiserver
 import (
 	"testing"
 
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
+	conversion "k8s.io/apiextensions-apiserver/pkg/apiserver/conversion"
 )
 
 func TestConvertFieldLabel(t *testing.T) {
@@ -64,10 +65,14 @@ func TestConvertFieldLabel(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 
-			c := crdObjectConverter{
-				UnstructuredObjectConverter: unstructured.UnstructuredObjectConverter{},
-				clusterScoped:               test.clusterScoped,
+			crd := apiextensions.CustomResourceDefinition{}
+
+			if test.clusterScoped {
+				crd.Spec.Scope = apiextensions.ClusterScoped
+			} else {
+				crd.Spec.Scope = apiextensions.NamespaceScoped
 			}
+			_, c := conversion.NewCRDConverter(&crd)
 
 			label, value, err := c.ConvertFieldLabel("", "", test.label, "value")
 			if e, a := test.expectError, err != nil; e != a {

staging/src/k8s.io/apiextensions-apiserver/pkg/registry/customresourcedefinition/strategy.go

@@ -62,6 +62,15 @@ func (strategy) PrepareForCreate(ctx context.Context, obj runtime.Object) {
 	if !utilfeature.DefaultFeatureGate.Enabled(apiextensionsfeatures.CustomResourceSubresources) {
 		crd.Spec.Subresources = nil
 	}
+
+	for _, v := range crd.Spec.Versions {
+		if v.Storage {
+			if !apiextensions.IsStoredVersion(crd, v.Name) {
+				crd.Status.StoredVersions = append(crd.Status.StoredVersions, v.Name)
+			}
+			break
+		}
+	}
 }
 
 // PrepareForUpdate clears fields that are not allowed to be set by end users on update.
@@ -90,6 +99,15 @@ func (strategy) PrepareForUpdate(ctx context.Context, obj, old runtime.Object) {
 		newCRD.Spec.Subresources = nil
 		oldCRD.Spec.Subresources = nil
 	}
+
+	for _, v := range newCRD.Spec.Versions {
+		if v.Storage {
+			if !apiextensions.IsStoredVersion(newCRD, v.Name) {
+				newCRD.Status.StoredVersions = append(newCRD.Status.StoredVersions, v.Name)
+			}
+			break
+		}
+	}
 }
 
 // Validate validates a new CustomResourceDefinition.

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

@@ -68,6 +68,42 @@ func instantiateCustomResource(t *testing.T, instanceToCreate *unstructured.Unst
 	return createdInstance, nil
 }
 
+func instantiateVersionedCustomResource(t *testing.T, instanceToCreate *unstructured.Unstructured, client dynamic.ResourceInterface, definition *apiextensionsv1beta1.CustomResourceDefinition, version string) (*unstructured.Unstructured, error) {
+	createdInstance, err := client.Create(instanceToCreate)
+	if err != nil {
+		t.Logf("%#v", createdInstance)
+		return nil, err
+	}
+	createdObjectMeta, err := meta.Accessor(createdInstance)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// it should have a UUID
+	if len(createdObjectMeta.GetUID()) == 0 {
+		t.Errorf("missing uuid: %#v", createdInstance)
+	}
+	createdTypeMeta, err := meta.TypeAccessor(createdInstance)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if e, a := definition.Spec.Group+"/"+version, createdTypeMeta.GetAPIVersion(); e != a {
+		t.Errorf("expected %v, got %v", e, a)
+	}
+	if e, a := definition.Spec.Names.Kind, createdTypeMeta.GetKind(); e != a {
+		t.Errorf("expected %v, got %v", e, a)
+	}
+	return createdInstance, nil
+}
+
+func NewNamespacedCustomResourceVersionedClient(ns string, client dynamic.Interface, crd *apiextensionsv1beta1.CustomResourceDefinition, version string) dynamic.ResourceInterface {
+	gvr := schema.GroupVersionResource{Group: crd.Spec.Group, Version: version, Resource: crd.Spec.Names.Plural}
+
+	if crd.Spec.Scope != apiextensionsv1beta1.ClusterScoped {
+		return client.Resource(gvr).Namespace(ns)
+	}
+	return client.Resource(gvr)
+}
+
 func NewNamespacedCustomResourceClient(ns string, client dynamic.Interface, crd *apiextensionsv1beta1.CustomResourceDefinition) dynamic.ResourceInterface {
 	gvr := schema.GroupVersionResource{Group: crd.Spec.Group, Version: crd.Spec.Version, Resource: crd.Spec.Names.Plural}
 

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

@@ -100,6 +100,62 @@ func NewNoxuInstance(namespace, name string) *unstructured.Unstructured {
 	}
 }
 
+func NewMultipleVersionNoxuCRD(scope apiextensionsv1beta1.ResourceScope) *apiextensionsv1beta1.CustomResourceDefinition {
+	return &apiextensionsv1beta1.CustomResourceDefinition{
+		ObjectMeta: metav1.ObjectMeta{Name: "noxus.mygroup.example.com"},
+		Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{
+			Group:   "mygroup.example.com",
+			Version: "v1beta1",
+			Names: apiextensionsv1beta1.CustomResourceDefinitionNames{
+				Plural:     "noxus",
+				Singular:   "nonenglishnoxu",
+				Kind:       "WishIHadChosenNoxu",
+				ShortNames: []string{"foo", "bar", "abc", "def"},
+				ListKind:   "NoxuItemList",
+				Categories: []string{"all"},
+			},
+			Scope: scope,
+			Versions: []apiextensionsv1beta1.CustomResourceDefinitionVersion{
+				{
+					Name:    "v1beta1",
+					Served:  true,
+					Storage: false,
+				},
+				{
+					Name:    "v1beta2",
+					Served:  true,
+					Storage: true,
+				},
+				{
+					Name:    "v0",
+					Served:  false,
+					Storage: false,
+				},
+			},
+		},
+	}
+}
+
+func NewVersionedNoxuInstance(namespace, name, version string) *unstructured.Unstructured {
+	return &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"apiVersion": "mygroup.example.com/" + version,
+			"kind":       "WishIHadChosenNoxu",
+			"metadata": map[string]interface{}{
+				"namespace": namespace,
+				"name":      name,
+			},
+			"content": map[string]interface{}{
+				"key": "value",
+			},
+			"num": map[string]interface{}{
+				"num1": noxuInstanceNum,
+				"num2": 1000000,
+			},
+		},
+	}
+}
+
 func NewNoxu2CustomResourceDefinition(scope apiextensionsv1beta1.ResourceScope) *apiextensionsv1beta1.CustomResourceDefinition {
 	return &apiextensionsv1beta1.CustomResourceDefinition{
 		ObjectMeta: metav1.ObjectMeta{Name: "noxus2.mygroup.example.com"},

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

@@ -0,0 +1,304 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package integration
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
+	"k8s.io/apiextensions-apiserver/test/integration/testserver"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/dynamic"
+)
+
+func TestVersionedNamspacedScopedCRD(t *testing.T) {
+	stopCh, apiExtensionClient, dynamicClient, err := testserver.StartDefaultServerWithClients()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer close(stopCh)
+
+	noxuDefinition := testserver.NewMultipleVersionNoxuCRD(apiextensionsv1beta1.NamespaceScoped)
+	err = testserver.CreateNewCustomResourceDefinition(noxuDefinition, apiExtensionClient, dynamicClient)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ns := "not-the-default"
+	testSimpleVersionedCRUD(t, ns, noxuDefinition, dynamicClient)
+}
+
+func TestVersionedClusterScopedCRD(t *testing.T) {
+	stopCh, apiExtensionClient, dynamicClient, err := testserver.StartDefaultServerWithClients()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer close(stopCh)
+
+	noxuDefinition := testserver.NewMultipleVersionNoxuCRD(apiextensionsv1beta1.ClusterScoped)
+	err = testserver.CreateNewCustomResourceDefinition(noxuDefinition, apiExtensionClient, dynamicClient)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ns := ""
+	testSimpleVersionedCRUD(t, ns, noxuDefinition, dynamicClient)
+}
+
+func TestStoragedVersionInNamespacedCRDStatus(t *testing.T) {
+	noxuDefinition := testserver.NewMultipleVersionNoxuCRD(apiextensionsv1beta1.NamespaceScoped)
+	ns := "not-the-default"
+	testStoragedVersionInCRDStatus(t, ns, noxuDefinition)
+}
+
+func TestStoragedVersionInClusterScopedCRDStatus(t *testing.T) {
+	noxuDefinition := testserver.NewMultipleVersionNoxuCRD(apiextensionsv1beta1.ClusterScoped)
+	ns := ""
+	testStoragedVersionInCRDStatus(t, ns, noxuDefinition)
+}
+
+func testStoragedVersionInCRDStatus(t *testing.T, ns string, noxuDefinition *apiextensionsv1beta1.CustomResourceDefinition) {
+	versionsV1Beta1Storage := []apiextensionsv1beta1.CustomResourceDefinitionVersion{
+		{
+			Name:    "v1beta1",
+			Served:  true,
+			Storage: true,
+		},
+		{
+			Name:    "v1beta2",
+			Served:  true,
+			Storage: false,
+		},
+	}
+	versionsV1Beta2Storage := []apiextensionsv1beta1.CustomResourceDefinitionVersion{
+		{
+			Name:    "v1beta1",
+			Served:  true,
+			Storage: false,
+		},
+		{
+			Name:    "v1beta2",
+			Served:  true,
+			Storage: true,
+		},
+	}
+	stopCh, apiExtensionClient, dynamicClient, err := testserver.StartDefaultServerWithClients()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer close(stopCh)
+
+	noxuDefinition.Spec.Versions = versionsV1Beta1Storage
+	err = testserver.CreateNewCustomResourceDefinition(noxuDefinition, apiExtensionClient, dynamicClient)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// The storage version list should be initilized to storage version
+	crd, err := testserver.GetCustomResourceDefinition(noxuDefinition, apiExtensionClient)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if e, a := []string{"v1beta1"}, crd.Status.StoredVersions; !reflect.DeepEqual(e, a) {
+		t.Errorf("expected %v, got %v", e, a)
+	}
+
+	// Changing CRD storage version should be reflected immediately
+	crd.Spec.Versions = versionsV1Beta2Storage
+	_, err = apiExtensionClient.ApiextensionsV1beta1().CustomResourceDefinitions().Update(crd)
+	if err != nil {
+		t.Fatal(err)
+	}
+	crd, err = testserver.GetCustomResourceDefinition(noxuDefinition, apiExtensionClient)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if e, a := []string{"v1beta1", "v1beta2"}, crd.Status.StoredVersions; !reflect.DeepEqual(e, a) {
+		t.Errorf("expected %v, got %v", e, a)
+	}
+
+	err = testserver.DeleteCustomResourceDefinition(crd, apiExtensionClient)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func testSimpleVersionedCRUD(t *testing.T, ns string, noxuDefinition *apiextensionsv1beta1.CustomResourceDefinition, dynamicClient dynamic.Interface) {
+	noxuResourceClients := map[string]dynamic.ResourceInterface{}
+	noxuWatchs := map[string]watch.Interface{}
+	disbaledVersions := map[string]bool{}
+	for _, v := range noxuDefinition.Spec.Versions {
+		disbaledVersions[v.Name] = !v.Served
+	}
+	for _, v := range noxuDefinition.Spec.Versions {
+		noxuResourceClients[v.Name] = NewNamespacedCustomResourceVersionedClient(ns, dynamicClient, noxuDefinition, v.Name)
+
+		noxuWatch, err := noxuResourceClients[v.Name].Watch(metav1.ListOptions{})
+		if disbaledVersions[v.Name] {
+			if err == nil {
+				t.Errorf("expected the watch creation fail for disabled version %s", v.Name)
+			}
+		} else {
+			if err != nil {
+				t.Fatal(err)
+			}
+			noxuWatchs[v.Name] = noxuWatch
+		}
+	}
+	defer func() {
+		for _, w := range noxuWatchs {
+			w.Stop()
+		}
+	}()
+
+	for version, noxuResourceClient := range noxuResourceClients {
+		createdNoxuInstance, err := instantiateVersionedCustomResource(t, testserver.NewVersionedNoxuInstance(ns, "foo", version), noxuResourceClient, noxuDefinition, version)
+		if disbaledVersions[version] {
+			if err == nil {
+				t.Errorf("expected the CR creation fail for disabled version %s", version)
+			}
+			continue
+		}
+		if err != nil {
+			t.Fatalf("unable to create noxu Instance:%v", err)
+		}
+		if e, a := noxuDefinition.Spec.Group+"/"+version, createdNoxuInstance.GetAPIVersion(); e != a {
+			t.Errorf("expected %v, got %v", e, a)
+		}
+		for watchVersion, noxuWatch := range noxuWatchs {
+			select {
+			case watchEvent := <-noxuWatch.ResultChan():
+				if e, a := watch.Added, watchEvent.Type; e != a {
+					t.Errorf("expected %v, got %v", e, a)
+					break
+				}
+				createdObjectMeta, err := meta.Accessor(watchEvent.Object)
+				if err != nil {
+					t.Fatal(err)
+				}
+				// it should have a UUID
+				if len(createdObjectMeta.GetUID()) == 0 {
+					t.Errorf("missing uuid: %#v", watchEvent.Object)
+				}
+				if e, a := ns, createdObjectMeta.GetNamespace(); e != a {
+					t.Errorf("expected %v, got %v", e, a)
+				}
+				createdTypeMeta, err := meta.TypeAccessor(watchEvent.Object)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if e, a := noxuDefinition.Spec.Group+"/"+watchVersion, createdTypeMeta.GetAPIVersion(); e != a {
+					t.Errorf("expected %v, got %v", e, a)
+				}
+				if e, a := noxuDefinition.Spec.Names.Kind, createdTypeMeta.GetKind(); e != a {
+					t.Errorf("expected %v, got %v", e, a)
+				}
+			case <-time.After(5 * time.Second):
+				t.Errorf("missing watch event")
+			}
+		}
+
+		// Check get for all versions
+		for version2, noxuResourceClient2 := range noxuResourceClients {
+			// Get test
+			gottenNoxuInstance, err := noxuResourceClient2.Get("foo", metav1.GetOptions{})
+
+			if disbaledVersions[version2] {
+				if err == nil {
+					t.Errorf("expected the get operation fail for disabled version %s", version2)
+				}
+			} else {
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				if e, a := version2, gottenNoxuInstance.GroupVersionKind().Version; !reflect.DeepEqual(e, a) {
+					t.Errorf("expected %v, got %v", e, a)
+				}
+			}
+
+			// List test
+			listWithItem, err := noxuResourceClient2.List(metav1.ListOptions{})
+			if disbaledVersions[version2] {
+				if err == nil {
+					t.Errorf("expected the list operation fail for disabled version %s", version2)
+				}
+			} else {
+				if err != nil {
+					t.Fatal(err)
+				}
+				if e, a := 1, len(listWithItem.Items); e != a {
+					t.Errorf("expected %v, got %v", e, a)
+				}
+				if e, a := version2, listWithItem.GroupVersionKind().Version; !reflect.DeepEqual(e, a) {
+					t.Errorf("expected %v, got %v", e, a)
+				}
+				if e, a := version2, listWithItem.Items[0].GroupVersionKind().Version; !reflect.DeepEqual(e, a) {
+					t.Errorf("expected %v, got %v", e, a)
+				}
+			}
+		}
+
+		// Delete test
+		if err := noxuResourceClient.Delete("foo", metav1.NewDeleteOptions(0)); err != nil {
+			t.Fatal(err)
+		}
+
+		listWithoutItem, err := noxuResourceClient.List(metav1.ListOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if e, a := 0, len(listWithoutItem.Items); e != a {
+			t.Errorf("expected %v, got %v", e, a)
+		}
+
+		for _, noxuWatch := range noxuWatchs {
+			select {
+			case watchEvent := <-noxuWatch.ResultChan():
+				if e, a := watch.Deleted, watchEvent.Type; e != a {
+					t.Errorf("expected %v, got %v", e, a)
+					break
+				}
+				deletedObjectMeta, err := meta.Accessor(watchEvent.Object)
+				if err != nil {
+					t.Fatal(err)
+				}
+				// it should have a UUID
+				createdObjectMeta, err := meta.Accessor(createdNoxuInstance)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if e, a := createdObjectMeta.GetUID(), deletedObjectMeta.GetUID(); e != a {
+					t.Errorf("expected %v, got %v", e, a)
+				}
+
+			case <-time.After(5 * time.Second):
+				t.Errorf("missing watch event")
+			}
+		}
+
+		// Delete test
+		if err := noxuResourceClient.DeleteCollection(metav1.NewDeleteOptions(0), metav1.ListOptions{}); err != nil {
+			t.Fatal(err)
+		}
+
+	}
+}