router: coordinate readiness check with reload

Previously, if either no routes or no endpoints were returned by the
initial List operation of a router, the readiness status would not be
reloaded even though state had been synchronized.  This change ensures
that empty initial Lists still result in a 'ready' router.

Author: marun

pkg/client/cache/eventqueue.go

@@ -49,10 +49,10 @@ type EventQueue struct {
 	// item it refers to is explicitly deleted from the store or the
 	// event is read via Pop().
 	lastReplaceKey string
-	// Tracks whether the Replace() method has been called at least
-	// once.  This indicates that a List operation was successfully
-	// completed against the api even if no items were queued.
+	// Tracks whether the Replace() method has been called at least once.
 	replaceCalled bool
+	// Tracks the number of items queued by the last Replace() call.
+	replaceCount int
 }
 
 // EventQueue implements kcache.Store
@@ -327,6 +327,7 @@ func (eq *EventQueue) Replace(objects []interface{}, resourceVersion string) err
 	defer eq.lock.Unlock()
 
 	eq.replaceCalled = true
+	eq.replaceCount = len(objects)
 
 	eq.events = map[string]watch.EventType{}
 	eq.queue = eq.queue[:0]
@@ -361,6 +362,14 @@ func (eq *EventQueue) ListSuccessfulAtLeastOnce() bool {
 	return eq.replaceCalled
 }
 
+// ListCount returns how many objects were queued by the most recent List operation.
+func (eq *EventQueue) ListCount() int {
+	eq.lock.Lock()
+	defer eq.lock.Unlock()
+
+	return eq.replaceCount
+}
+
 // ListConsumed indicates whether the items queued by a List/Relist
 // operation have been consumed.
 func (eq *EventQueue) ListConsumed() bool {

pkg/router/controller/controller.go

@@ -39,6 +39,8 @@ type RouterController struct {
 
 	RoutesListSuccessfulAtLeastOnce    func() bool
 	EndpointsListSuccessfulAtLeastOnce func() bool
+	RoutesListCount                    func() int
+	EndpointsListCount                 func() int
 
 	WatchNodes bool
 
@@ -63,20 +65,47 @@ func (c *RouterController) Run() {
 	go c.watchForFirstSync()
 }
 
-// watchForFirstSync signals the router when it sees that the various
+// handleFirstSync signals the router when it sees that the various
 // watchers have successfully listed data from the api.
+func (c *RouterController) handleFirstSync() bool {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	synced := c.EndpointsListSuccessfulAtLeastOnce() &&
+		c.RoutesListSuccessfulAtLeastOnce() &&
+		(c.Namespaces == nil || c.filteredByNamespace)
+	if !synced {
+		return false
+	}
+
+	// If either of the event queues were empty after the initial
+	// List, the tracking listConsumed variable's default value of
+	// 'false' will 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.
+	if c.EndpointsListCount() == 0 {
+		c.endpointsListConsumed = true
+	}
+	if c.RoutesListCount() == 0 {
+		c.routesListConsumed = true
+	}
+	c.updateLastSyncProcessed()
+
+	err := c.Plugin.SetSyncedAtLeastOnce()
+	if err == nil {
+		return true
+	}
+	utilruntime.HandleError(err)
+	return false
+}
+
+// watchForFirstSync loops until the first sync has been handled.
 func (c *RouterController) watchForFirstSync() {
 	for {
-		syncedAtLeastOnce := c.RoutesListSuccessfulAtLeastOnce() && c.EndpointsListSuccessfulAtLeastOnce() &&
-			(c.Namespaces == nil || c.filteredByNamespace)
-		if syncedAtLeastOnce {
-			if err := c.Plugin.SetSyncedAtLeastOnce(); err != nil {
-				utilruntime.HandleError(err)
-			} else {
-				return
-			}
+		if c.handleFirstSync() {
+			return
 		}
-		time.Sleep(1)
+		time.Sleep(50 * time.Millisecond)
 	}
 }
 

pkg/router/controller/controller_test.go

@@ -12,6 +12,7 @@ import (
 
 type fakeRouterPlugin struct {
 	lastSyncProcessed bool
+	syncedAtLeastOnce bool
 }
 
 func (p *fakeRouterPlugin) HandleRoute(t watch.EventType, route *routeapi.Route) error {
@@ -32,7 +33,8 @@ func (p *fakeRouterPlugin) SetLastSyncProcessed(processed bool) error {
 	return nil
 }
 
-func (p *fakeRouterPlugin) SetSyncedAtLeastOnce() {
+func (p *fakeRouterPlugin) SetSyncedAtLeastOnce() error {
+	p.syncedAtLeastOnce = true
 	return nil
 }
 
@@ -94,3 +96,5 @@ func TestRouterController_updateLastSyncProcessed(t *testing.T) {
 	}
 
 }
+
+// TODO add test for handleFirstSync

pkg/router/controller/factory/factory.go

@@ -99,6 +99,12 @@ func (factory *RouterControllerFactory) Create(plugin router.Plugin, watchNodes
 			}
 			return eventType, obj.(*kapi.Node), nil
 		},
+		EndpointsListCount: func() int {
+			return endpointsEventQueue.ListCount()
+		},
+		RoutesListCount: func() int {
+			return routeEventQueue.ListCount()
+		},
 		EndpointsListSuccessfulAtLeastOnce: func() bool {
 			return endpointsEventQueue.ListSuccessfulAtLeastOnce()
 		},

pkg/router/template/plugin.go

@@ -246,6 +246,7 @@ func (p *TemplatePlugin) SetLastSyncProcessed(processed bool) error {
 
 func (p *TemplatePlugin) SetSyncedAtLeastOnce() error {
 	p.Router.SetSyncedAtLeastOnce()
+	p.Router.Commit()
 	return nil
 }
 

test/integration/router_test.go

@@ -1599,7 +1599,18 @@ func waitForRouterToBecomeAvailable(host string, port int) error {
 
 // waitForRouterToBecomeReady checks that the router is ready.
 func waitForRouterToBecomeReady(host string, port int) error {
-	return waitForRouter(host, port, "ready")
+	// Need to poll because before the router becomes ready the
+	// readiness uri will return 401 instead of 503.
+	return wait.Poll(time.Millisecond*100, 30*time.Second, func() (bool, error) {
+		err := waitForRouter(host, port, "ready")
+		if err == nil {
+			return true, nil
+		} else if err == ErrUnauthenticated {
+			return false, nil
+		} else {
+			return false, err
+		}
+	})
 }
 
 func makeRouterRequest(t *testing.T, scheme string) (*http.Response, error) {
@@ -1648,7 +1659,7 @@ func TestRouterBindsPortsAfterSync(t *testing.T) {
 		t.Fatalf("Unexpected error while waiting for the router to become available: %v", err)
 	}
 
-	err = waitForRouterToBecomeReady(routerIP, statsPort)
+	err = waitForRouter(routerIP, statsPort, "ready")
 	if err == nil {
 		t.Fatalf("Router is unexpectedly ready")
 	} else if err != ErrUnauthenticated {
@@ -1674,41 +1685,8 @@ func TestRouterBindsPortsAfterSync(t *testing.T) {
 		t.Fatalf("Unable to start http server: %v", err)
 	}
 
-	// Create events that will allow the router to have enough state to consider itself 'ready'
-	httpEndpoint, err := getEndpoint(fakeMasterAndPod.PodHttpAddr)
-	if err != nil {
-		t.Fatalf("Couldn't get http endpoint: %v", err)
-	}
-	endpointEvent := &watch.Event{
-		Type: watch.Added,
-		Object: &kapi.Endpoints{
-			ObjectMeta: kapi.ObjectMeta{
-				Name:      "myService",
-				Namespace: "default",
-			},
-			Subsets: []kapi.EndpointSubset{httpEndpoint},
-		},
-	}
-	routeEvent := &watch.Event{
-		Type: watch.Added,
-		Object: &routeapi.Route{
-			ObjectMeta: kapi.ObjectMeta{
-				Name:      "path",
-				Namespace: "default",
-			},
-			Spec: routeapi.RouteSpec{
-				Host: "www.example.com",
-				Path: "/test",
-				To: routeapi.RouteTargetReference{
-					Name: "myService",
-				},
-				TLS: &routeapi.TLSConfig{},
-			},
-		},
-	}
-	sendTimeout(t, fakeMasterAndPod.EndpointChannel, eventString(endpointEvent), 30*time.Second)
-	sendTimeout(t, fakeMasterAndPod.RouteChannel, eventString(routeEvent), 30*time.Second)
-
+	// The router should now be able to sync with the api and become
+	// ready even without any routes or endopints.
 	if err = waitForRouterToBecomeReady(routerIP, statsPort); err != nil {
 		t.Fatalf("Unexpected error while waiting for router to become ready: %v", err)
 	}