feat: add support for s3 multipart upload

grusell · grusell · commit d8e530a7e28b · 2025-09-17T12:06:56.000+02:00
Signed-off-by: Gustav Grusell &lt;gustav.grusell@eyevinn.se&gt;
diff --git a/encore-common/build.gradle.kts b/encore-common/build.gradle.kts
@@ -16,8 +16,9 @@ dependencies {
 
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core")
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-slf4j")
-    implementation(platform("software.amazon.awssdk:bom:2.29.2"))
+    implementation(platform("software.amazon.awssdk:bom:2.33.9"))
     implementation("software.amazon.awssdk:s3")
+    implementation("software.amazon.awssdk:s3-transfer-manager")
 
     testImplementation(project(":encore-web"))
     testImplementation("org.springframework.security:spring-security-test")
diff --git a/encore-common/src/main/kotlin/se/svt/oss/encore/S3RemoteFilesConfiguration.kt b/encore-common/src/main/kotlin/se/svt/oss/encore/S3RemoteFilesConfiguration.kt
@@ -17,6 +17,7 @@ import software.amazon.awssdk.regions.Region
 import software.amazon.awssdk.services.s3.S3AsyncClient
 import software.amazon.awssdk.services.s3.S3Configuration
 import software.amazon.awssdk.services.s3.presigner.S3Presigner
+import software.amazon.awssdk.transfer.s3.S3TransferManager
 import java.net.URI
 
 @ConditionalOnProperty("remote-files.s3.enabled", havingValue = "true")
@@ -28,6 +29,12 @@ class S3RemoteFilesConfiguration {
     fun s3Region() =
         Region.of(System.getProperty("aws.region") ?: System.getenv("AWS_REGION") ?: "us-east-1")
 
+    @Bean
+    fun transferManager(s3Client: S3AsyncClient, s3Properties: S3Properties): S3TransferManager =
+        S3TransferManager.builder()
+            .s3Client(s3Client)
+            .build()
+
     @Bean
     fun s3Client(s3Region: Region, s3Properties: S3Properties) = S3AsyncClient.builder()
         .region(s3Region)
@@ -42,11 +49,11 @@ class S3RemoteFilesConfiguration {
             if (s3Properties.anonymousAccess) {
                 AnonymousCredentialsProvider.create()
             } else {
-                DefaultCredentialsProvider.create()
+                DefaultCredentialsProvider.builder().build()
             },
         )
         .apply {
-            if (!s3Properties.endpoint.isNullOrBlank()) {
+            if (s3Properties.endpoint.isNotBlank()) {
                 endpointOverride(URI.create(s3Properties.endpoint))
             }
         }
@@ -76,6 +83,7 @@ class S3RemoteFilesConfiguration {
         s3Presigner: S3Presigner,
         s3Properties: S3Properties,
         s3UriConverter: S3UriConverter,
+        transferManager: S3TransferManager,
     ) =
-        S3RemoteFileHandler(s3Client, s3Presigner, s3Properties, s3UriConverter)
+        S3RemoteFileHandler(s3Client, s3Presigner, s3Properties, s3UriConverter, transferManager)
 }
diff --git a/encore-common/src/main/kotlin/se/svt/oss/encore/service/remotefiles/s3/S3Properties.kt b/encore-common/src/main/kotlin/se/svt/oss/encore/service/remotefiles/s3/S3Properties.kt
@@ -11,6 +11,7 @@ import java.time.Duration
 data class S3Properties(
     val enabled: Boolean = false,
     val anonymousAccess: Boolean = false,
+    val usePathStyle: Boolean = false,
     val endpoint: String = "",
     val presignDurationSeconds: Long = Duration.ofHours(12).seconds,
     val uploadTimeoutSeconds: Long = Duration.ofHours(1).seconds,
diff --git a/encore-common/src/main/kotlin/se/svt/oss/encore/service/remotefiles/s3/S3RemoteFileHandler.kt b/encore-common/src/main/kotlin/se/svt/oss/encore/service/remotefiles/s3/S3RemoteFileHandler.kt
@@ -11,6 +11,8 @@ import software.amazon.awssdk.services.s3.model.GetObjectRequest
 import software.amazon.awssdk.services.s3.model.PutObjectRequest
 import software.amazon.awssdk.services.s3.presigner.S3Presigner
 import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest
+import software.amazon.awssdk.transfer.s3.S3TransferManager
+import software.amazon.awssdk.transfer.s3.model.UploadFileRequest
 import java.net.URI
 import java.nio.file.Paths
 import java.util.concurrent.TimeUnit
@@ -22,6 +24,7 @@ class S3RemoteFileHandler(
     private val presigner: S3Presigner,
     private val s3Properties: S3Properties,
     private val s3UriConverter: S3UriConverter,
+    private val transferManager: S3TransferManager,
 ) : RemoteFileHandler {
 
     override fun getAccessUri(uri: String): String {
@@ -54,11 +57,30 @@ class S3RemoteFileHandler(
         log.info { "Uploading $localFile to $remoteFile" }
         val s3Uri = URI.create(remoteFile)
         val (bucket, key) = s3UriConverter.getBucketAndKey(s3Uri)
+        if (s3Properties.anonymousAccess) {
+            standardUpload(localFile, bucket, key)
+        } else {
+            transferManagerUpload(localFile, bucket, key)
+        }
+    }
+
+    private fun transferManagerUpload(localFile: String, bucket: String, key: String) {
+        val uploadRequest = UploadFileRequest.builder()
+            .putObjectRequest { por -> por.bucket(bucket).key(key) }
+            .source(Paths.get(localFile))
+            .build()
+        val fileUpload = transferManager.uploadFile(uploadRequest)
+        val res = fileUpload.completionFuture().get(s3Properties.uploadTimeoutSeconds, TimeUnit.SECONDS)
+        log.info { "Upload completed: $res" }
+    }
+
+    private fun standardUpload(localFile: String, bucket: String, key: String) {
         val putObjectRequest: PutObjectRequest = PutObjectRequest.builder()
             .bucket(bucket)
             .key(key)
             .build()
-        val res = client.putObject(putObjectRequest, Paths.get(localFile)).get(s3Properties.presignDurationSeconds, TimeUnit.SECONDS)
+        val res = client.putObject(putObjectRequest, Paths.get(localFile))
+            .get(s3Properties.presignDurationSeconds, TimeUnit.SECONDS)
         log.info { "Upload result: $res" }
     }
 
diff --git a/encore-common/src/main/kotlin/se/svt/oss/encore/service/remotefiles/s3/S3UriConverter.kt b/encore-common/src/main/kotlin/se/svt/oss/encore/service/remotefiles/s3/S3UriConverter.kt
@@ -4,6 +4,7 @@
 
 package se.svt.oss.encore.service.remotefiles.s3
 
+import org.springframework.web.util.UriComponentsBuilder
 import software.amazon.awssdk.regions.Region
 import java.net.URI
 
@@ -20,7 +21,17 @@ class S3UriConverter(
         val key = s3Uri.path.stripLeadingSlash()
 
         if (s3Properties.endpoint.isNotBlank()) {
-            return "https://$bucket.${s3Properties.endpoint}/$key"
+            val endpointUri = URI.create(s3Properties.endpoint)
+            val uriBuilder = UriComponentsBuilder.fromUri(endpointUri)
+            val pathSegments = key.split("/").toTypedArray()
+            if (s3Properties.usePathStyle) {
+                uriBuilder.pathSegment(bucket, *pathSegments)
+            } else {
+                uriBuilder
+                    .host("$bucket.${endpointUri.host}")
+                    .pathSegment(*pathSegments)
+            }
+            return uriBuilder.toUriString()
         }
         return "https://$bucket.s3.$region.amazonaws.com/$key"
     }
diff --git a/encore-common/src/test/kotlin/se/svt/oss/encore/EncoreS3IntegrationTest.kt b/encore-common/src/test/kotlin/se/svt/oss/encore/EncoreS3IntegrationTest.kt
@@ -11,11 +11,13 @@ import mu.KotlinLogging
 import org.awaitility.Durations
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Nested
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.extension.ExtendWith
 import org.junit.jupiter.api.io.TempDir
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.test.context.ActiveProfiles
+import org.springframework.test.context.TestPropertySource
 import se.svt.oss.encore.Assertions.assertThat
 import se.svt.oss.encore.model.Status
 import se.svt.oss.encore.model.callback.JobProgress
@@ -25,9 +27,7 @@ import java.io.File
 import java.nio.file.Paths
 
 @ExtendWith(S3StorageExtension::class)
-@ActiveProfiles(profiles = ["test-local", "test-s3"])
-@WireMockTest
-class EncoreS3IntegrationTest(wireMockRuntimeInfo: WireMockRuntimeInfo) : EncoreIntegrationTestBase(wireMockRuntimeInfo) {
+abstract class EncoreS3IntegrationTest(wireMockRuntimeInfo: WireMockRuntimeInfo) : EncoreIntegrationTestBase(wireMockRuntimeInfo) {
     private val log = KotlinLogging.logger {}
 
     @Autowired
@@ -61,7 +61,6 @@ class EncoreS3IntegrationTest(wireMockRuntimeInfo: WireMockRuntimeInfo) : Encore
         }
     }
 
-    @Test
     fun jobWiths3InputAndOutputIsSuccessful(@TempDir outputDir: File) {
         val filename = "test.mp4"
         val remoteInput = uploadInputfile(testFileSurround.file.absolutePath, filename)
@@ -104,4 +103,30 @@ class EncoreS3IntegrationTest(wireMockRuntimeInfo: WireMockRuntimeInfo) : Encore
 
         return "s3://$inputBucket/$key"
     }
+
+    @Nested
+    @ActiveProfiles(profiles = ["test-local", "test-s3"])
+    @WireMockTest
+    class StandardS3Access(wireMockRuntimeInfo: WireMockRuntimeInfo) : EncoreS3IntegrationTest(wireMockRuntimeInfo) {
+        @Test
+        fun jobWithS3InputAndOutputIsSuccessful(@TempDir outputDir: File) {
+            super.jobWiths3InputAndOutputIsSuccessful(outputDir)
+        }
+    }
+
+    @Nested
+    @ActiveProfiles(profiles = ["test-local", "test-s3"])
+    @TestPropertySource(
+        properties = [
+            "remote-files.s3.anonymous-access=true",
+            "remote-files.s3.use-path-style=true", // localstack requires path style access
+        ],
+    )
+    @WireMockTest
+    class AnonymousS3Access(wireMockRuntimeInfo: WireMockRuntimeInfo) : EncoreS3IntegrationTest(wireMockRuntimeInfo) {
+        @Test
+        fun jobWithS3InputAndOutputIsSuccessful(@TempDir outputDir: File) {
+            super.jobWiths3InputAndOutputIsSuccessful(outputDir)
+        }
+    }
 }
diff --git a/encore-common/src/test/kotlin/se/svt/oss/encore/service/remotefiles/S3UriConverterTest.kt b/encore-common/src/test/kotlin/se/svt/oss/encore/service/remotefiles/S3UriConverterTest.kt
diff --git a/encore-common/src/test/kotlin/se/svt/oss/encore/service/remotefiles/s3/S3UriConverterTest.kt b/encore-common/src/test/kotlin/se/svt/oss/encore/service/remotefiles/s3/S3UriConverterTest.kt
@@ -0,0 +1,89 @@
+// SPDX-FileCopyrightText: 2024 Eyevinn Technology AB
+//
+// SPDX-License-Identifier: EUPL-1.2
+
+package se.svt.oss.encore.service.remotefiles.s3
+
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.jupiter.api.Nested
+import org.junit.jupiter.api.Test
+import software.amazon.awssdk.regions.Region
+import java.net.URI
+
+class S3UriConverterTest {
+
+    private val s3Properties = S3Properties(enabled = true, anonymousAccess = true)
+    private val region = Region.of("eu-west-1")
+    private val s3Uri = URI.create("s3://my-bucket/test2/test1_x264_3100.mp4")
+    private val s3UriConverter = S3UriConverter(s3Properties, region)
+
+    @Nested
+    inner class ToHttp {
+
+        @Test
+        fun returnsCorrectUri() {
+            val httpUri = s3UriConverter.toHttp(s3Uri)
+            assertThat(httpUri).isEqualTo("https://my-bucket.s3.eu-west-1.amazonaws.com/test2/test1_x264_3100.mp4")
+        }
+
+        @Test
+        fun differentRegionReturnsCorrectUri() {
+            val s3UriConverter = S3UriConverter(s3Properties, Region.of("eu-north-1"))
+
+            val httpUri = s3UriConverter.toHttp(s3Uri)
+            assertThat(httpUri).isEqualTo("https://my-bucket.s3.eu-north-1.amazonaws.com/test2/test1_x264_3100.mp4")
+        }
+
+        @Test
+        fun customEndpointReturnsCorrectUri() {
+            val endpoint = "https://some-host:1234"
+            val s3UriConverter = S3UriConverter(s3Properties.copy(endpoint = endpoint), region)
+
+            val httpUri = s3UriConverter.toHttp(s3Uri)
+            assertThat(httpUri).isEqualTo("https://my-bucket.some-host:1234/test2/test1_x264_3100.mp4")
+        }
+
+        @Test
+        fun customEndpointWithoutPortReturnsCorrectUri() {
+            val endpoint = "https://some-host"
+            val s3UriConverter = S3UriConverter(s3Properties.copy(endpoint = endpoint), region)
+
+            val httpUri = s3UriConverter.toHttp(s3Uri)
+            assertThat(httpUri).isEqualTo("https://my-bucket.some-host/test2/test1_x264_3100.mp4")
+        }
+
+        @Test
+        fun customEndpointReturnsCorrectPathStyleUri() {
+            val endpoint = "http://some-host:1234"
+            val s3UriConverter = S3UriConverter(s3Properties.copy(endpoint = endpoint, usePathStyle = true), region)
+
+            val httpUri = s3UriConverter.toHttp(s3Uri)
+            assertThat(httpUri).isEqualTo("http://some-host:1234/my-bucket/test2/test1_x264_3100.mp4")
+        }
+
+        @Test
+        fun nonS3UriThrowsException() {
+            val uri = URI.create("https://my-bucket/test2/test1_x264_3100.mp4")
+            assertThatThrownBy { s3UriConverter.toHttp(uri) }.isInstanceOf(IllegalArgumentException::class.java)
+                .hasMessage("Invalid URI: $uri")
+        }
+    }
+
+    @Nested
+    inner class GetBucketAndKey {
+        @Test
+        fun returnsCorrectValues() {
+            val (bucket, key) = s3UriConverter.getBucketAndKey(s3Uri)
+            assertThat(bucket).isEqualTo("my-bucket")
+            assertThat(key).isEqualTo("test2/test1_x264_3100.mp4")
+        }
+
+        @Test
+        fun nonS3UriThrowsException() {
+            val uri = URI.create("https://my-bucket/test2/test1_x264_3100.mp4")
+            assertThatThrownBy { s3UriConverter.getBucketAndKey(uri) }.isInstanceOf(IllegalArgumentException::class.java)
+                .hasMessage("Invalid URI: $uri")
+        }
+    }
+}
diff --git a/encore-common/src/testFixtures/kotlin/se/svt/oss/encore/S3StorageExtension.kt b/encore-common/src/testFixtures/kotlin/se/svt/oss/encore/S3StorageExtension.kt
@@ -24,7 +24,6 @@ class S3StorageExtension : BeforeAllCallback {
         localstack.start()
 
         log.info { "localstack endpoint: ${localstack.endpoint}" }
-
         System.setProperty("aws.accessKeyId", localstack.accessKey)
         System.setProperty("aws.secretAccessKey", localstack.secretKey)
         System.setProperty("remote-files.s3.endpoint", localstack.endpoint.toString())
