-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #326 from VEuPathDB/issue-259
Admin Reconciler Endpoint
- Loading branch information
Showing
30 changed files
with
323 additions
and
139 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| plugins { | ||
| kotlin("jvm") | ||
| } | ||
|
|
||
| dependencies { | ||
| implementation(platform(project(":platform"))) | ||
|
|
||
|
|
||
| implementation(project(":lib:app-db")) | ||
| implementation(project(":lib:cache-db")) | ||
| implementation(project(":lib:env")) | ||
| implementation(project(":lib:plugin-client")) | ||
| implementation(project(":lib:kafka")) | ||
| implementation(project(":lib:metrics")) | ||
| implementation(project(":lib:module-core")) | ||
| implementation(project(":lib:plugin-mapping")) | ||
| implementation(project(":lib:s3")) | ||
|
|
||
| implementation("org.veupathdb.vdi:vdi-component-common") | ||
|
|
||
| implementation("org.veupathdb.lib.s3:s34k-minio") | ||
|
|
||
| implementation("org.apache.logging.log4j:log4j-api-kotlin") | ||
|
|
||
| implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core") | ||
|
|
||
| testImplementation(kotlin("test")) | ||
| testImplementation("org.junit.jupiter:junit-jupiter-api:5.9.2") | ||
| testImplementation("org.mockito:mockito-core:5.2.0") | ||
| testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:5.9.2") | ||
| testRuntimeOnly("org.apache.logging.log4j:log4j-slf4j-impl") | ||
| } | ||
|
|
||
| tasks.test { | ||
| useJUnitPlatform() | ||
|
|
||
| testLogging.showStandardStreams = true | ||
| } |
2 changes: 1 addition & 1 deletion
2
...tlin/vdi/daemon/reconciler/AppDBTarget.kt → .../kotlin/vdi/lib/reconciler/AppDBTarget.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
...in/vdi/daemon/reconciler/CacheDBTarget.kt → ...otlin/vdi/lib/reconciler/CacheDBTarget.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
111 changes: 111 additions & 0 deletions
111
lib/reconciler/src/main/kotlin/vdi/lib/reconciler/Reconciler.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| package vdi.lib.reconciler | ||
|
|
||
| import kotlinx.coroutines.coroutineScope | ||
| import kotlinx.coroutines.launch | ||
| import kotlinx.coroutines.runBlocking | ||
| import kotlinx.coroutines.sync.Mutex | ||
| import kotlinx.coroutines.sync.withLock | ||
| import org.apache.logging.log4j.kotlin.CoroutineThreadContext | ||
| import org.apache.logging.log4j.kotlin.ThreadContextData | ||
| import org.apache.logging.log4j.kotlin.logger | ||
| import org.veupathdb.lib.s3.s34k.S3Api | ||
| import vdi.component.db.app.AppDatabaseRegistry | ||
| import vdi.component.kafka.router.KafkaRouter | ||
| import vdi.component.kafka.router.KafkaRouterFactory | ||
| import vdi.component.metrics.Metrics | ||
| import vdi.component.modules.AbortCB | ||
| import vdi.component.s3.DatasetManager | ||
|
|
||
| object Reconciler { | ||
| private val logger = logger().delegate | ||
|
|
||
| private val config = ReconcilerConfig() | ||
|
|
||
| private val active = Mutex() | ||
|
|
||
| private var initialized = false | ||
|
|
||
| private lateinit var datasetManager: DatasetManager | ||
|
|
||
| private lateinit var kafkaRouter: KafkaRouter | ||
|
|
||
| fun initialize(abortCB: AbortCB) { | ||
| // We lock while initializing to avoid double init. | ||
| if (active.isLocked) | ||
| return | ||
|
|
||
| // If we've already successfully initialized, nothing to do. | ||
| if (initialized) | ||
| return | ||
|
|
||
| // Initialize | ||
| runBlocking { | ||
| active.withLock { | ||
| try { | ||
| val s3 = S3Api.newClient(config.s3Config) | ||
| val bucket = s3.buckets[config.s3Bucket] | ||
| ?: throw IllegalStateException("S3 bucket ${config.s3Bucket} does not exist!") | ||
|
|
||
| datasetManager = DatasetManager(bucket) | ||
| } catch (e: Throwable) { | ||
| logger.error("failed to init dataset manager", e) | ||
| abortCB(e.message) | ||
| } | ||
|
|
||
| try { | ||
| kafkaRouter = KafkaRouterFactory(config.kafkaRouterConfig).newKafkaRouter() | ||
| } catch (e: Throwable) { | ||
| logger.error("failed to init kafka router", e) | ||
| abortCB(e.message) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| suspend fun runFull(): Boolean { | ||
| if (active.isLocked) | ||
| return false | ||
|
|
||
| val targets = AppDatabaseRegistry.asSequence() | ||
| .map { (project, _) -> AppDBTarget(project, project) } | ||
| .toMutableList<ReconcilerTarget>() | ||
| targets.add(CacheDBTarget()) | ||
|
|
||
| val timer = Metrics.Reconciler.Full.reconcilerTimes.startTimer() | ||
|
|
||
| logger.info("running full reconciler for ${targets.size} targets") | ||
|
|
||
| coroutineScope { | ||
| targets.forEach { | ||
| launch(CoroutineThreadContext(ThreadContextData(mapOf("workerID" to workerName(false, it))))) { | ||
| ReconcilerInstance(it, datasetManager, kafkaRouter, false, config.deletesEnabled).reconcile() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| timer.observeDuration() | ||
|
|
||
| return true | ||
| } | ||
|
|
||
| suspend fun runSlim(): Boolean { | ||
| if (active.isLocked) | ||
| return false | ||
|
|
||
| val timer = Metrics.Reconciler.Slim.executionTime.startTimer() | ||
|
|
||
| val target = CacheDBTarget() | ||
|
|
||
| coroutineScope { | ||
| launch(CoroutineThreadContext(ThreadContextData(mapOf("workerID" to workerName(true, target))))) { | ||
| ReconcilerInstance(target, datasetManager, kafkaRouter, true, false).reconcile() | ||
| } | ||
| } | ||
|
|
||
| timer.observeDuration() | ||
|
|
||
| return true | ||
| } | ||
|
|
||
| private fun workerName(slim: Boolean, tgt: ReconcilerTarget) = if (slim) "slim-${tgt.name}" else "full-${tgt.name}" | ||
| } |
31 changes: 31 additions & 0 deletions
31
lib/reconciler/src/main/kotlin/vdi/lib/reconciler/ReconcilerConfig.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| package vdi.lib.reconciler | ||
|
|
||
| import org.veupathdb.lib.s3.s34k.S3Config | ||
| import org.veupathdb.lib.s3.s34k.fields.BucketName | ||
| import org.veupathdb.vdi.lib.common.env.Environment | ||
| import org.veupathdb.vdi.lib.common.env.optBool | ||
| import org.veupathdb.vdi.lib.common.env.require | ||
| import vdi.component.env.EnvKey | ||
| import vdi.component.kafka.router.KafkaRouterConfig | ||
| import vdi.component.s3.util.S3Config | ||
|
|
||
| internal data class ReconcilerConfig( | ||
| val kafkaRouterConfig: KafkaRouterConfig, | ||
| val s3Config: S3Config, | ||
| val s3Bucket: BucketName, | ||
| val deletesEnabled: Boolean, | ||
| ) { | ||
| constructor() : this(System.getenv()) | ||
|
|
||
| constructor(env: Environment) : this( | ||
| kafkaRouterConfig = KafkaRouterConfig(env, "reconciler"), | ||
| s3Config = S3Config(env), | ||
| s3Bucket = BucketName(env.require(EnvKey.S3.BucketName)), | ||
| deletesEnabled = env.optBool(EnvKey.Reconciler.DeletesEnabled) ?: DefaultDeletesEnabled, | ||
| ) | ||
|
|
||
| companion object { | ||
| const val DefaultDeletesEnabled = true | ||
| } | ||
| } | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
...vdi/daemon/reconciler/ReconcilerTarget.kt → ...in/vdi/lib/reconciler/ReconcilerTarget.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
...daemon/reconciler/ReconcilerTargetType.kt → ...di/lib/reconciler/ReconcilerTargetType.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,3 @@ | ||
| package vdi.daemon.reconciler | ||
| package vdi.lib.reconciler | ||
|
|
||
| internal enum class ReconcilerTargetType { Cache, Install } |
2 changes: 1 addition & 1 deletion
2
...on/reconciler/UnsupportedTypeException.kt → ...ib/reconciler/UnsupportedTypeException.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,3 @@ | ||
| package vdi.daemon.reconciler | ||
| package vdi.lib.reconciler | ||
|
|
||
| internal class UnsupportedTypeException(message: String) : Exception(message) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.