Added S3 storage driver

Author: blootsvoets

build.gradle

@@ -40,6 +40,8 @@ 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 group: 'org.apache.avro', name: 'avro-mapred', version: avroVersion
     implementation group: 'org.apache.hadoop', name: 'hadoop-common', version: hadoopVersion
 

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/data/FileCache.kt

@@ -16,25 +16,15 @@
 
 package org.radarbase.hdfs.data
 
-import java.io.BufferedOutputStream
-import java.io.ByteArrayInputStream
-import java.io.Closeable
-import java.io.Flushable
-import java.io.IOException
-import java.io.InputStream
-import java.io.InputStreamReader
-import java.io.OutputStream
-import java.io.OutputStreamWriter
-import java.io.Writer
-import java.nio.file.Files
-import java.nio.file.Path
-import java.util.concurrent.atomic.AtomicBoolean
 import org.apache.avro.generic.GenericRecord
 import org.radarbase.hdfs.FileStoreFactory
 import org.radarbase.hdfs.accounting.Accountant
-import org.radarbase.hdfs.util.Timer
 import org.radarbase.hdfs.util.Timer.time
 import org.slf4j.LoggerFactory
+import java.io.*
+import java.nio.file.Files
+import java.nio.file.Path
+import java.util.concurrent.atomic.AtomicBoolean
 
 /** Keeps path handles of a path.  */
 class FileCache(
@@ -60,7 +50,7 @@ class FileCache(
     private val hasError: AtomicBoolean = AtomicBoolean(false)
 
     init {
-        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,
@@ -165,7 +155,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/data/FileCacheStore.kt

@@ -111,10 +111,13 @@ 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)
+                Files.delete(tmpSchemaPath)
             }
         }
     }

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

@@ -18,14 +18,11 @@ package org.radarbase.hdfs.data
 
 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 {
@@ -37,20 +34,18 @@ class LocalStorageDriver : StorageDriver {
         gid = properties["local-gid"]?.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/data/S3StorageDriver.kt

@@ -0,0 +1,84 @@
+/*
+ * 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.data
+
+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.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)
+    }
+
+    @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/data/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)
 }