Misc fixes

Author: blootsvoets

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

@@ -39,7 +39,7 @@ constructor(factory: FileStoreFactory, topic: String) : Flushable, Closeable {
         val offsetsDirectory = factory.config.paths.output
                 .resolve(OFFSETS_FILE_NAME)
 
-        Files.createDirectories(offsetsDirectory)
+        factory.storageDriver.createDirectories(offsetsDirectory)
 
         val offsetPath = offsetsDirectory.resolve("$topic.csv")
         this.offsetFile = OffsetRangeFile.read(factory.storageDriver, offsetPath)

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

@@ -16,13 +16,16 @@
 
 package org.radarbase.hdfs.storage
 
+import org.slf4j.LoggerFactory
 import java.io.IOException
 import java.io.InputStream
 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.attribute.FileAttribute
+import java.nio.file.attribute.PosixFileAttributes
 import java.nio.file.attribute.PosixFilePermissions
 
 class LocalStorageDriver : StorageDriver {
@@ -32,6 +35,10 @@ class LocalStorageDriver : StorageDriver {
     override fun init(properties: Map<String, String>) {
         uid = properties["localUid"]?.toIntOrNull() ?: -1
         gid = properties["localGid"]?.toIntOrNull() ?: -1
+
+        logger.info("Local storage configured with user id {}:{} (-1 if not configured)",
+                uid,
+                gid)
     }
 
     @Throws(IOException::class)
@@ -57,16 +64,31 @@ class LocalStorageDriver : StorageDriver {
 
     @Throws(IOException::class)
     override fun store(localPath: Path, newPath: Path) {
+        localPath.updateUser()
+        Files.setPosixFilePermissions(localPath, PosixFilePermissions.fromString("rw-r--r--"))
+        move(localPath, newPath)
+    }
+
+    override fun createDirectories(directory: Path) {
+        Files.createDirectories(directory, PosixFilePermissions.asFileAttribute(
+                PosixFilePermissions.fromString("rwxr-xr-x")))
+
+        directory.updateUser()
+    }
+
+    private fun Path.updateUser() {
         if (uid >= 0) {
-            Files.setAttribute(localPath, "unix:uid", uid)
+            Files.setAttribute(this, "unix:uid", uid)
         }
         if (gid >= 0) {
-            Files.setAttribute(localPath, "unix:gid", gid)
+            Files.setAttribute(this, "unix:gid", gid)
         }
-        Files.setPosixFilePermissions(localPath, PosixFilePermissions.fromString("rw-r--r--"))
-        move(localPath, newPath)
     }
 
     @Throws(IOException::class)
     override fun delete(path: Path) = Files.delete(path)
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(LocalStorageDriver::class.java)
+    }
 }

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

@@ -24,6 +24,7 @@ import java.io.InputStream
 import java.net.URI
 import java.nio.file.Files
 import java.nio.file.Path
+import java.nio.file.Paths
 
 class S3StorageDriver : StorageDriver {
     private lateinit var bucket: String
@@ -44,11 +45,15 @@ class S3StorageDriver : StorageDriver {
         }.build()
 
         bucket = requireNotNull(properties["s3Bucket"]) { "No AWS bucket provided" }
+
+        logger.info("Object storage configured with endpoint {} in bucket {}",
+                properties["s3EndpointUrl"],
+                properties["s3Bucket"])
     }
 
     override fun status(path: Path): StorageDriver.PathStatus? {
         return try {
-            awsClient.headObject { it.bucket(bucket).key(path.toString()) }
+            awsClient.headObject { it.bucket(bucket).key(path.toKey()) }
                     .let { StorageDriver.PathStatus(it.contentLength()) }
         } catch (ex: NoSuchKeyException) {
             null
@@ -57,30 +62,44 @@ class S3StorageDriver : StorageDriver {
 
     @Throws(IOException::class)
     override fun newInputStream(path: Path): InputStream = awsClient.getObject {
-        it.bucket(bucket).key(path.toString())
+        it.bucket(bucket).key(path.toKey())
     }
 
     @Throws(IOException::class)
     override fun move(oldPath: Path, newPath: Path) {
         awsClient.copyObject {
-            it.bucket(bucket).key(newPath.toString())
+            it.bucket(bucket).key(newPath.toKey())
                     .copySource("$bucket/$oldPath")
         }
         delete(oldPath)
     }
 
     @Throws(IOException::class)
     override fun store(localPath: Path, newPath: Path) {
-        awsClient.putObject({ it.bucket(bucket).key(newPath.toString()) }, localPath)
+        awsClient.putObject({ it.bucket(bucket).key(newPath.toKey()) }, localPath)
         Files.delete(localPath)
     }
 
     @Throws(IOException::class)
     override fun delete(path: Path) {
-        awsClient.deleteObject { it.bucket(bucket).key(path.toString()) }
+        awsClient.deleteObject { it.bucket(bucket).key(path.toKey()) }
+    }
+
+    override fun createDirectories(directory: Path) {
+        // noop
     }
 
     companion object {
         private val logger = LoggerFactory.getLogger(S3StorageDriver::class.java)
+
+        private val rootPath = Paths.get("/")
+
+        private fun Path.toKey(): String {
+            return if (this.startsWith(rootPath)) {
+                rootPath.relativize(this).toString()
+            } else {
+                toString()
+            }
+        }
     }
 }

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

@@ -45,5 +45,8 @@ interface StorageDriver : Plugin {
         return BufferedReader(reader)
     }
 
+    @Throws(IOException::class)
+    fun createDirectories(directory: Path)
+
     data class PathStatus(val size: Long)
 }

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

@@ -69,7 +69,7 @@ constructor(private val factory: FileStoreFactory, private val accountant: Accou
             ensureCapacity()
 
             val dir = path.parent
-            Files.createDirectories(dir)
+            factory.storageDriver.createDirectories(dir)
 
             try {
                 time("write.open") { FileCache(factory, transaction.topicPartition.topic, path, record, tmpDir.path, accountant) }