Add router option to bind ports only when ready

By default the router binds ports 80 and 443 even if no routing
configuration is available.  This may be desirable when haproxy is
serving traffic directly to clients.  However, the f5 loadbalancer will
treat bound ports as indicating that a backend is ready to receive
traffic.  This means that router pods that are starting or restarting
may be put into rotation before they are ready to serve the current
route state and clients may see 503s for perfectly healthy backend
endpoints.  To avoid this possibility, this change adds an
option (BIND_ONLY_WHEN_READY/--bind-only-when-ready) to the router that
ensures that ports are bound only when a router instance has route and
endpoint state available.

Reference: bz1383663

Author: marun

docs/man/man1/openshift-infra-router.1

@@ -45,6 +45,10 @@ You may restrict the set of routes exposed to a single project (with \-\-namespa
 \fB\-\-as\fP=""
     Username to impersonate for the operation
 
+.PP
+\fB\-\-bind\-ports\-after\-sync\fP=false
+    Bind ports only after route state has been synchronized
+
 .PP
 \fB\-\-certificate\-authority\fP=""
     Path to a cert. file for the certificate authority

images/router/haproxy/conf/haproxy-config.template

@@ -94,6 +94,7 @@ listen stats :1936
     stats auth {{.StatsUser}}:{{.StatsPassword}}
 {{ end }}
 
+{{ if .BindPorts }}
 frontend public
   bind :{{env "ROUTER_SERVICE_HTTP_PORT" "80"}}
   mode http
@@ -523,6 +524,9 @@ backend be_secure_{{$cfgIdx}}
     {{ end }}{{/* end range over serviceUnitNames */}}
   {{ end }}{{/* end tls==reencrypt */}}
 {{ end }}{{/* end loop over routes */}}
+{{ else }}
+# Avoiding binding ports until routing configuration has been synchronized.
+{{ end }}{{/* end bind ports after sync */}}
 {{ end }}{{/* end haproxy config template */}}
 
 {{/*--------------------------------- END OF HAPROXY CONFIG, BELOW ARE MAPPING FILES ------------------------*/}}

pkg/client/cache/eventqueue.go

@@ -49,6 +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.
+	replaceCalled bool
+	// Tracks the number of items queued by the last Replace() call.
+	replaceCount int
 }
 
 // EventQueue implements kcache.Store
@@ -322,6 +326,9 @@ func (eq *EventQueue) Replace(objects []interface{}, resourceVersion string) err
 	eq.lock.Lock()
 	defer eq.lock.Unlock()
 
+	eq.replaceCalled = true
+	eq.replaceCount = len(objects)
+
 	eq.events = map[string]watch.EventType{}
 	eq.queue = eq.queue[:0]
 
@@ -346,6 +353,23 @@ func (eq *EventQueue) Replace(objects []interface{}, resourceVersion string) err
 	return nil
 }
 
+// ListSuccessfulAtLeastOnce indicates whether a List operation was
+// successfully completed regardless of whether any items were queued.
+func (eq *EventQueue) ListSuccessfulAtLeastOnce() bool {
+	eq.lock.Lock()
+	defer eq.lock.Unlock()
+
+	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/cmd/infra/router/template.go

@@ -62,6 +62,7 @@ type TemplateRouter struct {
 	DefaultCertificateDir  string
 	ExtendedValidation     bool
 	RouterService          *ktypes.NamespacedName
+	BindPortsAfterSync     bool
 }
 
 // reloadInterval returns how often to run the router reloads. The interval
@@ -86,6 +87,7 @@ func (o *TemplateRouter) Bind(flag *pflag.FlagSet) {
 	flag.StringVar(&o.ReloadScript, "reload", util.Env("RELOAD_SCRIPT", ""), "The path to the reload script to use")
 	flag.DurationVar(&o.ReloadInterval, "interval", reloadInterval(), "Controls how often router reloads are invoked. Mutiple router reload requests are coalesced for the duration of this interval since the last reload time.")
 	flag.BoolVar(&o.ExtendedValidation, "extended-validation", util.Env("EXTENDED_VALIDATION", "true") == "true", "If set, then an additional extended validation step is performed on all routes admitted in by this router. Defaults to true and enables the extended validation checks.")
+	flag.BoolVar(&o.BindPortsAfterSync, "bind-ports-after-sync", util.Env("ROUTER_BIND_PORTS_AFTER_SYNC", "") == "true", "Bind ports only after route state has been synchronized")
 }
 
 type RouterStats struct {
@@ -188,6 +190,7 @@ func (o *TemplateRouterOptions) Run() error {
 		StatsUsername:          o.StatsUsername,
 		StatsPassword:          o.StatsPassword,
 		PeerService:            o.RouterService,
+		BindPortsAfterSync:     o.BindPortsAfterSync,
 		IncludeUDP:             o.RouterSelection.IncludeUDP,
 		AllowWildcardRoutes:    o.RouterSelection.AllowWildcardRoutes,
 	}

pkg/router/controller/controller.go

@@ -37,6 +37,11 @@ type RouterController struct {
 	endpointsListConsumed bool
 	filteredByNamespace   bool
 
+	RoutesListSuccessfulAtLeastOnce    func() bool
+	EndpointsListSuccessfulAtLeastOnce func() bool
+	RoutesListCount                    func() int
+	EndpointsListCount                 func() int
+
 	WatchNodes bool
 
 	Namespaces            NamespaceLister
@@ -57,6 +62,51 @@ func (c *RouterController) Run() {
 	if c.WatchNodes {
 		go utilwait.Forever(c.HandleNode, 0)
 	}
+	go c.watchForFirstSync()
+}
+
+// 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.RoutesListSuccessfulAtLeastOnce() &&
+		c.EndpointsListSuccessfulAtLeastOnce() &&
+		(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' 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.
+	if c.RoutesListCount() == 0 {
+		c.routesListConsumed = true
+	}
+	if c.EndpointsListCount() == 0 {
+		c.endpointsListConsumed = 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 {
+		if c.handleFirstSync() {
+			return
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
 }
 
 func (c *RouterController) HandleNamespaces() {

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 {
@@ -26,11 +27,17 @@ func (p *fakeRouterPlugin) HandleEndpoints(watch.EventType, *kapi.Endpoints) err
 func (p *fakeRouterPlugin) HandleNamespaces(namespaces sets.String) error {
 	return nil
 }
+
 func (p *fakeRouterPlugin) SetLastSyncProcessed(processed bool) error {
 	p.lastSyncProcessed = processed
 	return nil
 }
 
+func (p *fakeRouterPlugin) SetSyncedAtLeastOnce() error {
+	p.syncedAtLeastOnce = true
+	return nil
+}
+
 type fakeNamespaceLister struct {
 }
 

pkg/router/controller/extended_validator.go

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

pkg/router/controller/factory/factory.go

@@ -99,6 +99,18 @@ 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()
+		},
+		RoutesListSuccessfulAtLeastOnce: func() bool {
+			return routeEventQueue.ListSuccessfulAtLeastOnce()
+		},
 		EndpointsListConsumed: func() bool {
 			return endpointsEventQueue.ListConsumed()
 		},

pkg/router/controller/host_admitter.go

@@ -152,6 +152,10 @@ func (p *HostAdmitter) SetLastSyncProcessed(processed bool) error {
 	return p.plugin.SetLastSyncProcessed(processed)
 }
 
+func (p *HostAdmitter) SetSyncedAtLeastOnce() error {
+	return p.plugin.SetSyncedAtLeastOnce()
+}
+
 // addRoute admits routes based on subdomain ownership - returns errors if the route is not admitted.
 func (p *HostAdmitter) addRoute(route *routeapi.Route) error {
 	// Find displaced routes (or error if an existing route displaces us)

pkg/router/controller/status.go

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

pkg/router/controller/status_test.go

@@ -43,6 +43,10 @@ func (p *fakePlugin) SetLastSyncProcessed(processed bool) error {
 	return fmt.Errorf("not expected")
 }
 
+func (p *fakePlugin) SetSyncedAtLeastOnce() error {
+	return fmt.Errorf("not expected")
+}
+
 func TestStatusNoOp(t *testing.T) {
 	now := nowFn()
 	touched := unversioned.Time{Time: now.Add(-time.Minute)}

pkg/router/controller/unique_host.go

@@ -259,6 +259,10 @@ func (p *UniqueHost) SetLastSyncProcessed(processed bool) error {
 	return p.plugin.SetLastSyncProcessed(processed)
 }
 
+func (p *UniqueHost) SetSyncedAtLeastOnce() error {
+	return p.plugin.SetSyncedAtLeastOnce()
+}
+
 // routeKeys returns the internal router key to use for the given Route.
 func routeKeys(route *routeapi.Route) []string {
 	keys := make([]string, 1+len(route.Spec.AlternateBackends))

pkg/router/f5/plugin.go

@@ -618,3 +618,8 @@ func (p *F5Plugin) HandleRoute(eventType watch.EventType,
 func (p *F5Plugin) SetLastSyncProcessed(processed bool) error {
 	return nil
 }
+
+// No-op since f5 has its own concept of what 'ready' means
+func (p *F5Plugin) SetSyncedAtLeastOnce() error {
+	return nil
+}

pkg/router/interfaces.go

@@ -17,4 +17,5 @@ type Plugin interface {
 	HandleNamespaces(namespaces sets.String) error
 	HandleNode(watch.EventType, *kapi.Node) error
 	SetLastSyncProcessed(processed bool) error
+	SetSyncedAtLeastOnce() error
 }

pkg/router/template/plugin.go

@@ -50,6 +50,7 @@ type TemplatePluginConfig struct {
 	IncludeUDP             bool
 	AllowWildcardRoutes    bool
 	PeerService            *ktypes.NamespacedName
+	BindPortsAfterSync     bool
 }
 
 // routerInterface controls the interaction of the plugin with the underlying router implementation
@@ -89,6 +90,9 @@ type routerInterface interface {
 
 	// 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()
 }
 
 func env(name, defaultValue string) string {
@@ -143,6 +147,7 @@ func NewTemplatePlugin(cfg TemplatePluginConfig, lookupSvc ServiceLookup) (*Temp
 		statsPort:              cfg.StatsPort,
 		allowWildcardRoutes:    cfg.AllowWildcardRoutes,
 		peerEndpointsKey:       peerKey,
+		bindPortsAfterSync:     cfg.BindPortsAfterSync,
 	}
 	router, err := newTemplateRouter(templateRouterCfg)
 	return newDefaultTemplatePlugin(router, cfg.IncludeUDP, lookupSvc), err
@@ -239,6 +244,12 @@ func (p *TemplatePlugin) SetLastSyncProcessed(processed bool) error {
 	return nil
 }
 
+func (p *TemplatePlugin) SetSyncedAtLeastOnce() error {
+	p.Router.SetSyncedAtLeastOnce()
+	p.Router.Commit()
+	return nil
+}
+
 // routeKeys returns the internal router keys to use for the given Route.
 // A route can have several services that it can point to, now
 func routeKeys(route *routeapi.Route) ([]string, []int32) {

pkg/router/template/plugin_test.go

@@ -243,6 +243,9 @@ func (r *TestRouter) Commit() {
 func (r *TestRouter) SetSkipCommit(skipCommit bool) {
 }
 
+func (r *TestRouter) SetSyncedAtLeastOnce() {
+}
+
 func (r *TestRouter) HasServiceUnit(key string) bool {
 	return false
 }

pkg/router/template/router.go

@@ -86,6 +86,10 @@ type templateRouter struct {
 	lock sync.Mutex
 	// the router should only reload when the value is false
 	skipCommit bool
+	// If true, haproxy should only bind ports when it has route and endpoint state
+	bindPortsAfterSync bool
+	// whether the router state has been read from the api at least once
+	syncedAtLeastOnce bool
 }
 
 // templateRouterCfg holds all configuration items required to initialize the template router
@@ -103,6 +107,7 @@ type templateRouterCfg struct {
 	allowWildcardRoutes    bool
 	peerEndpointsKey       string
 	includeUDP             bool
+	bindPortsAfterSync     bool
 }
 
 // templateConfig is a subset of the templateRouter information that should be passed to the template for generating
@@ -124,6 +129,8 @@ type templateData struct {
 	StatsPassword string
 	//port to expose stats with (if the template supports it)
 	StatsPort int
+	// whether the router should bind the default ports
+	BindPorts bool
 }
 
 func newTemplateRouter(cfg templateRouterCfg) (*templateRouter, error) {
@@ -162,6 +169,7 @@ func newTemplateRouter(cfg templateRouterCfg) (*templateRouter, error) {
 		allowWildcardRoutes:    cfg.allowWildcardRoutes,
 		peerEndpointsKey:       cfg.peerEndpointsKey,
 		peerEndpoints:          []Endpoint{},
+		bindPortsAfterSync:     cfg.bindPortsAfterSync,
 
 		rateLimitedCommitFunction:    nil,
 		rateLimitedCommitStopChannel: make(chan struct{}),
@@ -394,6 +402,7 @@ func (r *templateRouter) writeConfig() error {
 			StatsUser:          r.statsUser,
 			StatsPassword:      r.statsPassword,
 			StatsPort:          r.statsPort,
+			BindPorts:          !r.bindPortsAfterSync || r.syncedAtLeastOnce,
 		}
 		if err := template.Execute(file, data); err != nil {
 			file.Close()
@@ -729,6 +738,13 @@ func (r *templateRouter) SetSkipCommit(skipCommit bool) {
 	}
 }
 
+// SetSyncedAtLeastOnce indicates to the router that state has been
+// read from the api.
+func (r *templateRouter) SetSyncedAtLeastOnce() {
+	r.syncedAtLeastOnce = true
+	glog.V(4).Infof("Router state synchronized for the first time")
+}
+
 // HasServiceUnit attempts to retrieve a service unit for the given
 // key, returning a boolean indication of whether the key is known.
 func (r *templateRouter) HasServiceUnit(key string) bool {

test/integration/router_stress_test.go

@@ -257,6 +257,10 @@ func (p *DelayPlugin) SetLastSyncProcessed(processed bool) error {
 	return p.plugin.SetLastSyncProcessed(processed)
 }
 
+func (p *DelayPlugin) SetSyncedAtLeastOnce() error {
+	return p.plugin.SetSyncedAtLeastOnce()
+}
+
 // launchRouter launches a template router that communicates with the
 // api via the provided clients.
 func launchRouter(oc osclient.Interface, kc kclient.Interface, maxDelay int32, name string, reloadInterval int, reloadCounts map[string]int) (templatePlugin *templateplugin.TemplatePlugin) {