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 is 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
ensures that the router will always reload after the initial sync.

Author: marun

pkg/router/controller/controller.go

@@ -116,16 +116,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()
 
-			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 +165,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.updateLastSyncProcessed()
 }
 
 // HandleEndpoints handles a single Endpoints event and refreshes the router backend.
@@ -189,14 +190,14 @@ func (c *RouterController) HandleEndpoints() {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
+	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.updateLastSyncProcessed()
-
-	if err := c.Plugin.HandleEndpoints(eventType, endpoints); err != nil {
-		utilruntime.HandleError(err)
-	}
 }
 
 // updateLastSyncProcessed notifies the plugin if the most recent sync

pkg/router/template/router.go

@@ -90,6 +90,8 @@ type templateRouter struct {
 	bindPortsAfterSync bool
 	// whether the router state has been read from the api at least once
 	syncedAtLeastOnce bool
+	// whether a commit has been requested when skipCommit was true
+	commitRequested bool
 }
 
 // templateRouterCfg holds all configuration items required to initialize the template router
@@ -348,8 +350,10 @@ func (r *templateRouter) readState() error {
 func (r *templateRouter) Commit() {
 	if r.skipCommit {
 		glog.V(4).Infof("Skipping router commit until last sync has been processed")
+		r.commitRequested = true
 	} else {
 		r.rateLimitedCommitFunction.Invoke(r.rateLimitedCommitFunction)
+		r.commitRequested = false
 	}
 }
 
@@ -742,9 +746,13 @@ func (r *templateRouter) shouldWriteCerts(cfg *ServiceAliasConfig) bool {
 // SetSkipCommit indicates to the router whether requests to
 // commit/reload should be skipped.
 func (r *templateRouter) SetSkipCommit(skipCommit bool) {
-	if r.skipCommit != skipCommit {
-		glog.V(4).Infof("Updating skip commit to: %t", skipCommit)
-		r.skipCommit = skipCommit
+	if r.skipCommit == skipCommit {
+		return
+	}
+	glog.V(4).Infof("Updating skip commit to: %t", skipCommit)
+	r.skipCommit = skipCommit
+	if r.skipCommit && r.commitRequested {
+		r.Commit()
 	}
 }
 

test/integration/router_stress_test.go

@@ -85,6 +85,17 @@ func stressRouter(t *testing.T, namespaceCount, routesPerNamespace, routerCount,
 				t.Fatalf("unexpected error: %v", err)
 			}
 			routes = append(routes, route)
+
+		}
+		// Create a final route that uses the same host as the
+		// previous route to create a conflict.  This will validate
+		// whether a router still reloads if the last item in the
+		// initial list is rejected.
+		host := routes[len(routes)-1].Spec.Host
+		routeProperties := createRouteProperties("invalid-service", host)
+		_, err = oc.Routes(namespace.Name).Create(routeProperties)
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
 		}
 	}
 
@@ -105,9 +116,9 @@ func stressRouter(t *testing.T, namespaceCount, routesPerNamespace, routerCount,
 		plugins = append(plugins, router)
 	}
 
-	// Wait until the routers have processed all the routes.  The test
-	// runner is assumed enforce a timeout that ensures termination in
-	// the event of a failure.
+	// Wait until routers have processed all the routes.  The test
+	// runner is assumed to enforce a timeout that ensures termination
+	// in the event of failure.
 	expectedRouteCount := len(routes)
 	for {
 		routeCount := 0
@@ -126,7 +137,10 @@ func stressRouter(t *testing.T, namespaceCount, routesPerNamespace, routerCount,
 	}
 
 	for _, reloadCount := range reloadCounts {
-		if reloadCount > 1 {
+		switch {
+		case reloadCount == 0:
+			t.Fatalf("One or more routers didn't reload")
+		case reloadCount > 1:
 			t.Fatalf("One or more routers reloaded more than once")
 		}
 	}