Feature/queue based data ingestion epmdj 11219

admin-app/src/main/kotlin/com/epam/drill/admin/DrillAdminApplication.kt

@@ -52,8 +52,12 @@ import io.ktor.server.request.*
 import io.ktor.server.resources.*
 import io.ktor.server.response.*
 import io.ktor.server.routing.*
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.runBlocking
 import kotlinx.serialization.json.Json
 import mu.KotlinLogging
+import org.kodein.di.allInstances
 import org.kodein.di.instance
 import org.kodein.di.ktor.closestDI
 import org.kodein.di.ktor.di
@@ -92,6 +96,7 @@ fun Application.module() {
         if (oauth2Enabled) configureOAuthAuthentication(di)
         roleBasedAuthentication()
     }
+    shutdownQueues()
     routing {
         rootRoute()
         swaggerUI(path = "swagger", swaggerFile = "openapi.yml")
@@ -235,4 +240,18 @@ val Application.jsCoverageConverterAddress: String
         .propertyOrNull("jsCoverageConverterAddress")
         ?.getString()
         ?.takeIf { it.isNotBlank() }
-        ?: "http://localhost:8092" // TODO think of default
+        ?: "http://localhost:8092" // TODO think of default
+
+private fun Application.shutdownQueues() {
+    val closableComponents: List<AutoCloseable> by closestDI().allInstances()
+
+    environment.monitor.subscribe(ApplicationStopping) {
+        runBlocking {
+            closableComponents.map {
+                async {
+                    it.close()
+                }
+            }.awaitAll()
+        }
+    }
+}

admin-app/src/main/resources/application.conf

@@ -16,6 +16,20 @@ ktor {
     }
 }
 
+kafka {
+    bootstrapServers = "localhost:9092"
+    bootstrapServers = ${?DRILL_KAFKA_BOOTSTRAP_SERVERS}
+    securityProtocol = ${?DRILL_KAFKA_SECURITY_PROTOCOL}
+    saslMechanism = ${?DRILL_KAFKA_SASL_MECHANISM}
+    saslJaasConfig = ${?DRILL_KAFKA_SASL_JAAS_CONFIG}
+    sslTruststoreLocation = ${?DRILL_KAFKA_SSL_TRUSTSTORE_LOCATION}
+    sslTruststorePassword = ${?DRILL_KAFKA_SSL_TRUSTSTORE_PASSWORD}
+    sslKeystoreLocation = ${?DRILL_KAFKA_SSL_KEYSTORE_LOCATION}
+    sslKeystorePassword = ${?DRILL_KAFKA_SSL_KEYSTORE_PASSWORD}
+    sslKeyPassword = ${?DRILL_KAFKA_SSL_KEY_PASSWORD}
+    sslEndpointIdentificationAlgorithm = ${?DRILL_KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM}
+}
+
 drill {
     database {
         host = ${?DRILL_DB_HOST}
@@ -72,6 +86,29 @@ drill {
         }
     }
     rawData {
+        queue {
+            type = ${?DRILL_RAW_DATA_QUEUE_TYPE}
+            capacity = ${?DRILL_RAW_DATA_QUEUE_CAPACITY}
+            workers = ${?DRILL_RAW_DATA_QUEUE_WORKERS}
+            kafka {
+                topic = "drill-raw-data"
+                topic = ${?DRILL_RAW_DATA_QUEUE_KAFKA_TOPIC}
+                consumerGroupId = "drill-writer"
+                consumerGroupId = ${?DRILL_RAW_DATA_QUEUE_KAFKA_CONSUMER_GROUP_ID}
+                pollTimeoutMs = 500
+                pollTimeoutMs = ${?DRILL_RAW_DATA_QUEUE_KAFKA_POLL_TIMEOUT_MS}
+                shutdownTimeoutMs = 5000
+                shutdownTimeoutMs = ${?DRILL_RAW_DATA_QUEUE_KAFKA_SHUTDOWN_TIMEOUT_MS}
+                producer {
+                    clientId = "drill-admin-raw-data-producer"
+                    clientId = ${?DRILL_RAW_DATA_QUEUE_KAFKA_PRODUCER_CLIENT_ID}
+                }
+                consumer {
+                    clientId = "drill-admin-raw-data-consumer"
+                    clientId = ${?DRILL_RAW_DATA_QUEUE_KAFKA_CONSUMER_CLIENT_ID}
+                }
+            }
+        }
     }
     metrics {
         database {

admin-metrics/src/test/kotlin/com/epam/drill/admin/metrics/BuildsInfoApiTest.kt

@@ -18,6 +18,7 @@ package com.epam.drill.admin.metrics
 import com.epam.drill.admin.metrics.config.MetricsDatabaseConfig
 import com.epam.drill.admin.test.*
 import com.epam.drill.admin.writer.rawdata.config.RawDataWriterDatabaseConfig
+import com.epam.drill.admin.writer.rawdata.route.payload.BuildInfoPayload
 import com.epam.drill.admin.writer.rawdata.route.payload.BuildPayload
 import com.epam.drill.admin.writer.rawdata.route.payload.InstancePayload
 import com.epam.drill.admin.writer.rawdata.table.BuildTable
@@ -36,11 +37,11 @@ class BuildsInfoApiTest : MetricsDatabaseTests({ default, metrics ->
     MetricsDatabaseConfig.init(metrics)
 }) {
     private suspend fun TestDataDsl.initTestData() {
-        client.putBuild(BuildPayload(groupId = testGroup, appId = testApp, buildVersion = "1.0.0", branch = testBranch))
-        client.putBuild(BuildPayload(groupId = testGroup, appId = testApp, buildVersion = "2.0.0", branch = testBranch))
-        client.putBuild(BuildPayload(groupId = testGroup, appId = testApp, buildVersion = "3.0.0", branch = "develop"))
-        client.putBuild(BuildPayload(groupId = testGroup, appId = "app-2", buildVersion = "1.0.0", branch = testBranch))
-        client.putBuild(BuildPayload(groupId = "group-2", appId = testApp, buildVersion = "1.0.0", branch = testBranch))
+        client.putBuildInfo(BuildInfoPayload(groupId = testGroup, appId = testApp, buildVersion = "1.0.0", branch = testBranch))
+        client.putBuildInfo(BuildInfoPayload(groupId = testGroup, appId = testApp, buildVersion = "2.0.0", branch = testBranch))
+        client.putBuildInfo(BuildInfoPayload(groupId = testGroup, appId = testApp, buildVersion = "3.0.0", branch = "develop"))
+        client.putBuildInfo(BuildInfoPayload(groupId = testGroup, appId = "app-2", buildVersion = "1.0.0", branch = testBranch))
+        client.putBuildInfo(BuildInfoPayload(groupId = "group-2", appId = testApp, buildVersion = "1.0.0", branch = testBranch))
     }
 
     private suspend fun TestDataDsl.initEnvironmentData() {

admin-metrics/src/test/kotlin/com/epam/drill/admin/metrics/CoverageTreemapTest.kt

@@ -19,6 +19,7 @@ import com.epam.drill.admin.metrics.config.MetricsDatabaseConfig
 import com.epam.drill.admin.test.MetricsDatabaseTests
 import com.epam.drill.admin.test.withTransaction
 import com.epam.drill.admin.writer.rawdata.config.RawDataWriterDatabaseConfig
+import com.epam.drill.admin.writer.rawdata.route.payload.BuildInfoPayload
 import com.epam.drill.admin.writer.rawdata.route.payload.BuildPayload
 import com.epam.drill.admin.writer.rawdata.route.payload.InstancePayload
 import com.epam.drill.admin.writer.rawdata.route.payload.SingleMethodPayload
@@ -35,7 +36,7 @@ class CoverageTreemapTest : MetricsDatabaseTests({ default, metrics ->
 }) {
     @Test
     fun `given build with no methods no coverage, coverage-treemap should return empty list`() = havingData {
-        client.putBuild(BuildPayload(groupId = testGroup, appId = testApp, buildVersion = "1.0.0", branch = "main"))
+        client.putBuildInfo(BuildInfoPayload(groupId = testGroup, appId = testApp, buildVersion = "1.0.0", branch = "main"))
     }.expectThat {
         client.get("/metrics/coverage-treemap") {
             parameter("buildId", "${testGroup}:${testApp}:1.0.0")

admin-metrics/src/test/kotlin/com/epam/drill/admin/metrics/DataIngestClient.kt

@@ -15,7 +15,6 @@
  */
 package com.epam.drill.admin.metrics
 
-import com.epam.drill.admin.writer.rawdata.config.toBitString
 import com.epam.drill.admin.writer.rawdata.route.payload.*
 import com.jayway.jsonpath.JsonPath
 import io.ktor.client.*
@@ -100,6 +99,12 @@ val TestDetails.definitionId: String
     }
 
 suspend fun HttpClient.putBuild(payload: BuildPayload): HttpResponse {
+    return put("/data-ingest/builds") {
+        setBody(payload)
+    }.assertSuccessStatus()
+}
+
+suspend fun HttpClient.putBuildInfo(payload: BuildInfoPayload): HttpResponse {
     return put("/data-ingest/builds/info") {
         setBody(payload)
     }.assertSuccessStatus()

admin-metrics/src/test/kotlin/com/epam/drill/admin/metrics/RecommendedTestsApiTest.kt

@@ -19,6 +19,7 @@ import com.epam.drill.admin.metrics.config.MetricsDatabaseConfig
 import com.epam.drill.admin.test.MetricsDatabaseTests
 import com.epam.drill.admin.test.withTransaction
 import com.epam.drill.admin.writer.rawdata.config.RawDataWriterDatabaseConfig
+import com.epam.drill.admin.writer.rawdata.route.payload.BuildInfoPayload
 import com.epam.drill.admin.writer.rawdata.route.payload.BuildPayload
 import com.epam.drill.admin.writer.rawdata.table.*
 import io.ktor.client.request.*
@@ -140,8 +141,8 @@ class RecommendedTestsApiTest : MetricsDatabaseTests({ default, metrics ->
     @Test
     fun `given baselineBuildBranches parameter, recommended test service should suggest skipping tests if they are not impacted in baselines from specified branch`() {
         havingData {
-            client.putBuild(BuildPayload(groupId = build1.groupId, appId = build1.appId, buildVersion = build1.buildVersion, branch = "main"))
-            client.putBuild(BuildPayload(groupId = build2.groupId, appId = build2.appId, buildVersion = build2.buildVersion, branch = "feature"))
+            client.putBuildInfo(BuildInfoPayload(groupId = build1.groupId, appId = build1.appId, buildVersion = build1.buildVersion, branch = "main"))
+            client.putBuildInfo(BuildInfoPayload(groupId = build2.groupId, appId = build2.appId, buildVersion = build2.buildVersion, branch = "feature"))
             //build1 on main branch, test1 covers method2
             client.deployInstance(build1, arrayOf(method1, method2))
             client.launchTest(

admin-metrics/src/test/kotlin/com/epam/drill/admin/metrics/TestDataDsl.kt

@@ -26,6 +26,7 @@ import com.epam.drill.admin.metrics.views.TestImpactStatus
 import com.epam.drill.admin.test.StubDrillScheduler
 import com.epam.drill.admin.test.drillApplication
 import com.epam.drill.admin.test.drillClient
+import com.epam.drill.admin.test.waitUntilInBlocking
 import com.epam.drill.admin.writer.rawdata.config.rawDataServicesDIModule
 import com.epam.drill.admin.writer.rawdata.route.dataIngestRoutes
 import com.epam.drill.admin.writer.rawdata.route.payload.InstancePayload
@@ -43,12 +44,14 @@ import org.kodein.di.instance
 import org.kodein.di.singleton
 
 val scheduler = DI.Module("testModule") {
-        bind<DrillScheduler>() with singleton {
-            StubDrillScheduler(UpdateMetricsEtlJob(
+    bind<DrillScheduler>() with singleton {
+        StubDrillScheduler(
+            UpdateMetricsEtlJob(
                 instance(), instance()
-            ))
-        }
+            )
+        )
     }
+}
 
 fun havingData(testsData: suspend TestDataDsl.() -> Unit): HttpClient {
     return runBlocking {
@@ -214,7 +217,9 @@ class TestDataDsl(val client: HttpClient) {
 
 fun HttpClient.expectThat(checks: suspend ExpectationDsl.(HttpClient) -> Unit) {
     val client = this
-    return runBlocking {
+    return waitUntilInBlocking(
+        onAssertionFailed = { refreshMetrics() }
+    ) {
         checks(ExpectationDsl(client), client)
     }
 }

admin-test/build.gradle.kts

@@ -24,6 +24,7 @@ val junitJupiterVersion: String by parent!!.extra
 val postgresSqlVersion: String by parent!!.extra
 val zaxxerHikaricpVersion: String by parent!!.extra
 val quartzVersion: String by parent!!.extra
+val awaitilityVersion: String by parent!!.extra
 
 repositories {
     mavenLocal()
@@ -67,6 +68,7 @@ dependencies {
     implementation("com.zaxxer:HikariCP:$zaxxerHikaricpVersion")
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.5.2")
     implementation("org.postgresql:postgresql:$postgresSqlVersion")
+    implementation("org.awaitility:awaitility:${awaitilityVersion}")
 }
 
 tasks {

admin-test/src/main/kotlin/com/epam/drill/admin/test/AwaitDb.kt

@@ -0,0 +1,60 @@
+/**
+ * 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.admin.test
+
+import kotlinx.coroutines.runBlocking
+import org.awaitility.Awaitility.await
+import org.jetbrains.exposed.sql.transactions.transaction
+import java.time.Duration
+
+private val DEFAULT_DB_WAIT_TIMEOUT: Duration = Duration.ofSeconds(5)
+private val DEFAULT_DB_POLL_INTERVAL: Duration = Duration.ofMillis(100)
+
+fun waitUntilInTransaction(assertion: () -> Unit) {
+    await()
+        .atMost(DEFAULT_DB_WAIT_TIMEOUT)
+        .pollInterval(DEFAULT_DB_POLL_INTERVAL)
+        .untilAsserted {
+            transaction {
+                assertion()
+            }
+        }
+}
+
+fun waitUntilInBlocking(
+    onAssertionFailed: suspend (AssertionError) -> Unit = {},
+    assertion: suspend () -> Unit
+) {
+    await()
+        .atMost(DEFAULT_DB_WAIT_TIMEOUT)
+        .pollInterval(DEFAULT_DB_POLL_INTERVAL)
+        .untilAsserted {
+            runCatching {
+                runBlocking {
+                    assertion()
+                }
+            }.onFailure { e ->
+                if (e is AssertionError) {
+                    runCatching {
+                        runBlocking {
+                            onAssertionFailed(e)
+                        }
+                    }
+                }
+                throw e
+            }
+        }
+}

admin-test/src/main/kotlin/com/epam/drill/admin/test/TestUtils.kt

@@ -16,7 +16,6 @@
 package com.epam.drill.admin.test
 
 import io.ktor.serialization.kotlinx.json.*
-import io.ktor.server.application.install
 import io.ktor.server.plugins.contentnegotiation.*
 import io.ktor.server.plugins.statuspages.StatusPages
 import io.ktor.server.resources.*
@@ -29,8 +28,13 @@ import org.kodein.di.DI
 import org.kodein.di.ktor.di
 import kotlin.test.assertEquals
 import com.epam.drill.admin.common.route.commonStatusPages
+import io.ktor.server.application.ApplicationStopping
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
 import org.jetbrains.exposed.sql.Database
-import javax.sql.DataSource
+import org.kodein.di.allInstances
+import org.kodein.di.ktor.closestDI
+import kotlin.getValue
 
 
 fun withRollback(test: suspend () -> Unit) {
@@ -68,6 +72,16 @@ fun drillApplication(
         di {
             diModules.forEach { import(it) }
         }
+        environment.monitor.subscribe(ApplicationStopping) {
+            val closableComponents: List<AutoCloseable> by closestDI().allInstances()
+            runBlocking {
+                closableComponents.map {
+                    async {
+                        it.close()
+                    }
+                }.awaitAll()
+            }
+        }
     }
     routing {
         routes()

admin-writer/build.gradle.kts

@@ -23,6 +23,8 @@ val testContainersVersion: String by parent!!.extra
 val postgresSqlVersion: String by parent!!.extra
 val zaxxerHikaricpVersion: String by parent!!.extra
 val logbackVersion: String by parent!!.extra
+val kafkaClientsVersion: String by parent!!.extra
+val junitJupiterVersion: String by parent!!.extra
 
 repositories {
     mavenLocal()
@@ -62,6 +64,8 @@ dependencies {
     api("org.flywaydb:flyway-core:$flywaydbVersion")
     compileOnly("org.postgresql:postgresql:$postgresSqlVersion")
 
+    implementation("org.apache.kafka:kafka-clients:$kafkaClientsVersion")
+
     implementation("io.ktor:ktor-client-core:$ktorVersion")
     implementation("io.ktor:ktor-client-apache:$ktorVersion")
     implementation("io.ktor:ktor-client-json:$ktorVersion")
@@ -74,6 +78,11 @@ dependencies {
     testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.5.2")
     testImplementation("io.ktor:ktor-server-test-host:$ktorVersion")
     testImplementation("ch.qos.logback:logback-classic:$logbackVersion")
+    testImplementation("org.mockito.kotlin:mockito-kotlin:$mockitoKotlinVersion")
+    testImplementation("org.testcontainers:testcontainers:$testContainersVersion")
+    testImplementation("org.testcontainers:junit-jupiter:$testContainersVersion")
+    testImplementation("org.testcontainers:kafka:$testContainersVersion")
+    testImplementation("org.junit.jupiter:junit-jupiter:$junitJupiterVersion")
     testImplementation(project(":admin-test"))
 }