Merge pull request #528 from RADAR-base/perTopicFormat

Add per-topic formatting with record data lookup

Author: blootsvoets

.github/workflows/main.yml

@@ -24,7 +24,7 @@ jobs:
 
       - uses: actions/setup-java@v3
         with:
-          distribution: zulu
+          distribution: temurin
           java-version: 17
 
       - name: Setup Gradle

.github/workflows/publish_snapshots.yml

@@ -23,7 +23,7 @@ jobs:
 
       - uses: actions/setup-java@v3
         with:
-          distribution: zulu
+          distribution: temurin
           java-version: 17
 
       - name: Setup Gradle

.github/workflows/release.yml

@@ -20,7 +20,7 @@ jobs:
 
       - uses: actions/setup-java@v3
         with:
-          distribution: zulu
+          distribution: temurin
           java-version: 17
 
       - name: Setup Gradle

Dockerfile

@@ -10,7 +10,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=$BUILDPLATFORM gradle:7.4-jdk17 AS builder
+FROM --platform=$BUILDPLATFORM gradle:7.5-jdk17 AS builder
 
 RUN mkdir /code
 WORKDIR /code

README.md

@@ -138,19 +138,65 @@ target:
 
 Secrets can be provided as environment variables as well:
 
-| Environment variable | Corresponding value |
-| --- | --- |
-| `SOURCE_S3_ACCESS_TOKEN` | `source.s3.accessToken` |
-| `SOURCE_S3_SECRET_KEY` | `source.s3.secretKey` |
-| `SOURCE_AZURE_USERNAME` | `source.azure.username` |
-| `SOURCE_AZURE_PASSWORD` | `source.azure.password` |
+| Environment variable        | Corresponding value        |
+|-----------------------------|----------------------------|
+| `SOURCE_S3_ACCESS_TOKEN`    | `source.s3.accessToken`    |
+| `SOURCE_S3_SECRET_KEY`      | `source.s3.secretKey`      |
+| `SOURCE_AZURE_USERNAME`     | `source.azure.username`    |
+| `SOURCE_AZURE_PASSWORD`     | `source.azure.password`    |
 | `SOURCE_AZURE_ACCOUNT_NAME` | `source.azure.accountName` |
-| `SOURCE_AZURE_ACCOUNT_KEY` | `source.azure.accountKey` |
-| `SOURCE_AZURE_SAS_TOKEN` | `source.azure.sasToken` |
-| `REDIS_URL` | `redis.url` |
+| `SOURCE_AZURE_ACCOUNT_KEY`  | `source.azure.accountKey`  |
+| `SOURCE_AZURE_SAS_TOKEN`    | `source.azure.sasToken`    |
+| `REDIS_URL`                 | `redis.url`                |
 
 Replace `SOURCE` with `TARGET` in the variables above to configure the target storage.
 
+### Path format
+
+The output path at the target storage is determined by the path format. The class that handles path
+output by default is the `org.radarbase.output.path.FormattedPathFactory`. The default format is
+```
+${projectId}/${userId}/${topic}/${filename}
+```
+Each format parameter is enclosed by a dollar sign with curly brackets.
+
+The full set of parameters is listed here:
+```yaml
+paths:
+  # Input directories in source storage
+  inputs:
+    - /testIn
+  # Temporary directory for local file processing.
+  temp: ./output/+tmp
+  # Output directory in target storage
+  output: /output
+  # Output path construction factory
+  factory: org.radarbase.output.path.FormattedPathFactory
+  # Additional properties
+  # properties:
+  #   format: ${projectId}/${userId}/${topic}/${time:mm}/${time:YYYYmmDD_HH'00'}${attempt}${extension}
+  #   plugins: fixed time key value org.example.plugin.MyPathPlugin
+```
+
+The FormattedPathFactory can use multiple plugins to format paths based on a given record.
+The `fixed` plugin has a number of fixed parameters that can be used:
+
+| Parameter | Description                                                             |
+|-----------|-------------------------------------------------------------------------|
+| projectId | record project ID                                                       |
+| userId    | record user ID                                                          |
+| sourceId  | record source ID                                                        |
+| topic     | Kafka topic                                                             | 
+| filename  | default time binning with attempt suffix and file extension             |
+| attempt   | attempt suffix for if a file with an incompatible format already exists |
+| extension | file extension                                                          |
+
+At least `filename` should be used, or a combination of `attempt` and `extension`.
+
+Then there are also plugins that take their own format. The `time` plugin formats a parameter according to the record time. It takes parameters with format `time:<date format>` where `<date format>` should be replaced by a [Java date format](https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/time/format/DateTimeFormatter.html), such as `YYYY-mm-dd`. The plugin tries to use the following time fields, in this order: a double `time` in the value struct, `timeStart` double or `start` long in the key struct, `dateTime` string in the value struct, `date` string in the value struct, `timeReceived` double in the value struct or `timeCompleted` double in the value struct. The first valid value used. If no valid time values are found, `unknown-date` is returned.
+
+The `key` and `value` plugins read values from the key or value structs of a given record. For example, parameter `value:color.red` will attempt to read the value struct, finding first the `color` field and then the enclosed `red` field. If no such value exists, `unknown-value` will be used in the format.
+
 ### Cleaner
 
 Source files can be automatically be removed by a cleaner process. This checks whether the file has already been extracted and is older than a configured age. This feature is not enabled by default. It can be configured in the `cleaner` configuration section:
@@ -192,10 +238,12 @@ Now the `radar-output-restructure` command should be available.
 To implement alternative storage paths, storage drivers or storage formats, put your custom JAR in
 `$APP_DIR/lib/radar-output-plugins`. To load them, use the following options:
 
-| Parameter                   | Base class                                          | Behaviour                                  | Default                   |
-| --------------------------- | --------------------------------------------------- | ------------------------------------------ | ------------------------- |
-| `paths: factory: ...`       | `org.radarbase.output.path.RecordPathFactory`         | Factory to create output path names with.  | ObservationKeyPathFactory |
-| `format: factory: ...`      | `org.radarbase.output.format.FormatFactory`           | Factory for output formats.                | FormatFactory             |
-| `compression: factory: ...` | `org.radarbase.output.compression.CompressionFactory` | Factory class to use for data compression. | CompressionFactory        |
+| Parameter                   | Base class                                            | Behaviour                                  | Default              |
+|-----------------------------|-------------------------------------------------------|--------------------------------------------|----------------------|
+| `paths: factory: ...`       | `org.radarbase.output.path.RecordPathFactory`         | Factory to create output path names with.  | FormattedPathFactory |
+| `format: factory: ...`      | `org.radarbase.output.format.FormatFactory`           | Factory for output formats.                | FormatFactory        |
+| `compression: factory: ...` | `org.radarbase.output.compression.CompressionFactory` | Factory class to use for data compression. | CompressionFactory   |
 
 The respective `<type>: properties: {}` configuration parameters can be used to provide custom configuration of the factory. This configuration will be passed to the `Plugin#init(Map<String, String>)` method.
+
+By adding additional path format plugins to the classpath, the path format of FormattedPathFactory may be expanded with different parameters or lookup engines.

build.gradle.kts

@@ -141,6 +141,7 @@ tasks.withType<KotlinCompile> {
         jvmTarget = "17"
         apiVersion = "1.6"
         languageVersion = "1.6"
+        freeCompilerArgs = listOf("-opt-in=kotlin.RequiresOptIn")
     }
 }
 

restructure.yml

@@ -136,6 +136,7 @@ paths:
   # Additional properties
   # properties:
   #   format: ${projectId}/${userId}/${topic}/${time:mm}/${time:YYYYmmDD_HH'00'}${attempt}${extension}
+  #   plugins: fixed time key value org.example.plugin.MyPathPlugin
 
 # Individual topic configuration
 topics:
@@ -160,3 +161,8 @@ topics:
     # Disable deduplication
     deduplication:
       enable: false
+  questionnaire_response:
+    # Specify an alternative path format.
+    pathProperties:
+      format: ${projectId}/${userId}/${topic}/${value:name}/${filename}
+      plugins: fixed value

src/integrationTest/java/org/radarbase/output/RestructureS3IntegrationTest.kt

@@ -14,6 +14,7 @@ import org.radarbase.output.util.objectBuild
 import java.nio.charset.StandardCharsets.UTF_8
 import java.nio.file.Paths
 
+@OptIn(ExperimentalCoroutinesApi::class)
 class RestructureS3IntegrationTest {
     @Test
     fun integration() = runTest {
@@ -24,17 +25,20 @@ class RestructureS3IntegrationTest {
             secretKey = "minioadmin",
             bucket = "source",
         )
-        val targetConfig = S3Config(
-            endpoint = "http://localhost:9000",
-            accessToken = "minioadmin",
-            secretKey = "minioadmin",
-            bucket = "target",
+        val targetConfig = sourceConfig.copy(bucket = "target")
+        val topicConfig = mapOf(
+            "application_server_status" to TopicConfig(
+                pathProperties = mapOf(
+                    "format" to "\${projectId}/\${userId}/\${topic}/\${value:serverStatus}/\${filename}"
+                )
+            )
         )
         val config = RestructureConfig(
             source = ResourceConfig("s3", s3 = sourceConfig),
             target = ResourceConfig("s3", s3 = targetConfig),
             paths = PathConfig(inputs = listOf(Paths.get("in"))),
-            worker = WorkerConfig(minimumFileAge = 0L)
+            worker = WorkerConfig(minimumFileAge = 0L),
+            topics = topicConfig,
         )
         val application = Application(config)
         val sourceClient = sourceConfig.createS3Client()
@@ -72,7 +76,7 @@ class RestructureS3IntegrationTest {
         }
 
         val firstParticipantOutput =
-            "output/STAGING_PROJECT/1543bc93-3c17-4381-89a5-c5d6272b827c/application_server_status"
+            "output/STAGING_PROJECT/1543bc93-3c17-4381-89a5-c5d6272b827c/application_server_status/CONNECTED"
         val secondParticipantOutput =
             "output/radar-test-root/4ab9b985-6eec-4e51-9a29-f4c571c89f99/android_phone_acceleration"
 
@@ -96,16 +100,15 @@ class RestructureS3IntegrationTest {
                 assertEquals(csvContents, targetContent.toString(UTF_8))
             }
 
-            withContext(Dispatchers.IO) {
+            return@coroutineScope withContext(Dispatchers.IO) {
                 targetClient.listObjects(
                     ListObjectsArgs.Builder().bucketBuild(targetConfig.bucket) {
                         prefix("output")
                         recursive(true)
                         useUrlEncodingType(false)
                     }
                 )
-                    .map { it.get().objectName() }
-                    .toHashSet()
+                    .mapTo(HashSet()) { it.get().objectName() }
             }
         }
 

src/integrationTest/java/org/radarbase/output/accounting/OffsetRangeRedisTest.kt

@@ -1,5 +1,6 @@
 package org.radarbase.output.accounting
 
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runTest
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.Assertions.*
@@ -13,6 +14,7 @@ import java.nio.file.Path
 import java.nio.file.Paths
 import java.time.Instant
 
+@OptIn(ExperimentalCoroutinesApi::class)
 class OffsetRangeRedisTest {
     private lateinit var testFile: Path
     private lateinit var redisHolder: RedisHolder

src/integrationTest/java/org/radarbase/output/accounting/RedisRemoteLockManagerTest.kt

@@ -1,5 +1,6 @@
 package org.radarbase.output.accounting
 
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runTest
 import org.hamcrest.MatcherAssert.assertThat
 import org.hamcrest.Matchers.not
@@ -10,6 +11,7 @@ import org.junit.jupiter.api.Test
 import org.radarbase.output.util.SuspendedCloseable.Companion.useSuspended
 import redis.clients.jedis.JedisPool
 
+@OptIn(ExperimentalCoroutinesApi::class)
 internal class RedisRemoteLockManagerTest {
     private lateinit var redisHolder: RedisHolder
     private lateinit var lockManager1: RemoteLockManager