fix alert some npe

gaoyan1998 · Dec 4, 2023 · 76be35d · 76be35d
1 parent a1e1f5d
commit 76be35d
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 24 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/data/model/ext/JobAlertData.java b/dinky-admin/src/main/java/org/dinky/data/model/ext/JobAlertData.java
@@ -31,6 +31,9 @@
 import org.dinky.job.JobConfig;
 import org.dinky.utils.TimeUtil;
 
+import java.time.LocalDateTime;
+import java.util.Optional;
+
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 import cn.hutool.core.text.StrFormatter;
@@ -39,9 +42,6 @@
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
-import java.time.LocalDateTime;
-import java.util.Optional;
-
 @Data
 @Builder
 @NoArgsConstructor
@@ -188,7 +188,9 @@ public static JobAlertData buildData(JobInfoDetail jobInfoDetail) {
         } else if (exceptions != null && ExceptionRule.isException(id, exceptions)) {
             // The error message is too long to send an alarm,
             // and only the first line of abnormal information is used
-            String err = Optional.ofNullable(exceptions.getRootException()).orElse("dinky didn't get any ERROR!").split("\n")[0];
+            String err = Optional.ofNullable(exceptions.getRootException())
+                    .orElse("dinky didn't get any ERROR!")
+                    .split("\n")[0];
             if (err.length() > 100) {
                 err = err.substring(0, 100) + "...";
             }

diff --git a/dinky-admin/src/main/java/org/dinky/job/FlinkJobTask.java b/dinky-admin/src/main/java/org/dinky/job/FlinkJobTask.java
@@ -79,17 +79,12 @@ public DaemonTaskConfig getConfig() {
      */
     @Override
     public boolean dealTask() {
-        try {
-            volatilityBalance();
-
-            boolean isDone = JobRefreshHandler.refreshJob(jobInfoDetail, isNeedSave());
-            JobAlertHandler.getInstance().check(jobInfoDetail);
-            JobMetricsHandler.writeFlinkMetrics(jobInfoDetail);
-            return isDone;
-        }catch (Exception e){
-            log.error("Deal Flink Job daemon refesh failed,", e);
-            return true;
-        }
+        volatilityBalance();
+
+        boolean isDone = JobRefreshHandler.refreshJob(jobInfoDetail, isNeedSave());
+        JobAlertHandler.getInstance().check(jobInfoDetail);
+        JobMetricsHandler.writeFlinkMetrics(jobInfoDetail);
+        return isDone;
     }
 
     /**

diff --git a/dinky-admin/src/main/java/org/dinky/job/handler/JobAlertHandler.java b/dinky-admin/src/main/java/org/dinky/job/handler/JobAlertHandler.java
@@ -28,6 +28,7 @@
 import org.dinky.daemon.pool.FlinkJobThreadPool;
 import org.dinky.data.dto.AlertRuleDTO;
 import org.dinky.data.dto.TaskDTO;
+import org.dinky.data.enums.JobLifeCycle;
 import org.dinky.data.enums.Status;
 import org.dinky.data.exception.DinkyException;
 import org.dinky.data.model.alert.AlertGroup;
@@ -46,6 +47,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 import org.jeasy.rules.api.Facts;
@@ -118,8 +120,8 @@ public JobAlertHandler() {
     public void check(JobInfoDetail jobInfoDetail) {
         Facts ruleFacts = new Facts();
         JobAlertData jobAlertData = JobAlertData.buildData(jobInfoDetail);
-        JsonUtils.toMap(jobAlertData).forEach((k,v)->{
-            if (v == null){
+        JsonUtils.toMap(jobAlertData).forEach((k, v) -> {
+            if (v == null) {
                 throw new DinkyException(StrFormatter.format(
                         "When deal alert job data, the key [{}] value is null, its maybe dinky bug,please report", k));
             }
@@ -178,6 +180,10 @@ private Rule buildRule(AlertRuleDTO alertRuleDTO) {
      */
     private void executeAlertAction(Facts facts, AlertRuleDTO alertRuleDTO) {
         TaskDTO task = taskService.getTaskInfoById(facts.get(JobAlertRuleOptions.FIELD_TASK_ID));
+        if (!Objects.equals(task.getStep(), JobLifeCycle.PUBLISH.getValue())) {
+            // Only publish job can be alerted
+            return;
+        }
         Map<String, Object> dataModel = new HashMap<>(facts.asMap());
         dataModel.put(JobAlertRuleOptions.OPTIONS_JOB_ALERT_RULE, alertRuleDTO);
         String alertContent;

diff --git a/dinky-admin/src/main/java/org/dinky/job/handler/JobMetricsHandler.java b/dinky-admin/src/main/java/org/dinky/job/handler/JobMetricsHandler.java
@@ -39,10 +39,12 @@
 import cn.hutool.json.JSONArray;
 import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
+import lombok.extern.slf4j.Slf4j;
 
 /**
  * The Job Metrics Handler class is used to process operations related to job metrics。
  */
+@Slf4j
 public class JobMetricsHandler {
 
     /**
@@ -63,13 +65,17 @@ public static void writeFlinkMetrics(JobInfoDetail jobInfoDetail) {
                         () -> fetchFlinkMetrics(e.getKey(), e.getValue(), jobManagerUrls, jobId)))
                 .toArray(CompletableFuture[]::new);
         // Wait for all Completable Future executions to finish
-        AsyncUtil.waitAll(array);
-        MetricsVO metricsVO = new MetricsVO();
-        metricsVO.setContent(customMetricsList);
-        metricsVO.setHeartTime(LocalDateTime.now());
-        metricsVO.setModel(jobId);
-        metricsVO.setDate(TimeUtil.nowStr("yyyy-MM-dd"));
-        MetricsContextHolder.getInstances().sendAsync(metricsVO.getModel(), metricsVO);
+        try {
+            AsyncUtil.waitAll(array);
+            MetricsVO metricsVO = new MetricsVO();
+            metricsVO.setContent(customMetricsList);
+            metricsVO.setHeartTime(LocalDateTime.now());
+            metricsVO.setModel(jobId);
+            metricsVO.setDate(TimeUtil.nowStr("yyyy-MM-dd"));
+            MetricsContextHolder.getInstances().sendAsync(metricsVO.getModel(), metricsVO);
+        } catch (Exception e) {
+            log.error("Get and save Flink metrics error", e);
+        }
     }
 
     /**