Revert "Optimize: consistent hash algorithm" (#3926)

Reverts #3918

Turns out in production it was much worse in cpu usage

Signed-off-by: Sandor Szücs <sandor.szuecs@zalando.de>

Author: szuecs

loadbalancer/algorithm.go

@@ -4,7 +4,6 @@ import (
 	"errors"
 	"fmt"
 	"math/rand/v2"
-	"net/netip"
 	"sort"
 	"sync"
 	"sync/atomic"
@@ -119,7 +118,7 @@ func newConsistentHashInternal(endpoints []string, hashesPerEndpoint int) routin
 	}
 	for i, ep := range endpoints {
 		endpointStartIndex := hashesPerEndpoint * i
-		for j := range hashesPerEndpoint {
+		for j := 0; j < hashesPerEndpoint; j++ {
 			ch.hashRing[endpointStartIndex+j] = endpointHash{i, hash(fmt.Sprintf("%s-%d", ep, j))}
 		}
 	}
@@ -128,8 +127,7 @@ func newConsistentHashInternal(endpoints []string, hashesPerEndpoint int) routin
 }
 
 func newConsistentHash(endpoints []string) routing.LBAlgorithm {
-	n := 10000 / len(endpoints)
-	return newConsistentHashInternal(endpoints, n)
+	return newConsistentHashInternal(endpoints, 100)
 }
 
 func hash(s string) uint64 {
@@ -138,10 +136,6 @@ func hash(s string) uint64 {
 
 func skipEndpoint(c *routing.LBContext, index int) bool {
 	host := c.Route.LBEndpoints[index].Host
-	if len(c.LBEndpoints) > 300 { // 300 see https://github.com/zalando/skipper/pull/3918/
-		return !existsBinarySearch(host, c.LBEndpoints)
-	}
-	// linear scan for low numbers
 	for i := range c.LBEndpoints {
 		if c.LBEndpoints[i].Host == host {
 			return false
@@ -150,30 +144,6 @@ func skipEndpoint(c *routing.LBContext, index int) bool {
 	return true
 }
 
-func existsBinarySearch(target string, list []routing.LBEndpoint) bool {
-	targetAP, err := netip.ParseAddrPort(target)
-	if err != nil {
-		return false
-	}
-
-	idx := sort.Search(len(list), func(i int) bool {
-		itemAP, err := netip.ParseAddrPort(list[i].Host)
-		if err != nil {
-			return false
-		}
-
-		if itemAP.Addr() != targetAP.Addr() {
-			return !itemAP.Addr().Less(targetAP.Addr())
-		}
-		return itemAP.Port() >= targetAP.Port()
-	})
-
-	if idx < len(list) && list[idx].Host == target {
-		return true
-	}
-	return false
-}
-
 // Returns index in hash ring with the closest hash to key's hash
 func (ch *consistentHash) searchRing(key string, ctx *routing.LBContext) int {
 	h := hash(key)

loadbalancer/algorithm_test.go

@@ -4,8 +4,6 @@ import (
 	"fmt"
 	"math"
 	"net/http"
-	"net/netip"
-	"sort"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -231,11 +229,10 @@ func TestApply(t *testing.T) {
 	const R = 1000
 	const N = 10
 	eps := make([]string, 0, N)
-	for i := range N {
+	for i := 0; i < N; i++ {
 		ep := fmt.Sprintf("http://127.0.0.1:123%d/foo", i)
 		eps = append(eps, ep)
 	}
-	sortLBEndpoints(eps)
 
 	for _, tt := range []struct {
 		name          string
@@ -286,7 +283,7 @@ func TestApply(t *testing.T) {
 			}
 
 			h := make(map[string]int)
-			for range R {
+			for i := 0; i < R; i++ {
 				lbe := tt.algorithm.Apply(lbctx)
 				h[lbe.Host] += 1
 			}
@@ -319,7 +316,6 @@ func TestConsistentHashSearch(t *testing.T) {
 	}
 
 	endpoints := []string{"http://127.0.0.1:8080", "http://127.0.0.2:8080", "http://127.0.0.3:8080"}
-	sortLBEndpoints(endpoints)
 	const key = "192.168.0.1"
 
 	ep := apply(key, endpoints)
@@ -343,7 +339,6 @@ func TestConsistentHashSearch(t *testing.T) {
 
 func TestConsistentHashBoundedLoadSearch(t *testing.T) {
 	endpoints := []string{"http://127.0.0.1:8080", "http://127.0.0.2:8080", "http://127.0.0.3:8080"}
-	sortLBEndpoints(endpoints)
 	r, _ := http.NewRequest("GET", "http://127.0.0.1:1234/foo", nil)
 	route := NewAlgorithmProvider().Do([]*routing.Route{{
 		Route: eskip.Route{
@@ -364,7 +359,7 @@ func TestConsistentHashBoundedLoadSearch(t *testing.T) {
 	defer endpointRegistry.Close()
 	endpointRegistry.Do([]*routing.Route{route})
 	noLoad := ch.Apply(ctx)
-	nonBounded := ch.Apply(&routing.LBContext{Request: r, Route: route, LBEndpoints: route.LBEndpoints, Params: map[string]any{}})
+	nonBounded := ch.Apply(&routing.LBContext{Request: r, Route: route, LBEndpoints: route.LBEndpoints, Params: map[string]interface{}{}})
 
 	if noLoad != nonBounded {
 		t.Error("When no endpoints are overloaded, the chosen endpoint should be the same as standard consistentHash")
@@ -393,7 +388,6 @@ func TestConsistentHashBoundedLoadSearch(t *testing.T) {
 
 func TestConsistentHashKey(t *testing.T) {
 	endpoints := []string{"http://127.0.0.1:8080", "http://127.0.0.2:8080", "http://127.0.0.3:8080"}
-	sortLBEndpoints(endpoints)
 	ch := newConsistentHash(endpoints)
 
 	r, _ := http.NewRequest("GET", "http://127.0.0.1:1234/foo", nil)
@@ -407,16 +401,16 @@ func TestConsistentHashKey(t *testing.T) {
 		},
 	}})[0]
 
-	defaultEndpoint := ch.Apply(&routing.LBContext{Request: r, Route: rt, LBEndpoints: rt.LBEndpoints, Params: make(map[string]any)})
-	remoteHostEndpoint := ch.Apply(&routing.LBContext{Request: r, Route: rt, LBEndpoints: rt.LBEndpoints, Params: map[string]any{ConsistentHashKey: net.RemoteHost(r).String()}})
+	defaultEndpoint := ch.Apply(&routing.LBContext{Request: r, Route: rt, LBEndpoints: rt.LBEndpoints, Params: make(map[string]interface{})})
+	remoteHostEndpoint := ch.Apply(&routing.LBContext{Request: r, Route: rt, LBEndpoints: rt.LBEndpoints, Params: map[string]interface{}{ConsistentHashKey: net.RemoteHost(r).String()}})
 
 	if defaultEndpoint != remoteHostEndpoint {
 		t.Error("remote host should be used as a default key")
 	}
 
 	for i, ep := range endpoints {
 		key := fmt.Sprintf("%s-%d", ep, 1) // "ep-0" to "ep-99" is the range of keys for this endpoint. If we use this as the hash key it should select endpoint ep.
-		selected := ch.Apply(&routing.LBContext{Request: r, Route: rt, LBEndpoints: rt.LBEndpoints, Params: map[string]any{ConsistentHashKey: key}})
+		selected := ch.Apply(&routing.LBContext{Request: r, Route: rt, LBEndpoints: rt.LBEndpoints, Params: map[string]interface{}{ConsistentHashKey: key}})
 		if selected != rt.LBEndpoints[i] {
 			t.Errorf("expected: %v, got %v", rt.LBEndpoints[i], selected)
 		}
@@ -425,7 +419,6 @@ func TestConsistentHashKey(t *testing.T) {
 
 func TestConsistentHashBoundedLoadDistribution(t *testing.T) {
 	endpoints := []string{"http://127.0.0.1:8080", "http://127.0.0.2:8080", "http://127.0.0.3:8080"}
-	sortLBEndpoints(endpoints)
 	r, _ := http.NewRequest("GET", "http://127.0.0.1:1234/foo", nil)
 	route := NewAlgorithmProvider().Do([]*routing.Route{{
 		Route: eskip.Route{
@@ -441,13 +434,13 @@ func TestConsistentHashBoundedLoadDistribution(t *testing.T) {
 		Request:     r,
 		Route:       route,
 		LBEndpoints: route.LBEndpoints,
-		Params:      map[string]any{ConsistentHashBalanceFactor: balanceFactor},
+		Params:      map[string]interface{}{ConsistentHashBalanceFactor: balanceFactor},
 	}
 	endpointRegistry := routing.NewEndpointRegistry(routing.RegistryOptions{})
 	defer endpointRegistry.Close()
 	endpointRegistry.Do([]*routing.Route{route})
 
-	for range 100 {
+	for i := 0; i < 100; i++ {
 		ep := ch.Apply(ctx)
 		ifr0 := route.LBEndpoints[0].Metrics.InflightRequests()
 		ifr1 := route.LBEndpoints[1].Metrics.InflightRequests()
@@ -482,7 +475,7 @@ func TestConsistentHashKeyDistribution(t *testing.T) {
 }
 
 func addInflightRequests(registry *routing.EndpointRegistry, endpoint routing.LBEndpoint, count int) {
-	for range count {
+	for i := 0; i < count; i++ {
 		endpoint.Metrics.IncInflightRequest()
 		registry.GetMetrics(endpoint.Host).IncInflightRequest()
 	}
@@ -552,88 +545,9 @@ func BenchmarkRandomAlgorithm(b *testing.B) {
 		LBEndpoints: lbeps,
 	}
 
-	for b.Loop() {
-		alg.Apply(lbc)
-	}
-}
-
-// similar to datasource processRouteDef() use of sort.SliceStable
-func sortLBEndpoints(lbEndpoints []string) {
-	sort.SliceStable(lbEndpoints, func(i, j int) bool {
-		apI, errI := netip.ParseAddrPort(lbEndpoints[i])
-		apJ, errJ := netip.ParseAddrPort(lbEndpoints[j])
-
-		if errI != nil || errJ != nil {
-			return errI == nil
-		}
-
-		ipI := apI.Addr()
-		ipJ := apJ.Addr()
-
-		if ipI != ipJ {
-			return ipI.Less(ipJ)
-		}
-
-		return apI.Port() < apJ.Port()
-	})
-
-}
-
-func BenchmarkConsistentHash(b *testing.B) {
-	for _, N := range []int{10, 100, 1000} {
-		eps := make([]string, 0, N)
-		var j int
-		for i := range N {
-			j = i / 255
-			ep := fmt.Sprintf("http://10.0.%d.%d:8080/", j, i%255)
-			eps = append(eps, ep)
-		}
-		sortLBEndpoints(eps)
-
-		req, err := http.NewRequest("GET", "http://consistent.bench.test", nil)
-		if err != nil {
-			b.Fatalf("Failed to create request: %v", err)
-		}
+	b.ResetTimer()
 
-		route := NewAlgorithmProvider().Do([]*routing.Route{{
-			Route: eskip.Route{
-				BackendType: eskip.LBBackend,
-				LBAlgorithm: ConsistentHash.String(),
-				LBEndpoints: eps,
-			},
-		}})[0]
-
-		alg := newConsistentHash(eps)
-		lbCtx := &routing.LBContext{
-			Request:     req,
-			Route:       route,
-			LBEndpoints: route.LBEndpoints,
-			Params: map[string]any{
-				ConsistentHashKey:           "Foo",
-				ConsistentHashBalanceFactor: 0.2,
-			},
-		}
-
-		// populate metrics
-		endpointRegistry := routing.NewEndpointRegistry(routing.RegistryOptions{})
-		defer endpointRegistry.Close()
-		endpointRegistry.Do([]*routing.Route{route})
-
-		// set Foo header value
-		headerValues := [10000]string{}
-		for i := range len(headerValues) {
-			headerValues[i] = fmt.Sprintf("foo-%d", i)
-		}
-
-		b.Run(fmt.Sprintf("%d endpoints", N), func(b *testing.B) {
-			var iter int64
-			b.ResetTimer()
-			b.ReportAllocs()
-			for b.Loop() {
-				iter++
-				req.Header.Set("Foo", headerValues[iter%10000])
-				alg.Apply(lbCtx)
-			}
-		})
+	for n := 0; n < b.N; n++ {
+		alg.Apply(lbc)
 	}
 }

proxy/proxy_test.go

@@ -2607,38 +2607,3 @@ func benchmarkCopyStream(b *testing.B, size int64, cpStream func(to flushWriter,
 		}
 	}
 }
-
-func BenchmarkConsistentHashSelectEndpoint(b *testing.B) {
-	for _, N := range []int{10, 100, 250, 300, 400, 500, 1000, 5000} {
-		eps := make([]string, 0, N)
-		var j int
-		for i := range N {
-			j = i / 255
-			ep := fmt.Sprintf("http://10.0.%d.%d:8080", j, i%255)
-			eps = append(eps, ep)
-		}
-		req, err := http.NewRequest("GET", "http://consistent.bench.test", nil)
-		if err != nil {
-			b.Fatalf("Failed to create request: %v", err)
-		}
-		doc := fmt.Sprintf(`r: * -> <consistentHash, "%s">`, strings.Join(eps, `", "`))
-		tp, err := newTestProxyWithParams(doc, Params{
-			AccessLogDisabled: false,
-		})
-		if err != nil {
-			b.Error(err)
-			return
-		}
-		defer tp.close()
-		route, _ := tp.routing.Get().Do(req)
-		ctx := newContext(nil, req, tp.proxy)
-		ctx.route = route
-		b.Run(fmt.Sprintf("%d endpoints", N), func(b *testing.B) {
-			b.ResetTimer()
-			b.ReportAllocs()
-			for b.Loop() {
-				tp.proxy.selectEndpoint(ctx)
-			}
-		})
-	}
-}

routing/datasource.go

@@ -3,7 +3,6 @@ package routing
 import (
 	"errors"
 	"fmt"
-	"net/netip"
 	"sort"
 	"sync"
 	"time"
@@ -535,26 +534,6 @@ func processRouteDef(o *Options, cpm map[string]PredicateSpec, def *eskip.Route)
 	}
 
 	r := &Route{Route: *def, Scheme: scheme, Host: host, Predicates: cps, Filters: fs, weight: weight}
-	if def.LBAlgorithm == "consistentHash" {
-		sort.SliceStable(def.LBEndpoints, func(i, j int) bool {
-			apI, errI := netip.ParseAddrPort(def.LBEndpoints[i])
-			apJ, errJ := netip.ParseAddrPort(def.LBEndpoints[j])
-
-			if errI != nil || errJ != nil {
-				return errI == nil
-			}
-
-			ipI := apI.Addr()
-			ipJ := apJ.Addr()
-
-			if ipI != ipJ {
-				return ipI.Less(ipJ)
-			}
-
-			return apI.Port() < apJ.Port()
-		})
-	}
-
 	if err := processTreePredicates(r, def.Predicates); err != nil {
 		return nil, err
 	}