#624 Set flat priority only for services with traffic splitting

allegro · Mar 27, 2024 · 539c435 · 539c435
1 parent 8e902af
commit 539c435
Show file tree

Hide file tree

Showing 5 changed files with 69 additions and 22 deletions.
diff --git a/...e/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/snapshot/SnapshotProperties.kt b/...e/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/snapshot/SnapshotProperties.kt
@@ -166,6 +166,7 @@ class TrafficSplittingProperties {
     var zoneName = ""
     var headerName = ""
     var weightsByService: Map<String, ZoneWeights> = mapOf()
+    var zonesAllowingTrafficSplitting = listOf<String>()
 }
 
 class ZoneWeights {

diff --git a/...llegro/tech/servicemesh/envoycontrol/snapshot/resource/endpoints/EnvoyEndpointsFactory.kt b/...llegro/tech/servicemesh/envoycontrol/snapshot/resource/endpoints/EnvoyEndpointsFactory.kt
@@ -34,26 +34,27 @@ class EnvoyEndpointsFactory(
 ) {
     companion object {
         private val logger by logger()
+        private const val HIGHEST_PRIORITY = 0
     }
 
     fun createLoadAssignment(
         clusters: Set<String>,
         multiClusterState: MultiClusterState
     ): List<ClusterLoadAssignment> {
-
         return clusters
             .map { serviceName ->
                 val localityLbEndpoints = multiClusterState
                     .map {
                         val locality = it.locality
                         val cluster = it.cluster
-
                         createEndpointsGroup(it.servicesState[serviceName], cluster, locality)
                     }
+                val endpoints = duplicateAndOverridePriorityForLocalityLbEndpoints(serviceName, localityLbEndpoints)
+                    ?.let { localityLbEndpoints + it } ?: localityLbEndpoints
 
                 ClusterLoadAssignment.newBuilder()
                     .setClusterName(serviceName)
-                    .addAllEndpoints(localityLbEndpoints)
+                    .addAllEndpoints(endpoints)
                     .build()
             }
     }
@@ -90,6 +91,25 @@ class EnvoyEndpointsFactory(
         } else loadAssignment
     }
 
+    private fun duplicateAndOverridePriorityForLocalityLbEndpoints(
+        serviceName: String,
+        localityLbEndpoints: List<LocalityLbEndpoints>
+    ): LocalityLbEndpoints? {
+        val weightsByService = properties.loadBalancing.trafficSplitting.weightsByService
+        val trafficSplittingEnabled = weightsByService.isNotEmpty()
+        val serviceNotListed = !weightsByService.containsKey(serviceName)
+        val zoneAllowsTrafficSplitting =
+            properties.loadBalancing.trafficSplitting.zonesAllowingTrafficSplitting.contains(currentZone)
+        return if (trafficSplittingEnabled && serviceNotListed && zoneAllowsTrafficSplitting) {
+            localityLbEndpoints
+                .find { properties.loadBalancing.trafficSplitting.zoneName == it.locality.zone }
+                ?.let { LocalityLbEndpoints.newBuilder(it)
+                    .setPriority(HIGHEST_PRIORITY)
+                    .build()
+                }
+        } else null
+    }
+
     private fun assignWeights(
         llbEndpointsList: List<LocalityLbEndpoints>, weights: ZoneWeights
     ): List<LocalityLbEndpoints> {

diff --git a/...tests/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/config/envoy/EnvoyAdmin.kt b/...tests/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/config/envoy/EnvoyAdmin.kt
@@ -69,8 +69,8 @@ class EnvoyAdmin(
         }
     }
 
-    private fun configDump(): String {
-        val response = get("config_dump")
+    fun configDump(): String {
+        val response = get("config_dump?include_eds=on")
         return response.body.use { it!!.string() }
     }
 

diff --git a/...csplitting/WeightedClustersRoutingTest.kt → ...calityWeightedLoadBalancingRoutingTest.kt b/...csplitting/WeightedClustersRoutingTest.kt → ...calityWeightedLoadBalancingRoutingTest.kt
@@ -11,10 +11,11 @@ import pl.allegro.tech.servicemesh.envoycontrol.config.envoy.EnvoyExtension
 import pl.allegro.tech.servicemesh.envoycontrol.config.envoycontrol.EnvoyControlClusteredExtension
 import pl.allegro.tech.servicemesh.envoycontrol.config.service.EchoServiceExtension
 import verifyCallsCountCloseTo
+import verifyCallsCountEq
 import verifyIsReachable
 import java.time.Duration
 
-class WeightedClustersRoutingTest {
+class LocalityWeightedLoadBalancingRoutingTest {
     companion object {
         private const val forceTrafficZone = "dc2"
 
@@ -24,17 +25,18 @@ class WeightedClustersRoutingTest {
             "envoy-control.envoy.snapshot.stateSampleDuration" to Duration.ofSeconds(0),
             "envoy-control.sync.enabled" to true,
             "envoy-control.envoy.snapshot.load-balancing.trafficSplitting.zoneName" to forceTrafficZone,
+            "envoy-control.envoy.snapshot.load-balancing.trafficSplitting.zonesAllowingTrafficSplitting" to "dc1",
             "envoy-control.envoy.snapshot.load-balancing.trafficSplitting.weightsByService.$serviceName.weightByZone.dc1" to 30,
             "envoy-control.envoy.snapshot.load-balancing.trafficSplitting.weightsByService.$serviceName.weightByZone.dc2" to 10,
             "envoy-control.envoy.snapshot.load-balancing.trafficSplitting.weightsByService.$serviceName.weightByZone.dc3" to 1,
             "envoy-control.envoy.snapshot.load-balancing.priorities.zonePriorities" to mapOf(
                 "dc1" to mapOf(
                     "dc1" to 0,
-                    "dc2" to 0,
+                    "dc2" to 1,
                     "dc3" to 3,
                 ),
                 "dc2" to mapOf(
-                    "dc1" to 0,
+                    "dc1" to 1,
                     "dc2" to 0,
                     "dc3" to 3,
                 ),
@@ -53,9 +55,11 @@ class WeightedClustersRoutingTest {
                   outgoing:
                     dependencies:
                       - service: "service-1"
+                      - service: "service-2"
         """.trimIndent()
 
         private val config = Xds.copy(configOverride = echo2Config, serviceName = "echo2")
+        private val configEcho = Xds.copy(configOverride = echo2Config, serviceName = "echo")
 
         @JvmField
         @RegisterExtension
@@ -76,6 +80,10 @@ class WeightedClustersRoutingTest {
         val envoyControl3 =
             EnvoyControlClusteredExtension(consul.serverThird, { properties }, listOf(consul))
 
+        @JvmField
+        @RegisterExtension
+        val echo2ServiceDC1 = EchoServiceExtension()
+
         @JvmField
         @RegisterExtension
         val echoServiceDC1 = EchoServiceExtension()
@@ -94,44 +102,62 @@ class WeightedClustersRoutingTest {
 
         @JvmField
         @RegisterExtension
-        val echoEnvoyDC1 = EnvoyExtension(envoyControl, localService = echoServiceDC1, config)
+        val echo2EnvoyDC1 = EnvoyExtension(envoyControl, localService = echo2ServiceDC1, config)
+
         @JvmField
         @RegisterExtension
-        val echoEnvoyDC2 = EnvoyExtension(envoyControl2)
+        val echoEnvoyDC1 = EnvoyExtension(envoyControl, localService = echoServiceDC1, configEcho)
 
         @JvmField
         @RegisterExtension
-        val echoEnvoyDC3 = EnvoyExtension(envoyControl3)
+        val echoEnvoyDC2 = EnvoyExtension(envoyControl2)
     }
 
     @Test
     fun `should route traffic according to weights`() {
-        consul.serverFirst.operations.registerServiceWithEnvoyOnEgress(echoEnvoyDC1, name = serviceName)
+        consul.serverFirst.operations.registerServiceWithEnvoyOnEgress(echo2EnvoyDC1, name = serviceName)
 
         consul.serverFirst.operations.registerService(upstreamServiceDC1, name = upstreamServiceName)
-        echoEnvoyDC1.verifyIsReachable(upstreamServiceDC1, upstreamServiceName)
+        echo2EnvoyDC1.verifyIsReachable(upstreamServiceDC1, upstreamServiceName)
 
         consul.serverSecond.operations.registerService(upstreamServiceDC2, name = upstreamServiceName)
-        echoEnvoyDC1.verifyIsReachable(upstreamServiceDC2, upstreamServiceName)
+        echo2EnvoyDC1.verifyIsReachable(upstreamServiceDC2, upstreamServiceName)
 
-        echoEnvoyDC1.callUpstreamServiceRepeatedly(upstreamServiceDC1, upstreamServiceDC2)
+        echo2EnvoyDC1.callUpstreamServiceRepeatedly(upstreamServiceDC1, upstreamServiceDC2)
             .verifyCallsCountCloseTo(upstreamServiceDC1, 75)
             .verifyCallsCountCloseTo(upstreamServiceDC2, 25)
-        println("snapshot: " + envoyControl.app.getGlobalSnapshot(false).toString())
+            .verifyCallsCountEq(upstreamServiceDC3, 0)
     }
 
     @Test
     fun `should route traffic according to weights with service tag`() {
-        consul.serverFirst.operations.registerServiceWithEnvoyOnEgress(echoEnvoyDC1, name = serviceName)
+        consul.serverFirst.operations.registerServiceWithEnvoyOnEgress(echo2EnvoyDC1, name = serviceName)
 
         consul.serverFirst.operations.registerService(upstreamServiceDC1, name = upstreamServiceName, tags = listOf("tag"))
-        echoEnvoyDC1.verifyIsReachable(upstreamServiceDC1, upstreamServiceName)
+        echo2EnvoyDC1.verifyIsReachable(upstreamServiceDC1, upstreamServiceName)
 
         consul.serverSecond.operations.registerService(upstreamServiceDC2, name = upstreamServiceName, tags = listOf("tag"))
-        echoEnvoyDC1.verifyIsReachable(upstreamServiceDC2, upstreamServiceName)
+        echo2EnvoyDC1.verifyIsReachable(upstreamServiceDC2, upstreamServiceName)
 
-        echoEnvoyDC1.callUpstreamServiceRepeatedly(upstreamServiceDC1, upstreamServiceDC2, tag = "tag")
+        echo2EnvoyDC1.callUpstreamServiceRepeatedly(upstreamServiceDC1, upstreamServiceDC2, tag = "tag")
             .verifyCallsCountCloseTo(upstreamServiceDC1, 75)
             .verifyCallsCountCloseTo(upstreamServiceDC2, 25)
+            .verifyCallsCountEq(upstreamServiceDC3, 0)
+    }
+
+    @Test
+    fun `should not split traffic for not listed service`() {
+        consul.serverFirst.operations.registerServiceWithEnvoyOnEgress(echoEnvoyDC1, name = "echo")
+
+        consul.serverFirst.operations.registerService(upstreamServiceDC1, name = upstreamServiceName,)
+        echoEnvoyDC1.verifyIsReachable(upstreamServiceDC1, upstreamServiceName)
+
+        consul.serverSecond.operations.registerService(upstreamServiceDC2, name = upstreamServiceName)
+        echoEnvoyDC2.verifyIsReachable(upstreamServiceDC2, upstreamServiceName)
+
+        echoEnvoyDC1.callUpstreamServiceRepeatedly(upstreamServiceDC1, upstreamServiceDC2)
+            .verifyCallsCountEq(upstreamServiceDC1, 100)
+            .verifyCallsCountEq(upstreamServiceDC2, 0)
+            .verifyCallsCountEq(upstreamServiceDC3, 0)
     }
 }
diff --git a/...main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/trafficsplitting/TrafficSplitting.kt b/...main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/trafficsplitting/TrafficSplitting.kt
@@ -28,8 +28,8 @@ fun CallStats.verifyCallsCountCloseTo(service: EchoServiceExtension, expectedCou
     return this
 }
 
-fun CallStats.verifyCallsCountGreaterThan(service: EchoServiceExtension, hits: Int): CallStats {
-    Assertions.assertThat(this.hits(service)).isGreaterThan(hits)
+fun CallStats.verifyCallsCountEq(service: EchoServiceExtension, expectedCount: Int): CallStats {
+    Assertions.assertThat(this.hits(service)).isEqualTo(expectedCount)
     return this
 }