Merge pull request #13134 from jszczepkowski/client-proxy

pkg/client/unversioned/request.go

@@ -52,6 +52,13 @@ type HTTPClient interface {
 	Do(req *http.Request) (*http.Response, error)
 }
 
+// ResponseWrapper is an interface for getting a response.
+// The response may be either accessed as a raw data (the whole output is put into memory) or as a stream.
+type ResponseWrapper interface {
+	DoRaw() ([]byte, error)
+	Stream() (io.ReadCloser, error)
+}
+
 // RequestConstructionError is returned when there's an error assembling a request.
 type RequestConstructionError struct {
 	Err error

pkg/client/unversioned/services.go

@@ -36,15 +36,16 @@ type ServiceInterface interface {
 	Update(srv *api.Service) (*api.Service, error)
 	Delete(name string) error
 	Watch(label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
+	ProxyGet(name, path string, params map[string]string) ResponseWrapper
 }
 
-// services implements PodsNamespacer interface
+// services implements ServicesNamespacer interface
 type services struct {
 	r  *Client
 	ns string
 }
 
-// newServices returns a PodsClient
+// newServices returns a services
 func newServices(c *Client, namespace string) *services {
 	return &services{c, namespace}
 }
@@ -98,3 +99,17 @@ func (c *services) Watch(label labels.Selector, field fields.Selector, resourceV
 		FieldsSelectorParam(field).
 		Watch()
 }
+
+// ProxyGet returns a response of the service by calling it through the proxy.
+func (c *services) ProxyGet(name, path string, params map[string]string) ResponseWrapper {
+	request := c.r.Get().
+		Prefix("proxy").
+		Namespace(c.ns).
+		Resource("services").
+		Name(name).
+		Suffix(path)
+	for k, v := range params {
+		request = request.Param(k, v)
+	}
+	return request
+}

pkg/client/unversioned/services_test.go

@@ -152,3 +152,18 @@ func TestDeleteService(t *testing.T) {
 	err := c.Setup().Services(ns).Delete("1")
 	c.Validate(t, nil, err)
 }
+
+func TestServiceProxyGet(t *testing.T) {
+	body := "OK"
+	ns := api.NamespaceDefault
+	c := &testClient{
+		Request: testRequest{
+			Method: "GET",
+			Path:   testapi.ResourcePathWithPrefix("proxy", "services", ns, "service-1") + "/foo",
+			Query:  buildQueryValues(url.Values{"param-name": []string{"param-value"}}),
+		},
+		Response: Response{StatusCode: 200, RawBody: &body},
+	}
+	response, err := c.Setup().Services(ns).ProxyGet("service-1", "foo", map[string]string{"param-name": "param-value"}).DoRaw()
+	c.ValidateRaw(t, response, err)
+}

pkg/client/unversioned/testclient/actions.go

@@ -138,6 +138,17 @@ func NewWatchAction(resource, namespace string, label labels.Selector, field fie
 	return action
 }
 
+func NewProxyGetAction(resource, namespace, name, path string, params map[string]string) ProxyGetActionImpl {
+	action := ProxyGetActionImpl{}
+	action.Verb = "get"
+	action.Resource = resource
+	action.Namespace = namespace
+	action.Name = name
+	action.Path = path
+	action.Params = params
+	return action
+}
+
 type ListRestrictions struct {
 	Labels labels.Selector
 	Fields fields.Selector
@@ -191,6 +202,13 @@ type WatchAction interface {
 	GetWatchRestrictions() WatchRestrictions
 }
 
+type ProxyGetAction interface {
+	Action
+	GetName() string
+	GetPath() string
+	GetParams() map[string]string
+}
+
 type ActionImpl struct {
 	Namespace   string
 	Verb        string
@@ -277,3 +295,22 @@ type WatchActionImpl struct {
 func (a WatchActionImpl) GetWatchRestrictions() WatchRestrictions {
 	return a.WatchRestrictions
 }
+
+type ProxyGetActionImpl struct {
+	ActionImpl
+	Name   string
+	Path   string
+	Params map[string]string
+}
+
+func (a ProxyGetActionImpl) GetName() string {
+	return a.Name
+}
+
+func (a ProxyGetActionImpl) GetPath() string {
+	return a.Path
+}
+
+func (a ProxyGetActionImpl) GetParams() map[string]string {
+	return a.Params
+}

pkg/client/unversioned/testclient/fake_services.go

@@ -18,6 +18,7 @@ package testclient
 
 import (
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/fields"
 	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/watch"
@@ -75,3 +76,8 @@ func (c *FakeServices) Watch(label labels.Selector, field fields.Selector, resou
 	c.Fake.Invokes(NewWatchAction("services", c.Namespace, label, field, resourceVersion), nil)
 	return c.Fake.Watch, nil
 }
+
+func (c *FakeServices) ProxyGet(name, path string, params map[string]string) unversioned.ResponseWrapper {
+	c.Fake.Invokes(NewProxyGetAction("services", c.Namespace, name, path, params), nil)
+	return nil
+}

pkg/controller/autoscaler/horizontalpodautoscaler_controller.go

@@ -40,7 +40,7 @@ const (
 )
 
 type HorizontalPodAutoscalerController struct {
-	client    *client.Client
+	client    client.Interface
 	expClient client.ExperimentalInterface
 }
 
@@ -71,8 +71,7 @@ var heapsterQueryStart, _ = time.ParseDuration("-5m")
 var downscaleForbiddenWindow, _ = time.ParseDuration("20m")
 var upscaleForbiddenWindow, _ = time.ParseDuration("3m")
 
-func New(client *client.Client, expClient client.ExperimentalInterface) *HorizontalPodAutoscalerController {
-	//TODO: switch to client.Interface
+func New(client client.Interface, expClient client.ExperimentalInterface) *HorizontalPodAutoscalerController {
 	return &HorizontalPodAutoscalerController{
 		client:    client,
 		expClient: expClient,
@@ -126,16 +125,9 @@ func (a *HorizontalPodAutoscalerController) reconcileAutoscalers() error {
 			strings.Join(podNames, ","),
 			metricSpec.name)
 
-		resultRaw, err := a.client.
-			Get().
-			Prefix("proxy").
-			Resource("services").
-			Namespace(heapsterNamespace).
-			Name(heapsterService).
-			Suffix(metricPath).
-			Param("start", startTime.Format(time.RFC3339)).
-			Do().
-			Raw()
+		resultRaw, err := a.client.Services(heapsterNamespace).
+			ProxyGet(heapsterService, metricPath, map[string]string{"start": startTime.Format(time.RFC3339)}).
+			DoRaw()
 
 		if err != nil {
 			glog.Warningf("Failed to get pods metrics for %s: %v", reference, err)