Build backend config using the K8S endpoint resource.

* Potentialy saves a network hop
* Ability to configure LB algothim (given some work to expose an
anotation etc...)
* K8s config Watch is triggered far less often

provider/k8s/client.go

@@ -22,6 +22,7 @@ const (
 type Client interface {
 	GetIngresses(predicate func(Ingress) bool) ([]Ingress, error)
 	GetServices(predicate func(Service) bool) ([]Service, error)
+	GetEndpoints(name, namespace string) (Endpoints, error)
 	WatchAll(stopCh <-chan bool) (chan interface{}, chan error, error)
 }
 
@@ -104,21 +105,26 @@ func (c *clientImpl) WatchServices(stopCh <-chan bool) (chan interface{}, chan e
 	return c.watch(getURL, stopCh)
 }
 
-// WatchEvents returns events in the cluster
-func (c *clientImpl) WatchEvents(stopCh <-chan bool) (chan interface{}, chan error, error) {
-	getURL := c.endpointURL + APIEndpoint + "/events"
-	return c.watch(getURL, stopCh)
+// GetEndpoints returns the named Endpoints
+// Endpoints have the same name as the coresponding service
+func (c *clientImpl) GetEndpoints(name, namespace string) (Endpoints, error) {
+	getURL := c.endpointURL + APIEndpoint + "/namespaces/" + namespace + "/endpoints/" + name
+
+	body, err := c.do(c.request(getURL))
+	if err != nil {
+		return Endpoints{}, fmt.Errorf("failed to create endpoints request: GET %q : %v", getURL, err)
+	}
+
+	var endpoints Endpoints
+	if err := json.Unmarshal(body, &endpoints); err != nil {
+		return Endpoints{}, fmt.Errorf("failed to decode endpoints resources: %v", err)
+	}
+	return endpoints, nil
 }
 
-// WatchPods returns pods in the cluster
-func (c *clientImpl) WatchPods(stopCh <-chan bool) (chan interface{}, chan error, error) {
-	getURL := c.endpointURL + APIEndpoint + "/pods"
-	return c.watch(getURL, stopCh)
-}
-
-// WatchReplicationControllers returns ReplicationControllers in the cluster
-func (c *clientImpl) WatchReplicationControllers(stopCh <-chan bool) (chan interface{}, chan error, error) {
-	getURL := c.endpointURL + APIEndpoint + "/replicationcontrollers"
+// WatchEndpoints returns endpoints in the cluster
+func (c *clientImpl) WatchEndpoints(stopCh <-chan bool) (chan interface{}, chan error, error) {
+	getURL := c.endpointURL + APIEndpoint + "/endpoints"
 	return c.watch(getURL, stopCh)
 }
 
@@ -137,13 +143,8 @@ func (c *clientImpl) WatchAll(stopCh <-chan bool) (chan interface{}, chan error,
 	if err != nil {
 		return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
 	}
-	stopPods := make(chan bool)
-	chanPods, chanPodsErr, err := c.WatchPods(stopPods)
-	if err != nil {
-		return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
-	}
-	stopReplicationControllers := make(chan bool)
-	chanReplicationControllers, chanReplicationControllersErr, err := c.WatchReplicationControllers(stopReplicationControllers)
+	stopEndpoints := make(chan bool)
+	chanEndpoints, chanEndpointsErr, err := c.WatchEndpoints(stopEndpoints)
 	if err != nil {
 		return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
 	}
@@ -152,32 +153,26 @@ func (c *clientImpl) WatchAll(stopCh <-chan bool) (chan interface{}, chan error,
 		defer close(errCh)
 		defer close(stopIngresses)
 		defer close(stopServices)
-		defer close(stopPods)
-		defer close(stopReplicationControllers)
+		defer close(stopEndpoints)
 
 		for {
 			select {
 			case <-stopCh:
 				stopIngresses <- true
 				stopServices <- true
-				stopPods <- true
-				stopReplicationControllers <- true
+				stopEndpoints <- true
 				return
 			case err := <-chanIngressesErr:
 				errCh <- err
 			case err := <-chanServicesErr:
 				errCh <- err
-			case err := <-chanPodsErr:
-				errCh <- err
-			case err := <-chanReplicationControllersErr:
+			case err := <-chanEndpointsErr:
 				errCh <- err
 			case event := <-chanIngresses:
 				watchCh <- event
 			case event := <-chanServices:
 				watchCh <- event
-			case event := <-chanPods:
-				watchCh <- event
-			case event := <-chanReplicationControllers:
+			case event := <-chanEndpoints:
 				watchCh <- event
 			}
 		}

provider/k8s/endpoints.go

@@ -0,0 +1,84 @@
+package k8s
+
+// Endpoints is a collection of endpoints that implement the actual service.  Example:
+//   Name: "mysvc",
+//   Subsets: [
+//     {
+//       Addresses: [{"ip": "10.10.1.1"}, {"ip": "10.10.2.2"}],
+//       Ports: [{"name": "a", "port": 8675}, {"name": "b", "port": 309}]
+//     },
+//     {
+//       Addresses: [{"ip": "10.10.3.3"}],
+//       Ports: [{"name": "a", "port": 93}, {"name": "b", "port": 76}]
+//     },
+//  ]
+type Endpoints struct {
+	TypeMeta   `json:",inline"`
+	ObjectMeta `json:"metadata,omitempty"`
+
+	// The set of all endpoints is the union of all subsets.
+	Subsets []EndpointSubset
+}
+
+// EndpointSubset is a group of addresses with a common set of ports.  The
+// expanded set of endpoints is the Cartesian product of Addresses x Ports.
+// For example, given:
+//   {
+//     Addresses: [{"ip": "10.10.1.1"}, {"ip": "10.10.2.2"}],
+//     Ports:     [{"name": "a", "port": 8675}, {"name": "b", "port": 309}]
+//   }
+// The resulting set of endpoints can be viewed as:
+//     a: [ 10.10.1.1:8675, 10.10.2.2:8675 ],
+//     b: [ 10.10.1.1:309, 10.10.2.2:309 ]
+type EndpointSubset struct {
+	Addresses         []EndpointAddress
+	NotReadyAddresses []EndpointAddress
+	Ports             []EndpointPort
+}
+
+// EndpointAddress is a tuple that describes single IP address.
+type EndpointAddress struct {
+	// The IP of this endpoint.
+	// IPv6 is also accepted but not fully supported on all platforms. Also, certain
+	// kubernetes components, like kube-proxy, are not IPv6 ready.
+	// TODO: This should allow hostname or IP, see #4447.
+	IP string
+	// Optional: Hostname of this endpoint
+	// Meant to be used by DNS servers etc.
+	Hostname string `json:"hostname,omitempty"`
+	// Optional: The kubernetes object related to the entry point.
+	TargetRef *ObjectReference
+}
+
+// EndpointPort is a tuple that describes a single port.
+type EndpointPort struct {
+	// The name of this port (corresponds to ServicePort.Name).  Optional
+	// if only one port is defined.  Must be a DNS_LABEL.
+	Name string
+
+	// The port number.
+	Port int32
+
+	// The IP protocol for this port.
+	Protocol Protocol
+}
+
+// ObjectReference contains enough information to let you inspect or modify the referred object.
+type ObjectReference struct {
+	Kind            string `json:"kind,omitempty"`
+	Namespace       string `json:"namespace,omitempty"`
+	Name            string `json:"name,omitempty"`
+	UID             UID    `json:"uid,omitempty"`
+	APIVersion      string `json:"apiVersion,omitempty"`
+	ResourceVersion string `json:"resourceVersion,omitempty"`
+
+	// Optional. If referring to a piece of an object instead of an entire object, this string
+	// should contain information to identify the sub-object. For example, if the object
+	// reference is to a container within a pod, this would take on a value like:
+	// "spec.containers{name}" (where "name" refers to the name of the container that triggered
+	// the event) or if no container name is specified "spec.containers[2]" (container with
+	// index 2 in this pod). This syntax is chosen only to have some well-defined way of
+	// referencing a part of an object.
+	// TODO: this design is not final and this field is subject to change in the future.
+	FieldPath string `json:"fieldPath,omitempty"`
+}

provider/kubernetes.go

@@ -209,9 +209,27 @@ func (provider *Kubernetes) loadIngresses(k8sClient k8s.Client) (*types.Configur
 							if port.Port == 443 {
 								protocol = "https"
 							}
-							templateObjects.Backends[r.Host+pa.Path].Servers[string(service.UID)] = types.Server{
-								URL:    protocol + "://" + service.Spec.ClusterIP + ":" + strconv.Itoa(port.Port),
-								Weight: 1,
+							endpoints, err := k8sClient.GetEndpoints(service.ObjectMeta.Name, service.ObjectMeta.Namespace)
+							if err != nil {
+								log.Errorf("Error retrieving endpoints: %v", err)
+								continue
+							}
+							if len(endpoints.Subsets) == 0 {
+								log.Warnf("Endpoints not found for %s/%s, falling back to Service ClusterIP", service.ObjectMeta.Namespace, service.ObjectMeta.Name)
+								templateObjects.Backends[r.Host+pa.Path].Servers[string(service.UID)] = types.Server{
+									URL:    protocol + "://" + service.Spec.ClusterIP + ":" + strconv.Itoa(port.Port),
+									Weight: 1,
+								}
+							} else {
+								for _, subset := range endpoints.Subsets {
+									for _, address := range subset.Addresses {
+										url := protocol + "://" + address.IP + ":" + strconv.Itoa(endpointPortNumber(port, subset.Ports))
+										templateObjects.Backends[r.Host+pa.Path].Servers[url] = types.Server{
+											URL:    url,
+											Weight: 1,
+										}
+									}
+								}
 							}
 							break
 						}
@@ -223,6 +241,20 @@ func (provider *Kubernetes) loadIngresses(k8sClient k8s.Client) (*types.Configur
 	return &templateObjects, nil
 }
 
+func endpointPortNumber(servicePort k8s.ServicePort, endpointPorts []k8s.EndpointPort) int {
+	if len(endpointPorts) > 0 {
+		//name is optional if there is only one port
+		port := endpointPorts[0]
+		for _, endpointPort := range endpointPorts {
+			if servicePort.Name == endpointPort.Name {
+				port = endpointPort
+			}
+		}
+		return int(port.Port)
+	}
+	return servicePort.Port
+}
+
 func equalPorts(servicePort k8s.ServicePort, ingressPort k8s.IntOrString) bool {
 	if servicePort.Port == ingressPort.IntValue() {
 		return true

provider/kubernetes_test.go

@@ -10,6 +10,9 @@ import (
 
 func TestLoadIngresses(t *testing.T) {
 	ingresses := []k8s.Ingress{{
+		ObjectMeta: k8s.ObjectMeta{
+			Namespace: "testing",
+		},
 		Spec: k8s.IngressSpec{
 			Rules: []k8s.IngressRule{
 				{
@@ -55,23 +58,25 @@ func TestLoadIngresses(t *testing.T) {
 	services := []k8s.Service{
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service1",
-				UID:  "1",
+				Name:      "service1",
+				UID:       "1",
+				Namespace: "testing",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.1",
 				Ports: []k8s.ServicePort{
 					{
 						Name: "http",
-						Port: 801,
+						Port: 80,
 					},
 				},
 			},
 		},
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service2",
-				UID:  "2",
+				Name:      "service2",
+				UID:       "2",
+				Namespace: "testing",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.2",
@@ -84,8 +89,9 @@ func TestLoadIngresses(t *testing.T) {
 		},
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service3",
-				UID:  "3",
+				Name:      "service3",
+				UID:       "3",
+				Namespace: "testing",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.3",
@@ -98,10 +104,46 @@ func TestLoadIngresses(t *testing.T) {
 			},
 		},
 	}
+	endpoints := []k8s.Endpoints{
+		{
+			ObjectMeta: k8s.ObjectMeta{
+				Name:      "service1",
+				UID:       "1",
+				Namespace: "testing",
+			},
+			Subsets: []k8s.EndpointSubset{
+				{
+					Addresses: []k8s.EndpointAddress{
+						{
+							IP: "10.10.0.1",
+						},
+					},
+					Ports: []k8s.EndpointPort{
+						{
+							Port: 8080,
+						},
+					},
+				},
+				{
+					Addresses: []k8s.EndpointAddress{
+						{
+							IP: "10.21.0.1",
+						},
+					},
+					Ports: []k8s.EndpointPort{
+						{
+							Port: 8080,
+						},
+					},
+				},
+			},
+		},
+	}
 	watchChan := make(chan interface{})
 	client := clientMock{
 		ingresses: ingresses,
 		services:  services,
+		endpoints: endpoints,
 		watchChan: watchChan,
 	}
 	provider := Kubernetes{}
@@ -114,8 +156,12 @@ func TestLoadIngresses(t *testing.T) {
 		Backends: map[string]*types.Backend{
 			"foo/bar": {
 				Servers: map[string]types.Server{
-					"1": {
-						URL:    "http://10.0.0.1:801",
+					"http://10.10.0.1:8080": {
+						URL:    "http://10.10.0.1:8080",
+						Weight: 1,
+					},
+					"http://10.21.0.1:8080": {
+						URL:    "http://10.21.0.1:8080",
 						Weight: 1,
 					},
 				},
@@ -1150,6 +1196,7 @@ func TestHostlessIngress(t *testing.T) {
 type clientMock struct {
 	ingresses []k8s.Ingress
 	services  []k8s.Service
+	endpoints []k8s.Endpoints
 	watchChan chan interface{}
 }
 
@@ -1174,6 +1221,16 @@ func (c clientMock) GetServices(predicate func(k8s.Service) bool) ([]k8s.Service
 	}
 	return services, nil
 }
+
+func (c clientMock) GetEndpoints(name, namespace string) (k8s.Endpoints, error) {
+	for _, endpoints := range c.endpoints {
+		if endpoints.Namespace == namespace && endpoints.Name == name {
+			return endpoints, nil
+		}
+	}
+	return k8s.Endpoints{}, nil
+}
+
 func (c clientMock) WatchAll(stopCh <-chan bool) (chan interface{}, chan error, error) {
 	return c.watchChan, make(chan error), nil
 }