Fix AWS credentials forwarding

The GitHub action we use in our workflow tests-aws.yml exposes the AWS credentials by setting environment variables. We copy them to the Spark containers.

CONTRIBUTING.md

@@ -28,13 +28,19 @@ Tests are implemented in the `tests` sbt submodule. They simulate the submission
    sbt testOnly com.scylladb.migrator.BasicMigrationTest
    ~~~
 
-  Or, to run the tests that access AWS, first configure your AWS credentials with `aws configure`, and then:
+  Or, to run the tests that access AWS:
 
   ~~~ sh
-  AWS_REGION=us-east-1 \
+  export AWS_REGION=us-east-1
+  export AWS_ACCESS_KEY_ID=xxx
+  export AWS_SECRET_ACCESS_KEY=yyy
+  export AWS_SESSION_TOKEN=zzz
+  docker compose -f docker-compose-tests.yml up -d
   sbt "testOnly -- --include-categories=com.scylladb.migrator.AWS"
   ~~~
 
+  Replace `xxx`, `yyy`, and `zzz` with your actual credentials.
+
 4. Ultimately, stop the Docker containers
 
    ~~~ sh

build.sbt

@@ -50,7 +50,7 @@ lazy val migrator = (project in file("migrator")).enablePlugins(BuildInfoPlugin)
     "software.amazon.awssdk"    % "sts"      % awsSdkVersion,
     "com.scylladb" %% "spark-scylladb-connector" % "4.0.0",
     "com.github.jnr" % "jnr-posix" % "3.1.19", // Needed by the Spark ScyllaDB connector
-    "com.scylladb.alternator" % "emr-dynamodb-hadoop" % "5.7.1",
+    "com.scylladb.alternator" % "emr-dynamodb-hadoop" % "5.8.0",
     "com.scylladb.alternator" % "load-balancing" % "1.0.0",
     "io.circe"       %% "circe-generic"      % "0.14.7",
     "io.circe"       %% "circe-parser"       % "0.14.7",

docker-compose-tests.yml

@@ -53,6 +53,10 @@ services:
     command: master
     environment:
       SPARK_PUBLIC_DNS: localhost
+      AWS_REGION: ${AWS_REGION}
+      AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
+      AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
+      AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN}
     expose:
       - 5005
     ports:
@@ -63,7 +67,6 @@ services:
       - ./migrator/target/scala-2.13:/jars
       - ./tests/src/test/configurations:/app/configurations
       - ./tests/docker/spark-master:/app/savepoints
-      - ./tests/docker/aws-profile:/root/.aws
       - ./tests/docker/parquet:/app/parquet
 
   spark-worker:
@@ -74,6 +77,10 @@ services:
       SPARK_WORKER_MEMORY: 4G
       SPARK_WORKER_WEBUI_PORT: 8081
       SPARK_PUBLIC_DNS: localhost
+      AWS_REGION: ${AWS_REGION}
+      AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
+      AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
+      AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN}
     expose:
       - 5006
     ports:

tests/src/test/configurations/dynamodb-to-alternator-streaming-skip-snapshot.yaml

@@ -1,11 +1,11 @@
 source:
   type: dynamodb
-  table: StreamedItemsSkipSnapshotTest
+  table: migrator-StreamedItemsSkipSnapshotTest
   region: us-east-1
 
 target:
   type: dynamodb
-  table: StreamedItemsSkipSnapshotTest
+  table: migrator-StreamedItemsSkipSnapshotTest
   region: dummy
   endpoint:
     host: http://scylla

tests/src/test/configurations/dynamodb-to-alternator-streaming.yaml

@@ -1,11 +1,11 @@
 source:
   type: dynamodb
-  table: StreamedItemsTest
+  table: migrator-StreamedItemsTest
   region: us-east-1
 
 target:
   type: dynamodb
-  table: StreamedItemsTest
+  table: migrator-StreamedItemsTest
   region: dummy
   endpoint:
     host: http://scylla

tests/src/test/scala/com/scylladb/migrator/alternator/MigratorSuite.scala

@@ -198,20 +198,7 @@ abstract class MigratorSuiteWithAWS extends MigratorSuite {
   lazy val sourceDDb: Fixture[DynamoDbClient] = new Fixture[DynamoDbClient]("sourceDDb") {
     private var client: DynamoDbClient = null
     def apply(): DynamoDbClient = client
-    override def beforeAll(): Unit = {
-      // Provision the AWS credentials on the Spark nodes via a Docker volume
-      val localAwsCredentials =
-        Paths.get(sys.props("user.home"), ".aws", "credentials")
-          .toAbsolutePath
-      (s"cp ${localAwsCredentials} docker/aws-profile/credentials").!!
-
-      val region = Region.of(sys.env("AWS_REGION"))
-      client =
-        DynamoDbClient
-          .builder()
-          .region(region)
-          .build()
-    }
+    override def beforeAll(): Unit = client = DynamoDbClient.create()
     override def afterAll(): Unit = client.close()
   }
 

tests/src/test/scala/com/scylladb/migrator/alternator/StreamedItemsTest.scala

@@ -10,9 +10,9 @@ import scala.util.chaining.scalaUtilChainingOps
 
 class StreamedItemsTest extends MigratorSuiteWithAWS {
 
-  override val munitTimeout: Duration = 120.seconds
+  override val munitTimeout: Duration = 600.seconds
 
-  withTable("StreamedItemsTest").test("Stream changes") { tableName =>
+  withTable("migrator-StreamedItemsTest").test("Stream changes") { tableName =>
     val configFileName = "dynamodb-to-alternator-streaming.yaml"
 
     // Populate the source table
@@ -27,7 +27,7 @@ class StreamedItemsTest extends MigratorSuiteWithAWS {
       submitSparkJobProcess(configFileName, "com.scylladb.migrator.Migrator")
         .run(ProcessLogger { log =>
           sparkLogs ++= log
-//          println(log) // Uncomment to see the logs
+          println(log)
         })
 
     awaitAtMost(60.seconds) {
@@ -53,7 +53,7 @@ class StreamedItemsTest extends MigratorSuiteWithAWS {
     sourceDDb().putItem(PutItemRequest.builder().tableName(tableName).item(item2Data.asJava).build())
 
     // Check that the added item has also been migrated
-    awaitAtMost(60.seconds) {
+    awaitAtMost(300.seconds) {
       targetAlternator()
         .getItem(GetItemRequest.builder().tableName(tableName).key(keys2.asJava).build())
         .tap { itemResult =>
@@ -71,7 +71,7 @@ class StreamedItemsTest extends MigratorSuiteWithAWS {
     deleteStreamTable(tableName)
   }
 
-  withTable("StreamedItemsSkipSnapshotTest").test("Stream changes but skip initial snapshot") { tableName =>
+  withTable("migrator-StreamedItemsSkipSnapshotTest").test("Stream changes but skip initial snapshot") { tableName =>
     val configFileName = "dynamodb-to-alternator-streaming-skip-snapshot.yaml"
 
     // Populate the source table
@@ -86,7 +86,7 @@ class StreamedItemsTest extends MigratorSuiteWithAWS {
       submitSparkJobProcess(configFileName, "com.scylladb.migrator.Migrator")
         .run(ProcessLogger { (log: String) =>
           sparkLogs ++= log
-//          println(log) // Uncomment to see the logs
+          println(log)
         })
 
     // Wait for the changes to start being streamed
@@ -104,7 +104,7 @@ class StreamedItemsTest extends MigratorSuiteWithAWS {
     sourceDDb().putItem(PutItemRequest.builder().tableName(tableName).item(item2Data.asJava).build())
 
     // Check that only the second item has been migrated
-    awaitAtMost(60.seconds) {
+    awaitAtMost(300.seconds) {
       targetAlternator()
         .getItem(GetItemRequest.builder().tableName(tableName).key(keys2.asJava).build())
         .tap { itemResult =>