Merge pull request #1077 from brendandburns/service

Step #1 in migrating the service controller away from the apiserver.

pkg/client/client.go

@@ -318,6 +318,11 @@ func (c *Client) WatchReplicationControllers(label, field labels.Selector, resou
 		Watch()
 }
 
+func (c *Client) ListServices(selector labels.Selector) (list api.ServiceList, err error) {
+	err = c.Get().Path("services").SelectorParam("labels", selector).Do().Into(&list)
+	return
+}
+
 // GetService returns information about a particular service.
 func (c *Client) GetService(name string) (result api.Service, err error) {
 	err = c.Get().Path("services").Path(name).Do().Into(&result)

pkg/client/client_test.go

@@ -407,6 +407,57 @@ func (c *testClient) Validate(t *testing.T, received interface{}, err error) {
 	}
 }
 
+func TestListServices(t *testing.T) {
+	c := &testClient{
+		Request: testRequest{Method: "GET", Path: "/services"},
+		Response: Response{StatusCode: 200,
+			Body: api.ServiceList{
+				Items: []api.Service{
+					{
+						JSONBase: api.JSONBase{ID: "name"},
+						Labels: map[string]string{
+							"foo":  "bar",
+							"name": "baz",
+						},
+						Selector: map[string]string{
+							"one": "two",
+						},
+					},
+				},
+			},
+		},
+	}
+	receivedServiceList, err := c.Setup().ListServices(labels.Everything())
+	c.Validate(t, receivedServiceList, err)
+}
+
+func TestListServicesLabels(t *testing.T) {
+	c := &testClient{
+		Request: testRequest{Method: "GET", Path: "/services", Query: url.Values{"labels": []string{"foo=bar,name=baz"}}},
+		Response: Response{StatusCode: 200,
+			Body: api.ServiceList{
+				Items: []api.Service{
+					{
+						JSONBase: api.JSONBase{ID: "name"},
+						Labels: map[string]string{
+							"foo":  "bar",
+							"name": "baz",
+						},
+						Selector: map[string]string{
+							"one": "two",
+						},
+					},
+				},
+			},
+		},
+	}
+	c.Setup()
+	c.QueryValidator["labels"] = validateLabels
+	selector := labels.Set{"foo": "bar", "name": "baz"}.AsSelector()
+	receivedServiceList, err := c.ListServices(selector)
+	c.Validate(t, receivedServiceList, err)
+}
+
 func TestGetService(t *testing.T) {
 	c := &testClient{
 		Request:  testRequest{Method: "GET", Path: "/services/1"},

pkg/service/endpoints_controller.go

@@ -46,9 +46,9 @@ func NewEndpointController(serviceRegistry service.Registry, client *client.Clie
 
 // SyncServiceEndpoints syncs service endpoints.
 func (e *EndpointController) SyncServiceEndpoints() error {
-	services, err := e.serviceRegistry.ListServices()
+	services, err := e.client.ListServices(labels.Everything())
 	if err != nil {
-		glog.Errorf("Failed to list services!")
+		glog.Errorf("Failed to list services: %v", err)
 		return err
 	}
 	var resultErr error

pkg/service/endpoints_controller_test.go

@@ -17,8 +17,8 @@ limitations under the License.
 package service
 
 import (
-	"encoding/json"
 	"fmt"
+	"net/http"
 	"net/http/httptest"
 	"testing"
 
@@ -121,13 +121,34 @@ func TestFindPort(t *testing.T) {
 	}
 }
 
-func TestSyncEndpointsEmpty(t *testing.T) {
+type serverResponse struct {
-	body, _ := json.Marshal(newPodList(0))
+	statusCode int
-	fakeHandler := util.FakeHandler{
+	obj        interface{}
-		StatusCode:   200,
+}
-		ResponseBody: string(body),
+
+func makeTestServer(t *testing.T, podResponse serverResponse, serviceResponse serverResponse) *httptest.Server {
+	fakePodHandler := util.FakeHandler{
+		StatusCode:   podResponse.statusCode,
+		ResponseBody: util.EncodeJSON(podResponse.obj),
 	}
-	testServer := httptest.NewTLSServer(&fakeHandler)
+	fakeServiceHandler := util.FakeHandler{
+		StatusCode:   serviceResponse.statusCode,
+		ResponseBody: util.EncodeJSON(serviceResponse.obj),
+	}
+	mux := http.NewServeMux()
+	mux.Handle("/api/v1beta1/pods", &fakePodHandler)
+	mux.Handle("/api/v1beta1/services", &fakeServiceHandler)
+	mux.HandleFunc("/", func(res http.ResponseWriter, req *http.Request) {
+		t.Errorf("unexpected request: %v", req.RequestURI)
+		res.WriteHeader(http.StatusNotFound)
+	})
+	return httptest.NewServer(mux)
+}
+
+func TestSyncEndpointsEmpty(t *testing.T) {
+	testServer := makeTestServer(t,
+		serverResponse{http.StatusOK, newPodList(0)},
+		serverResponse{http.StatusOK, api.ServiceList{}})
 	client := client.NewOrDie(testServer.URL, nil)
 	serviceRegistry := registrytest.ServiceRegistry{}
 	endpoints := NewEndpointController(&serviceRegistry, client)
@@ -137,55 +158,57 @@ func TestSyncEndpointsEmpty(t *testing.T) {
 }
 
 func TestSyncEndpointsError(t *testing.T) {
-	body, _ := json.Marshal(newPodList(0))
+	testServer := makeTestServer(t,
-	fakeHandler := util.FakeHandler{
+		serverResponse{http.StatusOK, newPodList(0)},
-		StatusCode:   200,
+		serverResponse{http.StatusInternalServerError, api.ServiceList{}})
-		ResponseBody: string(body),
-	}
-	testServer := httptest.NewTLSServer(&fakeHandler)
 	client := client.NewOrDie(testServer.URL, nil)
 	serviceRegistry := registrytest.ServiceRegistry{
 		Err: fmt.Errorf("test error"),
 	}
 	endpoints := NewEndpointController(&serviceRegistry, client)
-	if err := endpoints.SyncServiceEndpoints(); err != serviceRegistry.Err {
+	if err := endpoints.SyncServiceEndpoints(); err == nil {
-		t.Errorf("Errors don't match: %#v %#v", err, serviceRegistry.Err)
+		t.Errorf("unexpected non-error")
 	}
 }
 
 func TestSyncEndpointsItems(t *testing.T) {
-	body, _ := json.Marshal(newPodList(1))
+	serviceList := api.ServiceList{
-	fakeHandler := util.FakeHandler{
+		Items: []api.Service{
-		StatusCode:   200,
+			{
-		ResponseBody: string(body),
+				Selector: map[string]string{
-	}
+					"foo": "bar",
-	testServer := httptest.NewTLSServer(&fakeHandler)
-	client := client.NewOrDie(testServer.URL, nil)
-	serviceRegistry := registrytest.ServiceRegistry{
-		List: api.ServiceList{
-			Items: []api.Service{
-				{
-					Selector: map[string]string{
-						"foo": "bar",
-					},
 				},
 			},
 		},
 	}
+	testServer := makeTestServer(t,
+		serverResponse{http.StatusOK, newPodList(1)},
+		serverResponse{http.StatusOK, serviceList})
+	client := client.NewOrDie(testServer.URL, nil)
+	serviceRegistry := registrytest.ServiceRegistry{}
 	endpoints := NewEndpointController(&serviceRegistry, client)
 	if err := endpoints.SyncServiceEndpoints(); err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}
-	if len(serviceRegistry.Endpoints.Endpoints) != 1 {
+	if len(serviceRegistry.Endpoints.Endpoints) != 1 ||
+		serviceRegistry.Endpoints.Endpoints[0] != "1.2.3.4:8080" {
 		t.Errorf("Unexpected endpoints update: %#v", serviceRegistry.Endpoints)
 	}
 }
 
 func TestSyncEndpointsPodError(t *testing.T) {
-	fakeHandler := util.FakeHandler{
+	serviceList := api.ServiceList{
-		StatusCode: 500,
+		Items: []api.Service{
+			{
+				Selector: map[string]string{
+					"foo": "bar",
+				},
+			},
+		},
 	}
-	testServer := httptest.NewTLSServer(&fakeHandler)
+	testServer := makeTestServer(t,
+		serverResponse{http.StatusInternalServerError, api.PodList{}},
+		serverResponse{http.StatusOK, serviceList})
 	client := client.NewOrDie(testServer.URL, nil)
 	serviceRegistry := registrytest.ServiceRegistry{
 		List: api.ServiceList{