feat: support s3 urls for input and output

This commit add support for using s3 urls on the format
s3://<BUCKET>/<KEY> in both input and output.

If ans s3 URL is used as input, a presigned URL is created
and used as input to ffmpeg. The duration of the presigned URLs can be
controlled with the 'remote-files.s3.presignDurationSeconds' config
property.

If an s3 URL is used for 'outputFolder', output will first be stored
locally and then uploaded to s3 once transcoding is finished.

Aws credentials are read with DefaultCredentialsProvider, meaning
aws credentials can be provided in a number of ways, see
https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/DefaultCredentialsProvider.html;

Not that when using s3 urls for input, the presigned URLs will be
shown in the logs. If this is not desirable, setting
logging.config (or env variable LOGGING_CONFIG) to
'classpath:logback-json-mask-s3-presign.xml'
will use a log config that masks the presign query parameters.

By setting env variable REMOTEFILES_S3_ANONYMOUSACCESS to true, s3 urls will be
accessed in anonymous mode, corresponding to using the '--no-sign-request' flag
with the aws cli. Any s3 access key or secrets key configured will be
ignored. Multipart upload will be disabled in this case since the s3
sdk does not support multipart upload when using anonymous access.

S3 multipart upload can be configured through configuration properties:
- remote-files.s3.multipart.minimumPartSize
- remote-files.s3.multipartthreshold
- remote-files.s3.multipart.apicallbuffersize

These properties corresponds to the properties in
software.amazon.awssdk.services.s3.multipart.MultipartConfiguration
See https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/s3/multipart/MultipartConfiguration.html

Signed-off-by: Gustav Grusell <[email protected]>

Author: grusell

checks.gradle

@@ -11,6 +11,8 @@ jacocoTestCoverageVerification {
                     '*.static {...}',
                     '*.model.*.get*',
                     '*.service.localencode.LocalEncodeService.moveFile*',
+                    '*.S3Properties*.get*()',
+                    '*RemoteFileService.DefaultHandler.*',
                     '*QueueService.getQueue*',
                     '*QueueService.migrateQueues()',
                     '*.ShutdownHandler.*',

encore-common/build.gradle.kts

@@ -16,6 +16,9 @@ dependencies {
 
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core")
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-slf4j")
+    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")
@@ -26,6 +29,7 @@ dependencies {
     testFixturesImplementation("com.redis:testcontainers-redis:2.2.4")
     testFixturesImplementation("io.github.oshai:kotlin-logging-jvm:7.0.13")
     testFixturesImplementation("org.junit.jupiter:junit-jupiter-api")
+    testFixturesImplementation("org.testcontainers:localstack:1.20.3")
     testFixturesRuntimeOnly("org.junit.platform:junit-platform-launcher")
 }
 

encore-common/src/main/kotlin/se/svt/oss/encore/S3RemoteFilesConfiguration.kt

@@ -0,0 +1,95 @@
+// SPDX-FileCopyrightText: 2024 Eyevinn Technology AB
+//
+// SPDX-License-Identifier: EUPL-1.2
+
+package se.svt.oss.encore
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.boot.context.properties.EnableConfigurationProperties
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import se.svt.oss.encore.service.remotefiles.s3.S3Properties
+import se.svt.oss.encore.service.remotefiles.s3.S3RemoteFileHandler
+import se.svt.oss.encore.service.remotefiles.s3.S3UriConverter
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider
+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")
+@EnableConfigurationProperties(S3Properties::class)
+@Configuration
+class S3RemoteFilesConfiguration {
+
+    @Bean
+    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)
+        .crossRegionAccessEnabled(true)
+        .multipartConfiguration { multipartConfigurationBuilder ->
+            multipartConfigurationBuilder
+                .minimumPartSizeInBytes(s3Properties.multipart.minimumPartSize)
+                .thresholdInBytes(s3Properties.multipart.threshold)
+                .apiCallBufferSizeInBytes(s3Properties.multipart.apiCallBufferSize)
+        }
+        .multipartEnabled(!s3Properties.anonymousAccess) // Multipart upload requires credentials
+        .serviceConfiguration(
+            S3Configuration.builder()
+                .pathStyleAccessEnabled(s3Properties.usePathStyle)
+                .build(),
+        )
+        .credentialsProvider(
+            if (s3Properties.anonymousAccess) {
+                AnonymousCredentialsProvider.create()
+            } else {
+                DefaultCredentialsProvider.builder().build()
+            },
+        )
+        .apply {
+            if (s3Properties.endpoint.isNotBlank()) {
+                endpointOverride(URI.create(s3Properties.endpoint))
+            }
+        }
+        .build()
+
+    @Bean
+    fun s3Presigner(s3Region: Region, s3Properties: S3Properties) = S3Presigner.builder()
+        .region(s3Region)
+        .serviceConfiguration(
+            S3Configuration.builder()
+                .pathStyleAccessEnabled(true)
+                .build(),
+        )
+        .apply {
+            if (!s3Properties.endpoint.isNullOrBlank()) {
+                endpointOverride(URI.create(s3Properties.endpoint))
+            }
+        }
+        .build()
+
+    @Bean
+    fun s3UriConverter(s3Properties: S3Properties, s3Region: Region) = S3UriConverter(s3Properties, s3Region)
+
+    @Bean
+    fun s3RemoteFileHandler(
+        s3Client: S3AsyncClient,
+        s3Presigner: S3Presigner,
+        s3Properties: S3Properties,
+        s3UriConverter: S3UriConverter,
+        transferManager: S3TransferManager,
+    ) =
+        S3RemoteFileHandler(s3Client, s3Presigner, s3Properties, s3UriConverter, transferManager)
+}

encore-common/src/main/kotlin/se/svt/oss/encore/model/input/Input.kt

@@ -35,6 +35,8 @@ sealed interface Input {
     @get:Schema(description = "URI of input file", required = true, example = "/path/to/file.mp4")
     val uri: String
 
+    var accessUri: String
+
     @get:Schema(description = "Input params required to properly decode input", example = """{ "ac": "2" }""")
     val params: LinkedHashMap<String, String?>
 
@@ -167,6 +169,9 @@ data class AudioInput(
     override val type: String
         get() = TYPE_AUDIO
 
+    @JsonIgnore
+    override var accessUri: String = uri
+
     override fun withSeekTo(seekTo: Double) = copy(seekTo = seekTo)
 
     val duration: Double
@@ -188,6 +193,9 @@ data class VideoInput(
     override val seekTo: Double? = null,
     override val copyTs: Boolean = false,
 ) : VideoIn {
+    @JsonIgnore
+    override var accessUri: String = uri
+
     override val analyzedVideo: VideoFile
         @JsonIgnore
         get() = analyzed as? VideoFile ?: throw RuntimeException("Analyzed video for $uri is ${analyzed?.type}")
@@ -221,6 +229,9 @@ data class AudioVideoInput(
     override val copyTs: Boolean = false,
 ) : VideoIn,
     AudioIn {
+    @JsonIgnore
+    override var accessUri: String = uri
+
     override val analyzedVideo: VideoFile
         @JsonIgnore
         get() = analyzed as? VideoFile ?: throw RuntimeException("Analyzed audio/video for $uri is ${analyzed?.type}")
@@ -245,7 +256,7 @@ fun List<Input>.inputParams(readDuration: Double?): List<String> =
             (readDuration?.let { listOf("-t", "$it") } ?: emptyList()) +
             (input.seekTo?.let { listOf("-ss", "$it") } ?: emptyList()) +
             (if (input.copyTs) listOf("-copyts") else emptyList()) +
-            listOf("-i", input.uri)
+            listOf("-i", input.accessUri)
     }
 
 fun List<Input>.maxDuration(): Double? = maxOfOrNull {

encore-common/src/main/kotlin/se/svt/oss/encore/service/EncoreService.kt

@@ -45,6 +45,7 @@ import se.svt.oss.encore.service.callback.CallbackService
 import se.svt.oss.encore.service.localencode.LocalEncodeService
 import se.svt.oss.encore.service.mediaanalyzer.MediaAnalyzerService
 import se.svt.oss.encore.service.queue.QueueService
+import se.svt.oss.encore.service.remotefiles.RemoteFileService
 import se.svt.oss.mediaanalyzer.file.MediaContainer
 import se.svt.oss.mediaanalyzer.file.MediaFile
 import java.io.File
@@ -67,6 +68,7 @@ class EncoreService(
     private val localEncodeService: LocalEncodeService,
     private val encoreProperties: EncoreProperties,
     private val queueService: QueueService,
+    private val remoteFileService: RemoteFileService,
 ) {
 
     private val cancelTopicName = "cancel"
@@ -227,7 +229,7 @@ class EncoreService(
             callbackService.sendProgressCallback(encoreJob)
         } finally {
             redisMessageListerenerContainer.removeMessageListener(cancelListener)
-            localEncodeService.cleanup(outputFolder)
+            localEncodeService.cleanup(outputFolder, encoreJob)
         }
     }
 
@@ -270,6 +272,10 @@ class EncoreService(
     }
 
     private fun initJob(encoreJob: EncoreJob) {
+        encoreJob.inputs.forEach { input ->
+            input.accessUri = remoteFileService.getAccessUri(input.uri)
+        }
+
         encoreJob.inputs.forEach { input ->
             mediaAnalyzerService.analyzeInput(input)
         }

encore-common/src/main/kotlin/se/svt/oss/encore/service/localencode/LocalEncodeService.kt

@@ -9,11 +9,13 @@ import org.springframework.stereotype.Service
 import se.svt.oss.encore.config.EncoreProperties
 import se.svt.oss.encore.model.EncoreJob
 import se.svt.oss.encore.process.createTempDir
+import se.svt.oss.encore.service.remotefiles.RemoteFileService
 import se.svt.oss.mediaanalyzer.file.AudioFile
 import se.svt.oss.mediaanalyzer.file.ImageFile
 import se.svt.oss.mediaanalyzer.file.MediaFile
 import se.svt.oss.mediaanalyzer.file.VideoFile
 import java.io.File
+import java.net.URI
 import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.StandardCopyOption
@@ -23,11 +25,12 @@ private val log = KotlinLogging.logger {}
 @Service
 class LocalEncodeService(
     private val encoreProperties: EncoreProperties,
+    private val remoteFileService: RemoteFileService,
 ) {
 
     fun outputFolder(
         encoreJob: EncoreJob,
-    ): String = if (encoreProperties.localTemporaryEncode) {
+    ): String = if (encoreProperties.localTemporaryEncode || remoteFileService.isRemoteFile(encoreJob.outputFolder)) {
         createTempDir("job_${encoreJob.id}").toString()
     } else {
         encoreJob.outputFolder
@@ -38,6 +41,23 @@ class LocalEncodeService(
         output: List<MediaFile>,
         encoreJob: EncoreJob,
     ): List<MediaFile> {
+        if (remoteFileService.isRemoteFile(encoreJob.outputFolder)) {
+            log.debug { "Moving files to output destination ${encoreJob.outputFolder}, from local temp $outputFolder" }
+            File(outputFolder).listFiles()?.forEach { localFile ->
+                val remoteFile = URI.create(encoreJob.outputFolder).resolve(localFile.name).toString()
+                remoteFileService.upload(localFile.toString(), remoteFile)
+            }
+            val files = output.map {
+                val resolvedPath = URI.create(encoreJob.outputFolder).resolve(Path.of(it.file).fileName.toString()).toString()
+                when (it) {
+                    is VideoFile -> it.copy(file = resolvedPath)
+                    is AudioFile -> it.copy(file = resolvedPath)
+                    is ImageFile -> it.copy(file = resolvedPath)
+                    else -> throw Exception("Invalid conversion")
+                }
+            }
+            return files
+        }
         if (encoreProperties.localTemporaryEncode) {
             val destination = File(encoreJob.outputFolder)
             log.debug { "Moving files to correct outputFolder ${encoreJob.outputFolder}, from local temp $outputFolder" }
@@ -50,8 +70,10 @@ class LocalEncodeService(
         return output
     }
 
-    fun cleanup(tempDirectory: String?) {
-        if (tempDirectory != null && encoreProperties.localTemporaryEncode) {
+    fun cleanup(tempDirectory: String?, encoreJob: EncoreJob) {
+        if (tempDirectory != null &&
+            (encoreProperties.localTemporaryEncode || remoteFileService.isRemoteFile(encoreJob.outputFolder))
+        ) {
             File(tempDirectory).deleteRecursively()
         }
     }

encore-common/src/main/kotlin/se/svt/oss/encore/service/mediaanalyzer/MediaAnalyzerService.kt

@@ -22,6 +22,8 @@ import se.svt.oss.mediaanalyzer.ffprobe.SideData
 import se.svt.oss.mediaanalyzer.ffprobe.UnknownSideData
 import se.svt.oss.mediaanalyzer.ffprobe.UnknownStream
 import se.svt.oss.mediaanalyzer.file.AudioFile
+import se.svt.oss.mediaanalyzer.file.ImageFile
+import se.svt.oss.mediaanalyzer.file.SubtitleFile
 import se.svt.oss.mediaanalyzer.file.VideoFile
 import se.svt.oss.mediaanalyzer.mediainfo.AudioTrack
 import se.svt.oss.mediaanalyzer.mediainfo.GeneralTrack
@@ -58,20 +60,25 @@ class MediaAnalyzerService(private val mediaAnalyzer: MediaAnalyzer) {
         val useFirstAudioStreams = (input as? AudioIn)?.channelLayout?.channels?.size
 
         input.analyzed = mediaAnalyzer.analyze(
-            file = input.uri,
+            file = input.accessUri,
             probeInterlaced = probeInterlaced,
             ffprobeInputParams = input.params,
-        ).let {
-            val selectedVideoStream = (input as? VideoIn)?.videoStream
-            val selectedAudioStream = (input as? AudioIn)?.audioStream
-            when (it) {
-                is VideoFile -> it.selectVideoStream(selectedVideoStream)
-                    .selectAudioStream(selectedAudioStream)
-                    .trimAudio(useFirstAudioStreams)
-                is AudioFile -> it.selectAudioStream(selectedAudioStream)
-                    .trimAudio(useFirstAudioStreams)
-                else -> it
+        )
+            .let {
+                val selectedVideoStream = (input as? VideoIn)?.videoStream
+                val selectedAudioStream = (input as? AudioIn)?.audioStream
+                when (it) {
+                    is VideoFile -> it.selectVideoStream(selectedVideoStream)
+                        .selectAudioStream(selectedAudioStream)
+                        .trimAudio(useFirstAudioStreams)
+                        .copy(file = input.uri)
+                    is AudioFile -> it.selectAudioStream(selectedAudioStream)
+                        .trimAudio(useFirstAudioStreams)
+                        .copy(file = input.uri)
+                    is ImageFile -> it.copy(file = input.uri)
+                    is SubtitleFile -> it.copy(file = input.uri)
+                    else -> it
+                }
             }
-        }
     }
 }

encore-common/src/main/kotlin/se/svt/oss/encore/service/remotefiles/RemoteFileHandler.kt

@@ -0,0 +1,11 @@
+// SPDX-FileCopyrightText: 2024 Eyevinn Technology AB
+//
+// SPDX-License-Identifier: EUPL-1.2
+
+package se.svt.oss.encore.service.remotefiles
+
+interface RemoteFileHandler {
+    fun getAccessUri(uri: String): String
+    fun upload(localFile: String, remoteFile: String)
+    val protocols: List<String>
+}

encore-common/src/main/kotlin/se/svt/oss/encore/service/remotefiles/RemoteFileService.kt

@@ -0,0 +1,57 @@
+// SPDX-FileCopyrightText: 2024 Eyevinn Technology AB
+//
+// SPDX-License-Identifier: EUPL-1.2
+
+package se.svt.oss.encore.service.remotefiles
+
+import io.github.oshai.kotlinlogging.KotlinLogging
+import org.springframework.stereotype.Service
+import java.net.URI
+
+private val log = KotlinLogging.logger {}
+
+@Service
+class RemoteFileService(private val remoteFileHandlers: List<RemoteFileHandler>) {
+
+    private val defaultHandler = DefaultHandler()
+
+    fun isRemoteFile(uriOrPath: String): Boolean {
+        val uri = URI.create(uriOrPath)
+        return !(uri.scheme.isNullOrEmpty() || uri.scheme.lowercase() == "file")
+    }
+
+    fun getAccessUri(uriOrPath: String): String {
+        val uri = URI.create(uriOrPath)
+        return getHandler(uri).getAccessUri(uriOrPath)
+    }
+
+    fun upload(localFile: String, remoteFile: String) {
+        val uri = URI.create(remoteFile)
+        getHandler(uri).upload(localFile, remoteFile)
+    }
+
+    private fun getHandler(uri: URI): RemoteFileHandler {
+        log.info { "Getting handler for uri $uri. Available protocols: ${remoteFileHandlers.flatMap {it.protocols} }" }
+        if (uri.scheme.isNullOrEmpty() || uri.scheme.lowercase() == "file") {
+            return defaultHandler
+        }
+        val handler = remoteFileHandlers.firstOrNull { it.protocols.contains(uri.scheme) }
+        if (handler != null) {
+            return handler
+        }
+        log.info { "No remote file handler found for protocol ${uri.scheme}. Using default handler." }
+        return defaultHandler
+    }
+
+    /** Handler user for protocols where no specific handler is defined. Works for local files and
+     * any protocols that ffmpeg supports natively */
+    private class DefaultHandler : RemoteFileHandler {
+        override fun getAccessUri(uri: String): String = uri
+
+        override fun upload(localFile: String, remoteFile: String) {
+            // Do nothing
+        }
+
+        override val protocols: List<String> = emptyList()
+    }
+}