Feature/agent health status epmdj 11240

java-agent/drill.properties

@@ -4,3 +4,6 @@ coverageCollectionEnabled=true
 classScanningEnabled=true
 contextPropagationEnabled=true
 testTracingEnabled=false
+#Heartbeat status reporting
+heartbeatEnabled=true
+heartbeatInterval=30000

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

@@ -79,4 +79,15 @@ object ParameterDefinitions: AgentParameterDefinitionCollection() {
 
     val TEST_TRACING_PER_SESSION_ENABLED = AgentParameterDefinition.forBoolean(name = "testTracingPerTestSessionEnabled", defaultValue = true).register()
     val TEST_TRACING_PER_TEST_LAUNCH_ENABLED = AgentParameterDefinition.forBoolean(name = "testTracingPerTestLaunchEnabled", defaultValue = true).register()
+
+    val HEARTBEAT_ENABLED = AgentParameterDefinition.forBoolean(
+        name = "heartbeatEnabled",
+        description = "Enables periodic agent heartbeat status reporting to the Backend.",
+        defaultValue = true
+    ).register()
+    val HEARTBEAT_INTERVAL = AgentParameterDefinition.forLong(
+        name = "heartbeatInterval",
+        description = "Interval in milliseconds between agent heartbeat status requests.",
+        defaultValue = 30_000L
+    ).register()
 }

java-agent/src/commonMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt

@@ -17,4 +17,5 @@ package com.epam.drill.agent.transport
 
 expect object JvmModuleMessageSender {
     fun sendAgentMetadata()
+    fun startHeartbeatReporting()
 }

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

@@ -58,6 +58,7 @@ fun premain(agentArgs: String?, inst: Instrumentation) {
         inst.addTransformer(DrillClassFileTransformer, true)
         if (isClassScanningEnabled() || isCoverageCollectionEnabled()) {
             JvmModuleMessageSender.sendAgentMetadata()
+            JvmModuleMessageSender.startHeartbeatReporting()
             JvmModuleLoader.loadJvmModule(Test2Code::class.java.name).load()
         }
 

java-agent/src/jvmMain/kotlin/com/epam/drill/agent/test/prioritization/RecommendedTestsReceiver.kt

@@ -98,10 +98,12 @@ class RecommendedTestsReceiverImpl(
         logger.debug { "Loading tests to skip from file: $filePath" }
         return runCatching {
             val content = File(filePath).readText()
-            val entries = fileJson.decodeFromString(ListSerializer(TestDefinitionResponse.serializer()), content)
-            entries.map { it.toTestMethodInfo() }.also {
-                logger.info { "Loaded ${it.size} tests to skip from file: $filePath" }
-            }
+            fileJson.decodeFromString(ListSerializer(TestDefinitionResponse.serializer()), content)
+                .filter { it.impactStatus == "NOT_IMPACTED" }
+                .map { it.toTestMethodInfo() }
+                .also {
+                    logger.info { "Loaded ${it.size} tests to skip from file: $filePath" }
+                }
         }.onFailure {
             logger.warn { "Unable to load tests to skip from file '$filePath'. Error message: $it" }
         }.getOrElse {

java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/AgentHeartbeatPayload.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 AgentHeartbeatPayload(
+    val groupId: String,
+    val appId: String,
+    val instanceId: String,
+    val status: AgentHeartbeatStatus,
+) : AgentMessage()
+
+@Serializable
+enum class AgentHeartbeatStatus {
+    RUNNING,
+    SHUTDOWN
+}

java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/AgentHeartbeatSender.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 AgentHeartbeatSender {
+    fun startSendingHeartbeat()
+    fun stopSendingHeartbeat(remainingMs: Long)
+}
+
+/**
+ * Periodically reports the agent heartbeat status to the Backend.
+ *
+ * While the agent is running it sends [AgentHeartbeatStatus.RUNNING] on a fixed
+ * interval. On graceful shutdown it stops the scheduler and sends a final
+ * [AgentHeartbeatStatus.SHUTDOWN] status. All requests are sent synchronously
+ * through the provided [sender] (a DIRECT sender, never the queued pipeline).
+ */
+class IntervalAgentHeartbeatSender(
+    private val sender: AgentMessageSender,
+    private val intervalMs: Long,
+    private val groupId: String,
+    private val appId: String,
+    private val instanceId: String,
+) : AgentHeartbeatSender {
+    private val logger = KotlinLogging.logger {}
+    private val destination = AgentMessageDestination("PUT", "instances/heartbeat")
+    private val scheduledThreadPool = Executors.newSingleThreadScheduledExecutor { runnable ->
+        Thread(runnable, "drill-heartbeat-sender").apply { isDaemon = true }
+    }
+
+    override fun startSendingHeartbeat() {
+        scheduledThreadPool.scheduleAtFixedRate(
+            {
+                try {
+                    sendStatus(AgentHeartbeatStatus.RUNNING)
+                } catch (t: Throwable) {
+                    logger.error(t) { "Heartbeat status sending job failed" }
+                }
+            },
+            intervalMs,
+            intervalMs,
+            TimeUnit.MILLISECONDS
+        )
+        logger.info { "Heartbeat status sending job is started." }
+    }
+
+    override fun stopSendingHeartbeat(remainingMs: Long) {
+        scheduledThreadPool.shutdown()
+        if (remainingMs > 0 && !scheduledThreadPool.awaitTermination(remainingMs, TimeUnit.MILLISECONDS)) {
+            logger.warn { "Heartbeat sending scheduler did not stop within ${remainingMs}ms; leaving it for JVM exit." }
+        }
+        try {
+            sendStatus(AgentHeartbeatStatus.SHUTDOWN)
+        } catch (t: Throwable) {
+            logger.error(t) { "Failed to send SHUTDOWN heartbeat status" }
+        }
+        logger.info { "Heartbeat status sending job is stopped." }
+    }
+
+    private fun sendStatus(status: AgentHeartbeatStatus) {
+        sender.send(
+            destination,
+            AgentHeartbeatPayload(
+                groupId = groupId,
+                appId = appId,
+                instanceId = instanceId,
+                status = status
+            ),
+            AgentHeartbeatPayload.serializer()
+        )
+    }
+}

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

@@ -76,6 +76,13 @@ fun messageSender(): AgentMessageSender {
     }
 }
 
+fun directMessageSender(): AgentMessageSender =
+    SimpleAgentMessageSender(
+        agentMessageTransport(),
+        JsonAgentMessageSerializer(),
+        HttpAgentMessageDestinationMapper()
+    )
+
 fun agentMessageTransport(): HttpAgentMessageTransport {
     val transport = HttpAgentMessageTransport(
         serverAddress = Configuration.parameters[ParameterDefinitions.API_URL],

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

@@ -16,10 +16,23 @@
 package com.epam.drill.agent.transport
 
 import com.epam.drill.agent.configuration.Configuration
+import com.epam.drill.agent.configuration.ParameterDefinitions
+import com.epam.drill.agent.common.lifecycle.AgentShutdownRegistry
 import com.epam.drill.agent.common.transport.AgentMessageDestination
 import com.epam.drill.agent.common.transport.AgentMessageSender
 
 actual object JvmModuleMessageSender : AgentMessageSender by DataIngestMessageSender {
+
+    private val heartbeatSender: AgentHeartbeatSender by lazy {
+        IntervalAgentHeartbeatSender(
+            sender = directMessageSender(),
+            intervalMs = Configuration.parameters[ParameterDefinitions.HEARTBEAT_INTERVAL],
+            groupId = Configuration.agentMetadata.groupId,
+            appId = Configuration.agentMetadata.appId,
+            instanceId = Configuration.agentMetadata.instanceId
+        )
+    }
+
     actual fun sendAgentMetadata() {
         send(
             AgentMessageDestination("PUT", "instances"),
@@ -35,6 +48,14 @@ actual object JvmModuleMessageSender : AgentMessageSender by DataIngestMessageSe
         )
     }
 
+    actual fun startHeartbeatReporting() {
+        if (!Configuration.parameters[ParameterDefinitions.HEARTBEAT_ENABLED]) return
+        heartbeatSender.startSendingHeartbeat()
+        AgentShutdownRegistry.register("instance-heartbeat-sender") { remainingMs ->
+            heartbeatSender.stopSendingHeartbeat(remainingMs)
+        }
+    }
+
     fun sendBuildMetadata() {
         send(
             AgentMessageDestination("PUT", "builds"),

java-agent/src/jvmTest/kotlin/com/epam/drill/agent/transport/IntervalAgentHeartbeatSenderTest.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 IntervalAgentHeartbeatSenderTest {
+
+    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) = IntervalAgentHeartbeatSender(
+        sender = recorder,
+        intervalMs = intervalMs,
+        groupId = "someGroupId",
+        appId = "someAppId",
+        instanceId = "someInstanceId"
+    )
+
+    @Test
+    fun `should periodically send RUNNING status to instances 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<AgentHeartbeatPayload>()
+            .filter { it.status == AgentHeartbeatStatus.RUNNING }
+        assertTrue(running.size >= 2, "Expected multiple RUNNING statuses, got ${running.size}")
+
+        val destination = recorder.sent.first().first
+        assertEquals("PUT", destination.type)
+        assertEquals("instances/heartbeat", destination.target)
+
+        val payload = running.first()
+        assertEquals("someGroupId", payload.groupId)
+        assertEquals("someAppId", payload.appId)
+        assertEquals("someInstanceId", payload.instanceId)
+    }
+
+    @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<AgentHeartbeatPayload>().map { it.status }
+        assertEquals(AgentHeartbeatStatus.SHUTDOWN, statuses.last())
+    }
+}

java-agent/src/nativeMain/kotlin/com/epam/drill/agent/Agent.kt

@@ -87,6 +87,7 @@ object Agent {
         loadJvmModule("com.epam.drill.agent.test2code.Test2Code")
         if (isClassScanningEnabled() || isCoverageCollectionEnabled()) {
             JvmModuleMessageSender.sendAgentMetadata()
+            JvmModuleMessageSender.startHeartbeatReporting()
         }
         SessionController.startSession()
         AgentShutdownCoordinator.install()

java-agent/src/nativeMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt

@@ -22,4 +22,7 @@ actual object JvmModuleMessageSender {
     actual fun sendAgentMetadata(): Unit =
         callObjectVoidMethod(JvmModuleMessageSender::class, JvmModuleMessageSender::sendAgentMetadata)
 
+    actual fun startHeartbeatReporting(): Unit =
+        callObjectVoidMethod(JvmModuleMessageSender::class, JvmModuleMessageSender::startHeartbeatReporting)
+
 }