Merge pull request kroxylicious#2185 from robobario/sni-clustername-replacement-support

Enable users to inject virtual cluster name into SNI bootstrap address and advertised broker address pattern

Author: robobario

CHANGELOG.md

@@ -7,7 +7,7 @@ Format `<github issue/pr number>: <short description>`.
 
 ## SNAPSHOT
 
-
+* [#2185](https://github.com/kroxylicious/kroxylicious/pull/2185) Add $(virtualClusterName) placeholders to SNI bootstrap address and advertised broker address pattern
 * [#2198](https://github.com/kroxylicious/kroxylicious/pull/2198) Require VirtualCluster name to be a valid DNS label
 * [#2188](https://github.com/kroxylicious/kroxylicious/pull/2188) Delete deprecated bootstrapAddressPattern SNI gateway property
 * [#2186](https://github.com/kroxylicious/kroxylicious/pull/2186) Remove deprecated FilterFactory implementations
@@ -29,6 +29,10 @@ Format `<github issue/pr number>: <short description>`.
 * Remove the deprecated configuration property `brokerAddressPattern` from `sniHostIdentifiesNode` gateway configuration. Use
   `advertisedBrokerAddressPattern` instead.
 * VirtualCluster names are now restricted to a maximum length of 63, and must match pattern `^[a-z0-9]([-a-z0-9]*[a-z0-9])?$` (case insensitive).
+* `virtualClusters[].gateways[].sniHostIdentifiesNode.bootstrapAddress` can now contain an optional replacement token `$(virtualClusterName)`.
+When this is present, it will be replaced with the name of that gateway's VirtualCluster.
+* `virtualClusters[].gateways[].sniHostIdentifiesNode.advertisedBrokerAddressPattern` can now contain an optional replacement token `$(virtualClusterName)`.
+When this is present, it will be replaced with the name of that gateway's VirtualCluster.
 
 ## 0.12.0
 

docs/modules/configuring/con-configuring-vc-gateways.adoc

@@ -225,6 +225,8 @@ The gateway exposes the following broker addresses:
 
 |===
 
+Both the `advertisedBrokerAddressPattern` and `bootstrapAddress` configuration parameters accept the `$(virtualClusterName)` replacement token, 
+which is optional. If included, `$(virtualClusterName)` is replaced with the name of the gateway's virtual cluster.
 
 .Example SNI Host Identifies Node configuration with customized advertised port
 [source,yaml]

kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/tester/KroxyliciousConfigUtils.java

@@ -116,7 +116,9 @@ static String bootstrapServersFor(String virtualCluster, Configuration config, S
             return c.getBootstrapAddress().toString();
         }
         else if (provider.config() instanceof SniRoutingClusterNetworkAddressConfigProviderConfig c) {
-            return new HostPort(c.getBootstrapAddress().host(), c.getAdvertisedPort()).toString();
+            String bootstrapAddressPattern = c.getBootstrapAddressPattern();
+            String replaced = bootstrapAddressPattern.replace("$(virtualClusterName)", virtualCluster);
+            return new HostPort(replaced, c.getAdvertisedPort()).toString();
         }
         else {
             throw new IllegalStateException("I don't know how to handle ClusterEndpointConfigProvider type:" + provider.type());
@@ -141,7 +143,7 @@ public static VirtualClusterGatewayBuilder defaultPortIdentifiesNodeGatewayBuild
     public static VirtualClusterGatewayBuilder defaultSniHostIdentifiesNodeGatewayBuilder(HostPort bootstrapAddress, String advertisedBrokerAddressPattern) {
         return defaultGatewayBuilder()
                 .withNewSniHostIdentifiesNode()
-                .withBootstrapAddress(bootstrapAddress)
+                .withBootstrapAddress(bootstrapAddress.toString())
                 .withAdvertisedBrokerAddressPattern(advertisedBrokerAddressPattern)
                 .endSniHostIdentifiesNode();
     }
@@ -183,7 +185,7 @@ else if (providerDefinition.config() instanceof RangeAwarePortPerNodeClusterNetw
                     strategy, null, cluster.tls()));
         }
         else if (providerDefinition.config() instanceof SniRoutingClusterNetworkAddressConfigProviderConfig sc) {
-            var strategy = new SniHostIdentifiesNodeIdentificationStrategy(sc.getBootstrapAddress(),
+            var strategy = new SniHostIdentifiesNodeIdentificationStrategy(new HostPort(sc.getBootstrapAddressPattern(), sc.getAdvertisedPort()).toString(),
                     sc.getAdvertisedBrokerAddressPattern());
 
             return Stream.of(new VirtualClusterGateway(DEFAULT_GATEWAY_NAME,

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/ExpositionIT.java

@@ -50,6 +50,7 @@
 import io.kroxylicious.net.PassthroughProxy;
 import io.kroxylicious.proxy.config.ConfigurationBuilder;
 import io.kroxylicious.proxy.config.NamedRange;
+import io.kroxylicious.proxy.config.SniHostIdentifiesNodeIdentificationStrategy;
 import io.kroxylicious.proxy.config.VirtualClusterBuilder;
 import io.kroxylicious.proxy.config.VirtualClusterGateway;
 import io.kroxylicious.proxy.config.VirtualClusterGatewayBuilder;
@@ -66,6 +67,7 @@
 
 import edu.umd.cs.findbugs.annotations.NonNull;
 
+import static io.kroxylicious.net.IntegrationTestInetAddressResolverProvider.TEST_DOMAIN;
 import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.DEFAULT_GATEWAY_NAME;
 import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.defaultGatewayBuilder;
 import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.defaultPortIdentifiesNodeGatewayBuilder;
@@ -94,6 +96,7 @@ class ExpositionIT extends BaseIT {
 
     public static final HostPort SNI_BOOTSTRAP = new HostPort("bootstrap." + SNI_BASE_ADDRESS, 9192);
     public static final String SNI_BROKER_ADDRESS_PATTERN = "broker-$(nodeId)." + SNI_BASE_ADDRESS;
+    public static final String SNI_BROKER_ADDRESS_PATTERN_WITH_CLUSTER_NAME = "$(virtualClusterName)-broker-$(nodeId)." + SNI_BASE_ADDRESS;
     public static final String SASL_USER = "user";
     public static final String SASL_PASSWORD = "password";
 
@@ -299,6 +302,58 @@ void exposesTwoSeparateUpstreamClustersUsingSniRouting(KafkaCluster cluster) thr
         }
     }
 
+    @Test
+    void exposeSingleUpstreamClustersWithTwoVirtualClustersUsingIdenticalSNIGatewayConfigurationWithClusterNameReplacement(KafkaCluster cluster) throws Exception {
+        var keystoreTrustStoreList = new ArrayList<KeystoreTrustStorePair>();
+        var virtualClusterCommonNamePattern = IntegrationTestInetAddressResolverProvider.generateFullyQualifiedDomainName(".$(virtualClusterName)");
+        var virtualClusterBootstrapPattern = "bootstrap" + virtualClusterCommonNamePattern;
+        var virtualClusterBrokerAddressPattern = "broker-$(nodeId)" + virtualClusterCommonNamePattern;
+        VirtualClusterGatewayBuilder sniBuilder = defaultSniHostIdentifiesNodeGatewayBuilder(virtualClusterBootstrapPattern + ":9192",
+                virtualClusterBrokerAddressPattern);
+
+        var builder = new ConfigurationBuilder();
+
+        int numberOfVirtualClusters = 2;
+        for (int i = 0; i < numberOfVirtualClusters; i++) {
+            String clusterName = "cluster" + i;
+            String withClusterNameReplaced = virtualClusterCommonNamePattern.replace("$(virtualClusterName)", clusterName);
+            String domain = "*" + withClusterNameReplaced;
+            var keystoreTrustStorePair = buildKeystoreTrustStorePair(domain);
+            keystoreTrustStoreList.add(keystoreTrustStorePair);
+            var virtualCluster = KroxyliciousConfigUtils.baseVirtualClusterBuilder(cluster, clusterName)
+                    .addToGateways(sniBuilder
+                            .withNewTls()
+                            .withNewKeyStoreKey()
+                            .withStoreFile(keystoreTrustStorePair.brokerKeyStore())
+                            .withNewInlinePasswordStoreProvider(keystoreTrustStorePair.password())
+                            .endKeyStoreKey()
+                            .endTls()
+                            .build())
+                    .withLogNetwork(true)
+                    .withLogFrames(true)
+                    .build();
+            builder.addToVirtualClusters(virtualCluster);
+        }
+        try (var tester = kroxyliciousTester(builder)) {
+            for (int i = 0; i < numberOfVirtualClusters; i++) {
+                var trust = keystoreTrustStoreList.get(i);
+                String clusterName = "cluster" + i;
+                try (var admin = tester.admin(clusterName, Map.of(
+                        CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name,
+                        SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, trust.clientTrustStore(),
+                        SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, trust.password()))) {
+                    // do some work to ensure virtual cluster is operational
+                    createTopic(admin, TOPIC + i, 1);
+                    Set<HostPort> nodeAdvertisedAddresses = admin.describeCluster().nodes().get(5, TimeUnit.SECONDS).stream()
+                            .map(node -> new HostPort(node.host(), node.port()))
+                            .collect(Collectors.toSet());
+                    String expectedAdvertisedBrokerHost = "broker-0.%s%s".formatted(clusterName, TEST_DOMAIN);
+                    assertThat(nodeAdvertisedAddresses).singleElement().isEqualTo(new HostPort(expectedAdvertisedBrokerHost, 9192));
+                }
+            }
+        }
+    }
+
     @Test
     void exposesSingleUpstreamClustersUsingMultipleSniGateways(KafkaCluster cluster) throws Exception {
         var keystoreTrustStoreList = new ArrayList<KeystoreTrustStorePair>();
@@ -318,7 +373,7 @@ void exposesSingleUpstreamClustersUsingMultipleSniGateways(KafkaCluster cluster)
                     .addToGateways(new VirtualClusterGatewayBuilder()
                             .withName("gateway-" + i)
                             .withNewSniHostIdentifiesNode()
-                            .withBootstrapAddress(new HostPort(virtualClusterFQDN, 9192))
+                            .withBootstrapAddress(new HostPort(virtualClusterFQDN, 9192).toString())
                             .withAdvertisedBrokerAddressPattern(virtualClusterBrokerAddressPattern.formatted(i))
                             .endSniHostIdentifiesNode()
                             .withNewTls()
@@ -457,6 +512,19 @@ private static Stream<Arguments> virtualClusterConfigurations() throws Exception
                                         .endKeyStoreKey()
                                         .endTls()
                                         .build()),
+                        Map.of(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name,
+                                SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sniKeystoreTrustStorePair.clientTrustStore(),
+                                SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, sniKeystoreTrustStorePair.password())),
+                argumentSet("SniHostIdentifiesNodeWithClusterNameReplacement",
+                        new VirtualClusterBuilder()
+                                .addToGateways(defaultSniHostIdentifiesNodeGatewayBuilder(SNI_BOOTSTRAP.toString(), SNI_BROKER_ADDRESS_PATTERN_WITH_CLUSTER_NAME)
+                                        .withNewTls()
+                                        .withNewKeyStoreKey()
+                                        .withStoreFile(sniKeystoreTrustStorePair.brokerKeyStore())
+                                        .withNewInlinePasswordStoreProvider(sniKeystoreTrustStorePair.password())
+                                        .endKeyStoreKey()
+                                        .endTls()
+                                        .build()),
                         Map.of(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name,
                                 SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sniKeystoreTrustStorePair.clientTrustStore(),
                                 SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, sniKeystoreTrustStorePair.password())));
@@ -567,8 +635,12 @@ void connectToDiscoveryAddress(VirtualClusterBuilder virtualClusterBuilder,
                 .addToVirtualClusters(virtualClusterBuilder.build());
 
         final HostPort discoveryBrokerAddressToProbe;
-        if (virtualClusterBuilder.buildFirstGateway().sniHostIdentifiesNode() != null) {
-            discoveryBrokerAddressToProbe = new HostPort(SNI_BROKER_ADDRESS_PATTERN.replace("$(nodeId)", Integer.toString(cluster.getNumOfBrokers())),
+        SniHostIdentifiesNodeIdentificationStrategy sniStrategy = virtualClusterBuilder.buildFirstGateway().sniHostIdentifiesNode();
+        if (sniStrategy != null) {
+            String pattern = sniStrategy.advertisedBrokerAddressPattern();
+            String replacedNodeId = pattern.replace("$(nodeId)", Integer.toString(cluster.getNumOfBrokers()));
+            String replacedVirtualClusterName = replacedNodeId.replace("$(virtualClusterName)", "demo");
+            discoveryBrokerAddressToProbe = new HostPort(replacedVirtualClusterName,
                     SNI_BOOTSTRAP.port());
         }
         else {

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/config/Configuration.java

@@ -172,21 +172,21 @@ private static VirtualClusterModel toVirtualClusterModel(@NonNull VirtualCluster
     private static void addGateways(@NonNull PluginFactoryRegistry pfr, List<VirtualClusterGateway> gateways, VirtualClusterModel virtualClusterModel) {
         gateways.forEach(gateway -> {
             var config = gateway.clusterNetworkAddressConfigProvider().config();
-            var networkAddress = createDeprecatedProvider(config);
+            var networkAddress = createDeprecatedProvider(config, virtualClusterModel);
             var tls = gateway.tls();
             virtualClusterModel.addGateway(gateway.name(), networkAddress, tls);
         });
     }
 
     @NonNull
     @SuppressWarnings("removal")
-    private static ClusterNetworkAddressConfigProvider createDeprecatedProvider(Object config) {
+    private static ClusterNetworkAddressConfigProvider createDeprecatedProvider(Object config, VirtualClusterModel virtualClusterModel) {
         // We avoid using the buildNetworkAddressProviderService in order to avoid the deprecation notice it will produce.
         if (config instanceof SniRoutingClusterNetworkAddressConfigProviderConfig sniConfig) {
-            return new SniRoutingClusterNetworkAddressConfigProvider().build(sniConfig);
+            return new SniRoutingClusterNetworkAddressConfigProvider().build(sniConfig, virtualClusterModel);
         }
         else if (config instanceof RangeAwarePortPerNodeClusterNetworkAddressConfigProviderConfig rangeConfig) {
-            return new RangeAwarePortPerNodeClusterNetworkAddressConfigProvider().build(rangeConfig);
+            return new RangeAwarePortPerNodeClusterNetworkAddressConfigProvider().build(rangeConfig, virtualClusterModel);
         }
         else {
             throw new IllegalStateException("unexpected provider config type : " + config.getClass().getName());
@@ -198,15 +198,16 @@ private static void addGatewayFromDeprecatedConfig(@NonNull VirtualCluster virtu
                                                        VirtualClusterModel virtualClusterModel) {
         // VirtualCluster config validation should mean this we always have a provider if we reach this point.
         Objects.requireNonNull(virtualCluster.clusterNetworkAddressConfigProvider(), "provider unexpectedly null");
-        var networkAddress = buildNetworkAddressProviderService(virtualCluster.clusterNetworkAddressConfigProvider(), pfr);
+        var networkAddress = buildNetworkAddressProviderService(virtualCluster.clusterNetworkAddressConfigProvider(), pfr, virtualClusterModel);
         virtualClusterModel.addGateway(VirtualCluster.DEFAULT_GATEWAY_NAME, networkAddress, virtualCluster.tls());
     }
 
     private static ClusterNetworkAddressConfigProvider buildNetworkAddressProviderService(@NonNull ClusterNetworkAddressConfigProviderDefinition definition,
-                                                                                          @NonNull PluginFactoryRegistry registry) {
+                                                                                          @NonNull PluginFactoryRegistry registry,
+                                                                                          VirtualClusterModel virtualClusterModel) {
         var provider = registry.pluginFactory(ClusterNetworkAddressConfigProviderService.class)
                 .pluginInstance(definition.type());
-        return provider.build(definition.config());
+        return provider.build(definition.config(), virtualClusterModel);
     }
 
     public List<MicrometerDefinition> getMicrometer() {

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/config/SniHostIdentifiesNodeIdentificationStrategy.java

@@ -24,7 +24,7 @@
  *        must be resolvable and routable from the client's network).  A port number can be included.  If the port number is not included, the port number assigned
  *        to the bootstrapAddress is used.  One pattern is supported: {@code $(nodeId)} which interpolates the node id into the address (required).
  */
-public record SniHostIdentifiesNodeIdentificationStrategy(@NonNull @JsonProperty(required = true) HostPort bootstrapAddress,
+public record SniHostIdentifiesNodeIdentificationStrategy(@NonNull @JsonProperty(required = true) String bootstrapAddress,
                                                           @NonNull @JsonProperty(required = true) String advertisedBrokerAddressPattern)
         implements NodeIdentificationStrategy {
     public SniHostIdentifiesNodeIdentificationStrategy {

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/clusternetworkaddressconfigprovider/BrokerAddressPatternUtils.java

@@ -12,17 +12,18 @@
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.function.Consumer;
+import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
 
 final class BrokerAddressPatternUtils {
-    private static final String LITERAL_NODE_ID = "$(nodeId)";
+    public static final String LITERAL_NODE_ID = "$(nodeId)";
+    public static final String LITERAL_VIRTUAL_CLUSTER_NAME = "$(virtualClusterName)";
     private static final Pattern LITERAL_NODE_ID_PATTERN = Pattern.compile(Pattern.quote(LITERAL_NODE_ID));
     private static final Pattern PORT_SPECIFIER_RE = Pattern.compile("^(.*):([1-9]\\d*)$");
     private static final Pattern TOKEN_RE = Pattern.compile("(\\$\\([^)]+\\))");
-    public static Set<String> EXPECTED_TOKEN_SET = Set.of(LITERAL_NODE_ID);
 
     private BrokerAddressPatternUtils() {
     }
@@ -91,6 +92,10 @@ static PatternAndPort parse(@NonNull String address) {
         }
     }
 
+    public static String replaceVirtualClusterName(String toReplace, String clusterName) {
+        return toReplace.replaceAll(Pattern.quote(LITERAL_VIRTUAL_CLUSTER_NAME), Matcher.quoteReplacement(clusterName));
+    }
+
     /**
      * Represents a pattern string and optional port.
      * @param addressPattern address pattern, must not be null
@@ -110,8 +115,11 @@ static String replaceLiteralNodeId(String brokerAddressPattern, int nodeId) {
         return LITERAL_NODE_ID_PATTERN.matcher(brokerAddressPattern).replaceAll(Integer.toString(nodeId));
     }
 
-    static Pattern createNodeIdCapturingRegex(String brokerAddressPattern) {
-        var partsQuoted = Arrays.stream(brokerAddressPattern.split(Pattern.quote(LITERAL_NODE_ID))).map(Pattern::quote);
-        return Pattern.compile(partsQuoted.collect(Collectors.joining("(\\d+)")), Pattern.CASE_INSENSITIVE);
+    static String createNodeIdCapturingRegexPattern(String brokerAddressPattern, String virtualClusterName) {
+        String clusterReplaced = replaceVirtualClusterName(brokerAddressPattern, virtualClusterName);
+        // the -1 split limit produces an empty final item in the array if the node id replacement is at the end of the string
+        var partsQuoted = Arrays.stream(clusterReplaced.split(Pattern.quote(LITERAL_NODE_ID), -1)).map(Pattern::quote);
+        return partsQuoted.collect(Collectors.joining("(\\d+)"));
     }
+
 }