routing: Improve first route load signaling

Dataclients may load routes with arbitrary delays. When multiple
dataclients are used, it's possible that one dataclient loads routes
multiple times while another doesn't load any routes at all. This
could incorrectly signal that the first route has been loaded, even
though the routing table only contains routes from a single
dataclient.

This change fixes route update logic to signal first load only after
routes from all configured dataclients are received at least once.

An alternative to zalando#3447

Signed-off-by: Aleksandr Ponimaskin <[email protected]>

Author: ponimas

Diff for: routing/datasource.go

@@ -65,7 +65,8 @@ func (d *incomingData) log(l logging.Logger, suppress bool) {
 // communication error occurs, it re-requests the whole valid set, and continues polling.
 // Currently, the routes with the same id coming from different sources are merged in an
 // undeterministic way, but this may change in the future.
-func receiveFromClient(c DataClient, o Options, out chan<- *incomingData, quit <-chan struct{}) {
+func receiveFromClient(c DataClient, o Options, out chan<- *incomingData, quit <-chan struct{}, firstLoad *sync.WaitGroup) {
+	var once sync.Once
 	initial := true
 	var ticker *time.Ticker
 	if o.PollTimeout != 0 {
@@ -96,6 +97,7 @@ func receiveFromClient(c DataClient, o Options, out chan<- *incomingData, quit <
 			continue
 		case initial || len(routes) > 0 || len(deletedIDs) > 0:
 			var incoming *incomingData
+			once.Do(firstLoad.Done)
 			if initial {
 				incoming = &incomingData{incomingReset, c, routes, nil}
 			} else {
@@ -162,13 +164,21 @@ func mergeDefs(defsByClient map[DataClient]routeDefs) []*eskip.Route {
 //
 // The active set of routes from last successful update are used until the
 // next successful update.
-func receiveRouteDefs(o Options, quit <-chan struct{}) <-chan []*eskip.Route {
+func receiveRouteDefs(o Options, quit <-chan struct{}, firstLoad chan<- struct{}) <-chan []*eskip.Route {
 	in := make(chan *incomingData)
 	out := make(chan []*eskip.Route)
 	defsByClient := make(map[DataClient]routeDefs)
 
+	wg := &sync.WaitGroup{}
+	wg.Add(len(o.DataClients))
+
 	for _, c := range o.DataClients {
-		go receiveFromClient(c, o, in, quit)
+		go receiveFromClient(c, o, in, quit, wg)
+	}
+
+	if o.SignalFirstLoad {
+		wg.Wait()
+		close(firstLoad)
 	}
 
 	go func() {
@@ -543,8 +553,8 @@ func (rt *routeTable) close() {
 
 // receives the next version of the routing table on the output channel,
 // when an update is received on one of the data clients.
-func receiveRouteMatcher(o Options, out chan<- *routeTable, quit <-chan struct{}) {
-	updates := receiveRouteDefs(o, quit)
+func receiveRouteMatcher(o Options, out chan<- *routeTable, quit <-chan struct{}, firstLoad chan<- struct{}) {
+	updates := receiveRouteDefs(o, quit, firstLoad)
 	var (
 		rt           *routeTable
 		outRelay     chan<- *routeTable

Diff for: routing/routing.go

@@ -244,12 +244,11 @@ type PreProcessor interface {
 // Routing ('router') instance providing live
 // updatable request matching.
 type Routing struct {
-	routeTable        atomic.Value // of struct routeTable
-	log               logging.Logger
-	firstLoad         chan struct{}
-	firstLoadSignaled bool
-	quit              chan struct{}
-	metrics           metrics.Metrics
+	routeTable atomic.Value // of struct routeTable
+	log        logging.Logger
+	firstLoad  chan struct{}
+	quit       chan struct{}
+	metrics    metrics.Metrics
 }
 
 // New initializes a routing instance, and starts listening for route
@@ -263,7 +262,6 @@ func New(o Options) *Routing {
 	r.metrics = o.Metrics
 	if !o.SignalFirstLoad {
 		close(r.firstLoad)
-		r.firstLoadSignaled = true
 	}
 
 	initialMatcher, _ := newMatcher(nil, MatchingOptionsNone)
@@ -338,21 +336,13 @@ func (r *Routing) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 }
 
 func (r *Routing) startReceivingUpdates(o Options) {
-	dc := len(o.DataClients)
 	c := make(chan *routeTable)
-	go receiveRouteMatcher(o, c, r.quit)
+	go receiveRouteMatcher(o, c, r.quit, r.firstLoad)
 	go func() {
 		for {
 			select {
 			case rt := <-c:
 				r.routeTable.Store(rt)
-				if !r.firstLoadSignaled {
-					dc--
-					if dc == 0 {
-						close(r.firstLoad)
-						r.firstLoadSignaled = true
-					}
-				}
 				r.log.Infof("route settings applied, id: %d", rt.id)
 				if r.metrics != nil { // existing codebases might not supply metrics instance
 					r.metrics.UpdateGauge("routes.total", float64(len(rt.validRoutes)))