Merge pull request #77817 from smarterclayton/apicrd

CRDs should support watch of protobuf PartialObjectMetadata

cmd/kube-apiserver/app/testing/BUILD

@@ -13,6 +13,8 @@ go_library(
     deps = [
         "//cmd/kube-apiserver/app:go_default_library",
         "//cmd/kube-apiserver/app/options:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/registry/generic/registry:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/storagebackend:go_default_library",

cmd/kube-apiserver/app/testing/testserver.go

@@ -27,6 +27,8 @@ import (
 
 	pflag "github.com/spf13/pflag"
 
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/registry/generic/registry"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
@@ -158,6 +160,8 @@ func StartTestServer(t Logger, instanceOptions *TestServerInstanceOptions, custo
 	if err != nil {
 		return result, fmt.Errorf("failed to create a client: %v", err)
 	}
+
+	// wait until healthz endpoint returns ok
 	err = wait.Poll(100*time.Millisecond, 30*time.Second, func() (bool, error) {
 		select {
 		case err := <-errCh:
@@ -177,6 +181,26 @@ func StartTestServer(t Logger, instanceOptions *TestServerInstanceOptions, custo
 		return result, fmt.Errorf("failed to wait for /healthz to return ok: %v", err)
 	}
 
+	// wait until default namespace is created
+	err = wait.Poll(100*time.Millisecond, 30*time.Second, func() (bool, error) {
+		select {
+		case err := <-errCh:
+			return false, err
+		default:
+		}
+
+		if _, err := client.CoreV1().Namespaces().Get("default", metav1.GetOptions{}); err != nil {
+			if !errors.IsNotFound(err) {
+				t.Logf("Unable to get default namespace: %v", err)
+			}
+			return false, nil
+		}
+		return true, nil
+	})
+	if err != nil {
+		return result, fmt.Errorf("failed to wait for default namespace to be created: %v", err)
+	}
+
 	// from here the caller must call tearDown
 	result.ClientConfig = server.LoopbackClientConfig
 	result.ClientConfig.QPS = 1000

staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/BUILD

@@ -53,6 +53,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/serializer/json:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/runtime/serializer/protobuf:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/serializer/versioning:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",

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

@@ -53,6 +53,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/apimachinery/pkg/runtime/serializer/json"
+	"k8s.io/apimachinery/pkg/runtime/serializer/protobuf"
 	"k8s.io/apimachinery/pkg/runtime/serializer/versioning"
 	"k8s.io/apimachinery/pkg/types"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -628,21 +629,32 @@ func (r *crdHandler) getOrCreateServingInfoFor(crd *apiextensions.CustomResource
 
 		clusterScoped := crd.Spec.Scope == apiextensions.ClusterScoped
 
-		requestScopes[v.Name] = &handlers.RequestScope{
+		// CRDs explicitly do not support protobuf, but some objects returned by the API server do
-			Namer: handlers.ContextBasedNaming{
+		negotiatedSerializer := unstructuredNegotiatedSerializer{
-				SelfLinker:         meta.NewAccessor(),
-				ClusterScoped:      clusterScoped,
-				SelfLinkPathPrefix: selfLinkPrefix,
-			},
-			Serializer: unstructuredNegotiatedSerializer{
 			typer:                 typer,
 			creator:               creator,
 			converter:             safeConverter,
 			structuralSchemas:     structuralSchemas,
 			structuralSchemaGK:    kind.GroupKind(),
 			preserveUnknownFields: *crd.Spec.PreserveUnknownFields,
+		}
+		var standardSerializers []runtime.SerializerInfo
+		for _, s := range negotiatedSerializer.SupportedMediaTypes() {
+			if s.MediaType == runtime.ContentTypeProtobuf {
+				continue
+			}
+			standardSerializers = append(standardSerializers, s)
+		}
+
+		requestScopes[v.Name] = &handlers.RequestScope{
+			Namer: handlers.ContextBasedNaming{
+				SelfLinker:         meta.NewAccessor(),
+				ClusterScoped:      clusterScoped,
+				SelfLinkPathPrefix: selfLinkPrefix,
 			},
+			Serializer:          negotiatedSerializer,
 			ParameterCodec:      parameterCodec,
+			StandardSerializers: standardSerializers,
 
 			Creater:         creator,
 			Convertor:       safeConverter,
@@ -763,6 +775,16 @@ func (s unstructuredNegotiatedSerializer) SupportedMediaTypes() []runtime.Serial
 			EncodesAsText:    true,
 			Serializer:       json.NewYAMLSerializer(json.DefaultMetaFactory, s.creator, s.typer),
 		},
+		{
+			MediaType:        "application/vnd.kubernetes.protobuf",
+			MediaTypeType:    "application",
+			MediaTypeSubType: "vnd.kubernetes.protobuf",
+			Serializer:       protobuf.NewSerializer(s.creator, s.typer),
+			StreamSerializer: &runtime.StreamSerializerInfo{
+				Serializer: protobuf.NewRawSerializer(s.creator, s.typer),
+				Framer:     protobuf.LengthDelimitedFramer,
+			},
+		},
 	}
 }
 

staging/src/k8s.io/apimachinery/pkg/api/errors/errors.go

@@ -394,7 +394,11 @@ func NewGenericServerResponse(code int, verb string, qualifiedResource schema.Gr
 	case http.StatusNotAcceptable:
 		reason = metav1.StatusReasonNotAcceptable
 		// the server message has details about what types are acceptable
+		if len(serverMessage) == 0 || serverMessage == "unknown" {
+			message = "the server was unable to respond with a content type that the client supports"
+		} else {
 			message = serverMessage
+		}
 	case http.StatusUnsupportedMediaType:
 		reason = metav1.StatusReasonUnsupportedMediaType
 		// the server message has details about what types are acceptable

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

@@ -69,8 +69,6 @@ func IsNotMarshalable(err error) bool {
 // NewSerializer creates a Protobuf serializer that handles encoding versioned objects into the proper wire form. If a typer
 // is passed, the encoded object will have group, version, and kind fields set. If typer is nil, the objects will be written
 // as-is (any type info passed with the object will be used).
-//
-// This encoding scheme is experimental, and is subject to change at any time.
 func NewSerializer(creater runtime.ObjectCreater, typer runtime.ObjectTyper) *Serializer {
 	return &Serializer{
 		prefix:  protoEncodingPrefix,

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/negotiation/negotiate.go

@@ -116,9 +116,10 @@ func isPrettyPrint(req *http.Request) bool {
 // EndpointRestrictions is an interface that allows content-type negotiation
 // to verify server support for specific options
 type EndpointRestrictions interface {
-	// AllowsConversion should return true if the specified group version kind
+	// AllowsMediaTypeTransform returns true if the endpoint allows either the requested mime type
-	// is an allowed target object.
+	// or the requested transformation. If false, the caller should ignore this mime type. If the
-	AllowsConversion(target schema.GroupVersionKind, mimeType, mimeSubType string) bool
+	// target is nil, the client is not requesting a transformation.
+	AllowsMediaTypeTransform(mimeType, mimeSubType string, target *schema.GroupVersionKind) bool
 	// AllowsServerVersion should return true if the specified version is valid
 	// for the server group.
 	AllowsServerVersion(version string) bool
@@ -133,8 +134,8 @@ var DefaultEndpointRestrictions = emptyEndpointRestrictions{}
 
 type emptyEndpointRestrictions struct{}
 
-func (emptyEndpointRestrictions) AllowsConversion(schema.GroupVersionKind, string, string) bool {
+func (emptyEndpointRestrictions) AllowsMediaTypeTransform(mimeType string, mimeSubType string, gvk *schema.GroupVersionKind) bool {
-	return false
+	return gvk == nil
 }
 func (emptyEndpointRestrictions) AllowsServerVersion(string) bool  { return false }
 func (emptyEndpointRestrictions) AllowsStreamSchema(s string) bool { return s == "watch" }
@@ -225,7 +226,7 @@ func acceptMediaTypeOptions(params map[string]string, accepts *runtime.Serialize
 		}
 	}
 
-	if options.Convert != nil && !endpoint.AllowsConversion(*options.Convert, accepts.MediaTypeType, accepts.MediaTypeSubType) {
+	if !endpoint.AllowsMediaTypeTransform(accepts.MediaTypeType, accepts.MediaTypeSubType, options.Convert) {
 		return MediaTypeOptions{}, false
 	}
 

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

@@ -50,6 +50,11 @@ type RequestScope struct {
 	Serializer runtime.NegotiatedSerializer
 	runtime.ParameterCodec
 
+	// StandardSerializers, if set, restricts which serializers can be used when
+	// we aren't transforming the output (into Table or PartialObjectMetadata).
+	// Used only by CRDs which do not yet support Protobuf.
+	StandardSerializers []runtime.SerializerInfo
+
 	Creater         runtime.ObjectCreater
 	Convertor       runtime.ObjectConvertor
 	Defaulter       runtime.ObjectDefaulter
@@ -78,7 +83,21 @@ func (scope *RequestScope) err(err error, w http.ResponseWriter, req *http.Reque
 	responsewriters.ErrorNegotiated(err, scope.Serializer, scope.Kind.GroupVersion(), w, req)
 }
 
-func (scope *RequestScope) AllowsConversion(gvk schema.GroupVersionKind, mimeType, mimeSubType string) bool {
+func (scope *RequestScope) AllowsMediaTypeTransform(mimeType, mimeSubType string, gvk *schema.GroupVersionKind) bool {
+	// some handlers like CRDs can't serve all the mime types that PartialObjectMetadata or Table can - if
+	// gvk is nil (no conversion) allow StandardSerializers to further restrict the set of mime types.
+	if gvk == nil {
+		if len(scope.StandardSerializers) == 0 {
+			return true
+		}
+		for _, info := range scope.StandardSerializers {
+			if info.MediaTypeType == mimeType && info.MediaTypeSubType == mimeSubType {
+				return true
+			}
+		}
+		return false
+	}
+
 	// TODO: this is temporary, replace with an abstraction calculated at endpoint installation time
 	if gvk.GroupVersion() == metav1beta1.SchemeGroupVersion || gvk.GroupVersion() == metav1.SchemeGroupVersion {
 		switch gvk.Kind {

test/integration/apiserver/apiserver_test.go

@@ -404,6 +404,7 @@ func TestNameInFieldSelector(t *testing.T) {
 }
 
 func TestAPICRDProtobuf(t *testing.T) {
+	testNamespace := "test-api-crd-protobuf"
 	tearDown, config, _, err := fixtures.StartDefaultServer(t)
 	if err != nil {
 		t.Fatal(err)
@@ -442,7 +443,7 @@ func TestAPICRDProtobuf(t *testing.T) {
 		t.Fatal(err)
 	}
 	crdGVR := schema.GroupVersionResource{Group: fooCRD.Spec.Group, Version: fooCRD.Spec.Version, Resource: "foos"}
-	crclient := dynamicClient.Resource(crdGVR).Namespace("default")
+	crclient := dynamicClient.Resource(crdGVR).Namespace(testNamespace)
 
 	testcases := []struct {
 		name     string
@@ -452,7 +453,7 @@ func TestAPICRDProtobuf(t *testing.T) {
 		wantBody func(*testing.T, io.Reader)
 	}{
 		{
-			name:   "server returns 406 when asking for protobuf for CRDs",
+			name:   "server returns 406 when asking for protobuf for CRDs, which dynamic client does not support",
 			accept: "application/vnd.kubernetes.protobuf",
 			object: func(t *testing.T) (metav1.Object, string, string) {
 				cr, err := crclient.Create(&unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "cr.bar.com/v1", "kind": "Foo", "metadata": map[string]interface{}{"name": "test-1"}}}, metav1.CreateOptions{})
@@ -468,9 +469,15 @@ func TestAPICRDProtobuf(t *testing.T) {
 				if !apierrors.IsNotAcceptable(err) {
 					t.Fatal(err)
 				}
-				// TODO: this should be a more specific error
+				status := err.(apierrors.APIStatus).Status()
-				if err.Error() != "only the following media types are accepted: application/json, application/yaml" {
+				data, _ := json.MarshalIndent(status, "", "  ")
-					t.Fatal(err)
+				// because the dynamic client only has a json serializer, the client processing of the error cannot
+				// turn the response into something meaningful, so we verify that fallback handling works correctly
+				if !apierrors.IsUnexpectedServerError(err) {
+					t.Fatal(string(data))
+				}
+				if status.Message != "the server was unable to respond with a content type that the client supports (get foos.cr.bar.com test-1)" {
+					t.Fatal(string(data))
 				}
 			},
 		},
@@ -548,6 +555,7 @@ func TestAPICRDProtobuf(t *testing.T) {
 }
 
 func TestTransform(t *testing.T) {
+	testNamespace := "test-transform"
 	tearDown, config, _, err := fixtures.StartDefaultServer(t)
 	if err != nil {
 		t.Fatal(err)
@@ -586,7 +594,7 @@ func TestTransform(t *testing.T) {
 		t.Fatal(err)
 	}
 	crdGVR := schema.GroupVersionResource{Group: fooCRD.Spec.Group, Version: fooCRD.Spec.Version, Resource: "foos"}
-	crclient := dynamicClient.Resource(crdGVR).Namespace("default")
+	crclient := dynamicClient.Resource(crdGVR).Namespace(testNamespace)
 
 	testcases := []struct {
 		name          string
@@ -658,7 +666,7 @@ func TestTransform(t *testing.T) {
 					t.Fatal(err)
 				}
 				// TODO: this should be a more specific error
-				if err.Error() != "only the following media types are accepted: application/json;stream=watch" {
+				if err.Error() != "only the following media types are accepted: application/json;stream=watch, application/vnd.kubernetes.protobuf;stream=watch" {
 					t.Fatal(err)
 				}
 			},
@@ -667,12 +675,11 @@ func TestTransform(t *testing.T) {
 			name:   "v1beta1 verify columns on services",
 			accept: "application/json;as=Table;g=meta.k8s.io;v=v1beta1",
 			object: func(t *testing.T) (metav1.Object, string, string) {
-				ns := "default"
+				svc, err := clientset.CoreV1().Services(testNamespace).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-1"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
-				svc, err := clientset.CoreV1().Services(ns).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-1"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
 				if err != nil {
 					t.Fatalf("unable to create service: %v", err)
 				}
-				if _, err := clientset.CoreV1().Services(ns).Patch(svc.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
+				if _, err := clientset.CoreV1().Services(testNamespace).Patch(svc.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
 					t.Fatalf("unable to update service: %v", err)
 				}
 				return svc, "", "services"
@@ -686,12 +693,11 @@ func TestTransform(t *testing.T) {
 			accept:        "application/json;as=Table;g=meta.k8s.io;v=v1beta1",
 			includeObject: metav1.IncludeNone,
 			object: func(t *testing.T) (metav1.Object, string, string) {
-				ns := "default"
+				obj, err := clientset.CoreV1().Services(testNamespace).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-2"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
-				obj, err := clientset.CoreV1().Services(ns).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-2"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
 				if err != nil {
 					t.Fatalf("unable to create object: %v", err)
 				}
-				if _, err := clientset.CoreV1().Services(ns).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
+				if _, err := clientset.CoreV1().Services(testNamespace).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
 					t.Fatalf("unable to update object: %v", err)
 				}
 				return obj, "", "services"
@@ -705,12 +711,11 @@ func TestTransform(t *testing.T) {
 			accept:        "application/json;as=Table;g=meta.k8s.io;v=v1beta1",
 			includeObject: metav1.IncludeObject,
 			object: func(t *testing.T) (metav1.Object, string, string) {
-				ns := "default"
+				obj, err := clientset.CoreV1().Services(testNamespace).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-3"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
-				obj, err := clientset.CoreV1().Services(ns).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-3"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
 				if err != nil {
 					t.Fatalf("unable to create object: %v", err)
 				}
-				if _, err := clientset.CoreV1().Services(ns).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
+				if _, err := clientset.CoreV1().Services(testNamespace).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
 					t.Fatalf("unable to update object: %v", err)
 				}
 				return obj, "", "services"
@@ -730,12 +735,11 @@ func TestTransform(t *testing.T) {
 			name:   "v1beta1 verify partial metadata object on config maps",
 			accept: "application/json;as=PartialObjectMetadata;g=meta.k8s.io;v=v1beta1",
 			object: func(t *testing.T) (metav1.Object, string, string) {
-				ns := "default"
+				obj, err := clientset.CoreV1().ConfigMaps(testNamespace).Create(&v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "test-1", Annotations: map[string]string{"test": "0"}}})
-				obj, err := clientset.CoreV1().ConfigMaps(ns).Create(&v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "test-1", Annotations: map[string]string{"test": "0"}}})
 				if err != nil {
 					t.Fatalf("unable to create object: %v", err)
 				}
-				if _, err := clientset.CoreV1().ConfigMaps(ns).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
+				if _, err := clientset.CoreV1().ConfigMaps(testNamespace).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
 					t.Fatalf("unable to update object: %v", err)
 				}
 				return obj, "", "configmaps"
@@ -748,12 +752,11 @@ func TestTransform(t *testing.T) {
 			name:   "v1beta1 verify partial metadata object on config maps in protobuf",
 			accept: "application/vnd.kubernetes.protobuf;as=PartialObjectMetadata;g=meta.k8s.io;v=v1beta1",
 			object: func(t *testing.T) (metav1.Object, string, string) {
-				ns := "default"
+				obj, err := clientset.CoreV1().ConfigMaps(testNamespace).Create(&v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "test-2", Annotations: map[string]string{"test": "0"}}})
-				obj, err := clientset.CoreV1().ConfigMaps(ns).Create(&v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "test-2", Annotations: map[string]string{"test": "0"}}})
 				if err != nil {
 					t.Fatalf("unable to create object: %v", err)
 				}
-				if _, err := clientset.CoreV1().ConfigMaps(ns).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
+				if _, err := clientset.CoreV1().ConfigMaps(testNamespace).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
 					t.Fatalf("unable to update object: %v", err)
 				}
 				return obj, "", "configmaps"
@@ -762,6 +765,23 @@ func TestTransform(t *testing.T) {
 				expectPartialObjectMetaEventsProtobuf(t, w, "0", "1")
 			},
 		},
+		{
+			name:   "v1beta1 verify partial metadata object on CRDs in protobuf",
+			accept: "application/vnd.kubernetes.protobuf;as=PartialObjectMetadata;g=meta.k8s.io;v=v1beta1",
+			object: func(t *testing.T) (metav1.Object, string, string) {
+				cr, err := crclient.Create(&unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "cr.bar.com/v1", "kind": "Foo", "metadata": map[string]interface{}{"name": "test-4", "annotations": map[string]string{"test": "0"}}}}, metav1.CreateOptions{})
+				if err != nil {
+					t.Fatalf("unable to create cr: %v", err)
+				}
+				if _, err := crclient.Patch("test-4", types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`), metav1.PatchOptions{}); err != nil {
+					t.Fatalf("unable to patch cr: %v", err)
+				}
+				return cr, crdGVR.Group, "foos"
+			},
+			wantBody: func(t *testing.T, w io.Reader) {
+				expectPartialObjectMetaEventsProtobuf(t, w, "0", "1")
+			},
+		},
 		{
 			name:   "v1beta1 verify error on unsupported mimetype protobuf for table conversion",
 			accept: "application/vnd.kubernetes.protobuf;as=Table;g=meta.k8s.io;v=v1beta1",
@@ -853,23 +873,6 @@ func TestTransform(t *testing.T) {
 		{
 			name:   "v1 verify columns on CRDs in json",
 			accept: "application/json;as=Table;g=meta.k8s.io;v=v1",
-			object: func(t *testing.T) (metav1.Object, string, string) {
-				cr, err := crclient.Create(&unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "cr.bar.com/v1", "kind": "Foo", "metadata": map[string]interface{}{"name": "test-4"}}}, metav1.CreateOptions{})
-				if err != nil {
-					t.Fatalf("unable to create cr: %v", err)
-				}
-				if _, err := crclient.Patch("test-4", types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`), metav1.PatchOptions{}); err != nil {
-					t.Fatalf("unable to patch cr: %v", err)
-				}
-				return cr, crdGVR.Group, "foos"
-			},
-			wantBody: func(t *testing.T, w io.Reader) {
-				expectTableV1WatchEvents(t, 2, 2, metav1.IncludeMetadata, json.NewDecoder(w))
-			},
-		},
-		{
-			name:   "v1 verify columns on CRDs in json;stream=watch",
-			accept: "application/json;stream=watch;as=Table;g=meta.k8s.io;v=v1",
 			object: func(t *testing.T) (metav1.Object, string, string) {
 				cr, err := crclient.Create(&unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "cr.bar.com/v1", "kind": "Foo", "metadata": map[string]interface{}{"name": "test-5"}}}, metav1.CreateOptions{})
 				if err != nil {
@@ -885,8 +888,8 @@ func TestTransform(t *testing.T) {
 			},
 		},
 		{
-			name:   "v1 verify columns on CRDs in yaml",
+			name:   "v1 verify columns on CRDs in json;stream=watch",
-			accept: "application/yaml;as=Table;g=meta.k8s.io;v=v1",
+			accept: "application/json;stream=watch;as=Table;g=meta.k8s.io;v=v1",
 			object: func(t *testing.T) (metav1.Object, string, string) {
 				cr, err := crclient.Create(&unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "cr.bar.com/v1", "kind": "Foo", "metadata": map[string]interface{}{"name": "test-6"}}}, metav1.CreateOptions{})
 				if err != nil {
@@ -897,12 +900,29 @@ func TestTransform(t *testing.T) {
 				}
 				return cr, crdGVR.Group, "foos"
 			},
+			wantBody: func(t *testing.T, w io.Reader) {
+				expectTableV1WatchEvents(t, 2, 2, metav1.IncludeMetadata, json.NewDecoder(w))
+			},
+		},
+		{
+			name:   "v1 verify columns on CRDs in yaml",
+			accept: "application/yaml;as=Table;g=meta.k8s.io;v=v1",
+			object: func(t *testing.T) (metav1.Object, string, string) {
+				cr, err := crclient.Create(&unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "cr.bar.com/v1", "kind": "Foo", "metadata": map[string]interface{}{"name": "test-7"}}}, metav1.CreateOptions{})
+				if err != nil {
+					t.Fatalf("unable to create cr: %v", err)
+				}
+				if _, err := crclient.Patch("test-7", types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`), metav1.PatchOptions{}); err != nil {
+					t.Fatalf("unable to patch cr: %v", err)
+				}
+				return cr, crdGVR.Group, "foos"
+			},
 			wantErr: func(t *testing.T, err error) {
 				if !apierrors.IsNotAcceptable(err) {
 					t.Fatal(err)
 				}
 				// TODO: this should be a more specific error
-				if err.Error() != "only the following media types are accepted: application/json;stream=watch" {
+				if err.Error() != "only the following media types are accepted: application/json;stream=watch, application/vnd.kubernetes.protobuf;stream=watch" {
 					t.Fatal(err)
 				}
 			},
@@ -911,12 +931,11 @@ func TestTransform(t *testing.T) {
 			name:   "v1 verify columns on services",
 			accept: "application/json;as=Table;g=meta.k8s.io;v=v1",
 			object: func(t *testing.T) (metav1.Object, string, string) {
-				ns := "default"
+				svc, err := clientset.CoreV1().Services(testNamespace).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-5"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
-				svc, err := clientset.CoreV1().Services(ns).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-4"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
 				if err != nil {
 					t.Fatalf("unable to create service: %v", err)
 				}
-				if _, err := clientset.CoreV1().Services(ns).Patch(svc.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
+				if _, err := clientset.CoreV1().Services(testNamespace).Patch(svc.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
 					t.Fatalf("unable to update service: %v", err)
 				}
 				return svc, "", "services"
@@ -930,12 +949,11 @@ func TestTransform(t *testing.T) {
 			accept:        "application/json;as=Table;g=meta.k8s.io;v=v1",
 			includeObject: metav1.IncludeNone,
 			object: func(t *testing.T) (metav1.Object, string, string) {
-				ns := "default"
+				obj, err := clientset.CoreV1().Services(testNamespace).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-6"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
-				obj, err := clientset.CoreV1().Services(ns).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-5"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
 				if err != nil {
 					t.Fatalf("unable to create object: %v", err)
 				}
-				if _, err := clientset.CoreV1().Services(ns).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
+				if _, err := clientset.CoreV1().Services(testNamespace).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
 					t.Fatalf("unable to update object: %v", err)
 				}
 				return obj, "", "services"
@@ -949,12 +967,11 @@ func TestTransform(t *testing.T) {
 			accept:        "application/json;as=Table;g=meta.k8s.io;v=v1",
 			includeObject: metav1.IncludeObject,
 			object: func(t *testing.T) (metav1.Object, string, string) {
-				ns := "default"
+				obj, err := clientset.CoreV1().Services(testNamespace).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-7"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
-				obj, err := clientset.CoreV1().Services(ns).Create(&v1.Service{ObjectMeta: metav1.ObjectMeta{Name: "test-6"}, Spec: v1.ServiceSpec{Ports: []v1.ServicePort{{Port: 1000}}}})
 				if err != nil {
 					t.Fatalf("unable to create object: %v", err)
 				}
-				if _, err := clientset.CoreV1().Services(ns).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
+				if _, err := clientset.CoreV1().Services(testNamespace).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
 					t.Fatalf("unable to update object: %v", err)
 				}
 				return obj, "", "services"
@@ -974,12 +991,11 @@ func TestTransform(t *testing.T) {
 			name:   "v1 verify partial metadata object on config maps",
 			accept: "application/json;as=PartialObjectMetadata;g=meta.k8s.io;v=v1",
 			object: func(t *testing.T) (metav1.Object, string, string) {
-				ns := "default"
+				obj, err := clientset.CoreV1().ConfigMaps(testNamespace).Create(&v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "test-3", Annotations: map[string]string{"test": "0"}}})
-				obj, err := clientset.CoreV1().ConfigMaps(ns).Create(&v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "test-3", Annotations: map[string]string{"test": "0"}}})
 				if err != nil {
 					t.Fatalf("unable to create object: %v", err)
 				}
-				if _, err := clientset.CoreV1().ConfigMaps(ns).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
+				if _, err := clientset.CoreV1().ConfigMaps(testNamespace).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
 					t.Fatalf("unable to update object: %v", err)
 				}
 				return obj, "", "configmaps"
@@ -992,12 +1008,11 @@ func TestTransform(t *testing.T) {
 			name:   "v1 verify partial metadata object on config maps in protobuf",
 			accept: "application/vnd.kubernetes.protobuf;as=PartialObjectMetadata;g=meta.k8s.io;v=v1",
 			object: func(t *testing.T) (metav1.Object, string, string) {
-				ns := "default"
+				obj, err := clientset.CoreV1().ConfigMaps(testNamespace).Create(&v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "test-4", Annotations: map[string]string{"test": "0"}}})
-				obj, err := clientset.CoreV1().ConfigMaps(ns).Create(&v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "test-4", Annotations: map[string]string{"test": "0"}}})
 				if err != nil {
 					t.Fatalf("unable to create object: %v", err)
 				}
-				if _, err := clientset.CoreV1().ConfigMaps(ns).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
+				if _, err := clientset.CoreV1().ConfigMaps(testNamespace).Patch(obj.Name, types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`)); err != nil {
 					t.Fatalf("unable to update object: %v", err)
 				}
 				return obj, "", "configmaps"
@@ -1006,6 +1021,23 @@ func TestTransform(t *testing.T) {
 				expectPartialObjectMetaV1EventsProtobuf(t, w, "0", "1")
 			},
 		},
+		{
+			name:   "v1 verify partial metadata object on CRDs in protobuf",
+			accept: "application/vnd.kubernetes.protobuf;as=PartialObjectMetadata;g=meta.k8s.io;v=v1",
+			object: func(t *testing.T) (metav1.Object, string, string) {
+				cr, err := crclient.Create(&unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "cr.bar.com/v1", "kind": "Foo", "metadata": map[string]interface{}{"name": "test-8", "annotations": map[string]string{"test": "0"}}}}, metav1.CreateOptions{})
+				if err != nil {
+					t.Fatalf("unable to create cr: %v", err)
+				}
+				if _, err := crclient.Patch(cr.GetName(), types.MergePatchType, []byte(`{"metadata":{"annotations":{"test":"1"}}}`), metav1.PatchOptions{}); err != nil {
+					t.Fatalf("unable to patch cr: %v", err)
+				}
+				return cr, crdGVR.Group, "foos"
+			},
+			wantBody: func(t *testing.T, w io.Reader) {
+				expectPartialObjectMetaV1EventsProtobuf(t, w, "0", "1")
+			},
+		},
 		{
 			name:   "v1 verify error on unsupported mimetype protobuf for table conversion",
 			accept: "application/vnd.kubernetes.protobuf;as=Table;g=meta.k8s.io;v=v1",
@@ -1046,6 +1078,18 @@ func TestTransform(t *testing.T) {
 				}
 			},
 		},
+		{
+			name:   "v1 verify error on invalid mimetype - only meta kinds accepted",
+			accept: "application/json;as=Service;g=;v=v1",
+			object: func(t *testing.T) (metav1.Object, string, string) {
+				return &metav1.ObjectMeta{Name: "kubernetes", Namespace: "default"}, "", "services"
+			},
+			wantErr: func(t *testing.T, err error) {
+				if !apierrors.IsNotAcceptable(err) {
+					t.Fatal(err)
+				}
+			},
+		},
 		{
 			name:   "v1 verify error on invalid mimetype - missing kind",
 			accept: "application/json;g=meta.k8s.io;v=v1",

test/integration/kubelet/watch_manager_test.go

@@ -22,7 +22,7 @@ import (
 	"testing"
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -35,6 +35,7 @@ import (
 )
 
 func TestWatchBasedManager(t *testing.T) {
+	testNamespace := "test-watch-based-manager"
 	server := kubeapiservertesting.StartTestServerOrDie(t, nil, nil, framework.SharedEtcd())
 	defer server.TearDownFn()
 
@@ -43,6 +44,9 @@ func TestWatchBasedManager(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
+	if _, err := client.CoreV1().Namespaces().Create((&v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: testNamespace}})); err != nil {
+		t.Fatal(err)
+	}
 
 	listObj := func(namespace string, options metav1.ListOptions) (runtime.Object, error) {
 		return client.CoreV1().Secrets(namespace).List(options)
@@ -62,7 +66,7 @@ func TestWatchBasedManager(t *testing.T) {
 			defer wg.Done()
 			for j := 0; j < 100; j++ {
 				name := fmt.Sprintf("s%d", i*100+j)
-				if _, err := client.CoreV1().Secrets("default").Create(&v1.Secret{ObjectMeta: metav1.ObjectMeta{Name: name}}); err != nil {
+				if _, err := client.CoreV1().Secrets(testNamespace).Create(&v1.Secret{ObjectMeta: metav1.ObjectMeta{Name: name}}); err != nil {
 					t.Fatal(err)
 				}
 			}
@@ -81,9 +85,9 @@ func TestWatchBasedManager(t *testing.T) {
 			for j := 0; j < 100; j++ {
 				name := fmt.Sprintf("s%d", i*100+j)
 				start := time.Now()
-				store.AddReference("default", name)
+				store.AddReference(testNamespace, name)
 				err := wait.PollImmediate(10*time.Millisecond, 10*time.Second, func() (bool, error) {
-					obj, err := store.Get("default", name)
+					obj, err := store.Get(testNamespace, name)
 					if err != nil {
 						t.Logf("failed on %s, retrying: %v", name, err)
 						return false, nil

test/integration/master/crd_test.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/go-openapi/spec"
 
+	v1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	apiextensionsclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
@@ -42,10 +43,14 @@ func TestCRDShadowGroup(t *testing.T) {
 	result := kubeapiservertesting.StartTestServerOrDie(t, nil, nil, framework.SharedEtcd())
 	defer result.TearDownFn()
 
+	testNamespace := "test-crd-shadow-group"
 	kubeclient, err := kubernetes.NewForConfig(result.ClientConfig)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
+	if _, err := kubeclient.CoreV1().Namespaces().Create((&v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: testNamespace}})); err != nil {
+		t.Fatal(err)
+	}
 
 	apiextensionsclient, err := apiextensionsclientset.NewForConfig(result.ClientConfig)
 	if err != nil {
@@ -53,8 +58,8 @@ func TestCRDShadowGroup(t *testing.T) {
 	}
 
 	t.Logf("Creating a NetworkPolicy")
-	nwPolicy, err := kubeclient.NetworkingV1().NetworkPolicies("default").Create(&networkingv1.NetworkPolicy{
+	nwPolicy, err := kubeclient.NetworkingV1().NetworkPolicies(testNamespace).Create(&networkingv1.NetworkPolicy{
-		ObjectMeta: metav1.ObjectMeta{Name: "abc", Namespace: metav1.NamespaceDefault},
+		ObjectMeta: metav1.ObjectMeta{Name: "abc", Namespace: testNamespace},
 		Spec: networkingv1.NetworkPolicySpec{
 			PodSelector: metav1.LabelSelector{MatchLabels: map[string]string{"foo": "bar"}},
 			Ingress:     []networkingv1.NetworkPolicyIngressRule{},
@@ -100,6 +105,15 @@ func TestCRD(t *testing.T) {
 	result := kubeapiservertesting.StartTestServerOrDie(t, nil, nil, framework.SharedEtcd())
 	defer result.TearDownFn()
 
+	testNamespace := "test-crd"
+	kubeclient, err := kubernetes.NewForConfig(result.ClientConfig)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if _, err := kubeclient.CoreV1().Namespaces().Create((&v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: testNamespace}})); err != nil {
+		t.Fatal(err)
+	}
+
 	apiextensionsclient, err := apiextensionsclientset.NewForConfig(result.ClientConfig)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -128,7 +142,7 @@ func TestCRD(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 	fooResource := schema.GroupVersionResource{Group: "cr.bar.com", Version: "v1", Resource: "foos"}
-	_, err = dynamicClient.Resource(fooResource).Namespace("default").List(metav1.ListOptions{})
+	_, err = dynamicClient.Resource(fooResource).Namespace(testNamespace).List(metav1.ListOptions{})
 	if err != nil {
 		t.Errorf("Failed to list foos.cr.bar.com instances: %v", err)
 	}