diff --git a/java-agent/drill.properties b/java-agent/drill.properties index b370a2024..b6dacfdb3 100644 --- a/java-agent/drill.properties +++ b/java-agent/drill.properties @@ -4,3 +4,6 @@ coverageCollectionEnabled=true classScanningEnabled=true contextPropagationEnabled=true testTracingEnabled=false +#Heartbeat status reporting +heartbeatEnabled=true +heartbeatInterval=30000 diff --git a/java-agent/src/commonMain/kotlin/com/epam/drill/agent/configuration/ParameterDefinitions.kt b/java-agent/src/commonMain/kotlin/com/epam/drill/agent/configuration/ParameterDefinitions.kt index 36b17a1bb..2e7bdce77 100644 --- a/java-agent/src/commonMain/kotlin/com/epam/drill/agent/configuration/ParameterDefinitions.kt +++ b/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() } diff --git a/java-agent/src/commonMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt b/java-agent/src/commonMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt index 96d924cdb..9092f5c18 100644 --- a/java-agent/src/commonMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt +++ b/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() } diff --git a/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/Agent.kt b/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/Agent.kt index 230c7687b..988f6b569 100644 --- a/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/Agent.kt +++ b/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() } diff --git a/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/test/prioritization/RecommendedTestsReceiver.kt b/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/test/prioritization/RecommendedTestsReceiver.kt index 3aaa84f0a..b6ab3a97f 100644 --- a/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/test/prioritization/RecommendedTestsReceiver.kt +++ b/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 { diff --git a/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/AgentHeartbeatPayload.kt b/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/AgentHeartbeatPayload.kt new file mode 100644 index 000000000..5e64b9350 --- /dev/null +++ b/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 +} diff --git a/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/AgentHeartbeatSender.kt b/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/AgentHeartbeatSender.kt new file mode 100644 index 000000000..c202112c1 --- /dev/null +++ b/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() + ) + } +} diff --git a/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/DataIngestMessageSender.kt b/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/DataIngestMessageSender.kt index 1279af9c4..bcae567d1 100644 --- a/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/DataIngestMessageSender.kt +++ b/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], diff --git a/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt b/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt index f5f7b4fc0..3f1df0587 100644 --- a/java-agent/src/jvmMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt +++ b/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"), diff --git a/java-agent/src/jvmTest/kotlin/com/epam/drill/agent/transport/IntervalAgentHeartbeatSenderTest.kt b/java-agent/src/jvmTest/kotlin/com/epam/drill/agent/transport/IntervalAgentHeartbeatSenderTest.kt new file mode 100644 index 000000000..575cbe893 --- /dev/null +++ b/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>()) + override fun send(destination: AgentMessageDestination, message: T, serializer: KSerializer) { + 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() + .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().map { it.status } + assertEquals(AgentHeartbeatStatus.SHUTDOWN, statuses.last()) + } +} diff --git a/java-agent/src/nativeMain/kotlin/com/epam/drill/agent/Agent.kt b/java-agent/src/nativeMain/kotlin/com/epam/drill/agent/Agent.kt index b72d53143..6ba187410 100644 --- a/java-agent/src/nativeMain/kotlin/com/epam/drill/agent/Agent.kt +++ b/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() diff --git a/java-agent/src/nativeMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt b/java-agent/src/nativeMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt index f3bab5c27..47279ee77 100644 --- a/java-agent/src/nativeMain/kotlin/com/epam/drill/agent/transport/JvmModuleMessageSender.kt +++ b/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) + }