refactor: register logic for route registry

Resolves: #468

Author: maxmoehl

src/code.cloudfoundry.org/gorouter/proxy/round_tripper/proxy_round_tripper_test.go

@@ -418,11 +418,11 @@ var _ = Describe("ProxyRoundTripper", func() {
 								if transport.RoundTripCallCount() == 1 {
 									endpoint := endpointFor(4)
 									updated := routePool.Put(endpoint)
-									Expect(updated).To(Equal(route.UPDATED))
+									Expect(updated).To(Equal(route.REFRESHED))
 
 									endpoint = endpointFor(5)
 									updated = routePool.Put(endpoint)
-									Expect(updated).To(Equal(route.UPDATED))
+									Expect(updated).To(Equal(route.REFRESHED))
 								}
 
 								return nil, &net.OpError{Op: "dial", Err: errors.New("connection refused")}

src/code.cloudfoundry.org/gorouter/registry/registry.go

@@ -88,29 +88,32 @@ func (r *RouteRegistry) Register(uri route.Uri, endpoint *route.Endpoint) {
 		return
 	}
 
-	endpointAdded := r.register(uri, endpoint)
+	poolPutResult := r.register(uri, endpoint)
 
-	r.reporter.CaptureRegistryMessage(endpoint, endpointAdded.String())
+	r.reporter.CaptureRegistryMessage(endpoint, poolPutResult.String())
 
-	if endpointAdded == route.ADDED && !endpoint.UpdatedAt.IsZero() {
+	if poolPutResult == route.ADDED && !endpoint.UpdatedAt.IsZero() {
 		r.reporter.CaptureRouteRegistrationLatency(time.Since(endpoint.UpdatedAt))
 	}
 
-	switch endpointAdded {
+	switch poolPutResult {
 	case route.ADDED:
 		if r.logger.Enabled(context.Background(), slog.LevelInfo) {
 			r.logger.Info("endpoint-registered", buildSlogAttrs(uri, endpoint)...)
 		}
 	case route.UPDATED:
-		if r.logger.Enabled(context.Background(), slog.LevelDebug) {
-			r.logger.Debug("endpoint-registered", buildSlogAttrs(uri, endpoint)...)
+		if r.logger.Enabled(context.Background(), slog.LevelInfo) {
+			r.logger.Info("endpoint-registered", buildSlogAttrs(uri, endpoint)...)
 		}
-	default:
+	case route.UNMODIFIED:
 		if r.logger.Enabled(context.Background(), slog.LevelDebug) {
 			r.logger.Debug("endpoint-not-registered", buildSlogAttrs(uri, endpoint)...)
 		}
+	case route.REFRESHED:
+		if r.logger.Enabled(context.Background(), slog.LevelDebug) {
+			r.logger.Debug("endpoint-refreshed", buildSlogAttrs(uri, endpoint)...)
+		}
 	}
-
 }
 
 func (r *RouteRegistry) register(uri route.Uri, endpoint *route.Endpoint) route.PoolPutResult {
@@ -132,11 +135,11 @@ func (r *RouteRegistry) register(uri route.Uri, endpoint *route.Endpoint) route.
 		endpoint.StaleThreshold = r.dropletStaleThreshold
 	}
 
-	endpointAdded := pool.Put(endpoint)
+	poolPutResult := pool.Put(endpoint)
 	// Overwrites the load balancing algorithm of a pool by that of a specified endpoint, if that is valid.
 	r.SetTimeOfLastUpdate(t)
 
-	return endpointAdded
+	return poolPutResult
 }
 
 // insertRouteKey acquires a write lock, inserts the route key into the registry and releases the write lock.

src/code.cloudfoundry.org/gorouter/route/pool.go

@@ -31,6 +31,8 @@ func (p PoolPutResult) String() string {
 		return "updated"
 	case ADDED:
 		return "added"
+	case REFRESHED:
+		return "refreshed"
 	default:
 		panic("invalid PoolPutResult")
 	}
@@ -40,6 +42,7 @@ const (
 	UNMODIFIED = PoolPutResult(iota)
 	UPDATED
 	ADDED
+	REFRESHED
 )
 
 func NewCounter(initial int64) *Counter {
@@ -277,35 +280,58 @@ func (p *EndpointPool) Put(endpoint *Endpoint) PoolPutResult {
 	p.Lock()
 	defer p.Unlock()
 
-	var result PoolPutResult
+	var equal bool
 	e, found := p.index[endpoint.CanonicalAddr()]
 	if found {
-		result = UPDATED
-		if !e.endpoint.Equal(endpoint) {
-			e.Lock()
-			defer e.Unlock()
+		// Only calculate equal once, it's expensive.
+		equal = e.endpoint.Equal(endpoint)
+	}
 
-			if !e.endpoint.ModificationTag.SucceededBy(&endpoint.ModificationTag) {
-				return UNMODIFIED
-			}
+	switch {
+	case found && equal:
+		// This is the most common case. The endpoint has not changed but was simply re-announced
+		// to ensure gorouter is still aware of it.
+		e.updated = time.Now()
+		p.Update()
 
-			oldEndpoint := e.endpoint
-			e.endpoint = endpoint
+		return REFRESHED
 
-			if oldEndpoint.PrivateInstanceId != endpoint.PrivateInstanceId {
-				delete(p.index, oldEndpoint.PrivateInstanceId)
-				p.index[endpoint.PrivateInstanceId] = e
-			}
+	case found && !e.endpoint.ModificationTag.SucceededBy(&endpoint.ModificationTag):
+		// This exists to protect against flapping when a route receives a change (e.g. a new
+		// route-service URL) and messages for the old and new config are still floating around.
+		return UNMODIFIED
 
-			if oldEndpoint.ServerCertDomainSAN == endpoint.ServerCertDomainSAN {
-				endpoint.SetRoundTripper(oldEndpoint.RoundTripper())
-			}
+	case found && !equal:
+		// The same endpoint was announced with different data, replace the old endpoint with the
+		// new one.
+		e.Lock()
+		defer e.Unlock()
+
+		oldEndpoint := e.endpoint
+		e.endpoint = endpoint
+
+		if oldEndpoint.PrivateInstanceId != endpoint.PrivateInstanceId {
+			delete(p.index, oldEndpoint.PrivateInstanceId)
+			p.index[endpoint.PrivateInstanceId] = e
+		}
+
+		if oldEndpoint.ServerCertDomainSAN == endpoint.ServerCertDomainSAN {
+			endpoint.SetRoundTripper(oldEndpoint.RoundTripper())
 		}
-	} else {
-		result = ADDED
+
+		p.RouteSvcUrl = e.endpoint.RouteServiceUrl
+		p.setPoolLoadBalancingAlgorithm(e.endpoint)
+		e.updated = time.Now()
+		p.Update()
+
+		return UPDATED
+
+	case !found:
+		// New endpoint.
 		e = &endpointElem{
 			endpoint:           endpoint,
 			index:              len(p.endpoints),
+			updated:            time.Now(),
 			maxConnsPerBackend: p.maxConnsPerBackend,
 		}
 
@@ -314,14 +340,15 @@ func (p *EndpointPool) Put(endpoint *Endpoint) PoolPutResult {
 		p.index[endpoint.CanonicalAddr()] = e
 		p.index[endpoint.PrivateInstanceId] = e
 
-	}
-	p.RouteSvcUrl = e.endpoint.RouteServiceUrl
-	p.setPoolLoadBalancingAlgorithm(e.endpoint)
-	e.updated = time.Now()
-	// set the update time of the pool
-	p.Update()
+		p.RouteSvcUrl = e.endpoint.RouteServiceUrl
+		p.setPoolLoadBalancingAlgorithm(e.endpoint)
+		p.Update()
+
+		return ADDED
 
-	return result
+	default:
+		panic("quantum state discovered")
+	}
 }
 
 func (p *EndpointPool) RouteServiceUrl() string {

src/code.cloudfoundry.org/gorouter/route/pool_test.go

@@ -152,7 +152,7 @@ var _ = Describe("EndpointPool", func() {
 			pool.MarkUpdated(time.Now().Add(-(10 * time.Minute)))
 
 			b := pool.Put(endpoint)
-			Expect(b).To(Equal(route.UPDATED))
+			Expect(b).To(Equal(route.REFRESHED))
 
 			prunedEndpoints := pool.PruneEndpoints()
 			Expect(prunedEndpoints).To(BeEmpty())
@@ -163,7 +163,7 @@ var _ = Describe("EndpointPool", func() {
 			endpoint2 := route.NewEndpoint(&route.EndpointOpts{Host: "1.2.3.4", Port: 5678})
 
 			pool.Put(endpoint1)
-			Expect(pool.Put(endpoint2)).To(Equal(route.UPDATED))
+			Expect(pool.Put(endpoint2)).To(Equal(route.REFRESHED))
 		})
 
 		Context("with modification tags", func() {
@@ -189,7 +189,7 @@ var _ = Describe("EndpointPool", func() {
 				BeforeEach(func() {
 					modTag2.Increment()
 					endpoint := route.NewEndpoint(&route.EndpointOpts{Host: "1.2.3.4", Port: 5678, ModificationTag: modTag2})
-					pool.Put(endpoint)
+					Expect(pool.Put(endpoint)).To(Equal(route.UPDATED))
 				})
 
 				It("doesnt update an endpoint", func() {
@@ -387,7 +387,9 @@ var _ = Describe("EndpointPool", func() {
 		Context("when any endpoint updates its route_service_url", func() {
 			It("returns the route_service_url most recently updated in the pool", func() {
 				endpointRS1 := route.NewEndpoint(&route.EndpointOpts{Host: "host-1", Port: 1234, RouteServiceUrl: "first-url"})
+				endpointRS1Updated := route.NewEndpoint(&route.EndpointOpts{Host: "host-1", Port: 1234, RouteServiceUrl: "third-url"})
 				endpointRS2 := route.NewEndpoint(&route.EndpointOpts{Host: "host-2", Port: 2234, RouteServiceUrl: "second-url"})
+				endpointRS2Updated := route.NewEndpoint(&route.EndpointOpts{Host: "host-2", Port: 2234, RouteServiceUrl: "fourth-url"})
 				b := pool.Put(endpointRS1)
 				Expect(b).To(Equal(route.ADDED))
 
@@ -399,14 +401,12 @@ var _ = Describe("EndpointPool", func() {
 				url = pool.RouteServiceUrl()
 				Expect(url).To(Equal("second-url"))
 
-				endpointRS1.RouteServiceUrl = "third-url"
-				b = pool.Put(endpointRS1)
+				b = pool.Put(endpointRS1Updated)
 				Expect(b).To(Equal(route.UPDATED))
 				url = pool.RouteServiceUrl()
 				Expect(url).To(Equal("third-url"))
 
-				endpointRS2.RouteServiceUrl = "fourth-url"
-				b = pool.Put(endpointRS2)
+				b = pool.Put(endpointRS2Updated)
 				Expect(b).To(Equal(route.UPDATED))
 				url = pool.RouteServiceUrl()
 				Expect(url).To(Equal("fourth-url"))