feat(Spanner): integrate SourceConfigParser to centralize shard configuration loading for SourceDbToSpanner pipelines.

it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/cloudsql/CloudSqlShardOrchestrator.java

@@ -420,38 +420,28 @@ protected void createLogicalDatabases() {
   protected String generateAndUploadConfig(String artifactName) {
     LOG.info("Generating and uploading shard configuration...");
     JSONObject config = new JSONObject();
-    config.put("configType", "dataflow");
-    JSONObject shardConfigBulk = new JSONObject();
-    JSONArray dataShards = new JSONArray();
+    JSONArray shardConfigs = new JSONArray();
 
     int shardIdx = 0;
     for (Map.Entry<String, List<String>> entry : requestedShardMap.entrySet()) {
       String instanceName = entry.getKey();
       String ip = instanceIpMap.get(instanceName);
       List<String> dbNames = entry.getValue();
 
-      JSONObject dataShard = new JSONObject();
-      dataShard.put("dataShardId", instanceName);
-      dataShard.put("host", ip);
-      dataShard.put("port", port);
-      dataShard.put("user", username);
-      dataShard.put("password", password);
-
-      JSONArray databases = new JSONArray();
       for (String dbName : dbNames) {
-        JSONObject db = new JSONObject();
-        db.put("dbName", dbName);
-        db.put("databaseId", String.format("%s%02d%s", "shard_", shardIdx, dbName));
-        db.put("refDataShardId", instanceName);
-        databases.put(db);
+        JSONObject shardConfig = new JSONObject();
+        shardConfig.put("logicalShardId", String.format("%s%02d_%s", "shard_", shardIdx, dbName));
+        shardConfig.put("host", ip);
+        shardConfig.put("port", port);
+        shardConfig.put("user", username);
+        shardConfig.put("password", password);
+        shardConfig.put("dbName", dbName);
+        shardConfigs.put(shardConfig);
       }
       shardIdx++;
-      dataShard.put("databases", databases);
-      dataShards.put(dataShard);
     }
 
-    shardConfigBulk.put("dataShards", dataShards);
-    config.put("shardConfigurationBulk", shardConfigBulk);
+    config.put("shardConfigs", shardConfigs);
 
     String configContent = config.toString();
     GcsArtifact artifact =

v2/sourcedb-to-spanner/README.md

@@ -62,7 +62,7 @@ mvn test
 ### Executing Template
 
 #### Required Parameters
-* **sourceConfigURL** (Configuration to connect to the source database): Can be the JDBC URL or the location of the sharding config. (Example: jdbc:mysql://10.10.10.10:3306/testdb or gs://test1/shard.conf). Refer to src/main/scripts/create_simple_shard_config.bash for steps to generate a shard configuration.
+* **sourceConfigURL** (Source connection config file URL): The URL of the source connection config file. The file format is dependent on the source type. For Astra, it will point to an Astra connection config file ([sample](src/test/resources/SourceConfig/astra-connection-config.json)). For JDBC, it will point to a JDBC sharding config file ([sample](src/test/resources/SourceConfig/jdbc-shard-config.json)). For Cassandra, it will point to a Cassandra driver config file ([sample](src/test/resources/SourceConfig/cassandra-driver-config.conf)). This parameter is required. Refer to src/main/scripts/create_simple_shard_config.bash for steps to generate a shard configuration.
 * **username** (username of the source database): The username which can be used to connect to the source database.
 * **password** (username of the source database): The username which can be used to connect to the source database.
 * **instanceId** (Cloud Spanner Instance Id.): The destination Cloud Spanner instance.
@@ -86,7 +86,7 @@ export JOB_NAME="${IMAGE_NAME}-`date +%Y%m%d-%H%M%S-%N`"
 gcloud dataflow flex-template run ${JOB_NAME} \
         --project=${PROJECT} --region=us-central1 \
         --template-file-gcs-location=${TEMPLATE_IMAGE_SPEC} \
-        --parameters sourceConfigURL="jdbc:mysql://<source_ip>:3306/<mysql_db_name>",username=<mysql user>,password=<mysql pass>,instanceId="<spanner instanceid>",databaseId="<spanner_database_id>",projectId="$PROJECT",outputDirectory=gs://<gcs-dir> \
+        --parameters sourceConfigURL="gs://<bucket-name>/source-config.json",username=<mysql user>,password=<mysql pass>,instanceId="<spanner instanceid>",databaseId="<spanner_database_id>",projectId="$PROJECT",outputDirectory=gs://<gcs-dir> \
         --additional-experiments=disable_runner_v2
 ```
 #### Replaying DLQ entries.

v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner.md

@@ -18,7 +18,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat
 ## Parameters
 
 #### Required Parameters
-* **sourceConfigURL** (Configuration to connect to the source database): Can be the JDBC URL or the location of the sharding config. (Example: jdbc:mysql://10.10.10.10:3306/testdb or gs://test1/shard.conf)
+* **sourceConfigURL** (Source connection config file URL.): The URL of the source connection config file. The file format is dependent on the source type. For Astra, it will point to an Astra connection config file ([sample](src/test/resources/SourceConfig/astra-connection-config.json)). For JDBC, it will point to a JDBC sharding config file ([sample](src/test/resources/SourceConfig/jdbc-shard-config.json)). For Cassandra, it will point to a Cassandra driver config file ([sample](src/test/resources/SourceConfig/cassandra-driver-config.conf)). This parameter is required.
 * **username** (username of the source database): The username which can be used to connect to the source database.
 * **password** (username of the source database): The username which can be used to connect to the source database.
 * **instanceId** (Cloud Spanner Instance Id.): The destination Cloud Spanner instance.
@@ -519,7 +519,7 @@ resource "google_dataflow_flex_template_job" "sourcedb_to_spanner_flex" {
     instanceId = "<instanceId>"
     databaseId = "<databaseId>"
     projectId = "<projectId>"
-    sourceConfigURL = "jdbc:mysql://some-host:3306/sampledb"
+    sourceConfigURL = "gs://your-bucket/source-config.json"
     username = "<username>"
     password = "<password>"
     outputDirectory = "gs://your-bucket/dir"

v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner_Flex.md

@@ -30,7 +30,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat
 * **sourceDbDialect**: Possible values are `CASSANDRA`, `MYSQL` and `POSTGRESQL`. Defaults to: MYSQL.
 * **jdbcDriverJars**: The comma-separated list of driver JAR files. For example, `gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar`. Defaults to empty.
 * **jdbcDriverClassName**: The JDBC driver class name. For example, `com.mysql.jdbc.Driver`. Defaults to: com.mysql.jdbc.Driver.
-* **sourceConfigURL**: The URL to connect to the source database host. This can be either: 1. A JDBC connection URL for a single source database, which must contain the  host, port and source db name and can optionally contain properties like  autoReconnect, maxReconnects etc. Format: `jdbc:{mysql|postgresql}://{host}:{port}/{dbName}?{parameters}`. For example,`jdbc:mysql://127.4.5.30:3306/my-db?autoReconnect=true&maxReconnects=10&unicode=true&characterEncoding=UTF-8`. 2. A Cloud Storage path to a shard config file for sharded migrations. For example, `gs://my-bucket/my-shard-config.yaml`. This parameter is required except for ASTRA_DB source. Defaults to empty.
+* **sourceConfigURL**: The URL of the source connection config file. The file format is dependent on the source type. For Astra, it will point to an Astra connection config file ([sample](src/test/resources/SourceConfig/astra-connection-config.json)). For JDBC, it will point to a JDBC sharding config file ([sample](src/test/resources/SourceConfig/jdbc-shard-config.json)). For Cassandra, it will point to a Cassandra driver config file ([sample](src/test/resources/SourceConfig/cassandra-driver-config.conf)). This parameter is required. Defaults to empty.
 * **username**: The username to be used for the JDBC connection. Defaults to empty.
 * **password**: The password to be used for the JDBC connection. Defaults to empty.
 * **tables**: Tables to migrate from source. Defaults to empty.

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/OptionsToConfigBuilder.java

@@ -21,6 +21,7 @@
 import com.google.cloud.teleport.v2.reader.io.schema.SourceSchemaReference;
 import com.google.cloud.teleport.v2.source.SourceConnectorFactory;
 import com.google.cloud.teleport.v2.source.jdbc.AbstractJdbcSrcToSpSourceConnector;
+import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
 import com.google.cloud.teleport.v2.spanner.migrations.utils.DataflowWorkerMachineTypeUtils;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
@@ -57,16 +58,8 @@ public static String extractWorkerZone(PipelineOptions options) {
   }
 
   public static JdbcIOWrapperConfig getJdbcIOWrapperConfigWithDefaults(
-      SourceDbToSpannerOptions options,
-      List<String> tables,
-      String shardId,
-      Wait.OnSignal<?> waitOn) {
+      SourceDbToSpannerOptions options, Shard shard, List<String> tables, Wait.OnSignal<?> waitOn) {
     SQLDialect sqlDialect = SQLDialect.valueOf(options.getSourceDbDialect());
-    String sourceDbURL = options.getSourceConfigURL();
-    String dbName = extractDbFromURL(sourceDbURL);
-    String username = options.getUsername();
-    String password = options.getPassword();
-    String namespace = options.getNamespace();
 
     String jdbcDriverClassName = options.getJdbcDriverClassName();
     String jdbcDriverJars = options.getJdbcDriverJars();
@@ -83,15 +76,7 @@ public static JdbcIOWrapperConfig getJdbcIOWrapperConfigWithDefaults(
     return getJdbcIOWrapperConfig(
         sqlDialect,
         tables,
-        sourceDbURL,
-        null,
-        null,
-        0,
-        username,
-        password,
-        dbName,
-        namespace,
-        shardId,
+        shard,
         jdbcDriverClassName,
         jdbcDriverJars,
         maxConnections,
@@ -107,15 +92,7 @@ public static JdbcIOWrapperConfig getJdbcIOWrapperConfigWithDefaults(
   public static JdbcIOWrapperConfig getJdbcIOWrapperConfig(
       SQLDialect sqlDialect,
       List<String> tables,
-      String sourceDbURL,
-      String host,
-      String connectionProperties,
-      int port,
-      String username,
-      String password,
-      String dbName,
-      String namespace,
-      String shardId,
+      Shard shard,
       String jdbcDriverClassName,
       String jdbcDriverJars,
       long maxConnections,
@@ -130,15 +107,16 @@ public static JdbcIOWrapperConfig getJdbcIOWrapperConfig(
         SourceConnectorFactory.getSourceConnectorByDialect(sqlDialect);
     JdbcIOWrapperConfig.Builder builder = connector.getJdbcIOWrapperConfigBuilder();
     SourceSchemaReference sourceSchemaReference =
-        connector.getSourceSchemaReference(dbName, namespace);
+        connector.getSourceSchemaReference(shard.getDbName(), shard.getNamespace());
     builder =
         builder
             .setSourceSchemaReference(sourceSchemaReference)
             .setDbAuth(
                 LocalCredentialsProvider.builder()
                     .setUserName(
-                        username) // TODO - support taking username and password from url as well
-                    .setPassword(password)
+                        shard.getUserName()) // TODO - support taking username and password from url
+                    // as well
+                    .setPassword(shard.getPassword())
                     .build())
             .setJdbcDriverClassName(jdbcDriverClassName)
             .setJdbcDriverJars(jdbcDriverJars);
@@ -154,13 +132,18 @@ public static JdbcIOWrapperConfig getJdbcIOWrapperConfig(
       builder = builder.setMaxConnections(maxConnections);
     }
 
-    sourceDbURL =
+    String sourceDbURL =
         connector.getJdbcUrl(
-            sourceDbURL, host, port, dbName, connectionProperties, namespace, fetchSize);
+            shard.getHost(),
+            Integer.parseInt(shard.getPort()),
+            shard.getDbName(),
+            shard.getConnectionProperties(),
+            shard.getNamespace(),
+            fetchSize);
 
     builder.setSourceDbURL(sourceDbURL);
-    if (!StringUtils.isEmpty(shardId)) {
-      builder.setShardID(shardId);
+    if (!StringUtils.isEmpty(shard.getLogicalShardId())) {
+      builder.setShardID(shard.getLogicalShardId());
     }
 
     if (waitOn != null) {

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/SourceDbToSpannerOptions.java

@@ -68,46 +68,20 @@ public interface SourceDbToSpannerOptions extends CommonTemplateOptions {
 
   @TemplateParameter.Text(
       order = 4,
-      optional = true,
-      regexes = {"(^jdbc:mysql://.*|^jdbc:postgresql://.*|^gs://.*|^$)"},
       groupName = "Source",
-      description = "Source database connection URL or shard config path.",
+      description = "Source connection config file URL.",
       helpText =
-          "The URL to connect to the source database host. This can be either:"
-              + " 1. A JDBC connection URL for a single source database, which"
-              + " must contain the  host, port and source db name and can"
-              + " optionally contain properties like  autoReconnect,"
-              + " maxReconnects etc. Format: `jdbc:{mysql|postgresql}://{host}:{port}/{dbName}?{parameters}`."
-              + " For example,`jdbc:mysql://127.4.5.30:3306/my-db?autoReconnect=true&maxReconnects=10&unicode=true&characterEncoding=UTF-8`."
-              + " 2. A Cloud Storage path to a shard config file for sharded"
-              + " migrations. For example, `gs://my-bucket/my-shard-config.yaml`."
-              + " This parameter is required except for ASTRA_DB source.")
+          "The URL of the source connection config file. The file format is dependent on the source type."
+              + " For Astra, it will point to an Astra connection config file ([sample](src/test/resources/SourceConfig/astra-connection-config.json))."
+              + " For JDBC, it will point to a JDBC sharding config file ([sample](src/test/resources/SourceConfig/jdbc-shard-config.json))."
+              + " For Cassandra, it will point to a Cassandra driver config file ([sample](src/test/resources/SourceConfig/cassandra-driver-config.conf))."
+              + " This parameter is required.",
+      example = "gs://your-bucket/source-config.json")
   @Default.String("")
   String getSourceConfigURL();
 
   void setSourceConfigURL(String url);
 
-  @TemplateParameter.Text(
-      order = 5,
-      optional = true,
-      regexes = {"^.+$"},
-      description = "JDBC connection username.",
-      helpText = "The username to be used for the JDBC connection.")
-  @Default.String("")
-  String getUsername(); // Make optional
-
-  void setUsername(String username);
-
-  @TemplateParameter.Password(
-      order = 6,
-      optional = true,
-      description = "JDBC connection password.",
-      helpText = "The password to be used for the JDBC connection.")
-  @Default.String("")
-  String getPassword(); // make optional
-
-  void setPassword(String password);
-
   @TemplateParameter.Text(
       order = 7,
       optional = true,
@@ -252,17 +226,6 @@ public interface SourceDbToSpannerOptions extends CommonTemplateOptions {
 
   void setTransformationCustomParameters(String value);
 
-  @TemplateParameter.Text(
-      order = 20,
-      optional = true,
-      description = "Namespace",
-      helpText =
-          "Namespace to exported. For PostgreSQL, if no namespace is provided, 'public' will be used")
-  @Default.String("")
-  String getNamespace();
-
-  void setNamespace(String value);
-
   @TemplateParameter.Text(
       order = 21,
       optional = true,
@@ -365,47 +328,6 @@ public interface SourceDbToSpannerOptions extends CommonTemplateOptions {
 
   void setUniformizationStageCountHint(Long value);
 
-  @TemplateParameter.Text(
-      order = 28,
-      optional = true,
-      description = "Astra DB token",
-      helpText =
-          "AstraDB token, ignored for non-AstraDB dialects. This token is used to automatically download the securebundle by the tempalte.")
-  @Default.String("")
-  String getAstraDBToken();
-
-  void setAstraDBToken(String value);
-
-  @TemplateParameter.Text(
-      order = 29,
-      optional = true,
-      description = "Astra DB databaseID",
-      helpText = "AstraDB databaseID, ignored for non-AstraDB dialects")
-  @Default.String("")
-  String getAstraDBDatabaseId();
-
-  void setAstraDBDatabaseId(String value);
-
-  @TemplateParameter.Text(
-      order = 30,
-      optional = true,
-      description = "Astra DB keySpace",
-      helpText = "AstraDB keySpace, ignored for non-AstraDB dialects")
-  @Default.String("")
-  String getAstraDBKeySpace();
-
-  void setAstraDBKeySpace(String value);
-
-  @TemplateParameter.Text(
-      order = 31,
-      optional = true,
-      description = "Astra DB Region",
-      helpText = "AstraDB region, ignored for non-AstraDB dialects")
-  @Default.String("")
-  String getAstraDBRegion();
-
-  void setAstraDBRegion(String value);
-
   @TemplateParameter.Text(
       order = 32,
       optional = true,

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/ISrcToSpSourceConnector.java

@@ -16,6 +16,7 @@
 package com.google.cloud.teleport.v2.source;
 
 import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions;
+import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
@@ -32,5 +33,8 @@ public interface ISrcToSpSourceConnector {
    * @return The pipeline result.
    */
   PipelineResult executeMigration(
-      SourceDbToSpannerOptions options, Pipeline pipeline, SpannerConfig spannerConfig);
+      SourceDbToSpannerOptions options,
+      SourceConnectionConfig sourceConnectionConfig,
+      Pipeline pipeline,
+      SpannerConfig spannerConfig);
 }

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/CassandraSrcToSpSourceConnector.java

@@ -18,6 +18,7 @@
 import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions;
 import com.google.cloud.teleport.v2.source.ISrcToSpSourceConnector;
 import com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.iowrapper.CassandraIOWrapperFactory;
+import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig;
 import com.google.cloud.teleport.v2.templates.DbConfigContainerDefaultImpl;
 import com.google.cloud.teleport.v2.templates.PipelineController;
 import org.apache.beam.sdk.Pipeline;
@@ -29,11 +30,15 @@ public class CassandraSrcToSpSourceConnector implements ISrcToSpSourceConnector
 
   @Override
   public PipelineResult executeMigration(
-      SourceDbToSpannerOptions options, Pipeline pipeline, SpannerConfig spannerConfig) {
+      SourceDbToSpannerOptions options,
+      SourceConnectionConfig sourceConnectionConfig,
+      Pipeline pipeline,
+      SpannerConfig spannerConfig) {
     return PipelineController.executeMigrationForDbConfigContainer(
         options,
         pipeline,
         spannerConfig,
-        new DbConfigContainerDefaultImpl(CassandraIOWrapperFactory.fromPipelineOptions(options)));
+        new DbConfigContainerDefaultImpl(
+            CassandraIOWrapperFactory.fromConfig(options, sourceConnectionConfig)));
   }
 }