route advertisement should account for services using externalTrafficPolicy=Local (#334)

Author: andrewsykim

app/controllers/network_routes_controller.go

@@ -31,14 +31,16 @@ import (
 	"github.com/osrg/gobgp/table"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/vishvananda/netlink"
+	v1core "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
 )
 
 // NetworkRoutingController is struct to hold necessary information required by controller
 type NetworkRoutingController struct {
 	nodeIP               net.IP
-	nodeHostName         string
+	nodeName             string
 	nodeSubnet           net.IPNet
 	nodeInterface        string
 	activeNodes          map[string]bool
@@ -370,8 +372,20 @@ func (nrc *NetworkRoutingController) advertiseClusterIPs() {
 			if svc.Spec.ClusterIP == "None" || svc.Spec.ClusterIP == "" {
 				continue
 			}
-
 			glog.V(2).Info("found a service of cluster ip type")
+
+			if svc.Spec.ExternalTrafficPolicy == v1core.ServiceExternalTrafficPolicyTypeLocal {
+				nodeHasEndpoints, err := nrc.nodeHasEndpointsForService(svc)
+				if err != nil {
+					glog.Errorf("error determining if node has endpoints for svc: %q error: %v", svc.Name, err)
+					continue
+				}
+
+				if !nodeHasEndpoints {
+					continue
+				}
+			}
+
 			err := nrc.AdvertiseClusterIp(svc.Spec.ClusterIP)
 			if err != nil {
 				glog.Errorf("error advertising cluster IP: %q error: %v", svc.Spec.ClusterIP, err)
@@ -388,6 +402,19 @@ func (nrc *NetworkRoutingController) advertiseExternalIPs() {
 			if svc.Spec.ClusterIP == "None" || svc.Spec.ClusterIP == "" {
 				continue
 			}
+
+			if svc.Spec.ExternalTrafficPolicy == v1core.ServiceExternalTrafficPolicyTypeLocal {
+				nodeHasEndpoints, err := nrc.nodeHasEndpointsForService(svc)
+				if err != nil {
+					glog.Errorf("error determining if node has endpoints for svc: %q error: %v", svc.Name, err)
+					continue
+				}
+
+				if !nodeHasEndpoints {
+					continue
+				}
+			}
+
 			for _, externalIP := range svc.Spec.ExternalIPs {
 				err := nrc.AdvertiseClusterIp(externalIP)
 				if err != nil {
@@ -452,6 +479,40 @@ func (nrc *NetworkRoutingController) getExternalIps() ([]string, error) {
 	return externalIpList, nil
 }
 
+// nodeHasEndpointsForService will get the corresponding Endpoints resource for a given Service
+// return true if any endpoint addresses has NodeName matching the node name of the route controller
+func (nrc *NetworkRoutingController) nodeHasEndpointsForService(svc *v1core.Service) (bool, error) {
+	// listers for endpoints and services should use the same keys since
+	// endpoint and service resources share the same object name and namespace
+	key, err := cache.MetaNamespaceKeyFunc(svc)
+	if err != nil {
+		return false, err
+	}
+	item, exists, err := watchers.EndpointsWatcher.GetByKey(key)
+	if err != nil {
+		return false, err
+	}
+
+	if !exists {
+		return false, fmt.Errorf("endpoint resource doesn't exist for service: %q", svc.Name)
+	}
+
+	ep, ok := item.(*v1core.Endpoints)
+	if !ok {
+		return false, errors.New("failed to convert cache item to Endpoints type")
+	}
+
+	for _, subset := range ep.Subsets {
+		for _, address := range subset.Addresses {
+			if *address.NodeName == nrc.nodeName {
+				return true, nil
+			}
+		}
+	}
+
+	return false, nil
+}
+
 // Used for processing Annotations that may contain multiple items
 // Pass this the string and the delimiter
 func stringToSlice(s, d string) []string {
@@ -1563,7 +1624,7 @@ func NewNetworkRoutingController(clientset *kubernetes.Clientset,
 		return nil, errors.New("Failed getting node object from API server: " + err.Error())
 	}
 
-	nrc.nodeHostName = node.Name
+	nrc.nodeName = node.Name
 
 	nodeIP, err := utils.GetNodeIP(node)
 	if err != nil {

app/controllers/network_routes_controller_test.go

@@ -394,6 +394,137 @@ func Test_advertiseExternalIPs(t *testing.T) {
 	}
 }
 
+func Test_nodeHasEndpointsForService(t *testing.T) {
+	testcases := []struct {
+		name             string
+		nrc              *NetworkRoutingController
+		existingService  *v1core.Service
+		existingEndpoint *v1core.Endpoints
+		nodeHasEndpoints bool
+		err              error
+	}{
+		{
+			"node has endpoints for service",
+			&NetworkRoutingController{
+				nodeName: "node-1",
+			},
+			&v1core.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "svc-1",
+					Namespace: "default",
+				},
+				Spec: v1core.ServiceSpec{
+					Type:        "ClusterIP",
+					ClusterIP:   "10.0.0.1",
+					ExternalIPs: []string{"1.1.1.1", "2.2.2.2"},
+				},
+			},
+			&v1core.Endpoints{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "svc-1",
+					Namespace: "default",
+				},
+				Subsets: []v1core.EndpointSubset{
+					{
+						Addresses: []v1core.EndpointAddress{
+							{
+								IP:       "172.20.1.1",
+								NodeName: ptrToString("node-1"),
+							},
+							{
+								IP:       "172.20.1.2",
+								NodeName: ptrToString("node-2"),
+							},
+						},
+					},
+				},
+			},
+			true,
+			nil,
+		},
+		{
+			"node has no endpoints for service",
+			&NetworkRoutingController{
+				nodeName: "node-1",
+			},
+			&v1core.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "svc-1",
+					Namespace: "default",
+				},
+				Spec: v1core.ServiceSpec{
+					Type:        "ClusterIP",
+					ClusterIP:   "10.0.0.1",
+					ExternalIPs: []string{"1.1.1.1", "2.2.2.2"},
+				},
+			},
+			&v1core.Endpoints{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "svc-1",
+					Namespace: "default",
+				},
+				Subsets: []v1core.EndpointSubset{
+					{
+						Addresses: []v1core.EndpointAddress{
+							{
+								IP:       "172.20.1.1",
+								NodeName: ptrToString("node-2"),
+							},
+							{
+								IP:       "172.20.1.2",
+								NodeName: ptrToString("node-3"),
+							},
+						},
+					},
+				},
+			},
+			false,
+			nil,
+		},
+	}
+
+	for _, testcase := range testcases {
+		t.Run(testcase.name, func(t *testing.T) {
+			clientset := fake.NewSimpleClientset()
+
+			_, err := watchers.StartServiceWatcher(clientset, 0)
+			if err != nil {
+				t.Fatalf("failed to initialize service watcher: %v", err)
+			}
+
+			_, err = watchers.StartEndpointsWatcher(clientset, 0)
+			if err != nil {
+				t.Fatalf("failed to initialize endpoints watcher: %v", err)
+			}
+
+			_, err = clientset.CoreV1().Endpoints("default").Create(testcase.existingEndpoint)
+			if err != nil {
+				t.Fatalf("failed to create existing endpoints: %v", err)
+			}
+
+			_, err = clientset.CoreV1().Services("default").Create(testcase.existingService)
+			if err != nil {
+				t.Fatalf("failed to create existing services: %v", err)
+			}
+
+			waitForListerWithTimeout(time.Second*10, t)
+
+			nodeHasEndpoints, err := testcase.nrc.nodeHasEndpointsForService(testcase.existingService)
+			if !reflect.DeepEqual(err, testcase.err) {
+				t.Logf("actual err: %v", err)
+				t.Logf("expected err: %v", testcase.err)
+				t.Error("unexpected error")
+			}
+			if nodeHasEndpoints != testcase.nodeHasEndpoints {
+				t.Logf("expected nodeHasEndpoints: %v", testcase.nodeHasEndpoints)
+				t.Logf("actual nodeHasEndpoints: %v", nodeHasEndpoints)
+				t.Error("unexpected nodeHasEndpoints")
+			}
+
+		})
+	}
+}
+
 func Test_advertiseRoute(t *testing.T) {
 	testcases := []struct {
 		name        string
@@ -969,3 +1100,7 @@ func waitForBGPWatchEventWithTimeout(timeout time.Duration, expectedNumEvents in
 		}
 	}
 }
+
+func ptrToString(str string) *string {
+	return &str
+}

app/watchers/endpoints_watcher.go

@@ -7,7 +7,8 @@ import (
 	"github.com/cloudnativelabs/kube-router/utils"
 	api "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 )
@@ -31,7 +32,7 @@ var (
 )
 
 type endpointsWatcher struct {
-	clientset           *kubernetes.Clientset
+	clientset           kubernetes.Interface
 	endpointsController cache.Controller
 	endpointsLister     cache.Indexer
 	broadcaster         *utils.Broadcaster
@@ -84,13 +85,17 @@ func (ew *endpointsWatcher) List() []*api.Endpoints {
 	return epInstances
 }
 
+func (ew *endpointsWatcher) GetByKey(key string) (item interface{}, exists bool, err error) {
+	return ew.endpointsLister.GetByKey(key)
+}
+
 func (ew *endpointsWatcher) HasSynced() bool {
 	return ew.endpointsController.HasSynced()
 }
 
 var endpointsStopCh chan struct{}
 
-func StartEndpointsWatcher(clientset *kubernetes.Clientset, resyncPeriod time.Duration) (*endpointsWatcher, error) {
+func StartEndpointsWatcher(clientset kubernetes.Interface, resyncPeriod time.Duration) (*endpointsWatcher, error) {
 
 	ew := endpointsWatcher{}
 	EndpointsWatcher = &ew
@@ -103,7 +108,14 @@ func StartEndpointsWatcher(clientset *kubernetes.Clientset, resyncPeriod time.Du
 
 	ew.clientset = clientset
 	ew.broadcaster = utils.NewBroadcaster()
-	lw := cache.NewListWatchFromClient(clientset.CoreV1().RESTClient(), "endpoints", metav1.NamespaceAll, fields.Everything())
+	lw := &cache.ListWatch{
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+			return clientset.CoreV1().Endpoints(metav1.NamespaceAll).List(options)
+		},
+		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+			return clientset.CoreV1().Endpoints(metav1.NamespaceAll).Watch(options)
+		},
+	}
 	ew.endpointsLister, ew.endpointsController = cache.NewIndexerInformer(
 		lw,
 		&api.Endpoints{}, resyncPeriod, eventHandler,