feat: Add a service to run maintenance jobs

mnonnenmacher · mnonnenmacher · commit 74c7dade15f9 · 2024-09-16T15:02:10.000+02:00
Add an initial implementation of the `MaintenanceService` to run
maintenance jobs. This is required to run the package deduplication
which will be added in the next commit.

This first version is sufficient for the package deduplication but it
has several shortcoming that will have to be addressed later:

* The service runs on core, on the long term maintenance jobs should run
  on a separate node.
* It only supports jobs which run exactly once, support for running jobs
  multiple times or on a schedule is missing.
* The mechanism to prevent that a job is executed in parallel is based
  is based on a heuristic and needs to be replaced with a more
  sophisticated approach.

Signed-off-by: Martin Nonnenmacher &lt;martin.nonnenmacher@bosch.com&gt;
diff --git a/services/maintenance/build.gradle.kts b/services/maintenance/build.gradle.kts
@@ -0,0 +1,47 @@
+/*
+ * Copyright (C) 2024 The ORT Server Authors (See <https://github.com/eclipse-apoapsis/ort-server/blob/main/NOTICE>)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * License-Filename: LICENSE
+ */
+
+plugins {
+    // Apply precompiled plugins.
+    id("ort-server-kotlin-jvm-conventions")
+    id("ort-server-publication-conventions")
+
+    // Apply third-party plugins.
+    alias(libs.plugins.kotlinSerialization)
+}
+
+group = "org.eclipse.apoapsis.ortserver.services"
+
+dependencies {
+    api(projects.model)
+
+    implementation(projects.dao)
+    implementation(projects.utils.logging)
+
+    implementation(libs.kotlinxSerializationJson)
+
+    runtimeOnly(libs.logback)
+
+    testImplementation(testFixtures(projects.dao))
+    testImplementation(projects.utils.test)
+
+    testImplementation(libs.kotestAssertionsCore)
+    testImplementation(libs.kotestRunnerJunit5)
+    testImplementation(libs.mockk)
+}
diff --git a/services/maintenance/src/main/kotlin/MaintenanceJob.kt b/services/maintenance/src/main/kotlin/MaintenanceJob.kt
@@ -0,0 +1,43 @@
+/*
+ * Copyright (C) 2024 The ORT Server Authors (See <https://github.com/eclipse-apoapsis/ort-server/blob/main/NOTICE>)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * License-Filename: LICENSE
+ */
+
+package org.eclipse.apoapsis.ortserver.services.maintenance
+
+import org.eclipse.apoapsis.ortserver.model.MaintenanceJobData
+import org.eclipse.apoapsis.ortserver.utils.logging.withMdcContext
+
+abstract class MaintenanceJob {
+    abstract val name: String
+
+    var active = false
+
+    suspend fun start(service: MaintenanceService, jobData: MaintenanceJobData) {
+        active = true
+
+        try {
+            withMdcContext("maintenanceJob" to name) {
+                execute(service, jobData)
+            }
+        } finally {
+            active = false
+        }
+    }
+
+    abstract suspend fun execute(service: MaintenanceService, jobData: MaintenanceJobData)
+}
diff --git a/services/maintenance/src/main/kotlin/MaintenanceService.kt b/services/maintenance/src/main/kotlin/MaintenanceService.kt
@@ -0,0 +1,194 @@
+/*
+ * Copyright (C) 2024 The ORT Server Authors (See <https://github.com/eclipse-apoapsis/ort-server/blob/main/NOTICE>)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * License-Filename: LICENSE
+ */
+
+package org.eclipse.apoapsis.ortserver.services.maintenance
+
+import java.sql.Connection
+
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.minutes
+
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
+import kotlinx.datetime.Clock
+import kotlinx.serialization.json.JsonObject
+
+import org.eclipse.apoapsis.ortserver.dao.blockingQuery
+import org.eclipse.apoapsis.ortserver.dao.tables.MaintenanceJobDao
+import org.eclipse.apoapsis.ortserver.dao.tables.MaintenanceJobsTable
+import org.eclipse.apoapsis.ortserver.model.MaintenanceJobData
+import org.eclipse.apoapsis.ortserver.model.MaintenanceJobStatus
+
+import org.jetbrains.exposed.sql.Database
+import org.jetbrains.exposed.sql.and
+
+import org.slf4j.LoggerFactory
+
+private val logger = LoggerFactory.getLogger(MaintenanceService::class.java)
+
+/**
+ * A service to run maintenance jobs. The current implementation has the following limitations:
+ *
+ * - The service runs on core, on the long-term maintenance jobs should run on a separate node.
+ * - It only supports jobs which run exactly once, support for running jobs multiple times or on a schedule is missing.
+ * - The mechanism to prevent that a job is executed in parallel is based on a heuristic and needs to be
+ *   replaced with a more sophisticated approach.
+ */
+class MaintenanceService(private val db: Database, private val updateInterval: Duration = 5.minutes) {
+    private val jobs: MutableList<MaintenanceJob> = mutableListOf()
+    private val mutex = Mutex()
+
+    /**
+     * Add a [job] to the list of jobs to run. This function should be called before [run].
+     */
+    suspend fun addJob(job: MaintenanceJob) {
+        mutex.withLock { jobs += job }
+    }
+
+    /**
+     * Run the previously added jobs. This function is blocking and should be called from a coroutine.
+     */
+    suspend fun run() {
+        logger.info("Starting maintenance service.")
+
+        withContext(Dispatchers.IO) {
+            createJobs()
+
+            var uncompletedJobs = getUncompletedJobs()
+
+            while (uncompletedJobs.isNotEmpty()) {
+                // Check which jobs need to be started.
+                db.blockingQuery(transactionIsolation = Connection.TRANSACTION_SERIALIZABLE) {
+                    jobs.filterNot { it.active }.forEach { job ->
+                        logger.info("Checking if maintenance job '${job.name}' needs to be started.")
+                        val jobData = MaintenanceJobDao.find { MaintenanceJobsTable.name eq job.name }.firstOrNull()
+
+                        if (jobData == null) {
+                            logger.warn("Could not find job '${job.name}' in the database.")
+                            return@forEach
+                        }
+
+                        if (jobData.status !in MaintenanceJobStatus.uncompletedStates) {
+                            logger.info("Not starting maintenance job '${job.name}' as it is already completed.")
+                            return@forEach
+                        }
+
+                        // Only start the job if it was not updated in the last five minutes. This heuristic needs
+                        // to be replaced with a more reliable mechanism to find out if the job is already running.
+                        if (jobData.updatedAt?.let { it > Clock.System.now() - 5.minutes } == true) {
+                            logger.info("Not starting maintenance job '${job.name}' as it is already running.")
+                            return@forEach
+                        }
+
+                        logger.info("Starting maintenance job '${job.name}'.")
+
+                        val jobDataModel = jobData.mapToModel()
+
+                        launch {
+                            job.start(this@MaintenanceService, jobDataModel)
+                        }
+                    }
+                }
+
+                delay(updateInterval)
+
+                uncompletedJobs = getUncompletedJobs()
+            }
+
+            logger.info("All maintenance jobs have been completed.")
+        }
+    }
+
+    /**
+     * Create entries in the database for all jobs that are not yet present.
+     */
+    private fun createJobs() {
+        db.blockingQuery(transactionIsolation = Connection.TRANSACTION_SERIALIZABLE) {
+            val existingJobs = MaintenanceJobDao.all()
+
+            jobs.forEach { job ->
+                if (existingJobs.none { it.name == job.name }) {
+                    logger.info("Creating maintenance job '${job.name}'.")
+
+                    MaintenanceJobDao.new {
+                        name = job.name
+                        status = MaintenanceJobStatus.STARTED
+                        startedAt = Clock.System.now()
+                    }
+                }
+            }
+        }
+    }
+
+    /**
+     * Get all uncompleted jobs from the database.
+     */
+    private suspend fun getUncompletedJobs(): List<MaintenanceJobData> {
+        return mutex.withLock {
+            db.blockingQuery(transactionIsolation = Connection.TRANSACTION_SERIALIZABLE) {
+                val jobNames = jobs.map { it.name }
+                MaintenanceJobDao.find {
+                    MaintenanceJobsTable.name inList jobNames and
+                            (MaintenanceJobsTable.status inList MaintenanceJobStatus.uncompletedStates)
+                }.map { it.mapToModel() }
+            }
+        }
+    }
+
+    /**
+     * Update the job with the given [id] with the given [data] and/or [status].
+     */
+    internal fun updateJob(id: Long, data: JsonObject? = null, status: MaintenanceJobStatus? = null) {
+        db.blockingQuery(transactionIsolation = Connection.TRANSACTION_SERIALIZABLE) {
+            val job = MaintenanceJobDao.findById(id)
+
+            if (job == null) {
+                logger.warn("Could not find job with ID $id.")
+                return@blockingQuery
+            }
+
+            if (job.status.completed) {
+                logger.warn("Job '${job.name}' with ID $id is already completed.")
+                return@blockingQuery
+            }
+
+            val verb = when (status) {
+                MaintenanceJobStatus.FINISHED -> "Finish"
+                MaintenanceJobStatus.FAILED -> "Fail"
+                else -> "Update"
+            }
+
+            logger.info("$verb job '${job.name}' with ID $id.")
+
+            job.data = data
+            job.updatedAt = Clock.System.now()
+
+            if (status != null) {
+                job.status = status
+                if (status.completed) {
+                    job.finishedAt = Clock.System.now()
+                }
+            }
+        }
+    }
+}
diff --git a/services/maintenance/src/test/kotlin/MaintenanceServiceTest.kt b/services/maintenance/src/test/kotlin/MaintenanceServiceTest.kt
@@ -0,0 +1,107 @@
+/*
+ * Copyright (C) 2024 The ORT Server Authors (See <https://github.com/eclipse-apoapsis/ort-server/blob/main/NOTICE>)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * License-Filename: LICENSE
+ */
+
+package org.eclipse.apoapsis.ortserver.services.maintenance
+
+import io.kotest.core.spec.style.WordSpec
+import io.kotest.matchers.collections.containExactlyInAnyOrder
+import io.kotest.matchers.should
+import io.kotest.matchers.shouldBe
+
+import kotlin.time.Duration.Companion.milliseconds
+
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.decodeFromJsonElement
+import kotlinx.serialization.json.encodeToJsonElement
+import kotlinx.serialization.json.jsonObject
+
+import org.eclipse.apoapsis.ortserver.dao.findSingle
+import org.eclipse.apoapsis.ortserver.dao.tables.MaintenanceJobDao
+import org.eclipse.apoapsis.ortserver.dao.tables.MaintenanceJobsTable
+import org.eclipse.apoapsis.ortserver.dao.test.DatabaseTestExtension
+import org.eclipse.apoapsis.ortserver.model.MaintenanceJobData
+import org.eclipse.apoapsis.ortserver.model.MaintenanceJobStatus
+
+import org.jetbrains.exposed.sql.transactions.transaction
+
+class MaintenanceServiceTest : WordSpec({
+    val dbExtension = extension(DatabaseTestExtension())
+
+    "run" should {
+        "run the supplied jobs" {
+            val service = MaintenanceService(dbExtension.db, 100.milliseconds)
+
+            (1..5).forEach { index ->
+                service.addJob(object : MaintenanceJob() {
+                    override val name = "TestJob$index"
+                    override suspend fun execute(service: MaintenanceService, jobData: MaintenanceJobData) {
+                        service.updateJob(jobData.id, status = MaintenanceJobStatus.FINISHED)
+                    }
+                })
+            }
+
+            service.run()
+
+            transaction {
+                val jobData = MaintenanceJobDao.all().map { it.mapToModel() }
+                jobData.forEach { it.status shouldBe MaintenanceJobStatus.FINISHED }
+                jobData.map { it.name } should
+                        containExactlyInAnyOrder("TestJob1", "TestJob2", "TestJob3", "TestJob4", "TestJob5")
+            }
+        }
+    }
+
+    "updateJob" should {
+        "update the job data" {
+            val service = MaintenanceService(dbExtension.db, 100.milliseconds)
+            val jobId = createJob()
+            val jobData = JobData(42)
+
+            service.updateJob(jobId, data = Json.Default.encodeToJsonElement(jobData).jsonObject)
+
+            transaction {
+                val jobDao = MaintenanceJobDao.findSingle { MaintenanceJobsTable.id eq jobId }
+                jobDao.data?.let { Json.Default.decodeFromJsonElement<JobData>(it) } shouldBe jobData
+            }
+        }
+
+        "update the status" {
+            val service = MaintenanceService(dbExtension.db, 100.milliseconds)
+            val jobId = createJob()
+
+            service.updateJob(jobId, status = MaintenanceJobStatus.FAILED)
+
+            transaction {
+                val jobDao = MaintenanceJobDao.findSingle { MaintenanceJobsTable.id eq jobId }
+                jobDao.status shouldBe MaintenanceJobStatus.FAILED
+            }
+        }
+    }
+})
+
+@Serializable
+private data class JobData(val progress: Int)
+
+private fun createJob() = transaction {
+    MaintenanceJobDao.new {
+        name = "TestJob"
+        status = MaintenanceJobStatus.STARTED
+    }.id.value
+}
diff --git a/settings.gradle.kts b/settings.gradle.kts
