Add testing support and basic tests for Scylla Cloud

Extends `@CCMConfig` annotation with new option to start SNI proxy.
Adds custom `clusterProvider` for tests with proxy enabled.
It uses special `MockSniEndPointFactory` in order to resolve hosts
correctly on local network.

The tests establish connection with single and multi node
SNI proxy enabled ccm clusters using ccm generated .yaml for configuration.
Simple queries like keyspace and table creation are executed and verified.

Adding nodes to already running proxy enabled ccm clusters is not working yet.
However as a side effect of the changes some `long` tests that do add nodes
to clusters without proxy have been fixed.

Author: Bouncheck

driver-core/src/test/java/com/datastax/driver/core/CCMAccess.java

@@ -81,6 +81,9 @@ enum Workload {
   /** @return The binary port for this CCM cluster. */
   int getBinaryPort();
 
+  /** @return The SNI proxy port for this CCM cluster. */
+  int getSniPort();
+
   /** Signals that logs for this CCM cluster should be kept after the cluster is stopped. */
   void setKeepLogs(boolean keepLogs);
 

driver-core/src/test/java/com/datastax/driver/core/CCMBridge.java

@@ -350,14 +350,20 @@ public static boolean isWindows() {
 
   private final int binaryPort;
 
+  private final int sniPort;
+
   private final String ipPrefix;
 
   private final File ccmDir;
 
   private final boolean isDSE;
 
+  private final boolean isScylla;
+
   private final String jvmArgs;
 
+  private final boolean startSniProxy;
+
   private boolean keepLogs = false;
 
   private boolean started = false;
@@ -376,8 +382,10 @@ protected CCMBridge(
       int storagePort,
       int thriftPort,
       int binaryPort,
+      int sniPort,
       int[] jmxPorts,
       String jvmArgs,
+      boolean startSniProxy,
       int[] nodes) {
 
     this.clusterName = clusterName;
@@ -387,8 +395,11 @@ protected CCMBridge(
     this.storagePort = storagePort;
     this.thriftPort = thriftPort;
     this.binaryPort = binaryPort;
+    this.sniPort = sniPort;
     this.isDSE = dseVersion != null;
+    this.isScylla = (getGlobalScyllaVersion() != null);
     this.jvmArgs = jvmArgs;
+    this.startSniProxy = startSniProxy;
     this.nodes = nodes;
     this.ccmDir = Files.createTempDir();
     this.jmxPorts = jmxPorts;
@@ -489,6 +500,11 @@ public int getBinaryPort() {
     return binaryPort;
   }
 
+  @Override
+  public int getSniPort() {
+    return sniPort;
+  }
+
   @Override
   public void setKeepLogs(boolean keepLogs) {
     this.keepLogs = keepLogs;
@@ -554,6 +570,10 @@ public synchronized void start() {
       if (isWindows() && this.cassandraVersion.compareTo(VersionNumber.parse("2.2.4")) >= 0) {
         cmd += " --quiet-windows";
       }
+      if (startSniProxy) {
+        cmd += " --sni-proxy";
+        cmd += " --sni-port " + sniPort;
+      }
       execute(cmd);
 
       // Wait for binary interface on each node.
@@ -692,7 +712,8 @@ public void add(int dc, int n) {
     execute(
         CCM_COMMAND
             + " add node%d -d dc%s -i %s%d -t %s -l %s --binary-itf %s -j %d -r %s -s -b"
-            + (isDSE ? " --dse" : ""),
+            + (isDSE ? " --dse" : "")
+            + (isScylla ? " --scylla" : ""),
         n,
         dc,
         ipPrefix,
@@ -932,6 +953,7 @@ public static class Builder {
     private int[] jmxPorts = {};
     private boolean start = true;
     private boolean dse = isDse();
+    private boolean startSniProxy = false;
     private VersionNumber version = null;
     private final Set<String> createOptions = new LinkedHashSet<String>();
     private final Set<String> jvmArgs = new LinkedHashSet<String>();
@@ -964,6 +986,11 @@ public Builder withoutNodes() {
       return withNodes();
     }
 
+    public Builder withSniProxy() {
+      this.startSniProxy = true;
+      return this;
+    }
+
     /** Enables SSL encryption. */
     public Builder withSSL() {
       cassandraConfiguration.put("client_encryption_options.enabled", "true");
@@ -1111,6 +1138,8 @@ public CCMBridge build() {
       int binaryPort =
           Integer.parseInt(cassandraConfiguration.get("native_transport_port").toString());
 
+      int sniPort = TestUtils.findAvailablePort();
+
       // Copy any supplied jmx ports over, and find available ports for the rest
       int numNodes = 0;
       for (int i : nodes) {
@@ -1157,8 +1186,10 @@ public CCMBridge build() {
               storagePort,
               thriftPort,
               binaryPort,
+              sniPort,
               generatedJmxPorts,
               joinJvmArgs(),
+              startSniProxy,
               nodes);
 
       Runtime.getRuntime()
@@ -1365,6 +1396,7 @@ public boolean equals(Object o) {
         return false;
       if (!createOptions.equals(builder.createOptions)) return false;
       if (!jvmArgs.equals(builder.jvmArgs)) return false;
+      if (startSniProxy != builder.startSniProxy) return false;
       if (!cassandraConfiguration.equals(builder.cassandraConfiguration)) return false;
       if (!dseConfiguration.equals(builder.dseConfiguration)) return false;
       return workloads.equals(builder.workloads);
@@ -1379,6 +1411,7 @@ public int hashCode() {
       result = 31 * result + (version != null ? version.hashCode() : 0);
       result = 31 * result + createOptions.hashCode();
       result = 31 * result + jvmArgs.hashCode();
+      result = 31 * result + (startSniProxy ? 1 : 0);
       result = 31 * result + cassandraConfiguration.hashCode();
       result = 31 * result + dseConfiguration.hashCode();
       result = 31 * result + workloads.hashCode();

driver-core/src/test/java/com/datastax/driver/core/CCMCache.java

@@ -99,6 +99,11 @@ public int getBinaryPort() {
       return ccm.getBinaryPort();
     }
 
+    @Override
+    public int getSniPort() {
+      return ccm.getSniPort();
+    }
+
     @Override
     public void setKeepLogs(boolean keepLogs) {
       ccm.setKeepLogs(keepLogs);

driver-core/src/test/java/com/datastax/driver/core/CCMConfig.java

@@ -13,6 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+/*
+ * Copyright (C) 2022 ScyllaDB
+ *
+ * Modified by ScyllaDB
+ */
 package com.datastax.driver.core;
 
 import static java.lang.annotation.ElementType.METHOD;
@@ -210,6 +215,8 @@ final class Undefined {}
    */
   boolean[] createKeyspace() default {};
 
+  boolean[] startSniProxy() default {};
+
   /**
    * Returns {@code true} if the test class or the test method alters the CCM cluster, e.g. by
    * adding or removing nodes, in which case, it should not be reused after the test is finished.

driver-core/src/test/java/com/datastax/driver/core/CCMTestsSupport.java

@@ -24,6 +24,7 @@
 import static com.datastax.driver.core.CreateCCM.TestMode.PER_CLASS;
 import static com.datastax.driver.core.CreateCCM.TestMode.PER_METHOD;
 import static com.datastax.driver.core.TestUtils.CREATE_KEYSPACE_SIMPLE_FORMAT;
+import static com.datastax.driver.core.TestUtils.addressOfNode;
 import static com.datastax.driver.core.TestUtils.executeNoFail;
 import static com.datastax.driver.core.TestUtils.ipOfNode;
 import static org.assertj.core.api.Assertions.fail;
@@ -39,6 +40,7 @@
 import com.google.common.util.concurrent.Uninterruptibles;
 import java.io.Closeable;
 import java.io.File;
+import java.io.FileInputStream;
 import java.io.IOException;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Constructor;
@@ -154,6 +156,11 @@ public int getBinaryPort() {
       return delegate.getBinaryPort();
     }
 
+    @Override
+    public int getSniPort() {
+      return delegate.getSniPort();
+    }
+
     @Override
     public void setKeepLogs(boolean keepLogs) {
       delegate.setKeepLogs(keepLogs);
@@ -484,6 +491,14 @@ private boolean dirtiesContext() {
       return false;
     }
 
+    @SuppressWarnings("SimplifiableIfStatement")
+    private boolean startSniProxy() {
+      for (CCMConfig ann : annotations) {
+        if (ann != null && ann.startSniProxy().length > 0) return ann.startSniProxy()[0];
+      }
+      return false;
+    }
+
     private CCMBridge.Builder ccmBuilder(Object testInstance) throws Exception {
       if (ccmBuilder == null) {
         ccmBuilder = ccmProvider(testInstance);
@@ -501,6 +516,7 @@ private CCMBridge.Builder ccmBuilder(Object testInstance) throws Exception {
         if (dse != null) ccmBuilder.withDSE(dse);
         if (ssl()) ccmBuilder.withSSL();
         if (auth()) ccmBuilder.withAuth();
+        if (startSniProxy()) ccmBuilder.withSniProxy();
         for (Map.Entry<String, Object> entry : config().entrySet()) {
           ccmBuilder.withCassandraConfiguration(entry.getKey(), entry.getValue());
         }
@@ -742,6 +758,49 @@ public Cluster.Builder createClusterBuilderNoDebouncing() {
     return createClusterBuilder().withQueryOptions(TestUtils.nonDebouncingQueryOptions());
   }
 
+  /**
+   * Returns the cluster builder to test against Scylla Cloud (sniProxy enabled).
+   *
+   * <p>This implementation returns a vanilla builder with contact points and port that match
+   * datacenter description in CCM generated yaml configuration file. This configuration may contain
+   * domain names that cannot be resolved on local machine, therefore we overwrite EndPointFactory
+   * afterwards and add sniProxy contact point using raw ip addresses (with ports from configuration
+   * file). It's not required to call {@link Cluster.Builder#addContactPointsWithPorts}, it will be
+   * done automatically.
+   *
+   * @return The cluster builder to use for the tests.
+   */
+  public Cluster.Builder createClusterBuilderScyllaCloud() throws IOException {
+    assert ccmTestConfig.startSniProxy();
+    Cluster.Builder builder = Cluster.builder();
+
+    File ccmdir = ccm.getCcmDir();
+    File clusterFile = new File(ccmdir, ccm.getClusterName());
+    File yamlFile = new File(clusterFile, "config_data.yaml");
+
+    final ScyllaCloudConnectionConfig cloudConfig =
+        ScyllaCloudConnectionConfig.fromInputStream(new FileInputStream(yamlFile));
+
+    builder.withScyllaCloudConnectionConfig(cloudConfig);
+    builder.withEndPointFactory(
+        new MockSniEndPointFactory(
+            InetSocketAddress.createUnresolved(
+                addressOfNode(1).getHostAddress(),
+                cloudConfig.getCurrentDatacenter().getServer().getPort()),
+            builder.getConfiguration().getPolicies().getEndPointFactory()));
+    builder.addContactPoint(
+        new SniEndPoint(
+            InetSocketAddress.createUnresolved(
+                addressOfNode(1).getHostAddress(),
+                cloudConfig.getCurrentDatacenter().getServer().getPort()),
+            cloudConfig.getCurrentDatacenter().getServer().getHostName()));
+
+    builder
+        .withCodecRegistry(new CodecRegistry())
+        .withPort(cloudConfig.getCurrentDatacenter().getServer().getPort());
+    return builder;
+  }
+
   /**
    * Configures the builder with contact points and port that match the running CCM cluster.
    * Therefore it's not required to call {@link Cluster.Builder#addContactPointsWithPorts}, it will

driver-core/src/test/java/com/datastax/driver/core/MockSniEndPointFactory.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2022 ScyllaDB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.driver.core;
+
+import java.net.InetSocketAddress;
+
+class MockSniEndPointFactory implements EndPointFactory {
+  private final InetSocketAddress proxyAddress;
+  private final EndPointFactory childSniFactory;
+
+  public MockSniEndPointFactory(InetSocketAddress proxyAddress, EndPointFactory childSniFactory) {
+    this.proxyAddress = proxyAddress;
+    this.childSniFactory = childSniFactory;
+  }
+
+  @Override
+  public void init(Cluster cluster) {
+    childSniFactory.init(cluster);
+  }
+
+  @Override
+  public EndPoint create(Row peersRow) {
+    SniEndPoint originalEndPoint = (SniEndPoint) childSniFactory.create(peersRow);
+    return new SniEndPoint(proxyAddress, originalEndPoint.getServerName());
+  }
+}

driver-core/src/test/java/com/datastax/driver/core/ScyllaSniProxyTest.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright (C) 2022 ScyllaDB
+ */
+package com.datastax.driver.core;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.atMost;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.util.Collection;
+import org.testng.annotations.Test;
+
+@CreateCCM(CreateCCM.TestMode.PER_METHOD)
+public class ScyllaSniProxyTest extends CCMTestsSupport {
+
+  private void test_ccm_cluster(int testNodes) {
+    Cluster c = cluster().init();
+    Session s = c.connect();
+    TestUtils.waitForUp(TestUtils.ipOfNode(1), c);
+
+    Collection<Host> hosts = s.getState().getConnectedHosts();
+    assert hosts.size() == testNodes;
+    for (Host host : hosts) {
+      assert (host.getListenAddress() == null
+          || host.getListenAddress().equals(TestUtils.addressOfNode(1)));
+      assert host.getEndPoint().resolve().getAddress().equals(TestUtils.addressOfNode(1));
+      assert host.getEndPoint().resolve().getPort() == ccm().getSniPort();
+      assert !host.getEndPoint().toString().contains("any.");
+    }
+    ((SessionManager) s).cluster.manager.controlConnection.triggerReconnect();
+
+    SchemaChangeListener listener = mock(SchemaChangeListenerBase.class);
+    c.register(listener);
+
+    s.execute(String.format(TestUtils.CREATE_KEYSPACE_SIMPLE_FORMAT, "testks", testNodes));
+    s.execute("CREATE TABLE testks.testtab (a int PRIMARY KEY, b int);");
+
+    verify(listener, times(1)).onTableAdded(any(TableMetadata.class));
+    verify(listener, atLeast(1)).onKeyspaceAdded(any(KeyspaceMetadata.class));
+    verify(listener, atMost(2)).onKeyspaceAdded(any(KeyspaceMetadata.class));
+
+    s.close();
+    c.close();
+  }
+
+  @Test(groups = "short")
+  @CCMConfig(
+      startSniProxy = true,
+      numberOfNodes = 3,
+      clusterProvider = "createClusterBuilderScyllaCloud",
+      dirtiesContext = true)
+  public void test_ccm_cluster_3node() throws InterruptedException {
+    test_ccm_cluster(3);
+  }
+
+  @Test(groups = "short")
+  @CCMConfig(
+      startSniProxy = true,
+      numberOfNodes = 1,
+      clusterProvider = "createClusterBuilderScyllaCloud",
+      dirtiesContext = true)
+  public void test_ccm_cluster_1node() throws InterruptedException {
+    test_ccm_cluster(1);
+  }
+}