Browse Source
* Refactor worker (#2) * Refactor worker (#1993) * Refactor worker (#1) * add TaskResponseProcessor (#1983) * 1, master persistent task 2. extract master and worker communication model (#1992) * 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 Co-authored-by: qiaozhanwei <qiaozhanwei@outlook.com> * updates Co-authored-by: qiaozhanwei <qiaozhanwei@outlook.com> * TaskExecutionContext create modify (#1994) * 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 Co-authored-by: qiaozhanwei <qiaozhanwei@outlook.com> * updates Co-authored-by: qiaozhanwei <qiaozhanwei@outlook.com>pull/2/head
Tboy
5 years ago
committed by
GitHub
36 changed files with 771 additions and 257 deletions
@ -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;
}
@Override
public String toString() {
return "ExecuteTaskAckCommand{" +
"taskInstanceId=" + taskInstanceId +
", startTime=" + startTime +
", host='" + host + '\'' +
", status=" + status +
", logPath='" + logPath + '\'' +
", executePath='" + executePath + '\'' +
'}';
}
} |
/*
* 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(){
Command command = new Command();
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 + '\'' +
'}';
}
} |
||||||
@ -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;
/**
* execute task request command
*/
public class ExecuteTaskRequestCommand implements Serializable {
/**
* task instance json
*/
private String taskInfoJson;
public String getTaskInfoJson() {
return taskInfoJson;
}
public void setTaskInfoJson(String taskInfoJson) {
this.taskInfoJson = taskInfoJson;
}
instance json
}
instance json
instance json
instance json
*/
;
}
/**
* package request command
*
* @return command
*/
public Command convert2Command(){
Command command = new Command();
command.setType(CommandType.EXECUTE_TASK_REQUEST);
byte[] body = FastJsonSerializer.serialize(this);
command.setBody(body);
return command;
}
@Override
public String toString() {
return "ExecuteTaskRequestCommand{" +
"taskInfoJson='" + taskInfoJson + '\'' +
'}';
}
} |
/*
* 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;
/**
* execute task request command
*/
public class ExecuteTaskRequestCommand implements Serializable {
/**
* task instance json
private String taskInfoJson;
instance json
instance json
public String getTaskInfoJson() {
instance json
return taskInfoJson;
instance json
}
instance json
public void setTaskInfoJson(String taskInfoJson) {
instance json
this.taskInfoJson = taskInfoJson;
*/
this.taskExecutionContext = taskExecutionContext;
}
instance json
return taskInfoJson;
public ExecuteTaskRequestCommand() {
instance json
this.taskInfoJson = taskInfoJson;
instance json
return taskInfoJson;
*/
private String taskInfoJson;
this.taskExecutionContext = taskExecutionContext;
}
/**
* package request command
*
* @return command
*/
public Command convert2Command(){
Command command = new Command();
command.setType(CommandType.EXECUTE_TASK_REQUEST);
byte[] body = FastJsonSerializer.serialize(this);
command.setBody(body);
return command;
}
@Override
public String toString() {
return "ExecuteTaskRequestCommand{" +
"taskExecutionContext='" + taskExecutionContext + '\'' +
'}';
}
} |
||||||
@ -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 response command
*/
public class ExecuteTaskResponseCommand implements Serializable {
public ExecuteTaskResponseCommand() {
}
public ExecuteTaskResponseCommand(int taskInstanceId) {
this.taskInstanceId = taskInstanceId;
}
/**
* task instance id
*/
private int taskInstanceId;
/**
* status
*/
private int status;
/**
* end time
*/
private Date endTime;
public int getTaskInstanceId() {
return taskInstanceId;
}
public void setTaskInstanceId(int taskInstanceId) {
this.taskInstanceId = taskInstanceId;
}
public int getStatus() {
return status;
}
public void setStatus(int status) {
this.status = status;
}
public Date getEndTime() {
return endTime;
}
public void setEndTime(Date endTime) {
this.endTime = endTime;
}
/**
* package response command
*
* @param opaque request unique identification
* @return command
*/
public Command convert2Command(long opaque){
Command command = new Command(opaque);
command.setType(CommandType.EXECUTE_TASK_RESPONSE);
byte[] body = FastJsonSerializer.serialize(this);
command.setBody(body);
return command;
}
} |
/*
* 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 response command
*/
public class ExecuteTaskResponseCommand implements Serializable {
public ExecuteTaskResponseCommand() {
}
public ExecuteTaskResponseCommand(int taskInstanceId) {
this.taskInstanceId = taskInstanceId;
}
/**
* task instance id
*/
private int taskInstanceId;
/**
* status
*/
private int status;
/**
* end time
*/
private Date endTime;
public int getTaskInstanceId() {
return taskInstanceId;
}
public void setTaskInstanceId(int taskInstanceId) {
this.taskInstanceId = taskInstanceId;
}
public int getStatus() {
return status;
}
public void setStatus(int status) {
this.status = status;
}
public Date getEndTime() {
return endTime;
}
public void setEndTime(Date endTime) {
this.endTime = endTime;
}
/**
* package response command
* @return command
*/
public Command convert2Command(){
Command command = new Command();
command.setType(CommandType.EXECUTE_TASK_RESPONSE);
byte[] body = FastJsonSerializer.serialize(this);
command.setBody(body);
return command;
}
} |
||||||
@ -1,96 +0,0 @@ |
|||||||
/* |
|
||||||
* 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.utils; |
|
||||||
|
|
||||||
import java.io.Serializable; |
|
||||||
|
|
||||||
/** |
|
||||||
* server address |
|
||||||
*/ |
|
||||||
public class Address implements Serializable { |
|
||||||
|
|
||||||
/** |
|
||||||
* host |
|
||||||
*/ |
|
||||||
private String host; |
|
||||||
|
|
||||||
/** |
|
||||||
* port |
|
||||||
*/ |
|
||||||
private int port; |
|
||||||
|
|
||||||
public Address(){ |
|
||||||
//NOP
|
|
||||||
} |
|
||||||
|
|
||||||
public Address(String host, int port){ |
|
||||||
this.host = host; |
|
||||||
this.port = port; |
|
||||||
} |
|
||||||
|
|
||||||
public String getHost() { |
|
||||||
return host; |
|
||||||
} |
|
||||||
|
|
||||||
public void setHost(String host) { |
|
||||||
this.host = host; |
|
||||||
} |
|
||||||
|
|
||||||
public int getPort() { |
|
||||||
return port; |
|
||||||
} |
|
||||||
|
|
||||||
public void setPort(int port) { |
|
||||||
this.port = port; |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public int hashCode() { |
|
||||||
final int prime = 31; |
|
||||||
int result = 1; |
|
||||||
result = prime * result + ((host == null) ? 0 : host.hashCode()); |
|
||||||
result = prime * result + port; |
|
||||||
return result; |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public boolean equals(Object obj) { |
|
||||||
if (this == obj) { |
|
||||||
return true; |
|
||||||
} |
|
||||||
if (obj == null) { |
|
||||||
return false; |
|
||||||
} |
|
||||||
if (getClass() != obj.getClass()) { |
|
||||||
return false; |
|
||||||
} |
|
||||||
Address other = (Address) obj; |
|
||||||
if (host == null) { |
|
||||||
if (other.host != null) { |
|
||||||
return false; |
|
||||||
} |
|
||||||
} else if (!host.equals(other.host)) { |
|
||||||
return false; |
|
||||||
} |
|
||||||
return port == other.port; |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public String toString() { |
|
||||||
return "Address [host=" + host + ", port=" + port + "]"; |
|
||||||
} |
|
||||||
} |
|
||||||
@ -0,0 +1,77 @@ |
|||||||
|
/* |
||||||
|
* 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.dispatch; |
||||||
|
|
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.common.utils.StringUtils; |
||||||
|
import org.apache.dolphinscheduler.remote.utils.Host; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.enums.ExecutorType; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteException; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.executor.ExecutorManager; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.host.RoundRobinHostManager; |
||||||
|
import org.springframework.beans.factory.InitializingBean; |
||||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||||
|
import org.springframework.stereotype.Service; |
||||||
|
|
||||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||||
|
|
||||||
|
@Service |
||||||
|
public class ExecutorDispatcher implements InitializingBean { |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private NettyExecutorManager nettyExecutorManager; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private RoundRobinHostManager hostManager; |
||||||
|
|
||||||
|
private final ConcurrentHashMap<ExecutorType, ExecutorManager> executorManagers; |
||||||
|
|
||||||
|
public ExecutorDispatcher(){ |
||||||
|
this.executorManagers = new ConcurrentHashMap<>(); |
||||||
|
} |
||||||
|
|
||||||
|
public void dispatch(final ExecutionContext executeContext) throws ExecuteException { |
||||||
|
ExecutorManager executorManager = this.executorManagers.get(executeContext.getExecutorType()); |
||||||
|
if(executorManager == null){ |
||||||
|
throw new ExecuteException("no ExecutorManager for type : " + executeContext.getExecutorType()); |
||||||
|
} |
||||||
|
Host host = hostManager.select(executeContext); |
||||||
|
if (StringUtils.isEmpty(host.getAddress())) { |
||||||
|
throw new ExecuteException(String.format("fail to execute : %s due to no worker ", executeContext.getContext())); |
||||||
|
} |
||||||
|
executeContext.setHost(host); |
||||||
|
executorManager.beforeExecute(executeContext); |
||||||
|
try { |
||||||
|
executorManager.execute(executeContext); |
||||||
|
} finally { |
||||||
|
executorManager.afterExecute(executeContext); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public void afterPropertiesSet() throws Exception { |
||||||
|
register(ExecutorType.WORKER, nettyExecutorManager); |
||||||
|
register(ExecutorType.CLIENT, nettyExecutorManager); |
||||||
|
} |
||||||
|
|
||||||
|
public void register(ExecutorType type, ExecutorManager executorManager){ |
||||||
|
executorManagers.put(type, executorManager); |
||||||
|
} |
||||||
|
} |
||||||
@ -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.dispatch.context; |
||||||
|
|
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.remote.utils.Host; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.enums.ExecutorType; |
||||||
|
|
||||||
|
public class ExecutionContext { |
||||||
|
|
||||||
|
private Host host; |
||||||
|
|
||||||
|
private final Object context; |
||||||
|
|
||||||
|
private final ExecutorType executorType; |
||||||
|
|
||||||
|
public ExecutionContext(Object context, ExecutorType executorType) { |
||||||
|
this.context = context; |
||||||
|
this.executorType = executorType; |
||||||
|
} |
||||||
|
|
||||||
|
public ExecutorType getExecutorType() { |
||||||
|
return executorType; |
||||||
|
} |
||||||
|
|
||||||
|
public Object getContext() { |
||||||
|
return context; |
||||||
|
} |
||||||
|
|
||||||
|
public Host getHost() { |
||||||
|
return host; |
||||||
|
} |
||||||
|
|
||||||
|
public void setHost(Host host) { |
||||||
|
this.host = host; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,95 @@ |
|||||||
|
/* |
||||||
|
* 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.dispatch.exceptions; |
||||||
|
|
||||||
|
|
||||||
|
public class ExecuteException extends Exception{ |
||||||
|
|
||||||
|
public ExecuteException() { |
||||||
|
super(); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* Constructs a new exception with the specified detail message. The |
||||||
|
* cause is not initialized, and may subsequently be initialized by |
||||||
|
* a call to {@link #initCause}. |
||||||
|
* |
||||||
|
* @param message the detail message. The detail message is saved for |
||||||
|
* later retrieval by the {@link #getMessage()} method. |
||||||
|
*/ |
||||||
|
public ExecuteException(String message) { |
||||||
|
super(message); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* Constructs a new exception with the specified detail message and |
||||||
|
* cause. <p>Note that the detail message associated with |
||||||
|
* {@code cause} is <i>not</i> automatically incorporated in |
||||||
|
* this exception's detail message. |
||||||
|
* |
||||||
|
* @param message the detail message (which is saved for later retrieval |
||||||
|
* by the {@link #getMessage()} method). |
||||||
|
* @param cause the cause (which is saved for later retrieval by the |
||||||
|
* {@link #getCause()} method). (A <tt>null</tt> value is |
||||||
|
* permitted, and indicates that the cause is nonexistent or |
||||||
|
* unknown.) |
||||||
|
* @since 1.4 |
||||||
|
*/ |
||||||
|
public ExecuteException(String message, Throwable cause) { |
||||||
|
super(message, cause); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* Constructs a new exception with the specified cause and a detail |
||||||
|
* message of <tt>(cause==null ? null : cause.toString())</tt> (which |
||||||
|
* typically contains the class and detail message of <tt>cause</tt>). |
||||||
|
* This constructor is useful for exceptions that are little more than |
||||||
|
* wrappers for other throwables (for example, {@link |
||||||
|
* java.security.PrivilegedActionException}). |
||||||
|
* |
||||||
|
* @param cause the cause (which is saved for later retrieval by the |
||||||
|
* {@link #getCause()} method). (A <tt>null</tt> value is |
||||||
|
* permitted, and indicates that the cause is nonexistent or |
||||||
|
* unknown.) |
||||||
|
* @since 1.4 |
||||||
|
*/ |
||||||
|
public ExecuteException(Throwable cause) { |
||||||
|
super(cause); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* Constructs a new exception with the specified detail message, |
||||||
|
* cause, suppression enabled or disabled, and writable stack |
||||||
|
* trace enabled or disabled. |
||||||
|
* |
||||||
|
* @param message the detail message. |
||||||
|
* @param cause the cause. (A {@code null} value is permitted, |
||||||
|
* and indicates that the cause is nonexistent or unknown.) |
||||||
|
* @param enableSuppression whether or not suppression is enabled |
||||||
|
* or disabled |
||||||
|
* @param writableStackTrace whether or not the stack trace should |
||||||
|
* be writable |
||||||
|
* @since 1.7 |
||||||
|
*/ |
||||||
|
protected ExecuteException(String message, Throwable cause, |
||||||
|
boolean enableSuppression, |
||||||
|
boolean writableStackTrace) { |
||||||
|
super(message, cause, enableSuppression, writableStackTrace); |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,36 @@ |
|||||||
|
/* |
||||||
|
* 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.dispatch.executor; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteException; |
||||||
|
|
||||||
|
|
||||||
|
public abstract class AbstractExecutorManager implements ExecutorManager{ |
||||||
|
|
||||||
|
@Override |
||||||
|
public void beforeExecute(ExecutionContext executeContext) throws ExecuteException { |
||||||
|
//TODO add time monitor
|
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public void afterExecute(ExecutionContext executeContext) throws ExecuteException { |
||||||
|
//TODO add dispatch monitor
|
||||||
|
|
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,31 @@ |
|||||||
|
/* |
||||||
|
* 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.dispatch.executor; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteException; |
||||||
|
|
||||||
|
|
||||||
|
public interface ExecutorManager { |
||||||
|
|
||||||
|
void beforeExecute(ExecutionContext executeContext) throws ExecuteException; |
||||||
|
|
||||||
|
void execute(ExecutionContext executeContext) throws ExecuteException; |
||||||
|
|
||||||
|
void afterExecute(ExecutionContext executeContext) throws ExecuteException; |
||||||
|
} |
||||||
@ -0,0 +1,144 @@ |
|||||||
|
/* |
||||||
|
* 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.dispatch.executor; |
||||||
|
|
||||||
|
import org.apache.commons.collections.CollectionUtils; |
||||||
|
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.config.NettyClientConfig; |
||||||
|
import org.apache.dolphinscheduler.remote.entity.TaskExecutionContext; |
||||||
|
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer; |
||||||
|
import org.apache.dolphinscheduler.remote.utils.Host; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.enums.ExecutorType; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteException; |
||||||
|
import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager; |
||||||
|
import org.slf4j.Logger; |
||||||
|
import org.slf4j.LoggerFactory; |
||||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||||
|
import org.springframework.stereotype.Service; |
||||||
|
|
||||||
|
import java.util.Collection; |
||||||
|
import java.util.Collections; |
||||||
|
import java.util.HashSet; |
||||||
|
import java.util.Set; |
||||||
|
|
||||||
|
|
||||||
|
@Service |
||||||
|
public class NettyExecutorManager extends AbstractExecutorManager{ |
||||||
|
|
||||||
|
private final Logger logger = LoggerFactory.getLogger(NettyExecutorManager.class); |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ZookeeperNodeManager zookeeperNodeManager; |
||||||
|
|
||||||
|
private final NettyRemotingClient nettyRemotingClient; |
||||||
|
|
||||||
|
public NettyExecutorManager(){ |
||||||
|
final NettyClientConfig clientConfig = new NettyClientConfig(); |
||||||
|
this.nettyRemotingClient = new NettyRemotingClient(clientConfig); |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public void execute(ExecutionContext executeContext) throws ExecuteException { |
||||||
|
Set<String> allNodes = getAllNodes(executeContext); |
||||||
|
Set<String> failNodeSet = new HashSet<>(); |
||||||
|
//
|
||||||
|
Command command = buildCommand(executeContext); |
||||||
|
Host host = executeContext.getHost(); |
||||||
|
boolean success = false; |
||||||
|
//
|
||||||
|
while (!success) { |
||||||
|
try { |
||||||
|
doExecute(host, command); |
||||||
|
success = true; |
||||||
|
executeContext.setHost(host); |
||||||
|
} catch (ExecuteException ex) { |
||||||
|
logger.error(String.format("execute context : %s error", executeContext.getContext()), ex); |
||||||
|
try { |
||||||
|
failNodeSet.add(host.getAddress()); |
||||||
|
Set<String> tmpAllIps = new HashSet<>(allNodes); |
||||||
|
Collection<String> remained = CollectionUtils.subtract(tmpAllIps, failNodeSet); |
||||||
|
if (remained != null && remained.size() > 0) { |
||||||
|
host = Host.of(remained.iterator().next()); |
||||||
|
logger.error("retry execute context : {} host : {}", executeContext.getContext(), host); |
||||||
|
} else { |
||||||
|
throw new ExecuteException("fail after try all nodes"); |
||||||
|
} |
||||||
|
} catch (Throwable t) { |
||||||
|
throw new ExecuteException("fail after try all nodes"); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
private Command buildCommand(ExecutionContext context) { |
||||||
|
ExecuteTaskRequestCommand requestCommand = new ExecuteTaskRequestCommand(); |
||||||
|
ExecutorType executorType = context.getExecutorType(); |
||||||
|
switch (executorType){ |
||||||
|
case WORKER: |
||||||
|
TaskExecutionContext taskExecutionContext = (TaskExecutionContext)context.getContext(); |
||||||
|
requestCommand.setTaskExecutionContext(FastJsonSerializer.serializeToString(taskExecutionContext)); |
||||||
|
break; |
||||||
|
case CLIENT: |
||||||
|
break; |
||||||
|
default: |
||||||
|
throw new IllegalArgumentException("invalid executor type : " + executorType); |
||||||
|
|
||||||
|
} |
||||||
|
return requestCommand.convert2Command(); |
||||||
|
} |
||||||
|
|
||||||
|
private void doExecute(final Host host, final Command command) throws ExecuteException { |
||||||
|
int retryCount = 3; |
||||||
|
boolean success = false; |
||||||
|
do { |
||||||
|
try { |
||||||
|
nettyRemotingClient.send(host, command); |
||||||
|
success = true; |
||||||
|
} catch (Exception ex) { |
||||||
|
logger.error(String.format("send command : %s to %s error", command, host), ex); |
||||||
|
retryCount--; |
||||||
|
try { |
||||||
|
Thread.sleep(100); |
||||||
|
} catch (InterruptedException ignore) {} |
||||||
|
} |
||||||
|
} while (retryCount >= 0 && !success); |
||||||
|
|
||||||
|
if (!success) { |
||||||
|
throw new ExecuteException(String.format("send command : %s to %s error", command, host)); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
private Set<String> getAllNodes(ExecutionContext context){ |
||||||
|
Set<String> nodes = Collections.EMPTY_SET; |
||||||
|
ExecutorType executorType = context.getExecutorType(); |
||||||
|
switch (executorType){ |
||||||
|
case WORKER: |
||||||
|
nodes = zookeeperNodeManager.getWorkerNodes(); |
||||||
|
break; |
||||||
|
case CLIENT: |
||||||
|
break; |
||||||
|
default: |
||||||
|
throw new IllegalArgumentException("invalid executor type : " + executorType); |
||||||
|
|
||||||
|
} |
||||||
|
return nodes; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,28 @@ |
|||||||
|
/* |
||||||
|
* 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.dispatch.host; |
||||||
|
|
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.remote.utils.Host; |
||||||
|
import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext; |
||||||
|
|
||||||
|
public interface HostManager { |
||||||
|
|
||||||
|
Host select(ExecutionContext context); |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,61 @@ |
|||||||
|
/* |
||||||
|
* 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 io.netty.channel.Channel; |
||||||
|
import org.apache.dolphinscheduler.common.enums.ExecutionStatus; |
||||||
|
import org.apache.dolphinscheduler.common.utils.Preconditions; |
||||||
|
import org.apache.dolphinscheduler.remote.command.Command; |
||||||
|
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||||
|
import org.apache.dolphinscheduler.remote.command.ExecuteTaskAckCommand; |
||||||
|
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor; |
||||||
|
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer; |
||||||
|
import org.apache.dolphinscheduler.service.process.ProcessService; |
||||||
|
import org.slf4j.Logger; |
||||||
|
import org.slf4j.LoggerFactory; |
||||||
|
|
||||||
|
/** |
||||||
|
* task ack processor |
||||||
|
*/ |
||||||
|
public class TaskAckProcessor implements NettyRequestProcessor { |
||||||
|
|
||||||
|
private final Logger logger = LoggerFactory.getLogger(TaskAckProcessor.class); |
||||||
|
|
||||||
|
/** |
||||||
|
* process service |
||||||
|
*/ |
||||||
|
private final ProcessService processService; |
||||||
|
|
||||||
|
public TaskAckProcessor(ProcessService processService){ |
||||||
|
this.processService = processService; |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public void process(Channel channel, Command command) { |
||||||
|
Preconditions.checkArgument(CommandType.EXECUTE_TASK_ACK == command.getType(), String.format("invalid command type : %s", command.getType())); |
||||||
|
ExecuteTaskAckCommand taskAckCommand = FastJsonSerializer.deserialize(command.getBody(), ExecuteTaskAckCommand.class); |
||||||
|
logger.info("taskAckCommand : {}",taskAckCommand); |
||||||
|
processService.changeTaskState(ExecutionStatus.of(taskAckCommand.getStatus()), |
||||||
|
taskAckCommand.getStartTime(), |
||||||
|
taskAckCommand.getHost(), |
||||||
|
taskAckCommand.getExecutePath(), |
||||||
|
taskAckCommand.getLogPath(), |
||||||
|
taskAckCommand.getTaskInstanceId()); |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
Loading…
Reference in new issue