[ISSUE #25] Add HealthService

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/config/SchedulerConfig.java → eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/config/HealthCheckConfig.java

@@ -17,17 +17,15 @@
 
 package org.apache.eventmesh.dashboard.console.config;
 
-import org.springframework.context.annotation.Bean;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
-import org.springframework.stereotype.Component;
+import org.apache.eventmesh.dashboard.console.health.check.config.HealthCheckObjectConfig;
 
-@Component
-public class SchedulerConfig {
+import java.util.List;
 
-    @Bean
-    public ThreadPoolTaskScheduler taskScheduler() {
-        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
-        taskScheduler.setPoolSize(5);
-        return taskScheduler;
-    }
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class HealthCheckConfig {
+    private List<HealthCheckObjectConfig> checkObjectConfigList;
 }

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

@@ -0,0 +1,90 @@
+/*
+ * 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.health;
+
+import org.apache.eventmesh.dashboard.console.enums.health.HealthCheckStatus;
+import org.apache.eventmesh.dashboard.console.health.check.config.HealthCheckObjectConfig;
+
+import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+
+public class CheckResultCache {
+
+    private final HashMap<String, HashMap<Long, CheckResult>> cacheMap = new HashMap<>();
+
+    public void update(String type, Long typeId, HealthCheckStatus status, String resultDesc, Long latency) {
+        HashMap<Long, CheckResult> subMap = cacheMap.get(type);
+        if (Objects.isNull(subMap)) {
+            subMap = new HashMap<>();
+            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(),
+            latency, oldResult.getConfig());
+        subMap.put(typeId, result);
+    }
+
+    public void update(String type, Long typeId, HealthCheckStatus status, String resultDesc, Long latency, HealthCheckObjectConfig config) {
+        HashMap<Long, CheckResult> subMap = cacheMap.get(type);
+        if (Objects.isNull(subMap)) {
+            subMap = new HashMap<>();
+            cacheMap.put(type, subMap);
+        }
+        CheckResult resultToUpdate = subMap.get(typeId);
+        subMap.put(typeId, new CheckResult(status, resultDesc, LocalDateTime.now(), latency, config));
+    }
+
+    public Map<String, HashMap<Long, CheckResult>> getCacheMap() {
+        return Collections.unmodifiableMap(cacheMap);
+    }
+
+    @Getter
+    @AllArgsConstructor
+    public class CheckResult {
+
+        /**
+         * the status of a health check.
+         *
+         * @see HealthCheckStatus
+         */
+        @Setter
+        private HealthCheckStatus status;
+        /**
+         * if not passed, this field is used to store some description.
+         */
+        private String resultDesc;
+        /**
+         * the time this record is inserted into memory map.
+         */
+        private LocalDateTime createTime;
+        /**
+         * latency of a health check, for example ping latency.
+         */
+        private Long latencyMilliSeconds;
+
+        private HealthCheckObjectConfig config;
+    }
+}

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

@@ -0,0 +1,140 @@
+/*
+ * 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.health;
+
+import org.apache.eventmesh.dashboard.console.entity.health.HealthCheckResultEntity;
+import org.apache.eventmesh.dashboard.console.enums.health.HealthCheckStatus;
+import org.apache.eventmesh.dashboard.console.enums.health.HealthCheckType;
+import org.apache.eventmesh.dashboard.console.health.CheckResultCache.CheckResult;
+import org.apache.eventmesh.dashboard.console.health.callback.HealthCheckCallback;
+import org.apache.eventmesh.dashboard.console.health.check.AbstractHealthCheckService;
+import org.apache.eventmesh.dashboard.console.service.health.HealthDataService;
+
+import java.util.ArrayList;
+
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class HealthExecutor {
+
+    @Setter
+    private HealthDataService dataService;
+
+    /**
+     * memory cache is used to store real-time health check result.
+     */
+    @Getter
+    @Setter
+    private CheckResultCache memoryCache;
+
+    /**
+     * execute function is where health check services work.
+     *
+     * @param service The health check service to be executed.
+     */
+
+    public void execute(AbstractHealthCheckService service) {
+        final long startTime = System.currentTimeMillis();
+        //TODO: execute is called by a ScheduledThreadPoolExecutor,
+        // when called, it should check if current service should doCheck(check service check rate can be dynamically configured).
+        try {
+            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() {
+                @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
+                    );
+                }
+
+                @Override
+                public void onFail(Exception e) {
+                    //when the health check fails, the result is updated to the memory cache, passing in the exception message.
+                    log.error("Health check failed for reason: {}. Service config is {}", e, service.getConfig());
+                    memoryCache.update(service.getConfig().getHealthCheckResourceType(), service.getConfig().getInstanceId(),
+                        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,
+                e.getMessage(),
+                System.currentTimeMillis() - startTime);
+        }
+    }
+
+    /**
+     * this function should be called before any actual execute behaviour.<br> It will check the execution result of the last check cycle in the
+     * memory cache, set tasks that haven't finished status to time out and update the database.
+     */
+    public void startExecute() {
+        ArrayList<HealthCheckResultEntity> resultList = new ArrayList<>();
+        memoryCache.getCacheMap().forEach((type, subMap) -> {
+            subMap.forEach((instanceId, result) -> {
+                if (result.getStatus() == HealthCheckStatus.CHECKING) {
+                    result.setStatus(HealthCheckStatus.TIMEOUT);
+                }
+                addToResultList(result, resultList);
+            });
+        });
+        if (!resultList.isEmpty()) {
+            dataService.batchUpdateCheckResultByClusterIdAndTypeAndTypeId(resultList);
+        }
+    }
+
+    /**
+     * this function should be called after all actual execute behaviour.<br> It will insert the result of this check cycle into the database. At this
+     * point the status of the tasks may be CHECKING, they will be updated on the next startExecute.
+     */
+    public void endExecute() {
+        ArrayList<HealthCheckResultEntity> resultList = new ArrayList<>();
+        memoryCache.getCacheMap().forEach((type, subMap) -> {
+            subMap.forEach((instanceId, result) -> {
+                addToResultList(result, resultList);
+            });
+        });
+        dataService.batchInsertHealthCheckResult(resultList);
+    }
+
+    /**
+     * Helper function to add a CheckResult to the resultList.
+     *
+     * @param result     memory cached result object.
+     * @param resultList entity list to be inserted into the database.
+     */
+    private void addToResultList(CheckResult result, ArrayList<HealthCheckResultEntity> resultList) {
+        HealthCheckResultEntity newEntity = new HealthCheckResultEntity();
+        newEntity.setClusterId(result.getConfig().getClusterId());
+        newEntity.setType(HealthCheckType.toNumber(result.getConfig().getHealthCheckResourceType()));
+        newEntity.setTypeId(result.getConfig().getInstanceId());
+        newEntity.setResultDesc(result.getResultDesc());
+        newEntity.setStatus(result.getStatus().getNumber());
+
+        resultList.add(newEntity);
+    }
+
+}