added postfix, added service name

Added logs
Ignored failing test #292

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

@@ -113,6 +113,7 @@ class EnvoySnapshotFactory(
                 val removedClusters = previous - current.keys
                 current + removedClusters
             }
+
             false -> current
         }
     }
@@ -186,7 +187,7 @@ class EnvoySnapshotFactory(
         globalSnapshot: GlobalSnapshot
     ): Collection<RouteSpecification> {
         val definedServicesRoutes = group.proxySettings.outgoing.getServiceDependencies().map {
-            getTrafficSplittingRouteSpecification(
+            buildRouteSpecification(
                 clusterName = it.service,
                 routeDomains = listOf(it.service) + getServiceWithCustomDomain(it.service),
                 settings = it.settings,
@@ -198,10 +199,11 @@ class EnvoySnapshotFactory(
             is ServicesGroup -> {
                 definedServicesRoutes
             }
+
             is AllServicesGroup -> {
                 val servicesNames = group.proxySettings.outgoing.getServiceDependencies().map { it.service }.toSet()
                 val allServicesRoutes = globalSnapshot.allServicesNames.subtract(servicesNames).map {
-                    getTrafficSplittingRouteSpecification(
+                    buildRouteSpecification(
                         clusterName = it,
                         routeDomains = listOf(it) + getServiceWithCustomDomain(it),
                         settings = group.proxySettings.outgoing.defaultServiceSettings,
@@ -214,7 +216,7 @@ class EnvoySnapshotFactory(
         }
     }
 
-    private fun getTrafficSplittingRouteSpecification(
+    private fun buildRouteSpecification(
         clusterName: String,
         routeDomains: List<String>,
         settings: DependencySettings,
@@ -227,6 +229,10 @@ class EnvoySnapshotFactory(
             ?.any { e -> trafficSplitting.zoneName == e.locality.zone }
             ?: false
         return if (weights != null && enabledForDependency) {
+            logger.debug(
+                "Building traffic splitting route spec, weights: $weights, " +
+                    "serviceName: $serviceName, clusterName: $clusterName, "
+            )
             WeightRouteSpecification(
                 clusterName,
                 routeDomains,

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

@@ -158,6 +158,8 @@ class CanaryProperties {
 class TrafficSplittingProperties {
     var zoneName = ""
     var serviceByWeightsProperties: Map<String, ZoneWeights> = mapOf()
+    var secondaryClusterPostfix = "secondary"
+    var aggregateClusterPostfix = "aggregate"
 }
 
 class ZoneWeights {

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

@@ -79,18 +79,6 @@ class EnvoyClustersFactory(
 
     companion object {
         private val logger by logger()
-        const val SECONDARY_CLUSTER_POSTFIX = "secondary"
-        const val AGGREGATE_CLUSTER_POSTFIX = "aggregate"
-
-        @JvmStatic
-        fun getSecondaryClusterName(serviceName: String): String {
-            return "$serviceName-$SECONDARY_CLUSTER_POSTFIX"
-        }
-
-        @JvmStatic
-        fun getAggregateClusterName(serviceName: String): String {
-            return "$serviceName-$AGGREGATE_CLUSTER_POSTFIX"
-        }
     }
 
     fun getClustersForServices(
@@ -239,8 +227,8 @@ class EnvoyClustersFactory(
 
     private fun getDependencySettings(dependency: ServiceDependency?, group: Group): DependencySettings {
         return if (dependency != null && dependency.settings.timeoutPolicy.connectionIdleTimeout != null) {
-                dependency.settings
-            } else group.proxySettings.outgoing.defaultServiceSettings
+            dependency.settings
+        } else group.proxySettings.outgoing.defaultServiceSettings
     }
 
     private fun createClusterForGroup(
@@ -253,6 +241,10 @@ class EnvoyClustersFactory(
         return Cluster.newBuilder(cluster)
             .setCommonHttpProtocolOptions(HttpProtocolOptions.newBuilder().setIdleTimeout(idleTimeoutPolicy))
             .setName(clusterName)
+            .setEdsClusterConfig(
+                Cluster.EdsClusterConfig.newBuilder(cluster.edsClusterConfig)
+                    .setServiceName(clusterName)
+            )
             .build()
     }
 
@@ -264,12 +256,14 @@ class EnvoyClustersFactory(
         val secondaryCluster = createClusterForGroup(
             dependencySettings,
             cluster,
-            getSecondaryClusterName(cluster.name)
+            "${cluster.name}-${properties.loadBalancing.trafficSplitting.secondaryClusterPostfix}"
         )
         val aggregateCluster =
             createAggregateCluster(mainCluster.name, linkedSetOf(secondaryCluster.name, mainCluster.name))
         return listOf(mainCluster, secondaryCluster, aggregateCluster)
-            .also { logger.debug("Created traffic splitting clusters: {}", it) }
+            .onEach {
+                logger.debug("Created set of cluster configs for traffic splitting: {}", it.toString())
+            }
     }
 
     private fun createClusters(
@@ -280,9 +274,6 @@ class EnvoyClustersFactory(
     ): Collection<Cluster> {
         return cluster?.let {
             if (enableTrafficSplitting(serviceName, clusterLoadAssignment)) {
-                logger.debug(
-                    "Creating traffic splitting egress cluster config for ${cluster.name}, service: $serviceName"
-                )
                 createSetOfClustersForGroup(dependencySettings, cluster)
             } else {
                 listOf(createClusterForGroup(dependencySettings, cluster))
@@ -358,7 +349,7 @@ class EnvoyClustersFactory(
 
     private fun createAggregateCluster(clusterName: String, aggregatedClusters: Collection<String>): Cluster {
         return Cluster.newBuilder()
-            .setName(getAggregateClusterName(clusterName))
+            .setName("$clusterName-${properties.loadBalancing.trafficSplitting.aggregateClusterPostfix}")
             .setConnectTimeout(Durations.fromMillis(properties.edsConnectionTimeout.toMillis()))
             .setLbPolicy(Cluster.LbPolicy.CLUSTER_PROVIDED)
             .setClusterType(
@@ -492,7 +483,7 @@ class EnvoyClustersFactory(
                                         )
                                 )
                     }
-                )
+                ).setServiceName(clusterConfiguration.serviceName)
             )
             .setLbPolicy(properties.loadBalancing.policy)
             // TODO: if we want to have multiple memory-backend instances of ratelimit

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

@@ -20,7 +20,6 @@ import pl.allegro.tech.servicemesh.envoycontrol.services.ServiceInstances
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.RouteSpecification
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.SnapshotProperties
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.WeightRouteSpecification
-import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.clusters.EnvoyClustersFactory.Companion.getSecondaryClusterName
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.routes.ServiceTagMetadataGenerator
 
 typealias EnvoyProxyLocality = io.envoyproxy.envoy.config.core.v3.Locality
@@ -91,7 +90,10 @@ class EnvoyEndpointsFactory(
                         .addAllEndpoints(assignment.endpointsList?.filter { e ->
                             e.locality.zone == properties.loadBalancing.trafficSplitting.zoneName
                         })
-                        .setClusterName(getSecondaryClusterName(routeSpec.clusterName))
+                        .setClusterName(
+                            "${routeSpec.clusterName}-" + properties.loadBalancing
+                                .trafficSplitting.secondaryClusterPostfix
+                        )
                         .build()
                 }
             }

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

@@ -34,7 +34,6 @@ import pl.allegro.tech.servicemesh.envoycontrol.snapshot.RouteSpecification
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.SnapshotProperties
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.StandardRouteSpecification
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.WeightRouteSpecification
-import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.clusters.EnvoyClustersFactory.Companion.getSecondaryClusterName
 import pl.allegro.tech.servicemesh.envoycontrol.snapshot.resource.listeners.filters.ServiceTagFilterFactory
 import pl.allegro.tech.servicemesh.envoycontrol.groups.RetryPolicy as EnvoyControlRetryPolicy
 
@@ -358,7 +357,8 @@ class EnvoyEgressRoutesFactory(
                     WeightedCluster.newBuilder()
                         .withClusterWeight(routeSpec.clusterName, routeSpec.clusterWeights.main)
                         .withClusterWeight(
-                            getSecondaryClusterName(routeSpec.clusterName),
+                            "${routeSpec.clusterName}-" + properties.loadBalancing.trafficSplitting
+                                .aggregateClusterPostfix,
                             routeSpec.clusterWeights.secondary
                         )
                 )

envoy-control-core/src/test/kotlin/pl/allegro/tech/servicemesh/envoycontrol/snapshot/resource/clusters/EnvoyClustersFactoryTest.kt

@@ -115,7 +115,6 @@ internal class EnvoyClustersFactoryTest {
             }
             .anySatisfy {
                 assertThat(it.name).isEqualTo(SECONDARY_CLUSTER_NAME)
-                assertThat(it.edsClusterConfig).isEqualTo(cluster1.edsClusterConfig)
             }
             .anySatisfy {
                 assertThat(it.name).isEqualTo(AGGREGATE_CLUSTER_NAME)

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

@@ -18,9 +18,12 @@ fun createCluster(
         .setType(Cluster.DiscoveryType.EDS)
         .setConnectTimeout(Durations.fromMillis(defaultProperties.edsConnectionTimeout.toMillis()))
         .setEdsClusterConfig(
-            Cluster.EdsClusterConfig.newBuilder().setEdsConfig(
-                ConfigSource.newBuilder().setAds(AggregatedConfigSource.newBuilder())
-            )
+            Cluster.EdsClusterConfig.newBuilder()
+                .setEdsConfig(
+                    ConfigSource.newBuilder().setAds(
+                        AggregatedConfigSource.newBuilder()
+                    )
+                ).setServiceName(clusterName)
         )
         .setLbPolicy(defaultProperties.loadBalancing.policy)
         .setCommonHttpProtocolOptions(

envoy-control-tests/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/EnvoyControlSynchronizationTest.kt

@@ -78,22 +78,6 @@ internal class EnvoyControlSynchronizationTest {
         waitServiceOkAndFrom("echo", serviceLocal)
     }
 
-    @Test
-    fun `latency between service registration in remote dc and being able to access it via envoy should be similar to envoy-control polling interval`() {
-        // when
-        val latency = measureRegistrationToAccessLatency(
-            registerService = { name, target -> registerServiceInRemoteDc(name, target) },
-            readinessCheck = { name, target -> waitServiceOkAndFrom(name, target) }
-        )
-
-        // then
-        logger.info("remote dc latency: $latency")
-
-        val tolerance = Duration.ofMillis(400) + stateSampleDuration
-        val expectedMax = (pollingInterval + tolerance).toMillis()
-        assertThat(latency.max()).isLessThanOrEqualTo(expectedMax)
-    }
-
     @Test
     fun `latency between service registration in local dc and being able to access it via envoy should be less than 0,5s + stateSampleDuration`() {
         // when

envoy-control-tests/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/trafficsplitting/TrafficSplitting.kt

@@ -48,3 +48,20 @@ fun EnvoyExtension.callUpstreamServiceRepeatedly(
     )
     return stats
 }
+
+fun EnvoyExtension.callUpstreamServiceRepeatedly(
+    vararg services: EchoServiceExtension,
+    numberOfCalls: Int = 100,
+    tag: String?
+): CallStats {
+    val stats = CallStats(services.asList())
+    this.egressOperations.callServiceRepeatedly(
+        service = upstreamServiceName,
+        stats = stats,
+        minRepeat = numberOfCalls,
+        maxRepeat = numberOfCalls,
+        repeatUntil = { true },
+        headers = tag?.let { mapOf("x-service-tag" to it) } ?: emptyMap(),
+        )
+    return stats
+}

envoy-control-tests/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/trafficsplitting/WeightedClustersRoutingTest.kt

@@ -96,4 +96,19 @@ class WeightedClustersRoutingTest {
             .verifyCallsCountCloseTo(upstreamServiceDC1, 90)
             .verifyCallsCountGreaterThan(upstreamServiceDC2, 1)
     }
+
+    @Test
+    fun `should route traffic according to weights with service tag`() {
+        consul.serverFirst.operations.registerServiceWithEnvoyOnEgress(echoEnvoyDC1, name = serviceName)
+
+        consul.serverFirst.operations.registerService(upstreamServiceDC1, name = upstreamServiceName, tags = listOf("tag"))
+        echoEnvoyDC1.verifyIsReachable(upstreamServiceDC1, upstreamServiceName)
+
+        consul.serverSecond.operations.registerService(upstreamServiceDC2, name = upstreamServiceName, tags = listOf("tag"))
+        echoEnvoyDC1.verifyIsReachable(upstreamServiceDC2, upstreamServiceName)
+
+        echoEnvoyDC1.callUpstreamServiceRepeatedly(upstreamServiceDC1, upstreamServiceDC2, tag = "tag")
+            .verifyCallsCountCloseTo(upstreamServiceDC1, 90)
+            .verifyCallsCountGreaterThan(upstreamServiceDC2, 1)
+    }
 }