E2E Test Influxdb2 (#340)

* extracting TagFieldProvider

* adding proper names

* rename package name tests on infludb publisher

* testing new influxDB docker container

* ktlint format

library/core/talaiot-test-utils/src/main/java/org/testcontainers/influxdb2/InfluxDBContainerV2.kt

@@ -0,0 +1,159 @@
+package org.testcontainers.influxdb2
+
+import org.testcontainers.containers.GenericContainer
+import org.testcontainers.containers.wait.strategy.Wait
+import org.testcontainers.containers.wait.strategy.WaitAllStrategy
+import org.testcontainers.utility.DockerImageName
+import java.util.Optional
+
+/**
+ * Following code was extracted from https://github.com/testcontainers/testcontainers-java/pull/3669
+ * The PR is not merged and implements the InfluxDb 2.x testcontainer component
+
+ * Refer to
+ * [ the official InfluxDB 2.x container repository](https://hub.docker.com/_/influxdb)
+ * on docker hub for detailed documentation and newest tags.
+ */
+open class InfluxDBContainerV2(imageName: DockerImageName) : GenericContainer<InfluxDBContainerV2?>(imageName) {
+    private var username = "test-user"
+
+    private var password = "test-password"
+
+    private var bucket = "test-bucket"
+
+    private var organization = "test-org"
+
+    private var retention = Optional.empty<String>()
+
+    private var adminToken = "test-token"
+
+    constructor(imageName: String?) : this(DockerImageName.parse(imageName)) {}
+
+    /**
+     *
+     *
+     * The InfluxDB image contains some extra functionality to automatically bootstrap the system. This functionality is
+     * enabled by setting the DOCKER_INFLUXDB_INIT_MODE environment variable to the value "setup" when running the
+     * container. Additional environment variables are used to configure the setup logic:
+     *
+     *  *  DOCKER_INFLUXDB_INIT_USERNAME: The username to set for the system's initial super-user (Required).
+     *  *  DOCKER_INFLUXDB_INIT_PASSWORD: The password to set for the system's initial super-user (Required).
+     *  *  DOCKER_INFLUXDB_INIT_ORG: The name to set for the system's initial organization (Required).
+     *  *  DOCKER_INFLUXDB_INIT_BUCKET: The name to set for the system's initial bucket (Required).
+     *  *  DOCKER_INFLUXDB_INIT_RETENTION: The duration the system's initial bucket should retain data. If not set,
+     * the initial bucket will retain data forever.
+     *  *  DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: The authentication token to associate with the system's initial
+     * super-user. If not set, a token will be auto-generated by the system.
+     *
+     * <br></br>
+     * See
+     * [ full documentation](https://hub.docker.com/_/influxdb)
+     */
+    override fun configure() {
+        addEnv("DOCKER_INFLUXDB_INIT_MODE", "setup")
+        addEnv("DOCKER_INFLUXDB_INIT_USERNAME", username)
+        addEnv("DOCKER_INFLUXDB_INIT_PASSWORD", password)
+        addEnv("DOCKER_INFLUXDB_INIT_ORG", organization)
+        addEnv("DOCKER_INFLUXDB_INIT_BUCKET", bucket)
+        retention.ifPresent { ret: String? ->
+            addEnv(
+                "DOCKER_INFLUXDB_INIT_RETENTION",
+                ret
+            )
+        }
+        addEnv("DOCKER_INFLUXDB_INIT_ADMIN_TOKEN", adminToken)
+    }
+
+    /**
+     * Set env variable `DOCKER_INFLUXDB_INIT_USERNAME`.
+     *
+     * @param username The name of a user to be created with no privileges. If `INFLUXDB_BUCKET` is set, this user will
+     * be granted read and write permissions for that database.
+     * @return a reference to this container instance
+     */
+    fun withUsername(username: String): InfluxDBContainerV2 {
+        this.username = username
+        return this
+    }
+
+    /**
+     * Set env variable `DOCKER_INFLUXDB_INIT_PASSWORD`.
+     *
+     * @param password The password for the user configured with `INFLUXDB_USER`. If this is unset, a random password is
+     * generated and printed to standard out.
+     * @return a reference to this container instance
+     */
+    fun withPassword(password: String): InfluxDBContainerV2 {
+        this.password = password
+        return this
+    }
+
+    /**
+     * Set env variable `DOCKER_INFLUXDB_INIT_ORG`.
+     *
+     * @param organization The organization for the initial setup of influxDB.
+     * @return a reference to this container instance
+     */
+    fun withOrganization(organization: String): InfluxDBContainerV2 {
+        this.organization = organization
+        return this
+    }
+
+    /**
+     * Set env variable `DOCKER_INFLUXDB_INIT_BUCKET`.
+     *
+     * @param bucket Automatically initializes a bucket with the name of this environment variable.
+     * @return a reference to this container instance
+     */
+    fun withBucket(bucket: String): InfluxDBContainerV2 {
+        this.bucket = bucket
+        return this
+    }
+
+    /**
+     * Set env variable `DOCKER_INFLUXDB_INIT_RETENTION`.
+     *
+     * @param retention Duration bucket will retain data (0 is infinite, default is 0).
+     * @return a reference to this container instance
+     */
+    fun withRetention(retention: String): InfluxDBContainerV2 {
+        this.retention = Optional.of(retention)
+        return this
+    }
+
+    /**
+     * Set env variable `DOCKER_INFLUXDB_INIT_ADMIN_TOKEN`.
+     *
+     * @param adminToken Authentication token to associate with the admin user.
+     * @return a reference to this container instance
+     */
+    fun withAdminToken(adminToken: String): InfluxDBContainerV2 {
+        this.adminToken = adminToken
+        return this
+    }
+
+    /**
+     * @return a url to influxDb
+     */
+    val url: String
+        get() = "http://" + this.host + ":" + getMappedPort(INFLUXDB_PORT)
+
+    companion object {
+        private const val INFLUXDB_PORT = 8086
+        private val DEFAULT_IMAGE_NAME = DockerImageName.parse("influxdb")
+        private const val NO_CONTENT_STATUS_CODE = 204
+    }
+
+    init {
+        imageName.assertCompatibleWith(DEFAULT_IMAGE_NAME)
+        waitStrategy = WaitAllStrategy()
+            .withStrategy(
+                Wait
+                    .forHttp("/ping")
+                    .withBasicCredentials(username, password)
+                    .forStatusCode(NO_CONTENT_STATUS_CODE)
+            )
+            .withStrategy(Wait.forListeningPort())
+        addExposedPort(INFLUXDB_PORT)
+    }
+}

library/core/talaiot-test-utils/src/main/java/org/testcontainers/influxdb2/InfluxDb2Container.kt

@@ -0,0 +1,57 @@
+package org.testcontainers.pushgateway
+
+import org.testcontainers.containers.GenericContainer
+import org.testcontainers.containers.wait.strategy.Wait
+import org.testcontainers.containers.wait.strategy.WaitAllStrategy
+import org.testcontainers.utility.Base58
+import java.net.InetSocketAddress
+
+/**
+ * Custom TestContainer to support E2E tests for PushGateway
+ */
+open class InfluxDb2Container :
+    GenericContainer<PushGatewayContainer>("$INFLUXDB2_DEFAULT_IMAGE:$INFLUXDB2_DEFAULT_VERSION") {
+    init {
+
+        logger().info("Starting an  container using [{}]", dockerImageName)
+        withNetworkAliases("influxdb2-" + Base58.randomString(6))
+        withEnv("discovery.type", "single-node")
+        withEnv("bucket", "aaaaaa")
+        withEnv("org", "alo")
+        addExposedPorts(
+            INFLUXDB2_DEFAULT_PORT,
+            INFLUXDB2_DEFAULT_TCP_PORT
+        )
+        setWaitStrategy(
+            WaitAllStrategy()
+                .withStrategy(
+                    Wait.forListeningPort()
+                )
+        )
+    }
+
+    val httpHostAddress: String
+        get() = containerIpAddress + ":" + getMappedPort(INFLUXDB2_DEFAULT_PORT)
+
+    fun getTcpHost(): InetSocketAddress {
+        return InetSocketAddress(containerIpAddress, getMappedPort(INFLUXDB2_DEFAULT_TCP_PORT)!!)
+    }
+
+    companion object {
+
+        /**
+         * Pushgateway Default HTTP port
+         */
+        private val INFLUXDB2_DEFAULT_PORT = 8086
+
+        /**
+         * Pushgateway Docker base URL
+         */
+        private val INFLUXDB2_DEFAULT_IMAGE = "influxdb"
+        private val INFLUXDB2_DEFAULT_TCP_PORT = 8086
+        /**
+         * Pushgateway Default version
+         */
+        protected val INFLUXDB2_DEFAULT_VERSION = "latest"
+    }
+}

library/core/talaiot-test-utils/src/main/java/org/testcontainers/influxdb2/KInfluxDb2Container.kt

@@ -0,0 +1,2 @@
+package org.testcontainers.influxdb2
+class KInfluxDb2Container : InfluxDBContainerV2("influxdb")

library/plugins/influxdb/influxdb2-publisher/src/main/kotlin/io/github/cdsap/talaiot/publisher/influxdb2/InfluxDb2Publisher.kt

@@ -37,8 +37,8 @@ class InfluxDb2Publisher(
             influxDbPublisherConfiguration.buildMetricName.isEmpty()
         ) {
             logTracker.error(
-                "InfluxDbPublisher not executed. Configuration requires url, token, org, bucket taskMetricName and buildMetricName: \n" +
-                    "influxDbPublisher {\n" +
+                "InfluxDb2Publisher not executed. Configuration requires url, token, org, bucket taskMetricName and buildMetricName: \n" +
+                    "influxDb2Publisher {\n" +
                     "            url = \"http://localhost:8086\"\n" +
                     "            token = \"<token>\"\n" +
                     "            bucket = \"myBucketName\"\n" +

library/plugins/influxdb/influxdb2-publisher/src/test/kotlin/io/github/cdsap/talaiot/publisher/influxdb2/InfluxDb2PublisherTest.kt

@@ -0,0 +1,109 @@
+package io.github.cdsap.talaiot.publisher.influxdb2
+
+import com.influxdb.client.InfluxDBClientFactory
+import io.github.cdsap.talaiot.entities.CustomProperties
+import io.github.cdsap.talaiot.entities.Environment
+import io.github.cdsap.talaiot.entities.ExecutionReport
+import io.github.cdsap.talaiot.entities.TaskLength
+import io.github.cdsap.talaiot.entities.TaskMessageState
+import io.github.cdsap.talaiot.logger.TestLogTrackerRecorder
+import io.kotlintest.Spec
+import io.kotlintest.specs.BehaviorSpec
+import org.testcontainers.influxdb2.KInfluxDb2Container
+
+class InfluxDb2PublisherTest : BehaviorSpec() {
+
+    val container = KInfluxDb2Container()
+
+    override fun beforeSpec(spec: Spec) {
+        super.beforeSpec(spec)
+        container.start()
+    }
+
+    override fun afterSpec(spec: Spec) {
+        super.afterSpec(spec)
+        container.stop()
+    }
+
+    init {
+        given("InfluxDb2Publisher instance") {
+            val logger = TestLogTrackerRecorder
+
+            `when`("Simple configuration is provided") {
+                val influxDbConfiguration = InfluxDb2PublisherConfiguration().apply {
+                    url = container.url
+                    bucket = "test-bucket"
+                    token = "test-token"
+                    org = "test-org"
+                    taskMetricName = "task"
+                    buildMetricName = "build"
+                }
+                val influxDbPublisher = InfluxDb2Publisher(
+                    influxDbConfiguration, logger
+                )
+                val influxDBClient = InfluxDBClientFactory.create(container.url, "test-token".toCharArray(), "test-org")
+                influxDbPublisher.publish(executionReport())
+
+                then("build and task data is stored correctly on the InfluxDb2 Instance") {
+                    val queryApi = influxDBClient.queryApi
+                    val fluxBuild =
+                        "from(bucket:\"test-bucket\") |> range(start: 0) |> filter(fn: (r) => r._measurement == \"build\")"
+                    val tablesBuild = queryApi.query(fluxBuild)
+
+                    assert(tablesBuild.filter { it.records.filter { it.field == "cpuCount" && it.value.toString() == "12" }.size == 1 }.size == 1)
+                    assert(tablesBuild.filter { it.records.filter { it.field == "metric3" && it.value == "value3" }.size == 1 }.size == 1)
+                    assert(tablesBuild.filter { it.records.filter { it.field == "duration" && it.value.toString() == "10" }.size == 1 }.size == 1)
+
+                    val fluxTask =
+                        "from(bucket:\"test-bucket\") |> range(start: 0) |> filter(fn: (r) => r._measurement == \"task\")"
+
+                    val tablesTask = queryApi.query(fluxTask)
+
+                    assert(
+                        tablesTask[0].records.filter {
+                            it.values.filter {
+                                it.key == "task" && it.value == ":assemble"
+                            }.size == 1
+                        }.size == 1
+                    )
+                    assert(
+                        tablesTask[0].records.filter {
+                            it.values.filter {
+                                it.key == "metric1" && it.value == "value1"
+                            }.size == 1
+                        }.size == 1
+                    )
+                    influxDBClient.close()
+                }
+            }
+        }
+    }
+
+    private fun executionReport(): ExecutionReport {
+        return ExecutionReport(
+            requestedTasks = "assemble",
+            durationMs = "10",
+            success = true,
+            environment = Environment(
+                cpuCount = "12", maxWorkers = "4"
+            ),
+            customProperties = CustomProperties(
+                taskProperties = mutableMapOf(
+                    "metric1" to "value1",
+                    "metric2" to "value2"
+                ),
+                buildProperties = mutableMapOf(
+                    "metric3" to "value3",
+                    "metric4" to "value4"
+                )
+            ),
+
+            tasks = listOf(
+                TaskLength(
+                    1, "assemble", ":assemble", TaskMessageState.EXECUTED, false,
+                    "app", emptyList()
+                )
+            )
+        )
+    }
+}