diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/cloudsql/CloudSqlShardOrchestrator.java b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/cloudsql/CloudSqlShardOrchestrator.java index 6e1704f104..519d32ea96 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/cloudsql/CloudSqlShardOrchestrator.java +++ b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/cloudsql/CloudSqlShardOrchestrator.java @@ -420,9 +420,7 @@ 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> entry : requestedShardMap.entrySet()) { @@ -430,28 +428,20 @@ protected String generateAndUploadConfig(String artifactName) { String ip = instanceIpMap.get(instanceName); List 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 = diff --git a/v2/sourcedb-to-spanner/README.md b/v2/sourcedb-to-spanner/README.md index 9e9883051e..1241f11fd1 100644 --- a/v2/sourcedb-to-spanner/README.md +++ b/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://:3306/",username=,password=,instanceId="",databaseId="",projectId="$PROJECT",outputDirectory=gs:// \ + --parameters sourceConfigURL="gs:///source-config.json",username=,password=,instanceId="",databaseId="",projectId="$PROJECT",outputDirectory=gs:// \ --additional-experiments=disable_runner_v2 ``` #### Replaying DLQ entries. diff --git a/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner.md b/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner.md index 445931ea4d..289e1d82ac 100644 --- a/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner.md +++ b/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 = "" databaseId = "" projectId = "" - sourceConfigURL = "jdbc:mysql://some-host:3306/sampledb" + sourceConfigURL = "gs://your-bucket/source-config.json" username = "" password = "" outputDirectory = "gs://your-bucket/dir" diff --git a/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner_Flex.md b/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner_Flex.md index b99d12ac72..5d90460ad4 100644 --- a/v2/sourcedb-to-spanner/README_Sourcedb_to_Spanner_Flex.md +++ b/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. diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/OptionsToConfigBuilder.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/OptionsToConfigBuilder.java index 46a6ab09d6..b4a4dac671 100644 --- a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/OptionsToConfigBuilder.java +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/OptionsToConfigBuilder.java @@ -24,6 +24,7 @@ import com.google.cloud.teleport.v2.reader.io.jdbc.iowrapper.config.SQLDialect; import com.google.cloud.teleport.v2.reader.io.schema.SourceSchemaReference; import com.google.cloud.teleport.v2.source.mysql.reader.io.jdbc.iowrapper.config.defaults.MySqlConfigDefaults; +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; @@ -63,15 +64,11 @@ public static String extractWorkerZone(PipelineOptions options) { public static JdbcIOWrapperConfig getJdbcIOWrapperConfigWithDefaults( SourceDbToSpannerOptions options, + Shard shard, List tables, String shardId, 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(); @@ -88,14 +85,13 @@ public static JdbcIOWrapperConfig getJdbcIOWrapperConfigWithDefaults( return getJdbcIOWrapperConfig( sqlDialect, tables, - sourceDbURL, - null, - null, - 0, - username, - password, - dbName, - namespace, + shard.getHost(), + shard.getConnectionProperties(), + Integer.parseInt(shard.getPort()), + shard.getUserName(), + shard.getPassword(), + shard.getDbName(), + shard.getNamespace(), shardId, jdbcDriverClassName, jdbcDriverJars, @@ -112,7 +108,6 @@ public static JdbcIOWrapperConfig getJdbcIOWrapperConfigWithDefaults( public static JdbcIOWrapperConfig getJdbcIOWrapperConfig( SQLDialect sqlDialect, List tables, - String sourceDbURL, String host, String connectionProperties, int port, @@ -157,13 +152,12 @@ public static JdbcIOWrapperConfig getJdbcIOWrapperConfig( builder = builder.setMaxConnections(maxConnections); } + String sourceDbURL = ""; switch (sqlDialect) { case MYSQL: - if (sourceDbURL == null) { - sourceDbURL = "jdbc:mysql://" + host + ":" + port + "/" + dbName; - if (StringUtils.isNotBlank(connectionProperties)) { - sourceDbURL = sourceDbURL + "?" + connectionProperties; - } + sourceDbURL = "jdbc:mysql://" + host + ":" + port + "/" + dbName; + if (StringUtils.isNotBlank(connectionProperties)) { + sourceDbURL = sourceDbURL + "?" + connectionProperties; } for (Entry entry : MySqlConfigDefaults.DEFAULT_MYSQL_URL_PROPERTIES.entrySet()) { @@ -172,14 +166,14 @@ public static JdbcIOWrapperConfig getJdbcIOWrapperConfig( sourceDbURL = mysqlSetCursorModeIfNeeded(sqlDialect, sourceDbURL, fetchSize); break; case POSTGRESQL: - if (sourceDbURL == null) { - sourceDbURL = "jdbc:postgresql://" + host + ":" + port + "/" + dbName; - } + sourceDbURL = "jdbc:postgresql://" + host + ":" + port + "/" + dbName; sourceDbURL = sourceDbURL + "?currentSchema=" + sourceSchemaReference.jdbc().namespace(); if (StringUtils.isNotBlank(connectionProperties)) { sourceDbURL = sourceDbURL + "&" + connectionProperties; } break; + default: + throw new IllegalArgumentException("Unsupported SQL Dialect: " + sqlDialect); } builder.setSourceDbURL(sourceDbURL); diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/SourceDbToSpannerOptions.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/SourceDbToSpannerOptions.java index 58a2ba7e6d..61150db087 100644 --- a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/SourceDbToSpannerOptions.java +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/SourceDbToSpannerOptions.java @@ -69,19 +69,15 @@ public interface SourceDbToSpannerOptions extends CommonTemplateOptions { @TemplateParameter.Text( order = 4, optional = true, - regexes = {"(^jdbc:mysql://.*|^jdbc:postgresql://.*|^gs://.*|^$)"}, + regexes = {"(^.+$)"}, 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.") @Default.String("") String getSourceConfigURL(); @@ -252,17 +248,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, diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperFactory.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperFactory.java index 1f843fdd10..78e93535fd 100644 --- a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperFactory.java +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperFactory.java @@ -15,23 +15,27 @@ */ package com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.iowrapper; +import com.datastax.oss.driver.api.core.config.OptionsMap; import com.google.auto.value.AutoValue; import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions; import com.google.cloud.teleport.v2.reader.IoWrapperFactory; import com.google.cloud.teleport.v2.reader.auth.dbauth.GuardedStringValueProvider; import com.google.cloud.teleport.v2.reader.io.IoWrapper; import com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.iowrapper.CassandraDataSource.CassandraDialect; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.AstraConnectionConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.CassandraConnectionConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig; import com.google.common.base.Preconditions; import java.util.List; import javax.annotation.Nullable; import org.apache.beam.sdk.transforms.Wait.OnSignal; -import org.apache.commons.lang3.StringUtils; @AutoValue public abstract class CassandraIOWrapperFactory implements IoWrapperFactory { - /** GCS Path for Cassandra Driver Config. */ - public abstract String gcsConfigPath(); + /** Options Map for Cassandra Driver Config. */ + @Nullable + public abstract OptionsMap optionsMap(); /** * Number of partitions to read from. Defaults to Null. @@ -44,7 +48,6 @@ public abstract class CassandraIOWrapperFactory implements IoWrapperFactory { /** Cassandra Dialect. */ public abstract CassandraDataSource.CassandraDialect cassandraDialect(); - /** Astra DB options. Empty for OSS dialect. */ /** Astra DB Token. * */ public abstract GuardedStringValueProvider astraDBToken(); @@ -54,11 +57,11 @@ public abstract class CassandraIOWrapperFactory implements IoWrapperFactory { /** Astra DB Keyspace. * */ public abstract String astraDBKeyspace(); - /** Astra DB Keyspace. * */ + /** Astra DB Region. * */ public abstract String astraDBRegion(); private static CassandraIOWrapperFactory create( - String gcsConfigPath, + OptionsMap optionsMap, Integer numPartions, String sourceDialect, GuardedStringValueProvider astraDBToken, @@ -71,7 +74,7 @@ private static CassandraIOWrapperFactory create( default -> CassandraDialect.OSS; }; return new AutoValue_CassandraIOWrapperFactory( - gcsConfigPath, + optionsMap, numPartions, cassandraDialect, astraDBToken, @@ -80,27 +83,42 @@ private static CassandraIOWrapperFactory create( astraDBRegion); } - public static CassandraIOWrapperFactory fromPipelineOptions(SourceDbToSpannerOptions options) { - String gcsPath = options.getSourceConfigURL(); - // Implementation Details. the pipeline options are strings. + public static CassandraIOWrapperFactory fromConfig( + SourceDbToSpannerOptions options, SourceConnectionConfig sourceConnectionConfig) { Preconditions.checkArgument( options.getSourceDbDialect().equals(SourceDbToSpannerOptions.CASSANDRA_SOURCE_DIALECT) || options .getSourceDbDialect() .equals(SourceDbToSpannerOptions.ASTRA_DB_SOURCE_DIALECT), "Unexpected Dialect " + options.getSourceDbDialect() + " for Cassandra Source"); - Preconditions.checkArgument( - options.getSourceDbDialect().equals(SourceDbToSpannerOptions.ASTRA_DB_SOURCE_DIALECT) - || StringUtils.startsWith(gcsPath, "gs://"), - "GCS path Expected in place of `" + gcsPath + "`."); + + GuardedStringValueProvider astraDBToken = GuardedStringValueProvider.create(""); + String astraDBDatabaseId = ""; + String astraDBKeyspace = ""; + String astraDBRegion = ""; + OptionsMap optionsMap = null; + + if (sourceConnectionConfig instanceof AstraConnectionConfig) { + AstraConnectionConfig astraConfig = (AstraConnectionConfig) sourceConnectionConfig; + astraDBToken = GuardedStringValueProvider.create(astraConfig.getAstraToken()); + astraDBDatabaseId = astraConfig.getDatabaseId(); + astraDBKeyspace = astraConfig.getKeySpace(); + astraDBRegion = astraConfig.getAstraDbRegion(); + } else if (sourceConnectionConfig instanceof CassandraConnectionConfig) { + optionsMap = ((CassandraConnectionConfig) sourceConnectionConfig).getOptionsMap(); + } else { + throw new IllegalArgumentException( + "Unsupported source connection config type: " + sourceConnectionConfig); + } + return CassandraIOWrapperFactory.create( - options.getSourceConfigURL(), + optionsMap, options.getNumPartitions(), options.getSourceDbDialect(), - GuardedStringValueProvider.create(options.getAstraDBToken()), - options.getAstraDBDatabaseId(), - options.getAstraDBKeySpace(), - options.getAstraDBRegion()); + astraDBToken, + astraDBDatabaseId, + astraDBKeyspace, + astraDBRegion); } /** Create an {@link IoWrapper} instance for a list of SourceTables. */ @@ -108,7 +126,7 @@ public static CassandraIOWrapperFactory fromPipelineOptions(SourceDbToSpannerOpt public IoWrapper getIOWrapper(List sourceTables, OnSignal waitOnSignal) { /** TODO(vardhanvthigle@) incorporate waitOnSignal */ return new CassandraIoWrapper( - gcsConfigPath(), + optionsMap(), sourceTables, numPartitions(), cassandraDialect(), diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperHelper.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperHelper.java index 23100cf9d7..108001473e 100644 --- a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperHelper.java +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperHelper.java @@ -17,9 +17,9 @@ import static com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.iowrapper.CassandraDefaults.DEFAULT_CASSANDRA_SCHEMA_DISCOVERY_BACKOFF; +import com.datastax.oss.driver.api.core.config.OptionsMap; import com.google.cloud.teleport.v2.reader.auth.dbauth.GuardedStringValueProvider; import com.google.cloud.teleport.v2.reader.io.datasource.DataSource; -import com.google.cloud.teleport.v2.reader.io.exception.SchemaDiscoveryException; import com.google.cloud.teleport.v2.reader.io.row.SourceRow; import com.google.cloud.teleport.v2.reader.io.schema.SchemaDiscovery; import com.google.cloud.teleport.v2.reader.io.schema.SchemaDiscoveryImpl; @@ -33,7 +33,6 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; -import java.io.FileNotFoundException; import java.util.List; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.values.PBegin; @@ -47,7 +46,7 @@ class CassandraIOWrapperHelper { private static final Logger LOG = LoggerFactory.getLogger(CassandraIOWrapperHelper.class); static DataSource buildDataSource( - String gcsPath, + OptionsMap optionsMap, Integer numPartitions, CassandraDataSource.CassandraDialect cassandraDialect, GuardedStringValueProvider astraDBToken, @@ -55,30 +54,25 @@ static DataSource buildDataSource( String astraDBKeyspace, String astraDBRegion) { DataSource dataSource; - try { - dataSource = - switch (cassandraDialect) { - case ASTRA -> DataSource.ofCassandra( - // TODO: Astra: Build from Pipeline Options. - CassandraDataSource.ofAstra( - AstraDbDataSource.builder() - .setAstraToken(astraDBToken) - .setDatabaseId(astraDBDatabaseId) - .setKeySpace(astraDBKeyspace) - .setAstraDbRegion(astraDBRegion) - .build())); + dataSource = + switch (cassandraDialect) { + case ASTRA -> DataSource.ofCassandra( + // TODO: Astra: Build from Pipeline Options. + CassandraDataSource.ofAstra( + AstraDbDataSource.builder() + .setAstraToken(astraDBToken) + .setDatabaseId(astraDBDatabaseId) + .setKeySpace(astraDBKeyspace) + .setAstraDbRegion(astraDBRegion) + .build())); - default -> DataSource.ofCassandra( - CassandraDataSource.ofOss( - CassandraDataSourceOss.builder() - .setOptionsMapFromGcsFile(gcsPath) - .setNumPartitions(numPartitions) - .build())); - }; - } catch (FileNotFoundException e) { - LOG.error("Unable to find driver config file in {}. Cause ", gcsPath, e); - throw (new SchemaDiscoveryException(e)); - } + default -> DataSource.ofCassandra( + CassandraDataSource.ofOss( + CassandraDataSourceOss.builder() + .setOptionsMap(optionsMap) + .setNumPartitions(numPartitions) + .build())); + }; return dataSource; } diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIoWrapper.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIoWrapper.java index 95ffa176d8..5881f56e69 100644 --- a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIoWrapper.java +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIoWrapper.java @@ -15,6 +15,7 @@ */ package com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.iowrapper; +import com.datastax.oss.driver.api.core.config.OptionsMap; import com.google.cloud.teleport.v2.reader.auth.dbauth.GuardedStringValueProvider; import com.google.cloud.teleport.v2.reader.io.IoWrapper; import com.google.cloud.teleport.v2.reader.io.datasource.DataSource; @@ -41,7 +42,7 @@ public final class CassandraIoWrapper implements IoWrapper { tableReaders; public CassandraIoWrapper( - String gcsPath, + OptionsMap optionsMap, List sourceTables, @Nullable Integer numPartitions, CassandraDialect cassandraDialect, @@ -51,7 +52,7 @@ public CassandraIoWrapper( String astraDBRegion) { DataSource dataSource = CassandraIOWrapperHelper.buildDataSource( - gcsPath, + optionsMap, numPartitions, cassandraDialect, astraDBToken, diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/PipelineController.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/PipelineController.java index 22f763355e..68c3a7df37 100644 --- a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/PipelineController.java +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/PipelineController.java @@ -31,13 +31,17 @@ import com.google.cloud.teleport.v2.spanner.migrations.schema.SchemaStringOverridesBasedMapper; import com.google.cloud.teleport.v2.spanner.migrations.schema.SessionBasedMapper; import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConfigParser; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig; import com.google.cloud.teleport.v2.spanner.migrations.spanner.SpannerSchema; +import com.google.cloud.teleport.v2.spanner.migrations.utils.ISecretManagerAccessor; +import com.google.cloud.teleport.v2.spanner.migrations.utils.SecretManagerAccessorImpl; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableList; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Optional; import java.util.stream.Collectors; import org.apache.beam.repackaged.core.org.apache.commons.lang3.StringUtils; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; @@ -86,12 +90,31 @@ protected static PipelineResult executeSingleInstanceMigrationForDbConfigContain } static PipelineResult executeJdbcSingleInstanceMigration( - SourceDbToSpannerOptions options, Pipeline pipeline, SpannerConfig spannerConfig) { - JdbcDbConfigContainer jdbcDbConfigContainer = new SingleInstanceJdbcDbConfigContainer(options); + SourceDbToSpannerOptions options, + Shard shard, + Pipeline pipeline, + SpannerConfig spannerConfig) { + JdbcDbConfigContainer jdbcDbConfigContainer = + new SingleInstanceJdbcDbConfigContainer(options, shard); return executeSingleInstanceMigrationForDbConfigContainer( options, pipeline, spannerConfig, jdbcDbConfigContainer); } + static PipelineResult executeJdbcMigration( + SourceDbToSpannerOptions options, + JdbcShardConfig jdbcShardConfig, + Pipeline pipeline, + SpannerConfig spannerConfig) { + if (jdbcShardConfig.getShardConfigs().size() > 1) { + List shards = jdbcShardConfig.getShardConfigs(); + return PipelineController.executeJdbcShardedMigration( + options, pipeline, shards, spannerConfig); + } else { + return PipelineController.executeJdbcSingleInstanceMigration( + options, jdbcShardConfig.getShardConfigs().get(0), pipeline, spannerConfig); + } + } + static PipelineResult executeJdbcShardedMigration( SourceDbToSpannerOptions options, Pipeline pipeline, @@ -128,12 +151,16 @@ static PipelineResult executeJdbcShardedMigration( } static PipelineResult executeCassandraMigration( - SourceDbToSpannerOptions options, Pipeline pipeline, SpannerConfig spannerConfig) { + SourceDbToSpannerOptions options, + SourceConnectionConfig sourceConnectionConfig, + Pipeline pipeline, + SpannerConfig spannerConfig) { return executeSingleInstanceMigrationForDbConfigContainer( options, pipeline, spannerConfig, - new DbConfigContainerDefaultImpl(CassandraIOWrapperFactory.fromPipelineOptions(options))); + new DbConfigContainerDefaultImpl( + CassandraIOWrapperFactory.fromConfig(options, sourceConnectionConfig))); } @VisibleForTesting @@ -346,16 +373,12 @@ public JdbcIoWrapperConfigGroup getJdbcIoWrapperConfigGroup( // Read data from source String shardId = entry.getValue(); - // If a namespace is configured for a shard uses that, otherwise uses the namespace - // configured in the options if there is one. - String namespace = - Optional.ofNullable(shard.getNamespace()).orElse(options.getNamespace()); + String namespace = shard.getNamespace(); String dbName = entry.getKey(); JdbcIOWrapperConfig shardConfig = OptionsToConfigBuilder.getJdbcIOWrapperConfig( sqlDialect, sourceTables, - null, shard.getHost(), shard.getConnectionProperties(), Integer.parseInt(shard.getPort()), @@ -382,10 +405,12 @@ public JdbcIoWrapperConfigGroup getJdbcIoWrapperConfigGroup( } static class SingleInstanceJdbcDbConfigContainer implements JdbcDbConfigContainer { - private SourceDbToSpannerOptions options; + private final SourceDbToSpannerOptions options; + private final Shard shard; - public SingleInstanceJdbcDbConfigContainer(SourceDbToSpannerOptions options) { + public SingleInstanceJdbcDbConfigContainer(SourceDbToSpannerOptions options, Shard shard) { this.options = options; + this.shard = shard; } @Override @@ -394,8 +419,24 @@ public JdbcIoWrapperConfigGroup getJdbcIoWrapperConfigGroup( return JdbcIoWrapperConfigGroup.builder() .addShardConfig( OptionsToConfigBuilder.getJdbcIOWrapperConfigWithDefaults( - options, sourceTables, null, waitOnSignal)) + options, shard, sourceTables, null, waitOnSignal)) .build(); } } + + public static SourceConnectionConfig getSourceConnectionConfig( + String sourceType, String sourceShardsFilePath) { + ISecretManagerAccessor secretManagerAccessor = new SecretManagerAccessorImpl(); + SourceConfigParser sourceConfigParser = new SourceConfigParser(secretManagerAccessor); + SourceConnectionConfig sourceConnectionConfig; + try { + // Parse the source shards configuration file to respective + // SourceConnectionConfig. + LOG.info("Parsing source shards configuration file: {}", sourceShardsFilePath); + return sourceConfigParser.parseConfiguration(sourceType, sourceShardsFilePath); + } catch (Exception e) { + LOG.error("Error parsing source config", e); + throw new RuntimeException("Error parsing source config", e); + } + } } diff --git a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/SourceDbToSpanner.java b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/SourceDbToSpanner.java index cb309aa4d8..42f20470f9 100644 --- a/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/SourceDbToSpanner.java +++ b/v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/SourceDbToSpanner.java @@ -20,20 +20,19 @@ import com.google.cloud.teleport.v2.common.CommonTemplateJvmInitializer; import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger; import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions; -import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.AstraConnectionConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.CassandraConnectionConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig; import com.google.cloud.teleport.v2.spanner.migrations.utils.DataflowWorkerMachineTypeUtils; -import com.google.cloud.teleport.v2.spanner.migrations.utils.SecretManagerAccessorImpl; -import com.google.cloud.teleport.v2.spanner.migrations.utils.ShardFileReader; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; -import java.util.List; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult; import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.options.ValueProvider; -import org.apache.commons.lang3.StringUtils; /** * A template that copies data from a relational database using JDBC to an existing Spanner @@ -107,40 +106,33 @@ static PipelineResult run(SourceDbToSpannerOptions options) { DataflowWorkerMachineTypeUtils.validateMachineSpecs(workerMachineType, 4); SpannerConfig spannerConfig = createSpannerConfig(options); + SourceConnectionConfig sourceConnectionConfig = + PipelineController.getSourceConnectionConfig( + options.getSourceDbDialect(), options.getSourceConfigURL()); // Decide type and source of migration // TODO(vardhanvthigle): Move this within pipelineController. switch (options.getSourceDbDialect()) { case SourceDbToSpannerOptions.CASSANDRA_SOURCE_DIALECT: Preconditions.checkArgument( - StringUtils.isNotEmpty(options.getSourceConfigURL()), - "Cassandra Dialect needs sourceConfigURL to be set."); - return PipelineController.executeCassandraMigration(options, pipeline, spannerConfig); + (sourceConnectionConfig instanceof CassandraConnectionConfig), + "Source config is not type of CassandraConnectionConfig."); + return PipelineController.executeCassandraMigration( + options, sourceConnectionConfig, pipeline, spannerConfig); case SourceDbToSpannerOptions.ASTRA_DB_SOURCE_DIALECT: - return PipelineController.executeCassandraMigration(options, pipeline, spannerConfig); - + Preconditions.checkArgument( + (sourceConnectionConfig instanceof AstraConnectionConfig), + "Source config is not type of AstraConnectionConfig."); + return PipelineController.executeCassandraMigration( + options, sourceConnectionConfig, pipeline, spannerConfig); default: /* Implementation detail, not having a default leads to failure in compile time checks enforced here */ /* Making jdbc as default case which includes MYSQL and PG. */ Preconditions.checkArgument( - StringUtils.isNotEmpty(options.getSourceConfigURL()), - "JDBC based source needs sourceConfigURL to be set."); - return executeJdbcMigration(options, pipeline, spannerConfig); - } - } - - // TODO(vardhanvthigle): Move this within pipelineController. - private static PipelineResult executeJdbcMigration( - SourceDbToSpannerOptions options, Pipeline pipeline, SpannerConfig spannerConfig) { - if (options.getSourceConfigURL().startsWith("gs://")) { - List shards = - new ShardFileReader(new SecretManagerAccessorImpl()) - .readForwardMigrationShardingConfig(options.getSourceConfigURL()); - return PipelineController.executeJdbcShardedMigration( - options, pipeline, shards, spannerConfig); - } else { - return PipelineController.executeJdbcSingleInstanceMigration( - options, pipeline, spannerConfig); + (sourceConnectionConfig instanceof JdbcShardConfig), + "Source config is not type of JdbcShardConfig."); + return PipelineController.executeJdbcMigration( + options, (JdbcShardConfig) sourceConnectionConfig, pipeline, spannerConfig); } } diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/options/OptionsToConfigBuilderTest.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/options/OptionsToConfigBuilderTest.java index a035ea9b4c..fa546dcf64 100644 --- a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/options/OptionsToConfigBuilderTest.java +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/options/OptionsToConfigBuilderTest.java @@ -20,8 +20,8 @@ import com.google.cloud.teleport.v2.reader.io.jdbc.iowrapper.config.JdbcIOWrapperConfig; import com.google.cloud.teleport.v2.reader.io.jdbc.iowrapper.config.SQLDialect; +import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; import java.net.URISyntaxException; -import java.util.ArrayList; import java.util.List; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; @@ -44,37 +44,38 @@ public class OptionsToConfigBuilderTest { public void testConfigWithMySqlDefaultsFromOptions() { final String testDriverClassName = "org.apache.derby.jdbc.EmbeddedDriver"; final String testUrl = "jdbc:mysql://localhost:3306/testDB"; - final String testUser = "user"; - final String testPassword = "password"; SourceDbToSpannerOptions sourceDbToSpannerOptions = PipelineOptionsFactory.as(SourceDbToSpannerOptions.class); sourceDbToSpannerOptions.setSourceDbDialect(SQLDialect.MYSQL.name()); - sourceDbToSpannerOptions.setSourceConfigURL(testUrl); sourceDbToSpannerOptions.setJdbcDriverClassName(testDriverClassName); sourceDbToSpannerOptions.setMaxConnections(150); sourceDbToSpannerOptions.setNumPartitions(4000); - sourceDbToSpannerOptions.setUsername(testUser); - sourceDbToSpannerOptions.setPassword(testPassword); sourceDbToSpannerOptions.setTables("table1,table2"); + Shard shard = new Shard("", "localhost", "3306", "user", "password", "testDB", "", "", ""); PCollection dummyPCollection = pipeline.apply(Create.of(1)); pipeline.run(); JdbcIOWrapperConfig config = OptionsToConfigBuilder.getJdbcIOWrapperConfigWithDefaults( - sourceDbToSpannerOptions, List.of("table1", "table2"), null, Wait.on(dummyPCollection)); + sourceDbToSpannerOptions, + shard, + List.of("table1", "table2"), + null, + Wait.on(dummyPCollection)); assertThat(config.jdbcDriverClassName()).isEqualTo(testDriverClassName); assertThat(config.sourceDbURL()) .isEqualTo( testUrl + "?allowMultiQueries=true&autoReconnect=true&maxReconnects=10&useCursorFetch=true"); assertThat(config.tables()).containsExactlyElementsIn(new String[] {"table1", "table2"}); - assertThat(config.dbAuth().getUserName().get()).isEqualTo(testUser); - assertThat(config.dbAuth().getPassword().get()).isEqualTo(testPassword); + assertThat(config.dbAuth().getUserName().get()).isEqualTo(shard.getUserName()); + assertThat(config.dbAuth().getPassword().get()).isEqualTo(shard.getPassword()); assertThat(config.waitOn()).isNotNull(); assertThat(config.maxFetchSize()).isNull(); sourceDbToSpannerOptions.setFetchSize(42); assertThat( OptionsToConfigBuilder.getJdbcIOWrapperConfigWithDefaults( sourceDbToSpannerOptions, + shard, List.of("table1", "table2"), null, Wait.on(dummyPCollection)) @@ -90,7 +91,6 @@ public void testConfigWithMySqlUrlFromOptions() { OptionsToConfigBuilder.getJdbcIOWrapperConfig( SQLDialect.MYSQL, List.of("table1", "table2"), - null, "myhost", "testParam=testValue", 3306, @@ -114,7 +114,6 @@ public void testConfigWithMySqlUrlFromOptions() { OptionsToConfigBuilder.getJdbcIOWrapperConfig( SQLDialect.MYSQL, List.of("table1", "table2"), - null, "myhost", null, 3306, @@ -151,18 +150,17 @@ public void testConfigWithPostgreSQLDefaultsFromOptions() { SourceDbToSpannerOptions sourceDbToSpannerOptions = PipelineOptionsFactory.as(SourceDbToSpannerOptions.class); sourceDbToSpannerOptions.setSourceDbDialect(SQLDialect.POSTGRESQL.name()); - sourceDbToSpannerOptions.setSourceConfigURL(testUrl); sourceDbToSpannerOptions.setJdbcDriverClassName(testDriverClassName); sourceDbToSpannerOptions.setMaxConnections(150); sourceDbToSpannerOptions.setNumPartitions(4000); - sourceDbToSpannerOptions.setUsername(testUser); - sourceDbToSpannerOptions.setPassword(testPassword); sourceDbToSpannerOptions.setTables("table1,table2,table3"); + Shard shard = new Shard("", "localhost", "5432", "user", "password", "testDB", "", "", ""); PCollection dummyPCollection = pipeline.apply(Create.of(1)); pipeline.run(); JdbcIOWrapperConfig config = OptionsToConfigBuilder.getJdbcIOWrapperConfigWithDefaults( sourceDbToSpannerOptions, + shard, List.of("table1", "table2", "table3"), null, Wait.on(dummyPCollection)); @@ -183,7 +181,6 @@ public void testConfigWithPostgreSqlUrlFromOptions() { OptionsToConfigBuilder.getJdbcIOWrapperConfig( SQLDialect.POSTGRESQL, List.of("table1", "table2"), - null, "myhost", "testParam=testValue", 5432, @@ -206,7 +203,6 @@ public void testConfigWithPostgreSqlUrlFromOptions() { OptionsToConfigBuilder.getJdbcIOWrapperConfig( SQLDialect.POSTGRESQL, List.of("table1", "table2"), - null, "myhost", "", 5432, @@ -241,7 +237,6 @@ public void testConfigWithPostgreSqlUrlWithNamespace() { OptionsToConfigBuilder.getJdbcIOWrapperConfig( SQLDialect.POSTGRESQL, List.of("table1", "table2"), - null, "myhost", "", 5432, @@ -264,20 +259,6 @@ public void testConfigWithPostgreSqlUrlWithNamespace() { .isEqualTo("jdbc:postgresql://myhost:5432/mydb?currentSchema=mynamespace"); } - @Test - public void testURIParsingException() { - final String testUrl = "jd#bc://localhost"; - SourceDbToSpannerOptions sourceDbToSpannerOptions = - PipelineOptionsFactory.as(SourceDbToSpannerOptions.class); - sourceDbToSpannerOptions.setSourceDbDialect(SQLDialect.MYSQL.name()); - sourceDbToSpannerOptions.setSourceConfigURL(testUrl); - assertThrows( - RuntimeException.class, - () -> - OptionsToConfigBuilder.getJdbcIOWrapperConfigWithDefaults( - sourceDbToSpannerOptions, new ArrayList<>(), null, null)); - } - @Test public void testaddParamToJdbcUrl() throws URISyntaxException { // No Parameters initially. @@ -356,13 +337,13 @@ public void testExtractWorkerZoneException() { public void testFetchSizeMinusOneBehavesLikeNull() { SourceDbToSpannerOptions options = PipelineOptionsFactory.as(SourceDbToSpannerOptions.class); options.setSourceDbDialect(SQLDialect.MYSQL.name()); - options.setSourceConfigURL("jdbc:mysql://localhost:3306/testDB"); options.setJdbcDriverClassName("com.mysql.jdbc.Driver"); options.setFetchSize(-1); // Should be normalized to null + Shard shard = new Shard("", "localhost", "5432", "user", "password", "testDB", "", "", ""); JdbcIOWrapperConfig config = OptionsToConfigBuilder.getJdbcIOWrapperConfigWithDefaults( - options, List.of("table1"), null, null); + options, shard, List.of("table1"), null, null); assertThat(config.maxFetchSize()).isNull(); } diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperFactoryTest.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperFactoryTest.java index d173b32776..707aac6380 100644 --- a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperFactoryTest.java +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperFactoryTest.java @@ -36,6 +36,9 @@ import com.google.cloud.teleport.v2.reader.io.schema.SourceTableReference; import com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.iowrapper.CassandraDataSource.CassandraDialect; import com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.schema.CassandraSchemaReference; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.AstraConnectionConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.CassandraConnectionConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import java.net.InetSocketAddress; @@ -96,7 +99,7 @@ public void setup() { .when( () -> CassandraIOWrapperHelper.buildDataSource( - TEST_BUCKET_CASSANDRA_CONFIG_CONF, + null, null, CassandraDialect.OSS, GuardedStringValueProvider.create(""), @@ -132,19 +135,15 @@ public void cleanup() { @Test public void testCassandraIoWrapperFactoryOssBasic() { - String testConfigPath = TEST_BUCKET_CASSANDRA_CONFIG_CONF; SourceDbToSpannerOptions mockOptions = mock(SourceDbToSpannerOptions.class, Mockito.withSettings().serializable()); when(mockOptions.getSourceDbDialect()).thenReturn("CASSANDRA"); - when(mockOptions.getSourceConfigURL()).thenReturn(testConfigPath); when(mockOptions.getNumPartitions()).thenReturn(null); - when(mockOptions.getAstraDBToken()).thenReturn(""); - when(mockOptions.getAstraDBDatabaseId()).thenReturn(""); - when(mockOptions.getAstraDBRegion()).thenReturn(""); - when(mockOptions.getAstraDBKeySpace()).thenReturn(""); + CassandraConnectionConfig mockSourceConfig = mock(CassandraConnectionConfig.class); + when(mockSourceConfig.getOptionsMap()).thenReturn(null); CassandraIOWrapperFactory cassandraIOWrapperFactory = - CassandraIOWrapperFactory.fromPipelineOptions(mockOptions); - assertThat(cassandraIOWrapperFactory.gcsConfigPath()).isEqualTo(testConfigPath); + CassandraIOWrapperFactory.fromConfig(mockOptions, mockSourceConfig); + assertThat(cassandraIOWrapperFactory.optionsMap()).isEqualTo(null); assertThat(cassandraIOWrapperFactory.getIOWrapper(TABLES_TO_READ, null).discoverTableSchema()) .isEqualTo(ImmutableList.of(mockSourceSchema)); assertThat(cassandraIOWrapperFactory.cassandraDialect()).isEqualTo(CassandraDialect.OSS); @@ -157,19 +156,20 @@ public void testCassandraIoWrapperFactoryOssBasic() { @Test public void testCassandraIoWrapperFactoryAstraBasic() { - String testConfigPath = ""; SourceDbToSpannerOptions mockOptions = mock(SourceDbToSpannerOptions.class, Mockito.withSettings().serializable()); when(mockOptions.getSourceDbDialect()).thenReturn("ASTRA_DB"); - when(mockOptions.getSourceConfigURL()).thenReturn(testConfigPath); when(mockOptions.getNumPartitions()).thenReturn(null); - when(mockOptions.getAstraDBToken()).thenReturn("AstraCS:testToken"); - when(mockOptions.getAstraDBDatabaseId()).thenReturn("testId"); - when(mockOptions.getAstraDBRegion()).thenReturn("testRegion"); - when(mockOptions.getAstraDBKeySpace()).thenReturn("testKeyspace"); + + AstraConnectionConfig mockSourceConfig = mock(AstraConnectionConfig.class); + when(mockSourceConfig.getAstraToken()).thenReturn("AstraCS:testToken"); + when(mockSourceConfig.getDatabaseId()).thenReturn("testId"); + when(mockSourceConfig.getAstraDbRegion()).thenReturn("testRegion"); + when(mockSourceConfig.getKeySpace()).thenReturn("testKeyspace"); + CassandraIOWrapperFactory cassandraIOWrapperFactory = - CassandraIOWrapperFactory.fromPipelineOptions(mockOptions); - assertThat(cassandraIOWrapperFactory.gcsConfigPath()).isEqualTo(testConfigPath); + CassandraIOWrapperFactory.fromConfig(mockOptions, mockSourceConfig); + assertThat(cassandraIOWrapperFactory.optionsMap()).isEqualTo(null); assertThat(cassandraIOWrapperFactory.cassandraDialect()).isEqualTo(CassandraDialect.ASTRA); assertThat(cassandraIOWrapperFactory.astraDBKeyspace()).isEqualTo("testKeyspace"); assertThat(cassandraIOWrapperFactory.astraDBRegion()).isEqualTo("testRegion"); @@ -180,16 +180,12 @@ public void testCassandraIoWrapperFactoryAstraBasic() { @Test public void testCassandraIoWrapperFactoryExceptions() { - String testConfigPath = "smt-test-bucket/test-conf.conf"; SourceDbToSpannerOptions mockOptions = mock(SourceDbToSpannerOptions.class, Mockito.withSettings().serializable()); - when(mockOptions.getSourceDbDialect()).thenReturn("MYSQL").thenReturn("CASSANDRA"); - when(mockOptions.getSourceConfigURL()).thenReturn(testConfigPath); - assertThrows( - IllegalArgumentException.class, - () -> CassandraIOWrapperFactory.fromPipelineOptions(mockOptions)); + when(mockOptions.getSourceDbDialect()).thenReturn("MYSQL"); + SourceConnectionConfig mockConfig = mock(SourceConnectionConfig.class); assertThrows( IllegalArgumentException.class, - () -> CassandraIOWrapperFactory.fromPipelineOptions(mockOptions)); + () -> CassandraIOWrapperFactory.fromConfig(mockOptions, mockConfig)); } } diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperHelperTest.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperHelperTest.java index 714951ce30..fbae2ea60c 100644 --- a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperHelperTest.java +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIOWrapperHelperTest.java @@ -37,15 +37,12 @@ import static com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.testutils.BasicTestSchema.TEST_KEYSPACE; import static com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.testutils.BasicTestSchema.TEST_TABLES; import static com.google.common.truth.Truth.assertThat; -import static org.junit.Assert.assertThrows; -import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockStatic; import com.datastax.oss.driver.api.core.config.OptionsMap; import com.datastax.oss.driver.api.core.config.TypedDriverOption; import com.google.cloud.teleport.v2.reader.auth.dbauth.GuardedStringValueProvider; import com.google.cloud.teleport.v2.reader.io.datasource.DataSource; -import com.google.cloud.teleport.v2.reader.io.exception.SchemaDiscoveryException; import com.google.cloud.teleport.v2.reader.io.row.SourceRow; import com.google.cloud.teleport.v2.reader.io.schema.SourceSchema; import com.google.cloud.teleport.v2.reader.io.schema.SourceSchemaReference; @@ -54,6 +51,7 @@ import com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.schema.CassandraSchemaDiscovery; import com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.schema.CassandraSchemaReference; import com.google.cloud.teleport.v2.source.cassandra.reader.io.cassandra.testutils.SharedEmbeddedCassandra; +import com.google.cloud.teleport.v2.spanner.migrations.utils.CassandraDriverConfigLoader; import com.google.cloud.teleport.v2.spanner.migrations.utils.JarFileReader; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; @@ -62,7 +60,6 @@ import java.net.URL; import java.util.List; import java.util.stream.Collectors; -import org.apache.beam.sdk.io.cassandra.CassandraIO; import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.values.PBegin; @@ -96,58 +93,43 @@ public static void stopEmbeddedCassandra() throws Exception { } @Test - public void testBuildDataSource() { - + public void testBuildDataSource() throws Exception { String testGcsPath = "gs://smt-test-bucket/cassandraConfig.conf"; URL testUrl = Resources.getResource("CassandraUT/test-cassandra-config.conf"); - CassandraIO.Read mockCassandraIORead = mock(CassandraIO.Read.class); + OptionsMap optionsMap; try (MockedStatic mockFileReader = mockStatic(JarFileReader.class)) { - mockFileReader .when(() -> JarFileReader.saveFilesLocally(testGcsPath)) - .thenReturn(new URL[] {testUrl}) - .thenReturn(new URL[] {testUrl}) - /* Empty URL List to test FileNotFoundException handling. */ - .thenReturn(new URL[] {}); - - DataSource dataSource = - CassandraIOWrapperHelper.buildDataSource( - testGcsPath, - null, - CassandraDialect.OSS, - GuardedStringValueProvider.create(""), - "", - "", - ""); - assertThat(dataSource.cassandra().oss().loggedKeySpace()).isEqualTo("test-keyspace"); - assertThat(dataSource.cassandra().oss().localDataCenter()).isEqualTo("datacenter1"); - assertThat(dataSource.cassandra().oss().numPartitions()).isEqualTo(null); - assertThat( - CassandraIOWrapperHelper.buildDataSource( - testGcsPath, - 42, - CassandraDialect.OSS, - GuardedStringValueProvider.create(""), - "", - "", - "") - .cassandra() - .oss() - .numPartitions()) - .isEqualTo(42); - assertThrows( - SchemaDiscoveryException.class, - () -> - CassandraIOWrapperHelper.buildDataSource( - testGcsPath, - null, - CassandraDialect.OSS, - GuardedStringValueProvider.create(""), - "", - "", - "")); + .thenReturn(new URL[] {testUrl}); + optionsMap = CassandraDriverConfigLoader.getOptionsMapFromFile(testGcsPath); } + + DataSource dataSource = + CassandraIOWrapperHelper.buildDataSource( + optionsMap, + null, + CassandraDialect.OSS, + GuardedStringValueProvider.create(""), + "", + "", + ""); + assertThat(dataSource.cassandra().oss().loggedKeySpace()).isEqualTo("test-keyspace"); + assertThat(dataSource.cassandra().oss().localDataCenter()).isEqualTo("datacenter1"); + assertThat(dataSource.cassandra().oss().numPartitions()).isEqualTo(null); + assertThat( + CassandraIOWrapperHelper.buildDataSource( + optionsMap, + 42, + CassandraDialect.OSS, + GuardedStringValueProvider.create(""), + "", + "", + "") + .cassandra() + .oss() + .numPartitions()) + .isEqualTo(42); } @Test @@ -275,7 +257,7 @@ public void testAstra() { DataSource dataSource = CassandraIOWrapperHelper.buildDataSource( - "", + null, null, CassandraDialect.ASTRA, testAstraDbToken, diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIoWrapperTest.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIoWrapperTest.java index 9b3b1eb7f9..aad79eadf3 100644 --- a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIoWrapperTest.java +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/cassandra/reader/io/cassandra/iowrapper/CassandraIoWrapperTest.java @@ -95,7 +95,7 @@ public void testCassandraIoWrapperBasic() { .when( () -> CassandraIOWrapperHelper.buildDataSource( - testGcsPath, + null, null, CassandraDialect.OSS, GuardedStringValueProvider.create(""), @@ -107,7 +107,7 @@ public void testCassandraIoWrapperBasic() { .when( () -> CassandraIOWrapperHelper.buildDataSource( - "", + null, null, CassandraDialect.ASTRA, astraDataSource.cassandra().astra().astraToken(), @@ -151,7 +151,7 @@ public void testCassandraIoWrapperBasic() { CassandraIoWrapper cassandraIoWrapper = new CassandraIoWrapper( - testGcsPath, + null, tablesToRead, null, CassandraDialect.OSS, @@ -165,7 +165,7 @@ public void testCassandraIoWrapperBasic() { CassandraIoWrapper cassandraIoWrapperAstra = new CassandraIoWrapper( - "", + null, tablesToRead, null, CassandraDialect.ASTRA, diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/AstraDbToSpannerIT.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/AstraDbToSpannerIT.java new file mode 100644 index 0000000000..54ea70753d --- /dev/null +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/AstraDbToSpannerIT.java @@ -0,0 +1,266 @@ +/* + * Copyright (C) 2026 Google LLC + * + * 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.google.cloud.teleport.v2.templates; + +import static com.google.common.truth.Truth.assertThat; + +import com.datastax.oss.driver.api.core.CqlSession; +import com.dtsx.astra.sdk.db.AstraDBOpsClient; +import com.dtsx.astra.sdk.db.DbOpsClient; +import com.dtsx.astra.sdk.db.domain.Database; +import com.dtsx.astra.sdk.db.domain.DatabaseCreationRequest; +import com.dtsx.astra.sdk.db.domain.DatabaseStatusType; +import com.dtsx.astra.sdk.utils.ApiLocator; +import com.google.cloud.spanner.Struct; +import com.google.cloud.teleport.metadata.TemplateIntegrationTest; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.AstraConnectionConfig; +import com.google.gson.Gson; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.Serializable; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.beam.it.common.PipelineLauncher; +import org.apache.beam.it.common.PipelineOperator; +import org.apache.beam.it.common.TestProperties; +import org.apache.beam.it.common.utils.ResourceManagerUtils; +import org.apache.beam.it.gcp.spanner.SpannerResourceManager; +import org.apache.beam.it.gcp.spanner.matchers.SpannerAsserts; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Integration test for {@link com.google.cloud.teleport.v2.templates.SourceDbToSpanner} from Astra + * DB. + */ +@RunWith(JUnit4.class) +@Category(TemplateIntegrationTest.class) +@TemplateIntegrationTest(SourceDbToSpanner.class) +public class AstraDbToSpannerIT extends SourceDbToSpannerITBase implements Serializable { + + private static final Logger LOGGER = LoggerFactory.getLogger(AstraDbToSpannerIT.class); + + private static final long NUM_ROWS = 50L; + private static final String ASTRA_DB = "dataflow_integration_tests"; + private static final String ASTRA_DB_REGION = TestProperties.region(); + private static final String ASTRA_KS = "beam"; + + private static final String[][] SCIENTISTS = { + new String[] {"phys", "Einstein"}, + new String[] {"bio", "Darwin"}, + new String[] {"phys", "Copernicus"}, + new String[] {"bio", "Pasteur"}, + new String[] {"bio", "Curie"} + }; + + private String astraTable; + + private static DbOpsClient dbClient; + private SpannerResourceManager spannerResourceManager; + + @Before + public void setup() throws Exception { + astraTable = "scientist_" + testId.replaceAll("-", "_"); + + spannerResourceManager = + SpannerResourceManager.builder(testName, PROJECT, REGION).maybeUseStaticInstance().build(); + + // Create Spanner table + String spannerDdl = + String.format( + "CREATE TABLE %s (" + + " person_department STRING(MAX)," + + " person_id INT64," + + " person_name STRING(MAX)," + + ") PRIMARY KEY(person_department, person_id)", + astraTable); + spannerResourceManager.executeDdlStatement(spannerDdl); + + // Setup Astra Db + createOrResumeAstraDatabase(); + // Setup Astra Data + createAndPopulateTables(); + LOGGER.info("Initialization Successful."); + } + + @Test + public void testAstraDbToSpanner() throws IOException { + // Generate shard.json + AstraConnectionConfig astraConfig = new AstraConnectionConfig(); + astraConfig.setAstraToken(dbClient.getToken()); + astraConfig.setDatabaseId(dbClient.getDatabaseId()); + astraConfig.setKeySpace(ASTRA_KS); + astraConfig.setAstraDbRegion(dbClient.get().getInfo().getRegion()); + + String configContents = new Gson().toJson(astraConfig); + artifactClient.createArtifact("input/shard.json", configContents); + String sourceConfigURL = getGcsPath("input/shard.json", artifactClient); + + Map jobParameters = new HashMap<>(); + jobParameters.put("sourceDbDialect", "ASTRA_DB"); + jobParameters.put("sourceConfigURL", sourceConfigURL); + jobParameters.put("outputDirectory", getGcsPath("output", artifactClient)); + jobParameters.put("tables", astraTable); + jobParameters.put( + "ipConfiguration", "WORKER_IP_UNSPECIFIED"); // Require internet access for Astra API + + PipelineLauncher.LaunchInfo info = + launchDataflowJob(testName, null, null, null, spannerResourceManager, jobParameters, null); + LOGGER.info("Pipeline is now running."); + + PipelineOperator.Result result = pipelineOperator().waitUntilDone(createConfig(info)); + + org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult(result).isLaunchFinished(); + LOGGER.info("Destination Table has been populated."); + + // Verify the row count matches what was inserted + List rows = + spannerResourceManager.readTableRecords( + astraTable, List.of("person_department", "person_id", "person_name")); + assertThat(rows).hasSize(Math.toIntExact(NUM_ROWS)); + + // Verify row data content using SpannerAsserts like other ITs + List> expectedData = getExpectedData(); + + SpannerAsserts.assertThatStructs(rows).hasRecordsUnorderedCaseInsensitiveColumns(expectedData); + } + + private List> getExpectedData() { + List> expectedData = new ArrayList<>(); + for (int i = 0; i < NUM_ROWS; i++) { + int index = i % SCIENTISTS.length; + Map expectedRow = new HashMap<>(); + expectedRow.put("person_department", SCIENTISTS[index][0]); + expectedRow.put("person_id", i); + expectedRow.put("person_name", SCIENTISTS[index][1]); + expectedData.add(expectedRow); + } + return expectedData; + } + + @After + public void tearDown() { + ResourceManagerUtils.cleanResources(spannerResourceManager); + + if (dbClient != null) { + try (CqlSession astraSession = + CqlSession.builder() + .withCloudSecureConnectBundle( + new ByteArrayInputStream(dbClient.downloadDefaultSecureConnectBundle())) + .withAuthCredentials("token", dbClient.getToken()) + .withKeyspace(ASTRA_KS) + .build()) { + astraSession.execute(String.format("DROP TABLE IF EXISTS %s.%s;", ASTRA_KS, astraTable)); + } catch (Exception e) { + LOGGER.warn("Failed to drop Astra table", e); + } + } + } + + private static String test() { + return "AstraCS:" + HASH; + } + + @SuppressWarnings("BusyWait") + private void createOrResumeAstraDatabase() throws InterruptedException { + AstraDBOpsClient databasesClient = new AstraDBOpsClient(test()); + if (databasesClient.findByName(ASTRA_DB).findAny().isEmpty()) { + LOGGER.info("Create a new Database {}", ASTRA_DB); + databasesClient.create( + DatabaseCreationRequest.builder() + .name(ASTRA_DB) + .keyspace(ASTRA_KS) + .cloudRegion(ASTRA_DB_REGION) + .build()); + } else { + LOGGER.info("Database {} exists in source organization.", ASTRA_DB); + } + dbClient = databasesClient.databaseByName(ASTRA_DB); + if (dbClient.get().getStatus() == DatabaseStatusType.HIBERNATED) { + resumeDb(dbClient.get()); + LOGGER.info("Resuming as DB was Hibernated."); + } + while (dbClient.get().getStatus() != DatabaseStatusType.ACTIVE) { + Thread.sleep(5000); + LOGGER.info("Waiting for DB to be ACTIVE."); + } + } + + private void resumeDb(Database db) { + try { + HttpClient.newBuilder() + .version(HttpClient.Version.HTTP_2) + .connectTimeout(Duration.ofSeconds(20)) + .build() + .send( + HttpRequest.newBuilder() + .timeout(Duration.ofSeconds(20)) + .uri( + URI.create( + ApiLocator.getApiRestEndpoint(db.getId(), db.getInfo().getRegion()) + + "/v2/schemas/keyspace")) + .timeout(Duration.ofSeconds(20)) + .header("Content-Type", "application/json") + .header("X-Cassandra-Token", test()) + .GET() + .build(), + HttpResponse.BodyHandlers.ofString()); + } catch (Exception e) { + throw new IllegalStateException("Cannot resume database", e); + } + } + + private void createAndPopulateTables() { + try (CqlSession astraSession = + CqlSession.builder() + .withCloudSecureConnectBundle( + new ByteArrayInputStream(dbClient.downloadDefaultSecureConnectBundle())) + .withAuthCredentials("token", dbClient.getToken()) + .withKeyspace(ASTRA_KS) + .build()) { + astraSession.execute( + String.format( + "CREATE TABLE IF NOT EXISTS %s.%s(person_department text, person_id int, person_name text, PRIMARY KEY" + + "((person_department), person_id));", + ASTRA_KS, astraTable)); + for (int i = 0; i < NUM_ROWS; i++) { + int index = i % SCIENTISTS.length; + String insertStr = + String.format( + "INSERT INTO %s.%s(person_department, person_id, person_name) values(" + + "'%s', %d, '%s');", + ASTRA_KS, astraTable, SCIENTISTS[index][0], i, SCIENTISTS[index][1]); + astraSession.execute(insertStr); + } + } + } + + private static final String HASH = + "AIpXbGsYPQCXtrwExZvOktGw:3d5bae1547a667608f10ab2d2e89a90b936f8ff8a3e9111efe23fc818ef344fd"; +} diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/PipelineControllerTest.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/PipelineControllerTest.java index 8a731dbf15..afb945bd09 100644 --- a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/PipelineControllerTest.java +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/PipelineControllerTest.java @@ -21,6 +21,7 @@ import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.when; import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions; @@ -38,6 +39,9 @@ import com.google.cloud.teleport.v2.spanner.migrations.schema.SchemaStringOverridesBasedMapper; import com.google.cloud.teleport.v2.spanner.migrations.schema.SessionBasedMapper; import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConfigParser; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.SourceConnectionConfig; import com.google.cloud.teleport.v2.spanner.migrations.spanner.SpannerSchema; import com.google.cloud.teleport.v2.templates.PipelineController.ShardedJdbcDbConfigContainer; import com.google.cloud.teleport.v2.templates.PipelineController.SingleInstanceJdbcDbConfigContainer; @@ -66,6 +70,7 @@ import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; import org.mockito.Mock; +import org.mockito.MockedConstruction; import org.mockito.MockedStatic; import org.mockito.Mockito; import org.mockito.junit.MockitoJUnitRunner; @@ -353,9 +358,8 @@ public void singleDbConfigContainerWithUrlTest() { sourceDbToSpannerOptions.setJdbcDriverClassName(testDriverClassName); sourceDbToSpannerOptions.setMaxConnections(150); sourceDbToSpannerOptions.setNumPartitions(4000); - sourceDbToSpannerOptions.setUsername(testUser); - sourceDbToSpannerOptions.setPassword(testPassword); sourceDbToSpannerOptions.setTables("table1,table2"); + Shard shard = new Shard("", "localhost", "3306", "user", "password", "testDB", "", "", ""); String sessionFilePath = Paths.get(Resources.getResource("session-file-with-dropped-column.json").getPath()) .toString(); @@ -365,7 +369,7 @@ public void singleDbConfigContainerWithUrlTest() { PCollection dummyPCollection = pipeline.apply(Create.of(1)); pipeline.run(); SingleInstanceJdbcDbConfigContainer dbConfigContainer = - new SingleInstanceJdbcDbConfigContainer(sourceDbToSpannerOptions); + new SingleInstanceJdbcDbConfigContainer(sourceDbToSpannerOptions, shard); JdbcIoWrapperConfigGroup configGroup = dbConfigContainer.getJdbcIoWrapperConfigGroup( List.of("table1", "table2"), Wait.on(dummyPCollection)); @@ -683,6 +687,87 @@ public void shardedDbConfigContainerPGTest() { assertThat(config.waitOn()).isNotNull(); } + @Test + public void testGetSourceConnectionConfig() { + String testFilePath = "gs://test-bucket/config.json"; + + try (MockedConstruction mockedParser = + Mockito.mockConstruction( + SourceConfigParser.class, + (mock, context) -> { + when(mock.parseConfiguration("MYSQL", testFilePath)) + .thenReturn(new JdbcShardConfig()); + when(mock.parseConfiguration("INVALID", testFilePath)) + .thenThrow(new IllegalArgumentException("Invalid type")); + })) { + + SourceConnectionConfig config = + PipelineController.getSourceConnectionConfig("MYSQL", testFilePath); + assertThat(config).isInstanceOf(JdbcShardConfig.class); + + RuntimeException exception = + assertThrows( + RuntimeException.class, + () -> PipelineController.getSourceConnectionConfig("INVALID", testFilePath)); + assertThat(exception.getMessage()).contains("Error parsing source config"); + } + } + + @Test + public void testExecuteJdbcMigration() { + SourceDbToSpannerOptions mockOptions = + PipelineOptionsFactory.as(SourceDbToSpannerOptions.class); + mockOptions.setSourceDbDialect(SQLDialect.MYSQL.name()); + + SpannerConfig spannerConfig = SpannerConfig.create(); + org.apache.beam.sdk.Pipeline mockPipeline = mock(org.apache.beam.sdk.Pipeline.class); + when(mockPipeline.getOptions()).thenReturn(mockOptions); + + JdbcShardConfig mockConfig = new JdbcShardConfig(); + + // Test Single Shard Routing + mockConfig.setShardConfigs( + List.of(new Shard("shard1", "localhost", "3306", "user", "pass", "db1", null, null, null))); + + try (MockedStatic mockedPipelineController = + Mockito.mockStatic(PipelineController.class)) { + mockedPipelineController + .when(() -> PipelineController.executeJdbcMigration(any(), any(), any(), any())) + .thenCallRealMethod(); + + PipelineController.executeJdbcMigration(mockOptions, mockConfig, mockPipeline, spannerConfig); + + mockedPipelineController.verify( + () -> PipelineController.executeJdbcSingleInstanceMigration(any(), any(), any(), any()), + times(1)); + mockedPipelineController.verify( + () -> PipelineController.executeJdbcShardedMigration(any(), any(), any(), any()), + times(0)); + } + + // Test Multi Shard Routing + mockConfig.setShardConfigs( + List.of( + new Shard("shard1", "localhost", "3306", "user", "pass", "db1", null, null, null), + new Shard("shard2", "localhost", "3306", "user", "pass", "db2", null, null, null))); + + try (MockedStatic mockedPipelineController = + Mockito.mockStatic(PipelineController.class)) { + mockedPipelineController + .when(() -> PipelineController.executeJdbcMigration(any(), any(), any(), any())) + .thenCallRealMethod(); + + PipelineController.executeJdbcMigration(mockOptions, mockConfig, mockPipeline, spannerConfig); + + mockedPipelineController.verify( + () -> PipelineController.executeJdbcShardedMigration(any(), any(), any(), any()), + times(1)); + mockedPipelineController.verify( + () -> PipelineController.executeJdbcSingleInstanceMigration(any(), any(), any(), any()), + times(0)); + } + } + @After public void cleanup() { if (mockedStaticJdbcIoWrapper != null) { diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/SourceDbToSpannerITBase.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/SourceDbToSpannerITBase.java index dbabfc4423..e7c228aa5d 100644 --- a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/SourceDbToSpannerITBase.java +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/SourceDbToSpannerITBase.java @@ -19,8 +19,11 @@ import com.google.cloud.spanner.Dialect; import com.google.cloud.teleport.v2.reader.io.jdbc.iowrapper.config.SQLDialect; +import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig; import com.google.cloud.teleport.v2.spanner.migrations.transformation.CustomTransformation; import com.google.common.io.Resources; +import com.google.gson.Gson; import java.io.BufferedWriter; import java.io.FileWriter; import java.io.IOException; @@ -233,7 +236,9 @@ protected PipelineLauncher.LaunchInfo launchDataflowJob( } }; if (sourceResourceManager instanceof JDBCResourceManager) { - params.putAll(getJdbcParameters((JDBCResourceManager) sourceResourceManager)); + params.putAll( + getJdbcParameters( + (JDBCResourceManager) sourceResourceManager, gcsPathPrefix, jobParameters)); } else if (sourceResourceManager instanceof CassandraResourceManager) { params.putAll( getCassandraParameters((CassandraResourceManager) sourceResourceManager, gcsPathPrefix)); @@ -258,8 +263,15 @@ protected PipelineLauncher.LaunchInfo launchDataflowJob( } // overridden parameters + String ipConfig = "WORKER_IP_PRIVATE"; if (jobParameters != null) { + if (jobParameters.containsKey("ipConfiguration")) { + ipConfig = jobParameters.get("ipConfiguration"); + } for (Map.Entry entry : jobParameters.entrySet()) { + if ("namespace".equals(entry.getKey()) || "ipConfiguration".equals(entry.getKey())) { + continue; + } params.put(entry.getKey(), entry.getValue()); } } @@ -272,7 +284,7 @@ protected PipelineLauncher.LaunchInfo launchDataflowJob( options.setParameters(params); options.addEnvironment("additionalExperiments", List.of("disable_runner_v2")); options.addEnvironment("numWorkers", 2); - options.addEnvironment("ipConfiguration", "WORKER_IP_PRIVATE"); + options.addEnvironment("ipConfiguration", ipConfig); // Run PipelineLauncher.LaunchInfo jobInfo = launchTemplate(options); assertThatPipeline(jobInfo).isRunning(); @@ -280,13 +292,70 @@ protected PipelineLauncher.LaunchInfo launchDataflowJob( return jobInfo; } - private Map getJdbcParameters(JDBCResourceManager jdbcResourceManager) { + protected String createAndUploadShardConfigToGcs( + String gcsPathPrefix, + JDBCResourceManager jdbcResourceManager, + Map jobParameters) + throws IOException { + Shard shard = new Shard(); + shard.setLogicalShardId("Shard1"); + shard.setUser(jdbcResourceManager.getUsername()); + shard.setPassword(jdbcResourceManager.getPassword()); + if (jdbcResourceManager instanceof PostgresResourceManager pgRm) { + shard.setHost(pgRm.getHost()); + shard.setPort(String.valueOf(pgRm.getPort())); + shard.setDbName(pgRm.getDatabaseName()); + } else if (jdbcResourceManager instanceof MySQLResourceManager mySqlRm) { + shard.setHost(mySqlRm.getHost()); + shard.setPort(String.valueOf(mySqlRm.getPort())); + shard.setDbName(mySqlRm.getDatabaseName()); + } else if (jdbcResourceManager + instanceof org.apache.beam.it.gcp.cloudsql.CloudSqlResourceManager cloudRm) { + shard.setHost(cloudRm.getHost()); + shard.setPort(String.valueOf(cloudRm.getPort())); + shard.setDbName(cloudRm.getDatabaseName()); + } else { + throw new IllegalArgumentException( + "Unsupported JDBC resource manager type: " + jdbcResourceManager.getClass().getName()); + } + + if (jobParameters != null && jobParameters.containsKey("namespace")) { + shard.setNamespace(jobParameters.get("namespace")); + } + + JdbcShardConfig jdbcShardConfig = new JdbcShardConfig(); + jdbcShardConfig.setShardConfigs(List.of(shard)); + String shardFileContents = new Gson().toJson(jdbcShardConfig); + LOG.info("Shard file contents: {}", shardFileContents); + + String configBasePath = (gcsPathPrefix == null) ? "null" : gcsPathPrefix; + if (configBasePath.endsWith("/")) { + configBasePath = configBasePath.substring(0, configBasePath.length() - 1); + } + String configGcsPath = getGcsPath(configBasePath + "/shard.json"); + + gcsClient.createArtifact(configBasePath + "/shard.json", shardFileContents); + + return configGcsPath; + } + + private Map getJdbcParameters( + JDBCResourceManager jdbcResourceManager, + String gcsPathPrefix, + Map jobParameters) { Map params = new HashMap<>() { { put("sourceDbDialect", sqlDialectFrom(jdbcResourceManager)); - put("sourceConfigURL", jdbcResourceManager.getUri()); + try { + put( + "sourceConfigURL", + createAndUploadShardConfigToGcs( + gcsPathPrefix, jdbcResourceManager, jobParameters)); + } catch (IOException e) { + throw new RuntimeException(e); + } put("username", jdbcResourceManager.getUsername()); put("password", jdbcResourceManager.getPassword()); put("jdbcDriverClassName", driverClassNameFrom(jdbcResourceManager)); diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/SourceDbToSpannerTest.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/SourceDbToSpannerTest.java index 1ebeac42cb..f31e27a82d 100644 --- a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/SourceDbToSpannerTest.java +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/SourceDbToSpannerTest.java @@ -17,6 +17,7 @@ import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockConstruction; @@ -27,6 +28,9 @@ import com.google.cloud.teleport.v2.common.CommonTemplateJvmInitializer; import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.AstraConnectionConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.CassandraConnectionConfig; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig; import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.joda.time.Duration; @@ -84,4 +88,48 @@ public void testCreateSpannerConfig() { config = SourceDbToSpanner.createSpannerConfig(mockOptions); assertEquals(config.getMaxCommitDelay().get(), Duration.millis(42L)); } + + @Test + public void testRun_ValidationFailures() { + SourceDbToSpannerOptions mockOptions = + PipelineOptionsFactory.as(SourceDbToSpannerOptions.class); + mockOptions.setProjectId("testProject"); + mockOptions.setInstanceId("testInstance"); + mockOptions.setDatabaseId("testDatabaseId"); + mockOptions + .as(org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions.class) + .setWorkerMachineType("n2-standard-4"); + + try (MockedStatic mockedPipelineController = + mockStatic(PipelineController.class)) { + + // Test MYSQL dialect with wrong config type (AstraConnectionConfig) + mockOptions.setSourceDbDialect(SourceDbToSpannerOptions.MYSQL_SOURCE_DIALECT); + mockedPipelineController + .when(() -> PipelineController.getSourceConnectionConfig(any(), any())) + .thenReturn(mock(AstraConnectionConfig.class)); + assertThrows(IllegalArgumentException.class, () -> SourceDbToSpanner.run(mockOptions)); + + // Test POSTGRESQL dialect with wrong config type (CassandraConnectionConfig) + mockOptions.setSourceDbDialect(SourceDbToSpannerOptions.PG_SOURCE_DIALECT); + mockedPipelineController + .when(() -> PipelineController.getSourceConnectionConfig(any(), any())) + .thenReturn(mock(CassandraConnectionConfig.class)); + assertThrows(IllegalArgumentException.class, () -> SourceDbToSpanner.run(mockOptions)); + + // Test CASSANDRA dialect with wrong config type (JdbcShardConfig) + mockOptions.setSourceDbDialect(SourceDbToSpannerOptions.CASSANDRA_SOURCE_DIALECT); + mockedPipelineController + .when(() -> PipelineController.getSourceConnectionConfig(any(), any())) + .thenReturn(mock(JdbcShardConfig.class)); + assertThrows(IllegalArgumentException.class, () -> SourceDbToSpanner.run(mockOptions)); + + // Test ASTRADB dialect with wrong config type (JdbcShardConfig) + mockOptions.setSourceDbDialect(SourceDbToSpannerOptions.ASTRA_DB_SOURCE_DIALECT); + mockedPipelineController + .when(() -> PipelineController.getSourceConnectionConfig(any(), any())) + .thenReturn(mock(JdbcShardConfig.class)); + assertThrows(IllegalArgumentException.class, () -> SourceDbToSpanner.run(mockOptions)); + } + } } diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/failureinjectiontesting/SourceDbToSpannerFTBase.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/failureinjectiontesting/SourceDbToSpannerFTBase.java index 10040578be..670757cdf2 100644 --- a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/failureinjectiontesting/SourceDbToSpannerFTBase.java +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/failureinjectiontesting/SourceDbToSpannerFTBase.java @@ -19,7 +19,10 @@ import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; import com.google.cloud.teleport.v2.spanner.migrations.constants.Constants; +import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig; import com.google.cloud.teleport.v2.spanner.migrations.transformation.CustomTransformation; +import com.google.gson.Gson; import com.google.pubsub.v1.SubscriptionName; import com.google.pubsub.v1.TopicName; import java.io.BufferedReader; @@ -45,8 +48,6 @@ import org.apache.beam.it.gcp.spanner.SpannerResourceManager; import org.apache.beam.it.gcp.storage.GcsResourceManager; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings; -import org.json.JSONArray; -import org.json.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -116,6 +117,21 @@ protected PipelineLauncher.LaunchInfo launchBulkDataflowJob( CloudSqlResourceManager cloudSqlResourceManager, CustomTransformation customTransformation) throws IOException { + + Shard shard = new Shard(); + shard.setLogicalShardId("Shard1"); + shard.setHost(cloudSqlResourceManager.getHost()); + shard.setPort(String.valueOf(cloudSqlResourceManager.getPort())); + shard.setDbName(cloudSqlResourceManager.getDatabaseName()); + shard.setUser(cloudSqlResourceManager.getUsername()); + shard.setPassword(cloudSqlResourceManager.getPassword()); + + JdbcShardConfig jdbcShardConfig = new JdbcShardConfig(); + jdbcShardConfig.setShardConfigs(List.of(shard)); + String shardFileContents = new Gson().toJson(jdbcShardConfig); + LOG.info("Shard file contents: {}", shardFileContents); + gcsResourceManager.createArtifact("input/shard.json", shardFileContents); + // launch dataflow template FlexTemplateDataflowJobResourceManager.Builder flexTemplateBuilder = FlexTemplateDataflowJobResourceManager.builder(jobName) @@ -125,9 +141,7 @@ protected PipelineLauncher.LaunchInfo launchBulkDataflowJob( .addParameter("databaseId", spannerResourceManager.getDatabaseId()) .addParameter("projectId", PROJECT) .addParameter("outputDirectory", getGcsPath("output", gcsResourceManager)) - .addParameter("sourceConfigURL", cloudSqlResourceManager.getUri()) - .addParameter("username", cloudSqlResourceManager.getUsername()) - .addParameter("password", cloudSqlResourceManager.getPassword()) + .addParameter("sourceConfigURL", getGcsPath("input/shard.json", gcsResourceManager)) .addParameter("jdbcDriverClassName", "com.mysql.jdbc.Driver") .addParameter("workerMachineType", "n2-standard-4") .addEnvironmentVariable( @@ -189,51 +203,29 @@ protected PipelineLauncher.LaunchInfo launchShardedBulkDataflowJob( protected void createAndUploadBulkShardConfigToGcs( ArrayList dataShardsList, GcsResourceManager gcsResourceManager) { - JSONObject bulkConfig = new JSONObject(); - bulkConfig.put("configType", "dataflow"); - - JSONObject shardConfigBulk = new JSONObject(); - - JSONObject schemaSourceJson = new JSONObject(); - schemaSourceJson.put("dataShardId", ""); - schemaSourceJson.put("host", ""); - schemaSourceJson.put("user", ""); - schemaSourceJson.put("password", ""); - schemaSourceJson.put("port", ""); - schemaSourceJson.put("dbName", ""); - shardConfigBulk.put("schemaSource", schemaSourceJson); - - JSONArray dataShardsArray = new JSONArray(); + List shards = new ArrayList<>(); if (dataShardsList != null) { for (DataShard shardData : dataShardsList) { - JSONObject shardJson = new JSONObject(); - - shardJson.put("dataShardId", shardData.dataShardId); - shardJson.put("host", shardData.host); - shardJson.put("user", shardData.user); - shardJson.put("password", shardData.password); - shardJson.put("port", shardData.port); - shardJson.put("dbName", shardData.dbName); - shardJson.put("namespace", shardData.namespace); - shardJson.put("connectionProperties", shardData.connectionProperties); - - JSONArray databasesArray = new JSONArray(); - - for (Database dbData : shardData.databases) { - JSONObject dbJson = new JSONObject(); - dbJson.put("dbName", dbData.dbName); - dbJson.put("databaseId", dbData.databaseId); - dbJson.put("refDataShardId", dbData.refDataShardId); - databasesArray.put(dbJson); + Shard shard = new Shard(); + shard.setLogicalShardId(shardData.dataShardId); + shard.setHost(shardData.host); + shard.setUser(shardData.user); + shard.setPassword(shardData.password); + shard.setPort(shardData.port); + shard.setDbName(shardData.dbName); + if (shardData.namespace != null) { + shard.setNamespace(shardData.namespace); + } + if (shardData.connectionProperties != null) { + shard.setConnectionProperties(shardData.connectionProperties); } - shardJson.put("databases", databasesArray); - dataShardsArray.put(shardJson); + shards.add(shard); } } - shardConfigBulk.put("dataShards", dataShardsArray); - bulkConfig.put("shardConfigurationBulk", shardConfigBulk); - String shardFileContents = bulkConfig.toString(); + JdbcShardConfig jdbcShardConfig = new JdbcShardConfig(); + jdbcShardConfig.setShardConfigs(shards); + String shardFileContents = new Gson().toJson(jdbcShardConfig); LOG.info("Shard file contents: {}", shardFileContents); gcsResourceManager.createArtifact("input/shard-bulk.json", shardFileContents); } diff --git a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/loadtesting/SourceDbToSpannerLTBase.java b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/loadtesting/SourceDbToSpannerLTBase.java index ac77f80079..314bc8b6bd 100644 --- a/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/loadtesting/SourceDbToSpannerLTBase.java +++ b/v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/loadtesting/SourceDbToSpannerLTBase.java @@ -19,11 +19,15 @@ import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult; import com.google.cloud.teleport.v2.reader.io.jdbc.iowrapper.config.SQLDialect; +import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig; +import com.google.gson.Gson; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.text.ParseException; import java.time.Duration; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -34,6 +38,7 @@ import org.apache.beam.it.common.utils.ResourceManagerUtils; import org.apache.beam.it.conditions.ConditionCheck; import org.apache.beam.it.gcp.TemplateLoadTestBase; +import org.apache.beam.it.gcp.artifacts.utils.ArtifactUtils; import org.apache.beam.it.gcp.secretmanager.SecretManagerResourceManager; import org.apache.beam.it.gcp.spanner.SpannerResourceManager; import org.apache.beam.it.gcp.spanner.conditions.SpannerRowsCheck; @@ -175,11 +180,40 @@ protected String getOutputDirectory() { } protected Map getJdbcParameters(StaticJDBCResource jdbcResource) { - return getJdbcParameters( - jdbcResource.getconnectionURL(), - jdbcResource.username(), - jdbcResource.password(), - driverClassName()); + try { + return getJdbcParameters( + createAndUploadShardConfigToGcs(jdbcResource), + jdbcResource.username(), + jdbcResource.password(), + driverClassName()); + } catch (IOException e) { + throw new RuntimeException("Failed to create and upload shard config", e); + } + } + + protected String createAndUploadShardConfigToGcs(StaticJDBCResource jdbcResource) + throws IOException { + Shard shard = new Shard(); + shard.setLogicalShardId("Shard1"); + shard.setUser(jdbcResource.username()); + shard.setPassword(jdbcResource.password()); + shard.setHost(jdbcResource.hostname()); + shard.setPort(String.valueOf(jdbcResource.port())); + shard.setDbName(jdbcResource.database()); + + JdbcShardConfig jdbcShardConfig = new JdbcShardConfig(); + jdbcShardConfig.setShardConfigs(Collections.singletonList(shard)); + String shardFileContents = new Gson().toJson(jdbcShardConfig); + gcsResourceManager.createArtifact("input/shard.json", shardFileContents); + return getGcsPath("input/shard.json", gcsResourceManager); + } + + protected String getGcsPath(String artifactId, GcsResourceManager gcsResourceManager) { + return ArtifactUtils.getFullGcsPath( + gcsResourceManager.getBucket(), + getClass().getSimpleName(), + gcsResourceManager.runId(), + artifactId); } protected Map getJdbcParameters( diff --git a/v2/sourcedb-to-spanner/src/test/resources/SourceConfig/astra-connection-config.json b/v2/sourcedb-to-spanner/src/test/resources/SourceConfig/astra-connection-config.json new file mode 100644 index 0000000000..ea4ba3e789 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/test/resources/SourceConfig/astra-connection-config.json @@ -0,0 +1,6 @@ +{ + "databaseId": "test-database-id", + "astraToken": "AstraCS:test-token", + "keySpace": "test-keyspace", + "astraDbRegion": "us-east1" +} diff --git a/v2/sourcedb-to-spanner/src/test/resources/SourceConfig/cassandra-driver-config.conf b/v2/sourcedb-to-spanner/src/test/resources/SourceConfig/cassandra-driver-config.conf new file mode 100644 index 0000000000..bfdf5a3af1 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/test/resources/SourceConfig/cassandra-driver-config.conf @@ -0,0 +1,11 @@ +datastax-java-driver { + basic.contact-points = ["127.0.0.1:9042", "127.0.0.1:9043"] + basic.session-keyspace = "test-keyspace" + basic.load-balancing-policy { + local-datacenter = "datacenter1" + } + advanced.auth-provider { + username = "testUserName" + password = "testPassword1234@" + } +} diff --git a/v2/sourcedb-to-spanner/src/test/resources/SourceConfig/jdbc-shard-config.json b/v2/sourcedb-to-spanner/src/test/resources/SourceConfig/jdbc-shard-config.json new file mode 100644 index 0000000000..e429e593a0 --- /dev/null +++ b/v2/sourcedb-to-spanner/src/test/resources/SourceConfig/jdbc-shard-config.json @@ -0,0 +1,14 @@ +{ + "shardConfigs": [ + { + "logicalShardId": "shard1", + "host": "10.0.0.1", + "port": "3306", + "user": "db_user", + "password": "db_password", + "dbName": "test_db", + "namespace": "public", + "connectionProperties": "autoReconnect=true&maxReconnects=10" + } + ] +} diff --git a/v2/sourcedb-to-spanner/terraform/samples/single-job-bulk-migration/variables.tf b/v2/sourcedb-to-spanner/terraform/samples/single-job-bulk-migration/variables.tf index 1e1231264f..8a0d6f4015 100644 --- a/v2/sourcedb-to-spanner/terraform/samples/single-job-bulk-migration/variables.tf +++ b/v2/sourcedb-to-spanner/terraform/samples/single-job-bulk-migration/variables.tf @@ -49,7 +49,7 @@ variable "jdbc_driver_class_name" { variable "source_config_url" { type = string - description = "JDBC connection url for the source database. Ex- jdbc:mysql://127.4.5.30:3306/my-db?autoReconnect=true&maxReconnects=10&unicode=true&characterEncoding=UTF-8" + description = "Source connection config file URL. The file format is dependent on the source type." } variable "username" {