Always negotiate a decoder using ClientNegotiator

This commit performs two refactors and fixes a bug.

Refactor 1 changes the signature of Request to take a RESTClient, which
removes the extra copy of everything on RESTClient from Request. A pair
of optional constructors are added for testing. The major functional
change is that Request no longer has the shim HTTPClient interface and
so some test cases change slightly because we are now going through
http.Client code paths instead of direct to our test stubs.

Refactor 2 changes the signature of RESTClient to take a
ClientContentConfig instead of ContentConfig - the primary difference
being that ClientContentConfig uses ClientNegotiator instead of
NegotiatedSerializer and the old Serializers type. We also collapse
some redundancies (like the rate limiter can be created outside the
constructor).

The bug fix is to negotiate the streaming content type on a Watch()
like we do for requests. We stop caching the decoder and simply
resolve it on the request. We also clean up the dynamic client
and remove the extra WatchSpecificVersions() method by providing
a properly wrapped dynamic client.

Kubernetes-commit: 3b780c64b89606f4e6b21f48fb9c305d5998b9e5

Author: smarterclayton

metadata/fake/simple.go

@@ -317,7 +317,6 @@ func (c *metadataResourceClient) List(opts metav1.ListOptions) (*metav1.PartialO
 	if !ok {
 		return nil, fmt.Errorf("incoming object is incorrect type %T", obj)
 	}
-	fmt.Printf("DEBUG: %#v\n", inputList)
 
 	list := &metav1.PartialObjectMetadataList{
 		ListMeta: inputList.ListMeta,

rest/client.go

@@ -17,8 +17,6 @@ limitations under the License.
 package rest
 
 import (
-	"fmt"
-	"mime"
 	"net/http"
 	"net/url"
 	"os"
@@ -51,6 +49,28 @@ type Interface interface {
 	APIVersion() schema.GroupVersion
 }
 
+// ClientContentConfig controls how RESTClient communicates with the server.
+//
+// TODO: ContentConfig will be updated to accept a Negotiator instead of a
+//   NegotiatedSerializer and NegotiatedSerializer will be removed.
+type ClientContentConfig struct {
+	// AcceptContentTypes specifies the types the client will accept and is optional.
+	// If not set, ContentType will be used to define the Accept header
+	AcceptContentTypes string
+	// ContentType specifies the wire format used to communicate with the server.
+	// This value will be set as the Accept header on requests made to the server if
+	// AcceptContentTypes is not set, and as the default content type on any object
+	// sent to the server. If not set, "application/json" is used.
+	ContentType string
+	// GroupVersion is the API version to talk to. Must be provided when initializing
+	// a RESTClient directly. When initializing a Client, will be set with the default
+	// code version. This is used as the default group version for VersionedParams.
+	GroupVersion schema.GroupVersion
+	// Negotiator is used for obtaining encoders and decoders for multiple
+	// supported media types.
+	Negotiator runtime.ClientNegotiator
+}
+
 // RESTClient imposes common Kubernetes API conventions on a set of resource paths.
 // The baseURL is expected to point to an HTTP or HTTPS path that is the parent
 // of one or more resources.  The server should return a decodable API resource
@@ -64,66 +84,42 @@ type RESTClient struct {
 	// versionedAPIPath is a path segment connecting the base URL to the resource root
 	versionedAPIPath string
 
-	// contentConfig is the information used to communicate with the server.
-	contentConfig ContentConfig
-
-	// serializers contain all serializers for underlying content type.
-	serializers Serializers
+	// content describes how a RESTClient encodes and decodes responses.
+	content ClientContentConfig
 
 	// creates BackoffManager that is passed to requests.
 	createBackoffMgr func() BackoffManager
 
-	// TODO extract this into a wrapper interface via the RESTClient interface in kubectl.
-	Throttle flowcontrol.RateLimiter
+	// rateLimiter is shared among all requests created by this client unless specifically
+	// overridden.
+	rateLimiter flowcontrol.RateLimiter
 
 	// Set specific behavior of the client.  If not set http.DefaultClient will be used.
 	Client *http.Client
 }
 
-type Serializers struct {
-	Encoder             runtime.Encoder
-	Decoder             runtime.Decoder
-	StreamingSerializer runtime.Serializer
-	Framer              runtime.Framer
-	RenegotiatedDecoder func(contentType string, params map[string]string) (runtime.Decoder, error)
-}
-
 // NewRESTClient creates a new RESTClient. This client performs generic REST functions
-// such as Get, Put, Post, and Delete on specified paths.  Codec controls encoding and
-// decoding of responses from the server.
-func NewRESTClient(baseURL *url.URL, versionedAPIPath string, config ContentConfig, maxQPS float32, maxBurst int, rateLimiter flowcontrol.RateLimiter, client *http.Client) (*RESTClient, error) {
+// such as Get, Put, Post, and Delete on specified paths.
+func NewRESTClient(baseURL *url.URL, versionedAPIPath string, config ClientContentConfig, rateLimiter flowcontrol.RateLimiter, client *http.Client) (*RESTClient, error) {
+	if len(config.ContentType) == 0 {
+		config.ContentType = "application/json"
+	}
+
 	base := *baseURL
 	if !strings.HasSuffix(base.Path, "/") {
 		base.Path += "/"
 	}
 	base.RawQuery = ""
 	base.Fragment = ""
 
-	if config.GroupVersion == nil {
-		config.GroupVersion = &schema.GroupVersion{}
-	}
-	if len(config.ContentType) == 0 {
-		config.ContentType = "application/json"
-	}
-	serializers, err := createSerializers(config)
-	if err != nil {
-		return nil, err
-	}
-
-	var throttle flowcontrol.RateLimiter
-	if maxQPS > 0 && rateLimiter == nil {
-		throttle = flowcontrol.NewTokenBucketRateLimiter(maxQPS, maxBurst)
-	} else if rateLimiter != nil {
-		throttle = rateLimiter
-	}
 	return &RESTClient{
 		base:             &base,
 		versionedAPIPath: versionedAPIPath,
-		contentConfig:    config,
-		serializers:      *serializers,
+		content:          config,
 		createBackoffMgr: readExpBackoffConfig,
-		Throttle:         throttle,
-		Client:           client,
+		rateLimiter:      rateLimiter,
+
+		Client: client,
 	}, nil
 }
 
@@ -132,7 +128,7 @@ func (c *RESTClient) GetRateLimiter() flowcontrol.RateLimiter {
 	if c == nil {
 		return nil
 	}
-	return c.Throttle
+	return c.rateLimiter
 }
 
 // readExpBackoffConfig handles the internal logic of determining what the
@@ -153,58 +149,6 @@ func readExpBackoffConfig() BackoffManager {
 			time.Duration(backoffDurationInt)*time.Second)}
 }
 
-// createSerializers creates all necessary serializers for given contentType.
-// TODO: the negotiated serializer passed to this method should probably return
-//   serializers that control decoding and versioning without this package
-//   being aware of the types. Depends on whether RESTClient must deal with
-//   generic infrastructure.
-func createSerializers(config ContentConfig) (*Serializers, error) {
-	mediaTypes := config.NegotiatedSerializer.SupportedMediaTypes()
-	contentType := config.ContentType
-	mediaType, _, err := mime.ParseMediaType(contentType)
-	if err != nil {
-		return nil, fmt.Errorf("the content type specified in the client configuration is not recognized: %v", err)
-	}
-	info, ok := runtime.SerializerInfoForMediaType(mediaTypes, mediaType)
-	if !ok {
-		if len(contentType) != 0 || len(mediaTypes) == 0 {
-			return nil, fmt.Errorf("no serializers registered for %s", contentType)
-		}
-		info = mediaTypes[0]
-	}
-
-	internalGV := schema.GroupVersions{
-		{
-			Group:   config.GroupVersion.Group,
-			Version: runtime.APIVersionInternal,
-		},
-		// always include the legacy group as a decoding target to handle non-error `Status` return types
-		{
-			Group:   "",
-			Version: runtime.APIVersionInternal,
-		},
-	}
-
-	s := &Serializers{
-		Encoder: config.NegotiatedSerializer.EncoderForVersion(info.Serializer, *config.GroupVersion),
-		Decoder: config.NegotiatedSerializer.DecoderToVersion(info.Serializer, internalGV),
-
-		RenegotiatedDecoder: func(contentType string, params map[string]string) (runtime.Decoder, error) {
-			info, ok := runtime.SerializerInfoForMediaType(mediaTypes, contentType)
-			if !ok {
-				return nil, fmt.Errorf("serializer for %s not registered", contentType)
-			}
-			return config.NegotiatedSerializer.DecoderToVersion(info.Serializer, internalGV), nil
-		},
-	}
-	if info.StreamSerializer != nil {
-		s.StreamingSerializer = info.StreamSerializer.Serializer
-		s.Framer = info.StreamSerializer.Framer
-	}
-
-	return s, nil
-}
-
 // Verb begins a request with a verb (GET, POST, PUT, DELETE).
 //
 // Example usage of RESTClient's request building interface:
@@ -219,12 +163,7 @@ func createSerializers(config ContentConfig) (*Serializers, error) {
 // list, ok := resp.(*api.PodList)
 //
 func (c *RESTClient) Verb(verb string) *Request {
-	backoff := c.createBackoffMgr()
-
-	if c.Client == nil {
-		return NewRequest(nil, verb, c.base, c.versionedAPIPath, c.contentConfig, c.serializers, backoff, c.Throttle, 0)
-	}
-	return NewRequest(c.Client, verb, c.base, c.versionedAPIPath, c.contentConfig, c.serializers, backoff, c.Throttle, c.Client.Timeout)
+	return NewRequest(c).Verb(verb)
 }
 
 // Post begins a POST request. Short for c.Verb("POST").
@@ -254,5 +193,5 @@ func (c *RESTClient) Delete() *Request {
 
 // APIVersion returns the APIVersion this RESTClient is expected to use.
 func (c *RESTClient) APIVersion() schema.GroupVersion {
-	return *c.contentConfig.GroupVersion
+	return c.content.GroupVersion
 }

rest/client_test.go

@@ -27,7 +27,7 @@ import (
 
 	"fmt"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	v1beta1 "k8s.io/api/extensions/v1beta1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -57,12 +57,14 @@ func TestSerializer(t *testing.T) {
 		NegotiatedSerializer: scheme.Codecs.WithoutConversion(),
 	}
 
-	serializer, err := createSerializers(contentConfig)
+	n := runtime.NewClientNegotiator(contentConfig.NegotiatedSerializer, gv)
+	d, err := n.Decoder("application/json", nil)
 	if err != nil {
 		t.Fatal(err)
 	}
+
 	// bytes based on actual return from API server when encoding an "unversioned" object
-	obj, err := runtime.Decode(serializer.Decoder, []byte(`{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Success"}`))
+	obj, err := runtime.Decode(d, []byte(`{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Success"}`))
 	t.Log(obj)
 	if err != nil {
 		t.Fatal(err)

rest/config.go

@@ -269,6 +269,9 @@ type ContentConfig struct {
 	GroupVersion *schema.GroupVersion
 	// NegotiatedSerializer is used for obtaining encoders and decoders for multiple
 	// supported media types.
+	//
+	// TODO: NegotiatedSerializer will be phased out as internal clients are removed
+	//   from Kubernetes.
 	NegotiatedSerializer runtime.NegotiatedSerializer
 }
 
@@ -283,14 +286,6 @@ func RESTClientFor(config *Config) (*RESTClient, error) {
 	if config.NegotiatedSerializer == nil {
 		return nil, fmt.Errorf("NegotiatedSerializer is required when initializing a RESTClient")
 	}
-	qps := config.QPS
-	if config.QPS == 0.0 {
-		qps = DefaultQPS
-	}
-	burst := config.Burst
-	if config.Burst == 0 {
-		burst = DefaultBurst
-	}
 
 	baseURL, versionedAPIPath, err := defaultServerUrlFor(config)
 	if err != nil {
@@ -310,7 +305,33 @@ func RESTClientFor(config *Config) (*RESTClient, error) {
 		}
 	}
 
-	return NewRESTClient(baseURL, versionedAPIPath, config.ContentConfig, qps, burst, config.RateLimiter, httpClient)
+	rateLimiter := config.RateLimiter
+	if rateLimiter == nil {
+		qps := config.QPS
+		if config.QPS == 0.0 {
+			qps = DefaultQPS
+		}
+		burst := config.Burst
+		if config.Burst == 0 {
+			burst = DefaultBurst
+		}
+		if qps > 0 {
+			rateLimiter = flowcontrol.NewTokenBucketRateLimiter(qps, burst)
+		}
+	}
+
+	var gv schema.GroupVersion
+	if config.GroupVersion != nil {
+		gv = *config.GroupVersion
+	}
+	clientContent := ClientContentConfig{
+		AcceptContentTypes: config.AcceptContentTypes,
+		ContentType:        config.ContentType,
+		GroupVersion:       gv,
+		Negotiator:         runtime.NewClientNegotiator(config.NegotiatedSerializer, gv),
+	}
+
+	return NewRESTClient(baseURL, versionedAPIPath, clientContent, rateLimiter, httpClient)
 }
 
 // UnversionedRESTClientFor is the same as RESTClientFor, except that it allows
@@ -338,13 +359,33 @@ func UnversionedRESTClientFor(config *Config) (*RESTClient, error) {
 		}
 	}
 
-	versionConfig := config.ContentConfig
-	if versionConfig.GroupVersion == nil {
-		v := metav1.SchemeGroupVersion
-		versionConfig.GroupVersion = &v
+	rateLimiter := config.RateLimiter
+	if rateLimiter == nil {
+		qps := config.QPS
+		if config.QPS == 0.0 {
+			qps = DefaultQPS
+		}
+		burst := config.Burst
+		if config.Burst == 0 {
+			burst = DefaultBurst
+		}
+		if qps > 0 {
+			rateLimiter = flowcontrol.NewTokenBucketRateLimiter(qps, burst)
+		}
+	}
+
+	gv := metav1.SchemeGroupVersion
+	if config.GroupVersion != nil {
+		gv = *config.GroupVersion
+	}
+	clientContent := ClientContentConfig{
+		AcceptContentTypes: config.AcceptContentTypes,
+		ContentType:        config.ContentType,
+		GroupVersion:       gv,
+		Negotiator:         runtime.NewClientNegotiator(config.NegotiatedSerializer, gv),
 	}
 
-	return NewRESTClient(baseURL, versionedAPIPath, versionConfig, config.QPS, config.Burst, config.RateLimiter, httpClient)
+	return NewRESTClient(baseURL, versionedAPIPath, clientContent, rateLimiter, httpClient)
 }
 
 // SetKubernetesDefaults sets default values on the provided client config for accessing the