[Improve]Improve e2e code (#2945)

hertzbeat-e2e/hertzbeat-collector-basic-e2e/src/test/java/org/apache/hertzbeat/collector/collect/basic/http/DockerMonitorE2eTest.java

@@ -22,6 +22,8 @@ import com.sun.net.httpserver.HttpServer;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.collector.collect.AbstractCollectE2eTest;
 import org.apache.hertzbeat.collector.collect.http.HttpCollectImpl;
+import org.apache.hertzbeat.collector.util.CollectUtil;
+import org.apache.hertzbeat.common.entity.job.Configmap;
 import org.apache.hertzbeat.common.entity.job.Job;
 import org.apache.hertzbeat.common.entity.job.Metrics;
 import org.apache.hertzbeat.common.entity.job.protocol.HttpProtocol;
@@ -39,6 +41,10 @@ import java.io.IOException;
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.nio.file.Files;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
 
 /**
  * Integration test for Docker monitoring functionality
@@ -52,9 +58,18 @@ public class DockerMonitorE2eTest extends AbstractCollectE2eTest {
     private static final String LOCALHOST = "127.0.0.1";
     private static HttpServer mockServer;
 
+    @AfterAll
+    public static void tearDown() {
+        if (mockServer != null) {
+            mockServer.stop(0);
+        }
+    }
+
     @BeforeEach
     public void setUp() throws Exception {
+
         super.setUp();
+        // Setup collect instance
         collect = new HttpCollectImpl();
 
         // Setup mock server and endpoints
@@ -65,6 +80,7 @@ public class DockerMonitorE2eTest extends AbstractCollectE2eTest {
         // Setup Docker API endpoints
         String containerResponse = loadResponseFromFile("classpath:http/docker/containers_result.txt");
         String infoResponse = loadResponseFromFile("classpath:http/docker/system_result.txt");
+        String containerStatsResponse = loadResponseFromFile("classpath:http/docker/containers_stats.txt");
 
         mockServer.createContext("/containers/json", exchange -> {
             String query = exchange.getRequestURI().getQuery();
@@ -77,6 +93,8 @@ public class DockerMonitorE2eTest extends AbstractCollectE2eTest {
         });
 
         mockServer.createContext("/info", exchange -> sendJsonResponse(exchange, infoResponse));
+        mockServer.createContext("/containers/34174a918eb2e38cdb097c910f74af845e7383b04765d26ad52f940f86342a64/stats", exchange -> sendJsonResponse(exchange, containerStatsResponse));
+
     }
 
     private String loadResponseFromFile(String resourcePath) throws Exception {
@@ -94,13 +112,12 @@ public class DockerMonitorE2eTest extends AbstractCollectE2eTest {
     @Test
     public void testDockerMonitor() {
         Job dockerJob = appService.getAppDefine("docker");
-        dockerJob.getMetrics().forEach(metricsDef -> {
-            // Skip metrics containing "^o^" as parameter substitution is not supported in e2e tests
-            if (metricsDef.getHttp().getUrl().contains("^o^")) {
-                return;
-            }
-            validateMetricsCollection(metricsDef, metricsDef.getName());
-        });
+        List<Map<String, Configmap>> configmapFromPreCollectData = new LinkedList<>();
+        for (Metrics metricsDef : dockerJob.getMetrics()) {
+            metricsDef = CollectUtil.replaceCryPlaceholderToMetrics(metricsDef, configmapFromPreCollectData.size() > 0 ? configmapFromPreCollectData.get(0) : new HashMap<>());
+            CollectRep.MetricsData metricsData = validateMetricsCollection(metricsDef, metricsDef.getName());
+            configmapFromPreCollectData = CollectUtil.getConfigmapFromPreCollectData(metricsData);
+        }
     }
 
     @Override
@@ -122,11 +139,4 @@ public class DockerMonitorE2eTest extends AbstractCollectE2eTest {
         metrics.setHttp(protocol);
         return collectMetricsData(metrics, metricsDef);
     }
-
-    @AfterAll
-    public static void tearDown() {
-        if (mockServer != null) {
-            mockServer.stop(0);
-        }
-    }
 }

hertzbeat-e2e/hertzbeat-collector-basic-e2e/src/test/resources/http/docker/containers_stats.txt

@@ -0,0 +1 @@
+{"read":"2000-01-01T00:00:15.864368408Z","preread":"2000-01-01T00:00:14.86209639Z","pids_stats":{"current":41},"blkio_stats":{"io_service_bytes_recursive":[{"major":8,"minor":0,"op":"Read","value":118784},{"major":8,"minor":0,"op":"Write","value":267395072},{"major":8,"minor":0,"op":"Sync","value":267395072},{"major":8,"minor":0,"op":"Async","value":118784},{"major":8,"minor":0,"op":"Total","value":267513856},{"major":253,"minor":0,"op":"Read","value":118784},{"major":253,"minor":0,"op":"Write","value":267395072},{"major":253,"minor":0,"op":"Sync","value":267395072},{"major":253,"minor":0,"op":"Async","value":118784},{"major":253,"minor":0,"op":"Total","value":267513856}],"io_serviced_recursive":[{"major":8,"minor":0,"op":"Read","value":8},{"major":8,"minor":0,"op":"Write","value":7892},{"major":8,"minor":0,"op":"Sync","value":7892},{"major":8,"minor":0,"op":"Async","value":8},{"major":8,"minor":0,"op":"Total","value":7900},{"major":253,"minor":0,"op":"Read","value":8},{"major":253,"minor":0,"op":"Write","value":7892},{"major":253,"minor":0,"op":"Sync","value":7892},{"major":253,"minor":0,"op":"Async","value":8},{"major":253,"minor":0,"op":"Total","value":7900}],"io_queue_recursive":[],"io_service_time_recursive":[],"io_wait_time_recursive":[],"io_merged_recursive":[],"io_time_recursive":[],"sectors_recursive":[]},"num_procs":0,"storage_stats":{},"cpu_stats":{"cpu_usage":{"total_usage":4810174080874,"percpu_usage":[733251885463,588352569755,649761861064,692942680516,618939038384,513206148155,431772846470,581947051067],"usage_in_kernelmode":1191390000000,"usage_in_usermode":1463160000000},"system_cpu_usage":21196078922904483,"online_cpus":8,"throttling_data":{"periods":0,"throttled_periods":0,"throttled_time":0}},"precpu_stats":{"cpu_usage":{"total_usage":4810165603627,"percpu_usage":[733250739914,588352569755,649761643572,692942567027,618937059553,513205711609,431769258762,581946053435],"usage_in_kernelmode":1191390000000,"usage_in_usermode":1463160000000},"system_cpu_usage":21196070922904483,"online_cpus":8,"throttling_data":{"periods":0,"throttled_periods":0,"throttled_time":0}},"memory_stats":{"usage":505090048,"max_usage":519450624,"stats":{"active_anon":396529664,"active_file":52400128,"cache":108560384,"dirty":0,"hierarchical_memory_limit":9223372036854771712,"hierarchical_memsw_limit":9223372036854771712,"inactive_anon":0,"inactive_file":56160256,"mapped_file":40960,"pgfault":409148,"pgmajfault":0,"pgpgin":343733,"pgpgout":220420,"rss":396529664,"rss_huge":0,"total_active_anon":396529664,"total_active_file":52400128,"total_cache":108560384,"total_dirty":0,"total_inactive_anon":0,"total_inactive_file":56160256,"total_mapped_file":40960,"total_pgfault":0,"total_pgmajfault":0,"total_pgpgin":0,"total_pgpgout":0,"total_rss":396529664,"total_rss_huge":0,"total_unevictable":0,"total_writeback":0,"unevictable":0,"writeback":0},"limit":33566269440},"name":"/mysql","id":"34174a918eb2e38cdb097c910f74af845e7383b04765d26ad52f940f86342a64","networks":{"eth0":{"rx_bytes":68182,"rx_packets":533,"rx_errors":0,"rx_dropped":0,"tx_bytes":211850,"tx_packets":361,"tx_errors":0,"tx_dropped":0}}}

hertzbeat-e2e/hertzbeat-collector-common-e2e/src/test/java/org/apache/hertzbeat/collector/collect/AbstractCollectE2eTest.java

@@ -18,6 +18,11 @@
 package org.apache.hertzbeat.collector.collect;
 
 import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.dispatch.CollectDataDispatch;
+import org.apache.hertzbeat.collector.dispatch.MetricsCollect;
+import org.apache.hertzbeat.collector.dispatch.timer.Timeout;
+import org.apache.hertzbeat.collector.dispatch.timer.WheelTimerTask;
+import org.apache.hertzbeat.common.entity.job.Job;
 import org.apache.hertzbeat.common.entity.job.Metrics;
 import org.apache.hertzbeat.common.entity.job.protocol.Protocol;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
@@ -26,9 +31,14 @@ import org.apache.hertzbeat.manager.service.impl.ObjectStoreConfigServiceImpl;
 import org.junit.jupiter.api.Assertions;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
 
+import java.util.List;
 import java.util.stream.Collectors;
 
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 /**
  * AbstractCollectE2eTest
  */
@@ -40,12 +50,26 @@ public abstract class AbstractCollectE2eTest {
 
     protected AbstractCollect collect;
 
+    protected MetricsCollect metricsCollect;
+
     protected Metrics metrics;
 
     @Mock
     protected ObjectStoreConfigServiceImpl objectStoreConfigService;
+    @Mock
+    private WheelTimerTask timerJob;
+    @Mock
+    private Timeout timeout;
+    @Mock
+    private Job job;
 
     public void setUp() throws Exception {
+        // Initialize mocks
+        MockitoAnnotations.openMocks(this);
+        when(timeout.task()).thenReturn(timerJob);
+        when(timerJob.getJob()).thenReturn(job);
+        metricsCollect = new MetricsCollect(mock(Metrics.class), timeout, mock(CollectDataDispatch.class), null, List.of());
+
         // Initialize services and components
         appService.run();
         metrics = new Metrics();
@@ -55,32 +79,36 @@ public abstract class AbstractCollectE2eTest {
      * Validate metrics collection, check if the metrics values are not empty <br/>
      * We believe that all monitoring metrics should have data
      */
-    protected void validateMetricsCollection(Metrics metricsDef, String metricName) {
+    protected CollectRep.MetricsData validateMetricsCollection(Metrics metricsDef, String metricName) {
         CollectRep.MetricsData.Builder metricsData = collectMetrics(metricsDef);
-        
+
+        metricsCollect.calculateFields(metricsDef, metricsData);
+
         Assertions.assertTrue(metricsData.getValuesList().size() > 0,
                 String.format("%s metrics values should not be empty", metricName));
 
-        CollectRep.ValueRow firstRow = metricsData.getValuesList().get(0);
-        for (int i = 0; i < firstRow.getColumnsCount(); i++) {
-            Assertions.assertFalse(firstRow.getColumns(i).isEmpty(),
-                    String.format("%s metric column %d should not be empty", metricName, i));
+        for (CollectRep.ValueRow valueRow : metricsData.getValuesList()) {
+            for (int i = 0; i < valueRow.getColumnsCount(); i++) {
+                Assertions.assertFalse(valueRow.getColumns(i).isEmpty(),
+                        String.format("%s metric column %d should not be empty", metricName, i));
+            }
         }
 
         log.info("{} metrics validation passed", metricName);
+        return metricsData.build();
     }
-    
+
     protected void setMetricsAliasFields(Metrics metrics, Metrics metricsDef) {
         metrics.setAliasFields(metricsDef.getAliasFields() == null
                 ? metricsDef.getFields().stream()
-                        .map(Metrics.Field::getField)
-                        .collect(Collectors.toList()) : 
+                .map(Metrics.Field::getField)
+                .collect(Collectors.toList()) :
                 metricsDef.getAliasFields());
     }
 
     protected abstract CollectRep.MetricsData.Builder collectMetrics(Metrics metricsDef);
 
-    protected CollectRep.MetricsData.Builder collectMetricsData(Metrics metrics, Metrics metricsDef){
+    protected CollectRep.MetricsData.Builder collectMetricsData(Metrics metrics, Metrics metricsDef) {
         setMetricsAliasFields(metrics, metricsDef);
 
         // Collect metrics