published by bot

(https://github.com/kubernetes/contrib/tree/master/mungegithub)

copied from https://github.com/kubernetes/kubernetes.git, branch master,
last commit is 71ba8a90f0a4f3a307cffeb8f8566d13277cb135

rest/client.go

@@ -18,6 +18,7 @@ package rest
 
 import (
 	"fmt"
+	"mime"
 	"net/http"
 	"net/url"
 	"os"
@@ -38,6 +39,18 @@ const (
 	envBackoffDuration = "KUBE_CLIENT_BACKOFF_DURATION"
 )
 
+// Interface captures the set of operations for generically interacting with Kubernetes REST apis.
+type Interface interface {
+	GetRateLimiter() flowcontrol.RateLimiter
+	Verb(verb string) *Request
+	Post() *Request
+	Put() *Request
+	Patch(pt api.PatchType) *Request
+	Get() *Request
+	Delete() *Request
+	APIVersion() unversioned.GroupVersion
+}
+
 // 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
@@ -141,34 +154,48 @@ func readExpBackoffConfig() BackoffManager {
 }
 
 // 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) {
-	negotiated := config.NegotiatedSerializer
+	mediaTypes := config.NegotiatedSerializer.SupportedMediaTypes()
 	contentType := config.ContentType
-	info, ok := negotiated.SerializerForMediaType(contentType, nil)
-	if !ok {
-		return nil, fmt.Errorf("serializer for %s not registered", 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)
 	}
-	streamInfo, ok := negotiated.StreamingSerializerForMediaType(contentType, nil)
+	info, ok := runtime.SerializerInfoForMediaType(mediaTypes, mediaType)
 	if !ok {
-		return nil, fmt.Errorf("streaming serializer for %s not registered", contentType)
+		if len(contentType) != 0 || len(mediaTypes) == 0 {
+			return nil, fmt.Errorf("no serializers registered for %s", contentType)
+		}
+		info = mediaTypes[0]
 	}
+
 	internalGV := unversioned.GroupVersion{
 		Group:   config.GroupVersion.Group,
 		Version: runtime.APIVersionInternal,
 	}
-	return &Serializers{
-		Encoder:             negotiated.EncoderForVersion(info.Serializer, *config.GroupVersion),
-		Decoder:             negotiated.DecoderToVersion(info.Serializer, internalGV),
-		StreamingSerializer: streamInfo.Serializer,
-		Framer:              streamInfo.Framer,
+
+	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) {
-			renegotiated, ok := negotiated.SerializerForMediaType(contentType, params)
+			info, ok := runtime.SerializerInfoForMediaType(mediaTypes, contentType)
 			if !ok {
 				return nil, fmt.Errorf("serializer for %s not registered", contentType)
 			}
-			return negotiated.DecoderToVersion(renegotiated.Serializer, internalGV), nil
+			return config.NegotiatedSerializer.DecoderToVersion(info.Serializer, internalGV), nil
 		},
-	}, 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).

rest/config_test.go

@@ -153,20 +153,8 @@ var fakeWrapperFunc = func(http.RoundTripper) http.RoundTripper {
 
 type fakeNegotiatedSerializer struct{}
 
-func (n *fakeNegotiatedSerializer) SupportedMediaTypes() []string {
-	return []string{}
-}
-
-func (n *fakeNegotiatedSerializer) SerializerForMediaType(mediaType string, params map[string]string) (s runtime.SerializerInfo, ok bool) {
-	return runtime.SerializerInfo{}, true
-}
-
-func (n *fakeNegotiatedSerializer) SupportedStreamingMediaTypes() []string {
-	return []string{}
-}
-
-func (n *fakeNegotiatedSerializer) StreamingSerializerForMediaType(mediaType string, params map[string]string) (s runtime.StreamSerializerInfo, ok bool) {
-	return runtime.StreamSerializerInfo{}, true
+func (n *fakeNegotiatedSerializer) SupportedMediaTypes() []runtime.SerializerInfo {
+	return nil
 }
 
 func (n *fakeNegotiatedSerializer) EncoderForVersion(serializer runtime.Encoder, gv runtime.GroupVersioner) runtime.Encoder {

rest/fake/fake.go

@@ -0,0 +1,127 @@
+/*
+Copyright 2014 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.
+*/
+
+// This is made a separate package and should only be imported by tests, because
+// it imports testapi
+package fake
+
+import (
+	"net/http"
+	"net/url"
+
+	"k8s.io/client-go/pkg/api"
+	"k8s.io/client-go/pkg/api/testapi"
+	"k8s.io/client-go/pkg/api/unversioned"
+	"k8s.io/client-go/pkg/apimachinery/registered"
+	"k8s.io/client-go/pkg/runtime"
+	"k8s.io/client-go/pkg/util/flowcontrol"
+	"k8s.io/client-go/rest"
+)
+
+func CreateHTTPClient(roundTripper func(*http.Request) (*http.Response, error)) *http.Client {
+	return &http.Client{
+		Transport: roundTripperFunc(roundTripper),
+	}
+}
+
+type roundTripperFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}
+
+// RESTClient provides a fake RESTClient interface.
+type RESTClient struct {
+	Client               *http.Client
+	NegotiatedSerializer runtime.NegotiatedSerializer
+	GroupName            string
+
+	Req  *http.Request
+	Resp *http.Response
+	Err  error
+}
+
+func (c *RESTClient) Get() *rest.Request {
+	return c.request("GET")
+}
+
+func (c *RESTClient) Put() *rest.Request {
+	return c.request("PUT")
+}
+
+func (c *RESTClient) Patch(_ api.PatchType) *rest.Request {
+	return c.request("PATCH")
+}
+
+func (c *RESTClient) Post() *rest.Request {
+	return c.request("POST")
+}
+
+func (c *RESTClient) Delete() *rest.Request {
+	return c.request("DELETE")
+}
+
+func (c *RESTClient) Verb(verb string) *rest.Request {
+	return c.request(verb)
+}
+
+func (c *RESTClient) APIVersion() unversioned.GroupVersion {
+	return *(testapi.Default.GroupVersion())
+}
+
+func (c *RESTClient) GetRateLimiter() flowcontrol.RateLimiter {
+	return nil
+}
+
+func (c *RESTClient) request(verb string) *rest.Request {
+	config := rest.ContentConfig{
+		ContentType:          runtime.ContentTypeJSON,
+		GroupVersion:         &registered.GroupOrDie(api.GroupName).GroupVersion,
+		NegotiatedSerializer: c.NegotiatedSerializer,
+	}
+
+	groupName := api.GroupName
+	if c.GroupName != "" {
+		groupName = c.GroupName
+	}
+	ns := c.NegotiatedSerializer
+	info, _ := runtime.SerializerInfoForMediaType(ns.SupportedMediaTypes(), runtime.ContentTypeJSON)
+	internalVersion := unversioned.GroupVersion{
+		Group:   registered.GroupOrDie(groupName).GroupVersion.Group,
+		Version: runtime.APIVersionInternal,
+	}
+	internalVersion.Version = runtime.APIVersionInternal
+	serializers := rest.Serializers{
+		Encoder: ns.EncoderForVersion(info.Serializer, registered.GroupOrDie(api.GroupName).GroupVersion),
+		Decoder: ns.DecoderToVersion(info.Serializer, internalVersion),
+	}
+	if info.StreamSerializer != nil {
+		serializers.StreamingSerializer = info.StreamSerializer.Serializer
+		serializers.Framer = info.StreamSerializer.Framer
+	}
+	return rest.NewRequest(c, verb, &url.URL{Host: "localhost"}, "", config, serializers, nil, nil)
+}
+
+func (c *RESTClient) Do(req *http.Request) (*http.Response, error) {
+	if c.Err != nil {
+		return nil, c.Err
+	}
+	c.Req = req
+	if c.Client != nil {
+		return c.Client.Do(req)
+	}
+	return c.Resp, nil
+}

rest/request.go

@@ -795,6 +795,12 @@ func (r *Request) request(fn func(*http.Request, *http.Response)) error {
 		req.Header = r.headers
 
 		r.backoffMgr.Sleep(r.backoffMgr.CalculateBackoff(r.URL()))
+		if retries > 0 {
+			// We are retrying the request that we already send to apiserver
+			// at least once before.
+			// This request should also be throttled with the client-internal throttler.
+			r.tryThrottle()
+		}
 		resp, err := client.Do(req)
 		updateURLMetrics(r, resp, err)
 		if err != nil {