You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@hertzbeat.apache.org by zh...@apache.org on 2024/12/09 15:57:33 UTC
(hertzbeat) branch feature/integrate_apache_arrow updated: [test] fix e2e test
This is an automated email from the ASF dual-hosted git repository.
zhengqiwei pushed a commit to branch feature/integrate_apache_arrow
in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
The following commit(s) were added to refs/heads/feature/integrate_apache_arrow by this push:
new 6dde4b53f [test] fix e2e test
6dde4b53f is described below
commit 6dde4b53faa1b7e07f975d92fc3f7b9632f8649c
Author: Calvin <na...@163.com>
AuthorDate: Mon Dec 9 23:57:25 2024 +0800
[test] fix e2e test
---
.../collect/basic/ssh/SshCollectE2eTest.java | 44 +++++++++++++--------
.../collect/kafka/KafkaCollectE2eTest.java | 45 +++++++++++++++-------
2 files changed, 60 insertions(+), 29 deletions(-)
diff --git a/hertzbeat-e2e/hertzbeat-collector-basic-e2e/src/test/java/org/apache/hertzbeat/collector/collect/basic/ssh/SshCollectE2eTest.java b/hertzbeat-e2e/hertzbeat-collector-basic-e2e/src/test/java/org/apache/hertzbeat/collector/collect/basic/ssh/SshCollectE2eTest.java
index 86099c024..10463dec4 100644
--- a/hertzbeat-e2e/hertzbeat-collector-basic-e2e/src/test/java/org/apache/hertzbeat/collector/collect/basic/ssh/SshCollectE2eTest.java
+++ b/hertzbeat-e2e/hertzbeat-collector-basic-e2e/src/test/java/org/apache/hertzbeat/collector/collect/basic/ssh/SshCollectE2eTest.java
@@ -17,8 +17,14 @@
package org.apache.hertzbeat.collector.collect.basic.ssh;
+import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.collect.common.MetricsDataBuilder;
import org.apache.hertzbeat.collector.collect.ssh.SshCollectImpl;
+import org.apache.hertzbeat.common.entity.arrow.ArrowVectorReader;
+import org.apache.hertzbeat.common.entity.arrow.ArrowVectorReaderImpl;
+import org.apache.hertzbeat.common.entity.arrow.ArrowVectorWriterImpl;
+import org.apache.hertzbeat.common.entity.arrow.RowWrapper;
import org.apache.hertzbeat.common.entity.job.Job;
import org.apache.hertzbeat.common.entity.job.Metrics;
import org.apache.hertzbeat.common.entity.job.protocol.SshProtocol;
@@ -97,8 +103,9 @@ public class SshCollectE2eTest {
log.info("Container started successfully with mapped port: {}", mappedPort);
}
+ @SneakyThrows
@Test
- public void testSshCollect() throws ExecutionException, InterruptedException, TimeoutException {
+ public void testSshCollect() {
// Verify container running status
Assertions.assertTrue(linuxContainer.isRunning(), "Ubuntu container should be running");
@@ -106,27 +113,29 @@ public class SshCollectE2eTest {
Job ubuntuJob = appService.getAppDefine("ubuntu");
List<Metrics> metricsDefinitions = ubuntuJob.getMetrics();
- metricsDefinitions.forEach(this::testMetricsCollection);
+ for (Metrics metricsDefinition : metricsDefinitions) {
+ testMetricsCollection(metricsDefinition);
+ }
}
- private void testMetricsCollection(Metrics metricsDef) {
+ private void testMetricsCollection(Metrics metricsDef) throws Exception {
String name = metricsDef.getName();
- CollectRep.MetricsData.Builder builder = executeCollection(metricsDef);
+ MetricsDataBuilder metricsDataBuilder = executeCollection(metricsDef);
- // Verify CPU metrics
- Assertions.assertTrue(builder.getValuesList().size() > 0, name + " metrics values should not be empty");
- CollectRep.ValueRow valueRow = builder.getValuesList().get(0);
+ final CollectRep.MetricsData metricsData = metricsDataBuilder.build();
+ try (ArrowVectorReader arrowVectorReader = new ArrowVectorReaderImpl(metricsData.getData().toByteArray())) {
+ Assertions.assertTrue(arrowVectorReader.getRowCount() > 0, name + " metrics values should not be empty");
- // Verify all columns have values
- for (int i = 0; i < valueRow.getColumnsCount(); i++) {
- Assertions.assertFalse(valueRow.getColumns(i).isEmpty(),
- String.format("%s metric column %d should not be empty", name, i));
+ // Verify CPU metrics
+ RowWrapper firstRowWrapper = arrowVectorReader.readRow().nextRow();
+ firstRowWrapper.cellStream().forEach(cell -> Assertions.assertFalse(cell.getValue().isEmpty(),
+ String.format("%s metric column %s should not be empty", name, cell.getField().getName())));
}
- log.info(name + " metrics validation passed");
+ log.info("{} metrics validation passed", name);
}
- private CollectRep.MetricsData.Builder executeCollection(Metrics metricsDef) {
+ private MetricsDataBuilder executeCollection(Metrics metricsDef) {
// Build SSH protocol configuration
SshProtocol sshProtocol = buildSshProtocol(metricsDef);
metrics.setSsh(sshProtocol);
@@ -138,9 +147,12 @@ public class SshCollectE2eTest {
metricsDef.getAliasFields());
// Execute collection
- CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
- sshCollect.collect(builder, 0, "ubuntu", metrics);
- return builder;
+ CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder().setId(0L).setApp("ubuntu");
+ try (final ArrowVectorWriterImpl arrowVectorWriter = new ArrowVectorWriterImpl(metrics.getAliasFields())) {
+ final MetricsDataBuilder metricsDataBuilder = new MetricsDataBuilder(builder, arrowVectorWriter);
+ sshCollect.collect(metricsDataBuilder, metrics);
+ return metricsDataBuilder;
+ }
}
private SshProtocol buildSshProtocol(Metrics metricsDef) {
diff --git a/hertzbeat-e2e/hertzbeat-collector-kafka-e2e/src/test/java/org/apache/hertzbeat/collector/collect/kafka/KafkaCollectE2eTest.java b/hertzbeat-e2e/hertzbeat-collector-kafka-e2e/src/test/java/org/apache/hertzbeat/collector/collect/kafka/KafkaCollectE2eTest.java
index 319a69902..4b6efb336 100644
--- a/hertzbeat-e2e/hertzbeat-collector-kafka-e2e/src/test/java/org/apache/hertzbeat/collector/collect/kafka/KafkaCollectE2eTest.java
+++ b/hertzbeat-e2e/hertzbeat-collector-kafka-e2e/src/test/java/org/apache/hertzbeat/collector/collect/kafka/KafkaCollectE2eTest.java
@@ -19,7 +19,10 @@ package org.apache.hertzbeat.collector.collect.kafka;
import lombok.extern.slf4j.Slf4j;
import org.apache.hertzbeat.collector.collect.common.MetricsDataBuilder;
+import org.apache.hertzbeat.common.entity.arrow.ArrowVectorReader;
+import org.apache.hertzbeat.common.entity.arrow.ArrowVectorReaderImpl;
import org.apache.hertzbeat.common.entity.arrow.ArrowVectorWriterImpl;
+import org.apache.hertzbeat.common.entity.arrow.RowWrapper;
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;
@@ -40,6 +43,7 @@ import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.DockerLoggerFactory;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
@@ -97,7 +101,7 @@ public class KafkaCollectE2eTest {
}
@Test
- public void testKafkaCollect() throws ExecutionException, InterruptedException, TimeoutException {
+ public void testKafkaCollect() throws Exception {
Assertions.assertTrue(zookeeperContainer.isRunning(), "Zookeeper container should be running");
Assertions.assertTrue(kafkaContainer.isRunning(), "Kafka container should be running");
@@ -126,9 +130,20 @@ public class KafkaCollectE2eTest {
try (final ArrowVectorWriterImpl arrowVectorWriter = new ArrowVectorWriterImpl(metrics.getAliasFields())) {
final MetricsDataBuilder metricsDataBuilder = new MetricsDataBuilder(builder, arrowVectorWriter);
kafkaCollect.collect(metricsDataBuilder, metrics);
+
+ final CollectRep.MetricsData metricsData = metricsDataBuilder.build();
+ try (ArrowVectorReader arrowVectorReader = new ArrowVectorReaderImpl(metricsData.getData().toByteArray())) {
+ RowWrapper rowWrapper = arrowVectorReader.readRow();
+
+ List<String> topicNameList = new ArrayList<>();
+ while (rowWrapper.hasNextRow()) {
+ rowWrapper = rowWrapper.nextRow();
+
+ topicNameList.add(rowWrapper.nextCell().getValue());
+ }
+ Assertions.assertTrue(topicNameList.contains(topicName));
+ }
}
- Assertions.assertTrue(builder.getValuesList().stream()
- .anyMatch(valueRow -> valueRow.getColumns(0).equals(topicName)));
// Verify the information monitored by topic description
kafkaProtocol.setCommand("topic-describe");
@@ -136,16 +151,20 @@ public class KafkaCollectE2eTest {
try (final ArrowVectorWriterImpl arrowVectorWriter = new ArrowVectorWriterImpl(metrics.getAliasFields())) {
final MetricsDataBuilder metricsDataBuilder = new MetricsDataBuilder(builder, arrowVectorWriter);
kafkaCollect.collect(metricsDataBuilder, metrics);
+
+ final CollectRep.MetricsData metricsData = metricsDataBuilder.build();
+ try (ArrowVectorReader arrowVectorReader = new ArrowVectorReaderImpl(metricsData.getData().toByteArray())) {
+ RowWrapper firstRowWrapper = arrowVectorReader.readRow().nextRow();
+
+ Assertions.assertAll(
+ () -> Assertions.assertEquals(topicName, firstRowWrapper.nextCell().getValue()),
+ () -> Assertions.assertEquals(String.valueOf(numPartitions), firstRowWrapper.nextCell().getValue()),
+ () -> Assertions.assertEquals("0", firstRowWrapper.nextCell().getValue()),
+ () -> Assertions.assertEquals(kafkaProtocol.getHost(), firstRowWrapper.nextCell().getValue()),
+ () -> Assertions.assertEquals(kafkaProtocol.getPort(), firstRowWrapper.nextCell().getValue()),
+ () -> Assertions.assertEquals(String.valueOf(replicationFactor), firstRowWrapper.nextCell().getValue())
+ );
+ }
}
- 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))
- );
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@hertzbeat.apache.org
For additional commands, e-mail: notifications-help@hertzbeat.apache.org