-
Notifications
You must be signed in to change notification settings - Fork 38
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
WIP Add basic migration test from Cassandra
- Loading branch information
Showing
8 changed files
with
242 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
| cassandra/ | ||
| scylla/ | ||
| spark-master/ |
43 changes: 43 additions & 0 deletions
43
tests/src/test/configurations/cassandra-to-scylla-basic.yaml
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| source: | ||
| type: cassandra | ||
| host: cassandra | ||
| port: 9042 | ||
| localDC: datacenter1 | ||
| credentials: | ||
| username: dummy | ||
| password: dummy | ||
| keyspace: test | ||
| table: basictest | ||
| consistencyLevel: LOCAL_QUORUM | ||
| preserveTimestamps: true | ||
| splitCount: 8 | ||
| connections: 8 | ||
| fetchSize: 1000 | ||
|
|
||
| target: | ||
| type: scylla | ||
| host: scylla | ||
| port: 9042 | ||
| localDC: datacenter1 | ||
| credentials: | ||
| username: dummy | ||
| password: dummy | ||
| keyspace: test | ||
| table: basictest | ||
| consistencyLevel: LOCAL_QUORUM | ||
| connections: 16 | ||
| stripTrailingZerosForDecimals: false | ||
|
|
||
| renames: [] | ||
|
|
||
| savepoints: | ||
| path: /app/savepoints | ||
| intervalSeconds: 300 | ||
| skipTokenRanges: [] | ||
| validation: | ||
| compareTimestamps: true | ||
| ttlToleranceMillis: 60000 | ||
| writetimeToleranceMillis: 1000 | ||
| failuresToFetch: 100 | ||
| floatingPointTolerance: 0.001 | ||
| timestampMsTolerance: 0 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
33 changes: 33 additions & 0 deletions
33
tests/src/test/scala/com/scylladb/migrator/scylla/BasicMigrationTest.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| package com.scylladb.migrator.scylla | ||
|
|
||
| import com.datastax.oss.driver.api.querybuilder.QueryBuilder | ||
|
|
||
| class BasicMigrationTest extends MigratorSuite { | ||
|
|
||
| withTable("BasicTest").test("Read from source and write to target") { tableName => | ||
| val insertStatement = | ||
| QueryBuilder | ||
| .insertInto(keyspace, tableName) | ||
| .json("""{ "id": "12345", "foo": "bar" }""") | ||
| .build() | ||
|
|
||
| // Insert some items | ||
| sourceCassandra.execute(insertStatement) | ||
|
|
||
| // Perform the migration | ||
| submitSparkJob("cassandra-to-scylla-basic.yaml") | ||
|
|
||
| // Check that the schema has been replicated to the target table | ||
| val selectQuery = | ||
| QueryBuilder.selectFrom(keyspace, tableName).all().build() | ||
| val sourceColumns = | ||
| sourceCassandra.execute(selectQuery).getColumnDefinitions | ||
| val targetColumns = | ||
| targetScylla.execute(selectQuery).getColumnDefinitions | ||
| assertEquals(targetColumns, sourceColumns) | ||
|
|
||
| // Check that the items have been migrated to the target table | ||
| // TODO | ||
| } | ||
|
|
||
| } |
137 changes: 137 additions & 0 deletions
137
tests/src/test/scala/com/scylladb/migrator/scylla/MigratorSuite.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,137 @@ | ||
| package com.scylladb.migrator.scylla | ||
|
|
||
| import com.datastax.oss.driver.api.core.CqlSession | ||
| import com.datastax.oss.driver.api.core.`type`.DataTypes | ||
| import com.datastax.oss.driver.api.querybuilder.SchemaBuilder | ||
|
|
||
| import java.net.InetSocketAddress | ||
| import scala.sys.process.Process | ||
| import scala.jdk.CollectionConverters._ | ||
| import scala.util.chaining._ | ||
|
|
||
| /** | ||
| * Base class for implementing end-to-end tests. | ||
| * | ||
| * It expects external services (Cassandra, Scylla, Spark, etc.) to be running. | ||
| * See the files `CONTRIBUTING.md` and `docker-compose-tests.yml` for more information. | ||
| */ | ||
| trait MigratorSuite extends munit.FunSuite { | ||
|
|
||
| val keyspace = "test" | ||
|
|
||
| /** Client of a source Cassandra instance */ | ||
| val sourceCassandra: CqlSession = CqlSession | ||
| .builder() | ||
| .addContactPoint(new InetSocketAddress("localhost", 9043)) | ||
| .withLocalDatacenter("datacenter1") | ||
| .withAuthCredentials("dummy", "dummy") | ||
| .build() | ||
|
|
||
| /** Client of a target ScyllaDB instance */ | ||
| val targetScylla: CqlSession = CqlSession | ||
| .builder() | ||
| .addContactPoint(new InetSocketAddress("localhost", 9042)) | ||
| .withLocalDatacenter("datacenter1") | ||
| .withAuthCredentials("dummy", "dummy") | ||
| .build() | ||
|
|
||
| /** | ||
| * Fixture automating the house-keeping work when migrating a table. | ||
| * | ||
| * It deletes the table from both the source and target databases in case it was already | ||
| * existing, and then recreates it in the source database. | ||
| * | ||
| * After the test is executed, it deletes the table from both the source and target | ||
| * databases. | ||
| * | ||
| * @param name Name of the table | ||
| */ | ||
| def withTable(name: String): FunFixture[String] = FunFixture( | ||
| setup = { _ => | ||
| def deleteTableIfExists(database: CqlSession): Unit = | ||
| try { | ||
| val query = SchemaBuilder.dropTable(keyspace, name).ifExists().build() | ||
| database | ||
| .execute(query) | ||
| .ensuring(_.wasApplied()) | ||
| } catch { | ||
| case any: Throwable => | ||
| fail(s"Something did not work as expected", any) | ||
| } | ||
| // Make sure the target database does not contain the table already | ||
| deleteTableIfExists(sourceCassandra) | ||
| deleteTableIfExists(targetScylla) | ||
| try { | ||
| // Create the table in the source database | ||
| val createTableRequest = | ||
| SchemaBuilder | ||
| .createTable(keyspace, name) | ||
| .withPartitionKey("id", DataTypes.TEXT) | ||
| .withColumn("foo", DataTypes.TEXT) | ||
| .build() | ||
| sourceCassandra.execute(createTableRequest) | ||
| } catch { | ||
| case any: Throwable => | ||
| fail(s"Failed to created table ${name} in database ${sourceCassandra}: ${any}") | ||
| } | ||
| name | ||
| }, | ||
| teardown = { _ => | ||
| // Clean-up both the source and target databases because we assume the test did replicate the table | ||
| // to the target database | ||
| val dropTableQuery = SchemaBuilder.dropTable(keyspace, name).build() | ||
| targetScylla.execute(dropTableQuery) | ||
| sourceCassandra.execute(dropTableQuery) | ||
| () | ||
| } | ||
| ) | ||
|
|
||
| /** | ||
| * Run a migration by submitting a Spark job to the Spark cluster. | ||
| * @param migratorConfigFile Configuration file to use. Write your | ||
| * configuration files in the directory | ||
| * `src/test/configurations`, which is | ||
| * automatically mounted to the Spark | ||
| * cluster by Docker Compose. | ||
| */ | ||
| def submitSparkJob(migratorConfigFile: String): Unit = { | ||
| Process( | ||
| Seq( | ||
| "docker", | ||
| "compose", | ||
| "-f", "docker-compose-tests.yml", | ||
| "exec", | ||
| "spark-master", | ||
| "/spark/bin/spark-submit", | ||
| "--class", "com.scylladb.migrator.Migrator", | ||
| "--master", "spark://spark-master:7077", | ||
| "--conf", "spark.driver.host=spark-master", | ||
| "--conf", s"spark.scylla.config=/app/configurations/${migratorConfigFile}", | ||
| // Uncomment one of the following lines to plug a remote debugger on the Spark master or worker. | ||
| // "--conf", "spark.driver.extraJavaOptions=-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5005", | ||
| // "--conf", "spark.executor.extraJavaOptions=-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5006", | ||
| "/jars/scylla-migrator-assembly-0.0.1.jar" | ||
| ) | ||
| ).run().exitValue().tap { statusCode => | ||
| assertEquals(statusCode, 0, "Spark job failed") | ||
| } | ||
| () | ||
| } | ||
|
|
||
| override def beforeAll(): Unit = { | ||
| val keyspaceStatement = | ||
| SchemaBuilder | ||
| .createKeyspace(keyspace) | ||
| .ifNotExists() | ||
| .withReplicationOptions(Map[String, AnyRef]("class" -> "SimpleStrategy", "replication_factor" -> new Integer(1)).asJava) | ||
| .build() | ||
| sourceCassandra.execute(keyspaceStatement) | ||
| targetScylla.execute(keyspaceStatement) | ||
| } | ||
|
|
||
| override def afterAll(): Unit = { | ||
| sourceCassandra.close() | ||
| targetScylla.close() | ||
| } | ||
|
|
||
| } |