Browse Source
* add workflow executing data query * fix sonar check for interrupted3.0.0/version-upgrade
caishunfeng
2 years ago
19 changed files with 673 additions and 39 deletions
@ -0,0 +1,51 @@
|
||||
/* |
||||
* 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.dolphinscheduler.server.master.controller; |
||||
|
||||
import org.apache.dolphinscheduler.remote.dto.WorkflowExecuteDto; |
||||
import org.apache.dolphinscheduler.server.master.service.ExecutingService; |
||||
|
||||
import java.util.Optional; |
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.http.HttpStatus; |
||||
import org.springframework.web.bind.annotation.GetMapping; |
||||
import org.springframework.web.bind.annotation.RequestMapping; |
||||
import org.springframework.web.bind.annotation.RequestParam; |
||||
import org.springframework.web.bind.annotation.ResponseStatus; |
||||
import org.springframework.web.bind.annotation.RestController; |
||||
|
||||
@RestController |
||||
@RequestMapping("/workflow/execute") |
||||
public class WorkflowExecuteController { |
||||
|
||||
@Autowired |
||||
private ExecutingService executingService; |
||||
|
||||
/** |
||||
* query workflow execute data in memory |
||||
* @param processInstanceId |
||||
* @return |
||||
*/ |
||||
@GetMapping("") |
||||
@ResponseStatus(HttpStatus.OK) |
||||
public WorkflowExecuteDto queryExecuteData(@RequestParam("id") int processInstanceId) { |
||||
Optional<WorkflowExecuteDto> workflowExecuteDtoOptional = executingService.queryWorkflowExecutingData(processInstanceId); |
||||
return workflowExecuteDtoOptional.orElse(null); |
||||
} |
||||
} |
||||
@ -0,0 +1,65 @@
|
||||
/* |
||||
* 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.dolphinscheduler.server.master.processor; |
||||
|
||||
import org.apache.dolphinscheduler.common.utils.JSONUtils; |
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.command.WorkflowExecutingDataRequestCommand; |
||||
import org.apache.dolphinscheduler.remote.command.WorkflowExecutingDataResponseCommand; |
||||
import org.apache.dolphinscheduler.remote.dto.WorkflowExecuteDto; |
||||
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor; |
||||
import org.apache.dolphinscheduler.server.master.service.ExecutingService; |
||||
|
||||
import java.util.Optional; |
||||
|
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.stereotype.Component; |
||||
|
||||
import com.google.common.base.Preconditions; |
||||
|
||||
import io.netty.channel.Channel; |
||||
|
||||
/** |
||||
* workflow executing data process from api/master |
||||
*/ |
||||
@Component |
||||
public class WorkflowExecutingDataRequestProcessor implements NettyRequestProcessor { |
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(WorkflowExecutingDataRequestProcessor.class); |
||||
|
||||
@Autowired |
||||
private ExecutingService executingService; |
||||
|
||||
@Override |
||||
public void process(Channel channel, Command command) { |
||||
Preconditions.checkArgument(CommandType.WORKFLOW_EXECUTING_DATA_REQUEST == command.getType(), String.format("invalid command type: %s", command.getType())); |
||||
|
||||
WorkflowExecutingDataRequestCommand requestCommand = JSONUtils.parseObject(command.getBody(), WorkflowExecutingDataRequestCommand.class); |
||||
|
||||
logger.info("received command, processInstanceId:{}", requestCommand.getProcessInstanceId()); |
||||
|
||||
Optional<WorkflowExecuteDto> workflowExecuteDtoOptional = executingService.queryWorkflowExecutingData(requestCommand.getProcessInstanceId()); |
||||
|
||||
WorkflowExecutingDataResponseCommand responseCommand = new WorkflowExecutingDataResponseCommand(); |
||||
workflowExecuteDtoOptional.ifPresent(responseCommand::setWorkflowExecuteDto); |
||||
channel.writeAndFlush(responseCommand.convert2Command(command.getOpaque())); |
||||
} |
||||
} |
||||
@ -0,0 +1,75 @@
|
||||
/* |
||||
* 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.dolphinscheduler.server.master.service; |
||||
|
||||
import org.apache.dolphinscheduler.dao.entity.TaskInstance; |
||||
import org.apache.dolphinscheduler.remote.dto.TaskInstanceExecuteDto; |
||||
import org.apache.dolphinscheduler.remote.dto.WorkflowExecuteDto; |
||||
import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager; |
||||
import org.apache.dolphinscheduler.server.master.controller.WorkflowExecuteController; |
||||
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable; |
||||
|
||||
import org.apache.commons.beanutils.BeanUtils; |
||||
import org.apache.commons.collections4.CollectionUtils; |
||||
import org.apache.commons.compress.utils.Lists; |
||||
|
||||
import java.lang.reflect.InvocationTargetException; |
||||
import java.util.List; |
||||
import java.util.Optional; |
||||
|
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.stereotype.Component; |
||||
|
||||
/** |
||||
* executing service, to query executing data from memory, such workflow instance |
||||
*/ |
||||
@Component |
||||
public class ExecutingService { |
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(WorkflowExecuteController.class); |
||||
|
||||
@Autowired |
||||
private ProcessInstanceExecCacheManager processInstanceExecCacheManager; |
||||
|
||||
public Optional<WorkflowExecuteDto> queryWorkflowExecutingData(Integer processInstanceId) { |
||||
WorkflowExecuteRunnable workflowExecuteRunnable = processInstanceExecCacheManager.getByProcessInstanceId(processInstanceId); |
||||
if (workflowExecuteRunnable == null) { |
||||
logger.info("workflow execute data not found, maybe it has finished, workflow id:{}", processInstanceId); |
||||
return Optional.empty(); |
||||
} |
||||
try { |
||||
WorkflowExecuteDto workflowExecuteDto = new WorkflowExecuteDto(); |
||||
BeanUtils.copyProperties(workflowExecuteDto, workflowExecuteRunnable.getProcessInstance()); |
||||
List<TaskInstanceExecuteDto> taskInstanceList = Lists.newArrayList(); |
||||
if (CollectionUtils.isNotEmpty(workflowExecuteRunnable.getAllTaskInstances())) { |
||||
for (TaskInstance taskInstance : workflowExecuteRunnable.getAllTaskInstances()) { |
||||
TaskInstanceExecuteDto taskInstanceExecuteDto = new TaskInstanceExecuteDto(); |
||||
BeanUtils.copyProperties(taskInstanceExecuteDto, taskInstance); |
||||
taskInstanceList.add(taskInstanceExecuteDto); |
||||
} |
||||
} |
||||
workflowExecuteDto.setTaskInstances(taskInstanceList); |
||||
return Optional.of(workflowExecuteDto); |
||||
} catch (IllegalAccessException | InvocationTargetException e) { |
||||
logger.error("query workflow execute data fail, workflow id:{}", processInstanceId, e); |
||||
} |
||||
return Optional.empty(); |
||||
} |
||||
} |
||||
@ -0,0 +1,50 @@
|
||||
/* |
||||
* 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.dolphinscheduler.remote.command; |
||||
|
||||
import org.apache.dolphinscheduler.common.utils.JSONUtils; |
||||
|
||||
import java.io.Serializable; |
||||
|
||||
import lombok.AllArgsConstructor; |
||||
import lombok.Data; |
||||
import lombok.NoArgsConstructor; |
||||
|
||||
/** |
||||
* workflow executing data request, from api to master |
||||
*/ |
||||
@Data |
||||
@NoArgsConstructor |
||||
@AllArgsConstructor |
||||
public class WorkflowExecutingDataRequestCommand implements Serializable { |
||||
|
||||
private Integer processInstanceId; |
||||
|
||||
/** |
||||
* package request command |
||||
* |
||||
* @return command |
||||
*/ |
||||
public Command convert2Command() { |
||||
Command command = new Command(); |
||||
command.setType(CommandType.WORKFLOW_EXECUTING_DATA_REQUEST); |
||||
byte[] body = JSONUtils.toJsonByteArray(this); |
||||
command.setBody(body); |
||||
return command; |
||||
} |
||||
} |
||||
@ -0,0 +1,51 @@
|
||||
/* |
||||
* 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.dolphinscheduler.remote.command; |
||||
|
||||
import org.apache.dolphinscheduler.common.utils.JSONUtils; |
||||
import org.apache.dolphinscheduler.remote.dto.WorkflowExecuteDto; |
||||
|
||||
import java.io.Serializable; |
||||
|
||||
import lombok.AllArgsConstructor; |
||||
import lombok.Data; |
||||
import lombok.NoArgsConstructor; |
||||
|
||||
/** |
||||
* workflow executing data response, from master to api |
||||
*/ |
||||
@Data |
||||
@NoArgsConstructor |
||||
@AllArgsConstructor |
||||
public class WorkflowExecutingDataResponseCommand implements Serializable { |
||||
|
||||
private WorkflowExecuteDto workflowExecuteDto; |
||||
|
||||
/** |
||||
* package request command |
||||
* |
||||
* @return command |
||||
*/ |
||||
public Command convert2Command(long opaque) { |
||||
Command command = new Command(opaque); |
||||
command.setType(CommandType.WORKFLOW_EXECUTING_DATA_RESPONSE); |
||||
byte[] body = JSONUtils.toJsonByteArray(this); |
||||
command.setBody(body); |
||||
return command; |
||||
} |
||||
} |
||||
@ -0,0 +1,109 @@
|
||||
/* |
||||
* 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.dolphinscheduler.remote.dto; |
||||
|
||||
import org.apache.dolphinscheduler.common.enums.Flag; |
||||
import org.apache.dolphinscheduler.common.enums.Priority; |
||||
import org.apache.dolphinscheduler.plugin.task.api.enums.ExecutionStatus; |
||||
|
||||
import java.util.Date; |
||||
import java.util.Map; |
||||
|
||||
import lombok.Data; |
||||
|
||||
@Data |
||||
public class TaskInstanceExecuteDto { |
||||
|
||||
private int id; |
||||
|
||||
private String name; |
||||
|
||||
private String taskType; |
||||
|
||||
private int processInstanceId; |
||||
|
||||
private long taskCode; |
||||
|
||||
private int taskDefinitionVersion; |
||||
|
||||
private String processInstanceName; |
||||
|
||||
private int taskGroupPriority; |
||||
|
||||
private ExecutionStatus state; |
||||
|
||||
private Date firstSubmitTime; |
||||
|
||||
private Date submitTime; |
||||
|
||||
private Date startTime; |
||||
|
||||
private Date endTime; |
||||
|
||||
private String host; |
||||
|
||||
private String executePath; |
||||
|
||||
private String logPath; |
||||
|
||||
private int retryTimes; |
||||
|
||||
private Flag alertFlag; |
||||
|
||||
private int pid; |
||||
|
||||
private String appLink; |
||||
|
||||
private Flag flag; |
||||
|
||||
private String duration; |
||||
|
||||
private int maxRetryTimes; |
||||
|
||||
private int retryInterval; |
||||
|
||||
private Priority taskInstancePriority; |
||||
|
||||
private Priority processInstancePriority; |
||||
|
||||
private String workerGroup; |
||||
|
||||
private Long environmentCode; |
||||
|
||||
private String environmentConfig; |
||||
|
||||
private int executorId; |
||||
|
||||
private String varPool; |
||||
|
||||
private String executorName; |
||||
|
||||
private Map<String, String> resources; |
||||
|
||||
private int delayTime; |
||||
|
||||
private String taskParams; |
||||
|
||||
private int dryRun; |
||||
|
||||
private int taskGroupId; |
||||
|
||||
private Integer cpuQuota; |
||||
|
||||
private Integer memoryMax; |
||||
} |
||||
@ -0,0 +1,154 @@
|
||||
/* |
||||
* 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.dolphinscheduler.remote.dto; |
||||
|
||||
import org.apache.dolphinscheduler.common.enums.CommandType; |
||||
import org.apache.dolphinscheduler.common.enums.FailureStrategy; |
||||
import org.apache.dolphinscheduler.common.enums.Flag; |
||||
import org.apache.dolphinscheduler.common.enums.Priority; |
||||
import org.apache.dolphinscheduler.common.enums.TaskDependType; |
||||
import org.apache.dolphinscheduler.common.enums.WarningType; |
||||
import org.apache.dolphinscheduler.plugin.task.api.enums.ExecutionStatus; |
||||
|
||||
import java.util.Collection; |
||||
import java.util.Date; |
||||
|
||||
import lombok.Getter; |
||||
import lombok.Setter; |
||||
|
||||
@Setter |
||||
@Getter |
||||
public class WorkflowExecuteDto { |
||||
|
||||
private int id; |
||||
|
||||
private String name; |
||||
|
||||
private Long processDefinitionCode; |
||||
|
||||
private int processDefinitionVersion; |
||||
|
||||
private ExecutionStatus state; |
||||
|
||||
/** |
||||
* recovery flag for failover |
||||
*/ |
||||
private Flag recovery; |
||||
|
||||
private Date startTime; |
||||
|
||||
private Date endTime; |
||||
|
||||
private int runTimes; |
||||
|
||||
private String host; |
||||
|
||||
private CommandType commandType; |
||||
|
||||
private String commandParam; |
||||
|
||||
/** |
||||
* node depend type |
||||
*/ |
||||
private TaskDependType taskDependType; |
||||
|
||||
private int maxTryTimes; |
||||
|
||||
/** |
||||
* failure strategy when task failed. |
||||
*/ |
||||
private FailureStrategy failureStrategy; |
||||
|
||||
/** |
||||
* warning type |
||||
*/ |
||||
private WarningType warningType; |
||||
|
||||
private Integer warningGroupId; |
||||
|
||||
private Date scheduleTime; |
||||
|
||||
private Date commandStartTime; |
||||
|
||||
/** |
||||
* user define parameters string |
||||
*/ |
||||
private String globalParams; |
||||
|
||||
/** |
||||
* executor id |
||||
*/ |
||||
private int executorId; |
||||
|
||||
/** |
||||
* executor name |
||||
*/ |
||||
private String executorName; |
||||
|
||||
/** |
||||
* tenant code |
||||
*/ |
||||
private String tenantCode; |
||||
|
||||
/** |
||||
* queue |
||||
*/ |
||||
private String queue; |
||||
|
||||
/** |
||||
* process is sub process |
||||
*/ |
||||
private Flag isSubProcess; |
||||
|
||||
/** |
||||
* history command |
||||
*/ |
||||
private String historyCmd; |
||||
|
||||
/** |
||||
* depend processes schedule time |
||||
*/ |
||||
private String dependenceScheduleTimes; |
||||
|
||||
private String duration; |
||||
|
||||
private Priority processInstancePriority; |
||||
|
||||
private String workerGroup; |
||||
|
||||
private Long environmentCode; |
||||
|
||||
private int timeout; |
||||
|
||||
private int tenantId; |
||||
|
||||
/** |
||||
* varPool string |
||||
*/ |
||||
private String varPool; |
||||
|
||||
private int nextProcessInstanceId; |
||||
|
||||
private int dryRun; |
||||
|
||||
private Date restartTime; |
||||
|
||||
private boolean isBlocked; |
||||
|
||||
private Collection<TaskInstanceExecuteDto> taskInstances; |
||||
} |
||||
Loading…
Reference in new issue