Merge pull request #55 from RADAR-base/s3Integration

S3 integration

Author: blootsvoets

README.md

@@ -50,7 +50,30 @@ radar-hdfs-restructure --compression gzip  --nameservice <hdfs_node> --output-di
 
 ### Storage
 
-When using local storage, to set the output user ID and group ID, specify the `-p local-uid=123` and `-p local-gid=12` properties.
+There are two storage drivers implemented: `org.radarbase.hdfs.storage.LocalStorageDriver` for an output directory on the local file system or `org.radarbase.hdfs.storage.S3StorageDriver` for storage on an object store.
+
+`LocalStorageDriver` takes the following properties:
+```yaml
+storage:
+  factory: org.radarbase.hdfs.storage.LocalStorageDriver
+  properties:
+    # User ID to write data as
+    localUid: 123
+    # Group ID to write data as
+    localGid: 123
+```
+
+With the `S3StorageDriver`, use the following configuration instead:
+```yaml
+storage:
+  factory: org.radarbase.hdfs.storage.S3StorageDriver
+  properties:
+    # Object store URL
+    s3EndpointUrl: s3://my-region.s3.aws.amazon.com
+    # Bucket to use
+    s3Bucket: myBucketName
+```
+Ensure that the environment variables contain the authorized AWS keys that allow the service to list, download and upload files to the respective bucket.
 
 ### Service
 

build.gradle

@@ -43,6 +43,7 @@ dependencies {
     implementation group: 'com.beust', name: 'jcommander', version: jCommanderVersion
     implementation group: 'com.almworks.integers', name: 'integers', version: almworksVersion
 
+    implementation 'software.amazon.awssdk:s3:2.10.3'
     implementation 'com.opencsv:opencsv:5.0'
 
     implementation group: 'org.apache.avro', name: 'avro-mapred', version: avroVersion
@@ -106,8 +107,8 @@ test {
     useJUnitPlatform()
     testLogging {
         events "passed", "skipped", "failed"
-        setExceptionFormat("full")
         showStandardStreams = true
+        setExceptionFormat("full")
     }
 }
 

src/main/java/org/radarbase/hdfs/Plugin.kt

@@ -19,6 +19,11 @@ package org.radarbase.hdfs
 import java.io.IOException
 
 interface Plugin {
+
+    /**
+     * Initialize plugin. Throws IllegalArgumentException if required properties are not provided
+     * or if they are of the wrong format.
+     */
     @Throws(IOException::class)
     fun init(properties: Map<String, String>) {
         // do nothing

src/main/java/org/radarbase/hdfs/accounting/OffsetRangeFile.kt

@@ -69,7 +69,7 @@ class OffsetRangeFile(private val storage: StorageDriver, private val path: Path
 
         fun read(storage: StorageDriver, path: Path): OffsetRangeFile {
             try {
-                if (storage.exists(path)) {
+                if (storage.status(path) != null) {
                     val set = OffsetRangeSet()
                     storage.newBufferedReader(path).use { br ->
                         // ignore header

src/main/java/org/radarbase/hdfs/storage/LocalStorageDriver.kt

@@ -18,39 +18,34 @@ package org.radarbase.hdfs.storage
 
 import java.io.IOException
 import java.io.InputStream
-import java.io.OutputStream
 import java.nio.file.AtomicMoveNotSupportedException
 import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.StandardCopyOption.ATOMIC_MOVE
 import java.nio.file.StandardCopyOption.REPLACE_EXISTING
-import java.nio.file.StandardOpenOption.APPEND
-import java.nio.file.StandardOpenOption.CREATE
 import java.nio.file.attribute.PosixFilePermissions
 
 class LocalStorageDriver : StorageDriver {
     private var uid = -1
     private var gid = -1
 
     override fun init(properties: Map<String, String>) {
-        uid = properties["local-uid"]?.toIntOrNull() ?: -1
-        gid = properties["local-gid"]?.toIntOrNull() ?: -1
+        uid = properties["localUid"]?.toIntOrNull() ?: -1
+        gid = properties["localGid"]?.toIntOrNull() ?: -1
     }
 
-    override fun exists(path: Path): Boolean = Files.exists(path)
-
     @Throws(IOException::class)
-    override fun newInputStream(path: Path): InputStream = Files.newInputStream(path)
-
-    @Throws(IOException::class)
-    override fun newOutputStream(path: Path, append: Boolean): OutputStream {
-        return if (append) {
-            Files.newOutputStream(path, APPEND, CREATE)
+    override fun status(path: Path): StorageDriver.PathStatus? {
+        return if (Files.exists(path)) {
+            StorageDriver.PathStatus(Files.size(path))
         } else {
-            Files.newOutputStream(path)
+            null
         }
     }
 
+    @Throws(IOException::class)
+    override fun newInputStream(path: Path): InputStream = Files.newInputStream(path)
+
     @Throws(IOException::class)
     override fun move(oldPath: Path, newPath: Path) {
         try {
@@ -72,9 +67,6 @@ class LocalStorageDriver : StorageDriver {
         move(localPath, newPath)
     }
 
-    @Throws(IOException::class)
-    override fun size(path: Path): Long = Files.size(path)
-
     @Throws(IOException::class)
     override fun delete(path: Path) = Files.delete(path)
 }

src/main/java/org/radarbase/hdfs/storage/S3StorageDriver.kt

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2018 The Hyve
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.radarbase.hdfs.storage
+
+import org.slf4j.LoggerFactory
+import software.amazon.awssdk.services.s3.S3Client
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException
+import java.io.IOException
+import java.io.InputStream
+import java.net.URI
+import java.nio.file.Files
+import java.nio.file.Path
+
+class S3StorageDriver : StorageDriver {
+    private lateinit var bucket: String
+    private lateinit var awsClient: S3Client
+
+    override fun init(properties: Map<String, String>) {
+        awsClient = S3Client.builder().also { s3Builder ->
+            properties["s3EndpointUrl"]?.let {
+                val endpoint = try {
+                    URI.create(it)
+                } catch (ex: IllegalArgumentException) {
+                    logger.warn("Invalid S3 URL", ex)
+                    throw ex
+                }
+
+                s3Builder.endpointOverride(endpoint)
+            }
+        }.build()
+
+        bucket = requireNotNull(properties["s3Bucket"]) { "No AWS bucket provided" }
+    }
+
+    override fun status(path: Path): StorageDriver.PathStatus? {
+        return try {
+            awsClient.headObject { it.bucket(bucket).key(path.toString()) }
+                    .let { StorageDriver.PathStatus(it.contentLength()) }
+        } catch (ex: NoSuchKeyException) {
+            null
+        }
+    }
+
+    @Throws(IOException::class)
+    override fun newInputStream(path: Path): InputStream = awsClient.getObject {
+        it.bucket(bucket).key(path.toString())
+    }
+
+    @Throws(IOException::class)
+    override fun move(oldPath: Path, newPath: Path) {
+        awsClient.copyObject {
+            it.bucket(bucket).key(newPath.toString())
+                    .copySource("$bucket/$oldPath")
+        }
+        delete(oldPath)
+    }
+
+    @Throws(IOException::class)
+    override fun store(localPath: Path, newPath: Path) {
+        awsClient.putObject({ it.bucket(bucket).key(newPath.toString()) }, localPath)
+        Files.delete(localPath)
+    }
+
+    @Throws(IOException::class)
+    override fun delete(path: Path) {
+        awsClient.deleteObject { it.bucket(bucket).key(path.toString()) }
+    }
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(S3StorageDriver::class.java)
+    }
+}

src/main/java/org/radarbase/hdfs/storage/StorageDriver.kt

@@ -20,28 +20,22 @@ import java.io.BufferedReader
 import java.io.IOException
 import java.io.InputStream
 import java.io.InputStreamReader
-import java.io.OutputStream
-import java.io.Reader
 import java.nio.file.Path
 import org.radarbase.hdfs.Plugin
 
 interface StorageDriver : Plugin {
-    fun exists(path: Path): Boolean
-    @Throws(IOException::class)
-    fun newInputStream(path: Path): InputStream
+    /** Query the path status. Returns null if the file does not exist. */
+    fun status(path: Path): PathStatus?
 
     @Throws(IOException::class)
-    fun newOutputStream(path: Path, append: Boolean): OutputStream
+    fun newInputStream(path: Path): InputStream
 
     @Throws(IOException::class)
     fun move(oldPath: Path, newPath: Path)
 
     @Throws(IOException::class)
     fun store(localPath: Path, newPath: Path)
 
-    @Throws(IOException::class)
-    fun size(path: Path): Long
-
     @Throws(IOException::class)
     fun delete(path: Path)
 
@@ -50,4 +44,6 @@ interface StorageDriver : Plugin {
         val reader = InputStreamReader(newInputStream(path))
         return BufferedReader(reader)
     }
+
+    data class PathStatus(val size: Long)
 }

src/main/java/org/radarbase/hdfs/util/PostponedWriter.kt

@@ -58,17 +58,18 @@ abstract class PostponedWriter
      * not yet taken place, the write will occur earlier.
      */
     fun triggerWrite() {
-        var localWriteFuture: Future<*>? = writeFuture.get()
-        if (localWriteFuture == null) {
-            localWriteFuture = executor.schedule({ this.startWrite() }, timeout, timeoutUnit)
-            if (!writeFuture.compareAndSet(null, localWriteFuture)) {
-                localWriteFuture!!.cancel(false)
-            }
+        if (writeFuture.get() == null) {
+            executor.schedule(::startWrite, timeout, timeoutUnit)
+                    .also { newWriteFuture ->
+                        if (!writeFuture.compareAndSet(null, newWriteFuture)) {
+                            newWriteFuture.cancel(false)
+                        }
+                    }
         }
     }
 
     /** Start the write in the writer thread.  */
-    protected fun startWrite() {
+    private fun startWrite() {
         writeFuture.set(null)
         doWrite()
     }
@@ -89,8 +90,9 @@ abstract class PostponedWriter
 
     @Throws(IOException::class)
     private fun doFlush(shutdown: Boolean) {
-        val localFuture = executor.submit { this.startWrite() }
-        writeFuture.set(localFuture)
+        val localFuture = executor.submit(::startWrite)
+                .also { writeFuture.set(it) }
+
         if (shutdown) {
             executor.shutdown()
         }

src/main/java/org/radarbase/hdfs/worker/FileCache.kt

@@ -63,7 +63,8 @@ class FileCache(
         val defaultDeduplicate = factory.config.format.deduplication
         deduplicate = topicConfig?.deduplication(defaultDeduplicate) ?: defaultDeduplicate
 
-        val fileIsNew = !storageDriver.exists(path) || storageDriver.size(path) == 0L
+        val fileIsNew = storageDriver.status(path)?.takeIf { it.size > 0L } == null
+
         this.tmpPath = Files.createTempFile(tmpDir, fileName, ".tmp" + compression.extension)
 
         var outStream = compression.compress(fileName,
@@ -127,7 +128,7 @@ class FileCache(
         writer.close()
 
         if (!hasError.get()) {
-            if (deduplicate.enable!!) {
+            if (deduplicate.enable == true) {
                 time("close.deduplicate") {
                     val dedupTmp = tmpPath.resolveSibling("${tmpPath.fileName}.dedup")
                     converterFactory.deduplicate(fileName, tmpPath, dedupTmp, compression, deduplicate.distinctFields ?: emptySet(), deduplicate.ignoreFields ?: emptySet())
@@ -174,7 +175,7 @@ class FileCache(
             var i = 0
             while (corruptPath == null && i < 100) {
                 val path = source.resolveSibling(source.fileName.toString() + ".corrupted" + suffix)
-                if (!storageDriver.exists(path)) {
+                if (storageDriver.status(path) == null) {
                     corruptPath = path
                 }
                 suffix = "-$i"

src/main/java/org/radarbase/hdfs/worker/FileCacheStore.kt

@@ -110,10 +110,12 @@ constructor(private val factory: FileStoreFactory, private val accountant: Accou
         if (schemasAdded.putIfAbsent(schemaPath, schemaPath) == null) {
             val storage = factory.storageDriver
 
-            if (!storage.exists(schemaPath)) {
-                storage.newOutputStream(schemaPath, false).use {
-                    out -> OutputStreamWriter(out).use {
-                    writer -> writer.write(schema.toString(true)) } }
+            if (storage.status(schemaPath) == null) {
+                val tmpSchemaPath = Files.createTempFile(tmpDir.path, "schema-$topic", ".json")
+                Files.newOutputStream(tmpSchemaPath).use { out ->
+                    out.write(schema.toString(true).toByteArray())
+                }
+                storage.store(tmpSchemaPath, schemaPath)
             }
         }
     }