Merge pull request scylladb#121 from julienrf/test-cassandra

Expand the tests to cover migrations from Cassandra, ScyllaDB, and Parquet

.github/attempt.sh

@@ -0,0 +1,11 @@
+attempt () {
+  command=$1
+  attempts=0
+  max_attempts=240
+  while ! eval "$command" ; do
+      [[ $attempts -ge $max_attempts ]] && echo "Failed!" && exit 1
+      attempts=$((attempts+1))
+      sleep 1;
+      echo "waiting... (${attempts}/${max_attempts})"
+  done
+}

.github/wait-for-cql.sh

@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+
+service=$1
+
+source .github/attempt.sh
+
+echo "Waiting for CQL to be ready in service ${service}"
+attempt 'docker compose -f docker-compose-tests.yml exec ${service} bash -c "cqlsh -e '"'"'describe cluster'"'"'" > /dev/null'

.github/wait-for-port.sh

@@ -1,12 +1,8 @@
 #!/usr/bin/env bash
 
 port=$1
+
+source .github/attempt.sh
+
 echo "Waiting for successful HTTP status code on port ${port}"
-attempts=0
-max_attempts=60
-while ! curl -s "http://127.0.0.1:$port" > /dev/null ; do
-    [[ $attempts -ge $max_attempts ]] && echo "Failed!" && exit 1
-    attempts=$((attempts+1))
-    sleep 1;
-    echo "waiting... (${attempts}/${max_attempts})"
-done
+attempt 'curl -s "http://127.0.0.1:$port" > /dev/null'

.github/workflows/tests.yml

@@ -25,8 +25,11 @@ jobs:
       - name: Set up services
         run: |
           docker compose -f docker-compose-tests.yml up -d
-          .github/wait-for-port.sh 8000 # ScyllaDB
+          .github/wait-for-port.sh 8000 # ScyllaDB Alternator
           .github/wait-for-port.sh 8001 # DynamoDB
+          .github/wait-for-cql.sh scylla
+          .github/wait-for-cql.sh cassandra
+          .github/wait-for-cql.sh scylla-source
           .github/wait-for-port.sh 8080 # Spark master
           .github/wait-for-port.sh 8081 # Spark worker
       - name: Run tests

build.sbt

@@ -6,7 +6,8 @@ val sparkVersion = "2.4.4"
 inThisBuild(
   List(
     organization := "com.scylladb",
-    scalaVersion := "2.11.12"
+    scalaVersion := "2.11.12",
+    scalacOptions += "-target:jvm-1.8"
   )
 )
 
@@ -74,6 +75,9 @@ lazy val migrator = (project in file("migrator")).settings(
 lazy val tests = project.in(file("tests")).settings(
   libraryDependencies ++= Seq(
     "com.amazonaws" % "aws-java-sdk-dynamodb" % awsSdkVersion,
+    "org.apache.cassandra" % "java-driver-query-builder" % "4.18.0",
+    "com.github.mjakubowski84" %% "parquet4s-core" % "1.9.4",
+    "org.apache.hadoop" % "hadoop-client" % "2.9.2",
     "org.scalameta" %% "munit" % "0.7.29",
     "org.scala-lang.modules" %% "scala-collection-compat" % "2.11.0"
   ),

docker-compose-tests.yml

@@ -14,6 +14,30 @@ services:
       - "8001:8000"
     working_dir: /home/dynamodblocal
 
+  cassandra:
+    image: cassandra:latest
+    networks:
+      - scylla
+    volumes:
+      - ./tests/docker/cassandra:/var/lib/cassandra
+    ports:
+      - "9043:9042"
+    expose:
+      - 9043
+
+  scylla-source:
+    image: scylladb/scylla:latest
+    networks:
+      - scylla
+    volumes:
+      - "./tests/docker/scylla-source:/var/lib/scylla"
+    ports:
+      - "8002:8000"
+      - "9044:9042"
+    expose:
+      - 9044
+    command: "--smp 1 --memory 2048M --alternator-port 8000 --alternator-write-isolation only_rmw_uses_lwt"
+
   scylla:
     image: scylladb/scylla:latest
     networks:
@@ -22,7 +46,8 @@ services:
       - "./tests/docker/scylla:/var/lib/scylla"
     ports:
       - "8000:8000"
-    command: "--smp 2 --memory 2048M --alternator-port 8000 --alternator-write-isolation always_use_lwt"
+      - "9042:9042"
+    command: "--smp 1 --memory 2048M --alternator-port 8000 --alternator-write-isolation only_rmw_uses_lwt"
 
   spark-master:
     image: bde2020/spark-master:2.4.4-hadoop2.7
@@ -54,6 +79,8 @@ services:
     volumes:
       - ./migrator/target/scala-2.11:/jars
       - ./tests/src/test/configurations:/app/configurations
+      - ./tests/docker/spark-master:/app/savepoints
+      - ./tests/docker/parquet:/app/parquet
       # Workaround for https://github.com/awslabs/emr-dynamodb-connector/issues/50
       - ${PWD}/tests/docker/job-flow.json:/mnt/var/lib/info/job-flow.json
 
@@ -78,6 +105,8 @@ services:
     ports:
       - 5006:5006
       - 8081:8081
+    volumes:
+      - ./tests/docker/parquet:/app/parquet
     depends_on:
       - spark-master
 

tests/docker/.gitignore

@@ -0,0 +1,3 @@
+cassandra/
+scylla/
+spark-master/

tests/docker/parquet/.gitignore

@@ -0,0 +1,4 @@
+# Ignore everything in this directory
+*
+# Except this file
+!.gitignore

tests/src/test/configurations/cassandra-to-scylla-basic.yaml

@@ -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

tests/src/test/configurations/cassandra-to-scylla-renames.yaml

@@ -0,0 +1,45 @@
+source:
+  type: cassandra
+  host: cassandra
+  port: 9042
+  localDC: datacenter1
+  credentials:
+    username: dummy
+    password: dummy
+  keyspace: test
+  table: renameditems
+  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: renameditems
+  consistencyLevel: LOCAL_QUORUM
+  connections: 16
+  stripTrailingZerosForDecimals: false
+
+renames:
+  - from: bar
+    to: quux
+
+savepoints:
+  path: /app/savepoints
+  intervalSeconds: 300
+skipTokenRanges: []
+validation:
+  compareTimestamps: true
+  ttlToleranceMillis: 60000
+  writetimeToleranceMillis: 1000
+  failuresToFetch: 100
+  floatingPointTolerance: 0.001
+  timestampMsTolerance: 0

tests/src/test/configurations/parquet-to-scylla-basic.yaml

@@ -0,0 +1,31 @@
+source:
+  type: parquet
+  path: /app/parquet/basic.parquet
+
+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

tests/src/test/configurations/scylla-to-scylla-basic.yaml

@@ -0,0 +1,43 @@
+source:
+  type: scylla
+  host: scylla-source
+  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

tests/src/test/resources/application.conf

@@ -0,0 +1,5 @@
+datastax-java-driver {
+  basic.request {
+    timeout = 10 seconds
+  }
+}

tests/src/test/scala/com/scylladb/migrator/CassandraUtils.scala

@@ -0,0 +1,37 @@
+package com.scylladb.migrator
+
+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
+
+object CassandraUtils {
+
+  /**
+   * Prepare a Scylla table for a test. Drop any existing table of the same name and recreates it.
+   *
+   * @param database   Database session to use
+   * @param keyspace   Keyspace name
+   * @param name       Name of the table to create
+   * @param columnName Function to possible transform the initial name of the columns
+   */
+  def dropAndRecreateTable(database: CqlSession, keyspace: String, name: String, columnName: String => String): Unit = {
+    val dropTableStatement =
+      SchemaBuilder
+        .dropTable(keyspace, name)
+        .ifExists()
+        .build()
+    database
+      .execute(dropTableStatement)
+      .ensuring(_.wasApplied())
+    val createTableStatement =
+      SchemaBuilder
+        .createTable(keyspace, name)
+        .withPartitionKey("id", DataTypes.TEXT)
+        .withColumn(columnName("foo"), DataTypes.TEXT)
+        .withColumn(columnName("bar"), DataTypes.INT)
+        .build()
+    database
+      .execute(createTableStatement)
+      .ensuring(_.wasApplied())
+  }
+}

tests/src/test/scala/com/scylladb/migrator/SparkUtils.scala

@@ -0,0 +1,44 @@
+package com.scylladb.migrator
+
+import scala.sys.process.Process
+
+object SparkUtils {
+
+  /**
+   * Run a migration by submitting a Spark job to the Spark cluster
+   * and waiting for its successful completion.
+   *
+   * @param configFile 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 submitMigrationJob(configFile: String): Unit = {
+    val process =
+      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/${configFile}",
+          // 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"
+        )
+      )
+    process
+      .run()
+      .exitValue()
+      .ensuring(statusCode => statusCode == 0, "Spark job failed")
+    ()
+  }
+
+}