#564 Implemented adding a header for locality weighted load balancing

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 Lists all changes with user impact.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
+## [0.20.11]
+### Changed
+- Implemented adding a header for locality weighted load balancing
+
 ## [0.20.10]
 ### Changed
 - Implemented locality weighted load balancing

envoy-control-core/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/snapshot/SnapshotProperties.kt

@@ -164,6 +164,7 @@ class CanaryProperties {
 
 class TrafficSplittingProperties {
     var zoneName = ""
+    var headerName = ""
     var weightsByService: Map<String, ZoneWeights> = mapOf()
 }
 

envoy-control-core/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/snapshot/resource/listeners/filters/EnvoyDefaultFilters.kt

@@ -10,16 +10,15 @@ import pl.allegro.tech.servicemesh.envoycontrol.snapshot.SnapshotProperties
 
 class EnvoyDefaultFilters(
     private val snapshotProperties: SnapshotProperties,
-    private val customLuaMetadata: LuaMetadataProperty.StructPropertyLua
+    private val customLuaMetadata: LuaMetadataProperty.StructPropertyLua,
+    private val currentZone: String
 ) {
     private val rbacFilterFactory = RBACFilterFactory(
         snapshotProperties.incomingPermissions,
         snapshotProperties.routes.status,
         jwtProperties = snapshotProperties.jwt
     )
-    private val luaFilterFactory = LuaFilterFactory(
-        snapshotProperties.incomingPermissions
-    )
+    private val luaFilterFactory = LuaFilterFactory(snapshotProperties)
     private val jwtFilterFactory = JwtFilterFactory(
         snapshotProperties.jwt
     )
@@ -68,6 +67,10 @@ class EnvoyDefaultFilters(
         defaultHeaderToMetadataFilter, defaultServiceTagFilter, defaultEnvoyRouterHttpFilter
     )
 
+    val defaultCurrentZoneHeaderFilter = { _: Group, _: GlobalSnapshot ->
+        luaFilterFactory.ingressCurrentZoneHeaderFilter()
+    }
+
     /**
      * Order matters:
      * * defaultClientNameHeaderFilter has to be before defaultRbacLoggingFilter, because the latter consumes results of
@@ -101,7 +104,8 @@ class EnvoyDefaultFilters(
         val preFilters = listOf(
             defaultClientNameHeaderFilter,
             defaultAuthorizationHeaderFilter,
-            defaultJwtHttpFilter
+            defaultJwtHttpFilter,
+            defaultCurrentZoneHeaderFilter
         )
         val postFilters = listOf(
             defaultRbacLoggingFilter,
@@ -113,7 +117,9 @@ class EnvoyDefaultFilters(
         return preFilters + filters.toList() + postFilters
     }
 
-    val defaultIngressMetadata = { group: Group -> luaFilterFactory.ingressScriptsMetadata(group, customLuaMetadata) }
+    val defaultIngressMetadata = { group: Group ->
+        luaFilterFactory.ingressScriptsMetadata(group, customLuaMetadata, currentZone)
+    }
 
     private fun headerToMetadataConfig(
         rules: List<Config.Rule>,
@@ -144,8 +150,12 @@ class EnvoyDefaultFilters(
     private fun envoyRouterHttpFilter(): HttpFilter = HttpFilter
         .newBuilder()
         .setName("envoy.filters.http.router")
-        .setTypedConfig(Any.pack(Router.newBuilder()
-            .build()))
+        .setTypedConfig(
+            Any.pack(
+                Router.newBuilder()
+                    .build()
+            )
+        )
         .build()
 
     private fun headerToMetadataHttpFilter(headerToMetadataConfig: Config.Builder): HttpFilter {

envoy-control-core/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/snapshot/resource/listeners/filters/EnvoyHttpFilters.kt

@@ -15,9 +15,10 @@ class EnvoyHttpFilters(
 
         fun defaultFilters(
             snapshotProperties: SnapshotProperties,
+            localDatacenter: String,
             customLuaMetadata: LuaMetadataProperty.StructPropertyLua = LuaMetadataProperty.StructPropertyLua()
         ): EnvoyHttpFilters {
-            val defaultFilters = EnvoyDefaultFilters(snapshotProperties, customLuaMetadata)
+            val defaultFilters = EnvoyDefaultFilters(snapshotProperties, customLuaMetadata, localDatacenter)
             return EnvoyHttpFilters(
                 defaultFilters.ingressFilters(),
                 defaultFilters.defaultEgressFilters,

envoy-control-core/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/snapshot/resource/listeners/filters/LuaFilterFactory.kt

@@ -8,17 +8,17 @@ import io.envoyproxy.envoy.config.core.v3.Metadata
 import io.envoyproxy.envoy.extensions.filters.http.lua.v3.Lua
 import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpFilter
 import pl.allegro.tech.servicemesh.envoycontrol.groups.Group
-import pl.allegro.tech.servicemesh.envoycontrol.snapshot.IncomingPermissionsProperties
+import pl.allegro.tech.servicemesh.envoycontrol.snapshot.SnapshotProperties
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.ListPropertyLua
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.StringPropertyLua
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.StructPropertyLua
 
-class LuaFilterFactory(private val incomingPermissionsProperties: IncomingPermissionsProperties) {
+class LuaFilterFactory(private val snapshotProperties: SnapshotProperties) {
 
     private val ingressRbacLoggingScript: String = this::class.java.classLoader
         .getResource("lua/ingress_rbac_logging.lua")!!.readText()
 
-    private val ingressRbacLoggingFilter: HttpFilter? = if (incomingPermissionsProperties.enabled) {
+    private val ingressRbacLoggingFilter: HttpFilter? = if (snapshotProperties.incomingPermissions.enabled) {
         HttpFilter.newBuilder()
             .setName("envoy.lua")
             .setTypedConfig(Any.pack(Lua.newBuilder().setInlineCode(ingressRbacLoggingScript).build()))
@@ -27,7 +27,7 @@ class LuaFilterFactory(private val incomingPermissionsProperties: IncomingPermis
         null
     }
 
-    private val trustedClientIdentityHeader = incomingPermissionsProperties.trustedClientIdentityHeader
+    private val trustedClientIdentityHeader = snapshotProperties.incomingPermissions.trustedClientIdentityHeader
 
     fun ingressRbacLoggingFilter(group: Group): HttpFilter? =
         ingressRbacLoggingFilter.takeIf { group.proxySettings.incoming.permissionsEnabled }
@@ -41,40 +41,61 @@ class LuaFilterFactory(private val incomingPermissionsProperties: IncomingPermis
             .setTypedConfig(Any.pack(Lua.newBuilder().setInlineCode(ingressClientNameHeaderScript).build()))
             .build()
 
-    private val sanUriWildcardRegexForLua = SanUriMatcherFactory(incomingPermissionsProperties.tlsAuthentication)
+    private val ingressCurrentZoneHeaderScript: String = this::class.java.classLoader
+        .getResource("lua/ingress_current_zone_header.lua")!!.readText()
+
+    private val ingressCurrentZoneHeaderFilter: HttpFilter =
+        HttpFilter.newBuilder()
+            .setName("ingress.client.lua")
+            .setTypedConfig(Any.pack(Lua.newBuilder().setInlineCode(ingressCurrentZoneHeaderScript).build()))
+            .build()
+
+    private val sanUriWildcardRegexForLua = SanUriMatcherFactory(
+        snapshotProperties.incomingPermissions.tlsAuthentication
+    )
         .sanUriWildcardRegexForLua
 
-    fun ingressScriptsMetadata(group: Group, customLuaMetadata: StructPropertyLua = StructPropertyLua()): Metadata {
+    fun ingressScriptsMetadata(
+        group: Group,
+        customLuaMetadata: StructPropertyLua = StructPropertyLua(),
+        currentZone: String
+    ): Metadata {
         val metadata = StructPropertyLua(
             "client_identity_headers" to ListPropertyLua(
-                incomingPermissionsProperties
+                snapshotProperties.incomingPermissions
                     .clientIdentityHeaders.map(::StringPropertyLua)
             ),
             "request_id_headers" to ListPropertyLua(
-                incomingPermissionsProperties.requestIdentificationHeaders.map(
+                snapshotProperties.incomingPermissions.requestIdentificationHeaders.map(
                     ::StringPropertyLua
                 )
             ),
             "trusted_client_identity_header" to StringPropertyLua(trustedClientIdentityHeader),
             "san_uri_lua_pattern" to StringPropertyLua(sanUriWildcardRegexForLua),
             "clients_allowed_to_all_endpoints" to ListPropertyLua(
-                incomingPermissionsProperties.clientsAllowedToAllEndpoints.map(
+                snapshotProperties.incomingPermissions.clientsAllowedToAllEndpoints.map(
                     ::StringPropertyLua
                 )
             ),
             "service_name" to StringPropertyLua(group.serviceName),
             "discovery_service_name" to StringPropertyLua(group.discoveryServiceName ?: ""),
             "rbac_headers_to_log" to ListPropertyLua(
-                incomingPermissionsProperties.headersToLogInRbac.map(::StringPropertyLua)
+                snapshotProperties.incomingPermissions.headersToLogInRbac.map(::StringPropertyLua)
+            ),
+            "traffic_splitting_zone_header_name" to StringPropertyLua(
+                snapshotProperties.loadBalancing.trafficSplitting.headerName
             ),
-            ) + customLuaMetadata
+            "current_zone" to StringPropertyLua(currentZone)
+        ) + customLuaMetadata
         return Metadata.newBuilder()
             .putFilterMetadata("envoy.filters.http.lua", metadata.toValue().structValue)
             .build()
     }
 
     fun ingressClientNameHeaderFilter(): HttpFilter? =
         ingressClientNameHeaderFilter.takeIf { trustedClientIdentityHeader.isNotEmpty() }
+
+    fun ingressCurrentZoneHeaderFilter(): HttpFilter = ingressCurrentZoneHeaderFilter
 }
 
 sealed class LuaMetadataProperty<T>(open val value: T) {

envoy-control-core/src/main/resources/lua/ingress_current_zone_header.lua

@@ -0,0 +1,11 @@
+function envoy_on_request(handle)
+    local header_name = handle:metadata():get("traffic_splitting_zone_header_name") or ""
+    local current_zone = handle:metadata():get("current_zone") or ""
+    if header_name == "" then
+        return
+    end
+    handle:headers():add(header_name, current_zone)
+end
+
+function envoy_on_response(handle)
+end

envoy-control-core/src/test/kotlin/pl/allegro/tech/servicemesh/envoycontrol/EnvoySnapshotFactoryTest.kt

@@ -458,7 +458,7 @@ class EnvoySnapshotFactoryTest {
         val egressRoutesFactory = EnvoyEgressRoutesFactory(properties)
         val clustersFactory = EnvoyClustersFactory(properties)
         val endpointsFactory = EnvoyEndpointsFactory(properties, ServiceTagMetadataGenerator(), CURRENT_ZONE)
-        val envoyHttpFilters = EnvoyHttpFilters.defaultFilters(properties)
+        val envoyHttpFilters = EnvoyHttpFilters.defaultFilters(properties, "dc1")
         val listenersFactory = EnvoyListenersFactory(properties, envoyHttpFilters)
         val snapshotsVersions = SnapshotsVersions()
         val meterRegistry: MeterRegistry = SimpleMeterRegistry()

envoy-control-core/src/test/kotlin/pl/allegro/tech/servicemesh/envoycontrol/snapshot/resource/listeners/filters/EnvoyDefaultFiltersTest.kt

@@ -11,7 +11,11 @@ import pl.allegro.tech.servicemesh.envoycontrol.snapshot.SnapshotProperties
 
 class EnvoyDefaultFiltersTest {
 
-    private val defaultFilters = EnvoyDefaultFilters(SnapshotProperties(), LuaMetadataProperty.StructPropertyLua())
+    private val defaultFilters = EnvoyDefaultFilters(
+        SnapshotProperties(),
+        LuaMetadataProperty.StructPropertyLua(),
+        "dc1"
+    )
 
     @Test
     fun `should create default filters`() {

envoy-control-core/src/test/kotlin/pl/allegro/tech/servicemesh/envoycontrol/snapshot/resource/listeners/filters/LuaFilterFactoryTest.kt

@@ -6,13 +6,15 @@ import pl.allegro.tech.servicemesh.envoycontrol.groups.CommunicationMode
 import pl.allegro.tech.servicemesh.envoycontrol.groups.Group
 import pl.allegro.tech.servicemesh.envoycontrol.groups.ServicesGroup
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.IncomingPermissionsProperties
-import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.StructPropertyLua
-import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.ListPropertyLua
-import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.StringPropertyLua
+import pl.allegro.tech.servicemesh.envoycontrol.snapshot.SnapshotProperties
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.BooleanPropertyLua
+import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.ListPropertyLua
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.NumberPropertyLua
+import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.StringPropertyLua
+import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.LuaMetadataProperty.StructPropertyLua
 
 internal class LuaFilterFactoryTest {
+    val properties = SnapshotProperties().also { it.incomingPermissions = IncomingPermissionsProperties() }
 
     @Test
     fun `should create metadata with service name and discovery service name`() {
@@ -24,10 +26,10 @@ internal class LuaFilterFactoryTest {
             serviceName = expectedServiceName,
             discoveryServiceName = expectedDiscoveryServiceName
         )
-        val factory = LuaFilterFactory(IncomingPermissionsProperties())
+        val factory = LuaFilterFactory(properties)
 
         // when
-        val metadata = factory.ingressScriptsMetadata(group)
+        val metadata = factory.ingressScriptsMetadata(group, currentZone = "dc1")
         val givenServiceName = metadata
             .getFilterMetadataOrThrow("envoy.filters.http.lua")
             .getFieldsOrThrow("service_name")
@@ -62,15 +64,36 @@ internal class LuaFilterFactoryTest {
             serviceName = expectedServiceName,
             discoveryServiceName = expectedDiscoveryServiceName
         )
-        val factory = LuaFilterFactory(IncomingPermissionsProperties())
+        val factory = LuaFilterFactory(properties)
 
         // when
-        val luaMetadata = factory.ingressScriptsMetadata(group, customMetadata)
+        val luaMetadata = factory.ingressScriptsMetadata(group, customMetadata, "dc1")
             .getFilterMetadataOrThrow("envoy.filters.http.lua").fieldsMap
 
         // then
         assertThat(luaMetadata["flags"]).isEqualTo(customMetadata["flags"]?.toValue())
         assertThat(luaMetadata["list-value"]).isEqualTo(customMetadata["list-value"]?.toValue())
         assertThat(luaMetadata["count"]).isEqualTo(customMetadata["count"]?.toValue())
     }
+
+    @Test
+    fun `should create metadata with current zone`() {
+        // given
+        val expectedServiceName = "service-1"
+        val expectedDiscoveryServiceName = "consul-service-1"
+        val expectedCurrentZone = "dc1"
+        val group: Group = ServicesGroup(
+            communicationMode = CommunicationMode.XDS,
+            serviceName = expectedServiceName,
+            discoveryServiceName = expectedDiscoveryServiceName
+        )
+        val factory = LuaFilterFactory(properties)
+
+        // when
+        val luaMetadata = factory.ingressScriptsMetadata(group, StructPropertyLua(), "dc1")
+            .getFilterMetadataOrThrow("envoy.filters.http.lua").fieldsMap
+
+        // then
+        assertThat(luaMetadata["current_zone"]).isEqualTo(expectedCurrentZone)
+    }
 }

envoy-control-runner/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/infrastructure/ControlPlaneConfig.kt

@@ -66,11 +66,11 @@ class ControlPlaneConfig {
         globalStateChanges: GlobalStateChanges,
         metrics: EnvoyControlMetrics,
         envoyHttpFilters: EnvoyHttpFilters,
-        consulProperties: ConsulProperties
+        localDatacenter: String
     ): ControlPlane =
         ControlPlane.builder(properties, meterRegistry)
             .withMetrics(metrics)
-            .withCurrentZone(localDatacenter(consulProperties))
+            .withCurrentZone(localDatacenter)
             .withEnvoyHttpFilters(envoyHttpFilters)
             .build(globalStateChanges.combined())
 
@@ -163,11 +163,13 @@ class ControlPlaneConfig {
     @Bean
     @ConditionalOnMissingBean(EnvoyHttpFilters::class)
     fun envoyHttpFilters(
-        properties: EnvoyControlProperties
+        properties: EnvoyControlProperties,
+        localDatacenter: String
     ): EnvoyHttpFilters {
-        return EnvoyHttpFilters.defaultFilters(properties.envoy.snapshot)
+        return EnvoyHttpFilters.defaultFilters(properties.envoy.snapshot, localDatacenter)
     }
 
+    @Bean
     fun localDatacenter(properties: ConsulProperties) =
         ConsulClient(properties.host, properties.port).agentSelf.value?.config?.datacenter ?: "local"