Ensure router reload on initial sync

Previously, the router wouldn't reload HAProxy after the initial sync
if the last item of the initial list of any of the watched resources
didn't reach the router to trigger a commit.  One possible trigger for
this condition was a route specifying a host already claimed by
another namespace. The router could be left in its initial state until
another commit-triggering event occurred (e.g. a watch event).

This change refactors the commit handling so that the plugin no longer
triggers a commit directly.  Instead:

 - the plugin tracks whether state has changed
 - a commit is only attempted if state has changed
 - assuming a sync is not in progress, the controller will trigger a
   commit attempt after calling one of the relevant plugin Handle methods

Author: marun

pkg/router/controller/controller.go

@@ -36,6 +36,7 @@ type RouterController struct {
 	routesListConsumed    bool
 	endpointsListConsumed bool
 	filteredByNamespace   bool
+	syncing               bool
 
 	RoutesListSuccessfulAtLeastOnce    func() bool
 	EndpointsListSuccessfulAtLeastOnce func() bool
@@ -78,25 +79,26 @@ func (c *RouterController) handleFirstSync() bool {
 		return false
 	}
 
+	err := c.Plugin.SetSyncedAtLeastOnce()
+	if err != nil {
+		utilruntime.HandleError(err)
+		return false
+	}
+
 	// If either of the event queues were empty after the initial
 	// List, the tracking listConsumed variable's default value of
-	// 'false' may prevent the router from reloading to indicate the
-	// readiness status.  Set the value to 'true' to ensure that a
-	// reload will be performed if necessary.
+	// 'false' may prevent the router from committing the readiness
+	// status.  Set the value to 'true' to ensure that state will be
+	// committed if necessary.
 	if c.RoutesListCount() == 0 {
 		c.routesListConsumed = true
 	}
 	if c.EndpointsListCount() == 0 {
 		c.endpointsListConsumed = true
 	}
-	c.updateLastSyncProcessed()
+	c.commit()
 
-	err := c.Plugin.SetSyncedAtLeastOnce()
-	if err == nil {
-		return true
-	}
-	utilruntime.HandleError(err)
-	return false
+	return true
 }
 
 // watchForFirstSync loops until the first sync has been handled.
@@ -116,16 +118,17 @@ func (c *RouterController) HandleNamespaces() {
 			c.lock.Lock()
 			defer c.lock.Unlock()
 
+			glog.V(4).Infof("Updating watched namespaces: %v", namespaces)
+			if err := c.Plugin.HandleNamespaces(namespaces); err != nil {
+				utilruntime.HandleError(err)
+			}
+
 			// Namespace filtering is assumed to be have been
 			// performed so long as the plugin event handler is called
 			// at least once.
 			c.filteredByNamespace = true
-			c.updateLastSyncProcessed()
+			c.commit()
 
-			glog.V(4).Infof("Updating watched namespaces: %v", namespaces)
-			if err := c.Plugin.HandleNamespaces(namespaces); err != nil {
-				utilruntime.HandleError(err)
-			}
 			return
 		}
 		utilruntime.HandleError(fmt.Errorf("unable to find namespaces for router: %v", err))
@@ -164,18 +167,18 @@ func (c *RouterController) HandleRoute() {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	// Change the local sync state within the lock to ensure that all
-	// event handlers have the same view of sync state.
-	c.routesListConsumed = c.RoutesListConsumed()
-	c.updateLastSyncProcessed()
-
 	glog.V(4).Infof("Processing Route: %s -> %s", route.Name, route.Spec.To.Name)
 	glog.V(4).Infof("           Alias: %s", route.Spec.Host)
 	glog.V(4).Infof("           Event: %s", eventType)
 
 	if err := c.Plugin.HandleRoute(eventType, route); err != nil {
 		utilruntime.HandleError(err)
 	}
+
+	// Change the local sync state within the lock to ensure that all
+	// event handlers have the same view of sync state.
+	c.routesListConsumed = c.RoutesListConsumed()
+	c.commit()
 }
 
 // HandleEndpoints handles a single Endpoints event and refreshes the router backend.
@@ -189,22 +192,32 @@ func (c *RouterController) HandleEndpoints() {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	// Change the local sync state within the lock to ensure that all
-	// event handlers have the same view of sync state.
-	c.endpointsListConsumed = c.EndpointsListConsumed()
-	c.updateLastSyncProcessed()
-
 	if err := c.Plugin.HandleEndpoints(eventType, endpoints); err != nil {
 		utilruntime.HandleError(err)
 	}
+
+	// Change the local sync state within the lock to ensure that all
+	// event handlers have the same view of sync state.
+	c.endpointsListConsumed = c.EndpointsListConsumed()
+	c.commit()
 }
 
-// updateLastSyncProcessed notifies the plugin if the most recent sync
-// of router resources has been completed.
-func (c *RouterController) updateLastSyncProcessed() {
-	lastSyncProcessed := c.endpointsListConsumed && c.routesListConsumed &&
-		(c.Namespaces == nil || c.filteredByNamespace)
-	if err := c.Plugin.SetLastSyncProcessed(lastSyncProcessed); err != nil {
+// commit notifies the plugin that it is safe to commit state.
+func (c *RouterController) commit() {
+	syncing := !(c.endpointsListConsumed && c.routesListConsumed &&
+		(c.Namespaces == nil || c.filteredByNamespace))
+	if syncing != c.syncing {
+		c.syncing = syncing
+		if c.syncing {
+			glog.V(4).Infof("Router sync in progress")
+		} else {
+			glog.V(4).Infof("Router sync complete")
+		}
+	}
+	if c.syncing {
+		return
+	}
+	if err := c.Plugin.Commit(); err != nil {
 		utilruntime.HandleError(err)
 	}
 }

pkg/router/controller/controller_test.go

@@ -11,8 +11,7 @@ import (
 )
 
 type fakeRouterPlugin struct {
-	lastSyncProcessed bool
-	syncedAtLeastOnce bool
+	commitRequested bool
 }
 
 func (p *fakeRouterPlugin) HandleRoute(t watch.EventType, route *routeapi.Route) error {
@@ -28,13 +27,12 @@ func (p *fakeRouterPlugin) HandleNamespaces(namespaces sets.String) error {
 	return nil
 }
 
-func (p *fakeRouterPlugin) SetLastSyncProcessed(processed bool) error {
-	p.lastSyncProcessed = processed
+func (p *fakeRouterPlugin) Commit() error {
+	p.commitRequested = true
 	return nil
 }
 
 func (p *fakeRouterPlugin) SetSyncedAtLeastOnce() error {
-	p.syncedAtLeastOnce = true
 	return nil
 }
 
@@ -45,7 +43,7 @@ func (n fakeNamespaceLister) NamespaceNames() (sets.String, error) {
 	return sets.NewString("foo"), nil
 }
 
-func TestRouterController_updateLastSyncProcessed(t *testing.T) {
+func TestRouterController_commit(t *testing.T) {
 	p := fakeRouterPlugin{}
 	routesListConsumed := true
 	c := RouterController{
@@ -69,30 +67,34 @@ func TestRouterController_updateLastSyncProcessed(t *testing.T) {
 		NamespaceRetries: 1,
 	}
 
+	expectedMsg := "commit not expected to have been requested"
+	notExpectedMsg := "commit expected to have been requested"
+
 	// Simulate the initial sync
 	c.HandleNamespaces()
-	if p.lastSyncProcessed {
-		t.Fatalf("last sync not expected to have been processed")
+	if p.commitRequested {
+		t.Fatalf(notExpectedMsg)
 	}
 	c.HandleEndpoints()
-	if p.lastSyncProcessed {
-		t.Fatalf("last sync not expected to have been processed")
+	if p.commitRequested {
+		t.Fatalf(notExpectedMsg)
 	}
 	c.HandleRoute()
-	if !p.lastSyncProcessed {
-		t.Fatalf("last sync expected to have been processed")
+	if !p.commitRequested {
+		t.Fatalf(expectedMsg)
 	}
 
 	// Simulate a relist
+	p.commitRequested = false
 	routesListConsumed = false
 	c.HandleRoute()
-	if p.lastSyncProcessed {
-		t.Fatalf("last sync not expected to have been processed")
+	if p.commitRequested {
+		t.Fatalf(notExpectedMsg)
 	}
 	routesListConsumed = true
 	c.HandleRoute()
-	if !p.lastSyncProcessed {
-		t.Fatalf("last sync expected to have been processed")
+	if !p.commitRequested {
+		t.Fatalf(expectedMsg)
 	}
 
 }

pkg/router/controller/extended_validator.go

@@ -79,8 +79,8 @@ func (p *ExtendedValidator) HandleNamespaces(namespaces sets.String) error {
 	return p.plugin.HandleNamespaces(namespaces)
 }
 
-func (p *ExtendedValidator) SetLastSyncProcessed(processed bool) error {
-	return p.plugin.SetLastSyncProcessed(processed)
+func (p *ExtendedValidator) Commit() error {
+	return p.plugin.Commit()
 }
 
 func (p *ExtendedValidator) SetSyncedAtLeastOnce() error {

pkg/router/controller/host_admitter.go

@@ -148,8 +148,8 @@ func (p *HostAdmitter) HandleNamespaces(namespaces sets.String) error {
 	return p.plugin.HandleNamespaces(namespaces)
 }
 
-func (p *HostAdmitter) SetLastSyncProcessed(processed bool) error {
-	return p.plugin.SetLastSyncProcessed(processed)
+func (p *HostAdmitter) Commit() error {
+	return p.plugin.Commit()
 }
 
 func (p *HostAdmitter) SetSyncedAtLeastOnce() error {

pkg/router/controller/status.go

@@ -312,8 +312,8 @@ func (a *StatusAdmitter) HandleNamespaces(namespaces sets.String) error {
 	return a.plugin.HandleNamespaces(namespaces)
 }
 
-func (a *StatusAdmitter) SetLastSyncProcessed(processed bool) error {
-	return a.plugin.SetLastSyncProcessed(processed)
+func (a *StatusAdmitter) Commit() error {
+	return a.plugin.Commit()
 }
 
 func (a *StatusAdmitter) SetSyncedAtLeastOnce() error {

pkg/router/controller/status_test.go

@@ -39,7 +39,7 @@ func (p *fakePlugin) HandleEndpoints(watch.EventType, *kapi.Endpoints) error {
 func (p *fakePlugin) HandleNamespaces(namespaces sets.String) error {
 	return fmt.Errorf("not expected")
 }
-func (p *fakePlugin) SetLastSyncProcessed(processed bool) error {
+func (p *fakePlugin) Commit() error {
 	return fmt.Errorf("not expected")
 }
 

pkg/router/controller/unique_host.go

@@ -255,8 +255,8 @@ func (p *UniqueHost) HandleNamespaces(namespaces sets.String) error {
 	return p.plugin.HandleNamespaces(namespaces)
 }
 
-func (p *UniqueHost) SetLastSyncProcessed(processed bool) error {
-	return p.plugin.SetLastSyncProcessed(processed)
+func (p *UniqueHost) Commit() error {
+	return p.plugin.Commit()
 }
 
 func (p *UniqueHost) SetSyncedAtLeastOnce() error {

pkg/router/f5/plugin.go

@@ -630,7 +630,7 @@ func (p *F5Plugin) HandleRoute(eventType watch.EventType,
 }
 
 // No-op since f5 configuration can be updated piecemeal
-func (p *F5Plugin) SetLastSyncProcessed(processed bool) error {
+func (p *F5Plugin) Commit() error {
 	return nil
 }
 

pkg/router/interfaces.go

@@ -16,6 +16,6 @@ type Plugin interface {
 	// If sent, filter the list of accepted routes and endpoints to this set
 	HandleNamespaces(namespaces sets.String) error
 	HandleNode(watch.EventType, *kapi.Node) error
-	SetLastSyncProcessed(processed bool) error
+	Commit() error
 	SetSyncedAtLeastOnce() error
 }

pkg/router/template/plugin.go

@@ -65,17 +65,15 @@ type routerInterface interface {
 	// FindServiceUnit finds the service with the given id.
 	FindServiceUnit(id string) (v ServiceUnit, ok bool)
 
-	// 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
+	// AddEndpoints adds new Endpoints for the given id.
+	AddEndpoints(id string, endpoints []Endpoint)
 	// DeleteEndpoints deletes the endpoints for the frontend with the given id.
 	DeleteEndpoints(id string)
 
 	// AddRoute adds a route for the given id and the calculated host. Weight
 	// suggests the weightage attached to it with respect to other services
-	// pointed to by the route. Returns true if a
-	// change was made and the state should be stored with Commit().
-	AddRoute(id string, weight int32, route *routeapi.Route, host string) bool
+	// pointed to by the route.
+	AddRoute(id string, weight int32, route *routeapi.Route, host string)
 	// RemoveRoute removes the given route
 	RemoveRoute(route *routeapi.Route)
 	// HasRoute indicates whether the router is configured with the given route
@@ -86,9 +84,6 @@ type routerInterface interface {
 	// commit (persist router state + refresh the backend) that coalesces multiple changes.
 	Commit()
 
-	// SetSkipCommit indicates to the router whether commits should be skipped
-	SetSkipCommit(skipCommit bool)
-
 	// SetSyncedAtLeastOnce indicates to the router that state has been read from the api at least once
 	SetSyncedAtLeastOnce()
 }
@@ -171,14 +166,10 @@ func (p *TemplatePlugin) HandleEndpoints(eventType watch.EventType, endpoints *k
 		glog.V(4).Infof("Modifying endpoints for %s", key)
 		routerEndpoints := createRouterEndpoints(endpoints, !p.IncludeUDP, p.ServiceFetcher)
 		key := endpointsKey(endpoints)
-		commit := p.Router.AddEndpoints(key, routerEndpoints)
-		if commit {
-			p.Router.Commit()
-		}
+		p.Router.AddEndpoints(key, routerEndpoints)
 	case watch.Deleted:
 		glog.V(4).Infof("Deleting endpoints for %s", key)
 		p.Router.DeleteEndpoints(key)
-		p.Router.Commit()
 	}
 
 	return nil
@@ -206,7 +197,6 @@ func (p *TemplatePlugin) HandleRoute(eventType watch.EventType, route *routeapi.
 		p.Router.RemoveRoute(route)
 
 		// Now add the route back again
-		commit := false
 		for i := range serviceKeys {
 			key := serviceKeys[i]
 			weight := weights[i]
@@ -216,16 +206,11 @@ func (p *TemplatePlugin) HandleRoute(eventType watch.EventType, route *routeapi.
 			}
 
 			glog.V(4).Infof("Modifying routes for %s", key)
-			commitRoute := p.Router.AddRoute(key, weight, route, host)
-			commit = (map[bool]bool{true: true, false: commit})[commitRoute]
-		}
-		if commit {
-			p.Router.Commit()
+			p.Router.AddRoute(key, weight, route, host)
 		}
 	case watch.Deleted:
 		glog.V(4).Infof("Deleting route %v", route)
 		p.Router.RemoveRoute(route)
-		p.Router.Commit()
 	}
 	return nil
 }
@@ -234,18 +219,16 @@ func (p *TemplatePlugin) HandleRoute(eventType watch.EventType, route *routeapi.
 // the provided namespace list.
 func (p *TemplatePlugin) HandleNamespaces(namespaces sets.String) error {
 	p.Router.FilterNamespaces(namespaces)
-	p.Router.Commit()
 	return nil
 }
 
-func (p *TemplatePlugin) SetLastSyncProcessed(processed bool) error {
-	p.Router.SetSkipCommit(!processed)
+func (p *TemplatePlugin) Commit() error {
+	p.Router.Commit()
 	return nil
 }
 
 func (p *TemplatePlugin) SetSyncedAtLeastOnce() error {
 	p.Router.SetSyncedAtLeastOnce()
-	p.Router.Commit()
 	return nil
 }
 

pkg/router/template/plugin_test.go

@@ -245,9 +245,6 @@ func (r *TestRouter) Commit() {
 	r.Committed = true
 }
 
-func (r *TestRouter) SetSkipCommit(skipCommit bool) {
-}
-
 func (r *TestRouter) SetSyncedAtLeastOnce() {
 }