mirror of
https://github.com/PowerJob/PowerJob.git
synced 2025-07-17 00:00:04 +08:00
feat: merge official processor upgrade
This commit is contained in:
commit
f1baef7de4
@ -14,7 +14,7 @@ public class RemoteConstant {
|
||||
|
||||
public static final String WORKER_ACTOR_SYSTEM_NAME = "oms";
|
||||
|
||||
public static final String Task_TRACKER_ACTOR_NAME = "task_tracker";
|
||||
public static final String TASK_TRACKER_ACTOR_NAME = "task_tracker";
|
||||
public static final String PROCESSOR_TRACKER_ACTOR_NAME = "processor_tracker";
|
||||
public static final String WORKER_ACTOR_NAME = "worker";
|
||||
public static final String TROUBLESHOOTING_ACTOR_NAME = "troubleshooting";
|
||||
|
@ -1,6 +1,7 @@
|
||||
package tech.powerjob.official.processors.impl;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.alibaba.fastjson.JSONObject;
|
||||
import com.alibaba.fastjson.JSONValidator;
|
||||
import com.github.kfcfans.powerjob.worker.core.processor.ProcessResult;
|
||||
import com.github.kfcfans.powerjob.worker.core.processor.TaskContext;
|
||||
@ -20,14 +21,16 @@ import java.util.concurrent.TimeUnit;
|
||||
* common http processor
|
||||
*
|
||||
* @author tjq
|
||||
* @author Jiang Jining
|
||||
* @since 2021/1/30
|
||||
*/
|
||||
public class HttpProcessor extends CommonBasicProcessor {
|
||||
|
||||
/**
|
||||
* 60 seconds
|
||||
* Default timeout is 60 seconds.
|
||||
*/
|
||||
private static final int DEFAULT_TIMEOUT = 60;
|
||||
private static final int HTTP_SUCCESS_CODE = 200;
|
||||
private static final Map<Integer, OkHttpClient> CLIENT_STORE = new ConcurrentHashMap<>();
|
||||
|
||||
@Override
|
||||
@ -35,6 +38,12 @@ public class HttpProcessor extends CommonBasicProcessor {
|
||||
OmsLogger omsLogger = taskContext.getOmsLogger();
|
||||
HttpParams httpParams = JSON.parseObject(CommonUtils.parseParams(taskContext), HttpParams.class);
|
||||
|
||||
if (httpParams == null) {
|
||||
String message = "httpParams is null, please check jobParam configuration.";
|
||||
omsLogger.warn(message);
|
||||
return new ProcessResult(false, message);
|
||||
}
|
||||
|
||||
if (StringUtils.isEmpty(httpParams.url)) {
|
||||
return new ProcessResult(false, "url can't be empty!");
|
||||
}
|
||||
@ -54,9 +63,16 @@ public class HttpProcessor extends CommonBasicProcessor {
|
||||
}
|
||||
|
||||
// set default mediaType
|
||||
if (!"GET".equals(httpParams.method) && StringUtils.isEmpty(httpParams.mediaType) && JSONValidator.from(httpParams.body).validate()) {
|
||||
httpParams.mediaType = "application/json";
|
||||
omsLogger.warn("try to use 'application/json' as media type");
|
||||
if (!"GET".equals(httpParams.method)) {
|
||||
// set default request body
|
||||
if (StringUtils.isEmpty(httpParams.body)) {
|
||||
httpParams.body = new JSONObject().toJSONString();
|
||||
omsLogger.warn("try to use default request body:{}", httpParams.body);
|
||||
}
|
||||
if (JSONValidator.from(httpParams.body).validate() && StringUtils.isEmpty(httpParams.mediaType)) {
|
||||
httpParams.mediaType = "application/json";
|
||||
omsLogger.warn("try to use 'application/json' as media type");
|
||||
}
|
||||
}
|
||||
|
||||
// set default timeout
|
||||
@ -95,9 +111,15 @@ public class HttpProcessor extends CommonBasicProcessor {
|
||||
msgBody = response.body().string();
|
||||
}
|
||||
|
||||
String res = String.format("code:%d,body:%s", response.code(), msgBody);
|
||||
|
||||
return new ProcessResult(true, res);
|
||||
int responseCode = response.code();
|
||||
String res = String.format("code:%d, body:%s", responseCode, msgBody);
|
||||
boolean success = true;
|
||||
if (responseCode != HTTP_SUCCESS_CODE) {
|
||||
success = false;
|
||||
omsLogger.warn("{} url: {} failed, response code is {}, response body is {}",
|
||||
httpParams.method, httpParams.url, responseCode, msgBody);
|
||||
}
|
||||
return new ProcessResult(success, res);
|
||||
}
|
||||
|
||||
@Data
|
||||
|
@ -7,6 +7,8 @@ import com.github.kfcfans.powerjob.worker.log.OmsLogger;
|
||||
import com.google.common.collect.Sets;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.io.FileUtils;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.apache.commons.lang3.SystemUtils;
|
||||
import tech.powerjob.official.processors.CommonBasicProcessor;
|
||||
import tech.powerjob.official.processors.util.CommonUtils;
|
||||
|
||||
@ -20,47 +22,62 @@ import java.util.concurrent.ForkJoinPool;
|
||||
* 脚本处理器
|
||||
*
|
||||
* @author tjq
|
||||
* @author Jiang Jining
|
||||
* @since 2020/4/16
|
||||
*/
|
||||
@Slf4j
|
||||
public abstract class AbstractScriptProcessor extends CommonBasicProcessor {
|
||||
|
||||
private static final ForkJoinPool pool = new ForkJoinPool(4 * Runtime.getRuntime().availableProcessors());
|
||||
private static final ForkJoinPool POOL = new ForkJoinPool(4 * Runtime.getRuntime().availableProcessors());
|
||||
private static final Set<String> DOWNLOAD_PROTOCOL = Sets.newHashSet("http", "https", "ftp");
|
||||
static final String SH_SHELL = "/bin/sh";
|
||||
|
||||
@Override
|
||||
protected ProcessResult process0(TaskContext context) throws Exception {
|
||||
OmsLogger omsLogger = context.getOmsLogger();
|
||||
omsLogger.info("SYSTEM ===> ScriptProcessor start to process");
|
||||
|
||||
String scriptPath = prepareScriptFile(context.getInstanceId(), CommonUtils.parseParams(context));
|
||||
|
||||
// 1. 授权
|
||||
ProcessBuilder chmodPb = new ProcessBuilder("/bin/chmod", "755", scriptPath);
|
||||
// 等待返回,这里不可能导致死锁(shell产生大量数据可能导致死锁)
|
||||
chmodPb.start().waitFor();
|
||||
String scriptParams = CommonUtils.parseParams(context);
|
||||
if (scriptParams == null) {
|
||||
String message = "scriptParams is null, please check jobParam configuration.";
|
||||
omsLogger.warn(message);
|
||||
return new ProcessResult(false, message);
|
||||
}
|
||||
String scriptPath = prepareScriptFile(context.getInstanceId(), scriptParams);
|
||||
|
||||
if (SystemUtils.IS_OS_WINDOWS) {
|
||||
if (StringUtils.equals(getRunCommand(), SH_SHELL)) {
|
||||
String message = String.format("Current OS is %s where shell scripts cannot run.", SystemUtils.OS_NAME);
|
||||
omsLogger.warn(message);
|
||||
return new ProcessResult(false, message);
|
||||
}
|
||||
} else {
|
||||
// 1. 授权
|
||||
ProcessBuilder chmodPb = new ProcessBuilder("/bin/chmod", "755", scriptPath);
|
||||
// 等待返回,这里不可能导致死锁(shell产生大量数据可能导致死锁)
|
||||
chmodPb.start().waitFor();
|
||||
}
|
||||
|
||||
// 2. 执行目标脚本
|
||||
ProcessBuilder pb = new ProcessBuilder(getRunCommand(), scriptPath);
|
||||
Process process = pb.start();
|
||||
|
||||
StringBuilder inputSB = new StringBuilder();
|
||||
StringBuilder errorSB = new StringBuilder();
|
||||
StringBuilder inputBuilder = new StringBuilder();
|
||||
StringBuilder errorBuilder = new StringBuilder();
|
||||
|
||||
boolean success = true;
|
||||
String result;
|
||||
|
||||
try (InputStream is = process.getInputStream(); InputStream es = process.getErrorStream()) {
|
||||
|
||||
pool.execute(() -> copyStream(is, inputSB, omsLogger));
|
||||
pool.execute(() -> copyStream(es, errorSB, omsLogger));
|
||||
POOL.execute(() -> copyStream(is, inputBuilder, omsLogger));
|
||||
POOL.execute(() -> copyStream(es, errorBuilder, omsLogger));
|
||||
|
||||
success = process.waitFor() == 0;
|
||||
|
||||
} catch (InterruptedException ie) {
|
||||
omsLogger.info("SYSTEM ===> ScriptProcessor has been interrupted");
|
||||
} finally {
|
||||
result = String.format("[INPUT]: %s;[ERROR]: %s", inputSB.toString(), errorSB.toString());
|
||||
result = String.format("[INPUT]: %s;[ERROR]: %s", inputBuilder.toString(), errorBuilder.toString());
|
||||
}
|
||||
return new ProcessResult(success, result);
|
||||
}
|
||||
@ -112,6 +129,7 @@ public abstract class AbstractScriptProcessor extends CommonBasicProcessor {
|
||||
|
||||
/**
|
||||
* 生成脚本名称
|
||||
* @param instanceId id of instance
|
||||
* @return 文件名称
|
||||
*/
|
||||
protected abstract String getScriptName(Long instanceId);
|
||||
|
@ -15,6 +15,6 @@ public class ShellProcessor extends AbstractScriptProcessor {
|
||||
|
||||
@Override
|
||||
protected String getRunCommand() {
|
||||
return "/bin/sh";
|
||||
return SH_SHELL;
|
||||
}
|
||||
}
|
||||
|
@ -11,6 +11,14 @@ import tech.powerjob.official.processors.TestUtils;
|
||||
* @since 2021/1/31
|
||||
*/
|
||||
class HttpProcessorTest {
|
||||
|
||||
@Test
|
||||
void testDefaultMethod() throws Exception {
|
||||
String url = "https://www.baidu.com";
|
||||
JSONObject params = new JSONObject();
|
||||
params.put("url", url);
|
||||
System.out.println(new HttpProcessor().process(TestUtils.genTaskContext(params.toJSONString())));
|
||||
}
|
||||
|
||||
@Test
|
||||
void testGet() throws Exception {
|
||||
@ -33,6 +41,25 @@ class HttpProcessorTest {
|
||||
|
||||
System.out.println(new HttpProcessor().process(TestUtils.genTaskContext(params.toJSONString())));
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPostDefaultJson() throws Exception {
|
||||
String url = "https://mock.uutool.cn/4f5qfgcdahj0?test=true";
|
||||
JSONObject params = new JSONObject();
|
||||
params.put("url", url);
|
||||
params.put("method", "POST");
|
||||
System.out.println(new HttpProcessor().process(TestUtils.genTaskContext(params.toJSONString())));
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPostDefaultWithMediaType() throws Exception {
|
||||
String url = "https://mock.uutool.cn/4f5qfgcdahj0?test=true";
|
||||
JSONObject params = new JSONObject();
|
||||
params.put("url", url);
|
||||
params.put("method", "POST");
|
||||
params.put("mediaType", "application/json");
|
||||
System.out.println(new HttpProcessor().process(TestUtils.genTaskContext(params.toJSONString())));
|
||||
}
|
||||
|
||||
@Test
|
||||
void testTimeout() throws Exception {
|
||||
|
@ -5,7 +5,8 @@
|
||||
<parent>
|
||||
<artifactId>powerjob-server</artifactId>
|
||||
<groupId>com.github.kfcfans</groupId>
|
||||
<version>3.4.6</version>
|
||||
<version>4.0.0</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
|
@ -5,7 +5,8 @@
|
||||
<parent>
|
||||
<artifactId>powerjob-server</artifactId>
|
||||
<groupId>com.github.kfcfans</groupId>
|
||||
<version>3.4.6</version>
|
||||
<version>4.0.0</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
|
@ -5,7 +5,8 @@
|
||||
<parent>
|
||||
<artifactId>powerjob-server</artifactId>
|
||||
<groupId>com.github.kfcfans</groupId>
|
||||
<version>3.4.6</version>
|
||||
<version>4.0.0</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
|
@ -5,7 +5,8 @@
|
||||
<parent>
|
||||
<artifactId>powerjob-server</artifactId>
|
||||
<groupId>com.github.kfcfans</groupId>
|
||||
<version>3.4.6</version>
|
||||
<version>4.0.0</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
|
@ -5,7 +5,8 @@
|
||||
<parent>
|
||||
<artifactId>powerjob-server</artifactId>
|
||||
<groupId>com.github.kfcfans</groupId>
|
||||
<version>3.4.6</version>
|
||||
<version>4.0.0</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
|
@ -5,7 +5,8 @@
|
||||
<parent>
|
||||
<artifactId>powerjob-server</artifactId>
|
||||
<groupId>com.github.kfcfans</groupId>
|
||||
<version>3.4.6</version>
|
||||
<version>4.0.0</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
|
@ -5,7 +5,8 @@
|
||||
<parent>
|
||||
<artifactId>powerjob-server</artifactId>
|
||||
<groupId>com.github.kfcfans</groupId>
|
||||
<version>3.4.6</version>
|
||||
<version>4.0.0</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
|
@ -120,7 +120,7 @@ public class OhMyWorker implements ApplicationContextAware, InitializingBean, Di
|
||||
|
||||
ActorRef taskTrackerActorRef = actorSystem.actorOf(TaskTrackerActor.props(workerRuntime)
|
||||
.withDispatcher("akka.task-tracker-dispatcher")
|
||||
.withRouter(new RoundRobinPool(cores * 2)), RemoteConstant.Task_TRACKER_ACTOR_NAME);
|
||||
.withRouter(new RoundRobinPool(cores * 2)), RemoteConstant.TASK_TRACKER_ACTOR_NAME);
|
||||
actorSystem.actorOf(ProcessorTrackerActor.props(workerRuntime)
|
||||
.withDispatcher("akka.processor-tracker-dispatcher")
|
||||
.withRouter(new RoundRobinPool(cores)), RemoteConstant.PROCESSOR_TRACKER_ACTOR_NAME);
|
||||
|
@ -53,7 +53,8 @@ public abstract class MapProcessor implements BasicProcessor {
|
||||
ProcessorMapTaskRequest req = new ProcessorMapTaskRequest(task, taskList, taskName);
|
||||
|
||||
// 2. 可靠发送请求(任务不允许丢失,需要使用 ask 方法,失败抛异常)
|
||||
String akkaRemotePath = AkkaUtils.getAkkaWorkerPath(task.getAddress(), RemoteConstant.Task_TRACKER_ACTOR_NAME);
|
||||
|
||||
String akkaRemotePath = AkkaUtils.getAkkaWorkerPath(task.getAddress(), RemoteConstant.TASK_TRACKER_ACTOR_NAME);
|
||||
boolean requestSucceed = AkkaUtils.reliableTransmit(workerRuntime.getActorSystem().actorSelection(akkaRemotePath), req);
|
||||
|
||||
if (requestSucceed) {
|
||||
|
@ -108,7 +108,8 @@ public class ProcessorTracker {
|
||||
this.instanceInfo = request.getInstanceInfo();
|
||||
this.instanceId = request.getInstanceInfo().getInstanceId();
|
||||
this.taskTrackerAddress = request.getTaskTrackerAddress();
|
||||
String akkaRemotePath = AkkaUtils.getAkkaWorkerPath(taskTrackerAddress, RemoteConstant.Task_TRACKER_ACTOR_NAME);
|
||||
|
||||
String akkaRemotePath = AkkaUtils.getAkkaWorkerPath(taskTrackerAddress, RemoteConstant.TASK_TRACKER_ACTOR_NAME);
|
||||
this.taskTrackerActorRef = workerRuntime.getActorSystem().actorSelection(akkaRemotePath);
|
||||
|
||||
this.omsLogger = new OmsServerLogger(instanceId, workerRuntime.getOmsLogHandler());
|
||||
|
@ -37,7 +37,7 @@ public class CommonTaskTrackerTest {
|
||||
worker.init();
|
||||
|
||||
ActorSystem testAS = ActorSystem.create("oms-test", ConfigFactory.load("oms-akka-test.conf"));
|
||||
String akkaRemotePath = AkkaUtils.getAkkaWorkerPath(NetUtils.getLocalHost() + ":" + RemoteConstant.DEFAULT_WORKER_PORT, RemoteConstant.Task_TRACKER_ACTOR_NAME);
|
||||
String akkaRemotePath = AkkaUtils.getAkkaWorkerPath(NetUtils.getLocalHost() + ":" + RemoteConstant.DEFAULT_WORKER_PORT, RemoteConstant.TASK_TRACKER_ACTOR_NAME);
|
||||
remoteTaskTracker = testAS.actorSelection(akkaRemotePath);
|
||||
}
|
||||
|
||||
|
@ -40,7 +40,7 @@ public class CommonTest {
|
||||
String address = NetUtils.getLocalHost() + ":27777";
|
||||
|
||||
remoteProcessorTracker = testAS.actorSelection(AkkaUtils.getAkkaWorkerPath(address, RemoteConstant.PROCESSOR_TRACKER_ACTOR_NAME));
|
||||
remoteTaskTracker = testAS.actorSelection(AkkaUtils.getAkkaWorkerPath(address, RemoteConstant.Task_TRACKER_ACTOR_NAME));
|
||||
remoteTaskTracker = testAS.actorSelection(AkkaUtils.getAkkaWorkerPath(address, RemoteConstant.TASK_TRACKER_ACTOR_NAME));
|
||||
}
|
||||
|
||||
@AfterAll
|
||||
|
@ -35,7 +35,7 @@ public class FrequentTaskTrackerTest {
|
||||
worker.init();
|
||||
|
||||
ActorSystem testAS = ActorSystem.create("oms-test", ConfigFactory.load("oms-akka-test.conf"));
|
||||
String akkaRemotePath = AkkaUtils.getAkkaWorkerPath(NetUtils.getLocalHost() + ":" + RemoteConstant.DEFAULT_WORKER_PORT, RemoteConstant.Task_TRACKER_ACTOR_NAME);
|
||||
String akkaRemotePath = AkkaUtils.getAkkaWorkerPath(NetUtils.getLocalHost() + ":" + RemoteConstant.DEFAULT_WORKER_PORT, RemoteConstant.TASK_TRACKER_ACTOR_NAME);
|
||||
remoteTaskTracker = testAS.actorSelection(akkaRemotePath);
|
||||
}
|
||||
|
||||
|
Loading…
x
Reference in New Issue
Block a user