Propagate options-only route changes immediately in route-emitter

src/code.cloudfoundry.org/route-emitter/routingtable/endpoint.go

@@ -100,6 +100,10 @@ func (info ExternalEndpointInfo) Hash() interface{} {
 	return info
 }
 
+func (info ExternalEndpointInfo) HashWithOptions() interface{} {
+	return info.Hash()
+}
+
 func (info ExternalEndpointInfo) MessageFor(e Endpoint, directInstanceRoute, _ bool) (*RegistryMessage, *tcpmodels.TcpRouteMapping, *RegistryMessage) {
 	tlsHostPort := -1
 	tlsContainerPort := -1
@@ -168,6 +172,13 @@ type routeHash struct {
 	Protocol         string
 }
 
+// routeHashWithOptions extends routeHash to include options, normalized so
+// semantically identical option sets always compare equal.
+type routeHashWithOptions struct {
+	routeHash
+	Options string
+}
+
 // route hash is used to find route differences
 // it needs to be dereferenced so that it can be used as a key in a hash map
 func (r Route) Hash() interface{} {
@@ -180,6 +191,29 @@ func (r Route) Hash() interface{} {
 	}
 }
 
+// HashWithOptions includes normalized options so diffRoutes detects options-only changes.
+func (r Route) HashWithOptions() interface{} {
+	opts := string(r.Options)
+	if len(r.Options) > 0 {
+		var v interface{}
+		if err := json.Unmarshal(r.Options, &v); err == nil {
+			if b, err := json.Marshal(v); err == nil {
+				opts = string(b)
+			}
+		}
+	}
+	return routeHashWithOptions{
+		routeHash: routeHash{
+			Hostname:         r.Hostname,
+			RouteServiceUrl:  r.RouteServiceUrl,
+			IsolationSegment: r.IsolationSegment,
+			LogGUID:          r.LogGUID,
+			Protocol:         r.Protocol,
+		},
+		Options: opts,
+	}
+}
+
 func (r Route) MessageFor(endpoint Endpoint, directInstanceAddress, emitEndpointUpdatedAt bool) (*RegistryMessage, *tcpmodels.TcpRouteMapping, *RegistryMessage) {
 	generator := RegistryMessageFor
 	if endpoint.IsDirectInstanceRoute(directInstanceAddress) {
@@ -199,6 +233,10 @@ func (r InternalRoute) Hash() interface{} {
 	return r
 }
 
+func (r InternalRoute) HashWithOptions() interface{} {
+	return r.Hash()
+}
+
 func (r InternalRoute) MessageFor(endpoint Endpoint, _, emitEndpointUpdatedAt bool) (*RegistryMessage, *tcpmodels.TcpRouteMapping, *RegistryMessage) {
 	generator := InternalEndpointRegistryMessageFor
 	msg := generator(endpoint, r, emitEndpointUpdatedAt)

src/code.cloudfoundry.org/route-emitter/routingtable/endpoint_utils_test.go

@@ -1,6 +1,8 @@
 package routingtable_test
 
 import (
+	"encoding/json"
+
 	"code.cloudfoundry.org/bbs/models"
 	"code.cloudfoundry.org/route-emitter/routingtable"
 	"code.cloudfoundry.org/routing-info/tcp_routes"
@@ -50,6 +52,49 @@ var _ = Describe("LRP Utils", func() {
 		})
 	})
 
+	Describe("HashWithOptions", func() {
+		It("treats semantically identical JSON options as equal regardless of key order", func() {
+			routeAlpha := routingtable.Route{
+				Hostname: "foo.example.com",
+				Options:  json.RawMessage(`{"b":2,"a":1}`),
+			}
+			routeBeta := routingtable.Route{
+				Hostname: "foo.example.com",
+				Options:  json.RawMessage(`{"a":1,"b":2}`),
+			}
+			Expect(routeAlpha.HashWithOptions()).To(Equal(routeBeta.HashWithOptions()))
+		})
+
+		It("distinguishes routes with different option values", func() {
+			routeAlpha := routingtable.Route{
+				Hostname: "foo.example.com",
+				Options:  json.RawMessage(`{"loadbalancing":"hash"}`),
+			}
+			routeBeta := routingtable.Route{
+				Hostname: "foo.example.com",
+				Options:  json.RawMessage(`{"loadbalancing":"round-robin"}`),
+			}
+			Expect(routeAlpha.HashWithOptions()).NotTo(Equal(routeBeta.HashWithOptions()))
+		})
+
+		It("distinguishes a route with options from one without", func() {
+			withOptions := routingtable.Route{
+				Hostname: "foo.example.com",
+				Options:  json.RawMessage(`{"loadbalancing":"hash"}`),
+			}
+			withoutOptions := routingtable.Route{
+				Hostname: "foo.example.com",
+			}
+			Expect(withOptions.HashWithOptions()).NotTo(Equal(withoutOptions.HashWithOptions()))
+		})
+
+		It("treats nil and empty options as equal", func() {
+			withNil := routingtable.Route{Hostname: "foo.example.com", Options: nil}
+			withEmpty := routingtable.Route{Hostname: "foo.example.com", Options: json.RawMessage{}}
+			Expect(withNil.HashWithOptions()).To(Equal(withEmpty.HashWithOptions()))
+		})
+	})
+
 	Describe("NewEndpointsFromActual", func() {
 		Context("when actual is not evacuating", func() {
 			It("builds a map of container port to endpoint", func() {

src/code.cloudfoundry.org/route-emitter/routingtable/nats_routing_table_test.go

@@ -1,6 +1,7 @@
 package routingtable_test
 
 import (
+	"encoding/json"
 	"fmt"
 
 	mfakes "code.cloudfoundry.org/diego-logging-client/testhelpers"
@@ -650,6 +651,128 @@ var _ = Describe("RoutingTable", func() {
 			})
 		})
 
+		Context("when there is an existing routing key with options", func() {
+			var desiredLRP *models.DesiredLRP
+
+			createDesiredLRPWithOptions := func(options json.RawMessage) *models.DesiredLRP {
+				routingInfo := cfroutes.CFRoutes{
+					{
+						Hostnames: []string{hostname1, hostname2},
+						Port:      key.ContainerPort,
+						Options:   options,
+					},
+				}.RoutingInfo()
+				routes := models.Routes{}
+				for k, v := range routingInfo {
+					routes[k] = v
+				}
+				return createDesiredLRPWithRoutes(key.ProcessGUID, 3, routes, logGuid, *currentTag, runInfo)
+			}
+
+			BeforeEach(func() {
+				tempTable := routingtable.NewRoutingTable(false, false, fakeMetronClient)
+				desiredLRP = createDesiredLRPWithOptions(json.RawMessage(`{"loadbalancing":"hash"}`))
+				tempTable.SetRoutes(logger, nil, desiredLRP)
+				lrp := createActualLRP(key, endpoint1, domain)
+				tempTable.AddEndpoint(logger, lrp)
+				table.Swap(logger, tempTable, domains)
+			})
+
+			Context("when only options change in an event", func() {
+				BeforeEach(func() {
+					afterDesiredLRP := createDesiredLRPWithOptions(json.RawMessage(`{"loadbalancing":"round-robin"}`))
+					afterDesiredLRP.ModificationTag.Index++
+					_, messagesToEmit = table.SetRoutes(logger, desiredLRP, afterDesiredLRP)
+				})
+
+				It("registers the updated route without unregistering the old one", func() {
+					expected := routingtable.MessagesToEmit{
+						RegistrationMessages: []routingtable.RegistryMessage{
+							routingtable.RegistryMessageFor(endpoint1, routingtable.Route{Hostname: hostname1, LogGUID: logGuid, Options: json.RawMessage(`{"loadbalancing":"round-robin"}`)}, false),
+							routingtable.RegistryMessageFor(endpoint1, routingtable.Route{Hostname: hostname2, LogGUID: logGuid, Options: json.RawMessage(`{"loadbalancing":"round-robin"}`)}, false),
+						},
+					}
+					Expect(messagesToEmit).To(MatchMessagesToEmit(expected))
+				})
+			})
+
+			Context("when only options change during sync", func() {
+				BeforeEach(func() {
+					tempTable := routingtable.NewRoutingTable(false, false, fakeMetronClient)
+					afterDesiredLRP := createDesiredLRPWithOptions(json.RawMessage(`{"loadbalancing":"round-robin"}`))
+					tempTable.SetRoutes(logger, nil, afterDesiredLRP)
+					lrp := createActualLRP(key, endpoint1, domain)
+					tempTable.AddEndpoint(logger, lrp)
+					_, messagesToEmit = table.Swap(logger, tempTable, domains)
+				})
+
+				It("registers the updated route without unregistering the old one", func() {
+					expected := routingtable.MessagesToEmit{
+						RegistrationMessages: []routingtable.RegistryMessage{
+							routingtable.RegistryMessageFor(endpoint1, routingtable.Route{Hostname: hostname1, LogGUID: logGuid, Options: json.RawMessage(`{"loadbalancing":"round-robin"}`)}, false),
+							routingtable.RegistryMessageFor(endpoint1, routingtable.Route{Hostname: hostname2, LogGUID: logGuid, Options: json.RawMessage(`{"loadbalancing":"round-robin"}`)}, false),
+						},
+					}
+					Expect(messagesToEmit).To(MatchMessagesToEmit(expected))
+				})
+			})
+
+			Context("when options change and endpoints change simultaneously during sync", func() {
+				BeforeEach(func() {
+					tempTable := routingtable.NewRoutingTable(false, false, fakeMetronClient)
+					afterDesiredLRP := createDesiredLRPWithOptions(json.RawMessage(`{"loadbalancing":"round-robin"}`))
+					tempTable.SetRoutes(logger, nil, afterDesiredLRP)
+					lrp := createActualLRP(key, endpoint2, domain)
+					tempTable.AddEndpoint(logger, lrp)
+					_, messagesToEmit = table.Swap(logger, tempTable, domains)
+				})
+
+				It("registers the new endpoint with new options and unregisters the old endpoint with old options", func() {
+					expected := routingtable.MessagesToEmit{
+						RegistrationMessages: []routingtable.RegistryMessage{
+							routingtable.RegistryMessageFor(endpoint2, routingtable.Route{Hostname: hostname1, LogGUID: logGuid, Options: json.RawMessage(`{"loadbalancing":"round-robin"}`)}, false),
+							routingtable.RegistryMessageFor(endpoint2, routingtable.Route{Hostname: hostname2, LogGUID: logGuid, Options: json.RawMessage(`{"loadbalancing":"round-robin"}`)}, false),
+						},
+						UnregistrationMessages: []routingtable.RegistryMessage{
+							routingtable.RegistryMessageFor(endpoint1, routingtable.Route{Hostname: hostname1, LogGUID: logGuid, Options: json.RawMessage(`{"loadbalancing":"hash"}`)}, false),
+							routingtable.RegistryMessageFor(endpoint1, routingtable.Route{Hostname: hostname2, LogGUID: logGuid, Options: json.RawMessage(`{"loadbalancing":"hash"}`)}, false),
+						},
+					}
+					Expect(messagesToEmit).To(MatchMessagesToEmit(expected))
+				})
+			})
+
+			Context("when options change and a hostname is removed in the same event", func() {
+				BeforeEach(func() {
+					routingInfo := cfroutes.CFRoutes{
+						{
+							Hostnames: []string{hostname1},
+							Port:      key.ContainerPort,
+							Options:   json.RawMessage(`{"loadbalancing":"round-robin"}`),
+						},
+					}.RoutingInfo()
+					routes := models.Routes{}
+					for k, v := range routingInfo {
+						routes[k] = v
+					}
+					afterDesiredLRP := createDesiredLRPWithRoutes(key.ProcessGUID, 3, routes, logGuid, *newerTag, runInfo)
+					_, messagesToEmit = table.SetRoutes(logger, desiredLRP, afterDesiredLRP)
+				})
+
+				It("registers hostname1 with new options and unregisters hostname2 without a gap", func() {
+					expected := routingtable.MessagesToEmit{
+						RegistrationMessages: []routingtable.RegistryMessage{
+							routingtable.RegistryMessageFor(endpoint1, routingtable.Route{Hostname: hostname1, LogGUID: logGuid, Options: json.RawMessage(`{"loadbalancing":"round-robin"}`)}, false),
+						},
+						UnregistrationMessages: []routingtable.RegistryMessage{
+							routingtable.RegistryMessageFor(endpoint1, routingtable.Route{Hostname: hostname2, LogGUID: logGuid, Options: json.RawMessage(`{"loadbalancing":"hash"}`)}, false),
+						},
+					}
+					Expect(messagesToEmit).To(MatchMessagesToEmit(expected))
+				})
+			})
+		})
+
 		Context("when the routing key has an evacuating and instance endpoint", func() {
 			BeforeEach(func() {
 				tempTable := routingtable.NewRoutingTable(false, false, fakeMetronClient)

src/code.cloudfoundry.org/route-emitter/routingtable/routingtable.go

@@ -426,6 +426,7 @@ func (t *internalRoutingTable) GetRoutingEvents() (TCPRouteMappings, MessagesToE
 type routeMapping interface {
 	MessageFor(endpoint Endpoint, directInstanceAddress, emitEndpointUpdatedAt bool) (*RegistryMessage, *tcpmodels.TcpRouteMapping, *RegistryMessage)
 	Hash() interface{}
+	HashWithOptions() interface{}
 }
 
 func httpRoutesFrom(lrp *models.DesiredLRP) map[RoutingKey][]routeMapping {
@@ -638,7 +639,8 @@ func diffRoutes(before, after []routeMapping) routesDiff {
 	}
 
 	for routeHash := range newRoutes {
-		if _, ok := existingRoutes[routeHash]; !ok {
+		// Register routes that are new, or whose options changed.
+		if _, ok := existingRoutes[routeHash]; !ok || existingRoutes[routeHash].HashWithOptions() != newRoutes[routeHash].HashWithOptions() {
 			diff.added = append(diff.added, newRoutes[routeHash])
 		}
 	}