host add host:port format (#2070)

* 1, master persistent task 2. extract master and worker communication model * 1, master persistent task 2. extract master and worker communication model * 1, master persistent task 2. extract master and worker communication model * add license * modify javadoc error * TaskExecutionContext create modify * buildAckCommand taskInstanceId not set modify * java doc error modify * add comment * ExecutorManager interface add generic type * add TaskInstanceCacheManager receive Worker report result * TaskInstance setExecutePath * add TaskInstanceCacheManager to receive Worker Task result report * TaskInstanceCacheManager add remove method * add license * add dispatcht task method * AbstractCommandExecutor remove db access * AbstractCommandExecutor remove db access * AbstractCommandExecutor remove db access * AbstractCommandExecutor remove db access * AbstractCommandExecutor remove db access * AbstractCommandExecutor remove db access * AbstractCommandExecutor remove db access * taskInstanceCache is null ,need load from db * taskInstanceCache is null ,need load from db * taskInstanceCache is null ,need load from db * 1，worker TaskPros use TaskExecutionContext replase 2，Master kill Task , KillTaskProcessor modify * worker remove db * ShellTask modify * master persistence processId and appIds * master persistence processId and appIds * master add kill task logic * master add kill task logic * master add kill task logic * javadoc error modify * remove chinese log * executeDirectly method add Override * remote module modify * TaskKillResponseProcessor command type modify * create buildKillCommand * host add host:port format * host add host:port format
5 years ago · 72e8f8d195
7 changed files with 77 additions and 89 deletions
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/LoggerService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/LoggerService.java
@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.api.utils.Result;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.utils.StringUtils;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.remote.utils.Host;
 import org.apache.dolphinscheduler.service.log.LogClientService;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 import org.slf4j.Logger;
@ -68,7 +69,7 @@ public class LoggerService {
      return new Result(Status.TASK_INSTANCE_NOT_FOUND.getCode(), Status.TASK_INSTANCE_NOT_FOUND.getMsg());
    }

-    String host = taskInstance.getHost();
+    String host = Host.of(taskInstance.getHost()).getIp();
    if(StringUtils.isEmpty(host)){
      return new Result(Status.TASK_INSTANCE_NOT_FOUND.getCode(), Status.TASK_INSTANCE_NOT_FOUND.getMsg());
    }
@ -94,7 +95,7 @@ public class LoggerService {
    if (taskInstance == null){
      throw new RuntimeException("task instance is null");
    }
-    String host = taskInstance.getHost();
+    String host = Host.of(taskInstance.getHost()).getIp();
    return logClient.getLogBytes(host, Constants.RPC_PORT, taskInstance.getLogPath());
  }
 }
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskInstance.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/TaskInstance.java
@ -186,21 +186,11 @@ public class TaskInstance implements Serializable {
    @TableField(exist = false)
    private String dependentResult;

-
    /**
     * worker group id
     */
    private int workerGroupId;

-
-
-    public void init(String host,Date startTime,String executePath){
-        this.host = host;
-        this.startTime = startTime;
-        this.executePath = executePath;
-    }
-
-
    public ProcessInstance getProcessInstance() {
        return processInstance;
    }
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/builder/TaskExecutionContextBuilder.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/builder/TaskExecutionContextBuilder.java
@ -48,7 +48,6 @@ public class TaskExecutionContextBuilder {
        taskExecutionContext.setLogPath(taskInstance.getLogPath());
        taskExecutionContext.setExecutePath(taskInstance.getExecutePath());
        taskExecutionContext.setTaskJson(taskInstance.getTaskJson());
-        taskExecutionContext.setHost(taskInstance.getHost());
        taskExecutionContext.setWorkerGroup(Constants.DEFAULT_WORKER_GROUP);
        return this;
    }
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/entity/TaskExecutionContext.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/entity/TaskExecutionContext.java
@ -29,7 +29,7 @@ public class TaskExecutionContext implements Serializable{
    /**
     *  task id
     */
-    private Integer taskInstanceId;
+    private int taskInstanceId;


    /**
@ -51,7 +51,7 @@ public class TaskExecutionContext implements Serializable{
     * host
     */
    private String host;
-
+    
    /**
     *  task execute path
     */
@ -70,7 +70,7 @@ public class TaskExecutionContext implements Serializable{
    /**
     * processId
     */
-    private Integer processId;
+    private int processId;

    /**
     * appIds
@ -80,7 +80,7 @@ public class TaskExecutionContext implements Serializable{
    /**
     *  process instance id
     */
-    private Integer processInstanceId;
+    private int processInstanceId;


    /**
@ -97,13 +97,13 @@ public class TaskExecutionContext implements Serializable{
    /**
     *  execute user id
     */
-    private Integer executorId;
+    private int executorId;


    /**
     *  command type if complement
     */
-    private Integer cmdTypeIfComplement;
+    private int cmdTypeIfComplement;


    /**
@ -120,12 +120,12 @@ public class TaskExecutionContext implements Serializable{
    /**
     *  process define id
     */
-    private Integer processDefineId;
+    private int processDefineId;

    /**
     *  project id
     */
-    private Integer projectId;
+    private int projectId;

    /**
     * taskParams
@ -173,22 +173,11 @@ public class TaskExecutionContext implements Serializable{
     */
    private DataxTaskExecutionContext dataxTaskExecutionContext;

-
-
-
-    public String getWorkerGroup() {
-        return workerGroup;
-    }
-
-    public void setWorkerGroup(String workerGroup) {
-        this.workerGroup = workerGroup;
-    }
-
-    public Integer getTaskInstanceId() {
+    public int getTaskInstanceId() {
        return taskInstanceId;
    }

-    public void setTaskInstanceId(Integer taskInstanceId) {
+    public void setTaskInstanceId(int taskInstanceId) {
        this.taskInstanceId = taskInstanceId;
    }

@ -216,6 +205,14 @@ public class TaskExecutionContext implements Serializable{
        this.taskType = taskType;
    }

+    public String getHost() {
+        return host;
+    }
+
+    public void setHost(String host) {
+        this.host = host;
+    }
+
    public String getExecutePath() {
        return executePath;
    }
@ -224,6 +221,14 @@ public class TaskExecutionContext implements Serializable{
        this.executePath = executePath;
    }

+    public String getLogPath() {
+        return logPath;
+    }
+
+    public void setLogPath(String logPath) {
+        this.logPath = logPath;
+    }
+
    public String getTaskJson() {
        return taskJson;
    }
@ -232,11 +237,27 @@ public class TaskExecutionContext implements Serializable{
        this.taskJson = taskJson;
    }

-    public Integer getProcessInstanceId() {
+    public int getProcessId() {
+        return processId;
+    }
+
+    public void setProcessId(int processId) {
+        this.processId = processId;
+    }
+
+    public String getAppIds() {
+        return appIds;
+    }
+
+    public void setAppIds(String appIds) {
+        this.appIds = appIds;
+    }
+
+    public int getProcessInstanceId() {
        return processInstanceId;
    }

-    public void setProcessInstanceId(Integer processInstanceId) {
+    public void setProcessInstanceId(int processInstanceId) {
        this.processInstanceId = processInstanceId;
    }

@ -256,6 +277,22 @@ public class TaskExecutionContext implements Serializable{
        this.globalParams = globalParams;
    }

+    public int getExecutorId() {
+        return executorId;
+    }
+
+    public void setExecutorId(int executorId) {
+        this.executorId = executorId;
+    }
+
+    public int getCmdTypeIfComplement() {
+        return cmdTypeIfComplement;
+    }
+
+    public void setCmdTypeIfComplement(int cmdTypeIfComplement) {
+        this.cmdTypeIfComplement = cmdTypeIfComplement;
+    }
+
    public String getTenantCode() {
        return tenantCode;
    }
@ -272,46 +309,22 @@ public class TaskExecutionContext implements Serializable{
        this.queue = queue;
    }

-    public Integer getProcessDefineId() {
+    public int getProcessDefineId() {
        return processDefineId;
    }

-    public void setProcessDefineId(Integer processDefineId) {
+    public void setProcessDefineId(int processDefineId) {
        this.processDefineId = processDefineId;
    }

-    public Integer getProjectId() {
+    public int getProjectId() {
        return projectId;
    }

-    public void setProjectId(Integer projectId) {
+    public void setProjectId(int projectId) {
        this.projectId = projectId;
    }

-    public Integer getExecutorId() {
-        return executorId;
-    }
-
-    public void setExecutorId(Integer executorId) {
-        this.executorId = executorId;
-    }
-
-    public Integer getCmdTypeIfComplement() {
-        return cmdTypeIfComplement;
-    }
-
-    public void setCmdTypeIfComplement(Integer cmdTypeIfComplement) {
-        this.cmdTypeIfComplement = cmdTypeIfComplement;
-    }
-
-    public String getLogPath() {
-        return logPath;
-    }
-
-    public void setLogPath(String logPath) {
-        this.logPath = logPath;
-    }
-
    public String getTaskParams() {
        return taskParams;
    }
@ -360,28 +373,12 @@ public class TaskExecutionContext implements Serializable{
        this.taskTimeout = taskTimeout;
    }

-    public String getHost() {
-        return host;
-    }
-
-    public void setHost(String host) {
-        this.host = host;
-    }
-
-    public Integer getProcessId() {
-        return processId;
-    }
-
-    public void setProcessId(Integer processId) {
-        this.processId = processId;
-    }
-
-    public String getAppIds() {
-        return appIds;
+    public String getWorkerGroup() {
+        return workerGroup;
    }

-    public void setAppIds(String appIds) {
-        this.appIds = appIds;
+    public void setWorkerGroup(String workerGroup) {
+        this.workerGroup = workerGroup;
    }

    public SQLTaskExecutionContext getSqlTaskExecutionContext() {
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/ExecutorDispatcher.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/ExecutorDispatcher.java
@ -90,6 +90,7 @@ public class ExecutorDispatcher implements InitializingBean {
            throw new ExecuteException(String.format("fail to execute : %s due to no worker ", context.getContext()));
        }
        context.setHost(host);
+        context.getContext().setHost(host.getAddress());
        executorManager.beforeExecute(context);
        try {
            /**
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterTaskExecThread.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterTaskExecThread.java
@ -186,9 +186,7 @@ public class MasterTaskExecThread extends MasterBaseTaskExecThread {
        TaskExecutionContext taskExecutionContext = super.getTaskExecutionContext(taskInstance);
        ExecutionContext executionContext = new ExecutionContext(taskExecutionContext, ExecutorType.WORKER);

-        Host host = new Host();
-        host.setIp(taskInstance.getHost());
-        host.setPort(12346);
+        Host host = Host.of(taskInstance.getHost());
        executionContext.setHost(host);

        nettyExecutorManager.executeDirectly(executionContext);
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskExecuteProcessor.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskExecuteProcessor.java
@ -78,10 +78,12 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
    public void process(Channel channel, Command command) {
        Preconditions.checkArgument(CommandType.TASK_EXECUTE_REQUEST == command.getType(),
                String.format("invalid command type : %s", command.getType()));
-        logger.info("received command : {}", command);
+
        TaskExecuteRequestCommand taskRequestCommand = FastJsonSerializer.deserialize(
                command.getBody(), TaskExecuteRequestCommand.class);

+        logger.info("received command : {}", taskRequestCommand);
+
        String contextJson = taskRequestCommand.getTaskExecutionContext();

        TaskExecutionContext taskExecutionContext = JSONObject.parseObject(contextJson, TaskExecutionContext.class);
@ -141,7 +143,7 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
        ackCommand.setTaskInstanceId(taskExecutionContext.getTaskInstanceId());
        ackCommand.setStatus(ExecutionStatus.RUNNING_EXEUTION.getCode());
        ackCommand.setLogPath(getTaskLogPath(taskExecutionContext));
-        ackCommand.setHost(OSUtils.getHost());
+        ackCommand.setHost(taskExecutionContext.getHost());
        ackCommand.setStartTime(new Date());
        if(taskExecutionContext.getTaskType().equals(TaskType.SQL.name()) || taskExecutionContext.getTaskType().equals(TaskType.PROCEDURE.name())){
            ackCommand.setExecutePath(null);