Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Fix][Server] Fix the error when execute job through restapi #395

Merged
merged 1 commit into from
May 4, 2024
Merged

[Fix][Server] Fix the error when execute job through restapi #395

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion datavines-runner/src/main/java/io/datavines/runner/JobExecuteBootstrap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ public static void main(String[] args) {
JobExecutionRequest jobExecutionRequest = new JobExecutionRequest();
jobExecutionRequest.setJobExecutionName(submitJob.getName());
jobExecutionRequest.setJobExecutionId(id);
jobExecutionRequest.setJobExecutionUniqueId(submitJob.getName()+"_"+ id);
jobExecutionRequest.setJobExecutionUniqueId(submitJob.getName() + "_" + id);
jobExecutionRequest.setApplicationParameter(JSONUtils.toJsonString(qualityConfig));
jobExecutionRequest.setEngineType(submitJob.getEngineType());
jobExecutionRequest.setEngineParameter(JSONUtils.toJsonString(submitJob.getEngineParameter()));
Expand Down
103 changes: 60 additions & 43 deletions ...server/src/main/java/io/datavines/server/dqc/coordinator/operator/JobResultValidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,10 @@
package io.datavines.server.dqc.coordinator.operator;

import io.datavines.common.entity.JobExecutionRequest;
import io.datavines.common.enums.ExecutionStatus;
import io.datavines.common.enums.OperatorType;
import io.datavines.common.utils.JSONUtils;
import io.datavines.common.utils.ParameterUtils;
import io.datavines.common.utils.StringUtils;
import io.datavines.core.utils.LanguageUtils;
import io.datavines.metric.api.*;
import io.datavines.notification.api.entity.SlaConfigMessage;
Expand Down Expand Up @@ -73,25 +73,29 @@ public void operateDqExecuteResult(JobExecutionRequest jobExecutionRequest) {
List<JobExecutionResult> jobExecutionResultList =
jobExternalService.listJobExecutionResultByJobExecutionId(jobExecutionRequest.getJobExecutionId());

if (CollectionUtils.isNotEmpty(jobExecutionResultList)) {
boolean isSuccess = true;
Long jobExecutionId = 0L;
for (JobExecutionResult jobExecutionResult : jobExecutionResultList) {
if (jobExecutionResult != null) {
jobExecutionId = jobExecutionResult.getJobExecutionId();
// 判断期望值是否为空，如果为空并且不是固定值类型，则将期望值设置为实际值
if (jobExecutionResult.getExpectedValue() == null && !FIX_VALUE.equalsIgnoreCase(jobExecutionResult.getExpectedType())) {
jobExecutionResult.setExpectedValue(jobExecutionResult.getActualValue());
jobExternalService.getJobExecutionResultService().updateById(jobExecutionResult);
}
//check the result ,if result is failure do some operator by failure strategy
isSuccess &= checkDqExecuteResult(jobExecutionResult);
}
if (CollectionUtils.isEmpty(jobExecutionResultList)) {
return;
}

boolean isSuccess = true;
Long jobExecutionId = 0L;
for (JobExecutionResult jobExecutionResult : jobExecutionResultList) {
if (jobExecutionResult == null) {
continue;
}

if (!isSuccess) {
sendErrorEmail(jobExecutionId);
jobExecutionId = jobExecutionResult.getJobExecutionId();
// 判断期望值是否为空，如果为空并且不是固定值类型，则将期望值设置为实际值
if (jobExecutionResult.getExpectedValue() == null && !FIX_VALUE.equalsIgnoreCase(jobExecutionResult.getExpectedType())) {
jobExecutionResult.setExpectedValue(jobExecutionResult.getActualValue());
jobExternalService.getJobExecutionResultService().updateById(jobExecutionResult);
}
//check the result ,if result is failure do some operator by failure strategy
isSuccess &= checkDqExecuteResult(jobExecutionResult);
}

if (!isSuccess) {
sendErrorEmail(jobExecutionId);
}
}

Expand Down Expand Up @@ -120,44 +124,57 @@ private void sendErrorEmail(Long jobExecutionId) {
SlaNotificationMessage message = new SlaNotificationMessage();
JobExecution jobExecution = jobExternalService.getJobExecutionById(jobExecutionId);
Long jobId = jobExecution.getJobId();
JobService jobService = jobExternalService.getJobService();
Job job = jobService.getById(jobId);
String jobName = job.getName();
Long dataSourceId = job.getDataSourceId();
DataSource dataSource = jobExternalService.getDataSourceService().getDataSourceById(dataSourceId);
String dataSourceName = dataSource.getName();
String dataSourceType = dataSource.getType();
String jobName;
String dataSourceName = null;
String dataSourceType = null;
if (jobId == -1L) {
jobName = jobExecution.getName();
} else {
JobService jobService = jobExternalService.getJobService();
Job job = jobService.getById(jobId);
jobName = job.getName();
Long dataSourceId = job.getDataSourceId();
DataSource dataSource = jobExternalService.getDataSourceService().getDataSourceById(dataSourceId);
dataSourceName = dataSource.getName();
dataSourceType = dataSource.getType();
}

List<JobExecutionResult> errorJobExecutionResultList = jobExternalService.listErrorJobExecutionResultByJobExecutionId(jobExecution.getId());
if (CollectionUtils.isEmpty(errorJobExecutionResultList)) {
return;
}

boolean isEn = !LanguageUtils.isZhContext();
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);
for (JobExecutionResult errorJobExecutionResult : errorJobExecutionResultList) {
MetricExecutionResult metricExecutionResult = new MetricExecutionResult();
BeanUtils.copyProperties(errorJobExecutionResult, metricExecutionResult);
List<String> messages = new ArrayList<>();
messages.add((isEn ? "Job Name : ": "作业名称: ") + jobName);
if (StringUtils.isNotEmpty(dataSourceType)) {
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);
}
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);
}

} catch (Exception e) {
log.error("send email error: ", e);
}
}

private String buildAlertMessage(List<String> messages, MetricExecutionResult metricExecutionResult, String engineType, boolean isEn) {
Map<String,String> parameters = new HashMap<>();
parameters.put("actual_value", metricExecutionResult.getActualValue()+"");
parameters.put("expected_value", metricExecutionResult.getExpectedValue()+"");
parameters.put("threshold", metricExecutionResult.getThreshold()+"");
parameters.put("actual_value", String.valueOf(metricExecutionResult.getActualValue()));
parameters.put("expected_value", String.valueOf(metricExecutionResult.getExpectedValue()));
parameters.put("threshold", String.valueOf(metricExecutionResult.getThreshold()));
parameters.put("operator",OperatorType.of(metricExecutionResult.getOperator()).getSymbol());

SqlMetric sqlMetric = PluginLoader.getPluginLoader(SqlMetric.class).getOrCreatePlugin(metricExecutionResult.getMetricName());
Expand Down
Loading