[Fix][Server] Fixed Multiple metric are configured for a Job, resulti…

…ng in abnormal alarm information.
datavane · Feb 22, 2024 · 8056892 · 8056892
1 parent f31deb7
commit 8056892
Show file tree

Hide file tree

Showing 4 changed files with 32 additions and 18 deletions.
diff --git a/...server/src/main/java/io/datavines/server/dqc/coordinator/operator/JobResultValidator.java b/...server/src/main/java/io/datavines/server/dqc/coordinator/operator/JobResultValidator.java
@@ -128,27 +128,27 @@ private void sendErrorEmail(Long jobExecutionId){
         DataSource dataSource = jobExternalService.getDataSourceService().getDataSourceById(dataSourceId);
         String dataSourceName = dataSource.getName();
         String dataSourceType = dataSource.getType();
-        JobExecutionResult jobExecutionResult = jobExternalService.getJobExecutionResultByJobExecutionId(jobExecution.getId());
+        List<JobExecutionResult> errorJobExecutionResultList = jobExternalService.listErrorJobExecutionResultByJobExecutionId(jobExecution.getId());
         boolean isEn = !LanguageUtils.isZhContext();
-        if (jobExecutionResult != null) {
-            MetricExecutionResult metricExecutionResult = new MetricExecutionResult();
-            BeanUtils.copyProperties(jobExecutionResult, metricExecutionResult);
-            List<String> messages = new ArrayList<>();
-            messages.add((isEn ? "Job Name : ": "作业名称: ") + jobName);
-            messages.add(String.format((isEn ? "Datasource : %s [%s] : ": "数据源 : %s [%s]: ") ,dataSourceType.toUpperCase(), dataSourceName));
-            String title = buildAlertSubject(metricExecutionResult, isEn);
-            String content = buildAlertMessage(messages, metricExecutionResult, jobExecution.getEngineType(), isEn);
-            message.setSubject(title);
-            message.setMessage(content);
-
-            saveIssue(jobId, title, content);
-
-            Map<SlaSenderMessage, Set<SlaConfigMessage>> config = slaNotificationService.getSlasNotificationConfigurationByJobId(jobId);
-            if (config.isEmpty()){
-                return;
+        if (CollectionUtils.isNotEmpty(errorJobExecutionResultList)) {
+            for(JobExecutionResult errorJobExecutionResult : errorJobExecutionResultList){
+                MetricExecutionResult metricExecutionResult = new MetricExecutionResult();
+                BeanUtils.copyProperties(errorJobExecutionResult, metricExecutionResult);
+                List<String> messages = new ArrayList<>();
+                messages.add((isEn ? "Job Name : ": "作业名称: ") + jobName);
+                messages.add(String.format((isEn ? "Datasource : %s [%s] : ": "数据源 : %s [%s]: ") ,dataSourceType.toUpperCase(), dataSourceName));
+                String title = buildAlertSubject(metricExecutionResult, isEn);
+                String content = buildAlertMessage(messages, metricExecutionResult, jobExecution.getEngineType(), isEn);
+                message.setSubject(title);
+                message.setMessage(content);
+                saveIssue(jobId, title, content);
+                Map<SlaSenderMessage, Set<SlaConfigMessage>> config = slaNotificationService.getSlasNotificationConfigurationByJobId(jobId);
+                if (config.isEmpty()){
+                    return;
+                }
+                notificationClient.notify(message, config);
             }
 
-            notificationClient.notify(message, config);
         }
     }
 

diff --git a/...erver/src/main/java/io/datavines/server/repository/service/JobExecutionResultService.java b/...erver/src/main/java/io/datavines/server/repository/service/JobExecutionResultService.java
@@ -34,6 +34,8 @@ public interface JobExecutionResultService extends IService<JobExecutionResult>
 
     JobExecutionResult getByJobExecutionId(long jobExecutionId);
 
+    List<JobExecutionResult> listByErrorJobExecutionId(long jobExecutionId);
+
     List<JobExecutionResult> listByJobExecutionId(long jobExecutionId);
 
     JobExecutionResultVO getResultVOByJobExecutionId(long jobExecutionId);

diff --git a/.../main/java/io/datavines/server/repository/service/impl/JobExecutionResultServiceImpl.java b/.../main/java/io/datavines/server/repository/service/impl/JobExecutionResultServiceImpl.java
@@ -89,6 +89,14 @@ public List<JobExecutionResult> listByJobExecutionId(long jobExecutionId) {
                 .orderByDesc("update_time"));
     }
 
+    @Override
+    public List<JobExecutionResult> listByErrorJobExecutionId(long jobExecutionId) {
+        return baseMapper.selectList(new QueryWrapper<JobExecutionResult>()
+                .eq("job_execution_id", jobExecutionId)
+                .eq("state", DqJobExecutionState.FAILURE.getCode())
+                .orderByDesc("update_time"));
+    }
+
     @Override
     public JobExecutionResultVO getResultVOByJobExecutionId(long jobExecutionId) {
 

diff --git a/...-server/src/main/java/io/datavines/server/repository/service/impl/JobExternalService.java b/...-server/src/main/java/io/datavines/server/repository/service/impl/JobExternalService.java
@@ -172,6 +172,10 @@ public JobExecutionResult getJobExecutionResultByJobExecutionId(long jobExecutio
         return jobExecutionResultService.getByJobExecutionId(jobExecutionId);
     }
 
+    public List<JobExecutionResult> listErrorJobExecutionResultByJobExecutionId(long jobExecutionId) {
+        return jobExecutionResultService.listByErrorJobExecutionId(jobExecutionId);
+    }
+
     public List<JobExecutionResult> listJobExecutionResultByJobExecutionId(long jobExecutionId) {
         return jobExecutionResultService.listByJobExecutionId(jobExecutionId);
     }