diff --git a/query-processor/src/main/java/com/github/koop/queryprocessor/gateway/Main.java b/query-processor/src/main/java/com/github/koop/queryprocessor/gateway/Main.java index a9774e0..af3c175 100644 --- a/query-processor/src/main/java/com/github/koop/queryprocessor/gateway/Main.java +++ b/query-processor/src/main/java/com/github/koop/queryprocessor/gateway/Main.java @@ -371,6 +371,7 @@ private static void putOrUploadPartHandler(Context ctx, StorageService storage) } ctx.status(200); + ctx.header("ETag", "\"" + UUID.randomUUID().toString().replace("-", "") + "\""); ctx.result(""); } else { InputStream data = ctx.bodyInputStream(); @@ -381,6 +382,7 @@ private static void putOrUploadPartHandler(Context ctx, StorageService storage) return; } ctx.status(200); + ctx.header("ETag", "\"" + UUID.randomUUID().toString().replace("-", "") + "\""); ctx.result(""); } } catch (UnsupportedOperationException e) { @@ -572,11 +574,13 @@ private static String buildInitiateMultipartUploadXml(String bucket, String key, private static String buildCompleteMultipartUploadXml(String bucket, String key) { String encodedKey = URLEncoder.encode(key, StandardCharsets.UTF_8).replace("+", "%20"); + String etag = UUID.randomUUID().toString().replace("-", ""); return "\n" + "\n" + " http://localhost:8080/" + escapeXml(bucket) + "/" + encodedKey + "\n" + " " + escapeXml(bucket) + "\n" + " " + escapeXml(key) + "\n" + + " \"" + etag + "\"\n" + ""; } diff --git a/system-tests/pom.xml b/system-tests/pom.xml index 116a307..2e28e95 100644 --- a/system-tests/pom.xml +++ b/system-tests/pom.xml @@ -54,6 +54,14 @@ test + + + software.amazon.awssdk + s3-transfer-manager + 2.41.34 + test + + org.testcontainers testcontainers diff --git a/system-tests/src/test/java/koop/Boto3UploadFileIT.java b/system-tests/src/test/java/koop/Boto3UploadFileIT.java new file mode 100644 index 0000000..1b1afeb --- /dev/null +++ b/system-tests/src/test/java/koop/Boto3UploadFileIT.java @@ -0,0 +1,393 @@ +package koop; + +import com.github.koop.common.metadata.ErasureSetConfiguration; +import com.github.koop.common.metadata.ErasureSetConfiguration.ErasureSet; +import com.github.koop.common.metadata.ErasureSetConfiguration.Machine; +import com.github.koop.common.metadata.MemoryFetcher; +import com.github.koop.common.metadata.MetadataClient; +import com.github.koop.common.metadata.PartitionSpreadConfiguration; +import com.github.koop.common.metadata.PartitionSpreadConfiguration.PartitionSpread; +import com.github.koop.common.pubsub.MemoryPubSub; +import com.github.koop.common.pubsub.PubSubClient; +import com.github.koop.queryprocessor.gateway.Main; +import com.github.koop.queryprocessor.gateway.StorageServices.StorageService; +import com.github.koop.queryprocessor.gateway.StorageServices.StorageWorkerService; +import com.github.koop.queryprocessor.processor.CommitCoordinator; +import com.github.koop.queryprocessor.processor.StorageWorker; +import com.github.koop.queryprocessor.processor.cache.MemoryCacheClient; +import com.github.koop.storagenode.RocksDbRepairQueue; +import com.github.koop.storagenode.StorageNodeServerV2; +import com.github.koop.storagenode.db.Database; +import com.github.koop.storagenode.db.RocksDbStorageStrategy; +import io.javalin.Javalin; +import org.junit.jupiter.api.*; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.core.ResponseBytes; +import software.amazon.awssdk.core.checksums.RequestChecksumCalculation; +import software.amazon.awssdk.core.checksums.ResponseChecksumValidation; +import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration; +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.core.sync.ResponseTransformer; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.s3.S3AsyncClient; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.S3Configuration; +import software.amazon.awssdk.services.s3.model.*; +import software.amazon.awssdk.services.s3.multipart.MultipartConfiguration; +import software.amazon.awssdk.transfer.s3.S3TransferManager; +import software.amazon.awssdk.transfer.s3.model.CompletedFileUpload; +import software.amazon.awssdk.transfer.s3.model.FileUpload; +import software.amazon.awssdk.transfer.s3.model.UploadFileRequest; + +import java.io.IOException; +import java.io.RandomAccessFile; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.URI; +import java.nio.file.Files; +import java.nio.file.Path; +import java.security.SecureRandom; +import java.time.LocalTime; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Integration tests for file-based uploads via the Java S3 SDK — the equivalent + * of boto3's upload_file. Three approaches are exercised: + * + *
    + *
  1. Single-part PUT using {@code RequestBody.fromFile()} — reads directly from disk.
    2. + *
  2. Multipart upload reading file chunks with {@code RandomAccessFile}.
    3. + *
  3. {@link S3TransferManager#uploadFile} — the high-level SDK equivalent of + * boto3's {@code s3.upload_file()}, which selects single- or multi-part + * automatically based on file size.
    4. + *
+ */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class Boto3UploadFileIT { + + private static final int TOTAL_NODES = 9; + private static final String BUCKET = "upload-file-bucket"; + + private final List servers = new ArrayList<>(); + private final List dataDirs = new ArrayList<>(); + private final List databases = new ArrayList<>(); + + private MemoryFetcher sharedFetcher; + private MemoryPubSub sharedPubSub; + private MetadataClient sharedMetadataClient; + private PubSubClient sharedPubSubClient; + private CommitCoordinator commitCoordinator; + private StorageWorker worker; + + private Javalin app; + private S3Client s3; + private S3AsyncClient asyncS3; + private S3TransferManager transferManager; + + private static void log(String msg) { + System.out.printf("[%s] %s%n", LocalTime.now().withNano(0), msg); + } + + @BeforeEach + void startFullStack() throws Exception { + log("=== STARTING FULL STACK (Boto3UploadFileIT) ==="); + + sharedFetcher = new MemoryFetcher(); + sharedPubSub = new MemoryPubSub(); + + sharedPubSubClient = new PubSubClient(sharedPubSub); + sharedPubSubClient.start(); + + sharedMetadataClient = new MetadataClient(sharedFetcher); + sharedMetadataClient.start(); + + commitCoordinator = new CommitCoordinator(sharedPubSubClient, 0, 10); + worker = new StorageWorker(sharedMetadataClient, commitCoordinator); + + List addrs = new ArrayList<>(); + for (int i = 0; i < TOTAL_NODES; i++) { + int port = freePort(); + Path dir = Files.createTempDirectory("upload-file-node-" + i + "-"); + + RocksDbStorageStrategy strategy = new RocksDbStorageStrategy(dir.resolve("db").toString()); + RocksDbRepairQueue repairQueue = new RocksDbRepairQueue(strategy); + Database db = new Database(strategy); + StorageNodeServerV2 server = + new StorageNodeServerV2(port, "127.0.0.1", db, dir.resolve("data"), + sharedMetadataClient, sharedPubSubClient, repairQueue); + + servers.add(server); + dataDirs.add(dir); + databases.add(db); + addrs.add(new InetSocketAddress("127.0.0.1", port)); + } + + ErasureSetConfiguration esConfig = new ErasureSetConfiguration(); + ErasureSet es = new ErasureSet(); + es.setNumber(1); + es.setN(9); + es.setM(6); + es.setWriteQuorum(7); + List machines = new ArrayList<>(); + for (InetSocketAddress addr : addrs) { + Machine m = new Machine(); + m.setIp(addr.getHostString()); + m.setPort(addr.getPort()); + machines.add(m); + } + es.setMachines(machines); + esConfig.setErasureSets(List.of(es)); + + PartitionSpreadConfiguration psConfig = new PartitionSpreadConfiguration(); + PartitionSpread ps = new PartitionSpread(); + ps.setErasureSet(1); + List parts = new ArrayList<>(); + for (int i = 0; i < 3; i++) parts.add(i); + ps.setPartitions(parts); + psConfig.setPartitionSpread(List.of(ps)); + + sharedFetcher.update(esConfig); + sharedFetcher.update(psConfig); + + servers.forEach(StorageNodeServerV2::start); + + MemoryCacheClient cache = new MemoryCacheClient(); + StorageService storage = new StorageWorkerService(worker, cache); + app = Main.createApp(storage).start(0); + + URI endpoint = URI.create("http://localhost:" + app.port()); + StaticCredentialsProvider creds = StaticCredentialsProvider.create( + AwsBasicCredentials.create("fake-access-key", "fake-secret-key")); + S3Configuration s3Config = S3Configuration.builder() + .pathStyleAccessEnabled(true) + .chunkedEncodingEnabled(false) + .build(); + + s3 = S3Client.builder() + .endpointOverride(endpoint) + .credentialsProvider(creds) + .region(Region.US_EAST_1) + .serviceConfiguration(s3Config) + .requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED) + .responseChecksumValidation(ResponseChecksumValidation.WHEN_REQUIRED) + .build(); + + // Async client with a low multipart threshold so the transfer manager + // exercises multipart upload even for moderately sized test files. + asyncS3 = S3AsyncClient.builder() + .endpointOverride(endpoint) + .credentialsProvider(creds) + .region(Region.US_EAST_1) + .serviceConfiguration(s3Config) + .multipartEnabled(true) + .requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED) + .responseChecksumValidation(ResponseChecksumValidation.WHEN_REQUIRED) + .multipartConfiguration(MultipartConfiguration.builder() + .thresholdInBytes(1024 * 1024L) // 1 MB + .minimumPartSizeInBytes(1024 * 1024L) // 1 MB per part + .build()) + .overrideConfiguration(ClientOverrideConfiguration.builder() + .addExecutionInterceptor(new UnsignedPayloadInterceptor()) // Bypasses the hash requirement + .build()) + .build(); + + transferManager = S3TransferManager.builder() + .s3Client(asyncS3) + .build(); + + log("=== FULL STACK READY on port " + app.port() + " ==="); + } + + @AfterEach + void stopFullStack() throws Exception { + if (transferManager != null) transferManager.close(); + if (asyncS3 != null) asyncS3.close(); + if (s3 != null) s3.close(); + if (app != null) app.stop(); + if (worker != null) worker.shutdown(); + if (sharedMetadataClient != null) sharedMetadataClient.close(); + if (sharedPubSubClient != null) sharedPubSubClient.close(); + + for (StorageNodeServerV2 server : servers) server.stop(); + for (Database db : databases) { try { db.close(); } catch (Exception ignored) {} } + for (Path d : dataDirs) deleteRecursive(d); + + servers.clear(); + databases.clear(); + dataDirs.clear(); + + log("=== FULL STACK STOPPED ==="); + } + + // ===================================================================== + // TEST: single-part upload from file (RequestBody.fromFile) + // ===================================================================== + @Test + void uploadFile_singlePart_roundTrip() throws Exception { + log("\n--- TEST: uploadFile single-part (RequestBody.fromFile) ---"); + + byte[] data = new byte[2 * 1024 * 1024]; // 2 MB + new SecureRandom().nextBytes(data); + String key = "upload-file-single.bin"; + + Path tmpFile = Files.createTempFile("upload-single-", ".bin"); + try { + Files.write(tmpFile, data); + + log("Step 1: PUT from file path via RequestBody.fromFile"); + s3.putObject( + PutObjectRequest.builder() + .bucket(BUCKET) + .key(key) + .contentLength((long) data.length) + .build(), + RequestBody.fromFile(tmpFile) + ); + log("Step 1: PUT completed"); + + log("Step 2: GET and verify"); + ResponseBytes response = s3.getObject( + GetObjectRequest.builder().bucket(BUCKET).key(key).build(), + ResponseTransformer.toBytes() + ); + assertArrayEquals(data, response.asByteArray(), + "Downloaded bytes must match what was uploaded from file"); + log("--- TEST PASSED ---\n"); + } finally { + Files.deleteIfExists(tmpFile); + } + } + + // ===================================================================== + // TEST: multipart upload reading parts from disk via RandomAccessFile + // ===================================================================== + @Test + void uploadFile_multipart_roundTrip() throws Exception { + log("\n--- TEST: uploadFile multipart (parts read from file) ---"); + + int partSize = 1024 * 1024; // 1 MB per part + int numParts = 5; + byte[] data = new byte[numParts * partSize]; + new SecureRandom().nextBytes(data); + String key = "upload-file-multipart.bin"; + + Path tmpFile = Files.createTempFile("upload-multi-", ".bin"); + try { + Files.write(tmpFile, data); + + log("Step 1: CreateMultipartUpload"); + CreateMultipartUploadResponse initResp = s3.createMultipartUpload( + CreateMultipartUploadRequest.builder().bucket(BUCKET).key(key).build() + ); + String uploadId = initResp.uploadId(); + log("Step 1: Upload ID = " + uploadId); + + log("Step 2: Uploading " + numParts + " parts from file"); + List completedParts = new ArrayList<>(); + try (RandomAccessFile raf = new RandomAccessFile(tmpFile.toFile(), "r")) { + for (int i = 0; i < numParts; i++) { + byte[] chunk = new byte[partSize]; + raf.readFully(chunk); + + UploadPartResponse partResp = s3.uploadPart( + UploadPartRequest.builder() + .bucket(BUCKET) + .key(key) + .uploadId(uploadId) + .partNumber(i + 1) + .contentLength((long) partSize) + .build(), + RequestBody.fromBytes(chunk) + ); + log(" -> Part " + (i + 1) + " ETag: " + partResp.eTag()); + completedParts.add(CompletedPart.builder() + .partNumber(i + 1) + .eTag(partResp.eTag()) + .build()); + } + } + + log("Step 3: CompleteMultipartUpload"); + CompleteMultipartUploadResponse completeResp = s3.completeMultipartUpload( + CompleteMultipartUploadRequest.builder() + .bucket(BUCKET) + .key(key) + .uploadId(uploadId) + .multipartUpload(CompletedMultipartUpload.builder().parts(completedParts).build()) + .build() + ); + log("Step 3: Completed. ETag = " + completeResp.eTag()); + assertNotNull(completeResp.eTag(), "CompleteMultipartUpload response must include an ETag"); + + log("Step 4: GET and verify " + data.length + " bytes"); + ResponseBytes getResp = s3.getObject( + GetObjectRequest.builder().bucket(BUCKET).key(key).build(), + ResponseTransformer.toBytes() + ); + assertArrayEquals(data, getResp.asByteArray(), + "Downloaded bytes must match the multipart-uploaded file content"); + log("--- TEST PASSED ---\n"); + } finally { + Files.deleteIfExists(tmpFile); + } + } + + // ===================================================================== + // TEST: S3TransferManager.uploadFile — the direct Java SDK equivalent + // of boto3's s3.upload_file(). The async client is configured with a + // 1 MB multipart threshold so this exercises the full multipart path. + // ===================================================================== + @Test + void uploadFile_transferManager_roundTrip() throws Exception { + log("\n--- TEST: S3TransferManager.uploadFile (boto3 upload_file equivalent) ---"); + + byte[] data = new byte[5 * 1024 * 1024]; // 5 MB → 5 × 1 MB parts + new SecureRandom().nextBytes(data); + String key = "transfer-manager-upload.bin"; + + Path tmpFile = Files.createTempFile("transfer-upload-", ".bin"); + try { + Files.write(tmpFile, data); + + log("Step 1: S3TransferManager.uploadFile from " + tmpFile); + FileUpload upload = transferManager.uploadFile(UploadFileRequest.builder() + .putObjectRequest(r -> r.bucket(BUCKET).key(key)) + .source(tmpFile) + .build()); + + CompletedFileUpload completed = upload.completionFuture().join(); + log("Step 1: Upload complete. HTTP status = " + completed.response().sdkHttpResponse().statusCode()); + + log("Step 2: GET and verify " + data.length + " bytes"); + ResponseBytes getResp = s3.getObject( + GetObjectRequest.builder().bucket(BUCKET).key(key).build(), + ResponseTransformer.toBytes() + ); + assertArrayEquals(data, getResp.asByteArray(), + "Data downloaded after S3TransferManager.uploadFile must match source file"); + log("--- TEST PASSED ---\n"); + } finally { + Files.deleteIfExists(tmpFile); + } + } + + private static int freePort() throws IOException { + try (ServerSocket ss = new ServerSocket(0)) { + ss.setReuseAddress(true); + return ss.getLocalPort(); + } + } + + private static void deleteRecursive(Path root) throws IOException { + if (!Files.exists(root)) return; + try (var paths = Files.walk(root)) { + paths.sorted(Comparator.reverseOrder()) + .forEach(p -> { try { Files.deleteIfExists(p); } catch (IOException ignored) {} }); + } + } +} diff --git a/system-tests/src/test/java/koop/DockerComposeS3E2EIT.java b/system-tests/src/test/java/koop/DockerComposeS3E2EIT.java index 6d510a6..e234cd2 100644 --- a/system-tests/src/test/java/koop/DockerComposeS3E2EIT.java +++ b/system-tests/src/test/java/koop/DockerComposeS3E2EIT.java @@ -8,16 +8,27 @@ import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; import software.amazon.awssdk.core.ResponseBytes; +import software.amazon.awssdk.core.checksums.RequestChecksumCalculation; +import software.amazon.awssdk.core.checksums.ResponseChecksumValidation; +import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration; import software.amazon.awssdk.core.sync.RequestBody; import software.amazon.awssdk.core.sync.ResponseTransformer; import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.s3.S3AsyncClient; import software.amazon.awssdk.services.s3.S3Client; import software.amazon.awssdk.services.s3.S3Configuration; import software.amazon.awssdk.services.s3.model.*; +import software.amazon.awssdk.services.s3.multipart.MultipartConfiguration; +import software.amazon.awssdk.transfer.s3.S3TransferManager; +import software.amazon.awssdk.transfer.s3.model.CompletedFileUpload; +import software.amazon.awssdk.transfer.s3.model.FileUpload; +import software.amazon.awssdk.transfer.s3.model.UploadFileRequest; import java.io.File; import java.net.URI; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; import java.security.SecureRandom; import java.time.LocalTime; import java.util.ArrayList; @@ -31,6 +42,8 @@ public class DockerComposeS3E2EIT { private static final String BUCKET = "e2e-bucket"; private S3Client s3; + private S3AsyncClient asyncS3; + private S3TransferManager transferManager; @Container public static ComposeContainer environment = @@ -52,15 +65,42 @@ void setupS3Client() { String host = "127.0.0.1"; Integer port = 9001; + URI endpoint = URI.create("http://" + host + ":" + port); + StaticCredentialsProvider creds = StaticCredentialsProvider.create( + AwsBasicCredentials.create("fake-access-key", "fake-secret-key")); + S3Configuration s3Config = S3Configuration.builder() + .pathStyleAccessEnabled(true) + .chunkedEncodingEnabled(false) + .build(); + s3 = S3Client.builder() - .endpointOverride(URI.create("http://" + host + ":" + port)) - .credentialsProvider(StaticCredentialsProvider.create( - AwsBasicCredentials.create("fake-access-key", "fake-secret-key"))) + .endpointOverride(endpoint) + .credentialsProvider(creds) + .region(Region.US_EAST_1) + .serviceConfiguration(s3Config) + .requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED) + .responseChecksumValidation(ResponseChecksumValidation.WHEN_REQUIRED) + .build(); + + asyncS3 = S3AsyncClient.builder() + .endpointOverride(endpoint) + .credentialsProvider(creds) .region(Region.US_EAST_1) - .serviceConfiguration(S3Configuration.builder() - .pathStyleAccessEnabled(true) - .chunkedEncodingEnabled(false) + .serviceConfiguration(s3Config) + .multipartEnabled(true) + .requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED) + .responseChecksumValidation(ResponseChecksumValidation.WHEN_REQUIRED) + .multipartConfiguration(MultipartConfiguration.builder() + .thresholdInBytes(1024 * 1024L) + .minimumPartSizeInBytes(1024 * 1024L) .build()) + .overrideConfiguration(ClientOverrideConfiguration.builder() + .addExecutionInterceptor(new UnsignedPayloadInterceptor()) // Bypasses the hash requirement + .build()) + .build(); + + transferManager = S3TransferManager.builder() + .s3Client(asyncS3) .build(); log("3. Polling the Gateway S3 API to check if it is awake (Max 3 minutes)..."); @@ -103,6 +143,8 @@ void teardown() { log("======================================================"); log("PHASE: TEARDOWN"); log("======================================================"); + if (transferManager != null) transferManager.close(); + if (asyncS3 != null) asyncS3.close(); if (s3 != null) { log("Closing S3 Java Client connection..."); s3.close(); @@ -224,6 +266,40 @@ void e2e_multipartUpload_fullLifecycle_thenGet() { log("--- TEST PASSED ---\n"); } + @Test + void e2e_uploadFile_transferManager_roundTrip() throws Exception { + log("\n--- TEST: S3TransferManager.uploadFile (boto3 upload_file equivalent) ---"); + + byte[] data = new byte[5 * 1024 * 1024]; // 5 MB → 5 × 1 MB parts + new SecureRandom().nextBytes(data); + String key = "transfer-manager-upload.bin"; + + Path tmpFile = Files.createTempFile("docker-transfer-", ".bin"); + try { + Files.write(tmpFile, data); + + log("Step 1: S3TransferManager.uploadFile from " + tmpFile); + FileUpload upload = transferManager.uploadFile(UploadFileRequest.builder() + .putObjectRequest(r -> r.bucket(BUCKET).key(key)) + .source(tmpFile) + .build()); + + CompletedFileUpload completed = upload.completionFuture().join(); + log("Step 1: Upload complete. HTTP status = " + completed.response().sdkHttpResponse().statusCode()); + + log("Step 2: GET and verify " + data.length + " bytes"); + ResponseBytes getResp = s3.getObject( + GetObjectRequest.builder().bucket(BUCKET).key(key).build(), + ResponseTransformer.toBytes() + ); + assertArrayEquals(data, getResp.asByteArray(), + "Data downloaded after S3TransferManager.uploadFile must match source file"); + log("--- TEST PASSED ---\n"); + } finally { + Files.deleteIfExists(tmpFile); + } + } + @Test @Disabled("Buggy will address") void e2e_createThenDeleteBucket_headReturns404() { diff --git a/system-tests/src/test/java/koop/UnsignedPayloadInterceptor.java b/system-tests/src/test/java/koop/UnsignedPayloadInterceptor.java new file mode 100644 index 0000000..c409106 --- /dev/null +++ b/system-tests/src/test/java/koop/UnsignedPayloadInterceptor.java @@ -0,0 +1,17 @@ +package koop; +import software.amazon.awssdk.core.interceptor.Context; +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; +import software.amazon.awssdk.core.interceptor.ExecutionInterceptor; +import software.amazon.awssdk.http.SdkHttpRequest; + +public class UnsignedPayloadInterceptor implements ExecutionInterceptor { + @Override + public SdkHttpRequest modifyHttpRequest(Context.ModifyHttpRequest context, ExecutionAttributes executionAttributes) { + if (!context.httpRequest().headers().containsKey("x-amz-content-sha256")) { + return context.httpRequest().toBuilder() + .putHeader("x-amz-content-sha256", "UNSIGNED-PAYLOAD") + .build(); + } + return context.httpRequest(); + } +} \ No newline at end of file