do not bounce router for endpoint changes that do not change the data

plugins/router/template/plugin.go

@@ -31,13 +31,15 @@ type router interface {
 	// FindServiceUnit finds the service with the given id.
 	FindServiceUnit(id string) (v ServiceUnit, ok bool)
 
-	// AddEndpoints adds new Endpoints for the given id.
-	AddEndpoints(id string, endpoints []Endpoint)
+	// AddEndpoints adds new Endpoints for the given id.Returns true if a change was made
+	// and the state should be stored with Commit().
+	AddEndpoints(id string, endpoints []Endpoint) bool
 	// DeleteEndpoints deletes the endpoints for the frontend with the given id.
 	DeleteEndpoints(id string)
 
-	// AddRoute adds a route for the given id
-	AddRoute(id string, route *routeapi.Route)
+	// AddRoute adds a route for the given id.  Returns true if a change was made
+	// and the state should be stored with Commit().
+	AddRoute(id string, route *routeapi.Route) bool
 	// RemoveRoute removes the given route for the given id.
 	RemoveRoute(id string, route *routeapi.Route)
 
@@ -76,18 +78,18 @@ func (p *TemplatePlugin) HandleEndpoints(eventType watch.EventType, endpoints *k
 		p.Router.CreateServiceUnit(key)
 	}
 
-	// clear existing endpoints
-	p.Router.DeleteEndpoints(key)
-
 	switch eventType {
 	case watch.Added, watch.Modified:
 		glog.V(4).Infof("Modifying endpoints for %s", key)
 		routerEndpoints := createRouterEndpoints(endpoints)
 		key := endpointsKey(*endpoints)
-		p.Router.AddEndpoints(key, routerEndpoints)
+		commit := p.Router.AddEndpoints(key, routerEndpoints)
+		if commit {
+			return p.Router.Commit()
+		}
 	}
 
-	return p.Router.Commit()
+	return nil
 }
 
 // HandleRoute processes watch events on the Route resource.
@@ -101,13 +103,16 @@ func (p *TemplatePlugin) HandleRoute(eventType watch.EventType, route *routeapi.
 	switch eventType {
 	case watch.Added, watch.Modified:
 		glog.V(4).Infof("Modifying routes for %s", key)
-		p.Router.AddRoute(key, route)
+		commit := p.Router.AddRoute(key, route)
+		if commit {
+			return p.Router.Commit()
+		}
 	case watch.Deleted:
 		glog.V(4).Infof("Deleting routes for %s", key)
 		p.Router.RemoveRoute(key, route)
+		return p.Router.Commit()
 	}
-
-	return p.Router.Commit()
+	return nil
 }
 
 // routeKey returns the internal router key to use for the given Route.

plugins/router/template/plugin_test.go

@@ -2,6 +2,7 @@ package templaterouter
 
 import (
 	"fmt"
+	"reflect"
 	"testing"
 
 	kapi "github.com/GoogleCloudPlatform/kubernetes/pkg/api"
@@ -44,16 +45,17 @@ func (r *TestRouter) FindServiceUnit(id string) (v ServiceUnit, ok bool) {
 }
 
 // AddEndpoints adds the endpoints to the service unit identified by id
-func (r *TestRouter) AddEndpoints(id string, endpoints []Endpoint) {
+func (r *TestRouter) AddEndpoints(id string, endpoints []Endpoint) bool {
 	r.Committed = false //expect any call to this method to subsequently call commit
 	su, _ := r.FindServiceUnit(id)
 
-	for _, ep := range endpoints {
-		newEndpoint := Endpoint{ep.ID, ep.IP, ep.Port, "foo"}
-		su.EndpointTable = append(su.EndpointTable, newEndpoint)
+	// simulate the logic that compares endpoints
+	if reflect.DeepEqual(su.EndpointTable, endpoints) {
+		return false
 	}
-
+	su.EndpointTable = endpoints
 	r.State[id] = su
+	return true
 }
 
 // DeleteEndpoints removes all endpoints from the service unit
@@ -68,7 +70,7 @@ func (r *TestRouter) DeleteEndpoints(id string) {
 }
 
 // AddRoute adds a ServiceAliasConfig for the route to the ServiceUnit identified by id
-func (r *TestRouter) AddRoute(id string, route *routeapi.Route) {
+func (r *TestRouter) AddRoute(id string, route *routeapi.Route) bool {
 	r.Committed = false //expect any call to this method to subsequently call commit
 	su, _ := r.FindServiceUnit(id)
 	routeKey := r.routeKey(route)
@@ -80,6 +82,7 @@ func (r *TestRouter) AddRoute(id string, route *routeapi.Route) {
 
 	su.ServiceAliasConfigs[routeKey] = config
 	r.State[id] = su
+	return true
 }
 
 // RemoveRoute removes the service alias config for Route from the ServiceUnit
@@ -286,3 +289,71 @@ func TestHandleRoute(t *testing.T) {
 	}
 
 }
+
+func TestUnchangingEndpointsDoesNotCommit(t *testing.T) {
+	router := newTestRouter(make(map[string]ServiceUnit))
+	plugin := TemplatePlugin{Router: router}
+	endpoints := &kapi.Endpoints{
+		ObjectMeta: kapi.ObjectMeta{
+			Namespace: "foo",
+			Name:      "test",
+		},
+		Subsets: []kapi.EndpointSubset{{
+			Addresses: []kapi.EndpointAddress{{IP: "1.1.1.1"}, {IP: "2.2.2.2"}},
+			Ports:     []kapi.EndpointPort{{Port: 0}},
+		}},
+	}
+	changedEndpoints := &kapi.Endpoints{
+		ObjectMeta: kapi.ObjectMeta{
+			Namespace: "foo",
+			Name:      "test",
+		},
+		Subsets: []kapi.EndpointSubset{{
+			Addresses: []kapi.EndpointAddress{{IP: "3.3.3.3"}, {IP: "2.2.2.2"}},
+			Ports:     []kapi.EndpointPort{{Port: 0}},
+		}},
+	}
+
+	testCases := []struct {
+		name         string
+		event        watch.EventType
+		endpoints    *kapi.Endpoints
+		expectCommit bool
+	}{
+		{
+			name:         "initial add",
+			event:        watch.Added,
+			endpoints:    endpoints,
+			expectCommit: true,
+		},
+		{
+			name:         "mod with no change",
+			event:        watch.Modified,
+			endpoints:    endpoints,
+			expectCommit: false,
+		},
+		{
+			name:         "add with change",
+			event:        watch.Added,
+			endpoints:    changedEndpoints,
+			expectCommit: true,
+		},
+		{
+			name:         "add with no change",
+			event:        watch.Added,
+			endpoints:    changedEndpoints,
+			expectCommit: false,
+		},
+	}
+
+	for _, v := range testCases {
+		err := plugin.HandleEndpoints(v.event, v.endpoints)
+		if err != nil {
+			t.Errorf("%s had unexecpected error in handle endpoints %v", v.name, err)
+			continue
+		}
+		if router.Committed != v.expectCommit {
+			t.Errorf("%s expected router commit to be %v but found %v", v.name, v.expectCommit, router.Committed)
+		}
+	}
+}

plugins/router/template/router.go

@@ -271,7 +271,7 @@ func (r *templateRouter) routeKey(route *routeapi.Route) string {
 }
 
 // AddRoute adds a route for the given id
-func (r *templateRouter) AddRoute(id string, route *routeapi.Route) {
+func (r *templateRouter) AddRoute(id string, route *routeapi.Route) bool {
 	frontend, _ := r.FindServiceUnit(id)
 
 	backendKey := r.routeKey(route)
@@ -323,6 +323,7 @@ func (r *templateRouter) AddRoute(id string, route *routeapi.Route) {
 	//create or replace
 	frontend.ServiceAliasConfigs[backendKey] = config
 	r.state[id] = frontend
+	return true
 }
 
 // RemoveRoute removes the given route for the given id.
@@ -342,12 +343,13 @@ func (r *templateRouter) RemoveRoute(id string, route *routeapi.Route) {
 }
 
 // AddEndpoints adds new Endpoints for the given id.
-func (r *templateRouter) AddEndpoints(id string, endpoints []Endpoint) {
+func (r *templateRouter) AddEndpoints(id string, endpoints []Endpoint) bool {
 	frontend, _ := r.FindServiceUnit(id)
 
 	//only make the change if there is a difference
 	if reflect.DeepEqual(frontend.EndpointTable, endpoints) {
-		return
+		glog.V(4).Infof("Ignoring change for %s, endpoints are the same", id)
+		return false
 	}
 
 	frontend.EndpointTable = endpoints
@@ -357,6 +359,8 @@ func (r *templateRouter) AddEndpoints(id string, endpoints []Endpoint) {
 		r.peerEndpoints = frontend.EndpointTable
 		glog.V(4).Infof("Peer endpoints updated to: %#v", r.peerEndpoints)
 	}
+
+	return true
 }
 
 // cleanUpServiceAliasConfig performs any necessary steps to clean up a service alias config before deleting it from

plugins/router/template/router_test.go

@@ -68,6 +68,76 @@ func TestAddEndpoints(t *testing.T) {
 	}
 }
 
+// Test that AddEndpoints returns true and false correctly for changed endpoints.
+func TestAddEndpointDuplicates(t *testing.T) {
+	router := newFakeTemplateRouter()
+	suKey := "test"
+	router.CreateServiceUnit(suKey)
+	if _, ok := router.FindServiceUnit(suKey); !ok {
+		t.Fatalf("Unable to find serivce unit %s after creation", suKey)
+	}
+
+	endpoint := Endpoint{
+		ID:   "ep1",
+		IP:   "1.1.1.1",
+		Port: "80",
+	}
+	endpoint2 := Endpoint{
+		ID:   "ep2",
+		IP:   "2.2.2.2",
+		Port: "8080",
+	}
+	endpoint3 := Endpoint{
+		ID:   "ep3",
+		IP:   "3.3.3.3",
+		Port: "8888",
+	}
+
+	testCases := []struct {
+		name      string
+		endpoints []Endpoint
+		expected  bool
+	}{
+		{
+			name:      "initial add",
+			endpoints: []Endpoint{endpoint, endpoint2},
+			expected:  true,
+		},
+		{
+			name:      "add same endpoints",
+			endpoints: []Endpoint{endpoint, endpoint2},
+			expected:  false,
+		},
+		{
+			name:      "add changed endpoints",
+			endpoints: []Endpoint{endpoint3, endpoint2},
+			expected:  true,
+		},
+	}
+
+	for _, v := range testCases {
+		added := router.AddEndpoints(suKey, v.endpoints)
+		if added != v.expected {
+			t.Errorf("%s expected to return %v but got %v", v.name, v.expected, added)
+		}
+		su, ok := router.FindServiceUnit(suKey)
+		if !ok {
+			t.Errorf("%s was unable to find created service unit %s", v.name, suKey)
+			continue
+		}
+		if len(su.EndpointTable) != len(v.endpoints) {
+			t.Errorf("%s expected endpoint table to contain %d entries but found %v", v.name, len(v.endpoints), su.EndpointTable)
+			continue
+		}
+		for i, ep := range su.EndpointTable {
+			expected := v.endpoints[i]
+			if expected.IP != ep.IP || expected.Port != ep.Port {
+				t.Errorf("%s expected endpoint %v did not match actual endpoint %v", v.name, endpoint, ep)
+			}
+		}
+	}
+}
+
 // TestDeleteEndpoints tests removing endpoints from service units
 func TestDeleteEndpoints(t *testing.T) {
 	router := newFakeTemplateRouter()
@@ -147,7 +217,11 @@ func TestAddRoute(t *testing.T) {
 	suKey := "test"
 	router.CreateServiceUnit(suKey)
 
-	router.AddRoute(suKey, route)
+	// add route always returns true
+	added := router.AddRoute(suKey, route)
+	if !added {
+		t.Fatalf("expected AddRoute to return true but got false")
+	}
 
 	su, ok := router.FindServiceUnit(suKey)