Feature/test session heartbeat

java-agent/src/commonMain/kotlin/com/epam/drill/agent/configuration/ParameterDefinitions.kt

@@ -17,10 +17,12 @@ package com.epam.drill.agent.configuration
 
 import com.epam.drill.agent.common.configuration.AgentParameterDefinition
 import com.epam.drill.agent.common.configuration.AgentParameterDefinitionCollection
+import com.epam.drill.agent.common.configuration.AgentParameters
 import com.epam.drill.agent.common.configuration.NullableAgentParameterDefinition
 import com.epam.drill.agent.konform.validation.jsonschema.minLength
 
 object ParameterDefinitions: AgentParameterDefinitionCollection() {
+    val testTracingEnabled: (AgentParameters) -> Boolean = { config -> config[CapabilityParameterDefinitions.TEST_TRACING_ENABLED] }
 
     val API_URL = AgentParameterDefinition.forString(
         name = "apiUrl",
@@ -65,9 +67,10 @@ object ParameterDefinitions: AgentParameterDefinitionCollection() {
     val JS_AGENT_BUILD_VERSION = NullableAgentParameterDefinition.forString(name = "jsAgentBuildVersion").register()
     val JS_AGENT_ID = NullableAgentParameterDefinition.forString(name = "jsAgentId").register()
 
-    val TEST_PROJECT_ID = AgentParameterDefinition.forString(
+    val TEST_PROJECT_ID = NullableAgentParameterDefinition.forString(
         name = "testProjectId",
         description = "Unique arbitrary string identifying your test project. Example: my-test-project",
+        requiredIf = testTracingEnabled,
         validator = {
             identifier()
             minLength(3)

java-agent/src/jvmMain/kotlin/com/epam/drill/agent/test/session/SessionController.kt

@@ -32,10 +32,9 @@ import com.epam.drill.agent.test.sending.TestDefinitionPayload
 import com.epam.drill.agent.test.sending.TestLaunchPayload
 import com.epam.drill.agent.common.lifecycle.AgentShutdownRegistry
 import com.epam.drill.agent.test.execution.TestMethodInfo
-import com.epam.drill.agent.test.sending.AddTestDefinitionsPayload
-import com.epam.drill.agent.test.sending.AddTestLaunchesPayload
-import com.epam.drill.agent.test.session.SessionController.getSessionId
 import com.epam.drill.agent.transport.DataIngestMessageSender
+import com.epam.drill.agent.transport.IntervalTestSessionHeartbeatSender
+import com.epam.drill.agent.transport.directMessageSender
 import mu.KotlinLogging
 import java.time.Instant
 import java.time.ZoneId
@@ -73,6 +72,8 @@ actual object SessionController {
             testInfoSender.stopSendingTests(remainingMs)
         }
 
+        startHeartbeatReporting()
+
         val builds =
             takeIf { Configuration.parameters[ParameterDefinitions.RECOMMENDED_TESTS_TARGET_APP_ID].isNotEmpty() }?.let {
                 SingleSessionBuildPayload(
@@ -85,7 +86,8 @@ actual object SessionController {
             SessionPayload(
                 id = sessionId,
                 groupId = Configuration.parameters[DefaultParameterDefinitions.GROUP_ID],
-                testProjectId = Configuration.parameters[ParameterDefinitions.TEST_PROJECT_ID],
+                testProjectId = Configuration.parameters[ParameterDefinitions.TEST_PROJECT_ID]
+                    ?: error("Test project ID is not set"),
                 testTaskId = Configuration.parameters[ParameterDefinitions.TEST_TASK_ID],
                 startedAt = System.currentTimeMillis().toIsoTimeFormat(),
                 builds = builds
@@ -98,6 +100,22 @@ actual object SessionController {
     private fun isTestTracingEnabled(): Boolean = Configuration.parameters[TEST_TRACING_ENABLED]
     private fun isTestLaunchMetadataSendingEnabled(): Boolean =
         isTestTracingEnabled() && Configuration.parameters[ParameterDefinitions.TEST_TRACING_PER_TEST_LAUNCH_ENABLED]
+
+    private fun startHeartbeatReporting() {
+        if (!Configuration.parameters[ParameterDefinitions.HEARTBEAT_ENABLED]) return
+        val heartbeatSender = IntervalTestSessionHeartbeatSender(
+            sender = directMessageSender(),
+            intervalMs = Configuration.parameters[ParameterDefinitions.HEARTBEAT_INTERVAL],
+            groupId = Configuration.parameters[DefaultParameterDefinitions.GROUP_ID],
+            testProjectId = Configuration.parameters[ParameterDefinitions.TEST_PROJECT_ID]
+                ?: error("Test project ID is not set"),
+            testSessionId = sessionId
+        )
+        heartbeatSender.startSendingHeartbeat()
+        AgentShutdownRegistry.register("session-heartbeat-sender") { remainingMs ->
+            heartbeatSender.stopSendingHeartbeat(remainingMs)
+        }
+    }
 }
 
 private fun List<TestExecutionInfo>.toTestLaunchPayloads(): List<TestLaunchPayload> = map { info ->

java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/TestSessionHeartbeatPayload.kt

@@ -0,0 +1,33 @@
+/**
+ * Copyright 2020 - 2022 EPAM Systems
+ *
+ * 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
+ *
+ * http://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.
+ */
+package com.epam.drill.agent.transport
+
+import com.epam.drill.agent.common.transport.AgentMessage
+import kotlinx.serialization.Serializable
+
+@Serializable
+class TestSessionHeartbeatPayload(
+    val groupId: String,
+    val testProjectId: String,
+    val testSessionId: String,
+    val status: TestSessionStatus,
+) : AgentMessage()
+
+@Serializable
+enum class TestSessionStatus {
+    RUNNING,
+    FINISHED
+}

java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/TestSessionHeartbeatSender.kt

@@ -0,0 +1,91 @@
+/**
+ * Copyright 2020 - 2022 EPAM Systems
+ *
+ * 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
+ *
+ * http://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.
+ */
+package com.epam.drill.agent.transport
+
+import com.epam.drill.agent.common.transport.AgentMessageDestination
+import com.epam.drill.agent.common.transport.AgentMessageSender
+import mu.KotlinLogging
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
+interface TestSessionHeartbeatSender {
+    fun startSendingHeartbeat()
+    fun stopSendingHeartbeat(remainingMs: Long)
+}
+
+/**
+ * Periodically reports the test session heartbeat status to the Backend.
+ *
+ * While the session is active it sends [TestSessionStatus.RUNNING] on a fixed
+ * interval. On graceful shutdown it stops the scheduler and sends a final
+ * [TestSessionStatus.FINISHED] status. All requests are sent synchronously
+ * through the provided [sender] (a DIRECT sender, never the queued pipeline).
+ */
+class IntervalTestSessionHeartbeatSender(
+    private val sender: AgentMessageSender,
+    private val intervalMs: Long,
+    private val groupId: String,
+    private val testProjectId: String,
+    private val testSessionId: String,
+) : TestSessionHeartbeatSender {
+    private val logger = KotlinLogging.logger {}
+    private val destination = AgentMessageDestination("PUT", "sessions/heartbeat")
+    private val scheduledThreadPool = Executors.newSingleThreadScheduledExecutor { runnable ->
+        Thread(runnable, "drill-session-heartbeat-sender").apply { isDaemon = true }
+    }
+
+    override fun startSendingHeartbeat() {
+        scheduledThreadPool.scheduleAtFixedRate(
+            {
+                try {
+                    sendStatus(TestSessionStatus.RUNNING)
+                } catch (t: Throwable) {
+                    logger.error(t) { "Test session heartbeat status sending job failed" }
+                }
+            },
+            intervalMs,
+            intervalMs,
+            TimeUnit.MILLISECONDS
+        )
+        logger.info { "Test session heartbeat status sending job is started." }
+    }
+
+    override fun stopSendingHeartbeat(remainingMs: Long) {
+        scheduledThreadPool.shutdown()
+        if (remainingMs > 0 && !scheduledThreadPool.awaitTermination(remainingMs, TimeUnit.MILLISECONDS)) {
+            logger.warn { "Test session heartbeat sending scheduler did not stop within ${remainingMs}ms; leaving it for JVM exit." }
+        }
+        try {
+            sendStatus(TestSessionStatus.FINISHED)
+        } catch (t: Throwable) {
+            logger.error(t) { "Failed to send SHUTDOWN test session heartbeat status" }
+        }
+        logger.info { "Test session heartbeat status sending job is stopped." }
+    }
+
+    private fun sendStatus(status: TestSessionStatus) {
+        sender.send(
+            destination,
+            TestSessionHeartbeatPayload(
+                groupId = groupId,
+                testProjectId = testProjectId,
+                testSessionId = testSessionId,
+                status = status
+            ),
+            TestSessionHeartbeatPayload.serializer()
+        )
+    }
+}

java-agent/src/jvmTest/kotlin/com/epam/drill/agent/transport/IntervalTestSessionHeartbeatSenderTest.kt

@@ -0,0 +1,77 @@
+/**
+ * Copyright 2020 - 2022 EPAM Systems
+ *
+ * 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
+ *
+ * http://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.
+ */
+package com.epam.drill.agent.transport
+
+import java.util.Collections
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+import com.epam.drill.agent.common.transport.AgentMessageDestination
+import com.epam.drill.agent.common.transport.AgentMessageSender
+import kotlinx.serialization.KSerializer
+
+class IntervalTestSessionHeartbeatSenderTest {
+
+    private class RecordingSender : AgentMessageSender {
+        val sent = Collections.synchronizedList(mutableListOf<Pair<AgentMessageDestination, Any?>>())
+        override fun <T> send(destination: AgentMessageDestination, message: T, serializer: KSerializer<T>) {
+            sent.add(destination to message)
+        }
+    }
+
+    private fun newSender(recorder: RecordingSender, intervalMs: Long = 50L) = IntervalTestSessionHeartbeatSender(
+        sender = recorder,
+        intervalMs = intervalMs,
+        groupId = "someGroupId",
+        testProjectId = "someTestProjectId",
+        testSessionId = "someTestSessionId"
+    )
+
+    @Test
+    fun `should periodically send RUNNING status to sessions heartbeat`() {
+        val recorder = RecordingSender()
+        val sender = newSender(recorder, intervalMs = 50L)
+
+        sender.startSendingHeartbeat()
+        Thread.sleep(200)
+        sender.stopSendingHeartbeat(1000)
+
+        val running = recorder.sent.map { it.second }.filterIsInstance<TestSessionHeartbeatPayload>()
+            .filter { it.status == TestSessionStatus.RUNNING }
+        assertTrue(running.size >= 2, "Expected multiple RUNNING statuses, got ${running.size}")
+
+        val destination = recorder.sent.first().first
+        assertEquals("PUT", destination.type)
+        assertEquals("sessions/heartbeat", destination.target)
+
+        val payload = running.first()
+        assertEquals("someGroupId", payload.groupId)
+        assertEquals("someTestProjectId", payload.testProjectId)
+        assertEquals("someTestSessionId", payload.testSessionId)
+    }
+
+    @Test
+    fun `should send SHUTDOWN status on stop`() {
+        val recorder = RecordingSender()
+        val sender = newSender(recorder, intervalMs = 10_000L)
+
+        sender.startSendingHeartbeat()
+        sender.stopSendingHeartbeat(1000)
+
+        val statuses = recorder.sent.map { it.second }.filterIsInstance<TestSessionHeartbeatPayload>().map { it.status }
+        assertEquals(TestSessionStatus.FINISHED, statuses.last())
+    }
+}