Implement Hash-Based Routing

docs/03-how-to-add-new-route-option.md

@@ -22,6 +22,11 @@ applications:
   - route: example2.com
     options:
       loadbalancing: least-connection
+  - route: example3.com
+    options:
+      loadbalancing: hash
+      hash_header: tenant-id
+      hash_balance: 1.25
 ```
 
 **NOTE**: In the implementation, the `options` property of a route represents per-route features.

src/code.cloudfoundry.org/gorouter/handlers/helpers.go

@@ -63,13 +63,13 @@ func upgradeHeader(request *http.Request) string {
 	return ""
 }
 
-func EndpointIteratorForRequest(logger *slog.Logger, request *http.Request, stickySessionCookieNames config.StringSet, authNegotiateSticky bool, azPreference string, az string) (route.EndpointIterator, error) {
+func EndpointIteratorForRequest(logger *slog.Logger, request *http.Request, stickySessionCookieNames config.StringSet, authNegotiateSticky bool, azPreference string, az string, globalLB string) (route.EndpointIterator, error) {
 	reqInfo, err := ContextRequestInfo(request)
 	if err != nil {
 		return nil, fmt.Errorf("could not find reqInfo in context")
 	}
 	stickyEndpointID, mustBeSticky := GetStickySession(request, stickySessionCookieNames, authNegotiateSticky)
-	return reqInfo.RoutePool.Endpoints(logger, stickyEndpointID, mustBeSticky, azPreference, az), nil
+	return reqInfo.RoutePool.Endpoints(logger, stickyEndpointID, mustBeSticky, azPreference, az, globalLB, request), nil
 }
 
 func GetStickySession(request *http.Request, stickySessionCookieNames config.StringSet, authNegotiateSticky bool) (string, bool) {

src/code.cloudfoundry.org/gorouter/handlers/max_request_size.go

@@ -69,7 +69,7 @@ func (m *MaxRequestSize) ServeHTTP(rw http.ResponseWriter, r *http.Request, next
 		if err != nil {
 			logger.Error("request-info-err", log.ErrAttr(err))
 		} else {
-			endpointIterator, err := EndpointIteratorForRequest(logger, r, m.cfg.StickySessionCookieNames, m.cfg.StickySessionsForAuthNegotiate, m.cfg.LoadBalanceAZPreference, m.cfg.Zone)
+			endpointIterator, err := EndpointIteratorForRequest(logger, r, m.cfg.StickySessionCookieNames, m.cfg.StickySessionsForAuthNegotiate, m.cfg.LoadBalanceAZPreference, m.cfg.Zone, m.cfg.LoadBalance)
 			if err != nil {
 				logger.Error("failed-to-find-endpoint-for-req-during-431-short-circuit", log.ErrAttr(err))
 			} else {

src/code.cloudfoundry.org/gorouter/mbus/subscriber.go

@@ -43,7 +43,7 @@ type RegistryMessage struct {
 type RegistryMessageOpts struct {
 	LoadBalancingAlgorithm string  `json:"loadbalancing"`
 	HashHeaderName         string  `json:"hash_header"`
-	HashBalance            float64 `json:"hash_balance"`
+	HashBalance            float64 `json:"hash_balance,string"`
 }
 
 func (rm *RegistryMessage) makeEndpoint(http2Enabled bool) (*route.Endpoint, error) {

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

@@ -126,7 +126,7 @@ func (rt *roundTripper) RoundTrip(originalRequest *http.Request) (*http.Response
 
 	stickyEndpointID, mustBeSticky := handlers.GetStickySession(request, rt.config.StickySessionCookieNames, rt.config.StickySessionsForAuthNegotiate)
 	numberOfEndpoints := reqInfo.RoutePool.NumEndpoints()
-	iter := reqInfo.RoutePool.Endpoints(rt.logger, stickyEndpointID, mustBeSticky, rt.config.LoadBalanceAZPreference, rt.config.Zone)
+	iter := reqInfo.RoutePool.Endpoints(rt.logger, stickyEndpointID, mustBeSticky, rt.config.LoadBalanceAZPreference, rt.config.Zone, rt.config.LoadBalance, request)
 
 	// The selectEndpointErr needs to be tracked separately. If we get an error
 	// while selecting an endpoint we might just have run out of routes. In

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

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math/rand"
 	"net"
 	"net/http"
 	"net/http/httptest"
@@ -274,7 +275,7 @@ var _ = Describe("ProxyRoundTripper", func() {
 					res, err := proxyRoundTripper.RoundTrip(req)
 					Expect(err).NotTo(HaveOccurred())
 
-					iter := routePool.Endpoints(logger.Logger, "", false, AZPreference, AZ)
+					iter := routePool.Endpoints(logger.Logger, "", false, AZPreference, AZ, cfg.LoadBalance, req)
 					ep1 := iter.Next(0)
 					ep2 := iter.Next(1)
 					Expect(ep1.PrivateInstanceId).To(Equal(ep2.PrivateInstanceId))
@@ -608,7 +609,7 @@ var _ = Describe("ProxyRoundTripper", func() {
 					_, err := proxyRoundTripper.RoundTrip(req)
 					Expect(err).To(MatchError(ContainSubstring("tls: handshake failure")))
 
-					iter := routePool.Endpoints(logger.Logger, "", false, AZPreference, AZ)
+					iter := routePool.Endpoints(logger.Logger, "", false, AZPreference, AZ, cfg.LoadBalance, req)
 					ep1 := iter.Next(0)
 					ep2 := iter.Next(1)
 					Expect(ep1).To(Equal(ep2))
@@ -1700,6 +1701,167 @@ var _ = Describe("ProxyRoundTripper", func() {
 				})
 			})
 
+			Context("when load-balancing strategy is set to hash-based routing", func() {
+				JustBeforeEach(func() {
+					for i := 1; i <= 3; i++ {
+						endpoint = route.NewEndpoint(&route.EndpointOpts{
+							AppId:                  fmt.Sprintf("appID%d", i),
+							Host:                   fmt.Sprintf("%d.%d.%d.%d", i, i, i, i),
+							Port:                   9090,
+							PrivateInstanceId:      fmt.Sprintf("instanceID%d", i),
+							PrivateInstanceIndex:   fmt.Sprintf("%d", i),
+							AvailabilityZone:       AZ,
+							LoadBalancingAlgorithm: config.LOAD_BALANCE_HB,
+							HashHeaderName:         "X-Hash",
+						})
+
+						_ = routePool.Put(endpoint)
+						Expect(routePool.HashLookupTable).ToNot(BeNil())
+
+					}
+				})
+
+				It("routes requests with same hash header value to the same endpoint", func() {
+					req.Header.Set("X-Hash", "value")
+					reqInfo, err := handlers.ContextRequestInfo(req)
+					Expect(err).ToNot(HaveOccurred())
+					reqInfo.RoutePool = routePool
+
+					var selectedEndpoints []*route.Endpoint
+
+					// Make multiple requests with the same hash value
+					for i := 0; i < 5; i++ {
+						_, err = proxyRoundTripper.RoundTrip(req)
+						Expect(err).NotTo(HaveOccurred())
+						selectedEndpoints = append(selectedEndpoints, reqInfo.RouteEndpoint)
+					}
+
+					// All requests should go to the same endpoint
+					firstEndpoint := selectedEndpoints[0]
+					for _, ep := range selectedEndpoints[1:] {
+						Expect(ep.PrivateInstanceId).To(Equal(firstEndpoint.PrivateInstanceId))
+					}
+				})
+
+				It("routes requests with different hash header values to potentially different endpoints", func() {
+					reqInfo, err := handlers.ContextRequestInfo(req)
+					Expect(err).ToNot(HaveOccurred())
+					reqInfo.RoutePool = routePool
+
+					endpointDistribution := make(map[string]int)
+
+					// Make requests with different hash values
+					for i := 0; i < 10; i++ {
+						req.Header.Set("X-Hash", fmt.Sprintf("value-%d", i))
+						_, err = proxyRoundTripper.RoundTrip(req)
+						Expect(err).NotTo(HaveOccurred())
+						endpointDistribution[reqInfo.RouteEndpoint.PrivateInstanceId]++
+					}
+
+					// Should distribute across multiple endpoints (not all to one)
+					Expect(len(endpointDistribution)).To(BeNumerically(">", 1))
+				})
+
+				It("falls back to default load balancing algorithm when hash header is missing", func() {
+					reqInfo, err := handlers.ContextRequestInfo(req)
+					Expect(err).ToNot(HaveOccurred())
+
+					reqInfo.RoutePool = routePool
+
+					_, err = proxyRoundTripper.RoundTrip(req)
+					Expect(err).NotTo(HaveOccurred())
+
+					infoLogs := logger.Lines(zap.InfoLevel)
+					count := 0
+					for i := 0; i < len(infoLogs); i++ {
+						if strings.Contains(infoLogs[i], "hash-based-routing-header-not-found") {
+							count++
+						}
+					}
+					Expect(count).To(Equal(1))
+					// Verify it still selects an endpoint
+					Expect(reqInfo.RouteEndpoint).ToNot(BeNil())
+				})
+
+				Context("when sticky session cookies (JSESSIONID and VCAP_ID) are on the request", func() {
+					var (
+						sessionCookie *http.Cookie
+						cookies       []*http.Cookie
+					)
+
+					JustBeforeEach(func() {
+						sessionCookie = &http.Cookie{
+							Name: StickyCookieKey, //JSESSIONID
+						}
+						transport.RoundTripStub = func(req *http.Request) (*http.Response, error) {
+							resp := &http.Response{StatusCode: http.StatusTeapot, Header: make(map[string][]string)}
+							//Attach the same JSESSIONID on to the response if it exists on the request
+
+							if len(req.Cookies()) > 0 {
+								for _, cookie := range req.Cookies() {
+									if cookie.Name == StickyCookieKey {
+										resp.Header.Add(round_tripper.CookieHeader, cookie.String())
+										return resp, nil
+									}
+								}
+							}
+
+							sessionCookie.Value, _ = uuid.GenerateUUID()
+							resp.Header.Add(round_tripper.CookieHeader, sessionCookie.String())
+							return resp, nil
+						}
+						resp, err := proxyRoundTripper.RoundTrip(req)
+						Expect(err).ToNot(HaveOccurred())
+
+						cookies = resp.Cookies()
+						Expect(cookies).To(HaveLen(2))
+
+					})
+
+					Context("when there is a JSESSIONID and __VCAP_ID__ set on the request", func() {
+						It("will always route to the instance specified with the __VCAP_ID__ cookie", func() {
+
+							// Generate 20 random values for the hash header, so chance that all go to instanceID1
+							// by accident is 0.33^20
+							for i := 0; i < 20; i++ {
+								randomStr := make([]byte, 8)
+								for j := range randomStr {
+									randomStr[j] = byte('a' + rand.Intn(26))
+								}
+
+								req.Header.Set("X-Hash", string(randomStr))
+								reqInfo, err := handlers.ContextRequestInfo(req)
+								req.AddCookie(&http.Cookie{Name: round_tripper.VcapCookieId, Value: "instanceID1"})
+								req.AddCookie(&http.Cookie{Name: StickyCookieKey, Value: "abc"})
+
+								Expect(err).ToNot(HaveOccurred())
+								reqInfo.RoutePool = routePool
+
+								resp, err := proxyRoundTripper.RoundTrip(req)
+								Expect(err).ToNot(HaveOccurred())
+
+								new_cookies := resp.Cookies()
+								Expect(new_cookies).To(HaveLen(2))
+
+								for _, cookie := range new_cookies {
+									Expect(cookie.Name).To(SatisfyAny(
+										Equal(StickyCookieKey),
+										Equal(round_tripper.VcapCookieId),
+									))
+									if cookie.Name == StickyCookieKey {
+										Expect(cookie.Value).To(Equal("abc"))
+									} else {
+										Expect(cookie.Value).To(Equal("instanceID1"))
+									}
+								}
+
+							}
+
+						})
+					})
+				})
+			})
+
 			Context("when endpoint timeout is not 0", func() {
 				var reqCh chan *http.Request
 				BeforeEach(func() {