MBS-11465 Runner execution timeout (#1165)

subprojects/common/coroutines-extension/src/main/kotlin/com/avito/coroutines/extensions/ChannelExtensions.kt

@@ -0,0 +1,8 @@
+package com.avito.coroutines.extensions
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+
+@ExperimentalCoroutinesApi
+public val <E> Channel<E>.isClosedForSendAndReceive: Boolean
+    get() = this.isClosedForSend && this.isClosedForReceive

subprojects/test-runner/client/src/main/kotlin/com/avito/runner/config/InstrumentationConfigurationData.kt

@@ -4,6 +4,7 @@ import com.avito.android.runner.devices.model.DeviceType
 import com.avito.instrumentation.reservation.request.Device
 import java.io.File
 import java.io.Serializable
+import java.time.Duration
 
 public data class InstrumentationConfigurationData(
     val name: String,
@@ -12,7 +13,8 @@ public data class InstrumentationConfigurationData(
     val kubernetesNamespace: String,
     val targets: List<TargetConfigurationData>,
     val enableDeviceDebug: Boolean,
-    val timeoutInSeconds: Long,
+    val testRunnerExecutionTimeout: Duration,
+    val instrumentationTaskTimeout: Duration,
     val filter: InstrumentationFilterData,
     val outputFolder: File,
 ) : Serializable {

subprojects/test-runner/client/src/main/kotlin/com/avito/runner/scheduler/TestRunnerFactoryImpl.kt

@@ -80,7 +80,8 @@ internal class TestRunnerFactoryImpl(
             ),
             testSuiteListener = testSuiteListener,
             testRunRequestFactory = testRunnerRequestFactory,
-            targets = targets
+            targets = targets,
+            executionTimeout = params.instrumentationConfiguration.testRunnerExecutionTimeout
         )
     }
 

subprojects/test-runner/client/src/main/kotlin/com/avito/runner/scheduler/report/model/SummaryReport.kt

@@ -11,7 +11,7 @@ internal class SummaryReport(
         reports.count { it.result is TestCaseRequestMatchingReport.Result.Matched }
     }
 
-    val mismatched: Int by lazy {
+    val mismatchedCount: Int by lazy {
         reports.count { it.result is TestCaseRequestMatchingReport.Result.Mismatched }
     }
 

subprojects/test-runner/client/src/main/kotlin/com/avito/runner/scheduler/runner/TestRunnerImpl.kt

@@ -18,7 +18,10 @@ import com.avito.runner.service.DeviceWorkerPool
 import com.avito.test.model.DeviceName
 import com.avito.test.model.TestCase
 import com.avito.time.millisecondsToHumanReadableTime
+import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.withTimeout
+import java.time.Duration
 
 internal class TestRunnerImpl(
     private val scheduler: TestExecutionScheduler,
@@ -30,77 +33,83 @@ internal class TestRunnerImpl(
     private val devicesProvider: DevicesProvider,
     private val testRunRequestFactory: TestRunRequestFactory,
     private val targets: List<TargetConfigurationData>,
+    private val executionTimeout: Duration,
     loggerFactory: LoggerFactory
 ) : TestRunner {
 
     private val logger = loggerFactory.create<TestRunner>()
 
     override suspend fun runTests(tests: List<TestCase>): Result<TestRunnerResult> {
-        return coroutineScope {
-            val startTime = System.currentTimeMillis()
-            testSuiteListener.onTestSuiteStarted()
-            logger.info("Test run started")
-            val deviceWorkerPool: DeviceWorkerPool = devicesProvider.provideFor(
-                reservations = getReservations(tests),
-            )
-            try {
-                deviceWorkerPool.start()
-                reservationWatcher.watch(state.deviceSignals)
-                scheduler.start(
-                    requests = getTestRequests(tests),
+        return withTimeout(executionTimeout.toMillis()) {
+            coroutineScope {
+                val startTime = System.currentTimeMillis()
+                testSuiteListener.onTestSuiteStarted()
+                logger.info("Test run started")
+                val deviceWorkerPool: DeviceWorkerPool = devicesProvider.provideFor(
+                    reservations = getReservations(tests),
                 )
+                try {
+                    deviceWorkerPool.start()
+                    reservationWatcher.watch(state.deviceSignals)
+                    scheduler.start(
+                        requests = getTestRequests(tests),
+                    )
 
-                val expectedResultsCount = tests.count()
+                    val expectedResultsCount = tests.count()
 
-                val gottenResults = mutableListOf<TestRunResult>()
-                for (result in state.results) {
-                    gottenResults.add(result)
-                    val gottenCount = gottenResults.size
+                    val gottenResults = mutableListOf<TestRunResult>()
+                    for (result in state.results) {
+                        gottenResults.add(result)
+                        val gottenCount = gottenResults.size
 
-                    logger.debug(
-                        "Result for test: %s received after %d tries. Progress (%s)".format(
-                            result.request.testCase.name,
-                            result.result.size,
-                            "$gottenCount/$expectedResultsCount"
+                        logger.debug(
+                            "Result for test: %s received after %d tries. Progress (%s)".format(
+                                result.request.testCase.name,
+                                result.result.size,
+                                "$gottenCount/$expectedResultsCount"
+                            )
                         )
-                    )
 
-                    if (gottenCount == expectedResultsCount) {
-                        break
+                        if (gottenCount == expectedResultsCount) {
+                            break
+                        }
                     }
-                }
-                val result = TestRunnerResult(
-                    runs = gottenResults.associate {
-                        it.request.testCase to it.result
-                    }
-                )
+                    val result = TestRunnerResult(
+                        runs = gottenResults.associate {
+                            it.request.testCase to it.result
+                        }
+                    )
 
-                val summaryReport = summaryReportMaker.make(gottenResults, startTime)
-                reporter.report(report = summaryReport)
-                logger.debug(
-                    "Test run finished. The results: " +
-                        "passed = ${summaryReport.successRunsCount}, " +
-                        "failed = ${summaryReport.failedRunsCount}, " +
-                        "ignored = ${summaryReport.ignoredRunsCount}, " +
-                        "took ${summaryReport.durationMilliseconds.millisecondsToHumanReadableTime()}."
-                )
-                logger.debug(
-                    "Matching results: " +
-                        "matched = ${summaryReport.matchedCount}, " +
-                        "mismatched = ${summaryReport.mismatched}, " +
-                        "ignored = ${summaryReport.ignoredCount}."
-                )
+                    val summaryReport = summaryReportMaker.make(gottenResults, startTime)
+                    reporter.report(report = summaryReport)
+                    logger.debug(
+                        "Test run finished. The results: " +
+                            "passed = ${summaryReport.successRunsCount}, " +
+                            "failed = ${summaryReport.failedRunsCount}, " +
+                            "ignored = ${summaryReport.ignoredRunsCount}, " +
+                            "took ${summaryReport.durationMilliseconds.millisecondsToHumanReadableTime()}."
+                    )
+                    logger.debug(
+                        "Matching results: " +
+                            "matched = ${summaryReport.matchedCount}, " +
+                            "mismatched = ${summaryReport.mismatchedCount}, " +
+                            "ignored = ${summaryReport.ignoredCount}."
+                    )
 
-                testSuiteListener.onTestSuiteFinished()
-                logger.info("Test run end successfully")
-                Result.Success(result)
-            } catch (e: Throwable) {
-                logger.critical("Test run end with error", e)
-                Result.Failure(e)
-            } finally {
-                deviceWorkerPool.stop()
-                state.cancel()
-                devicesProvider.releaseDevices()
+                    testSuiteListener.onTestSuiteFinished()
+                    logger.info("Test run end successfully")
+                    Result.Success(result)
+                } catch (e: TimeoutCancellationException) {
+                    logger.critical("Test run end with timeout", e)
+                    Result.Failure(e)
+                } catch (e: Throwable) {
+                    logger.critical("Test run end with error", e)
+                    Result.Failure(e)
+                } finally {
+                    deviceWorkerPool.stop()
+                    state.cancel()
+                    devicesProvider.releaseDevices()
+                }
             }
         }
     }

subprojects/test-runner/client/src/test/kotlin/com/avito/runner/scheduler/runner/RunnerIntegrationTest.kt

@@ -1,9 +1,12 @@
 package com.avito.runner.scheduler.runner
 
 import com.avito.android.Result
+import com.avito.android.runner.devices.DevicesProvider
 import com.avito.android.runner.devices.StubDevicesProvider
 import com.avito.coroutines.extensions.Dispatchers
+import com.avito.coroutines.extensions.isClosedForSendAndReceive
 import com.avito.logger.StubLoggerFactory
+import com.avito.runner.config.InstrumentationConfigurationData
 import com.avito.runner.config.QuotaConfigurationData
 import com.avito.runner.config.TargetConfigurationData
 import com.avito.runner.config.createStubInstance
@@ -36,6 +39,7 @@ import com.avito.test.model.TestName
 import com.avito.time.StubTimeProvider
 import com.google.common.truth.Truth.assertThat
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.delay
@@ -47,6 +51,7 @@ import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
 import org.junit.jupiter.api.io.TempDir
 import java.io.File
+import java.time.Duration
 import java.util.concurrent.TimeUnit
 
 @ExperimentalCoroutinesApi
@@ -546,6 +551,37 @@ internal class RunnerIntegrationTest {
             .isEqualTo("devices channel was closed")
     }
 
+    @Test
+    fun `tests execution timeout - run failed`() = runBlockingTest {
+        val targets = listOf(createTarget())
+        val devicesProvider = createDevicesProvider()
+        val tests = createTests(2)
+        val runner = provideRunner(
+            targets = targets,
+            devicesProvider = devicesProvider,
+            executionTimeout = Duration.ofMinutes(15)
+        )
+        val device = StubDevice(
+            loggerFactory = loggerFactory,
+            model = deviceModel,
+            apiResult = StubActionResult.Success(deviceApi),
+            coordinate = DeviceCoordinate.Local.createStubInstance(),
+            installApplicationResults = List(2) { installApplicationSuccess() }, // main and test apps
+            gettingDeviceStatusResults = List(3) { DeviceStatus.Alive }, // initial and a try for each device
+            clearPackageResults = List(4) { succeedClearPackage() }, // main and test apps for each test
+            runTestsResults = List(2) { testPassed() },
+            testExecutionTime = Duration.ofMinutes(10)
+        )
+
+        devices.send(device)
+        assertThrows<TimeoutCancellationException> {
+            runner.runTests(tests)
+        }
+
+        assertThat(devicesProvider.isReleased).isTrue()
+        state.assertIsCancelled()
+    }
+
     private fun createTarget(quota: QuotaConfigurationData = defaultQuota) =
         TargetConfigurationData.createStubInstance(
             api = deviceApi,
@@ -622,7 +658,7 @@ internal class RunnerIntegrationTest {
         )
     }
 
-    private fun createSuccessfulDevice(testsCount: Int): StubDevice {
+    private fun createSuccessfulDevice(testsCount: Int, testExecutionTime: Duration = Duration.ZERO): StubDevice {
         return StubDevice(
             tag = "StubDevice:normal",
             model = deviceModel,
@@ -642,7 +678,8 @@ internal class RunnerIntegrationTest {
             gettingDeviceStatusResults = List(testsCount + 1) { DeviceStatus.Alive },
             runTestsResults = (0 until testsCount).map {
                 testPassed()
-            }
+            },
+            testExecutionTime = testExecutionTime
         )
     }
 
@@ -654,8 +691,27 @@ internal class RunnerIntegrationTest {
 
     private fun succeedClearPackage() = StubActionResult.Success<Result<Unit>>(Result.Success(Unit))
 
+    private fun createDevicesProvider() = StubDevicesProvider(
+        provider = DeviceWorkerPoolProvider(
+            timeProvider = StubTimeProvider(),
+            loggerFactory = loggerFactory,
+            deviceListener = StubDeviceListener(),
+            intentions = state.intentions,
+            intentionResults = state.intentionResults,
+            deviceSignals = state.deviceSignals,
+            dispatchers = object : Dispatchers {
+                override fun dispatcher() = testCoroutineDispatcher
+            },
+            testRunnerOutputDir = outputDirectory,
+            testListener = NoOpTestListener
+        ),
+        devices = devices
+    )
+
     private fun provideRunner(
-        targets: List<TargetConfigurationData>
+        targets: List<TargetConfigurationData>,
+        devicesProvider: DevicesProvider = createDevicesProvider(),
+        executionTimeout: Duration = InstrumentationConfigurationData.createStubInstance().testRunnerExecutionTimeout
     ): TestRunner {
         val testRunRequestFactory = TestRunRequestFactory(
             application = File("stub"),
@@ -682,24 +738,10 @@ internal class RunnerIntegrationTest {
             summaryReportMaker = SummaryReportMakerImpl(),
             reporter = CompositeReporter(emptyList()),
             testSuiteListener = StubTestMetricsListener,
-            devicesProvider = StubDevicesProvider(
-                provider = DeviceWorkerPoolProvider(
-                    timeProvider = StubTimeProvider(),
-                    loggerFactory = loggerFactory,
-                    deviceListener = StubDeviceListener(),
-                    intentions = state.intentions,
-                    intentionResults = state.intentionResults,
-                    deviceSignals = state.deviceSignals,
-                    dispatchers = object : Dispatchers {
-                        override fun dispatcher() = testCoroutineDispatcher
-                    },
-                    testRunnerOutputDir = outputDirectory,
-                    testListener = NoOpTestListener
-                ),
-                devices = devices
-            ),
+            devicesProvider = devicesProvider,
             testRunRequestFactory = testRunRequestFactory,
-            targets = targets
+            targets = targets,
+            executionTimeout = executionTimeout
         )
     }
 
@@ -717,6 +759,13 @@ internal class RunnerIntegrationTest {
         device = device.getData()
     )
 
+    private fun TestRunnerExecutionState.assertIsCancelled() {
+        assertThat(results.isClosedForSendAndReceive).isTrue()
+        assertThat(intentions.isClosedForSendAndReceive).isTrue()
+        assertThat(intentionResults.isClosedForSendAndReceive).isTrue()
+        assertThat(deviceSignals.isClosedForSendAndReceive).isTrue()
+    }
+
     private fun runBlockingTest(block: suspend TestCoroutineScope.() -> Unit) {
         testCoroutineDispatcher.runBlockingTest(block)
     }

subprojects/test-runner/client/src/testFixtures/kotlin/com/avito/runner/config/StubInstrumentationConfigurationData.kt

@@ -2,6 +2,7 @@ package com.avito.runner.config
 
 import com.avito.runner.scheduler.suite.filter.Filter
 import java.io.File
+import java.time.Duration
 
 public fun InstrumentationConfigurationData.Companion.createStubInstance(
     name: String = "name",
@@ -10,7 +11,8 @@ public fun InstrumentationConfigurationData.Companion.createStubInstance(
     kubernetesNamespace: String = "kubernetesNamespace",
     targets: List<TargetConfigurationData> = emptyList(),
     enableDeviceDebug: Boolean = false,
-    timeoutInSecond: Long = 100,
+    testRunnerExecutionTimeout: Duration = Duration.ofSeconds(100),
+    instrumentationTaskTimeout: Duration = Duration.ofSeconds(120),
     previousRunExcluded: Set<RunStatus> = emptySet(),
     outputFolder: File = File("")
 ): InstrumentationConfigurationData = InstrumentationConfigurationData(
@@ -20,7 +22,8 @@ public fun InstrumentationConfigurationData.Companion.createStubInstance(
     kubernetesNamespace = kubernetesNamespace,
     targets = targets,
     enableDeviceDebug = enableDeviceDebug,
-    timeoutInSeconds = timeoutInSecond,
+    testRunnerExecutionTimeout = testRunnerExecutionTimeout,
+    instrumentationTaskTimeout = instrumentationTaskTimeout,
     filter = InstrumentationFilterData(
         name = "stub",
         fromSource = InstrumentationFilterData.FromSource(

subprojects/test-runner/device-provider/impl/src/testFixtures/kotlin/com/avito/android/runner/devices/StubDevicesProvider.kt

@@ -12,12 +12,15 @@ public class StubDevicesProvider(
     private val devices: ReceiveChannel<Device>
 ) : DevicesProvider {
 
+    public var isReleased: Boolean = false
+        private set
+
     override suspend fun provideFor(
         reservations: Collection<ReservationData>,
     ): DeviceWorkerPool = provider.provide(devices)
 
     override suspend fun releaseDevices() {
-        // do nothing
+        isReleased = true
     }
 
     override suspend fun releaseDevice(coordinate: DeviceCoordinate) {