Browse Source

[FEATURE][Task Plugin]Add remote-shell task plugin (#13801)

3.2.0-release
JieguangZhou 2 years ago committed by GitHub
parent
commit
273a5e78eb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 16
      docs/configs/docsdev.js
  2. 15
      docs/docs/en/guide/datasource/ssh.md
  3. 31
      docs/docs/en/guide/task/remoteshell.md
  4. 15
      docs/docs/zh/guide/datasource/ssh.md
  5. 30
      docs/docs/zh/guide/task/remoteshell.md
  6. BIN
      docs/img/new_ui/dev/datasource/ssh.png
  7. BIN
      docs/img/tasks/demo/remote-shell.png
  8. BIN
      docs/img/tasks/icons/remoteshell.png
  9. 15
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/DataSourceServiceImpl.java
  10. 1
      dolphinscheduler-api/src/main/resources/task-type-config.yaml
  11. 12
      dolphinscheduler-bom/pom.xml
  12. 5
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-all/pom.xml
  13. 10
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-api/src/main/java/org/apache/dolphinscheduler/plugin/datasource/api/datasource/DataSourceProcessor.java
  14. 47
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/pom.xml
  15. 32
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/SSHDataSourceChannel.java
  16. 37
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/SSHDataSourceChannelFactory.java
  17. 30
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/SSHDataSourceClient.java
  18. 62
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/SSHUtils.java
  19. 39
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/param/SSHConnectionParam.java
  20. 35
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/param/SSHDataSourceParamDTO.java
  21. 139
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/param/SSHDataSourceProcessor.java
  22. 119
      dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/test/java/org/apache/dolphinscheduler/plugin/datasource/ssh/SSHDataSourceProcessorTest.java
  23. 1
      dolphinscheduler-datasource-plugin/pom.xml
  24. 3
      dolphinscheduler-dist/release-docs/LICENSE
  25. 202
      dolphinscheduler-dist/release-docs/licenses/LICENSE-sshd-scp.txt
  26. 202
      dolphinscheduler-dist/release-docs/licenses/LICENSE-sshd-sftp.txt
  27. 8
      dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/datasource/ConnectionParam.java
  28. 3
      dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/enums/DbType.java
  29. 5
      dolphinscheduler-task-plugin/dolphinscheduler-task-all/pom.xml
  30. 57
      dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/pom.xml
  31. 251
      dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/main/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteExecutor.java
  32. 50
      dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/main/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteShellParameters.java
  33. 185
      dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/main/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteShellTask.java
  34. 48
      dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/main/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteShellTaskChannel.java
  35. 65
      dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/main/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteShellTaskChannelFactory.java
  36. 136
      dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/test/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteExecutorTest.java
  37. 92
      dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/test/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteShellTaskTest.java
  38. 1
      dolphinscheduler-task-plugin/pom.xml
  39. BIN
      dolphinscheduler-ui/public/images/task-icons/remoteshell.png
  40. BIN
      dolphinscheduler-ui/public/images/task-icons/remoteshell_hover.png
  41. 3
      dolphinscheduler-ui/src/service/modules/data-source/types.ts
  42. 4
      dolphinscheduler-ui/src/store/project/task-type.ts
  43. 1
      dolphinscheduler-ui/src/store/project/types.ts
  44. 18
      dolphinscheduler-ui/src/views/datasource/list/detail.tsx
  45. 23
      dolphinscheduler-ui/src/views/datasource/list/use-form.ts
  46. 1
      dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts
  47. 7
      dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-datasource.ts
  48. 37
      dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-remote-shell.ts
  49. 5
      dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
  50. 4
      dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts
  51. 74
      dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-remote-shell.ts
  52. 5
      dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts
  53. 6
      dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss
  54. 7
      tools/dependencies/known-dependencies.txt

16
docs/configs/docsdev.js

@ -221,6 +221,10 @@ export default {
title: 'Apache Linkis', title: 'Apache Linkis',
link: '/en-us/docs/dev/user_doc/guide/task/linkis.html', link: '/en-us/docs/dev/user_doc/guide/task/linkis.html',
}, },
{
title: 'SSH',
link: '/en-us/docs/dev/user_doc/guide/task/ssh.html',
},
], ],
}, },
{ {
@ -319,6 +323,10 @@ export default {
title: 'OceanBase', title: 'OceanBase',
link: '/en-us/docs/dev/user_doc/guide/datasource/oceanbase.html', link: '/en-us/docs/dev/user_doc/guide/datasource/oceanbase.html',
}, },
{
title: 'SSH',
link: '/en-us/docs/dev/user_doc/guide/datasource/ssh.html',
},
], ],
}, },
{ {
@ -906,6 +914,10 @@ export default {
title: 'Apache Linkis', title: 'Apache Linkis',
link: '/zh-cn/docs/dev/user_doc/guide/task/linkis.html', link: '/zh-cn/docs/dev/user_doc/guide/task/linkis.html',
}, },
{
title: 'SSH',
link: '/zh-cn/docs/dev/user_doc/guide/task/ssh.html',
},
], ],
}, },
{ {
@ -988,6 +1000,10 @@ export default {
title: 'OceanBase', title: 'OceanBase',
link: '/zh-cn/docs/dev/user_doc/guide/datasource/oceanbase.html', link: '/zh-cn/docs/dev/user_doc/guide/datasource/oceanbase.html',
}, },
{
title: 'SSH',
link: '/zh-cn/docs/dev/user_doc/guide/datasource/ssh.html',
},
], ],
}, },
{ {

15
docs/docs/en/guide/datasource/ssh.md

@ -0,0 +1,15 @@
# SSH Data Source
This data source is used for RemoteShell component to execute commands remotely.
![sh](../../../../img/new_ui/dev/datasource/ssh.png)
- Data Source: SSH
- Data Source Name: Enter the name of the data source
- Description: Enter the description of the data source
- IP Hostname: Enter the IP to connect to SSH
- Port: Enter the port to connect to SSH
- Username: Set the username to connect to SSH
- Password: Set the password to connect to SSH
- Public Key: Set the public key to connect to SSH

31
docs/docs/en/guide/task/remoteshell.md

@ -0,0 +1,31 @@
# RemoteShell
## Overview
RemoteShell task type is used to execute commands on remote servers.
## Create Task
- Click Project Management-Project Name-Workflow Definition, click the "Create Workflow" button to enter the DAG editing page.
- Drag <img src="../../../../img/tasks/icons/remoteshell.png" width="15"/> from the toolbar to the canvas to complete the creation.
## Task Parameters
[//]: # (TODO: use the commented anchor below once our website template supports this syntax)
[//]: # (- Please refer to [DolphinScheduler Task Parameters Appendix]&#40;appendix.md#default-task-parameters&#41; `Default Task Parameters` section for default parameters.)
- Please refer to [DolphinScheduler Task Parameters Appendix](appendix.md) `Default Task Parameters` section for default parameters.
- SSH Data Source: Select SSH data source.
## Task Example
### View the path of the remote server (remote-server)
![remote-shell-demo](../../../../img/tasks/demo/remote-shell.png)
## Precautions
After the task connects to the server, it will not automatically source bashrc and other files. The required environment variables can be imported in the following ways
- Create environment variables in the security center-Environment Management, and then import them through the environment option in the task definition
- Enter the corresponding environment variables directly in the script

15
docs/docs/zh/guide/datasource/ssh.md

@ -0,0 +1,15 @@
# SSH 数据源
该数据源用于RemoteShell组件,用于远程执行命令。
![sh](../../../../img/new_ui/dev/datasource/ssh.png)
- 数据源:选择 SSH
- 数据源名称:输入数据源的名称
- 描述:输入数据源的描述
- IP 主机名:输入连接 SSH 的 IP
- 端口:输入连接 SSH 的端口
- 用户名:设置连接 SSH 的用户名
- 密码:设置连接 SSH 的密码
- 公钥:设置连接 SSH 的公钥

30
docs/docs/zh/guide/task/remoteshell.md

@ -0,0 +1,30 @@
# RemoteShell
## 综述
RemoteShell 任务类型,用于在远程服务器上执行命令。
## 创建任务
- 点击项目管理-项目名称-工作流定义,点击"创建工作流"按钮,进入 DAG 编辑页面。
- 工具栏中拖动 <img src="../../../../img/tasks/icons/remoteshell.png" width="15"/> 到画板中,即可完成创建。
## 任务参数
[//]: # (TODO: use the commented anchor below once our website template supports this syntax)
[//]: # (- 默认参数说明请参考[DolphinScheduler任务参数附录]&#40;appendix.md#默认任务参数&#41;`默认任务参数`一栏。)
- 默认参数说明请参考[DolphinScheduler任务参数附录](appendix.md)`默认任务参数`一栏。
- SSH Data Source: 选择SSH 数据源。
## 任务样例
### 查看远程服务器(remote-server)的路径
![remote-shell-demo](../../../../img/tasks/demo/remote-shell.png)
## 注意事项
该任务连接服务器后,不会自动source bashrc等文件,所需的环境变量,可以通过以下方式导入
- 在安全中心-环境管理中创建环境变量,然后通过任务定义中的环境选项引入
- 在脚本中直接输入对应的环境变量

BIN
docs/img/new_ui/dev/datasource/ssh.png

Binary file not shown.

After

Width:  |  Height:  |  Size: 40 KiB

BIN
docs/img/tasks/demo/remote-shell.png

Binary file not shown.

After

Width:  |  Height:  |  Size: 76 KiB

BIN
docs/img/tasks/icons/remoteshell.png

Binary file not shown.

After

Width:  |  Height:  |  Size: 747 B

15
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/DataSourceServiceImpl.java

@ -35,6 +35,7 @@ import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.DataSourceMapper; import org.apache.dolphinscheduler.dao.mapper.DataSourceMapper;
import org.apache.dolphinscheduler.dao.mapper.DataSourceUserMapper; import org.apache.dolphinscheduler.dao.mapper.DataSourceUserMapper;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO; import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.DataSourceProcessor;
import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceClientProvider; import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceClientProvider;
import org.apache.dolphinscheduler.plugin.datasource.api.utils.DataSourceUtils; import org.apache.dolphinscheduler.plugin.datasource.api.utils.DataSourceUtils;
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam; import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
@ -186,9 +187,10 @@ public class DataSourceServiceImpl extends BaseServiceImpl implements DataSource
return result; return result;
} }
// check password,if the password is not updated, set to the old password. // check password,if the password is not updated, set to the old password.
BaseConnectionParam connectionParam = ConnectionParam connectionParam = DataSourceUtils.buildConnectionParams(dataSourceParam);
(BaseConnectionParam) DataSourceUtils.buildConnectionParams(dataSourceParam);
String password = connectionParam.getPassword(); String password = connectionParam.getPassword();
if (StringUtils.isBlank(password)) { if (StringUtils.isBlank(password)) {
String oldConnectionParams = dataSource.getConnectionParams(); String oldConnectionParams = dataSource.getConnectionParams();
ObjectNode oldParams = JSONUtils.parseObject(oldConnectionParams); ObjectNode oldParams = JSONUtils.parseObject(oldConnectionParams);
@ -383,6 +385,15 @@ public class DataSourceServiceImpl extends BaseServiceImpl implements DataSource
@Override @Override
public Result<Object> checkConnection(DbType type, ConnectionParam connectionParam) { public Result<Object> checkConnection(DbType type, ConnectionParam connectionParam) {
Result<Object> result = new Result<>(); Result<Object> result = new Result<>();
if (type == DbType.SSH) {
DataSourceProcessor sshDataSourceProcessor = DataSourceUtils.getDatasourceProcessor(type);
if (sshDataSourceProcessor.testConnection(connectionParam)) {
putMsg(result, Status.SUCCESS);
} else {
putMsg(result, Status.CONNECT_DATASOURCE_FAILURE);
}
return result;
}
try (Connection connection = DataSourceClientProvider.getInstance().getConnection(type, connectionParam)) { try (Connection connection = DataSourceClientProvider.getInstance().getConnection(type, connectionParam)) {
if (connection == null) { if (connection == null) {
log.error("Connection test to {} datasource failed, connectionParam:{}.", type.getDescp(), log.error("Connection test to {} datasource failed, connectionParam:{}.", type.getDescp(),

1
dolphinscheduler-api/src/main/resources/task-type-config.yaml

@ -29,6 +29,7 @@ task:
- 'DINKY' - 'DINKY'
- 'FLINK_STREAM' - 'FLINK_STREAM'
- 'HIVECLI' - 'HIVECLI'
- 'REMOTESHELL'
cloud: cloud:
- 'EMR' - 'EMR'
- 'K8S' - 'K8S'

12
dolphinscheduler-bom/pom.xml

@ -106,6 +106,7 @@
<azure-resourcemanager.version>2.21.0</azure-resourcemanager.version> <azure-resourcemanager.version>2.21.0</azure-resourcemanager.version>
<azure-resourcemanager-datafactory.version>1.0.0-beta.19</azure-resourcemanager-datafactory.version> <azure-resourcemanager-datafactory.version>1.0.0-beta.19</azure-resourcemanager-datafactory.version>
<google-cloud-storage.version>2.18.0</google-cloud-storage.version> <google-cloud-storage.version>2.18.0</google-cloud-storage.version>
<sshd.version>2.8.0</sshd.version>
</properties> </properties>
<dependencyManagement> <dependencyManagement>
@ -738,6 +739,17 @@
</exclusions> </exclusions>
</dependency> </dependency>
<dependency>
<groupId>org.apache.sshd</groupId>
<artifactId>sshd-sftp</artifactId>
<version>${sshd.version}</version>
</dependency>
<dependency>
<groupId>org.apache.sshd</groupId>
<artifactId>sshd-scp</artifactId>
<version>${sshd.version}</version>
</dependency>
<dependency> <dependency>
<groupId>org.apache.spark</groupId> <groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId> <artifactId>spark-sql_2.12</artifactId>

5
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-all/pom.xml

@ -107,5 +107,10 @@
<artifactId>dolphinscheduler-datasource-dameng</artifactId> <artifactId>dolphinscheduler-datasource-dameng</artifactId>
<version>${project.version}</version> <version>${project.version}</version>
</dependency> </dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-datasource-ssh</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies> </dependencies>
</project> </project>

10
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-api/src/main/java/org/apache/dolphinscheduler/plugin/datasource/api/datasource/DataSourceProcessor.java

@ -95,6 +95,16 @@ public interface DataSourceProcessor {
*/ */
Connection getConnection(ConnectionParam connectionParam) throws ClassNotFoundException, SQLException, IOException; Connection getConnection(ConnectionParam connectionParam) throws ClassNotFoundException, SQLException, IOException;
/**
* test connection, use for not jdbc datasource
*
* @param connectionParam connectionParam
* @return true if connection is valid
*/
default boolean testConnection(ConnectionParam connectionParam) {
return false;
}
/** /**
* @return {@link DbType} * @return {@link DbType}
*/ */

47
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/pom.xml

@ -0,0 +1,47 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-datasource-plugin</artifactId>
<version>dev-SNAPSHOT</version>
</parent>
<artifactId>dolphinscheduler-datasource-ssh</artifactId>
<packaging>jar</packaging>
<name>${project.artifactId}</name>
<dependencies>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-spi</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-datasource-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.sshd</groupId>
<artifactId>sshd-scp</artifactId>
</dependency>
</dependencies>
</project>

32
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/SSHDataSourceChannel.java

@ -0,0 +1,32 @@
/*
* 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.datasource.ssh;
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
import org.apache.dolphinscheduler.spi.datasource.DataSourceClient;
import org.apache.dolphinscheduler.spi.enums.DbType;
public class SSHDataSourceChannel implements DataSourceChannel {
@Override
public DataSourceClient createDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
return new SSHDataSourceClient(baseConnectionParam, dbType);
}
}

37
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/SSHDataSourceChannelFactory.java

@ -0,0 +1,37 @@
/*
* 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.datasource.ssh;
import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
import org.apache.dolphinscheduler.spi.datasource.DataSourceChannelFactory;
import com.google.auto.service.AutoService;
@AutoService(DataSourceChannelFactory.class)
public class SSHDataSourceChannelFactory implements DataSourceChannelFactory {
@Override
public String getName() {
return "ssh";
}
@Override
public DataSourceChannel create() {
return new SSHDataSourceChannel();
}
}

30
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/SSHDataSourceClient.java

@ -0,0 +1,30 @@
/*
* 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.datasource.ssh;
import org.apache.dolphinscheduler.plugin.datasource.api.client.CommonDataSourceClient;
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
import org.apache.dolphinscheduler.spi.enums.DbType;
public class SSHDataSourceClient extends CommonDataSourceClient {
public SSHDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
super(baseConnectionParam, dbType);
}
}

62
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/SSHUtils.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.datasource.ssh;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHConnectionParam;
import org.apache.commons.lang3.StringUtils;
import org.apache.sshd.client.SshClient;
import org.apache.sshd.client.session.ClientSession;
import org.apache.sshd.common.config.keys.loader.KeyPairResourceLoader;
import org.apache.sshd.common.util.security.SecurityUtils;
import java.security.KeyPair;
import java.util.Collection;
public class SSHUtils {
private SSHUtils() {
throw new IllegalStateException("Utility class");
}
public static ClientSession getSession(SshClient client, SSHConnectionParam connectionParam) throws Exception {
ClientSession session;
session = client.connect(connectionParam.getUser(), connectionParam.getHost(), connectionParam.getPort())
.verify(5000).getSession();
// add password identity
String password = connectionParam.getPassword();
if (StringUtils.isNotEmpty(password)) {
session.addPasswordIdentity(password);
}
// add public key identity
String publicKey = connectionParam.getPublicKey();
if (StringUtils.isNotEmpty(publicKey)) {
try {
KeyPairResourceLoader loader = SecurityUtils.getKeyPairResourceParser();
Collection<KeyPair> keyPairCollection = loader.loadKeyPairs(null, null, null, publicKey);
for (KeyPair keyPair : keyPairCollection) {
session.addPublicKeyIdentity(keyPair);
}
} catch (Exception e) {
throw new Exception("Failed to add public key identity", e);
}
}
return session;
}
}

39
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/param/SSHConnectionParam.java

@ -0,0 +1,39 @@
/*
* 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.datasource.ssh.param;
import org.apache.dolphinscheduler.spi.datasource.ConnectionParam;
import lombok.Data;
import com.fasterxml.jackson.annotation.JsonInclude;
@Data
@JsonInclude(JsonInclude.Include.NON_NULL)
public class SSHConnectionParam implements ConnectionParam {
protected String user;
protected String password;
protected String publicKey;
protected String host;
protected int port = 22;
}

35
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/param/SSHDataSourceParamDTO.java

@ -0,0 +1,35 @@
/*
* 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.datasource.ssh.param;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO;
import org.apache.dolphinscheduler.spi.enums.DbType;
import lombok.Data;
@Data
public class SSHDataSourceParamDTO extends BaseDataSourceParamDTO {
protected String publicKey;
@Override
public DbType getType() {
return DbType.SSH;
}
}

139
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/main/java/org/apache/dolphinscheduler/plugin/datasource/ssh/param/SSHDataSourceProcessor.java

@ -0,0 +1,139 @@
/*
* 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.datasource.ssh.param;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.DataSourceProcessor;
import org.apache.dolphinscheduler.plugin.datasource.api.utils.PasswordUtils;
import org.apache.dolphinscheduler.plugin.datasource.ssh.SSHUtils;
import org.apache.dolphinscheduler.spi.datasource.ConnectionParam;
import org.apache.dolphinscheduler.spi.enums.DbType;
import org.apache.commons.lang3.StringUtils;
import org.apache.sshd.client.SshClient;
import org.apache.sshd.client.session.ClientSession;
import java.sql.Connection;
import java.text.MessageFormat;
import lombok.extern.slf4j.Slf4j;
import com.google.auto.service.AutoService;
@AutoService(DataSourceProcessor.class)
@Slf4j
public class SSHDataSourceProcessor implements DataSourceProcessor {
@Override
public BaseDataSourceParamDTO castDatasourceParamDTO(String paramJson) {
return JSONUtils.parseObject(paramJson, SSHDataSourceParamDTO.class);
}
@Override
public void checkDatasourceParam(BaseDataSourceParamDTO datasourceParamDTO) {
if (StringUtils.isEmpty(datasourceParamDTO.getHost())
|| StringUtils.isEmpty(datasourceParamDTO.getUserName())) {
throw new IllegalArgumentException("ssh datasource param is not valid");
}
}
@Override
public String getDatasourceUniqueId(ConnectionParam connectionParam, DbType dbType) {
SSHConnectionParam baseConnectionParam = (SSHConnectionParam) connectionParam;
return MessageFormat.format("{0}@{1}@{2}@{3}", dbType.getDescp(), baseConnectionParam.getHost(),
baseConnectionParam.getUser(),
PasswordUtils.encodePassword(baseConnectionParam.getPassword()));
}
@Override
public BaseDataSourceParamDTO createDatasourceParamDTO(String connectionJson) {
SSHConnectionParam connectionParams = (SSHConnectionParam) createConnectionParams(connectionJson);
SSHDataSourceParamDTO sshDataSourceParamDTO = new SSHDataSourceParamDTO();
sshDataSourceParamDTO.setUserName(connectionParams.getUser());
sshDataSourceParamDTO.setPassword(connectionParams.getPassword());
sshDataSourceParamDTO.setHost(connectionParams.getHost());
sshDataSourceParamDTO.setPort(connectionParams.getPort());
sshDataSourceParamDTO.setPublicKey(connectionParams.getPublicKey());
return sshDataSourceParamDTO;
}
@Override
public SSHConnectionParam createConnectionParams(BaseDataSourceParamDTO dataSourceParam) {
SSHDataSourceParamDTO sshDataSourceParam = (SSHDataSourceParamDTO) dataSourceParam;
SSHConnectionParam sshConnectionParam = new SSHConnectionParam();
sshConnectionParam.setUser(sshDataSourceParam.getUserName());
sshConnectionParam.setPassword(sshDataSourceParam.getPassword());
sshConnectionParam.setHost(sshDataSourceParam.getHost());
sshConnectionParam.setPort(sshDataSourceParam.getPort());
sshConnectionParam.setPublicKey(sshDataSourceParam.getPublicKey());
return sshConnectionParam;
}
@Override
public ConnectionParam createConnectionParams(String connectionJson) {
return JSONUtils.parseObject(connectionJson, SSHConnectionParam.class);
}
@Override
public String getDatasourceDriver() {
return "";
}
@Override
public String getValidationQuery() {
return "";
}
@Override
public String getJdbcUrl(ConnectionParam connectionParam) {
return "";
}
@Override
public Connection getConnection(ConnectionParam connectionParam) {
return null;
}
@Override
public boolean testConnection(ConnectionParam connectionParam) {
SSHConnectionParam baseConnectionParam = (SSHConnectionParam) connectionParam;
SshClient client = SshClient.setUpDefaultClient();
client.start();
try {
ClientSession session = SSHUtils.getSession(client, baseConnectionParam);
return session.auth().verify().isSuccess();
} catch (Exception e) {
return false;
}
}
@Override
public DbType getDbType() {
return DbType.SSH;
}
@Override
public DataSourceProcessor create() {
return new SSHDataSourceProcessor();
}
}

119
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-ssh/src/test/java/org/apache/dolphinscheduler/plugin/datasource/ssh/SSHDataSourceProcessorTest.java

@ -0,0 +1,119 @@
/*
* 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.datasource.ssh;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.when;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHConnectionParam;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHDataSourceParamDTO;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHDataSourceProcessor;
import org.apache.dolphinscheduler.spi.datasource.ConnectionParam;
import org.apache.dolphinscheduler.spi.enums.DbType;
import org.apache.sshd.client.session.ClientSession;
import java.io.IOException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
@ExtendWith(MockitoExtension.class)
public class SSHDataSourceProcessorTest {
private SSHDataSourceProcessor sshDataSourceProcessor;
private String connectJson =
"{\"user\":\"lucky\",\"password\":\"123456\",\"host\":\"dolphinscheduler.com\",\"port\":22, \"publicKey\":\"ssh-rsa AAAAB\"}";
@BeforeEach
public void init() {
sshDataSourceProcessor = new SSHDataSourceProcessor();
}
@Test
void testCheckDatasourceParam() {
BaseDataSourceParamDTO baseDataSourceParamDTO = new SSHDataSourceParamDTO();
Assertions.assertThrows(IllegalArgumentException.class,
() -> sshDataSourceProcessor.checkDatasourceParam(baseDataSourceParamDTO));
baseDataSourceParamDTO.setHost("localhost");
Assertions.assertThrows(IllegalArgumentException.class,
() -> sshDataSourceProcessor.checkDatasourceParam(baseDataSourceParamDTO));
baseDataSourceParamDTO.setUserName("root");
Assertions.assertDoesNotThrow(() -> sshDataSourceProcessor.checkDatasourceParam(baseDataSourceParamDTO));
}
@Test
void testGetDatasourceUniqueId() {
SSHConnectionParam sshConnectionParam = new SSHConnectionParam();
sshConnectionParam.setHost("localhost");
sshConnectionParam.setUser("root");
sshConnectionParam.setPassword("123456");
Assertions.assertEquals("ssh@localhost@root@123456",
sshDataSourceProcessor.getDatasourceUniqueId(sshConnectionParam, DbType.SSH));
}
@Test
void testCreateDatasourceParamDTO() {
SSHDataSourceParamDTO sshDataSourceParamDTO =
(SSHDataSourceParamDTO) sshDataSourceProcessor.createDatasourceParamDTO(connectJson);
Assertions.assertEquals("lucky", sshDataSourceParamDTO.getUserName());
Assertions.assertEquals("123456", sshDataSourceParamDTO.getPassword());
Assertions.assertEquals("dolphinscheduler.com", sshDataSourceParamDTO.getHost());
Assertions.assertEquals(22, sshDataSourceParamDTO.getPort());
Assertions.assertEquals("ssh-rsa AAAAB", sshDataSourceParamDTO.getPublicKey());
}
@Test
void testCreateConnectionParams() {
SSHDataSourceParamDTO sshDataSourceParamDTO =
(SSHDataSourceParamDTO) sshDataSourceProcessor.createDatasourceParamDTO(connectJson);
SSHConnectionParam sshConnectionParam = sshDataSourceProcessor.createConnectionParams(sshDataSourceParamDTO);
Assertions.assertEquals("lucky", sshConnectionParam.getUser());
Assertions.assertEquals("123456", sshConnectionParam.getPassword());
Assertions.assertEquals("dolphinscheduler.com", sshConnectionParam.getHost());
Assertions.assertEquals(22, sshConnectionParam.getPort());
Assertions.assertEquals("ssh-rsa AAAAB", sshConnectionParam.getPublicKey());
}
@Test
void testTestConnection() throws IOException {
SSHDataSourceParamDTO sshDataSourceParamDTO =
(SSHDataSourceParamDTO) sshDataSourceProcessor.createDatasourceParamDTO(connectJson);
ConnectionParam connectionParam = sshDataSourceProcessor.createConnectionParams(sshDataSourceParamDTO);
MockedStatic<SSHUtils> sshConnectionUtilsMockedStatic = org.mockito.Mockito.mockStatic(SSHUtils.class);
sshConnectionUtilsMockedStatic.when(() -> SSHUtils.getSession(Mockito.any(), Mockito.any())).thenReturn(null);
Assertions.assertFalse(sshDataSourceProcessor.testConnection(connectionParam));
ClientSession clientSession = Mockito.mock(ClientSession.class, RETURNS_DEEP_STUBS);
sshConnectionUtilsMockedStatic.when(() -> SSHUtils.getSession(Mockito.any(), Mockito.any()))
.thenReturn(clientSession);
when(clientSession.auth().verify().isSuccess()).thenReturn(true);
Assertions.assertTrue(sshDataSourceProcessor.testConnection(connectionParam));
}
}

1
dolphinscheduler-datasource-plugin/pom.xml

@ -46,6 +46,7 @@
<module>dolphinscheduler-datasource-starrocks</module> <module>dolphinscheduler-datasource-starrocks</module>
<module>dolphinscheduler-datasource-azure-sql</module> <module>dolphinscheduler-datasource-azure-sql</module>
<module>dolphinscheduler-datasource-dameng</module> <module>dolphinscheduler-datasource-dameng</module>
<module>dolphinscheduler-datasource-ssh</module>
</modules> </modules>
<dependencyManagement> <dependencyManagement>

3
dolphinscheduler-dist/release-docs/LICENSE vendored

@ -547,6 +547,9 @@ The text of each license is also included at licenses/LICENSE-[project].txt.
opencensus-proto 0.2.0: https://mvnrepository.com/artifact/io.opencensus/opencensus-proto/0.2.0, Apache 2.0 opencensus-proto 0.2.0: https://mvnrepository.com/artifact/io.opencensus/opencensus-proto/0.2.0, Apache 2.0
proto-google-cloud-storage-v2 2.18.0-alpha: https://mvnrepository.com/artifact/com.google.api.grpc/proto-google-cloud-storage-v2/2.18.0-alpha, Apache 2.0 proto-google-cloud-storage-v2 2.18.0-alpha: https://mvnrepository.com/artifact/com.google.api.grpc/proto-google-cloud-storage-v2/2.18.0-alpha, Apache 2.0
proto-google-iam-v1 1.9.0: https://mvnrepository.com/artifact/com.google.api.grpc/proto-google-iam-v1/1.9.0, Apache 2.0 proto-google-iam-v1 1.9.0: https://mvnrepository.com/artifact/com.google.api.grpc/proto-google-iam-v1/1.9.0, Apache 2.0
sshd-sftp https://mvnrepository.com/artifact/org.apache.sshd/sshd-sftp/2.8.0 Apache 2.0
sshd-scp https://mvnrepository.com/artifact/org.apache.sshd/sshd-scp/2.8.0 Aapache 2.0
jna-platform jna-platform

202
dolphinscheduler-dist/release-docs/licenses/LICENSE-sshd-scp.txt vendored

@ -0,0 +1,202 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed 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.

202
dolphinscheduler-dist/release-docs/licenses/LICENSE-sshd-sftp.txt vendored

@ -0,0 +1,202 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed 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.

8
dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/datasource/ConnectionParam.java

@ -23,4 +23,12 @@ import java.io.Serializable;
* The model of Datasource Connection param * The model of Datasource Connection param
*/ */
public interface ConnectionParam extends Serializable { public interface ConnectionParam extends Serializable {
default String getPassword() {
return "";
}
default void setPassword(String s) {
}
} }

3
dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/enums/DbType.java

@ -44,7 +44,8 @@ public enum DbType {
STARROCKS(13, "starrocks"), STARROCKS(13, "starrocks"),
AZURESQL(14, "azuresql"), AZURESQL(14, "azuresql"),
DAMENG(15, "dameng"), DAMENG(15, "dameng"),
OCEANBASE(16, "oceanbase"); OCEANBASE(16, "oceanbase"),
SSH(17, "ssh");
private static final Map<Integer, DbType> DB_TYPE_MAP = private static final Map<Integer, DbType> DB_TYPE_MAP =
Arrays.stream(DbType.values()).collect(toMap(DbType::getCode, Functions.identity())); Arrays.stream(DbType.values()).collect(toMap(DbType::getCode, Functions.identity()));

5
dolphinscheduler-task-plugin/dolphinscheduler-task-all/pom.xml

@ -217,6 +217,11 @@
<artifactId>dolphinscheduler-task-datafactory</artifactId> <artifactId>dolphinscheduler-task-datafactory</artifactId>
<version>${project.version}</version> <version>${project.version}</version>
</dependency> </dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-task-remoteshell</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies> </dependencies>
</project> </project>

57
dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/pom.xml

@ -0,0 +1,57 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-task-plugin</artifactId>
<version>dev-SNAPSHOT</version>
</parent>
<artifactId>dolphinscheduler-task-remoteshell</artifactId>
<packaging>jar</packaging>
<dependencies>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-datasource-all</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-spi</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-task-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-datasource-all</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.sshd</groupId>
<artifactId>sshd-sftp</artifactId>
</dependency>
</dependencies>
</project>

251
dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/main/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteExecutor.java

@ -0,0 +1,251 @@
/*
* 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.remoteshell;
import org.apache.dolphinscheduler.plugin.datasource.ssh.SSHUtils;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHConnectionParam;
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
import org.apache.dolphinscheduler.plugin.task.api.TaskException;
import org.apache.commons.lang3.StringUtils;
import org.apache.sshd.client.SshClient;
import org.apache.sshd.client.channel.ChannelExec;
import org.apache.sshd.client.channel.ClientChannelEvent;
import org.apache.sshd.client.session.ClientSession;
import org.apache.sshd.sftp.client.SftpClientFactory;
import org.apache.sshd.sftp.client.fs.SftpFileSystem;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.EnumSet;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class RemoteExecutor {
protected final Logger logger =
LoggerFactory.getLogger(String.format(TaskConstants.TASK_LOGGER_THREAD_NAME, getClass()));
protected static final Pattern SETVALUE_REGEX = Pattern.compile(TaskConstants.SETVALUE_REGEX);
static final String REMOTE_SHELL_HOME = "/tmp/dolphinscheduler-remote-shell-%s/";
static final String STATUS_TAG_MESSAGE = "DOLPHINSCHEDULER-REMOTE-SHELL-TASK-STATUS-";
static final int TRACK_INTERVAL = 5000;
protected StringBuilder varPool = new StringBuilder();
SshClient sshClient;
ClientSession session;
SSHConnectionParam sshConnectionParam;
public RemoteExecutor(SSHConnectionParam sshConnectionParam) {
this.sshConnectionParam = sshConnectionParam;
initClient();
}
private void initClient() {
sshClient = SshClient.setUpDefaultClient();
sshClient.start();
}
private ClientSession getSession() {
if (session != null && session.isOpen()) {
return session;
}
try {
session = SSHUtils.getSession(sshClient, sshConnectionParam);
if (session == null || !session.auth().verify().isSuccess()) {
throw new TaskException("SSH connection failed");
}
} catch (Exception e) {
throw new TaskException("SSH connection failed", e);
}
return session;
}
public int run(String taskId, String localFile) throws IOException {
try {
// only run task if no exist same task
String pid = getTaskPid(taskId);
if (StringUtils.isEmpty(pid)) {
saveCommand(taskId, localFile);
String runCommand = String.format(COMMAND.RUN_COMMAND, getRemoteShellHome(), taskId,
getRemoteShellHome(), taskId);
runRemote(runCommand);
}
track(taskId);
return getTaskExitCode(taskId);
} catch (Exception e) {
throw new TaskException("Remote shell task error", e);
}
}
public void track(String taskId) throws Exception {
int logN = 0;
String pid;
logger.info("Remote shell task log:");
do {
pid = getTaskPid(taskId);
String trackCommand = String.format(COMMAND.TRACK_COMMAND, logN + 1, getRemoteShellHome(), taskId);
String log = runRemote(trackCommand);
if (StringUtils.isEmpty(log)) {
Thread.sleep(TRACK_INTERVAL);
} else {
logN += log.split("\n").length;
setVarPool(log);
logger.info(log);
}
} while (StringUtils.isNotEmpty(pid));
}
public String getVarPool() {
return varPool.toString();
}
private void setVarPool(String log) {
String[] lines = log.split("\n");
for (String line : lines) {
if (line.startsWith("${setValue(") || line.startsWith("#{setValue(")) {
varPool.append(findVarPool(line));
varPool.append("$VarPool$");
}
}
}
private String findVarPool(String line) {
Matcher matcher = SETVALUE_REGEX.matcher(line);
if (matcher.find()) {
return matcher.group(1);
}
return null;
}
public Integer getTaskExitCode(String taskId) throws IOException {
String trackCommand = String.format(COMMAND.LOG_TAIL_COMMAND, getRemoteShellHome(), taskId);
String log = runRemote(trackCommand);
int exitCode = -1;
logger.info("Remote shell task run status: {}", log);
if (log.contains(STATUS_TAG_MESSAGE)) {
String status = log.replace(STATUS_TAG_MESSAGE, "").trim();
if (status.equals("0")) {
logger.info("Remote shell task success");
exitCode = 0;
} else {
logger.error("Remote shell task failed");
exitCode = Integer.parseInt(status);
}
}
cleanData(taskId);
logger.error("Remote shell task failed");
return exitCode;
}
public void cleanData(String taskId) {
String cleanCommand =
String.format(COMMAND.CLEAN_COMMAND, getRemoteShellHome(), taskId, getRemoteShellHome(), taskId);
try {
runRemote(cleanCommand);
} catch (Exception e) {
logger.error("Remote shell task clean data failed, but will not affect the task execution", e);
}
}
public void kill(String taskId) throws IOException {
String pid = getTaskPid(taskId);
String killCommand = String.format(COMMAND.KILL_COMMAND, pid);
runRemote(killCommand);
cleanData(taskId);
}
public String getTaskPid(String taskId) throws IOException {
String pidCommand = String.format(COMMAND.GET_PID_COMMAND, taskId);
return runRemote(pidCommand).trim();
}
public void saveCommand(String taskId, String localFile) throws IOException {
String checkDirCommand = String.format(COMMAND.CHECK_DIR, getRemoteShellHome(), getRemoteShellHome());
runRemote(checkDirCommand);
uploadScript(taskId, localFile);
logger.info("The final script is: \n{}",
runRemote(String.format(COMMAND.CAT_FINAL_SCRIPT, getRemoteShellHome(), taskId)));
}
public void uploadScript(String taskId, String localFile) throws IOException {
String remotePath = getRemoteShellHome() + taskId + ".sh";
logger.info("upload script from local:{} to remote: {}", localFile, remotePath);
try (SftpFileSystem fs = SftpClientFactory.instance().createSftpFileSystem(getSession())) {
Path path = fs.getPath(remotePath);
Files.copy(Paths.get(localFile), path);
}
}
public String runRemote(String command) throws IOException {
try (
ChannelExec channel = getSession().createExecChannel(command);
ByteArrayOutputStream out = new ByteArrayOutputStream();
ByteArrayOutputStream err = new ByteArrayOutputStream()) {
channel.setOut(System.out);
channel.setOut(out);
channel.setErr(err);
channel.open();
channel.waitFor(EnumSet.of(ClientChannelEvent.CLOSED), 0);
channel.close();
if (channel.getExitStatus() != 0) {
throw new TaskException("Remote shell task error, error message: " + err.toString());
}
return out.toString();
}
}
private String getRemoteShellHome() {
return String.format(REMOTE_SHELL_HOME, sshConnectionParam.getUser());
}
static class COMMAND {
private COMMAND() {
throw new IllegalStateException("Utility class");
}
static final String CHECK_DIR = "if [ ! -d %s ]; then mkdir -p %s; fi";
static final String RUN_COMMAND = "nohup /bin/bash %s%s.sh >%s%s.log 2>&1 &";
static final String TRACK_COMMAND = "tail -n +%s %s%s.log";
static final String LOG_TAIL_COMMAND = "tail -n 1 %s%s.log";
static final String GET_PID_COMMAND = "ps -ef | grep \"%s.sh\" | grep -v grep | awk '{print $2}'";
static final String KILL_COMMAND = "kill -9 %s";
static final String CLEAN_COMMAND = "rm %s%s.sh %s%s.log";
static final String HEADER = "#!/bin/bash\n";
static final String ADD_STATUS_COMMAND = "\necho %s$?";
static final String CAT_FINAL_SCRIPT = "cat %s%s.sh";
}
}

50
dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/main/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteShellParameters.java

@ -0,0 +1,50 @@
/*
* 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.remoteshell;
import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
import lombok.Data;
@Data
public class RemoteShellParameters extends AbstractParameters {
private String rawScript;
private String type;
/**
* datasource id
*/
private int datasource;
@Override
public boolean checkParameters() {
return rawScript != null && !rawScript.isEmpty();
}
@Override
public ResourceParametersHelper getResources() {
ResourceParametersHelper resources = super.getResources();
resources.put(ResourceType.DATASOURCE, datasource);
return resources;
}
}

185
dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/main/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteShellTask.java

@ -0,0 +1,185 @@
/*
* 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.remoteshell;
import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_FAILURE;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.datasource.api.utils.DataSourceUtils;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHConnectionParam;
import org.apache.dolphinscheduler.plugin.task.api.AbstractTask;
import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack;
import org.apache.dolphinscheduler.plugin.task.api.TaskException;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
import org.apache.dolphinscheduler.plugin.task.api.parser.ParamUtils;
import org.apache.dolphinscheduler.plugin.task.api.parser.ParameterUtils;
import org.apache.dolphinscheduler.plugin.task.api.utils.FileUtils;
import org.apache.dolphinscheduler.spi.enums.DbType;
import org.apache.commons.lang3.SystemUtils;
import java.io.File;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.Map;
/**
* shell task
*/
public class RemoteShellTask extends AbstractTask {
static final String TASK_ID_PREFIX = "dolphinscheduler-remoteshell-";
/**
* shell parameters
*/
private RemoteShellParameters remoteShellParameters;
/**
* taskExecutionContext
*/
private TaskExecutionContext taskExecutionContext;
private RemoteExecutor remoteExecutor;
private String taskId;
/**
* constructor
*
* @param taskExecutionContext taskExecutionContext
*/
public RemoteShellTask(TaskExecutionContext taskExecutionContext) {
super(taskExecutionContext);
this.taskExecutionContext = taskExecutionContext;
}
@Override
public void init() {
log.info("shell task params {}", taskExecutionContext.getTaskParams());
remoteShellParameters =
JSONUtils.parseObject(taskExecutionContext.getTaskParams(), RemoteShellParameters.class);
if (!remoteShellParameters.checkParameters()) {
throw new TaskException("sell task params is not valid");
}
taskId = taskExecutionContext.getAppIds();
if (taskId == null) {
taskId = TASK_ID_PREFIX + taskExecutionContext.getTaskInstanceId();
}
setAppIds(taskId);
taskExecutionContext.setAppIds(taskId);
initRemoteExecutor();
}
@Override
public void handle(TaskCallBack taskCallBack) throws TaskException {
try {
// construct process
String localFile = buildCommand();
int exitCode = remoteExecutor.run(taskId, localFile);
setExitStatusCode(exitCode);
remoteShellParameters.dealOutParam(remoteExecutor.getVarPool());
} catch (Exception e) {
log.error("shell task error", e);
setExitStatusCode(EXIT_CODE_FAILURE);
throw new TaskException("Execute shell task error", e);
}
}
@Override
public void cancel() throws TaskException {
// cancel process
try {
log.info("kill remote task {}", taskId);
remoteExecutor.kill(taskId);
} catch (Exception e) {
throw new TaskException("cancel application error", e);
}
}
/**
* create command
*
* @return file name
* @throws Exception exception
*/
public String buildCommand() throws Exception {
// generate scripts
String fileName = String.format("%s/%s_node.%s",
taskExecutionContext.getExecutePath(),
taskExecutionContext.getTaskAppId(), SystemUtils.IS_OS_WINDOWS ? "bat" : "sh");
File file = new File(fileName);
Path path = file.toPath();
if (Files.exists(path)) {
// this shouldn't happen
log.warn("The command file: {} is already exist", path);
return fileName;
}
String script = remoteShellParameters.getRawScript().replaceAll("\\r\\n", "\n");
script = parseScript(script);
String environment = taskExecutionContext.getEnvironmentConfig();
if (environment != null) {
environment = environment.replaceAll("\\r\\n", "\n");
environment = environment.replace("\r\n", "\n");
script = environment + "\n" + script;
}
script = String.format(RemoteExecutor.COMMAND.HEADER) + script;
script += String.format(RemoteExecutor.COMMAND.ADD_STATUS_COMMAND, RemoteExecutor.STATUS_TAG_MESSAGE);
FileUtils.createFileWith755(path);
Files.write(path, script.getBytes(), StandardOpenOption.APPEND);
log.info("raw script : {}", script);
return fileName;
}
@Override
public AbstractParameters getParameters() {
return remoteShellParameters;
}
private String parseScript(String script) {
// combining local and global parameters
Map<String, Property> paramsMap = taskExecutionContext.getPrepareParamsMap();
return ParameterUtils.convertParameterPlaceholders(script, ParamUtils.convert(paramsMap));
}
public void initRemoteExecutor() {
DataSourceParameters dbSource = (DataSourceParameters) taskExecutionContext.getResourceParametersHelper()
.getResourceParameters(ResourceType.DATASOURCE, remoteShellParameters.getDatasource());
taskExecutionContext.getResourceParametersHelper().getResourceParameters(ResourceType.DATASOURCE,
remoteShellParameters.getDatasource());
SSHConnectionParam sshConnectionParam = (SSHConnectionParam) DataSourceUtils.buildConnectionParams(
DbType.valueOf(remoteShellParameters.getType()),
dbSource.getConnectionParams());
remoteExecutor = new RemoteExecutor(sshConnectionParam);
}
}

48
dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/main/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteShellTaskChannel.java

@ -0,0 +1,48 @@
/*
* 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.remoteshell;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.task.api.TaskChannel;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.ParametersNode;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
public class RemoteShellTaskChannel implements TaskChannel {
@Override
public void cancelApplication(boolean status) {
}
@Override
public RemoteShellTask createTask(TaskExecutionContext taskRequest) {
return new RemoteShellTask(taskRequest);
}
@Override
public AbstractParameters parseParameters(ParametersNode parametersNode) {
return JSONUtils.parseObject(parametersNode.getTaskParams(), RemoteShellParameters.class);
}
@Override
public ResourceParametersHelper getResources(String parameters) {
return JSONUtils.parseObject(parameters, RemoteShellParameters.class).getResources();
}
}

65
dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/main/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteShellTaskChannelFactory.java

@ -0,0 +1,65 @@
/*
* 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.remoteshell;
import org.apache.dolphinscheduler.plugin.task.api.TaskChannel;
import org.apache.dolphinscheduler.plugin.task.api.TaskChannelFactory;
import org.apache.dolphinscheduler.spi.params.base.ParamsOptions;
import org.apache.dolphinscheduler.spi.params.base.PluginParams;
import org.apache.dolphinscheduler.spi.params.base.Validate;
import org.apache.dolphinscheduler.spi.params.input.InputParam;
import org.apache.dolphinscheduler.spi.params.radio.RadioParam;
import java.util.ArrayList;
import java.util.List;
import com.google.auto.service.AutoService;
@AutoService(TaskChannelFactory.class)
public class RemoteShellTaskChannelFactory implements TaskChannelFactory {
@Override
public TaskChannel create() {
return new RemoteShellTaskChannel();
}
@Override
public String getName() {
return "REMOTESHELL";
}
@Override
public List<PluginParams> getParams() {
List<PluginParams> paramsList = new ArrayList<>();
InputParam nodeName = InputParam.newBuilder("name", "$t('Node name')")
.addValidate(Validate.newBuilder()
.setRequired(true)
.build())
.build();
RadioParam runFlag = RadioParam.newBuilder("runFlag", "RUN_FLAG")
.addParamsOptions(new ParamsOptions("NORMAL", "NORMAL", false))
.addParamsOptions(new ParamsOptions("FORBIDDEN", "FORBIDDEN", false))
.build();
paramsList.add(nodeName);
paramsList.add(runFlag);
return paramsList;
}
}

136
dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/test/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteExecutorTest.java

@ -0,0 +1,136 @@
/*
* 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.remoteshell;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.apache.dolphinscheduler.plugin.datasource.ssh.SSHUtils;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHConnectionParam;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHDataSourceParamDTO;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHDataSourceProcessor;
import org.apache.dolphinscheduler.plugin.task.api.TaskException;
import org.apache.sshd.client.channel.ChannelExec;
import org.apache.sshd.client.session.ClientSession;
import java.io.IOException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
@ExtendWith(MockitoExtension.class)
public class RemoteExecutorTest {
private String connectJson =
"{\"user\":\"root\",\"password\":\"123456\",\"host\":\"dolphinscheduler.com\",\"port\":22, \"publicKey\":\"ssh-rsa AAAAB\"}";
SSHConnectionParam sshConnectionParam;
ClientSession clientSession;
MockedStatic<SSHUtils> sshConnectionUtilsMockedStatic = org.mockito.Mockito.mockStatic(SSHUtils.class);
@BeforeEach
void init() {
SSHDataSourceProcessor sshDataSourceProcessor = new SSHDataSourceProcessor();
SSHDataSourceParamDTO sshDataSourceParamDTO =
(SSHDataSourceParamDTO) sshDataSourceProcessor.createDatasourceParamDTO(connectJson);
sshConnectionParam = sshDataSourceProcessor.createConnectionParams(sshDataSourceParamDTO);
clientSession = Mockito.mock(ClientSession.class, RETURNS_DEEP_STUBS);
sshConnectionUtilsMockedStatic.when(() -> SSHUtils.getSession(Mockito.any(), Mockito.any()))
.thenReturn(clientSession);
}
@AfterEach
void tearDown() {
sshConnectionUtilsMockedStatic.close();
}
@Test
void testRunRemote() throws IOException {
RemoteExecutor remoteExecutor = spy(new RemoteExecutor(sshConnectionParam));
ChannelExec channel = Mockito.mock(ChannelExec.class, RETURNS_DEEP_STUBS);
when(clientSession.auth().verify().isSuccess()).thenReturn(true);
when(clientSession.createExecChannel(Mockito.anyString())).thenReturn(channel);
when(channel.getExitStatus()).thenReturn(1);
Assertions.assertThrows(TaskException.class, () -> remoteExecutor.runRemote("ls -l"));
when(channel.getExitStatus()).thenReturn(0);
Assertions.assertDoesNotThrow(() -> remoteExecutor.runRemote("ls -l"));
}
@Test
void testGetTaskPid() throws IOException {
RemoteExecutor remoteExecutor = spy(new RemoteExecutor(sshConnectionParam));
String taskId = "1234";
String command = String.format("ps -ef | grep \"%s.sh\" | grep -v grep | awk '{print $2}'", taskId);
doReturn("10001").when(remoteExecutor).runRemote(command);
Assertions.assertEquals("10001", remoteExecutor.getTaskPid(taskId));
}
@Test
void testSaveCommand() throws IOException {
RemoteExecutor remoteExecutor = spy(new RemoteExecutor(sshConnectionParam));
doNothing().when(remoteExecutor).uploadScript(Mockito.anyString(), Mockito.anyString());
String checkDirCommand =
"if [ ! -d /tmp/dolphinscheduler-remote-shell-root/ ]; then mkdir -p /tmp/dolphinscheduler-remote-shell-root/; fi";
String catScriptCommand = "cat /tmp/dolphinscheduler-remote-shell-root/1234.sh";
doReturn("").when(remoteExecutor).runRemote(checkDirCommand);
doReturn("").when(remoteExecutor).runRemote(catScriptCommand);
remoteExecutor.saveCommand("1234", "/tmp/dolphinscheduler/test.sh");
verify(remoteExecutor).runRemote(checkDirCommand);
}
@Test
void testCleanData() throws IOException {
RemoteExecutor remoteExecutor = spy(new RemoteExecutor(sshConnectionParam));
String cleanCommand =
"rm /tmp/dolphinscheduler-remote-shell-root/1234.sh /tmp/dolphinscheduler-remote-shell-root/1234.log";
doReturn("").when(remoteExecutor).runRemote(cleanCommand);
remoteExecutor.cleanData("1234");
String cleanCommandError =
"rm /tmp/dolphinscheduler-remote-shell-root/abcd.sh /tmp/dolphinscheduler-remote-shell-root/abcd.log";
doThrow(new TaskException()).when(remoteExecutor).runRemote(cleanCommandError);
remoteExecutor.cleanData("abcd");
}
@Test
void testGetTaskExitCode() throws IOException {
RemoteExecutor remoteExecutor = spy(new RemoteExecutor(sshConnectionParam));
String taskId = "1234";
doNothing().when(remoteExecutor).cleanData(taskId);
String trackCommand = "tail -n 1 /tmp/dolphinscheduler-remote-shell-root/1234.log";
doReturn("DOLPHINSCHEDULER-REMOTE-SHELL-TASK-STATUS-0").when(remoteExecutor).runRemote(trackCommand);
Assertions.assertEquals(0, remoteExecutor.getTaskExitCode(taskId));
doReturn("DOLPHINSCHEDULER-REMOTE-SHELL-TASK-STATUS-1").when(remoteExecutor).runRemote(trackCommand);
Assertions.assertEquals(1, remoteExecutor.getTaskExitCode(taskId));
}
}

92
dolphinscheduler-task-plugin/dolphinscheduler-task-remoteshell/src/test/java/org/apache/dolphinscheduler/plugin/task/remoteshell/RemoteShellTaskTest.java

@ -0,0 +1,92 @@
/*
* 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.remoteshell;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.spy;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHConnectionParam;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHDataSourceParamDTO;
import org.apache.dolphinscheduler.plugin.datasource.ssh.param.SSHDataSourceProcessor;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.sshd.client.session.ClientSession;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
@ExtendWith(MockitoExtension.class)
public class RemoteShellTaskTest {
private String connectJson =
"{\"user\":\"root\",\"password\":\"123456\",\"host\":\"dolphinscheduler.com\",\"port\":22, \"publicKey\":\"ssh-rsa AAAAB\"}";
SSHConnectionParam sshConnectionParam;
ClientSession clientSession;
@BeforeEach
void init() {
SSHDataSourceProcessor sshDataSourceProcessor = new SSHDataSourceProcessor();
SSHDataSourceParamDTO sshDataSourceParamDTO =
(SSHDataSourceParamDTO) sshDataSourceProcessor.createDatasourceParamDTO(connectJson);
sshConnectionParam = sshDataSourceProcessor.createConnectionParams(sshDataSourceParamDTO);
clientSession = Mockito.mock(ClientSession.class, RETURNS_DEEP_STUBS);
}
@Test
void testBuildCommand() throws Exception {
TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
taskExecutionContext.setTaskAppId("1");
taskExecutionContext
.setTaskParams("{\"localParams\":[],\"rawScript\":\"echo 1\",\"resourceList\":[],\"udfList\":[]}");
taskExecutionContext.setExecutePath("/tmp");
taskExecutionContext.setEnvironmentConfig("export PATH=/opt/anaconda3/bin:$PATH");
RemoteShellTask remoteShellTask = spy(new RemoteShellTask(taskExecutionContext));
doNothing().when(remoteShellTask).initRemoteExecutor();
remoteShellTask.init();
MockedStatic<Files> filesMockedStatic = org.mockito.Mockito.mockStatic(Files.class);
filesMockedStatic.when(() -> Files.exists(Mockito.any())).thenReturn(false);
String script = "#!/bin/bash\n" +
"export PATH=/opt/anaconda3/bin:$PATH\n" +
"echo 1\n" +
"echo DOLPHINSCHEDULER-REMOTE-SHELL-TASK-STATUS-$?";
Path path = Paths.get("/tmp/1_node.sh");
filesMockedStatic.when(() -> Files.write(path, script.getBytes(), StandardOpenOption.APPEND))
.thenThrow(new IOException("script match"));
IOException exception = Assertions.assertThrows(IOException.class, () -> {
remoteShellTask.buildCommand();
});
Assertions.assertEquals("script match", exception.getMessage());
}
}

1
dolphinscheduler-task-plugin/pom.xml

@ -62,6 +62,7 @@
<module>dolphinscheduler-task-kubeflow</module> <module>dolphinscheduler-task-kubeflow</module>
<module>dolphinscheduler-task-linkis</module> <module>dolphinscheduler-task-linkis</module>
<module>dolphinscheduler-task-datafactory</module> <module>dolphinscheduler-task-datafactory</module>
<module>dolphinscheduler-task-remoteshell</module>
</modules> </modules>
<dependencyManagement> <dependencyManagement>

BIN
dolphinscheduler-ui/public/images/task-icons/remoteshell.png

Binary file not shown.

After

Width:  |  Height:  |  Size: 747 B

BIN
dolphinscheduler-ui/public/images/task-icons/remoteshell_hover.png

Binary file not shown.

After

Width:  |  Height:  |  Size: 745 B

3
dolphinscheduler-ui/src/service/modules/data-source/types.ts

@ -32,6 +32,7 @@ type IDataBase =
| 'STARROCKS' | 'STARROCKS'
| 'DAMENG' | 'DAMENG'
| 'OCEANBASE' | 'OCEANBASE'
| 'SSH'
type IDataBaseLabel = type IDataBaseLabel =
| 'MYSQL' | 'MYSQL'
@ -50,6 +51,7 @@ type IDataBaseLabel =
| 'STARROCKS' | 'STARROCKS'
| 'DAMENG' | 'DAMENG'
| 'OCEANBASE' | 'OCEANBASE'
| 'SSH'
interface IDataSource { interface IDataSource {
id?: number id?: number
@ -76,6 +78,7 @@ interface IDataSource {
MSIClientId?: string MSIClientId?: string
dbUser?: string dbUser?: string
compatibleMode?: string compatibleMode?: string
publicKey?: string
} }
interface ListReq { interface ListReq {

4
dolphinscheduler-ui/src/store/project/task-type.ts

@ -153,6 +153,10 @@ export const TASK_TYPES_MAP = {
DATA_FACTORY: { DATA_FACTORY: {
alias: 'DATA_FACTORY', alias: 'DATA_FACTORY',
helperLinkDisable: true helperLinkDisable: true
},
REMOTESHELL: {
alias: 'REMOTESHELL',
helperLinkDisable: true
} }
} as { } as {
[key in TaskType]: { [key in TaskType]: {

1
dolphinscheduler-ui/src/store/project/types.ts

@ -57,6 +57,7 @@ type TaskType =
| 'KUBEFLOW' | 'KUBEFLOW'
| 'LINKIS' | 'LINKIS'
| 'DATA_FACTORY' | 'DATA_FACTORY'
| 'REMOTESHELL'
type ProgramType = 'JAVA' | 'SCALA' | 'PYTHON' type ProgramType = 'JAVA' | 'SCALA' | 'PYTHON'

18
dolphinscheduler-ui/src/views/datasource/list/detail.tsx

@ -162,6 +162,9 @@ const DetailModal = defineComponent({
showConnectType, showConnectType,
showPrincipal, showPrincipal,
showMode, showMode,
showDataBaseName,
showJDBCConnectParameters,
showPublicKey,
modeOptions, modeOptions,
redShitModeOptions, redShitModeOptions,
loading, loading,
@ -539,6 +542,7 @@ const DetailModal = defineComponent({
/> />
</NFormItem> </NFormItem>
<NFormItem <NFormItem
v-show={showDataBaseName}
label={t('datasource.database_name')} label={t('datasource.database_name')}
path='database' path='database'
show-require-mark={requiredDataBase} show-require-mark={requiredDataBase}
@ -585,6 +589,7 @@ const DetailModal = defineComponent({
/> />
</NFormItem> </NFormItem>
<NFormItem <NFormItem
v-show={showJDBCConnectParameters}
label={t('datasource.jdbc_connect_parameters')} label={t('datasource.jdbc_connect_parameters')}
path='other' path='other'
> >
@ -634,6 +639,19 @@ const DetailModal = defineComponent({
options={this.bindTestDataSourceExample} options={this.bindTestDataSourceExample}
/> />
</NFormItem> </NFormItem>
<NFormItem
v-show={showPublicKey}
label='PublicKey'
path='publicKey'
>
<NInput
v-model={[detailForm.publicKey, 'value']}
type='textarea'
autosize={{
minRows: 4
}}
/>
</NFormItem>
</NForm> </NForm>
</NSpin> </NSpin>
), ),

23
dolphinscheduler-ui/src/views/datasource/list/use-form.ts

@ -69,6 +69,9 @@ export function useForm(id?: number) {
showConnectType: false, showConnectType: false,
showPrincipal: false, showPrincipal: false,
showMode: false, showMode: false,
showDataBaseName: true,
showJDBCConnectParameters: true,
showPublicKey: false,
bindTestDataSourceExample: [] as { label: string; value: number }[], bindTestDataSourceExample: [] as { label: string; value: number }[],
rules: { rules: {
name: { name: {
@ -263,6 +266,19 @@ export function useForm(id?: number) {
} else { } else {
state.showPrincipal = false state.showPrincipal = false
} }
if (type === 'SSH') {
state.showDataBaseName = false
state.requiredDataBase = false
state.showJDBCConnectParameters = false
state.showPublicKey = true
}else {
state.showDataBaseName = true
state.requiredDataBase = true
state.showJDBCConnectParameters = true
state.showPublicKey = false
}
if (state.detailForm.id === undefined) { if (state.detailForm.id === undefined) {
await getSameTypeTestDataSource() await getSameTypeTestDataSource()
} }
@ -406,6 +422,11 @@ export const datasourceType: IDataBaseOptionKeys = {
value: 'OCEANBASE', value: 'OCEANBASE',
label: 'OCEANBASE', label: 'OCEANBASE',
defaultPort: 2881 defaultPort: 2881
},
SSH: {
value: 'SSH',
label: 'SSH',
defaultPort: 22
} }
} }
@ -414,4 +435,4 @@ export const datasourceTypeList: IDataBaseOption[] = Object.values(
).map((item) => { ).map((item) => {
item.class = 'options-datasource-type' item.class = 'options-datasource-type'
return item return item
}) })

1
dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts

@ -86,3 +86,4 @@ export { useDatasync } from './use-datasync'
export { useKubeflow } from './use-kubeflow' export { useKubeflow } from './use-kubeflow'
export { useLinkis } from './use-linkis' export { useLinkis } from './use-linkis'
export { useDataFactory } from './use-data-factory' export { useDataFactory } from './use-data-factory'
export { useRemoteShell } from './use-remote-shell'

7
dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-datasource.ts

@ -112,7 +112,12 @@ export function useDatasource(
id: 15, id: 15,
code: 'DAMENG', code: 'DAMENG',
disabled: false disabled: false
} },
{
id: 15,
code: 'SSH',
disabled: true
},
] ]
const getDatasourceTypes = async () => { const getDatasourceTypes = async () => {

37
dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-remote-shell.ts

@ -0,0 +1,37 @@
/*
* 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.
*/
import { useI18n } from 'vue-i18n'
import { useCustomParams } from '.'
import type { IJsonItem } from '../types'
export function useRemoteShell(model: { [field: string]: any }): IJsonItem[] {
const { t } = useI18n()
return [
{
type: 'editor',
field: 'rawScript',
name: t('project.node.script'),
validate: {
trigger: ['input', 'trigger'],
required: true,
message: t('project.node.script_tips')
}
},
...useCustomParams({ model, field: 'localParams', isSimple: false })
]
}

5
dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts

@ -465,6 +465,11 @@ export function formatParams(data: INodeData): {
taskParams.pipelineName = data.pipelineName taskParams.pipelineName = data.pipelineName
} }
if (data.taskType === 'REMOTESHELL') {
taskParams.type = data.type
taskParams.datasource = data.datasource
}
let timeoutNotifyStrategy = '' let timeoutNotifyStrategy = ''
if (data.timeoutNotifyStrategy) { if (data.timeoutNotifyStrategy) {
if (data.timeoutNotifyStrategy.length === 1) { if (data.timeoutNotifyStrategy.length === 1) {

4
dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts

@ -51,6 +51,7 @@ import { useDatasync } from './use-datasync'
import { useKubeflow } from './use-kubeflow' import { useKubeflow } from './use-kubeflow'
import { useLinkis } from './use-linkis' import { useLinkis } from './use-linkis'
import { useDataFactory } from './use-data-factory' import { useDataFactory } from './use-data-factory'
import { useRemoteShell } from './use-remote-shell'
export default { export default {
SHELL: useShell, SHELL: useShell,
@ -88,5 +89,6 @@ export default {
DATASYNC: useDatasync, DATASYNC: useDatasync,
KUBEFLOW: useKubeflow, KUBEFLOW: useKubeflow,
LINKIS: useLinkis, LINKIS: useLinkis,
DATA_FACTORY: useDataFactory DATA_FACTORY: useDataFactory,
REMOTESHELL: useRemoteShell
} }

74
dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-remote-shell.ts

@ -0,0 +1,74 @@
/*
* 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.
*/
import { reactive } from 'vue'
import * as Fields from '../fields/index'
import type { IJsonItem, INodeData, ITaskData } from '../types'
export function useRemoteShell({
projectCode,
from = 0,
readonly,
data
}: {
projectCode: number
from?: number
readonly?: boolean
data?: ITaskData
}) {
const model = reactive({
name: '',
taskType: 'REMOTESHELL',
flag: 'YES',
description: '',
timeoutFlag: false,
timeoutNotifyStrategy: ['WARN'],
timeout: 30,
localParams: [],
environmentCode: null,
failRetryInterval: 1,
failRetryTimes: 0,
workerGroup: 'default',
delayTime: 0,
type: 'SSH',
rawScript: ''
} as INodeData)
return {
json: [
Fields.useName(from),
...Fields.useTaskDefinition({ projectCode, from, readonly, data, model }),
Fields.useRunFlag(),
Fields.useCache(),
Fields.useDescription(),
Fields.useTaskPriority(),
Fields.useWorkerGroup(),
Fields.useEnvironmentName(model, !data?.id),
...Fields.useTaskGroup(model, projectCode),
...Fields.useFailed(),
...Fields.useResourceLimit(),
Fields.useDelayTime(model),
...Fields.useTimeoutAlarm(model),
...Fields.useDatasource(model, {
supportedDatasourceType: ['SSH']
}),
...Fields.useRemoteShell(model),
Fields.usePreTasks()
] as IJsonItem[],
model
}
}

5
dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts

@ -51,6 +51,7 @@ export type TaskType =
| 'KUBEFLOW' | 'KUBEFLOW'
| 'LINKIS' | 'LINKIS'
| 'DATA_FACTORY' | 'DATA_FACTORY'
| 'REMOTESHELL'
export type TaskExecuteType = 'STREAM' | 'BATCH' export type TaskExecuteType = 'STREAM' | 'BATCH'
@ -185,6 +186,10 @@ export const TASK_TYPES_MAP = {
DATA_FACTORY: { DATA_FACTORY: {
alias: 'DATA_FACTORY', alias: 'DATA_FACTORY',
helperLinkDisable: true helperLinkDisable: true
},
REMOTESHELL: {
alias: 'REMOTESHELL',
helperLinkDisable: true
} }
} as { } as {
[key in TaskType]: { [key in TaskType]: {

6
dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss

@ -207,6 +207,9 @@ $bgLight: #ffffff;
&.icon-data_factory { &.icon-data_factory {
background-image: url('/images/task-icons/data_factory.png'); background-image: url('/images/task-icons/data_factory.png');
} }
&.icon-remoteshell {
background-image: url('/images/task-icons/remoteshell.png');
}
} }
&:hover { &:hover {
@ -317,6 +320,9 @@ $bgLight: #ffffff;
&.icon-data_factory { &.icon-data_factory {
background-image: url('/images/task-icons/data_factory_hover.png'); background-image: url('/images/task-icons/data_factory_hover.png');
} }
&.icon-remoteshell {
background-image: url('/images/task-icons/remoteshell_hover.png');
}
} }
} }

7
tools/dependencies/known-dependencies.txt

@ -468,4 +468,9 @@ opencensus-proto-0.2.0.jar
proto-google-cloud-storage-v2-2.18.0-alpha.jar proto-google-cloud-storage-v2-2.18.0-alpha.jar
proto-google-iam-v1-1.9.0.jar proto-google-iam-v1-1.9.0.jar
re2j-1.6.jar re2j-1.6.jar
threetenbp-1.6.5.jar threetenbp-1.6.5.jar
sshd-scp-2.8.0.jar
sshd-sftp-2.8.0.jar
sshd-common-2.8.0.jar
sshd-core-2.8.0.jar
jcl-over-slf4j-1.7.36.jar

Loading…
Cancel
Save