[ISSUE #42] fix the problem HealthExcutor don't doCheck asynchronously

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/constant/HealthConstant.java

@@ -0,0 +1,22 @@
+/*
+ * 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.eventmesh.dashboard.console.constant;
+
+public class HealthConstant {
+    public static final String NEW_LINE_ENDING = "\n";
+}

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/health/CheckResultCache.java

@@ -17,6 +17,7 @@
 
 package org.apache.eventmesh.dashboard.console.health;
 
+import org.apache.eventmesh.dashboard.console.constant.HealthConstant;
 import org.apache.eventmesh.dashboard.console.enums.health.HealthCheckStatus;
 import org.apache.eventmesh.dashboard.console.health.check.config.HealthCheckObjectConfig;
 
@@ -41,8 +42,12 @@ public void update(String type, Long typeId, HealthCheckStatus status, String re
             cacheMap.put(type, subMap);
         }
         CheckResult oldResult = subMap.get(typeId);
-        String oldDesc = Objects.isNull(oldResult.getResultDesc()) ? "" : oldResult.getResultDesc() + "\n";
-        CheckResult result = new CheckResult(status, oldDesc + resultDesc, LocalDateTime.now(),
+        String description = resultDesc;
+        if (oldResult.getResultDesc() != null && !oldResult.getResultDesc().isEmpty()) {
+            description = oldResult.getResultDesc() + HealthConstant.NEW_LINE_ENDING + resultDesc;
+        }
+        description += " Latency: " + latency.toString() + "ms";
+        CheckResult result = new CheckResult(status, description, LocalDateTime.now(),
             latency, oldResult.getConfig());
         subMap.put(typeId, result);
     }
@@ -53,7 +58,6 @@ public void update(String type, Long typeId, HealthCheckStatus status, String re
             subMap = new HashMap<>();
             cacheMap.put(type, subMap);
         }
-        CheckResult resultToUpdate = subMap.get(typeId);
         subMap.put(typeId, new CheckResult(status, resultDesc, LocalDateTime.now(), latency, config));
     }
 

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/health/HealthExecutor.java

@@ -26,6 +26,8 @@
 import org.apache.eventmesh.dashboard.console.service.health.HealthDataService;
 
 import java.util.ArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 import lombok.Getter;
 import lombok.Setter;
@@ -44,6 +46,8 @@ public class HealthExecutor {
     @Setter
     private CheckResultCache memoryCache;
 
+    private final ExecutorService executorService = Executors.newCachedThreadPool();
+
     /**
      * execute function is where health check services work.
      *
@@ -58,15 +62,15 @@ public void execute(AbstractHealthCheckService service) {
             memoryCache.update(service.getConfig().getHealthCheckResourceType(), service.getConfig().getInstanceId(), HealthCheckStatus.CHECKING, "",
                 null, service.getConfig());
             //The callback interface is used to pass the processing methods for checking success and failure.
-            service.doCheck(new HealthCheckCallback() {
+            executorService.submit(() -> service.doCheck(new HealthCheckCallback() {
                 @Override
                 public void onSuccess() {
                     //when the health check is successful, the result is updated to the memory cache.
                     Long latency = System.currentTimeMillis() - startTime;
                     HealthCheckStatus status =
                         latency > service.getConfig().getRequestTimeoutMillis() ? HealthCheckStatus.TIMEOUT : HealthCheckStatus.PASSED;
                     memoryCache.update(service.getConfig().getHealthCheckResourceType(), service.getConfig().getInstanceId(),
-                        status, "health check success", latency
+                        status, "Health check succeed.", latency
                     );
                 }
 
@@ -78,7 +82,8 @@ public void onFail(Exception e) {
                         HealthCheckStatus.FAILED, e.getMessage(),
                         System.currentTimeMillis() - startTime);
                 }
-            });
+            }));
+
         } catch (Exception e) {
             log.error("Health check failed for reason: {}. Service config is {}", e, service.getConfig());
             memoryCache.update(service.getConfig().getHealthCheckResourceType(), service.getConfig().getInstanceId(), HealthCheckStatus.FAILED,

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/health/HealthService.java

@@ -68,7 +68,7 @@ private static void setClassCache(Class<?> clazz) {
      *
      * @see AbstractHealthCheckService
      */
-    private Map<String, Map<Long, AbstractHealthCheckService>> checkServiceMap = new ConcurrentHashMap<>();
+    private final Map<String, Map<Long, AbstractHealthCheckService>> checkServiceMap = new ConcurrentHashMap<>();
 
     private ScheduledThreadPoolExecutor scheduledExecutor;
 
@@ -87,10 +87,13 @@ public void insertCheckService(HealthCheckObjectConfig config) {
             if (Objects.nonNull(config.getSimpleClassName())) {
                 Class<?> clazz = HEALTH_CHECK_CLASS_CACHE.get(config.getSimpleClassName());
                 healthCheckService = createCheckService(clazz, config);
-                //if simpleClassName is null, use type(storage) and subType(redis) to create healthCheckService
+                // you can pass an object to create a HealthCheckService(not commonly used)
+            } else if (Objects.nonNull(config.getCheckClass())) {
+                healthCheckService = createCheckService(config.getCheckClass(), config);
+                //if simpleClassName and CheckClass are both null, use type(storage) and subType(redis) to create healthCheckService
+                //This is the default create method.
                 //healthCheckService is annotated with @HealthCheckType(type = "storage", subType = "redis")
-            } else if (Objects.isNull(config.getSimpleClassName())
-                && Objects.nonNull(config.getHealthCheckResourceType()) && Objects.nonNull(
+            } else if (Objects.nonNull(config.getHealthCheckResourceType()) && Objects.nonNull(
                 config.getHealthCheckResourceSubType())) {
                 for (Entry<String, Class<?>> entry : HEALTH_CHECK_CLASS_CACHE.entrySet()) {
                     Class<?> clazz = entry.getValue();
@@ -101,9 +104,6 @@ public void insertCheckService(HealthCheckObjectConfig config) {
                         break;
                     }
                 }
-                // you can pass an object to create a HealthCheckService(not commonly used)
-            } else if (Objects.nonNull(config.getCheckClass())) {
-                healthCheckService = createCheckService(config.getCheckClass(), config);
             }
         } catch (Exception e) {
             log.error("create healthCheckService failed, healthCheckObjectConfig:{}", config, e);
@@ -113,6 +113,13 @@ public void insertCheckService(HealthCheckObjectConfig config) {
         if (Objects.isNull(healthCheckService)) {
             throw new RuntimeException("No construct method of Health Check Service is found, config is {}" + config);
         }
+        insertCheckService(healthCheckService);
+    }
+
+    public void insertCheckService(AbstractHealthCheckService checkService) {
+        Map<Long, AbstractHealthCheckService> subMap = checkServiceMap.computeIfAbsent(checkService.getConfig().getHealthCheckResourceType(),
+            k -> new ConcurrentHashMap<>());
+        subMap.put(checkService.getConfig().getInstanceId(), checkService);
     }
 
     public void deleteCheckService(String resourceType, Long resourceId) {

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/health/check/config/HealthCheckObjectConfig.java

@@ -17,6 +17,8 @@
 
 package org.apache.eventmesh.dashboard.console.health.check.config;
 
+import java.util.Properties;
+
 import lombok.Data;
 
 @Data
@@ -32,6 +34,8 @@ public class HealthCheckObjectConfig {
 
     private Class<?> checkClass;
 
+    private Properties eventmeshProperties;
+
     private Long clusterId;
 
     private String connectUrl;

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/health/check/impl/StorageRedisCheck.java

@@ -77,6 +77,6 @@ public void init() {
             }
             redisUrl = builder.build().toString();
         }
-        RedisClient redisClient = RedisClient.create(redisUrl);
+        redisClient = RedisClient.create(redisUrl);
     }
 }

eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/health/HealthExecutorTest.java

@@ -113,9 +113,10 @@ public void initMock() {
     }
 
     @Test
-    public void testExecute() {
+    public void testExecute() throws InterruptedException {
         healthExecutor.execute(successHealthCheckService);
         healthExecutor.execute(failHealthCheckService);
+        Thread.sleep(1000);
         assertEquals(2, memoryCache.getCacheMap().get("storage").size());
         assertNotEquals(memoryCache.getCacheMap().get("storage").get(1L).getStatus(), memoryCache.getCacheMap().get("storage").get(2L).getStatus());
     }

eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/health/HealthServiceTest.java

@@ -17,8 +17,6 @@
 
 package org.apache.eventmesh.dashboard.console.health;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
 import org.apache.eventmesh.dashboard.console.health.callback.HealthCheckCallback;
 import org.apache.eventmesh.dashboard.console.health.check.AbstractHealthCheckService;
 import org.apache.eventmesh.dashboard.console.health.check.config.HealthCheckObjectConfig;
@@ -60,6 +58,8 @@ void testInsertCheckServiceWithSimpleClassName() {
         HealthCheckObjectConfig config = new HealthCheckObjectConfig();
         config.setInstanceId(1L);
         config.setSimpleClassName("StorageRedisCheck");
+        config.setHealthCheckResourceType("storage");
+        config.setHealthCheckResourceSubType("redis");
         config.setClusterId(1L);
         config.setConnectUrl("redis://localhost:6379");
         healthService.insertCheckService(config);
@@ -69,6 +69,8 @@ void testInsertCheckServiceWithSimpleClassName() {
     void testInsertCheckServiceWithClass() {
         HealthCheckObjectConfig config = new HealthCheckObjectConfig();
         config.setInstanceId(1L);
+        config.setHealthCheckResourceType("storage");
+        config.setHealthCheckResourceSubType("redis");
         config.setClusterId(1L);
         config.setCheckClass(TestHealthCheckService.class);
         healthService.insertCheckService(config);

eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/integration/health/HealthServiceIntegrateTest.java

@@ -0,0 +1,70 @@
+/*
+ * 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.eventmesh.dashboard.console.integration.health;
+
+import java.util.List;
+import org.apache.eventmesh.dashboard.console.entity.health.HealthCheckResultEntity;
+import org.apache.eventmesh.dashboard.console.enums.health.HealthCheckType;
+import org.apache.eventmesh.dashboard.console.health.CheckResultCache;
+import org.apache.eventmesh.dashboard.console.health.HealthService;
+import org.apache.eventmesh.dashboard.console.health.check.config.HealthCheckObjectConfig;
+import org.apache.eventmesh.dashboard.console.service.health.HealthDataService;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.jdbc.Sql;
+
+@SpringBootTest
+@ActiveProfiles("test")
+@Sql(executionPhase = Sql.ExecutionPhase.BEFORE_TEST_METHOD, scripts = {"classpath:use-test-schema.sql", "classpath:eventmesh-dashboard.sql"})
+public class HealthServiceIntegrateTest {
+    HealthService healthService = new HealthService();
+
+    @Autowired
+    private HealthDataService healthDataService;
+
+    private final CheckResultCache checkResultCache = new CheckResultCache();
+    @BeforeEach
+    void init() {
+        healthService.createExecutor(healthDataService, checkResultCache);
+    }
+
+    @Test
+    void testStorageRedis() throws InterruptedException {
+        HealthCheckObjectConfig config = new HealthCheckObjectConfig();
+        config.setClusterId(1L);
+        config.setInstanceId(1L);
+        config.setHealthCheckResourceType("storage");
+        config.setHealthCheckResourceSubType("redis");
+        config.setConnectUrl("redis://localhost:6379");
+        healthService.insertCheckService(config);
+        healthService.executeAll();
+        Thread.sleep(1000);
+        healthService.executeAll();
+
+        HealthCheckResultEntity queryEntity = new HealthCheckResultEntity();
+        queryEntity.setClusterId(1L);
+        queryEntity.setType(HealthCheckType.STORAGE.getNumber());
+        queryEntity.setTypeId(1L);
+        List<HealthCheckResultEntity> results =  healthDataService.queryHealthCheckResultByClusterIdAndTypeAndTypeId(queryEntity);
+        Assertions.assertEquals(2,results.size());
+    }
+}

pom.xml

@@ -135,6 +135,17 @@
           </execution>
         </executions>
       </plugin>
+
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <version>3.2.2</version>
+        <configuration>
+          <excludes>
+            <exclude>**/org/apache/eventmesh/dashboard/console/integration/**/*.java</exclude>
+          </excludes>
+        </configuration>
+      </plugin>
     </plugins>
   </build>
 </project>