Update agent statuses when they are assigned new jobs (#316)

* Update agent statuses when they are assigned new jobs * Repeat check for agent shutdown twice with 5 seconds delay * Add `STOPED_BY_ORCH` * Add debug logging in backend * Assign agent_id to test_execution in the DB * Store timestamps in the DB up to milliseconds Closes #305
saveourtool · Oct 15, 2021 · 3fd4296 · 3fd4296 · 0pdd · Oct 15, 2021
1 parent fa1ecd7
commit 3fd4296
Show file tree

Hide file tree

Showing 17 changed files with 178 additions and 51 deletions.
diff --git a/db/changelog/v-1/02-changeset-agent-status-table.xml b/db/changelog/v-1/02-changeset-agent-status-table.xml
@@ -18,10 +18,10 @@
             <column name="state" type="varchar(50)">
                 <constraints nullable="false"/>
             </column>
-            <column name="start_time" type="DATETIME">
+            <column name="start_time" type="DATETIME(3)">
                 <constraints nullable="false"/>
             </column>
-            <column name="end_time" type="DATETIME">
+            <column name="end_time" type="DATETIME(3)">
                 <constraints nullable="false"/>
             </column>
             <column name="agent_id" type="bigint" />

diff --git a/db/changelog/v-1/03-changeset-test-execution-table.xml b/db/changelog/v-1/03-changeset-test-execution-table.xml
@@ -27,10 +27,10 @@
             <column name="status" type="varchar(50)" defaultValue="READY">
                 <constraints nullable="false"/>
             </column>
-            <column name="start_time" type="DATETIME">
+            <column name="start_time" type="DATETIME(3)">
                 <constraints nullable="true"/>
             </column>
-            <column name="end_time" type="DATETIME">
+            <column name="end_time" type="DATETIME(3)">
                 <constraints nullable="true" />
             </column>
         </createTable>

diff --git a/db/changelog/v-1/11-changeset-test-agent-insert-data.xml b/db/changelog/v-1/11-changeset-test-agent-insert-data.xml
@@ -88,8 +88,8 @@
     <changeSet id="agent-status-insert" author="petertrr" context="dev">
         <loadData tableName="agent_status" encoding="UTF-8" separator=";" quotchar="&quot;" file="db/changelog/v-1/sqlRequests/agent-status.csv">
             <column header="id" name="id" type="NUMERIC" />
-            <column header="start_time" name="start_time" type="DATETIME" />
-            <column header="end_time" name="end_time" type="DATETIME" />
+            <column header="start_time" name="start_time" type="DATETIME(3)" />
+            <column header="end_time" name="end_time" type="DATETIME(3)" />
             <column header="state" name="state" type="STRING" />
             <column header="agent_id" name="agent_id" type="NUMERIC" />
         </loadData>

diff --git a/save-backend/src/main/kotlin/org/cqfn/save/backend/controllers/AgentsController.kt b/save-backend/src/main/kotlin/org/cqfn/save/backend/controllers/AgentsController.kt
@@ -7,7 +7,6 @@ import org.cqfn.save.entities.Agent
 import org.cqfn.save.entities.AgentStatus
 import org.cqfn.save.entities.AgentStatusDto
 import org.cqfn.save.entities.AgentStatusesForExecution
-import org.cqfn.save.entities.Execution
 import org.slf4j.LoggerFactory
 import org.springframework.transaction.annotation.Transactional
 import org.springframework.web.bind.annotation.GetMapping
@@ -60,7 +59,7 @@ class AgentsController(private val agentStatusRepository: AgentStatusRepository,
     @Transactional
     fun updateAgentStatusesWithDto(@RequestBody agentStates: List<AgentStatusDto>) {
         agentStates.forEach { dto ->
-            val agentStatus = agentStatusRepository.findTopByAgentContainerIdOrderByEndTimeDesc(dto.containerId)
+            val agentStatus = agentStatusRepository.findTopByAgentContainerIdOrderByEndTimeDescIdDesc(dto.containerId)
             if (agentStatus != null && agentStatus.state == dto.state) {
                 // updating time
                 agentStatus.endTime = dto.time
@@ -85,11 +84,9 @@ class AgentsController(private val agentStatusRepository: AgentStatusRepository,
     @Suppress("UnsafeCallOnNullableType")  // id will be available because it's retrieved from DB
     fun findAllAgentStatusesForSameExecution(@RequestParam agentId: String): AgentStatusesForExecution {
         val execution = getAgentByContainerId(agentId).execution
-        val agentStatuses = agentRepository.findAll { root, cq, cb ->
-            cb.equal(root.get<Execution>("execution"), execution)
-        }.map { agent ->
+        val agentStatuses = agentRepository.findByExecutionId(execution.id!!).map { agent ->
             val latestStatus = requireNotNull(
-                agentStatusRepository.findTopByAgentContainerIdOrderByEndTimeDesc(agent.containerId)
+                agentStatusRepository.findTopByAgentContainerIdOrderByEndTimeDescIdDesc(agent.containerId)
             ) {
                 "AgentStatus not found for agent id=${agent.containerId}"
             }

diff --git a/save-backend/src/main/kotlin/org/cqfn/save/backend/controllers/TestExecutionController.kt b/save-backend/src/main/kotlin/org/cqfn/save/backend/controllers/TestExecutionController.kt
@@ -2,6 +2,7 @@ package org.cqfn.save.backend.controllers
 
 import org.cqfn.save.agent.TestExecutionDto
 import org.cqfn.save.backend.service.TestExecutionService
+import org.cqfn.save.test.TestDto
 import org.springframework.dao.DataAccessException
 import org.springframework.http.HttpStatus
 import org.springframework.http.ResponseEntity
@@ -45,6 +46,15 @@ class TestExecutionController(private val testExecutionService: TestExecutionSer
     fun getTestExecutionsCount(@RequestParam executionId: Long) =
             testExecutionService.getTestExecutionsCount(executionId)
 
+    /**
+     * @param agentContainerId id of an agent
+     * @param testDtos test that will be executed by [agentContainerId] agent
+     */
+    @PostMapping(value = ["/testExecution/assignAgent"])
+    fun assignAgentByTest(@RequestParam agentContainerId: String, @RequestBody testDtos: List<TestDto>) {
+        testExecutionService.assignAgentByTest(agentContainerId, testDtos)
+    }
+
     /**
      * @param testExecutionsDto
      * @return response

diff --git a/save-backend/src/main/kotlin/org/cqfn/save/backend/repository/AgentStatusRepository.kt b/save-backend/src/main/kotlin/org/cqfn/save/backend/repository/AgentStatusRepository.kt
@@ -20,7 +20,7 @@ interface AgentStatusRepository : BaseEntityRepository<AgentStatus>, JpaSpecific
      * @param containerId id of an agent
      * @return [AgentStatus] of an agent
      */
-    fun findTopByAgentContainerIdOrderByEndTimeDesc(containerId: String): AgentStatus?
+    fun findTopByAgentContainerIdOrderByEndTimeDescIdDesc(containerId: String): AgentStatus?
 }
 
 /**

diff --git a/save-backend/src/main/kotlin/org/cqfn/save/backend/service/TestExecutionService.kt b/save-backend/src/main/kotlin/org/cqfn/save/backend/service/TestExecutionService.kt
@@ -8,11 +8,13 @@ import org.cqfn.save.backend.repository.TestRepository
 import org.cqfn.save.backend.utils.secondsToLocalDateTime
 import org.cqfn.save.domain.TestResultStatus
 import org.cqfn.save.entities.TestExecution
+import org.cqfn.save.test.TestDto
 
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.data.domain.PageRequest
 import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
 
 /**
  * Service for test result
@@ -56,6 +58,7 @@ class TestExecutionService(private val testExecutionRepository: TestExecutionRep
      */
     @Suppress("TOO_MANY_LINES_IN_LAMBDA", "UnsafeCallOnNullableType")
     fun saveTestResult(testExecutionsDtos: List<TestExecutionDto>): List<TestExecutionDto> {
+        log.debug("Saving ${testExecutionsDtos.size} test results from agent ${testExecutionsDtos.first().agentContainerId}")
         // we take agent id only from first element, because all test executions have same execution
         val agentContainerId = requireNotNull(testExecutionsDtos.first().agentContainerId) {
             "Attempt to save test results without assigned agent. testExecutionDtos=$testExecutionsDtos"
@@ -111,4 +114,33 @@ class TestExecutionService(private val testExecutionRepository: TestExecutionRep
             )
         }
     }
+
+    /**
+     * Set `agent` field of test executions corresponding to [testDtos] to [agentContainerId]
+     *
+     * @param agentContainerId id of an agent
+     * @param testDtos test that will be executed by [agentContainerId] agent
+     */
+    @Transactional
+    @Suppress("UnsafeCallOnNullableType")
+    fun assignAgentByTest(agentContainerId: String, testDtos: List<TestDto>) {
+        val agent = requireNotNull(agentRepository.findByContainerId(agentContainerId)) {
+            "Agent with containerId=[$agentContainerId] was not found in the DB"
+        }
+        val executionId = agent.execution.id!!
+        testDtos.forEach { test ->
+            val testExecution = testExecutionRepository.findByExecutionIdAndTestPluginNameAndTestFilePath(
+                executionId,
+                test.pluginName,
+                test.filePath
+            )
+                .orElseThrow {
+                    log.error("Can't find test_execution for executionId=$executionId, test.pluginName=${test.pluginName}, test.filePath=${test.filePath}")
+                    NoSuchElementException()
+                }
+            testExecutionRepository.save(testExecution.apply {
+                this.agent = agent
+            })
+        }
+    }
 }
diff --git a/save-backend/src/main/resources/logback.xml b/save-backend/src/main/resources/logback.xml
@@ -5,6 +5,8 @@
         </encoder>
     </appender>
 
+    <logger name="org.cqfn.save.backend.service.TestService" level="DEBUG"/>
+    <logger name="org.cqfn.save.backend.service.TestExecutionService" level="DEBUG"/>
     <root level="info">
         <appender-ref ref="STDOUT" />
     </root>

diff --git a/save-cloud-common/src/commonMain/kotlin/org/cqfn/save/agent/AgentState.kt b/save-cloud-common/src/commonMain/kotlin/org/cqfn/save/agent/AgentState.kt
@@ -41,5 +41,10 @@ enum class AgentState {
      * Agent has just started and hasn't received any heartbeats yet
      */
     STARTING,
+
+    /**
+     * Agent has been stopped by save-orchestartor, because there is no more work left
+     */
+    STOPPED_BY_ORCH,
     ;
 }
diff --git a/save-orchestrator/src/main/kotlin/org/cqfn/save/orchestrator/config/ConfigProperties.kt b/save-orchestrator/src/main/kotlin/org/cqfn/save/orchestrator/config/ConfigProperties.kt
@@ -15,6 +15,7 @@ import org.springframework.boot.context.properties.ConstructorBinding
  * @property docker configuration for docker API
  * @property agentsCount a number of agents to start for every [Execution]
  * @property executionLogs path to folder to store cli logs
+ * @property shutdownChecksIntervalMillis interval between checks whether agents are really finished
  */
 @ConstructorBinding
 @ConfigurationProperties(prefix = "orchestrator")
@@ -24,6 +25,7 @@ data class ConfigProperties(
     val docker: DockerSettings,
     val agentsCount: Int,
     val executionLogs: String,
+    val shutdownChecksIntervalMillis: Long,
 )
 
 /**

diff --git a/...orchestrator/src/main/kotlin/org/cqfn/save/orchestrator/controller/HeartbeatController.kt b/...orchestrator/src/main/kotlin/org/cqfn/save/orchestrator/controller/HeartbeatController.kt
@@ -3,7 +3,6 @@ package org.cqfn.save.orchestrator.controller
 import org.cqfn.save.agent.AgentState
 import org.cqfn.save.agent.ContinueResponse
 import org.cqfn.save.agent.Heartbeat
-import org.cqfn.save.agent.NewJobResponse
 import org.cqfn.save.agent.WaitResponse
 import org.cqfn.save.entities.AgentStatusDto
 import org.cqfn.save.orchestrator.config.ConfigProperties
@@ -15,8 +14,8 @@ import org.springframework.web.bind.annotation.PostMapping
 import org.springframework.web.bind.annotation.RequestBody
 import org.springframework.web.bind.annotation.RestController
 import reactor.core.publisher.Mono
-import reactor.core.scheduler.Schedulers
 
+import java.time.Duration
 import java.time.LocalDateTime
 
 import kotlinx.serialization.ExperimentalSerializationApi
@@ -34,7 +33,6 @@ class HeartbeatController(private val agentService: AgentService,
                           private val dockerService: DockerService,
                           private val configProperties: ConfigProperties) {
     private val logger = LoggerFactory.getLogger(HeartbeatController::class.java)
-    private val scheduler = Schedulers.boundedElastic().also { it.start() }
 
     /**
      * This controller accepts heartbeat and depending on the state it returns the needed response
@@ -66,18 +64,14 @@ class HeartbeatController(private val agentService: AgentService,
                         .doOnSuccess {
                             if (it is WaitResponse) {
                                 initiateShutdownSequence(heartbeat.agentId)
-                            } else if (it is NewJobResponse) {
-                                logger.debug("Agent ${heartbeat.agentId} will receive the following job: $it")
                             }
                         }
                     AgentState.FINISHED -> {
                         agentService.checkSavedData()
                         Mono.just(WaitResponse)
                     }
                     AgentState.BUSY -> Mono.just(ContinueResponse)
-                    AgentState.BACKEND_FAILURE -> Mono.just(WaitResponse)
-                    AgentState.BACKEND_UNREACHABLE -> Mono.just(WaitResponse)
-                    AgentState.CLI_FAILED -> Mono.just(WaitResponse)
+                    AgentState.BACKEND_FAILURE, AgentState.BACKEND_UNREACHABLE, AgentState.CLI_FAILED, AgentState.STOPPED_BY_ORCH -> Mono.just(WaitResponse)
                 }
             )
             .map {
@@ -93,21 +87,35 @@ class HeartbeatController(private val agentService: AgentService,
      * @param agentId an ID of the agent from the execution, that will be checked.
      */
     private fun initiateShutdownSequence(agentId: String) {
-        agentService.getAgentsAwaitingStop(agentId).doOnSuccess { (executionId, finishedAgentIds) ->
-            scheduler.schedule {
+        agentService.getAgentsAwaitingStop(agentId).flatMap { (_, finishedAgentIds) ->
+            if (finishedAgentIds.isNotEmpty()) {
+                // need to retry after some time, because for other agents BUSY state might have not been written completely
+                logger.debug("Waiting for ${configProperties.shutdownChecksIntervalMillis} seconds to repeat `getAgentsAwaitingStop` call for agentId=$agentId")
+                Mono.delay(Duration.ofMillis(configProperties.shutdownChecksIntervalMillis)).then(
+                    agentService.getAgentsAwaitingStop(agentId)
+                )
+            } else {
+                Mono.empty()
+            }
+        }
+            .flatMap { (executionId, finishedAgentIds) ->
                 if (finishedAgentIds.isNotEmpty()) {
                     logger.debug("Agents ids=$finishedAgentIds have completed execution, will make an attempt to terminate them")
                     val areAgentsStopped = dockerService.stopAgents(finishedAgentIds)
                     if (areAgentsStopped) {
                         logger.info("Agents have been stopped, will mark execution id=$executionId and agents $finishedAgentIds as FINISHED")
                         agentService
                             .markAgentsAndExecutionAsFinished(executionId, finishedAgentIds)
-                            .block()
+                    } else {
+                        logger.warn("Agents $finishedAgentIds are not stopped after stop command")
+                        Mono.empty()
                     }
+                } else {
+                    logger.debug("Agents other than $agentId are still running, so won't try to stop them")
+                    Mono.empty()
                 }
             }
-        }
-            .subscribeOn(scheduler)
+            .subscribeOn(agentService.scheduler)
             .subscribe()
     }
 }
diff --git a/save-orchestrator/src/main/kotlin/org/cqfn/save/orchestrator/service/AgentService.kt b/save-orchestrator/src/main/kotlin/org/cqfn/save/orchestrator/service/AgentService.kt
@@ -12,6 +12,7 @@ import org.cqfn.save.execution.ExecutionStatus
 import org.cqfn.save.execution.ExecutionUpdateDto
 import org.cqfn.save.orchestrator.BodilessResponseEntity
 import org.cqfn.save.test.TestBatch
+import org.cqfn.save.test.TestDto
 
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.beans.factory.annotation.Qualifier
@@ -20,6 +21,7 @@ import org.springframework.web.reactive.function.BodyInserters
 import org.springframework.web.reactive.function.client.WebClient
 import org.springframework.web.reactive.function.client.bodyToMono
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
 import reactor.util.Loggers
 
 import java.time.LocalDateTime
@@ -30,6 +32,11 @@ import java.util.logging.Level
  */
 @Service
 class AgentService {
+    /**
+     * A scheduler that executes long-running background tasks
+     */
+    internal val scheduler = Schedulers.boundedElastic().also { it.start() }
+
     @Autowired
     @Qualifier("webClientBackend")
     private lateinit var webClientBackend: WebClient
@@ -46,19 +53,10 @@ class AgentService {
                 .uri("/getTestBatches?agentId=$agentId")
                 .retrieve()
                 .bodyToMono<TestBatch>()
-                .map { batch ->
-                    if (batch.tests.isNotEmpty()) {
-                        // fixme: do we still need suitesToArgs, since we have execFlags in save.toml?
-                        NewJobResponse(
-                            batch.tests,
-                            batch.suitesToArgs.values.first() +
-                                    " --report-type json" +
-                                    " --result-output file" +
-                                    " " + batch.tests.joinToString(separator = " ") { it.filePath }
-                        )
-                    } else {
-                        log.info("Next test batch for agentId=$agentId is empty, setting it to wait")
-                        WaitResponse
+                .map { batch -> batch.toHeartbeatResponse(agentId) }
+                .doOnSuccess {
+                    if (it is NewJobResponse) {
+                        updateAssignedAgent(agentId, it)
                     }
                 }
 
@@ -130,7 +128,7 @@ class AgentService {
     fun markAgentsAndExecutionAsFinished(executionId: Long, finishedAgentIds: List<String>): Mono<BodilessResponseEntity> =
             updateAgentStatusesWithDto(
                 finishedAgentIds.map { agentId ->
-                    AgentStatusDto(LocalDateTime.now(), AgentState.FINISHED, agentId)
+                    AgentStatusDto(LocalDateTime.now(), AgentState.STOPPED_BY_ORCH, agentId)
                 }
             )
                 .then(
@@ -170,6 +168,7 @@ class AgentService {
             .retrieve()
             .bodyToMono<AgentStatusesForExecution>()
             .map { (executionId, agentStatuses) ->
+                log.debug("For executionId=$executionId agent statuses are $agentStatuses")
                 executionId to if (agentStatuses.areIdleOrFinished()) {
                     // We assume, that all agents will eventually have one of these statuses.
                     // Situations when agent gets stuck with a different status and for whatever reason is unable to update
@@ -181,6 +180,50 @@ class AgentService {
             }
     }
 
+    /**
+     * Perform two operations in arbitrary order: assign `agentContainerId` agent to test executions
+     * and mark this agent as BUSY
+     *
+     * @param agentContainerId id of an agent that receives tests
+     * @param newJobResponse a heartbeat response with tests
+     */
+    internal fun updateAssignedAgent(agentContainerId: String, newJobResponse: NewJobResponse) {
+        updateTestExecutionsWithAgent(agentContainerId, newJobResponse.tests).zipWith(
+            updateAgentStatusesWithDto(listOf(
+                AgentStatusDto(LocalDateTime.now(), AgentState.BUSY, agentContainerId)
+            ))
+        )
+            .doOnSuccess {
+                log.debug("Agent $agentContainerId has been set as executor for tests ${newJobResponse.tests} and its status has been set to BUSY")
+            }
+            .subscribeOn(scheduler)
+            .subscribe()
+    }
+
+    private fun updateTestExecutionsWithAgent(agentId: String, testDtos: List<TestDto>): Mono<BodilessResponseEntity> {
+        log.debug("Attempt to update test executions for tests=$testDtos for agent $agentId")
+        return webClientBackend.post()
+            .uri("/testExecution/assignAgent?agentContainerId=$agentId")
+            .bodyValue(testDtos)
+            .retrieve()
+            .toBodilessEntity()
+    }
+
+    private fun TestBatch.toHeartbeatResponse(agentId: String) =
+            if (tests.isNotEmpty()) {
+                // fixme: do we still need suitesToArgs, since we have execFlags in save.toml?
+                NewJobResponse(
+                    tests,
+                    suitesToArgs.values.first() +
+                            " --report-type json" +
+                            " --result-output file" +
+                            " " + tests.joinToString(separator = " ") { it.filePath }
+                )
+            } else {
+                log.info("Next test batch for agentId=$agentId is empty, setting it to wait")
+                WaitResponse
+            }
+
     private fun Collection<AgentStatusDto>.areIdleOrFinished() = all {
         it.state == AgentState.IDLE || it.state == AgentState.FINISHED
     }