diff --git a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongClusterNodeEntity.java b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongClusterNodeEntity.java
index b57aa6d268e..eddf41f641c 100644
--- a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongClusterNodeEntity.java
+++ b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongClusterNodeEntity.java
@@ -34,6 +34,9 @@ public class InlongClusterNodeEntity implements Serializable {
private String type;
private String ip;
private Integer port;
+ private String username;
+ private String password;
+ private Integer sshPort;
private String protocolType;
private Integer nodeLoad;
private String extParams;
diff --git a/inlong-manager/manager-dao/src/main/resources/mappers/InlongClusterNodeEntityMapper.xml b/inlong-manager/manager-dao/src/main/resources/mappers/InlongClusterNodeEntityMapper.xml
index b4b65da45b2..e3d54df45af 100644
--- a/inlong-manager/manager-dao/src/main/resources/mappers/InlongClusterNodeEntityMapper.xml
+++ b/inlong-manager/manager-dao/src/main/resources/mappers/InlongClusterNodeEntityMapper.xml
@@ -26,6 +26,9 @@
+
+
+
@@ -39,32 +42,34 @@
- id, parent_id, type, ip, port, protocol_type, node_load, ext_params, description,
+ id, parent_id, type, ip, port, username, password, ssh_port, protocol_type, node_load, ext_params, description,
status, is_deleted, creator, modifier, create_time, modify_time, version
insert into inlong_cluster_node (id, parent_id, type,
- ip, port, protocol_type,
- node_load, ext_params,
- description, status,
- creator, modifier)
+ ip, port, username,
+ password, ssh_port, protocol_type,
+ node_load, ext_params, description,
+ status, creator, modifier)
values (#{id,jdbcType=INTEGER}, #{parentId,jdbcType=INTEGER}, #{type,jdbcType=VARCHAR},
- #{ip,jdbcType=VARCHAR}, #{port,jdbcType=INTEGER}, #{protocolType,jdbcType=VARCHAR},
- #{nodeLoad,jdbcType=INTEGER}, #{extParams,jdbcType=LONGVARCHAR},
- #{description, jdbcType=VARCHAR}, #{status,jdbcType=INTEGER},
- #{creator,jdbcType=VARCHAR}, #{modifier,jdbcType=VARCHAR})
+ #{ip,jdbcType=VARCHAR}, #{port,jdbcType=INTEGER}, #{username,jdbcType=VARCHAR},
+ #{password,jdbcType=VARCHAR}, #{sshPort,jdbcType=INTEGER}, #{protocolType,jdbcType=VARCHAR},
+ #{nodeLoad,jdbcType=INTEGER}, #{extParams,jdbcType=LONGVARCHAR}, #{description, jdbcType=VARCHAR},
+ #{status,jdbcType=INTEGER}, #{creator,jdbcType=VARCHAR}, #{modifier,jdbcType=VARCHAR})
insert into inlong_cluster_node (id, parent_id, type,
- ip, port, protocol_type,
+ ip, port, username,
+ password, ssh_port, protocol_type,
node_load, ext_params, status,
creator, modifier)
values (#{id,jdbcType=INTEGER}, #{parentId,jdbcType=INTEGER}, #{type,jdbcType=VARCHAR},
- #{ip,jdbcType=VARCHAR}, #{port,jdbcType=INTEGER}, #{protocolType,jdbcType=VARCHAR},
+ #{ip,jdbcType=VARCHAR}, #{port,jdbcType=INTEGER}, #{username,jdbcType=VARCHAR},
+ #{password,jdbcType=VARCHAR}, #{sshPort,jdbcType=INTEGER}, #{protocolType,jdbcType=VARCHAR},
#{nodeLoad,jdbcType=INTEGER}, #{extParams,jdbcType=LONGVARCHAR},
#{status,jdbcType=INTEGER}, #{creator,jdbcType=VARCHAR}, #{modifier,jdbcType=VARCHAR})
ON DUPLICATE KEY UPDATE node_load = VALUES(node_load),
@@ -157,6 +162,9 @@
type = #{type,jdbcType=VARCHAR},
ip = #{ip,jdbcType=VARCHAR},
port = #{port,jdbcType=INTEGER},
+ username = #{username,jdbcType=VARCHAR},
+ password = #{password,jdbcType=VARCHAR},
+ ssh_port = #{sshPort,jdbcType=INTEGER},
protocol_type = #{protocolType,jdbcType=VARCHAR},
node_load = #{nodeLoad,jdbcType=INTEGER},
ext_params = #{extParams,jdbcType=LONGVARCHAR},
@@ -183,6 +191,15 @@
port = #{port,jdbcType=INTEGER},
+
+ username = #{username,jdbcType=VARCHAR},
+
+
+ password = #{password,jdbcType=VARCHAR},
+
+
+ ssh_port = #{sshPort,jdbcType=INTEGER},
+
protocol_type = #{protocolType,jdbcType=VARCHAR},
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/ClusterNodeRequest.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/ClusterNodeRequest.java
index 07960d23047..a7ea519fbe9 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/ClusterNodeRequest.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/ClusterNodeRequest.java
@@ -58,6 +58,15 @@ public class ClusterNodeRequest {
@NotNull(message = "port cannot be null")
private Integer port;
+ @ApiModelProperty(value = "Username")
+ private String username;
+
+ @ApiModelProperty(value = "password")
+ private String password;
+
+ @ApiModelProperty(value = "SSH port")
+ private Integer sshPort;
+
@ApiModelProperty(value = "Cluster protocol type")
@Length(min = 1, max = 20, message = "length must be less than or equal to 20")
private String protocolType;
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/ClusterNodeResponse.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/ClusterNodeResponse.java
index f16d582f8e2..51291fafdce 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/ClusterNodeResponse.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/ClusterNodeResponse.java
@@ -52,6 +52,15 @@ public class ClusterNodeResponse {
@ApiModelProperty(value = "Cluster port")
private Integer port;
+ @ApiModelProperty(value = "Username")
+ private String username;
+
+ @ApiModelProperty(value = "password")
+ private String password;
+
+ @ApiModelProperty(value = "SSH port")
+ private Integer sshPort;
+
@ApiModelProperty(value = "Cluster protocol type")
private String protocolType;
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/node/AgentClusterNodeInstallOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/node/AgentClusterNodeInstallOperator.java
index ae9615eedc4..22be7be9902 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/node/AgentClusterNodeInstallOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/node/AgentClusterNodeInstallOperator.java
@@ -18,24 +18,66 @@
package org.apache.inlong.manager.service.cluster.node;
import org.apache.inlong.manager.common.enums.ClusterType;
+import org.apache.inlong.manager.common.enums.ModuleType;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+import org.apache.inlong.manager.common.util.AESUtils;
+import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.InlongClusterEntity;
import org.apache.inlong.manager.dao.entity.InlongClusterNodeEntity;
+import org.apache.inlong.manager.dao.entity.ModuleConfigEntity;
+import org.apache.inlong.manager.dao.entity.PackageConfigEntity;
+import org.apache.inlong.manager.dao.entity.UserEntity;
import org.apache.inlong.manager.dao.mapper.InlongClusterEntityMapper;
+import org.apache.inlong.manager.dao.mapper.ModuleConfigEntityMapper;
+import org.apache.inlong.manager.dao.mapper.PackageConfigEntityMapper;
+import org.apache.inlong.manager.dao.mapper.UserEntityMapper;
import org.apache.inlong.manager.pojo.cluster.ClusterNodeRequest;
import org.apache.inlong.manager.pojo.cluster.agent.AgentClusterNodeRequest;
+import org.apache.inlong.manager.service.cmd.CommandExecutor;
+import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
@Service
public class AgentClusterNodeInstallOperator implements InlongClusterNodeInstallOperator {
private static final Logger LOGGER = LoggerFactory.getLogger(AgentClusterNodeInstallOperator.class);
+ public static final String INSTALLER_CONF_PATH = "/conf/installer.properties";
+ public static final String INSTALLER_START_CMD = "/bin/installer.sh start";
+ public static final String AGENT_MANAGER_AUTH_SECRET_ID = "agent.manager.auth.secretId";
+ public static final String AGENT_MANAGER_AUTH_SECRET_KEY = "agent.manager.auth.secretKey";
+ public static final String AGENT_MANAGER_ADDR = "agent.manager.addr";
+ public static final String AGENT_CLUSTER_NAME = "agent.cluster.name";
+ public static final String AGENT_CLUSTER_TAG = "agent.cluster.tag";
+ public static final String AUDIT_PROXYS_URL = "audit.proxys";
+ public static final String AGENT_LOCAL_IP = "agent.local.ip";
+
@Autowired
private InlongClusterEntityMapper clusterEntityMapper;
+ @Autowired
+ private CommandExecutor commandExecutor;
+ @Autowired
+ private ModuleConfigEntityMapper moduleConfigEntityMapper;
+ @Autowired
+ private PackageConfigEntityMapper packageConfigEntityMapper;
+ @Autowired
+ private UserEntityMapper userEntityMapper;
+
+ @Value("${metrics.audit.proxy.hosts:127.0.0.1:10081}")
+ private String auditProxyUrl;
+ @Value("${agent.install.path:inlong/inlong-installer/}")
+ private String agentInstallPath;
+ @Value("${manager.url:127.0.0.1:8083}")
+ private String managerUrl;
@Override
public Boolean accept(String clusterType) {
@@ -49,9 +91,38 @@ public String getClusterNodeType() {
@Override
public boolean install(ClusterNodeRequest clusterNodeRequest, String operator) {
- // todo Provide agent installation capability
- AgentClusterNodeRequest agentNodeRequest = (AgentClusterNodeRequest) clusterNodeRequest;
- InlongClusterEntity clusterEntity = clusterEntityMapper.selectById(clusterNodeRequest.getParentId());
+ LOGGER.info("begin to insert agent inlong cluster node={}", clusterNodeRequest);
+ try {
+ InlongClusterEntity clusterEntity = clusterEntityMapper.selectById(clusterNodeRequest.getParentId());
+ AgentClusterNodeRequest request = (AgentClusterNodeRequest) clusterNodeRequest;
+ commandExecutor.mkdir(request, agentInstallPath);
+ String downLoadUrl = getInstallerDownLoadUrl(request);
+ String fileName = downLoadUrl.substring(downLoadUrl.lastIndexOf('/') + 1);
+ commandExecutor.downLoadPackage(request, agentInstallPath, downLoadUrl);
+ commandExecutor.tarPackage(request, fileName, agentInstallPath);
+ String confFile = agentInstallPath + INSTALLER_CONF_PATH;
+ Map configMap = new HashMap<>();
+ configMap.put(AGENT_LOCAL_IP, request.getIp());
+ configMap.put(AGENT_MANAGER_ADDR, managerUrl);
+ UserEntity userInfo = userEntityMapper.selectByName(operator);
+ Preconditions.expectNotNull(userInfo, "User doesn't exist");
+ String secretKey =
+ new String(AESUtils.decryptAsString(userInfo.getSecretKey(), userInfo.getEncryptVersion()));
+ configMap.put(AGENT_MANAGER_AUTH_SECRET_ID, operator);
+ configMap.put(AGENT_MANAGER_AUTH_SECRET_KEY, secretKey);
+ configMap.put(AGENT_CLUSTER_TAG, clusterEntity.getClusterTags());
+ configMap.put(AGENT_CLUSTER_NAME, clusterEntity.getName());
+ configMap.put(AUDIT_PROXYS_URL, auditProxyUrl);
+ commandExecutor.modifyConfig(request, configMap, confFile);
+ String startCmd = agentInstallPath + INSTALLER_START_CMD;
+ commandExecutor.execRemote(request, startCmd);
+
+ } catch (Exception e) {
+ String errMsg = String.format("install installer failed for ip=%s", clusterNodeRequest.getIp());
+ LOGGER.error(errMsg, e);
+ throw new BusinessException(errMsg);
+ }
+ LOGGER.info("success to insert agent inlong cluster node={}", clusterNodeRequest);
return true;
}
@@ -61,4 +132,23 @@ public boolean unload(InlongClusterNodeEntity clusterNodeEntity, String operator
InlongClusterEntity clusterEntity = clusterEntityMapper.selectById(clusterNodeEntity.getParentId());
return true;
}
+
+ private String getInstallerDownLoadUrl(AgentClusterNodeRequest request) {
+ if (CollectionUtils.isEmpty(request.getModuleIdList())) {
+ throw new BusinessException(
+ String.format("install failed when module id list is null for ip=%s, type=%s", request.getIp(),
+ request.getType()));
+ }
+ for (Integer moduleId : request.getModuleIdList()) {
+ ModuleConfigEntity moduleConfigEntity = moduleConfigEntityMapper.selectByPrimaryKey(moduleId);
+ if (Objects.equals(moduleConfigEntity.getType(), ModuleType.INSTALLER.name())) {
+ PackageConfigEntity packageConfigEntity = packageConfigEntityMapper.selectByPrimaryKey(
+ moduleConfigEntity.getPackageId());
+ return packageConfigEntity.getDownloadUrl();
+ }
+ }
+ throw new BusinessException(
+ String.format("can't get installer download url for ip=%s, type=%s", request.getIp(),
+ request.getType()));
+ }
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/CommandExecutor.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/CommandExecutor.java
new file mode 100644
index 00000000000..e7f8eac0d83
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/CommandExecutor.java
@@ -0,0 +1,40 @@
+/*
+ * 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.inlong.manager.service.cmd;
+
+import org.apache.inlong.manager.pojo.cluster.agent.AgentClusterNodeRequest;
+
+import java.util.Map;
+
+public interface CommandExecutor {
+
+ CommandResult exec(String cmd) throws Exception;
+
+ CommandResult execRemote(AgentClusterNodeRequest clusterNodeRequest, String cmd) throws Exception;
+
+ CommandResult modifyConfig(AgentClusterNodeRequest clusterNodeRequest, Map configMap,
+ String confPath) throws Exception;
+
+ CommandResult tarPackage(AgentClusterNodeRequest clusterNodeRequest, String fileName, String tarPath)
+ throws Exception;
+
+ CommandResult downLoadPackage(AgentClusterNodeRequest clusterNodeRequest, String downLoadPath, String downLoadUrl)
+ throws Exception;
+
+ CommandResult mkdir(AgentClusterNodeRequest clusterNodeRequest, String path) throws Exception;
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/CommandExecutorImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/CommandExecutorImpl.java
new file mode 100644
index 00000000000..b1729b9b9ad
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/CommandExecutorImpl.java
@@ -0,0 +1,120 @@
+/*
+ * 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.inlong.manager.service.cmd;
+
+import org.apache.inlong.manager.common.consts.InlongConstants;
+import org.apache.inlong.manager.pojo.cluster.agent.AgentClusterNodeRequest;
+import org.apache.inlong.manager.service.cmd.shell.ShellExecutorImpl;
+import org.apache.inlong.manager.service.cmd.shell.ShellTracker;
+
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@Service
+public class CommandExecutorImpl implements CommandExecutor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CommandExecutorImpl.class);
+
+ @Override
+ public CommandResult exec(String cmd) throws Exception {
+ ShellTracker shellTracker = new ShellTracker();
+ ShellExecutorImpl shellExecutor = new ShellExecutorImpl(shellTracker);
+ shellExecutor.syncExec("sh", "-c", cmd);
+ String cmdMsg = String.join(InlongConstants.BLANK, "sh", "-c", cmd);
+ LOG.debug("run command : {}", cmdMsg);
+ CommandResult commandResult = new CommandResult();
+ commandResult.setCode(shellTracker.getCode());
+ commandResult.setResult(String.join(InlongConstants.BLANK, shellTracker.getResult()));
+ if (commandResult.getCode() != 0) {
+ throw new Exception("command " + cmdMsg + " exec failed, code = " +
+ commandResult.getCode() + ", output = " + commandResult.getResult());
+ }
+ LOG.debug(commandResult.getResult());
+ return commandResult;
+ }
+
+ @Override
+ public CommandResult execRemote(AgentClusterNodeRequest clusterNodeRequest, String cmd) throws Exception {
+ String cmdShell = "./conf/exec_cmd.exp";
+ String ip = clusterNodeRequest.getIp();
+ String port = String.valueOf(clusterNodeRequest.getSshPort());
+ String user = clusterNodeRequest.getUsername();
+ String password = clusterNodeRequest.getPassword();
+ String remoteCommandTimeout = "20000";
+
+ cmd = "sh -c \"" + cmd + "\"";
+ String cmdMsg =
+ String.join(InlongConstants.BLANK, cmdShell, ip, user, password, remoteCommandTimeout, cmd, port);
+ LOG.info("run remote command : {}", cmdMsg);
+
+ ShellTracker shellTracker = new ShellTracker();
+ ShellExecutorImpl shellExecutor = new ShellExecutorImpl(shellTracker);
+ shellExecutor.syncExec(cmdShell, ip, user, password, remoteCommandTimeout, cmd, port);
+
+ CommandResult commandResult = new CommandResult();
+ commandResult.setCode(shellTracker.getCode());
+ commandResult.setResult(String.join(InlongConstants.BLANK, shellTracker.getResult()));
+
+ LOG.debug(commandResult.getResult());
+ if (commandResult.getCode() != 0) {
+ throw new Exception(
+ "remote command " + cmdMsg + " exec failed, code = " + commandResult.getCode() + ", output = "
+ + commandResult.getResult());
+ }
+ return commandResult;
+ }
+
+ @Override
+ public CommandResult modifyConfig(AgentClusterNodeRequest clusterNodeRequest, Map configMap,
+ String confPath)
+ throws Exception {
+ List configList = configMap.entrySet().stream()
+ .map(entry -> "grep " + entry.getKey() + " " + confPath + " && sed -i 's%^" + entry.getKey() + ".*%"
+ + entry.getKey() + InlongConstants.EQUAL + entry.getValue() + "%' " + confPath
+ + " || echo " + entry.getKey() + InlongConstants.EQUAL + entry.getValue() + " >> "
+ + confPath)
+ .collect(Collectors.toList());
+ String modifyCmd = StringUtils.join(configList, ";");
+ return this.execRemote(clusterNodeRequest, modifyCmd);
+ }
+
+ @Override
+ public CommandResult tarPackage(AgentClusterNodeRequest clusterNodeRequest, String fileName,
+ String tarPath) throws Exception {
+ String tarCmd = "tar -zxvf " + tarPath + fileName + " -C " + tarPath;
+ return execRemote(clusterNodeRequest, tarCmd);
+ }
+
+ @Override
+ public CommandResult downLoadPackage(AgentClusterNodeRequest clusterNodeRequest, String downLoadPath,
+ String downLoadUrl) throws Exception {
+ return execRemote(clusterNodeRequest, "wget -P " + downLoadPath + InlongConstants.BLANK + downLoadUrl);
+ }
+
+ @Override
+ public CommandResult mkdir(AgentClusterNodeRequest clusterNodeRequest, String path) throws Exception {
+ return execRemote(clusterNodeRequest, "mkdir " + path);
+ }
+
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/CommandResult.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/CommandResult.java
new file mode 100644
index 00000000000..6cbb594d1f2
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/CommandResult.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.service.cmd;
+
+/**
+ * Command result
+ */
+public class CommandResult {
+
+ private int code = 0;
+ private String result;
+ private String errMsg;
+
+ public int getCode() {
+ return code;
+ }
+
+ public void setCode(int code) {
+ this.code = code;
+ }
+
+ public String getResult() {
+ return result;
+ }
+
+ public void setResult(String result) {
+ this.result = result;
+ }
+
+ public String getErrMsg() {
+ return errMsg;
+ }
+
+ public void setErrMsg(String errMsg) {
+ this.errMsg = errMsg;
+ }
+
+ @Override
+ public String toString() {
+ return "CommandResult{" +
+ "code=" + code +
+ ", stdout='" + result + '\'' +
+ ", stderr='" + errMsg + '\'' +
+ '}';
+ }
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/shell/ShellExecutor.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/shell/ShellExecutor.java
new file mode 100644
index 00000000000..0a34401bdf9
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/shell/ShellExecutor.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.inlong.manager.service.cmd.shell;
+
+public interface ShellExecutor {
+
+ /**
+ * Execute shell commands
+ *
+ * @param shellPath shell path
+ * @param params params
+ */
+ void syncExec(String shellPath, String... params);
+
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/shell/ShellExecutorImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/shell/ShellExecutorImpl.java
new file mode 100644
index 00000000000..6b910684cae
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/shell/ShellExecutorImpl.java
@@ -0,0 +1,131 @@
+/*
+ * 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.inlong.manager.service.cmd.shell;
+
+import org.apache.inlong.manager.common.consts.InlongConstants;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+@Slf4j
+public class ShellExecutorImpl implements ShellExecutor {
+
+ private static final String[] EXCEPTION_REG = new String[]{"(.*)Caused by: (.*)Exception(.*)",
+ "(.*)java.net.UnknownHostException(.*)",
+ "(.*)Copy failed: java.io.IOException: Job failed!(.*)"};
+ private ShellTracker tracker;
+
+ public ShellExecutorImpl(ShellTracker tracker) {
+ this.tracker = tracker;
+ }
+
+ private static long getPid(Process process) {
+ try {
+ Field f = process.getClass().getDeclaredField("pid");
+ f.setAccessible(true);
+ return f.getLong(process);
+ } catch (Exception e) {
+ log.error("get pid failed", e);
+ return -1;
+ }
+ }
+
+ private static String[] merge(String shellPath, String[] paths) {
+ List cmds = new ArrayList();
+ cmds.add(shellPath);
+ for (String path : paths) {
+ if (StringUtils.isBlank(path)) {
+ continue;
+ }
+ cmds.add(path);
+ }
+ String[] strings = new String[cmds.size()];
+ cmds.toArray(strings);
+ return strings;
+ }
+
+ private static String arrayToString(Object[] array, String split) {
+ if (array == null || array.length == 0) {
+ return InlongConstants.BLANK;
+ }
+ StringBuilder str = new StringBuilder();
+ for (int i = 0, length = array.length; i < length; i++) {
+ if (i != 0) {
+ str.append(split);
+ }
+ str.append(array[i]);
+ }
+ return str.toString();
+ }
+
+ private static boolean HasException(String str) {
+ for (String reg : EXCEPTION_REG) {
+ Pattern pattern = Pattern.compile(reg);
+ Matcher matcher = pattern.matcher(str);
+ if (matcher.find()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ public void syncExec(String shellPath, String... params) {
+ List result = new ArrayList();
+ String[] cmds = merge(shellPath, params);
+ try {
+ Process ps = Runtime.getRuntime().exec(cmds);
+ long pid = getPid(ps);
+ tracker.setProcessId(pid);
+ BufferedReader br = new BufferedReader(new InputStreamReader(ps.getInputStream()));
+ String line;
+ boolean hasException = false;
+ while ((line = br.readLine()) != null) {
+ if (HasException(line)) {
+ hasException = true;
+ }
+ result.add(line);
+ tracker.setRunResult(arrayToString(result.toArray(), InlongConstants.NEW_LINE));
+ tracker.lineChange(line);
+ }
+ if (hasException) {
+ tracker.lineChange("Java exception exist in output");
+ tracker.setCode(-1);
+ return;
+ }
+ ps.waitFor();
+ int exitValue = ps.exitValue();
+ if (exitValue != 0) {
+ tracker.setCode(exitValue);
+ }
+ } catch (Exception e) {
+ log.error("sync exec shell failed", e);
+ result.add(e.getMessage());
+ tracker.setRunResult(arrayToString(result.toArray(), InlongConstants.NEW_LINE));
+ tracker.lineChange(e.getMessage());
+ tracker.setCode(-1);
+ }
+ }
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/shell/ShellTracker.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/shell/ShellTracker.java
new file mode 100644
index 00000000000..e0635faa3cc
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cmd/shell/ShellTracker.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.service.cmd.shell;
+
+import lombok.Data;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Data
+public class ShellTracker {
+
+ private List result = new ArrayList<>();
+ private int code;
+ private Long processId;
+ private String runResult;
+
+ public void lineChange(String line) {
+ result.add(line);
+ }
+}
diff --git a/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql b/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql
index c9f9a424b79..8f201963527 100644
--- a/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql
+++ b/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql
@@ -132,6 +132,9 @@ CREATE TABLE IF NOT EXISTS `inlong_cluster_node`
`type` varchar(20) NOT NULL COMMENT 'Cluster type, such as: AGENT, DATAPROXY, etc',
`ip` varchar(512) NOT NULL COMMENT 'Cluster IP, separated by commas, such as: 127.0.0.1:8080,host2:8081',
`port` int(6) NULL COMMENT 'Cluster port',
+ `username` varchar(256) DEFAULT NULL COMMENT 'Username for ssh',
+ `password` varchar(256) DEFAULT NULL COMMENT 'Password for ssh',
+ `ssh_port` int(11) DEFAULT NULL COMMENT 'Ssh port',
`protocol_type` varchar(20) DEFAULT NULL COMMENT 'DATAPROXY Source listen protocol type, such as: TCP/HTTP',
`node_load` int(11) DEFAULT '-1' COMMENT 'Current load value of the node',
`ext_params` mediumtext DEFAULT NULL COMMENT 'Another fields will be saved as JSON string',
diff --git a/inlong-manager/manager-web/sql/apache_inlong_manager.sql b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
index cff675f287c..41e55c52068 100644
--- a/inlong-manager/manager-web/sql/apache_inlong_manager.sql
+++ b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
@@ -143,6 +143,9 @@ CREATE TABLE IF NOT EXISTS `inlong_cluster_node`
`type` varchar(20) NOT NULL COMMENT 'Cluster type, such as: AGENT, DATAPROXY, etc',
`ip` varchar(512) NOT NULL COMMENT 'Cluster IP, separated by commas, such as: 127.0.0.1:8080,host2:8081',
`port` int(6) NULL COMMENT 'Cluster port',
+ `username` varchar(256) DEFAULT NULL COMMENT 'Username for ssh',
+ `password` varchar(256) DEFAULT NULL COMMENT 'Password for ssh',
+ `ssh_port` int(11) DEFAULT NULL COMMENT 'Ssh port',
`protocol_type` varchar(20) DEFAULT NULL COMMENT 'DATAPROXY Source listen protocol type, such as: TCP/HTTP',
`node_load` int(11) DEFAULT '-1' COMMENT 'Current load value of the node',
`ext_params` mediumtext DEFAULT NULL COMMENT 'Another fields will be saved as JSON string',
diff --git a/inlong-manager/manager-web/sql/changes-1.12.0.sql b/inlong-manager/manager-web/sql/changes-1.12.0.sql
index 36cbaefe9a6..18c5979da90 100644
--- a/inlong-manager/manager-web/sql/changes-1.12.0.sql
+++ b/inlong-manager/manager-web/sql/changes-1.12.0.sql
@@ -15,8 +15,8 @@
* limitations under the License.
*/
--- This is the SQL change file from version 1.9.0 to the current version 1.10.0.
--- When upgrading to version 1.10.0, please execute those SQLs in the DB (such as MySQL) used by the Manager module.
+-- This is the SQL change file from version 1.11.0 to the current version 1.12.0.
+-- When upgrading to version 1.12.0, please execute those SQLs in the DB (such as MySQL) used by the Manager module.
SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
diff --git a/inlong-manager/manager-web/sql/changes-1.13.0.sql b/inlong-manager/manager-web/sql/changes-1.13.0.sql
new file mode 100644
index 00000000000..7013c2990f3
--- /dev/null
+++ b/inlong-manager/manager-web/sql/changes-1.13.0.sql
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- This is the SQL change file from version 1.12.0 to the current version 1.13.0.
+-- When upgrading to version 1.13.0, please execute those SQLs in the DB (such as MySQL) used by the Manager module.
+
+SET NAMES utf8mb4;
+SET FOREIGN_KEY_CHECKS = 0;
+
+USE `apache_inlong_manager`;
+
+ALTER TABLE `inlong_cluster_node` ADD COLUMN `username` varchar(256) DEFAULT NULL COMMENT 'username for ssh';
+ALTER TABLE `inlong_cluster_node` ADD COLUMN `password` varchar(256) DEFAULT NULL COMMENT 'password for ssh';
+ALTER TABLE `inlong_cluster_node` ADD COLUMN `ssh_port` int(11) DEFAULT NULL COMMENT 'ssh port';
diff --git a/inlong-manager/manager-web/src/main/resources/application-dev.properties b/inlong-manager/manager-web/src/main/resources/application-dev.properties
index d5d7c047da7..ab8b5e2c4aa 100644
--- a/inlong-manager/manager-web/src/main/resources/application-dev.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-dev.properties
@@ -102,3 +102,5 @@ metrics.audit.proxy.hosts=127.0.0.1:10081
cls.manager.endpoint=127.0.0.1
+manager.url=127.0.0.1:8083
+agent.install.path=
\ No newline at end of file
diff --git a/inlong-manager/manager-web/src/main/resources/exec_cmd.exp b/inlong-manager/manager-web/src/main/resources/exec_cmd.exp
new file mode 100644
index 00000000000..a56079f96dd
--- /dev/null
+++ b/inlong-manager/manager-web/src/main/resources/exec_cmd.exp
@@ -0,0 +1,41 @@
+#!/usr/bin/expect
+#
+# 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.
+#
+set remoteHost [lindex $argv 0]
+set remoteUser [lindex $argv 1]
+set password [lindex $argv 2]
+set cmdTimeout [lindex $argv 3]
+set runCommand [lindex $argv 4]
+set remotePort [lindex $argv 5]
+
+spawn ssh -p ${remotePort} ${remoteHost} -l ${remoteUser} "${runCommand} && echo \\#SUCCESS\\# || echo \\#fail\\#"
+
+set timeout ${cmdTimeout}
+expect {
+ "*yes/no)?" {send "yes\n"; exp_continue}
+ "*assword:" {send "${password}\n"; exp_continue }
+ "Last login:" {}
+ "#SUCCESS#" {}
+ "#fail#" {exit 1 }
+ "*No route to host" {exit 2}
+ "Permission denied" {exit 3}
+ "*Host key verification failed*" {exit 4}
+ timeout { exit 5 }
+ eof { exit 6 }
+}
\ No newline at end of file