diff --git a/run/src/main/kotlin/com/cosmotech/run/service/RunServiceImpl.kt b/run/src/main/kotlin/com/cosmotech/run/service/RunServiceImpl.kt index 4ff93ee40..5c5eb1f87 100644 --- a/run/src/main/kotlin/com/cosmotech/run/service/RunServiceImpl.kt +++ b/run/src/main/kotlin/com/cosmotech/run/service/RunServiceImpl.kt @@ -333,11 +333,10 @@ class RunServiceImpl( getRun(runner.organizationId, runner.workspaceId, runner.id, runner.lastRunInfo.lastRunId!!) run.hasPermission(PERMISSION_WRITE) - if (!(run.state!!.isTerminal())) { + check(!(run.state!!.isTerminal())) { logger.warn( "Run ${run.id} is already in a terminal state (${run.state}). It can't be stopped." ) - return // exiting, run already stopped } workflowService.stopWorkflow(run) diff --git a/run/src/test/kotlin/com/cosmotech/run/service/RunServiceImplTests.kt b/run/src/test/kotlin/com/cosmotech/run/service/RunServiceImplTests.kt new file mode 100644 index 000000000..ba26381a4 --- /dev/null +++ b/run/src/test/kotlin/com/cosmotech/run/service/RunServiceImplTests.kt @@ -0,0 +1,147 @@ +// Copyright (c) Cosmo Tech. +// Licensed under the MIT license. +package com.cosmotech.run.service + +import com.cosmotech.common.events.RunStop +import com.cosmotech.common.rbac.CsmRbac +import com.cosmotech.common.rbac.ROLE_ADMIN +import com.cosmotech.run.RunContainerFactory +import com.cosmotech.run.domain.Run +import com.cosmotech.run.domain.RunEditInfo +import com.cosmotech.run.domain.RunState +import com.cosmotech.run.domain.RunStatus +import com.cosmotech.run.repository.RunRepository +import com.cosmotech.run.workflow.WorkflowService +import com.cosmotech.runner.RunnerApiServiceInterface +import com.cosmotech.runner.domain.LastRunInfo +import com.cosmotech.runner.domain.Runner +import com.cosmotech.runner.domain.RunnerAccessControl +import com.cosmotech.runner.domain.RunnerDatasets +import com.cosmotech.runner.domain.RunnerEditInfo +import com.cosmotech.runner.domain.RunnerSecurity +import com.cosmotech.runner.domain.RunnerValidationStatus +import io.mockk.MockKAnnotations +import io.mockk.every +import io.mockk.impl.annotations.InjectMockKs +import io.mockk.impl.annotations.MockK +import io.mockk.impl.annotations.RelaxedMockK +import io.mockk.junit5.MockKExtension +import io.mockk.slot +import io.mockk.verify +import java.util.Optional +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertFailsWith +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.extension.ExtendWith + +private const val ORGANIZATION_ID = "o-organizationid" +private const val WORKSPACE_ID = "w-workspaceid" +private const val RUNNER_ID = "r-runnerid" +private const val RUN_ID = "run-runid" + +@ExtendWith(MockKExtension::class) +class RunServiceImplTests { + + @Suppress("unused") @MockK private lateinit var containerFactory: RunContainerFactory + @MockK private lateinit var workflowService: WorkflowService + @MockK private lateinit var runnerApiService: RunnerApiServiceInterface + @MockK private lateinit var runRepository: RunRepository + @Suppress("unused") @RelaxedMockK private lateinit var csmRbac: CsmRbac + + @InjectMockKs private lateinit var runServiceImpl: RunServiceImpl + + @BeforeTest + fun setUp() { + MockKAnnotations.init(this) + } + + private fun buildRunner(lastRunId: String? = RUN_ID) = + Runner( + id = RUNNER_ID, + name = "runner", + createInfo = RunnerEditInfo(timestamp = 0L, userId = "user"), + updateInfo = RunnerEditInfo(timestamp = 0L, userId = "user"), + solutionId = "sol-id", + runTemplateId = "runtemplate-id", + organizationId = ORGANIZATION_ID, + workspaceId = WORKSPACE_ID, + datasets = RunnerDatasets(bases = mutableListOf(), parameter = ""), + parametersValues = mutableListOf(), + lastRunInfo = + LastRunInfo( + lastRunId = lastRunId, + lastRunStatus = LastRunInfo.LastRunStatus.Running, + ), + validationStatus = RunnerValidationStatus.Validated, + security = + RunnerSecurity( + default = ROLE_ADMIN, + accessControlList = + mutableListOf(RunnerAccessControl(id = "user", role = ROLE_ADMIN)), + ), + ) + + private fun buildRun(state: RunState?) = + Run( + id = RUN_ID, + organizationId = ORGANIZATION_ID, + workspaceId = WORKSPACE_ID, + runnerId = RUNNER_ID, + state = state, + createInfo = RunEditInfo(timestamp = 0L, userId = "user"), + ) + + @Test + fun `onRunStop stops workflow and sets run state to Failed when run is not in terminal state`() { + val runner = buildRunner() + val run = buildRun(RunState.Running) + val runStopRequest = RunStop(this, runner) + + every { runnerApiService.getRunner(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID) } returns runner + every { runRepository.findBy(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID, RUN_ID) } returns + Optional.of(run) + every { workflowService.getRunStatus(run) } returns RunStatus(id = RUN_ID, phase = "Running") + every { workflowService.stopWorkflow(run) } returns RunStatus(id = RUN_ID) + + val savedRunSlot = slot() + every { runRepository.save(capture(savedRunSlot)) } answers { savedRunSlot.captured } + + runServiceImpl.onRunStop(runStopRequest) + + verify(exactly = 1) { workflowService.stopWorkflow(run) } + assertEquals(RunState.Failed, savedRunSlot.captured.state) + } + + @Test + fun `onRunStop throws IllegalStateException when run is Successful (terminal state)`() { + val runner = buildRunner() + val run = buildRun(RunState.Successful) + val runStopRequest = RunStop(this, runner) + + every { runnerApiService.getRunner(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID) } returns runner + every { runRepository.findBy(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID, RUN_ID) } returns + Optional.of(run) + + assertFailsWith { runServiceImpl.onRunStop(runStopRequest) } + + verify(exactly = 0) { workflowService.stopWorkflow(any()) } + verify(exactly = 0) { runRepository.save(any()) } + } + + @Test + fun `onRunStop throws IllegalStateException when run is Failed (terminal state)`() { + val runner = buildRunner() + val run = buildRun(RunState.Failed) + val runStopRequest = RunStop(this, runner) + + every { runnerApiService.getRunner(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID) } returns runner + every { runRepository.findBy(ORGANIZATION_ID, WORKSPACE_ID, RUNNER_ID, RUN_ID) } returns + Optional.of(run) + + assertFailsWith { runServiceImpl.onRunStop(runStopRequest) } + + verify(exactly = 0) { workflowService.stopWorkflow(any()) } + verify(exactly = 0) { runRepository.save(any()) } + } +} diff --git a/runner/src/main/kotlin/com/cosmotech/runner/service/RunnerApiServiceImpl.kt b/runner/src/main/kotlin/com/cosmotech/runner/service/RunnerApiServiceImpl.kt index 5bd363c73..85c175528 100644 --- a/runner/src/main/kotlin/com/cosmotech/runner/service/RunnerApiServiceImpl.kt +++ b/runner/src/main/kotlin/com/cosmotech/runner/service/RunnerApiServiceImpl.kt @@ -122,11 +122,17 @@ internal class RunnerApiServiceImpl( val runnerService = getRunnerService().inOrganization(organizationId).inWorkspace(workspaceId) val runnerInstance = runnerService.getInstance(runnerId).userHasPermission(PERMISSION_DELETE) + if (runnerInstance.getRunnerDataObject().status == RunnerStatus.Archived) { + throw IllegalStateException( + "This runner has already been mark as archived and so is going to be delete" + ) + } + + runnerService.archiveInstance(runnerInstance) + // Set runner status to Archived for future deletion (allow async process) runnerInstance.runner.status = RunnerStatus.Archived runnerService.saveInstance(runnerInstance.stamp()) - - runnerService.archiveInstance(runnerInstance) } override fun listRunners( diff --git a/runner/src/main/kotlin/com/cosmotech/runner/service/RunnerService.kt b/runner/src/main/kotlin/com/cosmotech/runner/service/RunnerService.kt index 88cb9a9ba..ea42701fe 100644 --- a/runner/src/main/kotlin/com/cosmotech/runner/service/RunnerService.kt +++ b/runner/src/main/kotlin/com/cosmotech/runner/service/RunnerService.kt @@ -122,7 +122,6 @@ class RunnerService( fun archiveInstance(runnerInstance: RunnerInstance) { val runner = runnerInstance.getRunnerDataObject() - // Check there are no running runs val hasRunningRuns = HasRunningRuns(this, runner.organizationId, runner.workspaceId, runner.id) this.eventPublisher.publishEvent(hasRunningRuns) @@ -131,7 +130,6 @@ class RunnerService( "Can't delete runner ${runner.id}: at least one run is still running" ) } - // Update parent and root references to deleted runner val newRoots = mutableListOf() listAllRunnerByParentId(runner.organizationId, runner.workspaceId, runner.id).forEach { @@ -287,13 +285,15 @@ class RunnerService( val runId = startEvent.response ?: throw IllegalStateException("Run Service did not respond") runnerInstance.setLastRunInfo(runId) runnerRepository.save(runnerInstance.getRunnerDataObject()) + return CreatedRun(id = runId) } fun stopLastRunOf(runnerInstance: RunnerInstance) { val runner = runnerInstance.getRunnerDataObject() runner.lastRunInfo.lastRunId ?: return // No run to stop - this.eventPublisher.publishEvent(RunStop(this, runner)) + val runStopEvent = RunStop(this, runner) + this.eventPublisher.publishEvent(runStopEvent) } fun cleanupArchived() { @@ -301,23 +301,28 @@ class RunnerService( val archivedRunners: List = runnerRepository.findAllByStatus(RunnerStatus.Archived) archivedRunners.forEach { runner -> - val cleanupEvent = CleanUpRun(this, runner.lastRunInfo.lastRunId!!) - this.eventPublisher.publishEvent(cleanupEvent) - if (cleanupEvent.response == true) { - runnerRepository.delete(runner) - val runnerEvent = - RunnerDeleted( - this, - runner.organizationId, - runner.workspaceId, - runner.datasets.parameter, - ) - this.eventPublisher.publishEvent(runnerEvent) + // If a runner has a last run, we need to cleanup the run before deleting the runner + if (runner.lastRunInfo.lastRunId != null) { + val cleanupEvent = CleanUpRun(this, runner.lastRunInfo.lastRunId!!) + this.eventPublisher.publishEvent(cleanupEvent) + if (cleanupEvent.response == false) + // The cleanup run process is not finished, we will not delete the runner yet + return@forEach } + + runnerRepository.delete(runner) + val runnerEvent = + RunnerDeleted( + this, + runner.organizationId, + runner.workspaceId, + runner.datasets.parameter, + ) + this.eventPublisher.publishEvent(runnerEvent) } } - @Suppress("TooManyFunctions") + @Suppress("TooManyFunctions", "LargeClass") inner class RunnerInstance { private val roleDefinition: RolesDefinition = getRunnerRolesDefinition() lateinit var runner: Runner diff --git a/runner/src/main/kotlin/com/cosmotech/runner/tasks/RunnerScheduledTasks.kt b/runner/src/main/kotlin/com/cosmotech/runner/tasks/RunnerScheduledTasks.kt index f61d68acf..936c6086c 100644 --- a/runner/src/main/kotlin/com/cosmotech/runner/tasks/RunnerScheduledTasks.kt +++ b/runner/src/main/kotlin/com/cosmotech/runner/tasks/RunnerScheduledTasks.kt @@ -4,10 +4,13 @@ package com.cosmotech.runner.tasks import com.cosmotech.runner.service.RunnerService import java.util.concurrent.TimeUnit +import org.slf4j.LoggerFactory import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty import org.springframework.scheduling.annotation.Scheduled import org.springframework.stereotype.Component +private val logger = LoggerFactory.getLogger(RunnerScheduledTasks::class.java) + @Component @ConditionalOnProperty( name = ["csm.platform.tasks.cleanUpArchivedRunners.enabled"], @@ -23,6 +26,7 @@ class RunnerScheduledTasks( fixedDelayString = "\${csm.platform.tasks.cleanUpArchivedRunners.delay}", ) fun cleanupArchivedRunners() { + logger.debug("Scheduled task: Cleaning up archived runners") runnerService.cleanupArchived() } }