Add support for connecting with Scylla Cloud clusters

Allows driver to build sessions with Scylla Cloud yaml configuration files
and to establish connection with Scylla Cloud clusters.
Takes care of control connection random node issue by replacing the endpoint
after the first connection with one that points to a specific node
in a slightly hacky way.

Author: Bouncheck

core/src/main/java/com/datastax/oss/driver/api/core/session/ProgrammaticArguments.java

@@ -64,6 +64,7 @@ public static Builder builder() {
   private final AuthProvider authProvider;
   private final SslEngineFactory sslEngineFactory;
   private final InetSocketAddress cloudProxyAddress;
+  private final String scyllaCloudNodeDomain;
   private final UUID startupClientId;
   private final String startupApplicationName;
   private final String startupApplicationVersion;
@@ -82,6 +83,7 @@ private ProgrammaticArguments(
       @Nullable AuthProvider authProvider,
       @Nullable SslEngineFactory sslEngineFactory,
       @Nullable InetSocketAddress cloudProxyAddress,
+      @Nullable String scyllaCloudNodeDomain,
       @Nullable UUID startupClientId,
       @Nullable String startupApplicationName,
       @Nullable String startupApplicationVersion,
@@ -99,6 +101,7 @@ private ProgrammaticArguments(
     this.authProvider = authProvider;
     this.sslEngineFactory = sslEngineFactory;
     this.cloudProxyAddress = cloudProxyAddress;
+    this.scyllaCloudNodeDomain = scyllaCloudNodeDomain;
     this.startupClientId = startupClientId;
     this.startupApplicationName = startupApplicationName;
     this.startupApplicationVersion = startupApplicationVersion;
@@ -163,6 +166,11 @@ public InetSocketAddress getCloudProxyAddress() {
     return cloudProxyAddress;
   }
 
+  @Nullable
+  public String getScyllaCloudNodeDomain() {
+    return scyllaCloudNodeDomain;
+  }
+
   @Nullable
   public UUID getStartupClientId() {
     return startupClientId;
@@ -203,6 +211,7 @@ public static class Builder {
     private AuthProvider authProvider;
     private SslEngineFactory sslEngineFactory;
     private InetSocketAddress cloudProxyAddress;
+    private String scyllaCloudNodeDomain;
     private UUID startupClientId;
     private String startupApplicationName;
     private String startupApplicationVersion;
@@ -366,6 +375,14 @@ public Builder withCloudProxyAddress(@Nullable InetSocketAddress cloudAddress) {
       return this;
     }
 
+    @NonNull
+    public Builder withScyllaCloudProxyAddress(
+        @Nullable InetSocketAddress cloudAddress, String scyllaCloudNodeDomain) {
+      this.cloudProxyAddress = cloudAddress;
+      this.scyllaCloudNodeDomain = scyllaCloudNodeDomain;
+      return this;
+    }
+
     @NonNull
     public Builder withAuthProvider(@Nullable AuthProvider authProvider) {
       this.authProvider = authProvider;
@@ -422,6 +439,7 @@ public ProgrammaticArguments build() {
           authProvider,
           sslEngineFactory,
           cloudProxyAddress,
+          scyllaCloudNodeDomain,
           startupClientId,
           startupApplicationName,
           startupApplicationVersion,

core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java

@@ -44,23 +44,24 @@
 import com.datastax.oss.driver.api.core.type.codec.registry.MutableCodecRegistry;
 import com.datastax.oss.driver.api.core.uuid.Uuids;
 import com.datastax.oss.driver.internal.core.ContactPoints;
-import com.datastax.oss.driver.internal.core.config.cloud.CloudConfig;
-import com.datastax.oss.driver.internal.core.config.cloud.CloudConfigFactory;
+import com.datastax.oss.driver.internal.core.config.scyllacloud.ConfigurationBundle;
+import com.datastax.oss.driver.internal.core.config.scyllacloud.ScyllaCloudConnectionConfig;
 import com.datastax.oss.driver.internal.core.config.typesafe.DefaultDriverConfigLoader;
 import com.datastax.oss.driver.internal.core.context.DefaultDriverContext;
 import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
 import com.datastax.oss.driver.internal.core.metadata.DefaultEndPoint;
+import com.datastax.oss.driver.internal.core.metadata.SniEndPoint;
 import com.datastax.oss.driver.internal.core.session.DefaultSession;
 import com.datastax.oss.driver.internal.core.util.concurrent.BlockingOperation;
 import com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.io.InputStream;
 import java.net.InetSocketAddress;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.nio.file.Path;
-import java.nio.file.Paths;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
@@ -96,6 +97,7 @@ public abstract class SessionBuilder<SelfT extends SessionBuilder, SessionT> {
   protected Set<EndPoint> programmaticContactPoints = new HashSet<>();
   protected CqlIdentifier keyspace;
   protected Callable<InputStream> cloudConfigInputStream;
+  protected Callable<InputStream> scyllaCloudConfigInputStream;
 
   protected ProgrammaticArguments.Builder programmaticArgumentsBuilder =
       ProgrammaticArguments.builder();
@@ -655,6 +657,17 @@ public SelfT withCloudSecureConnectBundle(@NonNull Path cloudConfigPath) {
     return self;
   }
 
+  @NonNull
+  public SelfT withScyllaCloudSecureConnectBundle(@NonNull Path cloudConfigPath) {
+    try {
+      URL cloudConfigUrl = cloudConfigPath.toAbsolutePath().normalize().toUri().toURL();
+      this.scyllaCloudConfigInputStream = cloudConfigUrl::openStream;
+    } catch (MalformedURLException e) {
+      throw new IllegalArgumentException("Incorrect format of cloudConfigPath", e);
+    }
+    return self;
+  }
+
   /**
    * Registers a CodecRegistry to use for the session.
    *
@@ -687,6 +700,12 @@ public SelfT withCloudSecureConnectBundle(@NonNull URL cloudConfigUrl) {
     return self;
   }
 
+  @NonNull
+  public SelfT withScyllaCloudSecureConnectBundle(@NonNull URL cloudConfigUrl) {
+    this.scyllaCloudConfigInputStream = cloudConfigUrl::openStream;
+    return self;
+  }
+
   /**
    * Configures this SessionBuilder for Cloud deployments by retrieving connection information from
    * the provided {@link InputStream}.
@@ -711,6 +730,12 @@ public SelfT withCloudSecureConnectBundle(@NonNull InputStream cloudConfigInputS
     return self;
   }
 
+  @NonNull
+  public SelfT withScyllaCloudSecureConnectBundle(@NonNull InputStream cloudConfigInputStream) {
+    this.scyllaCloudConfigInputStream = () -> cloudConfigInputStream;
+    return self;
+  }
+
   /**
    * Configures this SessionBuilder to use the provided Cloud proxy endpoint.
    *
@@ -733,6 +758,13 @@ public SelfT withCloudProxyAddress(@Nullable InetSocketAddress cloudProxyAddress
     return self;
   }
 
+  @NonNull
+  public SelfT withScyllaCloudProxyAddress(
+      @Nullable InetSocketAddress cloudProxyAddress, String nodeDomain) {
+    this.programmaticArgumentsBuilder.withScyllaCloudProxyAddress(cloudProxyAddress, nodeDomain);
+    return self;
+  }
+
   /**
    * A unique identifier for the created session.
    *
@@ -857,16 +889,9 @@ protected final CompletionStage<CqlSession> buildDefaultSessionAsync() {
               : defaultConfigLoader(programmaticArguments.getClassLoader());
 
       DriverExecutionProfile defaultConfig = configLoader.getInitialConfig().getDefaultProfile();
-      if (cloudConfigInputStream == null) {
-        String configUrlString =
-            defaultConfig.getString(DefaultDriverOption.CLOUD_SECURE_CONNECT_BUNDLE, null);
-        if (configUrlString != null) {
-          cloudConfigInputStream = () -> getURL(configUrlString).openStream();
-        }
-      }
       List<String> configContactPoints =
           defaultConfig.getStringList(DefaultDriverOption.CONTACT_POINTS, Collections.emptyList());
-      if (cloudConfigInputStream != null) {
+      if (scyllaCloudConfigInputStream != null) {
         if (!programmaticContactPoints.isEmpty() || !configContactPoints.isEmpty()) {
           LOG.info(
               "Both a secure connect bundle and contact points were provided. These are mutually exclusive. The contact points from the secure bundle will have priority.");
@@ -880,20 +905,27 @@ protected final CompletionStage<CqlSession> buildDefaultSessionAsync() {
           LOG.info(
               "Both a secure connect bundle and SSL options were provided. They are mutually exclusive. The SSL options from the secure bundle will have priority.");
         }
-        CloudConfig cloudConfig =
-            new CloudConfigFactory().createCloudConfig(cloudConfigInputStream.call());
-        addContactEndPoints(cloudConfig.getEndPoints());
+        ScyllaCloudConnectionConfig cloudConfig =
+            ScyllaCloudConnectionConfig.fromInputStream(scyllaCloudConfigInputStream.call());
+        InetSocketAddress proxyAddress = cloudConfig.getCurrentDatacenter().getServer();
+        addContactEndPoints(
+            ImmutableList.of(
+                new SniEndPoint(proxyAddress, cloudConfig.getCurrentDatacenter().getNodeDomain())));
 
         boolean localDataCenterDefined =
             anyProfileHasDatacenterDefined(configLoader.getInitialConfig());
         if (programmaticLocalDatacenter || localDataCenterDefined) {
           LOG.info(
-              "Both a secure connect bundle and a local datacenter were provided. They are mutually exclusive. The local datacenter from the secure bundle will have priority.");
+              "Both a secure connect bundle and a local datacenter were provided. They are mutually exclusive. The currentContext datacenter name from the secure bundle will be ignored.");
+        } else {
           programmaticArgumentsBuilder.clearDatacenters();
+          withLocalDatacenter(cloudConfig.getCurrentContext().getDatacenterName());
         }
-        withLocalDatacenter(cloudConfig.getLocalDatacenter());
-        withSslEngineFactory(cloudConfig.getSslEngineFactory());
-        withCloudProxyAddress(cloudConfig.getProxyAddress());
+        ConfigurationBundle bundle = cloudConfig.createBundle();
+        withSslEngineFactory(bundle.getSSLEngineFactory());
+        withScyllaCloudProxyAddress(
+            proxyAddress, cloudConfig.getCurrentDatacenter().getNodeDomain());
+
         programmaticArguments = programmaticArgumentsBuilder.build();
       }
 
@@ -912,7 +944,6 @@ protected final CompletionStage<CqlSession> buildDefaultSessionAsync() {
           (InternalDriverContext) buildContext(configLoader, programmaticArguments),
           contactPoints,
           keyspace);
-
     } catch (Throwable t) {
       // We construct the session synchronously (until the init() call), but async clients expect a
       // failed future if anything goes wrong. So wrap any error from that synchronous part.
@@ -929,27 +960,6 @@ private boolean anyProfileHasDatacenterDefined(DriverConfig driverConfig) {
     return false;
   }
 
-  /**
-   * Returns URL based on the configUrl setting. If the configUrl has no protocol provided, the
-   * method will fallback to file:// protocol and return URL that has file protocol specified.
-   *
-   * @param configUrl url to config secure bundle
-   * @return URL with file protocol if there was not explicit protocol provided in the configUrl
-   *     setting
-   */
-  private URL getURL(String configUrl) throws MalformedURLException {
-    try {
-      return new URL(configUrl);
-    } catch (MalformedURLException e1) {
-      try {
-        return Paths.get(configUrl).toAbsolutePath().normalize().toUri().toURL();
-      } catch (MalformedURLException e2) {
-        e2.addSuppressed(e1);
-        throw e2;
-      }
-    }
-  }
-
   /**
    * This <b>must</b> return an instance of {@code InternalDriverContext} (it's not expressed
    * directly in the signature to avoid leaking that type through the protected API).

core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannel.java

@@ -69,7 +69,7 @@ public class DriverChannel {
   @SuppressWarnings("RedundantStringConstructorCall")
   static final Object FORCEFUL_CLOSE_MESSAGE = new String("FORCEFUL_CLOSE_MESSAGE");
 
-  private final EndPoint endPoint;
+  private EndPoint endPoint;
   private final Channel channel;
   private final InFlightHandler inFlightHandler;
   private final WriteCoalescer writeCoalescer;
@@ -326,4 +326,9 @@ public SetKeyspaceEvent(CqlIdentifier keyspaceName, Promise<Void> promise) {
       this.promise = promise;
     }
   }
+
+  // Necessary for swapping ControlConnection endpoint when connecting with serverless clusters
+  public void setEndPoint(EndPoint endPoint) {
+    this.endPoint = endPoint;
+  }
 }

core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java

@@ -60,6 +60,7 @@
 import com.datastax.oss.driver.internal.core.metadata.MetadataManager;
 import com.datastax.oss.driver.internal.core.metadata.MultiplexingNodeStateListener;
 import com.datastax.oss.driver.internal.core.metadata.NoopNodeStateListener;
+import com.datastax.oss.driver.internal.core.metadata.ScyllaCloudTopologyMonitor;
 import com.datastax.oss.driver.internal.core.metadata.TopologyMonitor;
 import com.datastax.oss.driver.internal.core.metadata.schema.MultiplexingSchemaChangeListener;
 import com.datastax.oss.driver.internal.core.metadata.schema.NoopSchemaChangeListener;
@@ -236,6 +237,7 @@ public class DefaultDriverContext implements InternalDriverContext {
   private final Map<String, NodeDistanceEvaluator> nodeDistanceEvaluatorsFromBuilder;
   private final ClassLoader classLoader;
   private final InetSocketAddress cloudProxyAddress;
+  private final String scyllaCloudNodeDomain;
   private final LazyReference<RequestLogFormatter> requestLogFormatterRef =
       new LazyReference<>("requestLogFormatter", this::buildRequestLogFormatter, cycleDetector);
   private final UUID startupClientId;
@@ -291,6 +293,7 @@ public DefaultDriverContext(
     this.nodeDistanceEvaluatorsFromBuilder = programmaticArguments.getNodeDistanceEvaluators();
     this.classLoader = programmaticArguments.getClassLoader();
     this.cloudProxyAddress = programmaticArguments.getCloudProxyAddress();
+    this.scyllaCloudNodeDomain = programmaticArguments.getScyllaCloudNodeDomain();
     this.startupClientId = programmaticArguments.getStartupClientId();
     this.startupApplicationName = programmaticArguments.getStartupApplicationName();
     this.startupApplicationVersion = programmaticArguments.getStartupApplicationVersion();
@@ -492,6 +495,9 @@ protected TopologyMonitor buildTopologyMonitor() {
     if (cloudProxyAddress == null) {
       return new DefaultTopologyMonitor(this);
     }
+    if (scyllaCloudNodeDomain != null) {
+      return new ScyllaCloudTopologyMonitor(this, cloudProxyAddress, scyllaCloudNodeDomain);
+    }
     return new CloudTopologyMonitor(this, cloudProxyAddress);
   }
 

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultTopologyMonitor.java

@@ -68,11 +68,11 @@ public class DefaultTopologyMonitor implements TopologyMonitor {
   private static final int INFINITE_PAGE_SIZE = -1;
 
   private final String logPrefix;
-  private final InternalDriverContext context;
-  private final ControlConnection controlConnection;
+  protected final InternalDriverContext context;
+  protected final ControlConnection controlConnection;
   private final Duration timeout;
-  private final boolean reconnectOnInit;
-  private final CompletableFuture<Void> closeFuture;
+  protected final boolean reconnectOnInit;
+  protected final CompletableFuture<Void> closeFuture;
 
   @VisibleForTesting volatile boolean isSchemaV2;
   @VisibleForTesting volatile int port = -1;