From 2bb2d0ce7c86c8a77bd637cec68cea18a21c062a Mon Sep 17 00:00:00 2001 From: qiaozhanwei Date: Thu, 20 Feb 2020 14:55:53 +0800 Subject: [PATCH] master/worker basic communication --- .../remote/command/ExecuteTaskAckCommand.java | 2 +- .../server/log/LoggerRequestProcessor.java | 16 +++++----- .../runner/MasterBaseTaskExecThread.java | 30 +++++++++++++++---- .../worker/processor/TaskCallbackService.java | 5 ++-- .../processor/WorkerRequestProcessor.java | 3 -- .../worker/runner/TaskScheduleThread.java | 4 +-- 6 files changed, 39 insertions(+), 21 deletions(-) diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/ExecuteTaskAckCommand.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/ExecuteTaskAckCommand.java index 24ab68f260..7f7da0e445 100644 --- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/ExecuteTaskAckCommand.java +++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/ExecuteTaskAckCommand.java @@ -1 +1 @@ -/* * 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.remote.utils.FastJsonSerializer; import java.io.Serializable; import java.util.Date; /** * execute task request command */ public class ExecuteTaskAckCommand implements Serializable { private int taskInstanceId; private Date startTime; private String host; private int status; private String logPath; private String executePath; public Date getStartTime() { return startTime; } public void setStartTime(Date startTime) { this.startTime = startTime; } public String getHost() { return host; } public void setHost(String host) { this.host = host; } public int getStatus() { return status; } public void setStatus(int status) { this.status = status; } public int getTaskInstanceId() { return taskInstanceId; } public void setTaskInstanceId(int taskInstanceId) { this.taskInstanceId = taskInstanceId; } public String getLogPath() { return logPath; } public void setLogPath(String logPath) { this.logPath = logPath; } public String getExecutePath() { return executePath; } public void setExecutePath(String executePath) { this.executePath = executePath; } /** * package request command * * @return command */ public Command convert2Command(long opaque){ Command command = new Command(opaque); command.setType(CommandType.EXECUTE_TASK_ACK); byte[] body = FastJsonSerializer.serialize(this); command.setBody(body); return command; } } \ No newline at end of file +/* * 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.remote.utils.FastJsonSerializer; import java.io.Serializable; import java.util.Date; /** * execute task request command */ public class ExecuteTaskAckCommand implements Serializable { private int taskInstanceId; private Date startTime; private String host; private int status; private String logPath; private String executePath; public Date getStartTime() { return startTime; } public void setStartTime(Date startTime) { this.startTime = startTime; } public String getHost() { return host; } public void setHost(String host) { this.host = host; } public int getStatus() { return status; } public void setStatus(int status) { this.status = status; } public int getTaskInstanceId() { return taskInstanceId; } public void setTaskInstanceId(int taskInstanceId) { this.taskInstanceId = taskInstanceId; } public String getLogPath() { return logPath; } public void setLogPath(String logPath) { this.logPath = logPath; } public String getExecutePath() { return executePath; } public void setExecutePath(String executePath) { this.executePath = executePath; } /** * package request command * * @return command */ public Command convert2Command(long opaque){ Command command = new Command(opaque); command.setType(CommandType.EXECUTE_TASK_ACK); byte[] body = FastJsonSerializer.serialize(this); command.setBody(body); return command; } @Override public String toString() { return "ExecuteTaskAckCommand{" + "taskInstanceId=" + taskInstanceId + ", startTime=" + startTime + ", host='" + host + '\'' + ", status=" + status + ", logPath='" + logPath + '\'' + ", executePath='" + executePath + '\'' + '}'; } } \ No newline at end of file diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/log/LoggerRequestProcessor.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/log/LoggerRequestProcessor.java index 4e4404ea1c..818b453a1b 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/log/LoggerRequestProcessor.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/log/LoggerRequestProcessor.java @@ -59,14 +59,14 @@ public class LoggerRequestProcessor implements NettyRequestProcessor { */ final CommandType commandType = command.getType(); switch (commandType){ - case GET_LOG_BYTES_REQUEST: - GetLogBytesRequestCommand getLogRequest = FastJsonSerializer.deserialize( - command.getBody(), GetLogBytesRequestCommand.class); - byte[] bytes = getFileContentBytes(getLogRequest.getPath()); - GetLogBytesResponseCommand getLogResponse = new GetLogBytesResponseCommand(bytes); - channel.writeAndFlush(getLogResponse.convert2Command(command.getOpaque())); - break; - case VIEW_WHOLE_LOG_REQUEST: + case GET_LOG_BYTES_REQUEST: + GetLogBytesRequestCommand getLogRequest = FastJsonSerializer.deserialize( + command.getBody(), GetLogBytesRequestCommand.class); + byte[] bytes = getFileContentBytes(getLogRequest.getPath()); + GetLogBytesResponseCommand getLogResponse = new GetLogBytesResponseCommand(bytes); + channel.writeAndFlush(getLogResponse.convert2Command(command.getOpaque())); + break; + case VIEW_WHOLE_LOG_REQUEST: ViewLogRequestCommand viewLogRequest = FastJsonSerializer.deserialize( command.getBody(), ViewLogRequestCommand.class); String msg = readWholeFileContent(viewLogRequest.getPath()); diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterBaseTaskExecThread.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterBaseTaskExecThread.java index 27d527a59a..8104c42bd0 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterBaseTaskExecThread.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterBaseTaskExecThread.java @@ -22,11 +22,12 @@ import org.apache.dolphinscheduler.dao.entity.ProcessInstance; import org.apache.dolphinscheduler.dao.entity.TaskInstance; import org.apache.dolphinscheduler.dao.utils.BeanContext; import org.apache.dolphinscheduler.remote.NettyRemotingClient; -import org.apache.dolphinscheduler.remote.command.Command; -import org.apache.dolphinscheduler.remote.command.ExecuteTaskRequestCommand; +import org.apache.dolphinscheduler.remote.command.*; +import org.apache.dolphinscheduler.remote.command.log.RollViewLogRequestCommand; import org.apache.dolphinscheduler.remote.config.NettyClientConfig; import org.apache.dolphinscheduler.remote.exceptions.RemotingException; import org.apache.dolphinscheduler.remote.utils.Address; +import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer; import org.apache.dolphinscheduler.server.master.config.MasterConfig; import org.apache.dolphinscheduler.service.bean.SpringApplicationContext; import org.apache.dolphinscheduler.service.process.ProcessService; @@ -122,10 +123,29 @@ public class MasterBaseTaskExecThread implements Callable { // TODO send task to worker public void sendToWorker(String taskInstanceJson){ final Address address = new Address("127.0.0.1", 12346); - ExecuteTaskRequestCommand command = new ExecuteTaskRequestCommand(taskInstanceJson); + ExecuteTaskRequestCommand taskRequestCommand = new ExecuteTaskRequestCommand(taskInstanceJson); try { - Command response = nettyRemotingClient.sendSync(address, command.convert2Command(), Integer.MAX_VALUE); - logger.info("response result : {}",response); + Command responseCommand = nettyRemotingClient.sendSync(address, + taskRequestCommand.convert2Command(), Integer.MAX_VALUE); + + logger.info("receive command : {}", responseCommand); + + final CommandType commandType = responseCommand.getType(); + switch (commandType){ + case EXECUTE_TASK_ACK: + ExecuteTaskAckCommand taskAckCommand = FastJsonSerializer.deserialize( + responseCommand.getBody(), ExecuteTaskAckCommand.class); + logger.info("taskAckCommand : {}",taskAckCommand); + break; + case EXECUTE_TASK_RESPONSE: + ExecuteTaskResponseCommand taskResponseCommand = FastJsonSerializer.deserialize( + responseCommand.getBody(), ExecuteTaskResponseCommand.class); + logger.info("taskResponseCommand : {}",taskResponseCommand); + break; + default: + throw new IllegalArgumentException("unknown commandType"); + } + logger.info("response result : {}",responseCommand); } catch (InterruptedException | RemotingException ex) { logger.error(String.format("send command to : %s error", address), ex); } diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java index 762ee18c45..2b6ed957d0 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java @@ -56,7 +56,8 @@ public class TaskCallbackService { public void sendResult(int taskInstanceId, ExecuteTaskResponseCommand responseCommand){ CallbackChannel callbackChannel = getCallbackChannel(taskInstanceId); - callbackChannel.getChannel().writeAndFlush(responseCommand.convert2Command(0)).addListener(new ChannelFutureListener(){ + callbackChannel.getChannel().writeAndFlush(responseCommand.convert2Command( + callbackChannel.getOpaque())).addListener(new ChannelFutureListener(){ @Override public void operationComplete(ChannelFuture future) throws Exception { @@ -68,7 +69,7 @@ public class TaskCallbackService { }); } - //TODO + // TODO private Channel createChannel(){ return null; } diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/WorkerRequestProcessor.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/WorkerRequestProcessor.java index be6b95e489..0a1b7c4837 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/WorkerRequestProcessor.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/WorkerRequestProcessor.java @@ -123,9 +123,6 @@ public class WorkerRequestProcessor implements NettyRequestProcessor { // submit task workerExecService.submit(new TaskScheduleThread(taskInstance, processService, taskCallbackService)); - - ExecuteTaskResponseCommand executeTaskResponseCommand = new ExecuteTaskResponseCommand(taskInstance.getId()); - channel.writeAndFlush(executeTaskResponseCommand.convert2Command(command.getOpaque())); } private boolean verifyTenantIsNull(Tenant tenant, TaskInstance taskInstance) { diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskScheduleThread.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskScheduleThread.java index 3488fe0031..bb233b559e 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskScheduleThread.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskScheduleThread.java @@ -96,7 +96,7 @@ public class TaskScheduleThread implements Runnable { @Override public void run() { - // TODO 需要去掉，暂时保留 + // TODO Need to be removed and kept temporarily update task instance state updateTaskState(taskInstance.getTaskType()); ExecuteTaskResponseCommand responseCommand = new ExecuteTaskResponseCommand(taskInstance.getId()); @@ -172,7 +172,7 @@ public class TaskScheduleThread implements Runnable { logger.error("task scheduler failure", e); kill(); - //TODO 需要去掉，暂时保留 update task instance state + //TODO Need to be removed and kept temporarily update task instance state processService.changeTaskState(ExecutionStatus.FAILURE, new Date(), taskInstance.getId());