[feature] add e2e code (#2811)

Co-authored-by: aias00 <[email protected]>
Co-authored-by: tomsun28 <[email protected]>

.github/workflows/backend-build-test-reuse.yml

@@ -35,7 +35,7 @@ jobs:
     - uses: ./script/ci/github-actions/setup-deps
 
     - name: Build with Maven
-      run: mvn clean -B package -Prelease --file pom.xml
+      run: mvn clean -B package -Prelease -Dmaven.test.skip=false --file pom.xml
 
     - name: Upload coverage reports to Codecov
       uses: codecov/[email protected]

hertzbeat-e2e/hertzbeat-collector-kafka-e2e/pom.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.hertzbeat</groupId>
+        <artifactId>hertzbeat-e2e</artifactId>
+        <version>2.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>hertzbeat-collector-kafka-e2e</artifactId>
+
+    <properties>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>kafka</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hertzbeat</groupId>
+            <artifactId>hertzbeat-collector-kafka</artifactId>
+            <version>${hertzbeat.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hertzbeat</groupId>
+            <artifactId>hertzbeat-collector-common</artifactId>
+            <version>${hertzbeat.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

hertzbeat-e2e/hertzbeat-collector-kafka-e2e/src/test/java/org/apache/hertzbeat/collector/collect/kafka/KafkaCollectE2eTest.java

@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hertzbeat.collector.collect.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.job.protocol.KafkaProtocol;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Stream;
+
+/**
+ * KafkaCollectE2E
+ */
+@Slf4j
+public class KafkaCollectE2eTest {
+
+    private static final String ZOOKEEPER_IMAGE_NAME = "zookeeper:3.8.4";
+    private static final String ZOOKEEPER_NAME = "zookeeper";
+    private static final Integer ZOOKEEPER_PORT = 2181;
+    private static final String KAFKA_IMAGE_NAME = "confluentinc/cp-kafka:7.4.7";
+    private static final String KAFKA_NAME = "kafka";
+    private static GenericContainer<?> zookeeperContainer;
+    private static KafkaContainer kafkaContainer;
+    private Metrics metrics;
+    private KafkaCollectImpl kafkaCollect;
+    private CollectRep.MetricsData.Builder builder;
+
+    @AfterAll
+    public static void tearDown() {
+        kafkaContainer.stop();
+        zookeeperContainer.stop();
+    }
+
+    @BeforeEach
+    public void setUp() {
+        kafkaCollect = new KafkaCollectImpl();
+        metrics = new Metrics();
+        Network.NetworkImpl network = Network.builder().build();
+        zookeeperContainer = new GenericContainer<>(DockerImageName.parse(ZOOKEEPER_IMAGE_NAME))
+                .withExposedPorts(ZOOKEEPER_PORT)
+                .withNetwork(network)
+                .withNetworkAliases(ZOOKEEPER_NAME)
+                .waitingFor(Wait.forListeningPort())
+                .withStartupTimeout(Duration.ofSeconds(30));
+        zookeeperContainer.setPortBindings(Collections.singletonList(ZOOKEEPER_PORT + ":" + ZOOKEEPER_PORT));
+
+        Startables.deepStart(Stream.of(zookeeperContainer)).join();
+
+        kafkaContainer = new KafkaContainer(DockerImageName.parse(KAFKA_IMAGE_NAME))
+                .withExternalZookeeper(ZOOKEEPER_NAME + ":2181")
+                .withNetwork(network)
+                .withNetworkAliases(KAFKA_NAME)
+                .withLogConsumer(
+                        new Slf4jLogConsumer(
+                                DockerLoggerFactory.getLogger(KAFKA_IMAGE_NAME)));
+        Startables.deepStart(Stream.of(kafkaContainer)).join();
+    }
+
+    @Test
+    public void testKafkaCollect() throws ExecutionException, InterruptedException, TimeoutException {
+
+        Assertions.assertTrue(zookeeperContainer.isRunning(), "Zookeeper container should be running");
+        Assertions.assertTrue(kafkaContainer.isRunning(), "Kafka container should be running");
+
+        String topicName = "test-topic";
+
+        String bootstrapServers = kafkaContainer.getBootstrapServers().replace("PLAINTEXT://", "");
+        KafkaProtocol kafkaProtocol = new KafkaProtocol();
+        kafkaProtocol.setHost(bootstrapServers.split(":")[0]);
+        kafkaProtocol.setPort(bootstrapServers.split(":")[1]);
+        kafkaProtocol.setCommand("topic-list");
+        metrics.setKclient(kafkaProtocol);
+        log.info("bootstrapServers: {}", bootstrapServers);
+
+        // Create Topic
+        Properties properties = new Properties();
+        properties.put("bootstrap.servers", bootstrapServers);
+        AdminClient adminClient = KafkaAdminClient.create(properties);
+        int numPartitions = 1;
+        short replicationFactor = 1;
+        NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
+        adminClient.createTopics(Collections.singletonList(newTopic)).all().get(60, TimeUnit.SECONDS);
+
+        // Verify the information of topic list monitoring
+        builder = CollectRep.MetricsData.newBuilder();
+        kafkaCollect.collect(builder, 0, "kafka", metrics);
+        Assertions.assertTrue(builder.getValuesList().stream()
+                .anyMatch(valueRow -> valueRow.getColumns(0).equals(topicName)));
+
+        // Verify the information monitored by topic description
+        builder = CollectRep.MetricsData.newBuilder();
+        kafkaProtocol.setCommand("topic-describe");
+        kafkaCollect.collect(builder, 0, "kafka", metrics);
+        List<CollectRep.ValueRow> topicDescribeList = builder.getValuesList();
+        CollectRep.ValueRow firstRow = topicDescribeList.get(0);
+        Assertions.assertAll(
+                () -> Assertions.assertEquals(topicName, firstRow.getColumns(0)),
+                () -> Assertions.assertEquals(String.valueOf(numPartitions), firstRow.getColumns(1)),
+                () -> Assertions.assertEquals("0", firstRow.getColumns(2)),
+                () -> Assertions.assertEquals(kafkaProtocol.getHost(), firstRow.getColumns(3)),
+                () -> Assertions.assertEquals(kafkaProtocol.getPort(), firstRow.getColumns(4)),
+                () -> Assertions.assertEquals(String.valueOf(replicationFactor), firstRow.getColumns(5))
+        );
+    }
+}

hertzbeat-e2e/pom.xml

@@ -0,0 +1,69 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.hertzbeat</groupId>
+        <artifactId>hertzbeat</artifactId>
+        <version>2.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>hertzbeat-e2e</artifactId>
+    <packaging>pom</packaging>
+    <modules>
+        <module>hertzbeat-collector-kafka-e2e</module>
+    </modules>
+
+    <properties>
+<!--        <maven.test.skip>true</maven.test.skip>-->
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <testcontainers.version>1.20.2</testcontainers.version>
+        <junit.version>4.13.2</junit.version>
+        <awaitility.version>4.2.0</awaitility.version>
+    </properties>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>${testcontainers.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>${awaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

material/licenses/LICENSE

@@ -533,6 +533,7 @@ The text of each license is also included in licenses/LICENSE-[project].txt.
     https://mvnrepository.com/artifact/org.java-websocket/Java-WebSocket/1.5.2
     https://mvnrepository.com/artifact/com.taosdata.jdbc/taos-jdbcdriver/3.0.0
     https://mvnrepository.com/artifact/com.beetstra.jutf7/jutf7/1.0.0
+    https://mvnrepository.com/artifact/org.testcontainers/kafka/1.20.2 MIT
 
 ========================================================================
 MPL-1.1 licenses

material/licenses/backend/LICENSE

@@ -533,6 +533,7 @@ The text of each license is also included in licenses/LICENSE-[project].txt.
     https://mvnrepository.com/artifact/org.java-websocket/Java-WebSocket/1.5.2
     https://mvnrepository.com/artifact/com.taosdata.jdbc/taos-jdbcdriver/3.0.0
     https://mvnrepository.com/artifact/com.beetstra.jutf7/jutf7/1.0.0
+    https://mvnrepository.com/artifact/org.testcontainers/kafka/1.20.2 MIT
 
 ========================================================================
 MPL-1.1 licenses

pom.xml

@@ -88,6 +88,7 @@
         <module>hertzbeat-push</module>
         <module>hertzbeat-plugin</module>
         <module>hertzbeat-grafana</module>
+        <module>hertzbeat-e2e</module>
     </modules>
 
     <properties>