package tech.powerjob.client;
|
|
import com.alibaba.fastjson.JSON;
|
import tech.powerjob.common.enums.InstanceStatus;
|
import tech.powerjob.common.OmsConstant;
|
import tech.powerjob.common.OpenAPIConstant;
|
import tech.powerjob.common.exception.PowerJobException;
|
import tech.powerjob.common.request.http.SaveJobInfoRequest;
|
import tech.powerjob.common.request.http.SaveWorkflowNodeRequest;
|
import tech.powerjob.common.request.http.SaveWorkflowRequest;
|
import tech.powerjob.common.request.query.JobInfoQuery;
|
import tech.powerjob.common.response.*;
|
import tech.powerjob.common.utils.CommonUtils;
|
import tech.powerjob.common.utils.HttpUtils;
|
import tech.powerjob.common.serialize.JsonUtils;
|
import com.google.common.collect.Lists;
|
import lombok.extern.slf4j.Slf4j;
|
import okhttp3.FormBody;
|
import okhttp3.MediaType;
|
import okhttp3.RequestBody;
|
import org.apache.commons.lang3.StringUtils;
|
|
import java.io.IOException;
|
import java.util.List;
|
import java.util.Objects;
|
|
import static tech.powerjob.client.TypeStore.*;
|
|
/**
|
* PowerJobClient, the client for OpenAPI.
|
*
|
* @author tjq
|
* @since 2020/4/15
|
*/
|
@Slf4j
|
public class PowerJobClient implements IPowerJobClient {
|
|
private Long appId;
|
private String currentAddress;
|
private final List<String> allAddress;
|
|
private static final String URL_PATTERN = "http://%s%s%s";
|
|
/**
|
* Init PowerJobClient with domain, appName and password.
|
*
|
* @param domain like powerjob-server.apple-inc.com (Intranet Domain)
|
* @param appName name of the application
|
* @param password password of the application
|
*/
|
public PowerJobClient(String domain, String appName, String password) {
|
this(Lists.newArrayList(domain), appName, password);
|
}
|
|
|
/**
|
* Init PowerJobClient with server address, appName and password.
|
*
|
* @param addressList IP:Port address list, like 192.168.1.1:7700
|
* @param appName name of the application
|
* @param password password of the application
|
*/
|
public PowerJobClient(List<String> addressList, String appName, String password) {
|
|
CommonUtils.requireNonNull(addressList, "addressList can't be null!");
|
CommonUtils.requireNonNull(appName, "appName can't be null");
|
|
allAddress = addressList;
|
for (String addr : addressList) {
|
String url = getUrl(OpenAPIConstant.ASSERT, addr);
|
try {
|
String result = assertApp(appName, password, url);
|
if (StringUtils.isNotEmpty(result)) {
|
ResultDTO<Long> resultDTO = JSON.parseObject(result, LONG_RESULT_TYPE);
|
if (resultDTO.isSuccess()) {
|
appId = resultDTO.getData();
|
currentAddress = addr;
|
break;
|
} else {
|
throw new PowerJobException(resultDTO.getMessage());
|
}
|
}
|
} catch (IOException ignore) {
|
//
|
}
|
}
|
|
if (StringUtils.isEmpty(currentAddress)) {
|
throw new PowerJobException("no server available for PowerJobClient");
|
}
|
log.info("[PowerJobClient] {}'s PowerJobClient bootstrap successfully, using server: {}", appName, currentAddress);
|
}
|
|
private static String assertApp(String appName, String password, String url) throws IOException {
|
FormBody.Builder builder = new FormBody.Builder()
|
.add("appName", appName);
|
if (password != null) {
|
builder.add("password", password);
|
}
|
return HttpUtils.post(url, builder.build());
|
}
|
|
|
private static String getUrl(String path, String address) {
|
return String.format(URL_PATTERN, address, OpenAPIConstant.WEB_PATH, path);
|
}
|
|
/* ************* Job 区 ************* */
|
|
/**
|
* Save one Job
|
* When an ID exists in SaveJobInfoRequest, it is an update operation. Otherwise, it is a crate operation.
|
*
|
* @param request Job meta info
|
* @return jobId
|
*/
|
@Override
|
public ResultDTO<Long> saveJob(SaveJobInfoRequest request) {
|
|
request.setAppId(appId);
|
MediaType jsonType = MediaType.parse(OmsConstant.JSON_MEDIA_TYPE);
|
String json = JSON.toJSONString(request);
|
String post = postHA(OpenAPIConstant.SAVE_JOB, RequestBody.create(jsonType, json));
|
return JSON.parseObject(post, LONG_RESULT_TYPE);
|
}
|
|
|
/**
|
* Copy one Job
|
*
|
* @param jobId Job id
|
* @return Id of job copy
|
*/
|
@Override
|
public ResultDTO<Long> copyJob(Long jobId) {
|
RequestBody body = new FormBody.Builder()
|
.add("jobId", jobId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.COPY_JOB, body);
|
return JSON.parseObject(post, LONG_RESULT_TYPE);
|
}
|
|
@Override
|
public ResultDTO<SaveJobInfoRequest> exportJob(Long jobId) {
|
RequestBody body = new FormBody.Builder()
|
.add("jobId", jobId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.EXPORT_JOB, body);
|
return JSON.parseObject(post, SAVE_JOB_INFO_REQUEST_RESULT_TYPE);
|
}
|
|
/**
|
* Query JobInfo by jobId
|
*
|
* @param jobId jobId
|
* @return Job meta info
|
*/
|
@Override
|
public ResultDTO<JobInfoDTO> fetchJob(Long jobId) {
|
RequestBody body = new FormBody.Builder()
|
.add("jobId", jobId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.FETCH_JOB, body);
|
return JSON.parseObject(post, JOB_RESULT_TYPE);
|
}
|
|
/**
|
* Query all JobInfo
|
*
|
* @return All JobInfo
|
*/
|
@Override
|
public ResultDTO<List<JobInfoDTO>> fetchAllJob() {
|
RequestBody body = new FormBody.Builder()
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.FETCH_ALL_JOB, body);
|
return JSON.parseObject(post, LIST_JOB_RESULT_TYPE);
|
}
|
|
/**
|
* Query JobInfo by PowerQuery
|
*
|
* @param powerQuery JobQuery
|
* @return JobInfo
|
*/
|
@Override
|
public ResultDTO<List<JobInfoDTO>> queryJob(JobInfoQuery powerQuery) {
|
powerQuery.setAppIdEq(appId);
|
MediaType jsonType = MediaType.parse(OmsConstant.JSON_MEDIA_TYPE);
|
String json = JsonUtils.toJSONStringUnsafe(powerQuery);
|
String post = postHA(OpenAPIConstant.QUERY_JOB, RequestBody.create(jsonType, json));
|
return JSON.parseObject(post, LIST_JOB_RESULT_TYPE);
|
}
|
|
/**
|
* Disable one Job by jobId
|
*
|
* @param jobId jobId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> disableJob(Long jobId) {
|
RequestBody body = new FormBody.Builder()
|
.add("jobId", jobId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.DISABLE_JOB, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Enable one job by jobId
|
*
|
* @param jobId jobId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> enableJob(Long jobId) {
|
RequestBody body = new FormBody.Builder()
|
.add("jobId", jobId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.ENABLE_JOB, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Delete one job by jobId
|
*
|
* @param jobId jobId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> deleteJob(Long jobId) {
|
RequestBody body = new FormBody.Builder()
|
.add("jobId", jobId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.DELETE_JOB, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Run a job once
|
*
|
* @param jobId ID of the job to be run
|
* @param instanceParams Runtime parameters of the job (TaskContext#instanceParams)
|
* @param delayMS Delay time（Milliseconds）
|
* @return instanceId
|
*/
|
@Override
|
public ResultDTO<Long> runJob(Long jobId, String instanceParams, long delayMS) {
|
FormBody.Builder builder = new FormBody.Builder()
|
.add("jobId", jobId.toString())
|
.add("appId", appId.toString())
|
.add("delay", String.valueOf(delayMS));
|
|
if (StringUtils.isNotEmpty(instanceParams)) {
|
builder.add("instanceParams", instanceParams);
|
}
|
String post = postHA(OpenAPIConstant.RUN_JOB, builder.build());
|
return JSON.parseObject(post, LONG_RESULT_TYPE);
|
}
|
|
public ResultDTO<Long> runJob(Long jobId) {
|
return runJob(jobId, null, 0);
|
}
|
|
/* ************* Instance API list ************* */
|
|
/**
|
* Stop one job instance
|
*
|
* @param instanceId instanceId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> stopInstance(Long instanceId) {
|
RequestBody body = new FormBody.Builder()
|
.add("instanceId", instanceId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.STOP_INSTANCE, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Cancel a job instance that is not yet running
|
* Notice：There is a time interval between the call interface time and the expected execution time of the job instance to be cancelled, otherwise reliability is not guaranteed
|
*
|
* @param instanceId instanceId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> cancelInstance(Long instanceId) {
|
RequestBody body = new FormBody.Builder()
|
.add("instanceId", instanceId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.CANCEL_INSTANCE, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Retry failed job instance
|
* Notice: Only job instance with completion status (success, failure, manually stopped, cancelled) can be retried, and retries of job instances within workflows are not supported yet.
|
*
|
* @param instanceId instanceId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> retryInstance(Long instanceId) {
|
RequestBody body = new FormBody.Builder()
|
.add("instanceId", instanceId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.RETRY_INSTANCE, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Query status about a job instance
|
*
|
* @param instanceId instanceId
|
* @return {@link InstanceStatus}
|
*/
|
@Override
|
public ResultDTO<Integer> fetchInstanceStatus(Long instanceId) {
|
RequestBody body = new FormBody.Builder()
|
.add("instanceId", instanceId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.FETCH_INSTANCE_STATUS, body);
|
return JSON.parseObject(post, INTEGER_RESULT_TYPE);
|
}
|
|
/**
|
* Query detail about a job instance
|
*
|
* @param instanceId instanceId
|
* @return instance detail
|
*/
|
@Override
|
public ResultDTO<InstanceInfoDTO> fetchInstanceInfo(Long instanceId) {
|
RequestBody body = new FormBody.Builder()
|
.add("instanceId", instanceId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.FETCH_INSTANCE_INFO, body);
|
return JSON.parseObject(post, INSTANCE_RESULT_TYPE);
|
}
|
|
/* ************* Workflow API list ************* */
|
|
/**
|
* Save one workflow
|
* When an ID exists in SaveWorkflowRequest, it is an update operation. Otherwise, it is a crate operation.
|
*
|
* @param request Workflow meta info
|
* @return workflowId
|
*/
|
@Override
|
public ResultDTO<Long> saveWorkflow(SaveWorkflowRequest request) {
|
request.setAppId(appId);
|
MediaType jsonType = MediaType.parse(OmsConstant.JSON_MEDIA_TYPE);
|
// 中坑记录：用 FastJSON 序列化会导致 Server 接收时 pEWorkflowDAG 为 null，无语.jpg
|
String json = JsonUtils.toJSONStringUnsafe(request);
|
String post = postHA(OpenAPIConstant.SAVE_WORKFLOW, RequestBody.create(jsonType, json));
|
return JSON.parseObject(post, LONG_RESULT_TYPE);
|
}
|
|
/**
|
* Copy one workflow
|
*
|
* @param workflowId Workflow id
|
* @return Id of workflow copy
|
*/
|
@Override
|
public ResultDTO<Long> copyWorkflow(Long workflowId) {
|
RequestBody body = new FormBody.Builder()
|
.add("workflowId", workflowId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.COPY_WORKFLOW, body);
|
return JSON.parseObject(post, LONG_RESULT_TYPE);
|
}
|
|
|
/**
|
* 添加工作流节点
|
*
|
* @param requestList Node info list of Workflow
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<List<WorkflowNodeInfoDTO>> saveWorkflowNode(List<SaveWorkflowNodeRequest> requestList) {
|
for (SaveWorkflowNodeRequest saveWorkflowNodeRequest : requestList) {
|
saveWorkflowNodeRequest.setAppId(appId);
|
}
|
MediaType jsonType = MediaType.parse(OmsConstant.JSON_MEDIA_TYPE);
|
String json = JsonUtils.toJSONStringUnsafe(requestList);
|
String post = postHA(OpenAPIConstant.SAVE_WORKFLOW_NODE, RequestBody.create(jsonType, json));
|
return JSON.parseObject(post, WF_NODE_LIST_RESULT_TYPE);
|
}
|
|
|
|
/**
|
* Query Workflow by workflowId
|
*
|
* @param workflowId workflowId
|
* @return Workflow meta info
|
*/
|
@Override
|
public ResultDTO<WorkflowInfoDTO> fetchWorkflow(Long workflowId) {
|
RequestBody body = new FormBody.Builder()
|
.add("workflowId", workflowId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.FETCH_WORKFLOW, body);
|
return JSON.parseObject(post, WF_RESULT_TYPE);
|
}
|
|
/**
|
* Disable Workflow by workflowId
|
*
|
* @param workflowId workflowId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> disableWorkflow(Long workflowId) {
|
RequestBody body = new FormBody.Builder()
|
.add("workflowId", workflowId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.DISABLE_WORKFLOW, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Enable Workflow by workflowId
|
*
|
* @param workflowId workflowId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> enableWorkflow(Long workflowId) {
|
RequestBody body = new FormBody.Builder()
|
.add("workflowId", workflowId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.ENABLE_WORKFLOW, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Delete Workflow by workflowId
|
*
|
* @param workflowId workflowId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> deleteWorkflow(Long workflowId) {
|
RequestBody body = new FormBody.Builder()
|
.add("workflowId", workflowId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.DELETE_WORKFLOW, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Run a workflow once
|
*
|
* @param workflowId workflowId
|
* @param initParams workflow startup parameters
|
* @param delayMS Delay time（Milliseconds）
|
* @return workflow instanceId
|
*/
|
@Override
|
public ResultDTO<Long> runWorkflow(Long workflowId, String initParams, long delayMS) {
|
FormBody.Builder builder = new FormBody.Builder()
|
.add("workflowId", workflowId.toString())
|
.add("appId", appId.toString())
|
.add("delay", String.valueOf(delayMS));
|
if (StringUtils.isNotEmpty(initParams)) {
|
builder.add("initParams", initParams);
|
}
|
String post = postHA(OpenAPIConstant.RUN_WORKFLOW, builder.build());
|
return JSON.parseObject(post, LONG_RESULT_TYPE);
|
}
|
|
public ResultDTO<Long> runWorkflow(Long workflowId) {
|
return runWorkflow(workflowId, null, 0);
|
}
|
|
/* ************* Workflow Instance API list ************* */
|
|
/**
|
* Stop one workflow instance
|
*
|
* @param wfInstanceId workflow instanceId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> stopWorkflowInstance(Long wfInstanceId) {
|
RequestBody body = new FormBody.Builder()
|
.add("wfInstanceId", wfInstanceId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.STOP_WORKFLOW_INSTANCE, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Retry one workflow instance
|
*
|
* @param wfInstanceId workflow instanceId
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> retryWorkflowInstance(Long wfInstanceId) {
|
RequestBody body = new FormBody.Builder()
|
.add("wfInstanceId", wfInstanceId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.RETRY_WORKFLOW_INSTANCE, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* mark the workflow node as success
|
*
|
* @param wfInstanceId workflow instanceId
|
* @param nodeId node id
|
* @return Standard return object
|
*/
|
@Override
|
public ResultDTO<Void> markWorkflowNodeAsSuccess(Long wfInstanceId, Long nodeId) {
|
RequestBody body = new FormBody.Builder()
|
.add("wfInstanceId", wfInstanceId.toString())
|
.add("nodeId", nodeId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.MARK_WORKFLOW_NODE_AS_SUCCESS, body);
|
return JSON.parseObject(post, VOID_RESULT_TYPE);
|
}
|
|
/**
|
* Query detail about a workflow instance
|
*
|
* @param wfInstanceId workflow instanceId
|
* @return detail about a workflow
|
*/
|
@Override
|
public ResultDTO<WorkflowInstanceInfoDTO> fetchWorkflowInstanceInfo(Long wfInstanceId) {
|
RequestBody body = new FormBody.Builder()
|
.add("wfInstanceId", wfInstanceId.toString())
|
.add("appId", appId.toString())
|
.build();
|
String post = postHA(OpenAPIConstant.FETCH_WORKFLOW_INSTANCE_INFO, body);
|
return JSON.parseObject(post, WF_INSTANCE_RESULT_TYPE);
|
}
|
|
|
private String postHA(String path, RequestBody requestBody) {
|
|
// 先尝试默认地址
|
String url = getUrl(path, currentAddress);
|
try {
|
String res = HttpUtils.post(url, requestBody);
|
if (StringUtils.isNotEmpty(res)) {
|
return res;
|
}
|
} catch (IOException e) {
|
log.warn("[PowerJobClient] request url:{} failed, reason is {}.", url, e.toString());
|
}
|
|
// 失败，开始重试
|
for (String addr : allAddress) {
|
if (Objects.equals(addr, currentAddress)) {
|
continue;
|
}
|
url = getUrl(path, addr);
|
try {
|
String res = HttpUtils.post(url, requestBody);
|
if (StringUtils.isNotEmpty(res)) {
|
log.warn("[PowerJobClient] server change: from({}) -> to({}).", currentAddress, addr);
|
currentAddress = addr;
|
return res;
|
}
|
} catch (IOException e) {
|
log.warn("[PowerJobClient] request url:{} failed, reason is {}.", url, e.toString());
|
}
|
}
|
|
log.error("[PowerJobClient] do post for path: {} failed because of no server available in {}.", path, allAddress);
|
throw new PowerJobException("no server available when send post request");
|
}
|
}
|
