Use endpointregistry in the hotpath (#2823)

Signed-off-by: Roman Zavodskikh <[email protected]>
Co-authored-by: Roman Zavodskikh <[email protected]>

VERSION

@@ -1 +1 @@
-v0.18
+v0.19

docs/kubernetes/ingress-controller.md

@@ -753,7 +753,7 @@ the communication work with TCP and UDP to the specified `swarm-port`:
 ## Upgrades
 
 Please always read the announcements of the vX.Y.**0**
-[release page](https://github.com/zalando/skipper/releases/tag/v0.18.0),
+[release page](https://github.com/zalando/skipper/releases/tag/v0.19.0),
 because these will document in case we break something in a backwards non
 compatible way. Most of the time it will be safe to deploy minor
 version updates, but better to know in advance if something could

filters/fadein/fadein.go

@@ -202,7 +202,6 @@ func (p *postProcessor) Do(r []*routing.Route) []*routing.Route {
 				detected = now
 			}
 
-			ep.Detected = detected
 			if p.endpointRegisty != nil {
 				metrics := p.endpointRegisty.GetMetrics(ep.Host)
 				if endpointsCreated[key].After(metrics.DetectedTime()) {

filters/fadein/fadein_test.go

@@ -249,9 +249,6 @@ func TestPostProcessor(t *testing.T) {
 		}
 
 		for _, ep := range bar.LBEndpoints {
-			if ep.Detected.IsZero() {
-				t.Fatal("failed to set detection time")
-			}
 			if endpointRegistry.GetMetrics(ep.Host).DetectedTime().IsZero() {
 				t.Fatal("failed to set detection time")
 			}
@@ -284,7 +281,7 @@ func TestPostProcessor(t *testing.T) {
 		endpointRegisty := routing.NewEndpointRegistry(routing.RegistryOptions{})
 		rt, _ := createRouting(t, routes, endpointRegisty)
 		r := route(rt, "/")
-		if r == nil || len(r.LBEndpoints) == 0 || !r.LBEndpoints[0].Detected.IsZero() {
+		if r == nil || len(r.LBEndpoints) == 0 {
 			t.Fatal("failed to ignore negative duration")
 		}
 		if endpointRegisty.GetMetrics(r.LBEndpoints[0].Host).DetectedTime().IsZero() {
@@ -311,9 +308,6 @@ func TestPostProcessor(t *testing.T) {
 		var found bool
 		for _, ep := range r.LBEndpoints {
 			if ep.Host == "10.0.0.1:8080" {
-				if ep.Detected.After(firstDetected) {
-					t.Fatal("Failed to keep detection time.")
-				}
 				if endpointRegistry.GetMetrics(ep.Host).DetectedTime().After(firstDetected) {
 					t.Fatal("Failed to keep detection time.")
 				}
@@ -348,9 +342,6 @@ func TestPostProcessor(t *testing.T) {
 		var found bool
 		for _, ep := range r.LBEndpoints {
 			if ep.Host == "10.0.0.1:8080" {
-				if ep.Detected.After(firstDetected) {
-					t.Fatal("Failed to keep detection time.")
-				}
 				if endpointRegistry.GetMetrics(ep.Host).DetectedTime().After(firstDetected) {
 					t.Fatal("Failed to keep detection time.")
 				}
@@ -387,9 +378,6 @@ func TestPostProcessor(t *testing.T) {
 		var found bool
 		for _, ep := range r.LBEndpoints {
 			if ep.Host == "10.0.0.1:8080" {
-				if !ep.Detected.After(firstDetected) {
-					t.Fatal("Failed to clear detection time.")
-				}
 				if !endpointRegistry.GetMetrics(ep.Host).DetectedTime().After(firstDetected) {
 					t.Fatal("Failed to clear detection time.")
 				}
@@ -428,9 +416,6 @@ func TestPostProcessor(t *testing.T) {
 		var found bool
 		for _, ep := range r.LBEndpoints {
 			if ep.Host == "10.0.0.1:8080" {
-				if !ep.Detected.After(firstDetected) {
-					t.Fatal("Failed to reset detection time.")
-				}
 				if !endpointRegistry.GetMetrics(ep.Host).DetectedTime().After(firstDetected) {
 					t.Fatal("Failed to reset detection time.")
 				}

loadbalancer/algorithm.go

@@ -72,15 +72,15 @@ func shiftWeighted(rnd *rand.Rand, ctx *routing.LBContext, now time.Time) routin
 	rt := ctx.Route
 	ep := ctx.LBEndpoints
 	for _, epi := range ep {
-		wi := fadeIn(now, rt.LBFadeInDuration, rt.LBFadeInExponent, epi.Detected)
+		wi := fadeIn(now, rt.LBFadeInDuration, rt.LBFadeInExponent, epi.Metrics.DetectedTime())
 		sum += wi
 	}
 
 	choice := ep[len(ep)-1]
 	r := rnd.Float64() * sum
 	var upto float64
 	for i, epi := range ep {
-		upto += fadeIn(now, rt.LBFadeInDuration, rt.LBFadeInExponent, epi.Detected)
+		upto += fadeIn(now, rt.LBFadeInDuration, rt.LBFadeInExponent, epi.Metrics.DetectedTime())
 		if upto > r {
 			choice = ep[i]
 			break
@@ -113,15 +113,15 @@ func withFadeIn(rnd *rand.Rand, ctx *routing.LBContext, choice int, algo routing
 		now,
 		ctx.Route.LBFadeInDuration,
 		ctx.Route.LBFadeInExponent,
-		ctx.LBEndpoints[choice].Detected,
+		ctx.LBEndpoints[choice].Metrics.DetectedTime(),
 	)
 
 	if rnd.Float64() < f {
 		return ep[choice]
 	}
 	notFadingIndexes := make([]int, 0, len(ep))
 	for i := 0; i < len(ep); i++ {
-		if _, fadingIn := fadeInState(now, ctx.Route.LBFadeInDuration, ep[i].Detected); !fadingIn {
+		if _, fadingIn := fadeInState(now, ctx.Route.LBFadeInDuration, ep[i].Metrics.DetectedTime()); !fadingIn {
 			notFadingIndexes = append(notFadingIndexes, i)
 		}
 	}
@@ -260,7 +260,7 @@ func computeLoadAverage(ctx *routing.LBContext) float64 {
 	sum := 1.0 // add 1 to include the request that just arrived
 	endpoints := ctx.LBEndpoints
 	for _, v := range endpoints {
-		sum += float64(v.Metrics.GetInflightRequests())
+		sum += float64(v.Metrics.InflightRequests())
 	}
 	return sum / float64(len(endpoints))
 }
@@ -277,10 +277,10 @@ func (ch *consistentHash) boundedLoadSearch(key string, balanceFactor float64, c
 		if skipEndpoint(endpointIndex) {
 			continue
 		}
-		load := ctx.LBEndpoints[endpointIndex].Metrics.GetInflightRequests()
+		load := ctx.LBEndpoints[endpointIndex].Metrics.InflightRequests()
 		// We know there must be an endpoint whose load <= average load.
 		// Since targetLoad >= average load (balancerFactor >= 1), there must also be an endpoint with load <= targetLoad.
-		if load <= int(targetLoad) {
+		if float64(load) <= targetLoad {
 			break
 		}
 		ringIndex = (ringIndex + 1) % ch.Len()
@@ -372,7 +372,7 @@ func (p *powerOfRandomNChoices) Apply(ctx *routing.LBContext) routing.LBEndpoint
 // getScore returns negative value of inflightrequests count.
 func (p *powerOfRandomNChoices) getScore(e routing.LBEndpoint) int64 {
 	// endpoints with higher inflight request should have lower score
-	return -int64(e.Metrics.GetInflightRequests())
+	return -int64(e.Metrics.InflightRequests())
 }
 
 type (
@@ -431,9 +431,8 @@ func parseEndpoints(r *routing.Route) error {
 		}
 
 		r.LBEndpoints[i] = routing.LBEndpoint{
-			Scheme:  scheme,
-			Host:    host,
-			Metrics: &routing.LBMetrics{},
+			Scheme: scheme,
+			Host:   host,
 		}
 	}
 

loadbalancer/algorithm_test.go

@@ -30,6 +30,7 @@ func TestSelectAlgorithm(t *testing.T) {
 
 	t.Run("LB route with default algorithm", func(t *testing.T) {
 		p := NewAlgorithmProvider()
+		endpointRegistry := routing.NewEndpointRegistry(routing.RegistryOptions{})
 		r := &routing.Route{
 			Route: eskip.Route{
 				BackendType: eskip.LBBackend,
@@ -38,6 +39,7 @@ func TestSelectAlgorithm(t *testing.T) {
 		}
 
 		rr := p.Do([]*routing.Route{r})
+		endpointRegistry.Do([]*routing.Route{r})
 		if len(rr) != 1 {
 			t.Fatal("failed to process LB route")
 		}
@@ -56,6 +58,7 @@ func TestSelectAlgorithm(t *testing.T) {
 
 	t.Run("LB route with explicit round-robin algorithm", func(t *testing.T) {
 		p := NewAlgorithmProvider()
+		endpointRegistry := routing.NewEndpointRegistry(routing.RegistryOptions{})
 		r := &routing.Route{
 			Route: eskip.Route{
 				BackendType: eskip.LBBackend,
@@ -65,6 +68,7 @@ func TestSelectAlgorithm(t *testing.T) {
 		}
 
 		rr := p.Do([]*routing.Route{r})
+		endpointRegistry.Do([]*routing.Route{r})
 		if len(rr) != 1 {
 			t.Fatal("failed to process LB route")
 		}
@@ -83,6 +87,7 @@ func TestSelectAlgorithm(t *testing.T) {
 
 	t.Run("LB route with explicit consistentHash algorithm", func(t *testing.T) {
 		p := NewAlgorithmProvider()
+		endpointRegistry := routing.NewEndpointRegistry(routing.RegistryOptions{})
 		r := &routing.Route{
 			Route: eskip.Route{
 				BackendType: eskip.LBBackend,
@@ -92,6 +97,7 @@ func TestSelectAlgorithm(t *testing.T) {
 		}
 
 		rr := p.Do([]*routing.Route{r})
+		endpointRegistry.Do([]*routing.Route{r})
 		if len(rr) != 1 {
 			t.Fatal("failed to process LB route")
 		}
@@ -110,6 +116,7 @@ func TestSelectAlgorithm(t *testing.T) {
 
 	t.Run("LB route with explicit random algorithm", func(t *testing.T) {
 		p := NewAlgorithmProvider()
+		endpointRegistry := routing.NewEndpointRegistry(routing.RegistryOptions{})
 		r := &routing.Route{
 			Route: eskip.Route{
 				BackendType: eskip.LBBackend,
@@ -119,6 +126,7 @@ func TestSelectAlgorithm(t *testing.T) {
 		}
 
 		rr := p.Do([]*routing.Route{r})
+		endpointRegistry.Do([]*routing.Route{r})
 		if len(rr) != 1 {
 			t.Fatal("failed to process LB route")
 		}
@@ -137,6 +145,7 @@ func TestSelectAlgorithm(t *testing.T) {
 
 	t.Run("LB route with explicit powerOfRandomNChoices algorithm", func(t *testing.T) {
 		p := NewAlgorithmProvider()
+		endpointRegistry := routing.NewEndpointRegistry(routing.RegistryOptions{})
 		r := &routing.Route{
 			Route: eskip.Route{
 				BackendType: eskip.LBBackend,
@@ -146,6 +155,7 @@ func TestSelectAlgorithm(t *testing.T) {
 		}
 
 		rr := p.Do([]*routing.Route{r})
+		endpointRegistry.Do([]*routing.Route{r})
 		if len(rr) != 1 {
 			t.Fatal("failed to process LB route")
 		}
@@ -264,6 +274,7 @@ func TestApply(t *testing.T) {
 				LBEndpoints: rt[0].LBEndpoints,
 				Registry:    routing.NewEndpointRegistry(routing.RegistryOptions{}),
 			}
+			lbctx.Registry.Do([]*routing.Route{r})
 
 			h := make(map[string]int)
 			for i := 0; i < R; i++ {
@@ -316,6 +327,7 @@ func TestConsistentHashBoundedLoadSearch(t *testing.T) {
 			LBEndpoints: endpoints,
 		},
 	}})[0]
+
 	ch := route.LBAlgorithm.(*consistentHash)
 	ctx := &routing.LBContext{
 		Request:     r,
@@ -324,6 +336,7 @@ func TestConsistentHashBoundedLoadSearch(t *testing.T) {
 		Params:      map[string]interface{}{ConsistentHashBalanceFactor: 1.25},
 		Registry:    routing.NewEndpointRegistry(routing.RegistryOptions{}),
 	}
+	ctx.Registry.Do([]*routing.Route{route})
 	noLoad := ch.Apply(ctx)
 	nonBounded := ch.Apply(&routing.LBContext{Request: r, Route: route, LBEndpoints: route.LBEndpoints, Params: map[string]interface{}{}})
 
@@ -393,6 +406,7 @@ func TestConsistentHashBoundedLoadDistribution(t *testing.T) {
 			LBEndpoints: endpoints,
 		},
 	}})[0]
+
 	ch := route.LBAlgorithm.(*consistentHash)
 	balanceFactor := 1.25
 	ctx := &routing.LBContext{
@@ -402,23 +416,23 @@ func TestConsistentHashBoundedLoadDistribution(t *testing.T) {
 		Params:      map[string]interface{}{ConsistentHashBalanceFactor: balanceFactor},
 		Registry:    routing.NewEndpointRegistry(routing.RegistryOptions{}),
 	}
+	ctx.Registry.Do([]*routing.Route{route})
 
 	for i := 0; i < 100; i++ {
 		ep := ch.Apply(ctx)
-		ifr0 := route.LBEndpoints[0].Metrics.GetInflightRequests()
-		ifr1 := route.LBEndpoints[1].Metrics.GetInflightRequests()
-		ifr2 := route.LBEndpoints[2].Metrics.GetInflightRequests()
+		ifr0 := route.LBEndpoints[0].Metrics.InflightRequests()
+		ifr1 := route.LBEndpoints[1].Metrics.InflightRequests()
+		ifr2 := route.LBEndpoints[2].Metrics.InflightRequests()
 
 		assert.Equal(t, int64(ifr0), ctx.Registry.GetMetrics(route.LBEndpoints[0].Host).InflightRequests())
 		assert.Equal(t, int64(ifr1), ctx.Registry.GetMetrics(route.LBEndpoints[1].Host).InflightRequests())
 		assert.Equal(t, int64(ifr2), ctx.Registry.GetMetrics(route.LBEndpoints[2].Host).InflightRequests())
 
 		avg := float64(ifr0+ifr1+ifr2) / 3.0
-		limit := int(avg*balanceFactor) + 1
+		limit := int64(avg*balanceFactor) + 1
 		if ifr0 > limit || ifr1 > limit || ifr2 > limit {
 			t.Errorf("Expected in-flight requests for each endpoint to be less than %d. In-flight request counts: %d, %d, %d", limit, ifr0, ifr1, ifr2)
 		}
-		ep.Metrics.IncInflightRequest()
 		ctx.Registry.GetMetrics(ep.Host).IncInflightRequest()
 	}
 }

loadbalancer/fadein_test.go

@@ -67,8 +67,8 @@ func initializeEndpoints(endpointAges []time.Duration, fadeInDuration time.Durat
 
 	for i := range eps {
 		ctx.Route.LBEndpoints = append(ctx.Route.LBEndpoints, routing.LBEndpoint{
-			Host:     eps[i],
-			Detected: detectionTimes[i],
+			Host:    eps[i],
+			Metrics: ctx.Registry.GetMetrics(eps[i]),
 		})
 		ctx.Registry.GetMetrics(eps[i]).SetDetected(detectionTimes[i])
 	}
@@ -329,8 +329,8 @@ func benchmarkFadeIn(
 		registry := routing.NewEndpointRegistry(routing.RegistryOptions{})
 		for i := range eps {
 			route.LBEndpoints = append(route.LBEndpoints, routing.LBEndpoint{
-				Host:     eps[i],
-				Detected: detectionTimes[i],
+				Host:    eps[i],
+				Metrics: registry.GetMetrics(eps[i]),
 			})
 			registry.GetMetrics(eps[i]).SetDetected(detectionTimes[i])
 		}

proxy/proxy.go

@@ -484,8 +484,8 @@ func selectEndpoint(ctx *context, registry *routing.EndpointRegistry) *routing.L
 
 // creates an outgoing http request to be forwarded to the route endpoint
 // based on the augmented incoming request
-func mapRequest(ctx *context, requestContext stdlibcontext.Context, removeHopHeaders bool, registry *routing.EndpointRegistry) (*http.Request, *routing.LBEndpoint, error) {
-	var endpoint *routing.LBEndpoint
+func mapRequest(ctx *context, requestContext stdlibcontext.Context, removeHopHeaders bool, registry *routing.EndpointRegistry) (*http.Request, routing.Metrics, error) {
+	var endpointMetrics routing.Metrics
 	r := ctx.request
 	rt := ctx.route
 	host := ctx.outgoingHost
@@ -497,9 +497,13 @@ func mapRequest(ctx *context, requestContext stdlibcontext.Context, removeHopHea
 		setRequestURLFromRequest(u, r)
 		setRequestURLForDynamicBackend(u, stateBag)
 	case eskip.LBBackend:
-		endpoint = selectEndpoint(ctx, registry)
+		endpoint := selectEndpoint(ctx, registry)
+		endpointMetrics = endpoint.Metrics
 		u.Scheme = endpoint.Scheme
 		u.Host = endpoint.Host
+	case eskip.NetworkBackend:
+		endpointMetrics = registry.GetMetrics(rt.Host)
+		fallthrough
 	default:
 		u.Scheme = rt.Scheme
 		u.Host = rt.Host
@@ -512,7 +516,7 @@ func mapRequest(ctx *context, requestContext stdlibcontext.Context, removeHopHea
 
 	rr, err := http.NewRequestWithContext(requestContext, r.Method, u.String(), body)
 	if err != nil {
-		return nil, endpoint, err
+		return nil, nil, err
 	}
 
 	rr.ContentLength = r.ContentLength
@@ -543,7 +547,7 @@ func mapRequest(ctx *context, requestContext stdlibcontext.Context, removeHopHea
 		rr = forwardToProxy(r, rr)
 	}
 
-	return rr, endpoint, nil
+	return rr, endpointMetrics, nil
 }
 
 type proxyUrlContextKey struct{}
@@ -836,7 +840,7 @@ func (p *Proxy) makeUpgradeRequest(ctx *context, req *http.Request) {
 }
 
 func (p *Proxy) makeBackendRequest(ctx *context, requestContext stdlibcontext.Context) (*http.Response, *proxyError) {
-	req, endpoint, err := mapRequest(ctx, requestContext, p.flags.HopHeadersRemoval(), p.registry)
+	req, endpointMetrics, err := mapRequest(ctx, requestContext, p.flags.HopHeadersRemoval(), p.registry)
 	if err != nil {
 		return nil, &proxyError{err: fmt.Errorf("could not map backend request: %w", err)}
 	}
@@ -845,9 +849,9 @@ func (p *Proxy) makeBackendRequest(ctx *context, requestContext stdlibcontext.Co
 		return res, nil
 	}
 
-	if endpoint != nil {
-		endpoint.Metrics.IncInflightRequest()
-		defer endpoint.Metrics.DecInflightRequest()
+	if endpointMetrics != nil {
+		endpointMetrics.IncInflightRequest()
+		defer endpointMetrics.DecInflightRequest()
 	}
 
 	if p.experimentalUpgrade && isUpgradeRequest(req) {