Merge pull request #62913 from deads2k/client-04-dynamic

cmd/kube-controller-manager/app/core.go

@@ -36,7 +36,6 @@ import (
 	cacheddiscovery "k8s.io/client-go/discovery/cached"
 	"k8s.io/client-go/dynamic"
 	clientset "k8s.io/client-go/kubernetes"
-	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	"k8s.io/kubernetes/pkg/controller"
 	endpointcontroller "k8s.io/kubernetes/pkg/controller/endpoint"
 	"k8s.io/kubernetes/pkg/controller/garbagecollector"
@@ -294,9 +293,6 @@ func startResourceQuotaController(ctx ControllerContext) (bool, error) {
 }
 
 func startNamespaceController(ctx ControllerContext) (bool, error) {
-	// TODO: should use a dynamic RESTMapper built from the discovery results.
-	restMapper := legacyscheme.Registry.RESTMapper()
-
 	// the namespace cleanup controller is very chatty.  It makes lots of discovery calls and then it makes lots of delete calls
 	// the ratelimiter negatively affects its speed.  Deleting 100 total items in a namespace (that's only a few of each resource
 	// including events), takes ~10 seconds by default.
@@ -304,13 +300,17 @@ func startNamespaceController(ctx ControllerContext) (bool, error) {
 	nsKubeconfig.QPS *= 10
 	nsKubeconfig.Burst *= 10
 	namespaceKubeClient := clientset.NewForConfigOrDie(nsKubeconfig)
-	namespaceClientPool := dynamic.NewClientPool(nsKubeconfig, restMapper, dynamic.LegacyAPIPathResolverFunc)
 
 	discoverResourcesFn := namespaceKubeClient.Discovery().ServerPreferredNamespacedResources
 
+	dynamicClient, err := dynamic.NewForConfig(nsKubeconfig)
+	if err != nil {
+		return true, err
+	}
+
 	namespaceController := namespacecontroller.NewNamespaceController(
 		namespaceKubeClient,
-		namespaceClientPool,
+		dynamicClient,
 		discoverResourcesFn,
 		ctx.InformerFactory.Core().V1().Namespaces(),
 		ctx.ComponentConfig.NamespaceController.NamespaceSyncPeriod.Duration,

pkg/controller/garbagecollector/garbagecollector.go

@@ -303,7 +303,7 @@ func (gc *GarbageCollector) isDangling(reference metav1.OwnerReference, item *no
 	// TODO: It's only necessary to talk to the API server if the owner node
 	// is a "virtual" node. The local graph could lag behind the real
 	// status, but in practice, the difference is small.
-	owner, err = client.Resource(resource, item.identity.Namespace).Get(reference.Name, metav1.GetOptions{})
+	owner, err = client.Resource(resource, resourceDefaultNamespace(resource, item.identity.Namespace)).Get(reference.Name, metav1.GetOptions{})
 	switch {
 	case errors.IsNotFound(err):
 		gc.absentOwnerCache.Add(reference.UID)

pkg/controller/garbagecollector/garbagecollector_test.go

@@ -686,9 +686,13 @@ func TestOrphanDependentsFailure(t *testing.T) {
 		},
 	}
 	err := gc.orphanDependents(objectReference{}, dependents)
-	expected := `the server reported a conflict (patch pods pod)`
+	expected := `the server reported a conflict`
 	if err == nil || !strings.Contains(err.Error(), expected) {
-		t.Errorf("expected error contains text %s, got %v", expected, err)
+		if err != nil {
+			t.Errorf("expected error contains text %q, got %q", expected, err.Error())
+		} else {
+			t.Errorf("expected error contains text %q, got nil", expected)
+		}
 	}
 }
 

pkg/controller/garbagecollector/graph_builder.go

@@ -135,7 +135,7 @@ func listWatcher(client dynamic.Interface, resource schema.GroupVersionResource)
 			// namespaces if it's namespace scoped, so leave
 			// APIResource.Namespaced as false is all right.
 			apiResource := metav1.APIResource{Name: resource.Resource}
-			return client.ParameterCodec(dynamic.VersionedParameterEncoderWithV1Fallback).
+			return client.
 				Resource(&apiResource, metav1.NamespaceAll).
 				List(options)
 		},
@@ -145,7 +145,7 @@ func listWatcher(client dynamic.Interface, resource schema.GroupVersionResource)
 			// namespaces if it's namespace scoped, so leave
 			// APIResource.Namespaced as false is all right.
 			apiResource := metav1.APIResource{Name: resource.Resource}
-			return client.ParameterCodec(dynamic.VersionedParameterEncoderWithV1Fallback).
+			return client.
 				Resource(&apiResource, metav1.NamespaceAll).
 				Watch(options)
 		},

pkg/controller/garbagecollector/operations.go

@@ -30,6 +30,14 @@ import (
 	"k8s.io/client-go/util/retry"
 )
 
+// cluster scoped resources don't have namespaces.  Default to the item's namespace, but clear it for cluster scoped resources
+func resourceDefaultNamespace(resource *metav1.APIResource, defaultNamespace string) string {
+	if resource.Namespaced {
+		return defaultNamespace
+	}
+	return ""
+}
+
 // apiResource consults the REST mapper to translate an <apiVersion, kind,
 // namespace> tuple to a unversioned.APIResource struct.
 func (gc *GarbageCollector) apiResource(apiVersion, kind string) (*metav1.APIResource, error) {
@@ -60,7 +68,7 @@ func (gc *GarbageCollector) deleteObject(item objectReference, policy *metav1.De
 	uid := item.UID
 	preconditions := metav1.Preconditions{UID: &uid}
 	deleteOptions := metav1.DeleteOptions{Preconditions: &preconditions, PropagationPolicy: policy}
-	return client.Resource(resource, item.Namespace).Delete(item.Name, &deleteOptions)
+	return client.Resource(resource, resourceDefaultNamespace(resource, item.Namespace)).Delete(item.Name, &deleteOptions)
 }
 
 func (gc *GarbageCollector) getObject(item objectReference) (*unstructured.Unstructured, error) {
@@ -73,7 +81,7 @@ func (gc *GarbageCollector) getObject(item objectReference) (*unstructured.Unstr
 	if err != nil {
 		return nil, err
 	}
-	return client.Resource(resource, item.Namespace).Get(item.Name, metav1.GetOptions{})
+	return client.Resource(resource, resourceDefaultNamespace(resource, item.Namespace)).Get(item.Name, metav1.GetOptions{})
 }
 
 func (gc *GarbageCollector) updateObject(item objectReference, obj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
@@ -86,7 +94,7 @@ func (gc *GarbageCollector) updateObject(item objectReference, obj *unstructured
 	if err != nil {
 		return nil, err
 	}
-	return client.Resource(resource, item.Namespace).Update(obj)
+	return client.Resource(resource, resourceDefaultNamespace(resource, item.Namespace)).Update(obj)
 }
 
 func (gc *GarbageCollector) patchObject(item objectReference, patch []byte) (*unstructured.Unstructured, error) {
@@ -99,7 +107,7 @@ func (gc *GarbageCollector) patchObject(item objectReference, patch []byte) (*un
 	if err != nil {
 		return nil, err
 	}
-	return client.Resource(resource, item.Namespace).Patch(item.Name, types.StrategicMergePatchType, patch)
+	return client.Resource(resource, resourceDefaultNamespace(resource, item.Namespace)).Patch(item.Name, types.StrategicMergePatchType, patch)
 }
 
 // TODO: Using Patch when strategicmerge supports deleting an entry from a

pkg/controller/namespace/deletion/BUILD

@@ -31,7 +31,6 @@ go_test(
     srcs = ["namespaced_resources_deleter_test.go"],
     embed = [":go_default_library"],
     deps = [
-        "//pkg/api/legacyscheme:go_default_library",
         "//pkg/apis/core:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",

pkg/controller/namespace/deletion/namespaced_resources_deleter.go

@@ -43,13 +43,13 @@ type NamespacedResourcesDeleterInterface interface {
 }
 
 func NewNamespacedResourcesDeleter(nsClient v1clientset.NamespaceInterface,
-	clientPool dynamic.ClientPool, podsGetter v1clientset.PodsGetter,
+	dynamicClient dynamic.DynamicInterface, podsGetter v1clientset.PodsGetter,
 	discoverResourcesFn func() ([]*metav1.APIResourceList, error),
 	finalizerToken v1.FinalizerName, deleteNamespaceWhenDone bool) NamespacedResourcesDeleterInterface {
 	d := &namespacedResourcesDeleter{
-		nsClient:   nsClient,
-		clientPool: clientPool,
-		podsGetter: podsGetter,
+		nsClient:      nsClient,
+		dynamicClient: dynamicClient,
+		podsGetter:    podsGetter,
 		opCache: &operationNotSupportedCache{
 			m: make(map[operationKey]bool),
 		},
@@ -68,7 +68,7 @@ type namespacedResourcesDeleter struct {
 	// Client to manipulate the namespace.
 	nsClient v1clientset.NamespaceInterface
 	// Dynamic client to list and delete all namespaced resources.
-	clientPool dynamic.ClientPool
+	dynamicClient dynamic.DynamicInterface
 	// Interface to get PodInterface.
 	podsGetter v1clientset.PodsGetter
 	// Cache of what operations are not supported on each group version resource.
@@ -328,9 +328,7 @@ func (d *namespacedResourcesDeleter) finalizeNamespace(namespace *v1.Namespace)
 // deleteCollection is a helper function that will delete the collection of resources
 // it returns true if the operation was supported on the server.
 // it returns an error if the operation was supported on the server but was unable to complete.
-func (d *namespacedResourcesDeleter) deleteCollection(
-	dynamicClient dynamic.Interface, gvr schema.GroupVersionResource,
-	namespace string) (bool, error) {
+func (d *namespacedResourcesDeleter) deleteCollection(gvr schema.GroupVersionResource, namespace string) (bool, error) {
 	glog.V(5).Infof("namespace controller - deleteCollection - namespace: %s, gvr: %v", namespace, gvr)
 
 	key := operationKey{operation: operationDeleteCollection, gvr: gvr}
@@ -339,14 +337,12 @@ func (d *namespacedResourcesDeleter) deleteCollection(
 		return false, nil
 	}
 
-	apiResource := metav1.APIResource{Name: gvr.Resource, Namespaced: true}
-
 	// namespace controller does not want the garbage collector to insert the orphan finalizer since it calls
 	// resource deletions generically.  it will ensure all resources in the namespace are purged prior to releasing
 	// namespace itself.
 	background := metav1.DeletePropagationBackground
 	opts := &metav1.DeleteOptions{PropagationPolicy: &background}
-	err := dynamicClient.Resource(&apiResource, namespace).DeleteCollection(opts, metav1.ListOptions{})
+	err := d.dynamicClient.NamespacedResource(gvr, namespace).DeleteCollection(opts, metav1.ListOptions{})
 
 	if err == nil {
 		return true, nil
@@ -373,8 +369,7 @@ func (d *namespacedResourcesDeleter) deleteCollection(
 //  the list of items in the collection (if found)
 //  a boolean if the operation is supported
 //  an error if the operation is supported but could not be completed.
-func (d *namespacedResourcesDeleter) listCollection(
-	dynamicClient dynamic.Interface, gvr schema.GroupVersionResource, namespace string) (*unstructured.UnstructuredList, bool, error) {
+func (d *namespacedResourcesDeleter) listCollection(gvr schema.GroupVersionResource, namespace string) (*unstructured.UnstructuredList, bool, error) {
 	glog.V(5).Infof("namespace controller - listCollection - namespace: %s, gvr: %v", namespace, gvr)
 
 	key := operationKey{operation: operationList, gvr: gvr}
@@ -383,13 +378,8 @@ func (d *namespacedResourcesDeleter) listCollection(
 		return nil, false, nil
 	}
 
-	apiResource := metav1.APIResource{Name: gvr.Resource, Namespaced: true}
-	obj, err := dynamicClient.Resource(&apiResource, namespace).List(metav1.ListOptions{IncludeUninitialized: true})
+	unstructuredList, err := d.dynamicClient.NamespacedResource(gvr, namespace).List(metav1.ListOptions{IncludeUninitialized: true})
 	if err == nil {
-		unstructuredList, ok := obj.(*unstructured.UnstructuredList)
-		if !ok {
-			return nil, false, fmt.Errorf("resource: %s, expected *unstructured.UnstructuredList, got %#v", apiResource.Name, obj)
-		}
 		return unstructuredList, true, nil
 	}
 
@@ -409,22 +399,20 @@ func (d *namespacedResourcesDeleter) listCollection(
 }
 
 // deleteEachItem is a helper function that will list the collection of resources and delete each item 1 by 1.
-func (d *namespacedResourcesDeleter) deleteEachItem(
-	dynamicClient dynamic.Interface, gvr schema.GroupVersionResource, namespace string) error {
+func (d *namespacedResourcesDeleter) deleteEachItem(gvr schema.GroupVersionResource, namespace string) error {
 	glog.V(5).Infof("namespace controller - deleteEachItem - namespace: %s, gvr: %v", namespace, gvr)
 
-	unstructuredList, listSupported, err := d.listCollection(dynamicClient, gvr, namespace)
+	unstructuredList, listSupported, err := d.listCollection(gvr, namespace)
 	if err != nil {
 		return err
 	}
 	if !listSupported {
 		return nil
 	}
-	apiResource := metav1.APIResource{Name: gvr.Resource, Namespaced: true}
 	for _, item := range unstructuredList.Items {
 		background := metav1.DeletePropagationBackground
 		opts := &metav1.DeleteOptions{PropagationPolicy: &background}
-		if err = dynamicClient.Resource(&apiResource, namespace).Delete(item.GetName(), opts); err != nil && !errors.IsNotFound(err) && !errors.IsMethodNotSupported(err) {
+		if err = d.dynamicClient.NamespacedResource(gvr, namespace).Delete(item.GetName(), opts); err != nil && !errors.IsNotFound(err) && !errors.IsMethodNotSupported(err) {
 			return err
 		}
 	}
@@ -447,22 +435,15 @@ func (d *namespacedResourcesDeleter) deleteAllContentForGroupVersionResource(
 	}
 	glog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - estimate - namespace: %s, gvr: %v, estimate: %v", namespace, gvr, estimate)
 
-	// get a client for this group version...
-	dynamicClient, err := d.clientPool.ClientForGroupVersionResource(gvr)
-	if err != nil {
-		glog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - unable to get client - namespace: %s, gvr: %v, err: %v", namespace, gvr, err)
-		return estimate, err
-	}
-
 	// first try to delete the entire collection
-	deleteCollectionSupported, err := d.deleteCollection(dynamicClient, gvr, namespace)
+	deleteCollectionSupported, err := d.deleteCollection(gvr, namespace)
 	if err != nil {
 		return estimate, err
 	}
 
 	// delete collection was not supported, so we list and delete each item...
 	if !deleteCollectionSupported {
-		err = d.deleteEachItem(dynamicClient, gvr, namespace)
+		err = d.deleteEachItem(gvr, namespace)
 		if err != nil {
 			return estimate, err
 		}
@@ -471,7 +452,7 @@ func (d *namespacedResourcesDeleter) deleteAllContentForGroupVersionResource(
 	// verify there are no more remaining items
 	// it is not an error condition for there to be remaining items if local estimate is non-zero
 	glog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - checking for no more items in namespace: %s, gvr: %v", namespace, gvr)
-	unstructuredList, listSupported, err := d.listCollection(dynamicClient, gvr, namespace)
+	unstructuredList, listSupported, err := d.listCollection(gvr, namespace)
 	if err != nil {
 		glog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - error verifying no items in namespace: %s, gvr: %v, err: %v", namespace, gvr, err)
 		return estimate, err
@@ -497,8 +478,7 @@ func (d *namespacedResourcesDeleter) deleteAllContentForGroupVersionResource(
 // deleteAllContent will use the dynamic client to delete each resource identified in groupVersionResources.
 // It returns an estimate of the time remaining before the remaining resources are deleted.
 // If estimate > 0, not all resources are guaranteed to be gone.
-func (d *namespacedResourcesDeleter) deleteAllContent(
-	namespace string, namespaceDeletedAt metav1.Time) (int64, error) {
+func (d *namespacedResourcesDeleter) deleteAllContent(namespace string, namespaceDeletedAt metav1.Time) (int64, error) {
 	estimate := int64(0)
 	glog.V(4).Infof("namespace controller - deleteAllContent - namespace: %s", namespace)
 	resources, err := d.discoverResourcesFn()

pkg/controller/namespace/deletion/namespaced_resources_deleter_test.go

@@ -36,7 +36,6 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 	restclient "k8s.io/client-go/rest"
 	core "k8s.io/client-go/testing"
-	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	api "k8s.io/kubernetes/pkg/apis/core"
 )
 
@@ -173,14 +172,16 @@ func testSyncNamespaceThatIsTerminating(t *testing.T, versions *metav1.APIVersio
 		defer srv.Close()
 
 		mockClient := fake.NewSimpleClientset(testInput.testNamespace)
-		clientPool := dynamic.NewClientPool(clientConfig, legacyscheme.Registry.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
+		dynamicClient, err := dynamic.NewForConfig(clientConfig)
+		if err != nil {
+			t.Fatal(err)
+		}
 
 		fn := func() ([]*metav1.APIResourceList, error) {
 			return resources, nil
 		}
-		d := NewNamespacedResourcesDeleter(mockClient.Core().Namespaces(), clientPool, mockClient.Core(), fn, v1.FinalizerKubernetes, true)
-		err := d.Delete(testInput.testNamespace.Name)
-		if err != nil {
+		d := NewNamespacedResourcesDeleter(mockClient.Core().Namespaces(), dynamicClient, mockClient.Core(), fn, v1.FinalizerKubernetes, true)
+		if err := d.Delete(testInput.testNamespace.Name); err != nil {
 			t.Errorf("scenario %s - Unexpected error when synching namespace %v", scenario, err)
 		}
 

pkg/controller/namespace/namespace_controller.go

@@ -63,7 +63,7 @@ type NamespaceController struct {
 // NewNamespaceController creates a new NamespaceController
 func NewNamespaceController(
 	kubeClient clientset.Interface,
-	clientPool dynamic.ClientPool,
+	dynamicClient dynamic.DynamicInterface,
 	discoverResourcesFn func() ([]*metav1.APIResourceList, error),
 	namespaceInformer coreinformers.NamespaceInformer,
 	resyncPeriod time.Duration,
@@ -72,7 +72,7 @@ func NewNamespaceController(
 	// create the controller so we can inject the enqueue function
 	namespaceController := &NamespaceController{
 		queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "namespace"),
-		namespacedResourcesDeleter: deletion.NewNamespacedResourcesDeleter(kubeClient.CoreV1().Namespaces(), clientPool, kubeClient.CoreV1(), discoverResourcesFn, finalizerToken, true),
+		namespacedResourcesDeleter: deletion.NewNamespacedResourcesDeleter(kubeClient.CoreV1().Namespaces(), dynamicClient, kubeClient.CoreV1(), discoverResourcesFn, finalizerToken, true),
 	}
 
 	if kubeClient != nil && kubeClient.CoreV1().RESTClient().GetRateLimiter() != nil {

pkg/kubectl/cmd/util/factory_object_mapping.go

@@ -197,7 +197,7 @@ func genericDescriber(clientAccessFactory ClientAccessFactory, mapping *meta.RES
 	clientConfigCopy.GroupVersion = &gv
 
 	// used to fetch the resource
-	dynamicClient, err := dynamic.NewClient(&clientConfigCopy)
+	dynamicClient, err := dynamic.NewClient(&clientConfigCopy, gv)
 	if err != nil {
 		return nil, err
 	}

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/BUILD

@@ -49,6 +49,9 @@ filegroup(
 
 filegroup(
     name = "all-srcs",
-    srcs = [":package-srcs"],
+    srcs = [
+        ":package-srcs",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/unstructuredscheme:all-srcs",
+    ],
     tags = ["automanaged"],
 )

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/unstructured.go

@@ -58,6 +58,26 @@ func (obj *Unstructured) IsList() bool {
 	_, ok = field.([]interface{})
 	return ok
 }
+func (obj *Unstructured) ToList() (*UnstructuredList, error) {
+	if !obj.IsList() {
+		// return an empty list back
+		return &UnstructuredList{Object: obj.Object}, nil
+	}
+
+	ret := &UnstructuredList{}
+	ret.Object = obj.Object
+
+	err := obj.EachListItem(func(item runtime.Object) error {
+		castItem := item.(*Unstructured)
+		ret.Items = append(ret.Items, *castItem)
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return ret, nil
+}
 
 func (obj *Unstructured) EachListItem(fn func(runtime.Object) error) error {
 	field, ok := obj.Object["items"]

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/unstructuredscheme/BUILD

+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["scheme.go"],
+    importpath = "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/unstructuredscheme",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/serializer/json:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/serializer/versioning:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/unstructuredscheme/scheme.go

+/*
+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 unstructuredscheme
+
+import (
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"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/versioning"
+)
+
+var (
+	scheme = runtime.NewScheme()
+	codecs = serializer.NewCodecFactory(scheme)
+)
+
+// NewUnstructuredNegotiatedSerializer returns a simple, negotiated serializer
+func NewUnstructuredNegotiatedSerializer() runtime.NegotiatedSerializer {
+	return unstructuredNegotiatedSerializer{
+		scheme:  scheme,
+		typer:   NewUnstructuredObjectTyper(),
+		creator: NewUnstructuredCreator(),
+	}
+}
+
+type unstructuredNegotiatedSerializer struct {
+	scheme  *runtime.Scheme
+	typer   runtime.ObjectTyper
+	creator runtime.ObjectCreater
+}
+
+func (s unstructuredNegotiatedSerializer) SupportedMediaTypes() []runtime.SerializerInfo {
+	return []runtime.SerializerInfo{
+		{
+			MediaType:        "application/json",
+			EncodesAsText:    true,
+			Serializer:       json.NewSerializer(json.DefaultMetaFactory, s.creator, s.typer, false),
+			PrettySerializer: json.NewSerializer(json.DefaultMetaFactory, s.creator, s.typer, true),
+			StreamSerializer: &runtime.StreamSerializerInfo{
+				EncodesAsText: true,
+				Serializer:    json.NewSerializer(json.DefaultMetaFactory, s.creator, s.typer, false),
+				Framer:        json.Framer,
+			},
+		},
+		{
+			MediaType:     "application/yaml",
+			EncodesAsText: true,
+			Serializer:    json.NewYAMLSerializer(json.DefaultMetaFactory, s.creator, s.typer),
+		},
+	}
+}
+
+func (s unstructuredNegotiatedSerializer) EncoderForVersion(encoder runtime.Encoder, gv runtime.GroupVersioner) runtime.Encoder {
+	return versioning.NewDefaultingCodecForScheme(s.scheme, encoder, nil, gv, nil)
+}
+
+func (s unstructuredNegotiatedSerializer) DecoderToVersion(decoder runtime.Decoder, gv runtime.GroupVersioner) runtime.Decoder {
+	return versioning.NewDefaultingCodecForScheme(s.scheme, nil, decoder, nil, gv)
+}
+
+type unstructuredObjectTyper struct {
+}
+
+// NewUnstructuredObjectTyper returns an object typer that can deal with unstructured things
+func NewUnstructuredObjectTyper() runtime.ObjectTyper {
+	return unstructuredObjectTyper{}
+}
+
+func (t unstructuredObjectTyper) ObjectKinds(obj runtime.Object) ([]schema.GroupVersionKind, bool, error) {
+	// Delegate for things other than Unstructured.
+	if _, ok := obj.(runtime.Unstructured); !ok {
+		return nil, false, fmt.Errorf("cannot type %T", obj)
+	}
+	return []schema.GroupVersionKind{obj.GetObjectKind().GroupVersionKind()}, false, nil
+}
+
+func (t unstructuredObjectTyper) Recognizes(gvk schema.GroupVersionKind) bool {
+	return true
+}
+
+type unstructuredCreator struct{}
+
+// NewUnstructuredCreator returns a simple object creator that always returns an unstructured
+func NewUnstructuredCreator() runtime.ObjectCreater {
+	return unstructuredCreator{}
+}
+
+func (c unstructuredCreator) New(kind schema.GroupVersionKind) (runtime.Object, error) {
+	ret := &unstructured.Unstructured{}
+	ret.SetGroupVersionKind(kind)
+	return ret, nil
+}
+
+type unstructuredDefaulter struct {
+}
+
+// NewUnstructuredDefaulter returns defaulter suitable for unstructured types that doesn't default anything
+func NewUnstructuredDefaulter() runtime.ObjectDefaulter {
+	return unstructuredDefaulter{}
+}
+
+func (d unstructuredDefaulter) Default(in runtime.Object) {
+}

staging/src/k8s.io/client-go/dynamic/BUILD

@@ -30,25 +30,28 @@ go_test(
 go_library(
     name = "go_default_library",
     srcs = [
+        "bad_debt.go",
         "client.go",
         "client_pool.go",
         "dynamic_util.go",
+        "scheme.go",
+        "simple.go",
     ],
     importpath = "k8s.io/client-go/dynamic",
     deps = [
-        "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/conversion/queryparams:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/serializer/json:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/serializer/streaming:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime/serializer/versioning:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes/scheme:go_default_library",
         "//vendor/k8s.io/client-go/rest:go_default_library",
-        "//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
     ],
 )
 

staging/src/k8s.io/client-go/dynamic/bad_debt.go

+/*
+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 dynamic
+
+import (
+	"encoding/json"
+	"io"
+	"strings"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+)
+
+// dynamicCodec is a codec that wraps the standard unstructured codec
+// with special handling for Status objects.
+// Deprecated only used by test code and its wrong
+type dynamicCodec struct{}
+
+func (dynamicCodec) Decode(data []byte, gvk *schema.GroupVersionKind, obj runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) {
+	obj, gvk, err := unstructured.UnstructuredJSONScheme.Decode(data, gvk, obj)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if _, ok := obj.(*metav1.Status); !ok && strings.ToLower(gvk.Kind) == "status" {
+		obj = &metav1.Status{}
+		err := json.Unmarshal(data, obj)
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+
+	return obj, gvk, nil
+}
+
+func (dynamicCodec) Encode(obj runtime.Object, w io.Writer) error {
+	return unstructured.UnstructuredJSONScheme.Encode(obj, w)
+}
+
+// ContentConfig returns a rest.ContentConfig for dynamic types.
+// Deprecated only used by test code and its wrong
+func ContentConfig() rest.ContentConfig {
+	var jsonInfo runtime.SerializerInfo
+	// TODO: scheme.Codecs here should become "pkg/apis/server/scheme" which is the minimal core you need
+	// to talk to a kubernetes server
+	for _, info := range scheme.Codecs.SupportedMediaTypes() {
+		if info.MediaType == runtime.ContentTypeJSON {
+			jsonInfo = info
+			break
+		}
+	}
+
+	jsonInfo.Serializer = dynamicCodec{}
+	jsonInfo.PrettySerializer = nil
+	return rest.ContentConfig{
+		AcceptContentTypes:   runtime.ContentTypeJSON,
+		ContentType:          runtime.ContentTypeJSON,
+		NegotiatedSerializer: serializer.NegotiatedSerializerWrapper(jsonInfo),
+	}
+}

staging/src/k8s.io/client-go/dynamic/client_pool.go

@@ -113,7 +113,7 @@ func (c *clientPoolImpl) ClientForGroupVersionKind(kind schema.GroupVersionKind)
 	// we need to make a client
 	conf.GroupVersion = &gv
 
-	dynamicClient, err := NewClient(conf)
+	dynamicClient, err := NewClient(conf, gv)
 	if err != nil {
 		return nil, err
 	}

staging/src/k8s.io/client-go/dynamic/scheme.go

+/*
+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 dynamic
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"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/versioning"
+)
+
+var watchScheme = runtime.NewScheme()
+var basicScheme = runtime.NewScheme()
+var deleteScheme = runtime.NewScheme()
+var parameterScheme = runtime.NewScheme()
+var deleteOptionsCodec = serializer.NewCodecFactory(deleteScheme)
+var dynamicParameterCodec = runtime.NewParameterCodec(parameterScheme)
+
+var versionV1 = schema.GroupVersion{Version: "v1"}
+
+func init() {
+	metav1.AddToGroupVersion(watchScheme, versionV1)
+	metav1.AddToGroupVersion(basicScheme, versionV1)
+	metav1.AddToGroupVersion(parameterScheme, versionV1)
+	metav1.AddToGroupVersion(deleteScheme, versionV1)
+}
+
+var watchJsonSerializerInfo = runtime.SerializerInfo{
+	MediaType:        "application/json",
+	EncodesAsText:    true,
+	Serializer:       json.NewSerializer(json.DefaultMetaFactory, watchScheme, watchScheme, false),
+	PrettySerializer: json.NewSerializer(json.DefaultMetaFactory, watchScheme, watchScheme, true),
+	StreamSerializer: &runtime.StreamSerializerInfo{
+		EncodesAsText: true,
+		Serializer:    json.NewSerializer(json.DefaultMetaFactory, watchScheme, watchScheme, false),
+		Framer:        json.Framer,
+	},
+}
+
+// watchNegotiatedSerializer is used to read the wrapper of the watch stream
+type watchNegotiatedSerializer struct{}
+
+var watchNegotiatedSerializerInstance = watchNegotiatedSerializer{}
+
+func (s watchNegotiatedSerializer) SupportedMediaTypes() []runtime.SerializerInfo {
+	return []runtime.SerializerInfo{watchJsonSerializerInfo}
+}
+
+func (s watchNegotiatedSerializer) EncoderForVersion(encoder runtime.Encoder, gv runtime.GroupVersioner) runtime.Encoder {
+	return versioning.NewDefaultingCodecForScheme(watchScheme, encoder, nil, gv, nil)
+}
+
+func (s watchNegotiatedSerializer) DecoderToVersion(decoder runtime.Decoder, gv runtime.GroupVersioner) runtime.Decoder {
+	return versioning.NewDefaultingCodecForScheme(watchScheme, nil, decoder, nil, gv)
+}
+
+// basicNegotiatedSerializer is used to handle discovery and error handling serialization
+type basicNegotiatedSerializer struct{}
+
+func (s basicNegotiatedSerializer) SupportedMediaTypes() []runtime.SerializerInfo {
+	return []runtime.SerializerInfo{
+		{
+			MediaType:        "application/json",
+			EncodesAsText:    true,
+			Serializer:       json.NewSerializer(json.DefaultMetaFactory, basicScheme, basicScheme, false),
+			PrettySerializer: json.NewSerializer(json.DefaultMetaFactory, basicScheme, basicScheme, true),
+			StreamSerializer: &runtime.StreamSerializerInfo{
+				EncodesAsText: true,
+				Serializer:    json.NewSerializer(json.DefaultMetaFactory, basicScheme, basicScheme, false),
+				Framer:        json.Framer,
+			},
+		},
+	}
+}
+
+func (s basicNegotiatedSerializer) EncoderForVersion(encoder runtime.Encoder, gv runtime.GroupVersioner) runtime.Encoder {
+	return versioning.NewDefaultingCodecForScheme(watchScheme, encoder, nil, gv, nil)
+}
+
+func (s basicNegotiatedSerializer) DecoderToVersion(decoder runtime.Decoder, gv runtime.GroupVersioner) runtime.Decoder {
+	return versioning.NewDefaultingCodecForScheme(watchScheme, nil, decoder, nil, gv)
+}

staging/src/k8s.io/client-go/rest/request.go

@@ -317,10 +317,14 @@ func (r *Request) Param(paramName, s string) *Request {
 // VersionedParams will not write query parameters that have omitempty set and are empty. If a
 // parameter has already been set it is appended to (Params and VersionedParams are additive).
 func (r *Request) VersionedParams(obj runtime.Object, codec runtime.ParameterCodec) *Request {
+	return r.SpecificallyVersionedParams(obj, codec, *r.content.GroupVersion)
+}
+
+func (r *Request) SpecificallyVersionedParams(obj runtime.Object, codec runtime.ParameterCodec, version schema.GroupVersion) *Request {
 	if r.err != nil {
 		return r
 	}
-	params, err := codec.EncodeParameters(obj, *r.content.GroupVersion)
+	params, err := codec.EncodeParameters(obj, version)
 	if err != nil {
 		r.err = err
 		return r
@@ -485,6 +489,19 @@ func (r *Request) tryThrottle() {
 // Watch attempts to begin watching the requested location.
 // Returns a watch.Interface, or an error.
 func (r *Request) Watch() (watch.Interface, error) {
+	return r.WatchWithSpecificDecoders(
+		func(body io.ReadCloser) streaming.Decoder {
+			framer := r.serializers.Framer.NewFrameReader(body)
+			return streaming.NewDecoder(framer, r.serializers.StreamingSerializer)
+		},
+		r.serializers.Decoder,
+	)
+}
+
+// WatchWithSpecificDecoders attempts to begin watching the requested location with a *different* decoder.
+// Turns out that you want one "standard" decoder for the watch event and one "personal" decoder for the content
+// Returns a watch.Interface, or an error.
+func (r *Request) WatchWithSpecificDecoders(wrapperDecoderFn func(io.ReadCloser) streaming.Decoder, embeddedDecoder runtime.Decoder) (watch.Interface, error) {
 	// We specifically don't want to rate limit watches, so we
 	// don't use r.throttle here.
 	if r.err != nil {
@@ -532,9 +549,8 @@ func (r *Request) Watch() (watch.Interface, error) {
 		}
 		return nil, fmt.Errorf("for request '%+v', got status: %v", url, resp.StatusCode)
 	}
-	framer := r.serializers.Framer.NewFrameReader(resp.Body)
-	decoder := streaming.NewDecoder(framer, r.serializers.StreamingSerializer)
-	return watch.NewStreamWatcher(restclientwatch.NewDecoder(decoder, r.serializers.Decoder)), nil
+	wrapperDecoder := wrapperDecoderFn(resp.Body)
+	return watch.NewStreamWatcher(restclientwatch.NewDecoder(wrapperDecoder, embeddedDecoder)), nil
 }
 
 // updateURLMetrics is a convenience function for pushing metrics.

test/e2e_node/services/BUILD

@@ -23,7 +23,6 @@ go_library(
         "//cmd/kube-apiserver/app:go_default_library",
         "//cmd/kube-apiserver/app/options:go_default_library",
         "//cmd/kubelet/app/options:go_default_library",
-        "//pkg/api/legacyscheme:go_default_library",
         "//pkg/controller/namespace:go_default_library",
         "//pkg/features:go_default_library",
         "//pkg/kubelet/apis/kubeletconfig:go_default_library",

test/e2e_node/services/namespace_controller.go

@@ -24,7 +24,6 @@ import (
 	"k8s.io/client-go/informers"
 	clientset "k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
-	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	namespacecontroller "k8s.io/kubernetes/pkg/controller/namespace"
 )
 
@@ -56,12 +55,15 @@ func (n *NamespaceController) Start() error {
 	if err != nil {
 		return err
 	}
-	clientPool := dynamic.NewClientPool(config, legacyscheme.Registry.RESTMapper(), dynamic.LegacyAPIPathResolverFunc)
+	dynamicClient, err := dynamic.NewForConfig(config)
+	if err != nil {
+		return err
+	}
 	discoverResourcesFn := client.Discovery().ServerPreferredNamespacedResources
 	informerFactory := informers.NewSharedInformerFactory(client, ncResyncPeriod)
 	nc := namespacecontroller.NewNamespaceController(
 		client,
-		clientPool,
+		dynamicClient,
 		discoverResourcesFn,
 		informerFactory.Core().V1().Namespaces(),
 		ncResyncPeriod, v1.FinalizerKubernetes,

test/integration/client/dynamic_client_test.go

@@ -45,7 +45,7 @@ func TestDynamicClient(t *testing.T) {
 	}
 
 	client := clientset.NewForConfigOrDie(config)
-	dynamicClient, err := dynamic.NewClient(config)
+	dynamicClient, err := dynamic.NewClient(config, *gv)
 	_ = dynamicClient
 	if err != nil {
 		t.Fatalf("unexpected error creating dynamic client: %v", err)

test/integration/master/crd_test.go

@@ -153,7 +153,7 @@ func TestCRD(t *testing.T) {
 	barComConfig := *result.ClientConfig
 	barComConfig.GroupVersion = &schema.GroupVersion{Group: "cr.bar.com", Version: "v1"}
 	barComConfig.APIPath = "/apis"
-	barComClient, err := dynamic.NewClient(&barComConfig)
+	barComClient, err := dynamic.NewClient(&barComConfig, *barComConfig.GroupVersion)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}