tech10001
/
jianmu-ci-server
forked from jianmu-dev/jianmu-ci-server
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: HA部署终止流程，或任务在WAITTING状态下终止流程时，无法通知worker终止任务 [https://gitee.com/jianmu-dev/jianmu/issues/I6590R]

This commit is contained in:
daihw 2023-01-09 16:17:41 +08:00
parent 7b1bf67d1b
commit a84b6413d8
9 changed files with 144 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
api/src/main/java/dev/jianmu/api/controller/WorkerController.java
View File

@ -11,6 +11,7 @@ import dev.jianmu.infrastructure.GlobalProperties;
import dev.jianmu.infrastructure.storage.StorageService;
import dev.jianmu.infrastructure.worker.DeferredResultService;
import dev.jianmu.infrastructure.worker.unit.Unit;
import dev.jianmu.task.aggregate.InstanceStatus;
import dev.jianmu.worker.aggregate.Worker;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
@ -187,7 +188,16 @@ public class WorkerController {
@Parameter(name = "X-Jianmu-Token", in = ParameterIn.HEADER, description = "认证token")
})
public DeferredResult<ResponseEntity<?>> watchTasks(@PathVariable String workerId, @PathVariable("businessId") String businessId) {
return this.deferredResultService.newWatchDeferredResult(workerId, businessId);
var exist = this.deferredResultService.existWatchDeferredResult(workerId, businessId);
var deferredResult = this.deferredResultService.newWatchDeferredResult(workerId, businessId);
if (!exist) {
var taskInstance = this.taskInstanceApplication.findByBusinessIdAndMaxSerialNo(businessId)
.orElseThrow(() -> new RuntimeException("未找到任务实例，businessId: " + businessId));
if (taskInstance.getStatus() == InstanceStatus.EXECUTION_FAILED) {
deferredResult.setResult(ResponseEntity.status(HttpStatus.OK).body(businessId));
}
}
return deferredResult;
}
@Retryable(

29
api/src/main/java/dev/jianmu/api/eventhandler/WatchDeferredResultTerminateEventHandler.java Normal file
View File

@ -0,0 +1,29 @@
package dev.jianmu.api.eventhandler;
import dev.jianmu.event.Subscriber;
import dev.jianmu.event.impl.WatchDeferredResultTerminateEvent;
import dev.jianmu.infrastructure.worker.DeferredResultService;
import lombok.extern.slf4j.Slf4j;
/**
* @class WatchDeferredResultTerminateEventHandler
* @description WatchDeferredResultTerminateEventHandler
* @author Daihw
* @create 2023/1/9 3:10 下午
*/
@Slf4j
public class WatchDeferredResultTerminateEventHandler implements Subscriber<WatchDeferredResultTerminateEvent> {
private final DeferredResultService deferredResultService;
public WatchDeferredResultTerminateEventHandler(DeferredResultService deferredResultService) {
this.deferredResultService = deferredResultService;
}
@Override
public void subscribe(WatchDeferredResultTerminateEvent event) {
log.info("Get WatchDeferredResultTerminateEvent here -------------------------");
log.info(event.toString());
this.deferredResultService.terminateDeferredResult(event.getWorkerId(), event.getBusinessId());
log.info("-----------------------------------------------------");
}
}

30
api/src/main/java/dev/jianmu/api/eventhandler/local/WatchDeferredResultTerminateEventHandlerImpl.java Normal file
View File

@ -0,0 +1,30 @@
package dev.jianmu.api.eventhandler.local;
import dev.jianmu.api.eventhandler.WatchDeferredResultTerminateEventHandler;
import dev.jianmu.event.impl.WatchDeferredResultTerminateEvent;
import dev.jianmu.infrastructure.worker.DeferredResultService;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
/**
* @author Daihw
* @class WatchDeferredResultTerminateEventHandlerImpl
* @description WatchDeferredResultTerminateEventHandlerImpl
* @create 2023/1/9 3:14 下午
*/
@Component
@ConditionalOnProperty(prefix = "jianmu.event", name = "type", havingValue = "local", matchIfMissing = true)
public class WatchDeferredResultTerminateEventHandlerImpl extends WatchDeferredResultTerminateEventHandler {
public WatchDeferredResultTerminateEventHandlerImpl(DeferredResultService deferredResultService) {
super(deferredResultService);
}
@Async
@EventListener
@Override
public void subscribe(WatchDeferredResultTerminateEvent event) {
super.subscribe(event);
}
}

29
api/src/main/java/dev/jianmu/api/eventhandler/redis/WatchDeferredResultTerminateEventHandlerImpl.java Normal file
View File

@ -0,0 +1,29 @@
package dev.jianmu.api.eventhandler.redis;
import dev.jianmu.api.eventhandler.WatchDeferredResultTerminateEventHandler;
import dev.jianmu.event.impl.WatchDeferredResultTerminateEvent;
import dev.jianmu.infrastructure.redis.RedisSubscriber;
import dev.jianmu.infrastructure.redis.annotation.RedisEventListener;
import dev.jianmu.infrastructure.worker.DeferredResultService;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* @author Daihw
* @class WatchDeferredResultTerminateEventHandlerImpl
* @description WatchDeferredResultTerminateEventHandlerImpl
* @create 2023/1/9 3:14 下午
*/
@Component
@ConditionalOnProperty(prefix = "jianmu.event", name = "type", havingValue = "redis")
public class WatchDeferredResultTerminateEventHandlerImpl extends WatchDeferredResultTerminateEventHandler implements RedisSubscriber<WatchDeferredResultTerminateEvent> {
public WatchDeferredResultTerminateEventHandlerImpl(DeferredResultService deferredResultService) {
super(deferredResultService);
}
@RedisEventListener
@Override
public void subscribe(WatchDeferredResultTerminateEvent event) {
super.subscribe(event);
}
}

7
application/src/main/java/dev/jianmu/application/service/internal/AsyncTaskInstanceInternalApplication.java
View File

@ -3,9 +3,6 @@ package dev.jianmu.application.service.internal;
import dev.jianmu.application.command.AsyncTaskActivatingCmd;
import dev.jianmu.application.exception.DataNotFoundException;
import dev.jianmu.infrastructure.exception.DBException;
import dev.jianmu.task.aggregate.NodeInfo;
import dev.jianmu.task.aggregate.TaskInstance;
import dev.jianmu.task.repository.TaskInstanceRepository;
import dev.jianmu.workflow.aggregate.process.TaskStatus;
import dev.jianmu.workflow.repository.AsyncTaskInstanceRepository;
import dev.jianmu.workflow.repository.WorkflowInstanceRepository;
@ -14,8 +11,6 @@ import org.slf4j.MDC;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.UUID;
/**
* @author Ethan Liu
* @class AsyncTaskInstanceInternalApplication
@ -75,7 +70,7 @@ public class AsyncTaskInstanceInternalApplication {
asyncTaskInstances.stream()
.filter(asyncTaskInstance -> asyncTaskInstance.getStatus() == TaskStatus.WAITING)
.forEach(asyncTaskInstance -> {
asyncTaskInstance.fail();
asyncTaskInstance.terminate();
log.info("终止待执行任务: " + asyncTaskInstance.getAsyncTaskRef());
this.asyncTaskInstanceRepository.updateById(asyncTaskInstance);
});

18
application/src/main/java/dev/jianmu/application/service/internal/TaskInstanceInternalApplication.java
View File

@ -8,8 +8,9 @@ import dev.jianmu.application.exception.DataNotFoundException;
import dev.jianmu.application.query.NodeDef;
import dev.jianmu.application.query.NodeDefApi;
import dev.jianmu.el.ElContext;
import dev.jianmu.event.Publisher;
import dev.jianmu.event.impl.WatchDeferredResultTerminateEvent;
import dev.jianmu.infrastructure.storage.MonitoringFileService;
import dev.jianmu.infrastructure.worker.DeferredResultService;
import dev.jianmu.node.definition.aggregate.NodeParameter;
import dev.jianmu.task.aggregate.InstanceParameter;
import dev.jianmu.task.aggregate.InstanceStatus;
@ -56,8 +57,8 @@ public class TaskInstanceInternalApplication {
private final ExpressionLanguage expressionLanguage;
private final WorkflowInstanceRepository workflowInstanceRepository;
private final MonitoringFileService monitoringFileService;
private final DeferredResultService deferredResultService;
private final AsyncTaskInstanceRepository asyncTaskInstanceRepository;
private final Publisher publisher;
public TaskInstanceInternalApplication(
TaskInstanceRepository taskInstanceRepository,
@ -71,8 +72,8 @@ public class TaskInstanceInternalApplication {
ExpressionLanguage expressionLanguage,
WorkflowInstanceRepository workflowInstanceRepository,
MonitoringFileService monitoringFileService,
DeferredResultService deferredResultService,
AsyncTaskInstanceRepository asyncTaskInstanceRepository
AsyncTaskInstanceRepository asyncTaskInstanceRepository,
Publisher publisher
) {
this.taskInstanceRepository = taskInstanceRepository;
this.workflowRepository = workflowRepository;
@ -85,8 +86,8 @@ public class TaskInstanceInternalApplication {
this.expressionLanguage = expressionLanguage;
this.workflowInstanceRepository = workflowInstanceRepository;
this.monitoringFileService = monitoringFileService;
this.deferredResultService = deferredResultService;
this.asyncTaskInstanceRepository = asyncTaskInstanceRepository;
this.publisher = publisher;
}
public List<TaskInstance> findRunningTask() {
@ -228,7 +229,10 @@ public class TaskInstanceInternalApplication {
public void terminate(String asyncTaskInstanceId) {
var taskInstance = this.taskInstanceRepository.findByBusinessIdAndMaxSerialNo(asyncTaskInstanceId)
.orElseThrow(() -> new DataNotFoundException("未找到该任务实例"));
this.deferredResultService.terminateDeferredResult(taskInstance.getWorkerId(), taskInstance.getBusinessId());
this.publisher.publish(WatchDeferredResultTerminateEvent.builder()
.workerId(taskInstance.getWorkerId())
.businessId(taskInstance.getBusinessId())
.build());
}
@Transactional
@ -284,7 +288,7 @@ public class TaskInstanceInternalApplication {
.orElseThrow(() -> new DataNotFoundException("未找到该任务实例"));
workflowInstance.terminateInStart();
this.workflowInstanceRepository.save(workflowInstance);
}else {
} else {
log.error("清除Volume失败");
this.monitoringFileService.clearCallbackByLogId(taskInstance.getTriggerId());
}

4
application/src/main/java/dev/jianmu/application/service/internal/WorkflowInstanceInternalApplication.java
View File

@ -197,6 +197,10 @@ public class WorkflowInstanceInternalApplication {
.orElseThrow(() -> new DataNotFoundException("未找到项目最后执行记录"));
// 终止流程
MDC.put("triggerId", workflowInstance.getTriggerId());
if (workflowInstance.getStatus() == ProcessStatus.TERMINATED) {
log.warn("流程实例已终止，无需终止");
return;
}
workflowInstance.terminate();
projectLastExecution.end(workflowInstance.getId(), workflowInstance.getSerialNo(), workflowInstance.getStatus().name(), workflowInstance.getStartTime(), workflowInstance.getEndTime());
this.workflowInstanceRepository.save(workflowInstance);

19
event/src/main/java/dev/jianmu/event/impl/WatchDeferredResultTerminateEvent.java Normal file
View File

@ -0,0 +1,19 @@
package dev.jianmu.event.impl;
import lombok.*;
/**
* @class WatchDeferredResultTerminateEvent
* @description WatchDeferredResult终止事件
* @author Daihw
* @create 2023/1/9 3:23 下午
*/
@Getter
@Builder
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class WatchDeferredResultTerminateEvent extends BaseEvent {
private String workerId;
private String businessId;
}

10
infrastructure/src/main/java/dev/jianmu/infrastructure/worker/DeferredResultService.java
View File

@ -51,6 +51,16 @@ public class DeferredResultService {
this.pullDeferredResults.remove(workerId);
}
/**
* 是否存在监视任务
*/
public boolean existWatchDeferredResult(String workerId, String businessId) {
if (!this.watchDeferredResults.containsKey(workerId)) {
return false;
}
return this.watchDeferredResults.get(workerId).containsKey(businessId);
}
/**
* 创建获取终止任务的DeferredResult
*