diff --git a/dolphinscheduler-dist/release-docs/LICENSE b/dolphinscheduler-dist/release-docs/LICENSE index 2308359cdf..ef26c32dfb 100644 --- a/dolphinscheduler-dist/release-docs/LICENSE +++ b/dolphinscheduler-dist/release-docs/LICENSE @@ -416,7 +416,7 @@ The text of each license is also included at licenses/LICENSE-[project].txt. protostuff-runtime 1.7.2: https://github.com/protostuff/protostuff/protostuff-core Apache-2.0 protostuff-api 1.7.2: https://github.com/protostuff/protostuff/protostuff-api Apache-2.0 protostuff-collectionschema 1.7.2: https://github.com/protostuff/protostuff/protostuff-collectionschema Apache-2.0 - + async-http-client 2.12.3: https://mvnrepository.com/artifact/org.asynchttpclient/async-http-client Apache-2.0 ======================================================================== BSD licenses ======================================================================== diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/pom.xml b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/pom.xml new file mode 100644 index 0000000000..5abc8d01be --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/pom.xml @@ -0,0 +1,97 @@ + + + + + dolphinscheduler-task-plugin + org.apache.dolphinscheduler + 1.3.6-SNAPSHOT + + + 4.0.0 + + dolphinscheduler-task-tis + + + + + org.apache.dolphinscheduler + dolphinscheduler-task-api + ${project.version} + + + org.slf4j + slf4j-api + + + + + com.github.dreamhead + moco-core + 1.2.0 + test + + + com.github.dreamhead + moco-runner + 1.2.0 + test + + + commons-cli + commons-cli + + + + + + org.asynchttpclient + async-http-client + 2.12.3 + + + + org.apache.httpcomponents + httpclient + + + org.apache.httpcomponents + httpcore + + + org.apache.dolphinscheduler + dolphinscheduler-common + + + org.mockito + mockito-core + + + org.apache.dolphinscheduler + dolphinscheduler-server + provided + + + org.powermock + powermock-api-mockito2 + + + + + \ No newline at end of file diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/readme.md b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/readme.md new file mode 100644 index 0000000000..e69de29bb2 diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISParameters.java new file mode 100644 index 0000000000..f19c33ed43 --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISParameters.java @@ -0,0 +1,62 @@ +/* + * 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.plugin.task.tis; + +import org.apache.dolphinscheduler.spi.task.AbstractParameters; +import org.apache.dolphinscheduler.spi.task.ResourceInfo; +import org.apache.dolphinscheduler.spi.utils.StringUtils; + +import java.util.Collections; +import java.util.List; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * TIS parameter + */ +public class TISParameters extends AbstractParameters { + + private static final Logger logger = LoggerFactory.getLogger(TISParameters.class); + /** + * TIS target job name + */ + private String targetJobName; + + public String getTargetJobName() { + return targetJobName; + } + + public void setTargetJobName(String targetJobName) { + this.targetJobName = targetJobName; + } + + @Override + public boolean checkParameters() { + if (StringUtils.isBlank(this.targetJobName)) { + logger.error("checkParameters faild targetJobName can not be null"); + return false; + } + return true; + } + + @Override + public List getResourceFilesList() { + return Collections.emptyList(); + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISParamsConstants.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISParamsConstants.java new file mode 100644 index 0000000000..54dfcebb7a --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISParamsConstants.java @@ -0,0 +1,27 @@ +/* + * 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.plugin.task.tis; + +public class TISParamsConstants { + + public static String NAME_TARGET_JOB_NAME = "targetJobName"; + public static String TARGET_JOB_NAME = NAME_TARGET_JOB_NAME; + + private TISParamsConstants() { + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTask.java new file mode 100644 index 0000000000..4a70b6cace --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTask.java @@ -0,0 +1,345 @@ +/* + * 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.plugin.task.tis; + +import org.apache.dolphinscheduler.common.utils.CollectionUtils; +import org.apache.dolphinscheduler.spi.task.AbstractParameters; +import org.apache.dolphinscheduler.spi.task.AbstractTask; +import org.apache.dolphinscheduler.spi.task.TaskConstants; +import org.apache.dolphinscheduler.spi.task.TaskRequest; +import org.apache.dolphinscheduler.spi.utils.JSONUtils; +import org.apache.dolphinscheduler.spi.utils.StringUtils; + +import org.apache.http.HttpEntity; +import org.apache.http.StatusLine; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.util.EntityUtils; + +import java.net.HttpURLConnection; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; + +import org.asynchttpclient.Dsl; +import org.asynchttpclient.ws.WebSocket; +import org.asynchttpclient.ws.WebSocketListener; +import org.asynchttpclient.ws.WebSocketUpgradeHandler; +import org.slf4j.Logger; + +/** + * TIS DataX Task + **/ +public class TISTask extends AbstractTask { + + public static final String WS_REQUEST_PATH = "/tjs/download/logfeedback"; + public static final String KEY_POOL_VAR_TIS_HOST = "tisHost"; + private final TaskRequest taskExecutionContext; + + private TISParameters tisParameters; + + public TISTask(TaskRequest taskExecutionContext, Logger logger) { + super(taskExecutionContext, logger); + this.taskExecutionContext = taskExecutionContext; + } + + @Override + public void init() { + super.init(); + logger.info("tis task params {}", taskExecutionContext.getTaskParams()); + tisParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), TISParameters.class); + if (!tisParameters.checkParameters()) { + throw new RuntimeException("datax task params is not valid"); + } + } + + @Override + public void handle() throws Exception { + // Trigger TIS DataX pipeline + logger.info("start execute TIS task"); + long startTime = System.currentTimeMillis(); + String targetJobName = this.tisParameters.getTargetJobName(); + final String tisHost = taskExecutionContext.getDefinedParams().get(KEY_POOL_VAR_TIS_HOST); + if (StringUtils.isEmpty(tisHost)) { + throw new IllegalStateException("global var '" + KEY_POOL_VAR_TIS_HOST + "' can not be empty"); + } + try { + final String triggerUrl = String.format("http://%s/tjs/coredefine/coredefine.ajax", tisHost); + final String getStatusUrl = String.format("http://%s/tjs/config/config.ajax?action=collection_action&emethod=get_task_status", tisHost); + HttpPost post = new HttpPost(triggerUrl); + post.addHeader("appname", targetJobName); + addFormUrlencoded(post); + StringEntity entity = new StringEntity("action=datax_action&emethod=trigger_fullbuild_task", StandardCharsets.UTF_8); + post.setEntity(entity); + BizResult ajaxResult = null; + ExecResult execState = null; + int taskId; + WebSocket webSocket = null; + try (CloseableHttpClient client = HttpClients.createDefault(); + // trigger to start TIS dataX task + CloseableHttpResponse response = client.execute(post)) { + ajaxResult = processResponse(triggerUrl, response, BizResult.class); + if (!ajaxResult.isSuccess()) { + List errormsg = ajaxResult.getErrormsg(); + StringBuffer errs = new StringBuffer(); + if (CollectionUtils.isNotEmpty(errormsg)) { + errs.append(",errs:").append(errormsg.stream().collect(Collectors.joining(","))); + } + throw new Exception("trigger TIS job faild taskName:" + targetJobName + errs.toString()); + } + taskId = ajaxResult.getBizresult().getTaskid(); + + webSocket = receiveRealtimeLog(tisHost, targetJobName, taskId); + + setAppIds(String.valueOf(taskId)); + + CloseableHttpResponse status = null; + + while (true) { + try { + post = new HttpPost(getStatusUrl); + entity = new StringEntity("{\n taskid: " + taskId + "\n, log: false }", StandardCharsets.UTF_8); + post.setEntity(entity); + status = client.execute(post); + StatusResult execStatus = processResponse(getStatusUrl, status, StatusResult.class); + Map bizresult = execStatus.getBizresult(); + Map s = (Map) bizresult.get("status"); + execState = ExecResult.parse((Integer) s.get("state")); + if (execState == ExecResult.SUCCESS || execState == ExecResult.FAILD) { + break; + } + Thread.sleep(3000); + } finally { + status.close(); + } + } + } finally { + try { + webSocket.sendCloseFrame(); + } catch (Throwable e) { + logger.warn(e.getMessage(), e); + } + } + + long costTime = System.currentTimeMillis() - startTime; + logger.info("TIS task: {},taskId:{} costTime : {} milliseconds, statusCode : {}", + targetJobName, taskId, costTime, (execState == ExecResult.SUCCESS) ? "'success'" : "'failure'"); + setExitStatusCode((execState == ExecResult.SUCCESS) ? TaskConstants.EXIT_CODE_SUCCESS : TaskConstants.EXIT_CODE_FAILURE); + } catch (Exception e) { + logger.error("execute TIS dataX faild,TIS task name:" + targetJobName, e); + setExitStatusCode(TaskConstants.EXIT_CODE_FAILURE); + } + } + + private void addFormUrlencoded(HttpPost post) { + post.addHeader("content-type", "application/x-www-form-urlencoded"); + } + + @Override + public void cancelApplication(boolean status) throws Exception { + super.cancelApplication(status); + } + + private WebSocket receiveRealtimeLog(final String tisHost, String dataXName, int taskId) throws InterruptedException, java.util.concurrent.ExecutionException { + + WebSocketUpgradeHandler.Builder upgradeHandlerBuilder + = new WebSocketUpgradeHandler.Builder(); + WebSocketUpgradeHandler wsHandler = upgradeHandlerBuilder + .addWebSocketListener(new WebSocketListener() { + @Override + public void onOpen(WebSocket websocket) { + // WebSocket connection opened + } + + @Override + public void onClose(WebSocket websocket, int code, String reason) { + // WebSocket connection closed + } + + public void onTextFrame(String payload, boolean finalFragment, int rsv) { + ExecLog execLog = JSONUtils.parseObject(payload, ExecLog.class); + logger.info(execLog.getMsg()); + } + + @Override + public void onError(Throwable t) { + // WebSocket connection error + logger.error(t.getMessage(), t); + } + }).build(); + WebSocket webSocketClient = Dsl.asyncHttpClient() + .prepareGet(String.format("ws://%s" + WS_REQUEST_PATH, tisHost)) + // .addHeader("header_name", "header_value") + .addQueryParam("logtype", "full") + .addQueryParam("collection", dataXName) + .addQueryParam("taskid", String.valueOf(taskId)) + .setRequestTimeout(5000) + .execute(wsHandler) + .get(); + + return webSocketClient; + } + + private T processResponse(String applyUrl, CloseableHttpResponse response, Class clazz) throws Exception { + StatusLine resStatus = response.getStatusLine(); + if (HttpURLConnection.HTTP_OK != resStatus.getStatusCode()) { + throw new IllegalStateException("request server " + applyUrl + " faild:" + resStatus.getReasonPhrase()); + } + HttpEntity entity = response.getEntity(); + String resp = EntityUtils.toString(entity, StandardCharsets.UTF_8); + T result = JSONUtils.parseObject(resp, clazz); + return result; + } + + @Override + public AbstractParameters getParameters() { + Objects.requireNonNull(this.tisParameters, "tisParameters can not be null"); + return this.tisParameters; + } + + private static class BizResult extends AjaxResult { + private TriggerBuildResult bizresult; + + public TriggerBuildResult getBizresult() { + return this.bizresult; + } + + public void setBizresult(TriggerBuildResult bizresult) { + this.bizresult = bizresult; + } + } + + private static class StatusResult extends AjaxResult { + private Map bizresult; + + public Map getBizresult() { + return this.bizresult; + } + + public void setBizresult(Map bizresult) { + this.bizresult = bizresult; + } + } + + private abstract static class AjaxResult { + + private boolean success; + + private List errormsg; + + private List msg; + + public abstract T getBizresult(); + + public boolean isSuccess() { + return success; + } + + public void setSuccess(boolean success) { + this.success = success; + } + + public List getErrormsg() { + return this.errormsg; + } + + public void setErrormsg(List errormsg) { + this.errormsg = errormsg; + } + + public List getMsg() { + return this.msg; + } + + public void setMsg(List msg) { + this.msg = msg; + } + + } + + private static class TriggerBuildResult { + private int taskid; + + public int getTaskid() { + return taskid; + } + + public void setTaskid(int taskid) { + this.taskid = taskid; + } + } + + private enum ExecResult { + + SUCCESS(1), FAILD(-1), DOING(2), ASYN_DOING(22), CANCEL(3); + + private final int value; + + public static ExecResult parse(int value) { + for (ExecResult r : values()) { + if (r.value == value) { + return r; + } + } + throw new IllegalStateException("vale:" + value + " is illegal"); + } + + private ExecResult(int value) { + this.value = value; + } + + public int getValue() { + return this.value; + } + } + + private static class ExecLog { + private String logType; + private String msg; + private int taskId; + + public String getLogType() { + return logType; + } + + public void setLogType(String logType) { + this.logType = logType; + } + + public String getMsg() { + return msg; + } + + public void setMsg(String msg) { + this.msg = msg; + } + + public int getTaskId() { + return taskId; + } + + public void setTaskId(int taskId) { + this.taskId = taskId; + } + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskChannel.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskChannel.java new file mode 100644 index 0000000000..467ad7c89b --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskChannel.java @@ -0,0 +1,34 @@ +/* + * 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.plugin.task.tis; + +import org.apache.dolphinscheduler.spi.task.AbstractTask; +import org.apache.dolphinscheduler.spi.task.TaskChannel; +import org.apache.dolphinscheduler.spi.task.TaskRequest; + +public class TISTaskChannel implements TaskChannel { + @Override + public void cancelApplication(boolean status) { + + } + + @Override + public AbstractTask createTask(TaskRequest taskRequest, org.slf4j.Logger logger) { + return new TISTask(taskRequest, logger); + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskChannelFactory.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskChannelFactory.java new file mode 100644 index 0000000000..c00f5ee8f1 --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskChannelFactory.java @@ -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.plugin.task.tis; + +import org.apache.dolphinscheduler.spi.params.InputParam; +import org.apache.dolphinscheduler.spi.params.base.PluginParams; +import org.apache.dolphinscheduler.spi.params.base.Validate; +import org.apache.dolphinscheduler.spi.task.TaskChannel; +import org.apache.dolphinscheduler.spi.task.TaskChannelFactory; + +import java.util.Arrays; +import java.util.List; + +/** + * TIS endpoint + **/ +public class TISTaskChannelFactory implements TaskChannelFactory { + + @Override + public TaskChannel create() { + return new TISTaskChannel(); + } + + @Override + public String getName() { + return "TIS"; + } + + @Override + public List getParams() { + InputParam webHookParam = InputParam.newBuilder(TISParamsConstants.NAME_TARGET_JOB_NAME, TISParamsConstants.TARGET_JOB_NAME) + .addValidate(Validate.newBuilder().setRequired(true).build()) + .build(); + return Arrays.asList(webHookParam); + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/test/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/test/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskTest.java new file mode 100644 index 0000000000..a118edb9ca --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/test/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskTest.java @@ -0,0 +1,142 @@ +/* + * 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.plugin.task.tis; + +import static com.github.dreamhead.moco.Moco.pathResource; +import static com.github.dreamhead.moco.MocoJsonRunner.jsonHttpServer; +import static com.github.dreamhead.moco.Runner.running; + +import org.apache.dolphinscheduler.server.worker.task.TaskProps; +import org.apache.dolphinscheduler.spi.task.ExecutionStatus; +import org.apache.dolphinscheduler.spi.task.TaskRequest; + +import org.apache.commons.io.IOUtils; + +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.Date; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; +import org.powermock.api.mockito.PowerMockito; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.github.dreamhead.moco.HttpServer; + +public class TISTaskTest { + private static final Logger logger = LoggerFactory.getLogger(TISTaskTest.class); + private TISTask tisTask; + + private TaskRequest taskExecutionContext; + + @Before + public void before() throws Exception { + + TaskProps props = new TaskProps(); + props.setExecutePath("/tmp"); + props.setTaskAppId(String.valueOf(System.currentTimeMillis())); + props.setTaskInstanceId(1); + props.setTenantCode("1"); + props.setEnvFile(".dolphinscheduler_env.sh"); + props.setTaskStartTime(new Date()); + props.setTaskTimeout(0); + props.setTaskParams("{\"targetJobName\":\"mysql_elastic\"}"); + + taskExecutionContext = Mockito.mock(TaskRequest.class); + Mockito.when(taskExecutionContext.getTaskParams()).thenReturn(props.getTaskParams()); + Mockito.when(taskExecutionContext.getExecutePath()).thenReturn("/tmp"); + Mockito.when(taskExecutionContext.getTaskAppId()).thenReturn(UUID.randomUUID().toString()); + Mockito.when(taskExecutionContext.getTenantCode()).thenReturn("root"); + Mockito.when(taskExecutionContext.getStartTime()).thenReturn(new Date()); + Mockito.when(taskExecutionContext.getTaskTimeout()).thenReturn(10000); + Mockito.when(taskExecutionContext.getLogPath()).thenReturn("/tmp/dx"); + // Mockito.when(taskExecutionContext.getVarPool()) + // .thenReturn("[{\"direct\":\"IN\",\"prop\":\"" + TISTask.KEY_POOL_VAR_TIS_HOST + "\",\"type\":\"VARCHAR\",\"value\":\"127.0.0.1:8080\"}]"); + Map gloabParams = Collections.singletonMap(TISTask.KEY_POOL_VAR_TIS_HOST, "127.0.0.1:8080"); + Mockito.when(taskExecutionContext.getDefinedParams()).thenReturn(gloabParams); + + tisTask = PowerMockito.spy(new TISTask(taskExecutionContext, logger)); + tisTask.init(); + + } + + /** + * Method: DataxTask() + */ + @Test + public void testDataxTask() + throws Exception { + TaskProps props = new TaskProps(); + props.setExecutePath("/tmp"); + props.setTaskAppId(String.valueOf(System.currentTimeMillis())); + props.setTaskInstanceId(1); + props.setTenantCode("1"); + Assert.assertNotNull(new TISTask(null, logger)); + } + + @Test + public void testInit() + throws Exception { + try { + tisTask.init(); + } catch (Exception e) { + Assert.fail(e.getMessage()); + } + } + + @Test + public void testHandle() + throws Exception { + HttpServer server = jsonHttpServer(8080, pathResource("org/apache/dolphinscheduler/plugin/task/tis/TISTaskTest.json")); + + running(server, () -> { + tisTask.handle(); + + Assert.assertEquals("TIS execute be success", ExecutionStatus.SUCCESS, tisTask.getExitStatus()); + }); + } + + private String loadResContent(String resName) { + try (InputStream i = this.getClass().getResourceAsStream(resName)) { + Objects.requireNonNull(i, "resource " + resName + " relevant stream content can not be null"); + String content = IOUtils.toString(i, StandardCharsets.UTF_8); + + return content; + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + @Test + public void testCancelApplication() + throws Exception { + try { + tisTask.cancelApplication(true); + } catch (Exception e) { + Assert.fail(e.getMessage()); + } + } + +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/test/resources/org/apache/dolphinscheduler/plugin/task/tis/TISTaskTest.json b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/test/resources/org/apache/dolphinscheduler/plugin/task/tis/TISTaskTest.json new file mode 100644 index 0000000000..fe11b39fda --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/test/resources/org/apache/dolphinscheduler/plugin/task/tis/TISTaskTest.json @@ -0,0 +1,60 @@ +[ + { + "description": "trigger TIS task execute", + "request": { + "uri": "/tjs/coredefine/coredefine.ajax", + "method": "post", + "headers": { + "Content-Type": "application/x-www-form-urlencoded", + "appname": "mysql_elastic" + }, + "text": "action=datax_action&emethod=trigger_fullbuild_task" + }, + "response": { + "text": "{\n \"success\":true,\n \"errormsg\":[],\n \"msg\":[],\n \"bizresult\":{\"taskid\": \"1087\"}\n}" + } + }, + { + "description": "Get TIS task execute status", + "request": { + "uri": "/tjs/config/config.ajax", + "method": "post", + "headers": { + "Content-Type": "text/plain; charset=UTF-8" + }, + "queries": { + "action": "collection_action", + "emethod": "get_task_status" + }, + "text": "{\n taskid: 1087\n, log: false }" + }, + "response": { + "seq": [ + { + "text": "{\n \"success\": true,\n \"errormsg\": [\n \"err1\"\n ],\n \"bizresult\": {\n \"status\": {\n \"state\": 2\n }\n }\n}" + }, + { + "text": "{\n \"success\": true,\n \"errormsg\": [\n \"err1\"\n ],\n \"bizresult\": {\n \"status\": {\n \"state\": 1\n }\n }\n}" + } + ] + } + }, + { + "websocket": { + "uri": "/tjs/download/logfeedback", + "connected": "connected", + "sessions": [ + { + "request": { + "text": "logtype=full&collection=mysql_elastic&taskid=1087" + }, + "response": { + "broadcast": { + "content": "{\n \"logType\": \"FULL\",\n \"msg\": \"message 1\",\n \"taskId\": \"1087\"\n}" + } + } + } + ] + } + } +] \ No newline at end of file diff --git a/dolphinscheduler-task-plugin/pom.xml b/dolphinscheduler-task-plugin/pom.xml index 33ffb9d54f..d6459fee88 100644 --- a/dolphinscheduler-task-plugin/pom.xml +++ b/dolphinscheduler-task-plugin/pom.xml @@ -34,6 +34,7 @@ dolphinscheduler-task-flink dolphinscheduler-task-python dolphinscheduler-task-spark + dolphinscheduler-task-tis diff --git a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/config.js b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/config.js index fb6c12308a..b63c899cf2 100755 --- a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/config.js +++ b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/config.js @@ -301,6 +301,10 @@ const tasksType = { desc: 'DataX', color: '#1fc747' }, + TIS: { + desc: 'TIS', + color: '#1fc747' + }, SQOOP: { desc: 'SQOOP', color: '#E46F13' diff --git a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/dag.scss b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/dag.scss index 08f2feccbe..5627b903ae 100755 --- a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/dag.scss +++ b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/dag.scss @@ -107,6 +107,9 @@ .icos-DATAX { background: url("../img/toolbar_DATAX.png") no-repeat 50% 50%; } + .icos-TIS { + background: url("../img/toolbar_TIS.svg") no-repeat 50% 50%; + } .icos-SQOOP { background: url("../img/toolbar_SQOOP.png") no-repeat 50% 50%; } diff --git a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/formModel.vue b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/formModel.vue index 7dfcbc62bb..08223a28a6 100644 --- a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/formModel.vue +++ b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/formModel.vue @@ -243,6 +243,13 @@ ref="DATAX" :backfill-item="backfillItem"> + + +
+ +
{{$t('TargetJobName')}}
+
+ + +
+ +
+ + diff --git a/dolphinscheduler-ui/src/js/conf/home/pages/dag/img/toolbar_TIS.svg b/dolphinscheduler-ui/src/js/conf/home/pages/dag/img/toolbar_TIS.svg new file mode 100644 index 0000000000..3c1b452b22 --- /dev/null +++ b/dolphinscheduler-ui/src/js/conf/home/pages/dag/img/toolbar_TIS.svg @@ -0,0 +1,46 @@ + + + + + + + + + + + + + + + + + + + + + + diff --git a/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js b/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js index 07803e8e89..51a46eba16 100755 --- a/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js +++ b/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js @@ -586,6 +586,8 @@ export default { 'Spark Version': 'Spark版本', TargetDataBase: '目标库', TargetTable: '目标表', + TargetJobName: 'TIS目标任务名', + 'Please enter TIS DataX job name': '请输入TIS DataX任务名', 'Please enter the table of target': '请输入目标表名', 'Please enter a Target Table(required)': '请输入目标表(必填)', SpeedByte: '限流(字节数)', diff --git a/pom.xml b/pom.xml index 522d9b1ab9..c97650dec3 100644 --- a/pom.xml +++ b/pom.xml @@ -1093,6 +1093,7 @@ **/alert/processor/AlertRequestProcessorTest.java **/alert/runner/AlertSenderTest.java **/alert/AlertServerTest.java + **/plugin/task/tis/TISTaskTest.java diff --git a/tools/dependencies/known-dependencies.txt b/tools/dependencies/known-dependencies.txt index 8ff0c29053..17dfe9d138 100755 --- a/tools/dependencies/known-dependencies.txt +++ b/tools/dependencies/known-dependencies.txt @@ -16,7 +16,7 @@ api-util-1.0.0-M20.jar asm-3.1.jar asm-6.2.1.jar aspectjweaver-1.9.6.jar -async-http-client-1.6.5.jar +async-http-client-2.12.3.jar audience-annotations-0.5.0.jar avro-1.7.4.jar aws-java-sdk-1.7.4.jar @@ -252,4 +252,4 @@ xml-apis-1.4.01.jar xmlbeans-3.1.0.jar xmlenc-0.52.jar xz-1.0.jar -zookeeper-3.4.14.jar +zookeeper-3.4.14.jar \ No newline at end of file