sleepytime for minio

Author: Foxcapades

.github/dependabot.yml

@@ -0,0 +1,20 @@
+version: 2
+
+updates:
+- package-ecosystem: gradle
+  directory: "/"
+  registries:
+  - github-packages
+  schedule:
+    interval: weekly
+  commit-message:
+  open-pull-request-limit: 20
+  ignore:
+    dependency-name: org.gusdb:fgputil*
+
+registries:
+  github-packages:
+    type: maven-repository
+    url: https://maven.pkg.github.com/veupathdb/maven-packages
+    username: ${{ secrets.PACKAGES_USER }}
+    password: ${{ secrets.PACKAGES_KEY }}

build.gradle.kts

@@ -1,5 +1,5 @@
 plugins {
-  kotlin("jvm") version "2.0.20"
+  alias(libs.plugins.kotlin)
 }
 
 allprojects {

gradle/libs.versions.toml

@@ -0,0 +1,28 @@
+[plugins]
+kotlin = { id = "org.jetbrains.kotlin.jvm", version = "2.2.21" }
+dokka  = { id = "org.jetbrains.dokka", version = "2.1.0" }
+
+[libraries]
+logging = { module = "org.slf4j:slf4j-api", version = "2.0.17" }
+
+jackson = { module = "org.veupathdb.lib:jackson-singleton-json", version = "4.0.5" }
+
+db-pool = { module = "com.zaxxer:HikariCP", version = "5.1.0" }
+db-driver = { module = "org.postgresql:postgresql", version = "42.7.3" }
+
+s3-api = { module = "org.veupathdb.lib.s3:s34k", version = "0.11.0" }
+s3-dirs = { module = "org.veupathdb.lib.s3:workspaces-java", version = "5.1.0" }
+
+queue = { module = "org.veupathdb.lib:rabbit-job-queue", version = "2.0.1" }
+
+metrics-prometheus-api = { module = "io.prometheus:simpleclient", version = "0.16.0" }
+metrics-prometheus-client = { module = "io.prometheus:simpleclient_common", version = "0.16.0" }
+
+util-hashid = { module = "org.veupathdb.lib:hash-id", version = "1.1.0" }
+util-cache = { module = "com.github.ben-manes.caffeine:caffeine", version = "3.1.8" }
+
+
+[bundles]
+database = ["db-pool", "db-driver"]
+metrics = ["metrics-prometheus-api", "metrics-prometheus-client"]
+minio = ["s3-api", "s3-dirs"]

lib/build.gradle.kts

@@ -1,40 +1,27 @@
+import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
+
 plugins {
-  kotlin("jvm")
-  id("org.jetbrains.dokka") version "1.9.20"
+  alias(libs.plugins.kotlin)
+  alias(libs.plugins.dokka)
   `maven-publish`
 }
 
 group = "org.veupathdb.lib"
-version = "1.8.5"
+version = "1.8.6"
 
 
 dependencies {
-  // Logging
-  api("org.slf4j:slf4j-api:1.7.36")
-
-  // Jackson
-  implementation("org.veupathdb.lib:jackson-singleton:3.2.1")
-
-  // DB
-  implementation("com.zaxxer:HikariCP:5.1.0")
-  implementation("org.postgresql:postgresql:42.7.3")
+  api(libs.logging)
+  api(libs.util.hashid)
+  api(libs.bundles.minio)
 
-  // S3
-  api("org.veupathdb.lib.s3:s34k:0.11.0")
-  api("org.veupathdb.lib.s3:workspaces-java:5.1.0")
+  implementation(libs.bundles.database)
+  implementation(libs.bundles.metrics)
 
-  // Rabbit
-  implementation("org.veupathdb.lib:rabbit-job-queue:2.0.1")
+  implementation(libs.jackson)
+  implementation(libs.queue)
+  implementation(libs.util.cache)
 
-  // Metrics
-  implementation("io.prometheus:simpleclient:0.16.0")
-  implementation("io.prometheus:simpleclient_common:0.16.0")
-
-  // Misc & Utils
-  api("org.veupathdb.lib:hash-id:1.1.0")
-  implementation("com.github.ben-manes.caffeine:caffeine:3.1.8") // Used for self-expiring cache.
-
-  // Testing
   testImplementation(kotlin("test"))
 }
 
@@ -108,3 +95,8 @@ publishing {
     }
   }
 }
+
+val compileKotlin: KotlinCompile by tasks
+compileKotlin.compilerOptions {
+  freeCompilerArgs.set(listOf("-Xcontext-parameters"))
+}

lib/src/main/kotlin/org/veupathdb/lib/compute/platform/intern/minio/AbstractMinIOHack.kt

@@ -0,0 +1,28 @@
+package org.veupathdb.lib.compute.platform.intern.minio
+
+import org.slf4j.LoggerFactory
+import org.veupathdb.lib.compute.platform.intern.minio.MinIOHax.sleepCountDuration
+
+@Suppress("NOTHING_TO_INLINE")
+internal sealed class AbstractMinIOHack {
+  companion object {
+    private const val MaxDeleteAttempts = 5
+  }
+
+  protected val log = LoggerFactory.getLogger(javaClass)!!
+
+  protected inline fun <T> withRetries(noinline msg: () -> String, noinline fn: () -> T): T =
+    context(log) { MinIOHax.withRetries(msg, fn) }
+
+  protected inline fun <T> withRetries(count: Int, noinline msg: () -> String, noinline fn: () -> T): T =
+    context(log) { MinIOHax.withRetries(count, msg, fn) }
+
+  protected inline fun sleep() =
+    context(log) { MinIOHax.sleep() }
+
+  protected inline fun delete(
+    path: String,
+    noinline deleteFn: () -> Unit,
+    noinline statFn: () -> Boolean,
+  ) = context(log) { MinIOHax.delete(path, deleteFn, statFn) }
+}

lib/src/main/kotlin/org/veupathdb/lib/compute/platform/intern/minio/FuglyMinIOSubWorkspace.kt

@@ -0,0 +1,7 @@
+package org.veupathdb.lib.compute.platform.intern.minio
+
+import org.veupathdb.lib.s3.workspaces.java.SubS3Workspace
+
+internal class FuglyMinIOSubWorkspace(delegate: SubS3Workspace, override val parent: FuglyMinIOWorkspace)
+  : FuglyMinIOWorkspace(delegate)
+  , SubS3Workspace

lib/src/main/kotlin/org/veupathdb/lib/compute/platform/intern/minio/FuglyMinIOWorkspace.kt

@@ -0,0 +1,49 @@
+package org.veupathdb.lib.compute.platform.intern.minio
+
+import org.veupathdb.lib.hash_id.HashID
+import org.veupathdb.lib.s3.workspaces.java.S3Workspace
+import java.io.File
+
+internal open class FuglyMinIOWorkspace(private val delegate: S3Workspace)
+  : AbstractMinIOHack()
+  , S3Workspace by delegate
+{
+  override fun exists() =
+    withRetries({ "test for workspace $path" }, delegate::exists)
+
+  override fun open(path: String) =
+    withRetries({ "open object ${objectPath(path)}" }) { delegate.open(path) }
+
+  override fun get(path: String) =
+    withRetries({ "get object ${objectPath(path)}" }) { delegate[path] }
+      ?.let(::FuglyMinIOWorkspaceFile)
+
+  override fun copy(from: String, to: File) =
+    withRetries({ "copy object ${objectPath(path)} to file ${to.path}" }) { delegate.copy(from, to) }
+
+  override fun files() =
+    withRetries({ "list files in workspace $path" }, delegate::files)
+      .map(::FuglyMinIOWorkspaceFile)
+
+  override fun contains(path: String) =
+    withRetries({ "stat object ${objectPath(path)}" }) { path in delegate }
+
+  override fun delete() =
+    delete(path, delegate::delete, ::exists)
+
+  override fun hasSubWorkspace(id: HashID) =
+    withRetries({ "test for sub-workspace ${objectPath(id.string)}" }) { delegate.hasSubWorkspace(id) }
+
+  override fun openSubWorkspace(id: HashID) =
+    withRetries({ "open sub-workspace ${objectPath(id.string)}" }) { delegate.openSubWorkspace(id) }
+      ?.let { FuglyMinIOSubWorkspace(it, this@FuglyMinIOWorkspace) }
+
+  override fun createSubWorkspace(id: HashID) =
+    FuglyMinIOSubWorkspace(
+      withRetries({ "create sub-workspace ${objectPath(id.string)}" }) { delegate.createSubWorkspace(id) },
+      this@FuglyMinIOWorkspace,
+    )
+
+  @Suppress("NOTHING_TO_INLINE")
+  private inline fun objectPath(name: String) = "$path/$name"
+}

lib/src/main/kotlin/org/veupathdb/lib/compute/platform/intern/minio/FuglyMinIOWorkspaceFactory.kt

@@ -0,0 +1,17 @@
+package org.veupathdb.lib.compute.platform.intern.minio
+
+import org.veupathdb.lib.hash_id.HashID
+import org.veupathdb.lib.s3.workspaces.java.S3WorkspaceFactory
+
+/**
+ * Now with more hacks!!
+ */
+internal class FuglyMinIOWorkspaceFactory(private val delegate: S3WorkspaceFactory): AbstractMinIOHack() {
+
+  fun get(jobID: HashID) =
+    withRetries({ "open workspace for job $jobID" }) { delegate.get(jobID) }
+      ?.let(::FuglyMinIOWorkspace)
+
+  fun create(jobID: HashID) =
+    FuglyMinIOWorkspace(withRetries({ "create workspace for job $jobID" }) { delegate.create(jobID) })
+}

lib/src/main/kotlin/org/veupathdb/lib/compute/platform/intern/minio/FuglyMinIOWorkspaceFile.kt

@@ -0,0 +1,30 @@
+package org.veupathdb.lib.compute.platform.intern.minio
+
+import org.veupathdb.lib.s3.workspaces.java.WorkspaceFile
+import java.io.File
+
+internal class FuglyMinIOWorkspaceFile(private val delegate: WorkspaceFile): AbstractMinIOHack(),
+WorkspaceFile by delegate {
+  override fun size() =
+    withRetries({ "stat object $absolutePath" }, delegate::size)
+
+  override fun open() =
+    withRetries({ "open object $absolutePath" }, delegate::open)
+
+  override fun download(localFile: File) =
+    withRetries({ "download object $absolutePath to file ${localFile.path}" }) { delegate.download(localFile) }
+
+  override fun delete() =
+    delete(absolutePath, delegate::delete, ::stat)
+
+  // workaround to check for object existence.  If the object doesn't exist, the
+  // underlying method throws an NPE when attempting to access the file size
+  // property.
+  private fun stat() =
+    try {
+      size()
+      true
+    } catch (_: NullPointerException) {
+      false
+    }
+}

lib/src/main/kotlin/org/veupathdb/lib/compute/platform/intern/minio/MinIOHax.kt

@@ -0,0 +1,74 @@
+package org.veupathdb.lib.compute.platform.intern.minio
+
+import org.slf4j.Logger
+import org.veupathdb.lib.s3.s34k.errors.S34KError
+import org.veupathdb.lib.s3.s34k.objects.S3Object
+import kotlin.time.Duration.Companion.milliseconds
+
+@Suppress("NOTHING_TO_INLINE")
+internal object MinIOHax {
+  var MaxDeleteAttempts = 5
+
+  var SleepMillis = 1_000L
+
+  var RetryCount = 3
+
+  fun sleepCountDuration(sleepCount: Int) = (sleepCount * SleepMillis).milliseconds
+
+  context(log: Logger)
+  fun <T> withRetries(actionMsg: () -> String, fn: () -> T) =
+    withRetries(RetryCount, actionMsg, fn)
+
+  context(log: Logger)
+  fun <T> withRetries(retries: Int, actionMsg: () -> String, fn: () -> T): T {
+    var lastError: S34KError? = null
+
+    for (i in 1..retries) {
+      try {
+        return fn()
+      } catch (e: S34KError) {
+        log.warn("failed {} time(s) to {}", i, actionMsg())
+
+        if (lastError != null)
+          e.addSuppressed(lastError)
+
+        lastError = e
+
+        sleep()
+      }
+    }
+
+    log.error("failed {} time(s) to {}", RetryCount, actionMsg())
+    throw lastError!!
+  }
+
+  context(log: Logger)
+  fun sleep() {
+    log.debug("sleeping for {}", { SleepMillis.milliseconds })
+    Thread.sleep(SleepMillis)
+  }
+
+  context(log: Logger)
+  fun delete(obj: S3Object) = delete(obj.path, obj::delete, obj::exists)
+
+  context(log: Logger)
+  fun delete(path: String, deleteFn: () -> Unit, statFn: () -> Boolean) {
+    // Tell minio to delete the object
+    deleteFn()
+
+    // Sleep for a bit to let minio ponder the delete request
+    sleep()
+
+    // If minio is being particularly dim, give it some more time to work it out
+    for (i in 1..MaxDeleteAttempts) {
+      if (!statFn())
+        return
+
+      log.debug("waiting for object {} deletion: {}", path, sleepCountDuration(i))
+      sleep()
+    }
+
+    log.error("waited {} for object {} deletion, but MinIO reports that it still exists", sleepCountDuration(MaxDeleteAttempts), path)
+    throw RuntimeException("object deletion timeout for MinIO object $path")
+  }
+}