Tested restructure on Azure

Author: blootsvoets

restructure.yml

@@ -7,25 +7,49 @@ service:
 # Source data resource
 # @since: 0.7.0
 source:
-  type: s3  # hdfs or s3
+  type: s3  # hdfs, azure or s3
   s3:
     endpoint: http://localhost:9000  # using AWS S3 endpoint is also possible.
     bucket: radar
     accessToken: minioadmin
     secretKey: minioadmin
+  azure:
+    endpoint: https://MyBlobStorageAccount.blob.core.windows.net
+    # when using personal login
+    #username: User
+    #password: Password
+    # when using shared access tokens
+    #accountName: MyBlobStorageAccount
+    #accountKey: MyLongToken
+    # when using a specially made SAS token
+    #sasToken: MyLongToken
+    # if no credentials are supplied, this only works with a publicly writable blob storage
+    container: MySourceContainer
   # only actually needed if source type is hdfs
   hdfs:
     nameNodes: [hdfs-namenode-1, hdfs-namenode-2]
 
 # Target data resource
 # @since: 0.7.0
 target:
-  type: s3  # s3 or local
+  type: s3  # s3, azure or local
   s3:
     endpoint: http://localhost:9000
     bucket: out
     accessToken: minioadmin
     secretKey: minioadmin
+  azure:
+    endpoint: https://MyBlobStorageAccount.blob.core.windows.net
+    # when using personal login
+    #username: User
+    #password: Password
+    # when using shared access tokens
+    #accountName: MyBlobStorageAccount
+    #accountKey: MyLongToken
+    # when using a specially made SAS token
+    #sasToken: MyLongToken
+    # if no credentials are supplied, this only works with a publicly writable blob storage
+    container: MyTargetContainer
   # only actually needed if target type is local
   local:
     userId: 1000  # write as regular user, use -1 to use current user (default).
@@ -102,7 +126,7 @@ paths:
   # Output directory in target
   output: /output
   # Output path construction factory
-  factory: org.radarbase.output.path.MonthlyObservationKeyPathFactory
+  #factory: org.radarbase.output.path.MonthlyObservationKeyPathFactory
   # Additional properties
   # properties: {}
 

src/main/java/org/radarbase/output/accounting/OffsetRangeSet.kt

@@ -16,6 +16,7 @@
 
 package org.radarbase.output.accounting
 
+import com.fasterxml.jackson.annotation.JsonIgnore
 import org.radarbase.output.util.FunctionalValue
 import org.radarbase.output.util.LockedFunctionalValue
 import org.radarbase.output.util.ReadOnlyFunctionalValue
@@ -164,6 +165,7 @@ class OffsetRangeSet {
 
 
     data class Range(val from: Long, val to: Long?, val lastProcessed: Instant = Instant.now()) {
+        @JsonIgnore
         val size: Long? = to?.let { it - from + 1 }
         fun ensureToOffset(): Range = if (to == null) copy(to = from) else this
         override fun toString() = "($from - $to, $lastProcessed)"

src/main/java/org/radarbase/output/accounting/OffsetRedisPersistence.kt

@@ -18,6 +18,7 @@ package org.radarbase.output.accounting
 
 import com.fasterxml.jackson.databind.ObjectReader
 import com.fasterxml.jackson.databind.ObjectWriter
+import com.fasterxml.jackson.databind.SerializationFeature
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import org.radarbase.output.util.PostponedWriter
@@ -95,8 +96,10 @@ class OffsetRedisPersistence(
                 val ranges: List<OffsetRangeSet.Range>)
 
         private val logger = LoggerFactory.getLogger(OffsetRedisPersistence::class.java)
-        private val mapper = jacksonObjectMapper()
-                .registerModule(JavaTimeModule())
+        private val mapper = jacksonObjectMapper().apply {
+            registerModule(JavaTimeModule())
+            configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false)
+        }
         val redisOffsetWriter: ObjectWriter = mapper.writerFor(RedisOffsetRangeSet::class.java)
         val redisOffsetReader: ObjectReader = mapper.readerFor(RedisOffsetRangeSet::class.java)
     }

src/main/java/org/radarbase/output/config/RestructureConfig.kt

@@ -3,6 +3,7 @@ package org.radarbase.output.config
 import com.azure.core.credential.BasicAuthenticationCredential
 import com.azure.storage.blob.BlobServiceClient
 import com.azure.storage.blob.BlobServiceClientBuilder
+import com.azure.storage.common.StorageSharedKeyCredential
 import com.fasterxml.jackson.annotation.JsonIgnore
 import io.minio.MinioClient
 import org.apache.hadoop.conf.Configuration
@@ -329,11 +330,23 @@ data class S3Config(
 data class AzureConfig(
         val endpoint: String,
         val container: String,
-        val username: String,
-        val password: String
+        val username: String?,
+        val password: String?,
+        val accountName: String?,
+        val accountKey: String?,
+        val sasToken: String?
 ) {
-    fun createAzureClient(): BlobServiceClient = BlobServiceClientBuilder()
-            .endpoint(endpoint)
-            .credential(BasicAuthenticationCredential(username, password))
-            .buildClient()
+    fun createAzureClient(): BlobServiceClient = BlobServiceClientBuilder().apply {
+        endpoint(endpoint)
+        when {
+            username != null && password != null -> credential(BasicAuthenticationCredential(username, password))
+            accountName != null && accountKey != null -> credential(StorageSharedKeyCredential(accountName, accountKey))
+            sasToken != null -> sasToken(sasToken)
+            else -> logger.warn("No Azure credentials supplied. Assuming a public blob storage.")
+        }
+    }.buildClient()
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(AzureConfig::class.java)
+    }
 }

src/main/java/org/radarbase/output/source/AzureSourceStorage.kt

@@ -19,7 +19,7 @@ class AzureSourceStorage(
 
     override fun list(path: Path): Sequence<SimpleFileStatus> = blobContainerClient.listBlobsByHierarchy("$path/")
             .asSequence()
-            .map { SimpleFileStatus(Paths.get(it.name), it.isPrefix, it.properties.lastModified.toInstant()) }
+            .map { SimpleFileStatus(Paths.get(it.name), it.isPrefix ?: false, it.properties?.lastModified?.toInstant()) }
 
     override fun delete(path: Path) {
         blobContainerClient.getBlobClient(path.toKey())
@@ -38,7 +38,7 @@ class AzureSourceStorage(
 
             blobContainerClient
                     .getBlobClient(file.path.toKey())
-                    .downloadToFile(fileName.toString())
+                    .downloadToFile(fileName.toString(), true)
 
             return object : SeekableFileInput(fileName.toFile()) {
                 override fun close() {