You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2021/11/11 08:39:03 UTC
[incubator-doris] branch doris-manager updated: [Manager] Adapt to
manger product prototype (#7036)
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch doris-manager
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/doris-manager by this push:
new 2d23761 [Manager] Adapt to manger product prototype (#7036)
2d23761 is described below
commit 2d2376129008f9e418249d569fcf670bc49a3594
Author: weajun <we...@gmail.com>
AuthorDate: Thu Nov 11 16:38:49 2021 +0800
[Manager] Adapt to manger product prototype (#7036)
1. Support recording task status when installing a cluster
2. Integrated user permission verification, etc.
3. Use JDBC to add BE nodes
4. Increase the installation progress query, log query and other interfaces
4. Modify the interface document
5. Optimize the agent log code
---
manager/build.sh | 2 +-
.../doris/manager/agent/api/CommandController.java | 26 +-
.../doris/manager/agent/api/LogController.java | 117 ++++
.../doris/manager/agent/command/Command.java | 2 +-
.../doris/manager/agent/command/CommandResult.java | 58 --
.../agent/command/CommandResultService.java | 87 +--
.../manager/agent/command/WriteBeConfCommand.java | 3 +-
.../manager/agent/command/WriteFeConfCommand.java | 3 +-
.../doris/manager/agent/common/AgentConstants.java | 9 +-
.../doris/manager/agent/register/AgentContext.java | 2 +-
.../manager/agent/register/AgentHeartbeat.java | 2 +-
.../doris/manager/agent/service/BeService.java | 36 +-
.../manager/agent/service/ServiceContext.java | 4 +-
.../manager/agent/task/AbsAsyncTaskHandler.java | 2 +
.../doris/manager/agent/task/ScriptTask.java | 16 +-
.../org/apache/doris/manager/agent/task/Task.java | 24 +-
.../doris/manager/agent/task/TaskLruLog.java | 92 ----
.../dm-agent/src/main/resources/logback-spring.xml | 17 +
.../manager/common/domain/CommandResult.java} | 25 +-
.../doris/manager/common/domain/RResult.java | 4 +
.../doris/manager/common/domain}/TaskResult.java | 2 +-
.../doris/manager/common/domain}/TaskState.java | 8 +-
manager/dm-server/pom.xml | 26 +-
.../doris/stack/agent/AgentHeatbeatRunner.java | 10 +-
.../org/apache/doris/stack/agent/AgentRest.java | 22 +-
...tbeatRunner.java => AgentTaskStatusRunner.java} | 49 +-
.../doris/stack/bean/SpringApplicationContext.java | 46 ++
.../doris/stack/component/AgentComponent.java | 18 +-
.../doris/stack/component/AgentRoleComponent.java | 20 +-
.../stack/component/ProcessInstanceComponent.java | 82 +++
.../stack/component/TaskInstanceComponent.java | 132 +++++
.../apache/doris/stack/constants/AgentStatus.java | 2 +
.../apache/doris/stack/constants/CmdTypeEnum.java | 3 +-
.../apache/doris/stack/constants/Constants.java | 3 +-
.../doris/stack/constants/ExecutionStatus.java | 88 +++
.../{req/TaskInfoReq.java => constants/Flag.java} | 38 +-
.../doris/stack/constants/ProcessTypeEnum.java | 73 +++
.../{CmdTypeEnum.java => TaskTypeEnum.java} | 31 +-
.../doris/stack/controller/AgentController.java | 95 ++--
.../stack/controller/ProcessTaskController.java | 116 ++++
.../doris/stack/controller/ServerController.java | 42 +-
.../apache/doris/stack/dao/AgentRepository.java | 9 +-
.../doris/stack/dao/AgentRoleRepository.java | 20 +-
...ository.java => ProcessInstanceRepository.java} | 20 +-
...Repository.java => TaskInstanceRepository.java} | 23 +-
.../org/apache/doris/stack/entity/AgentEntity.java | 24 +-
.../apache/doris/stack/entity/AgentRoleEntity.java | 26 +-
...AgentEntity.java => ProcessInstanceEntity.java} | 52 +-
.../doris/stack/entity/TaskInstanceEntity.java | 109 ++++
.../AgentRegister.java => model/AgentInstall.java} | 42 +-
.../{req/BeJoinReq.java => model/BeJoin.java} | 23 +-
.../DeployConfig.java} | 22 +-
.../AgentRoleEntity.java => model/DorisStart.java} | 22 +-
.../InstallInfo.java} | 25 +-
.../request/AgentCommon.java} | 20 +-
.../request/AgentInstallReq.java} | 27 +-
.../request/AgentRegister.java} | 19 +-
.../stack/{req => model/request}/BeJoinReq.java | 19 +-
.../stack/model/request/DeployConfigReq.java} | 28 +-
.../request/DorisExecReq.java} | 24 +-
.../{req => model/request}/DorisInstallReq.java | 20 +-
.../request/DorisStartReq.java} | 25 +-
.../org/apache/doris/stack/req/DeployConfig.java | 56 --
.../java/org/apache/doris/stack/req/DorisExec.java | 51 --
.../org/apache/doris/stack/req/DorisExecReq.java | 44 --
.../org/apache/doris/stack/req/InstallInfo.java | 94 ----
.../java/org/apache/doris/stack/req/SshInfo.java | 73 ---
.../TaskContext.java} | 28 +-
.../doris/stack/runner/TaskExecCallback.java | 65 +++
.../doris/stack/runner/TaskExecuteThread.java | 76 +++
.../{ServerAgent.java => AgentProcess.java} | 37 +-
.../apache/doris/stack/service/ProcessTask.java | 57 ++
.../apache/doris/stack/service/ServerProcess.java | 14 +-
.../doris/stack/service/impl/AgentProcessImpl.java | 482 +++++++++++++++++
.../doris/stack/service/impl/ProcessTaskImpl.java | 164 ++++++
.../doris/stack/service/impl/ServerAgentImpl.java | 313 -----------
.../stack/service/impl/ServerProcessImpl.java | 196 +++----
.../java/org/apache/doris/stack/shell/SCP.java | 6 +-
.../java/org/apache/doris/stack/shell/SSH.java | 6 +-
.../TaskInfoReq.java => task/AbstractTask.java} | 28 +-
.../apache/doris/stack/task/InstallAgentTask.java | 162 ++++++
.../org/apache/doris/stack/task/JoinBeTask.java | 71 +++
.../java/org/apache/doris/stack/util/JdbcUtil.java | 14 +-
.../org/apache/doris/stack/util/Preconditions.java | 47 --
manager/dm-server/src/main/resources/dm-server.sql | 19 -
...216\245\345\217\243\346\226\207\346\241\243.md" | 588 +++++++++++++++------
manager/manager-bin/agent/bin/agent_start.sh | 3 +
manager/manager-bin/agent/bin/agent_stop.sh | 3 +
manager/manager-bin/agent/bin/install_be.sh | 30 +-
manager/manager-bin/agent/bin/install_fe.sh | 29 +-
90 files changed, 3034 insertions(+), 1758 deletions(-)
diff --git a/manager/build.sh b/manager/build.sh
index 6675257..36b562e 100644
--- a/manager/build.sh
+++ b/manager/build.sh
@@ -43,8 +43,8 @@ mkdir -p output/server/lib
mv manager-server/target/manager-server-1.0.0.jar output/server/lib/doris-manager.jar
cp -r conf output/server/
cp -r manager-bin output/
+mv output/manager-bin/agent output/
mv output/manager-bin output/server/bin
-cp -r dm-agent/src/main/resources/agent output/
mkdir -p output/agent/lib
mv dm-agent/target/dm-agent-1.0.0.jar output/agent/lib/dm-agent.jar
cp -r manager-server/src/main/resources/web-resource output/server/
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/api/CommandController.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/api/CommandController.java
index 8ba04cf..840138b 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/api/CommandController.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/api/CommandController.java
@@ -18,12 +18,10 @@
package org.apache.doris.manager.agent.api;
import org.apache.doris.manager.agent.command.CommandFactory;
-import org.apache.doris.manager.agent.command.CommandResult;
import org.apache.doris.manager.agent.command.CommandResultService;
-import org.apache.doris.manager.agent.common.AgentConstants;
import org.apache.doris.manager.agent.task.Task;
-import org.apache.doris.manager.agent.task.TaskContext;
import org.apache.doris.manager.common.domain.CommandRequest;
+import org.apache.doris.manager.common.domain.CommandResult;
import org.apache.doris.manager.common.domain.CommandType;
import org.apache.doris.manager.common.domain.RResult;
import org.springframework.web.bind.annotation.GetMapping;
@@ -42,27 +40,19 @@ public class CommandController {
@PostMapping("/execute")
public RResult execute(@RequestBody CommandRequest commandRequest) {
if (Objects.isNull(CommandType.findByName(commandRequest.getCommandType()))) {
- return RResult.error("unkonwn command");
+ return RResult.error("Unkonwn command");
}
Task task = CommandFactory.get(commandRequest).setup().execute();
- CommandResult commandResult = new CommandResult(task.getTaskResult(),
- task.getTasklog().allStdLog(),
- task.getTasklog().allErrLog());
+ CommandResult commandResult = new CommandResult(task.getTaskResult());
return RResult.success(commandResult);
}
@GetMapping("/result")
public RResult commandResult(@RequestParam String taskId) {
- return RResult.success(CommandResultService.commandResult(taskId));
- }
-
- @GetMapping("/stdlog")
- public RResult getTaskStdLog(@RequestParam String taskId, @RequestParam int offset) {
- return RResult.success(TaskContext.getTaskByTaskId(taskId) == null ? null : TaskContext.getTaskByTaskId(taskId).getTasklog().stdLog(offset, AgentConstants.COMMAND_LOG_PAGE_SIZE));
- }
-
- @GetMapping("/errlog")
- public RResult getTaskErrLog(@RequestParam String taskId, @RequestParam int offset) {
- return RResult.success(TaskContext.getTaskByTaskId(taskId) == null ? null : TaskContext.getTaskByTaskId(taskId).getTasklog().errLog(offset, AgentConstants.COMMAND_LOG_PAGE_SIZE));
+ CommandResult commandResult = CommandResultService.commandResult(taskId);
+ if (Objects.isNull(commandResult)) {
+ return RResult.error("Task not found");
+ }
+ return RResult.success(commandResult);
}
}
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/api/LogController.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/api/LogController.java
new file mode 100644
index 0000000..d523d15
--- /dev/null
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/api/LogController.java
@@ -0,0 +1,117 @@
+// 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.doris.manager.agent.api;
+
+import com.google.common.base.Preconditions;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.doris.manager.agent.common.AgentConstants;
+import org.apache.doris.manager.agent.exception.AgentException;
+import org.apache.doris.manager.agent.register.AgentContext;
+import org.apache.doris.manager.agent.service.Service;
+import org.apache.doris.manager.agent.service.ServiceContext;
+import org.apache.doris.manager.common.domain.RResult;
+import org.apache.doris.manager.common.domain.ServiceRole;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * fe be log
+ **/
+@Slf4j
+@RestController
+@RequestMapping("/log")
+public class LogController {
+
+ private static final long WEB_LOG_BYTES = 1024 * 1024; // 1MB
+
+ /**
+ * Obtain service logs according to type: fe.log/fe.out/be.log/be.out
+ */
+ @GetMapping
+ public RResult serviceLog(@RequestParam String type) {
+ String logPath = getLogPath(type);
+ Map<String, String> result = appendLogInfo(logPath);
+ return RResult.success(result);
+ }
+
+ /**
+ * Obtain task log
+ */
+ @GetMapping("/task")
+ public RResult taskLog(@RequestParam String taskId) {
+ String logPath = AgentContext.getAgentInstallDir() + AgentConstants.TASK_LOG_FILE_RELATIVE_PATH;
+ Map<String, String> result = appendLogInfo(logPath);
+ return RResult.success(result);
+ }
+
+ private String getLogPath(String type) {
+ Preconditions.checkArgument(StringUtils.isNotBlank(type), "type is required");
+ Map<ServiceRole, Service> serviceMap = ServiceContext.getServiceMap();
+ if (type.startsWith(ServiceRole.FE.name().toLowerCase())) {
+ return serviceMap.get(ServiceRole.FE).getInstallDir() + AgentConstants.LOG_FILE_RELATIVE_PATH + type;
+ } else if (type.startsWith(ServiceRole.BE.name().toLowerCase())) {
+ return serviceMap.get(ServiceRole.BE).getInstallDir() + AgentConstants.LOG_FILE_RELATIVE_PATH + type;
+ } else {
+ throw new AgentException("can not find this type log path:" + type);
+ }
+ }
+
+ private Map<String, String> appendLogInfo(String logPath) {
+ Map<String, String> map = new HashMap<>();
+ map.put("logPath", logPath);
+
+ RandomAccessFile raf = null;
+ try {
+ raf = new RandomAccessFile(logPath, "r");
+ long fileSize = raf.length();
+ long startPos = fileSize < WEB_LOG_BYTES ? 0L : fileSize - WEB_LOG_BYTES;
+ long webContentLength = fileSize < WEB_LOG_BYTES ? fileSize : WEB_LOG_BYTES;
+ raf.seek(startPos);
+ map.put("showingLast", webContentLength + " bytes of log");
+ StringBuilder sb = new StringBuilder();
+ String line = "";
+ while ((line = raf.readLine()) != null) {
+ sb.append(line).append("\n");
+ }
+ map.put("log", sb.toString());
+
+ } catch (FileNotFoundException e) {
+ map.put("error", "Couldn't open log file: " + logPath);
+ } catch (IOException e) {
+ map.put("error", "Failed to read log file: " + logPath);
+ } finally {
+ try {
+ if (raf != null) {
+ raf.close();
+ }
+ } catch (IOException e) {
+ log.warn("fail to close log file: " + logPath, e);
+ }
+ }
+ return map;
+ }
+}
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/Command.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/Command.java
index 1debb53..afbdde2 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/Command.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/Command.java
@@ -22,9 +22,9 @@ import org.apache.doris.manager.agent.task.Task;
import org.apache.doris.manager.common.domain.CommandType;
public abstract class Command {
+ protected CommandType commandType;
private Task task;
private ITaskHandlerFactory handlerFactory;
- protected CommandType commandType;
public Command setup() {
task = setupTask();
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/CommandResult.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/CommandResult.java
deleted file mode 100644
index 7431854..0000000
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/CommandResult.java
+++ /dev/null
@@ -1,58 +0,0 @@
-// 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.doris.manager.agent.command;
-
-import org.apache.doris.manager.agent.task.TaskResult;
-
-import java.util.List;
-
-public class CommandResult {
- private TaskResult taskResult;
- private List<String> stdlogs;
- private List<String> errlogs;
-
- public CommandResult(TaskResult taskResult, List<String> stdlogs, List<String> errlogs) {
- this.taskResult = taskResult;
- this.stdlogs = stdlogs;
- this.errlogs = errlogs;
- }
-
- public TaskResult getTaskResult() {
- return taskResult;
- }
-
- public void setTaskResult(TaskResult taskResult) {
- this.taskResult = taskResult;
- }
-
- public List<String> getStdlogs() {
- return stdlogs;
- }
-
- public void setStdlogs(List<String> stdlogs) {
- this.stdlogs = stdlogs;
- }
-
- public List<String> getErrlogs() {
- return errlogs;
- }
-
- public void setErrlogs(List<String> errlogs) {
- this.errlogs = errlogs;
- }
-}
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/CommandResultService.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/CommandResultService.java
index 1c2516f..7aaef5e 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/CommandResultService.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/CommandResultService.java
@@ -19,24 +19,27 @@ package org.apache.doris.manager.agent.command;
import org.apache.doris.manager.agent.common.AgentConstants;
import org.apache.doris.manager.agent.service.ServiceContext;
+import org.apache.doris.manager.agent.task.LRU;
import org.apache.doris.manager.agent.task.Task;
import org.apache.doris.manager.agent.task.TaskContext;
-import org.apache.doris.manager.agent.task.TaskResult;
+import org.apache.doris.manager.common.domain.CommandResult;
import org.apache.doris.manager.common.domain.CommandType;
import org.apache.doris.manager.common.domain.ServiceRole;
+import org.apache.doris.manager.common.domain.TaskResult;
+import org.apache.doris.manager.common.domain.TaskState;
import java.util.Objects;
public class CommandResultService {
+ private static LRU<String, Integer> taskFinalStatus = new LRU<>(AgentConstants.COMMAND_HISTORY_SAVE_MAX_COUNT);
+
public static CommandResult commandResult(String taskId) {
if (Objects.isNull(TaskContext.getTaskByTaskId(taskId))) {
return null;
}
Task task = TaskContext.getTaskByTaskId(taskId);
- CommandResult commandResult = new CommandResult(task.getTaskResult(),
- task.getTasklog().allStdLog(),
- task.getTasklog().allErrLog());
+ CommandResult commandResult = new CommandResult(task.getTaskResult());
switch (CommandType.valueOf(task.getTaskDesc().getTaskName())) {
case INSTALL_FE:
@@ -45,57 +48,67 @@ public class CommandResultService {
case WRITE_FE_CONF:
return commandResult;
case START_FE:
- return feStateResult(task, false);
+ return fetchCommandResult(task, CommandType.START_FE);
case STOP_FE:
- return feStateResult(task, true);
+ return fetchCommandResult(task, CommandType.STOP_FE);
case START_BE:
- return beStateResult(task, false);
+ return fetchCommandResult(task, CommandType.START_BE);
case STOP_BE:
- return beStateResult(task, true);
+ return fetchCommandResult(task, CommandType.STOP_BE);
default:
return null;
}
}
- private static CommandResult feStateResult(Task task, boolean isStop) {
+ private static CommandResult fetchCommandResult(Task task, CommandType commandType) {
if (Objects.isNull(task.getTaskResult().getRetCode()) || task.getTaskResult().getRetCode() != 0) {
- return new CommandResult(task.getTaskResult(),
- task.getTasklog().allStdLog(),
- task.getTasklog().allErrLog());
+ return new CommandResult(task.getTaskResult());
}
- // todo: save final status
TaskResult tmpTaskResult = new TaskResult(task.getTaskResult());
- boolean health = ServiceContext.getServiceMap().get(ServiceRole.FE).isHealth();
- if (isStop) {
- tmpTaskResult.setRetCode(!health ? 0 : AgentConstants.COMMAND_EXECUTE_UNHEALTH_CODE);
- } else {
- tmpTaskResult.setRetCode(health ? 0 : AgentConstants.COMMAND_EXECUTE_UNHEALTH_CODE);
+
+ Integer finalStatus = taskFinalStatus.get(task.getTaskId());
+ if (Objects.nonNull(finalStatus)) {
+ tmpTaskResult.setTaskState(TaskState.FINISHED);
+ tmpTaskResult.setRetCode(finalStatus);
+ return new CommandResult(tmpTaskResult);
}
- return new CommandResult(tmpTaskResult,
- task.getTasklog().allStdLog(),
- task.getTasklog().allErrLog());
- }
+ int retCode = AgentConstants.COMMAND_EXECUTE_UNHEALTH_CODE;
+ Boolean health = false;
+ TaskState taskState = TaskState.RUNNING;
+ if (CommandType.START_FE == commandType) {
+ health = ServiceContext.getServiceMap().get(ServiceRole.FE).isHealth();
+ retCode = health ? AgentConstants.COMMAND_EXECUTE_SUCCESS_CODE : AgentConstants.COMMAND_EXECUTE_UNHEALTH_CODE;
+ taskState = health ? TaskState.FINISHED : TaskState.RUNNING;
+ } else if (CommandType.STOP_FE == commandType) {
+ health = ServiceContext.getServiceMap().get(ServiceRole.FE).isHealth();
+ retCode = !health ? AgentConstants.COMMAND_EXECUTE_SUCCESS_CODE : AgentConstants.COMMAND_EXECUTE_UNHEALTH_CODE;
+ taskState = !health ? TaskState.FINISHED : TaskState.RUNNING;
+ } else if (CommandType.START_BE == commandType) {
+ health = ServiceContext.getServiceMap().get(ServiceRole.BE).isHealth();
+ retCode = health ? AgentConstants.COMMAND_EXECUTE_SUCCESS_CODE : AgentConstants.COMMAND_EXECUTE_UNHEALTH_CODE;
+ taskState = health ? TaskState.FINISHED : TaskState.RUNNING;
+ } else if (CommandType.STOP_BE == commandType) {
+ health = ServiceContext.getServiceMap().get(ServiceRole.BE).isHealth();
+ retCode = !health ? AgentConstants.COMMAND_EXECUTE_SUCCESS_CODE : AgentConstants.COMMAND_EXECUTE_UNHEALTH_CODE;
+ taskState = !health ? TaskState.FINISHED : TaskState.RUNNING;
+ }
- private static CommandResult beStateResult(Task task, boolean isStop) {
- if (Objects.isNull(task.getTaskResult().getRetCode()) || task.getTaskResult().getRetCode() != 0) {
- return new CommandResult(task.getTaskResult(),
- task.getTasklog().allStdLog(),
- task.getTasklog().allErrLog());
+ if (health && (CommandType.START_FE == commandType || CommandType.START_BE == commandType)) {
+ taskFinalStatus.put(task.getTaskId(), AgentConstants.COMMAND_EXECUTE_SUCCESS_CODE);
+ }
+ if (!health && (CommandType.STOP_FE == commandType || CommandType.STOP_BE == commandType)) {
+ taskFinalStatus.put(task.getTaskId(), AgentConstants.COMMAND_EXECUTE_SUCCESS_CODE);
}
- // todo: save final status
- TaskResult tmpTaskResult = new TaskResult(task.getTaskResult());
- boolean health = ServiceContext.getServiceMap().get(ServiceRole.BE).isHealth();
- if (isStop) {
- tmpTaskResult.setRetCode(!health ? 0 : AgentConstants.COMMAND_EXECUTE_UNHEALTH_CODE);
- } else {
- tmpTaskResult.setRetCode(health ? 0 : AgentConstants.COMMAND_EXECUTE_UNHEALTH_CODE);
+ if (task.getTaskResult().getSubmitTime().getTime() + AgentConstants.COMMAND_EXECUTE_TIMEOUT_MILSECOND < System.currentTimeMillis()) {
+ taskFinalStatus.put(task.getTaskId(), AgentConstants.COMMAND_EXECUTE_TIMEOUT_CODE);
+ tmpTaskResult.setTaskState(TaskState.FINISHED);
}
- return new CommandResult(tmpTaskResult,
- task.getTasklog().allStdLog(),
- task.getTasklog().allErrLog());
+ tmpTaskResult.setRetCode(retCode);
+ tmpTaskResult.setTaskState(taskState);
+ return new CommandResult(tmpTaskResult);
}
}
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/WriteBeConfCommand.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/WriteBeConfCommand.java
index e4dc807..f2160d0 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/WriteBeConfCommand.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/WriteBeConfCommand.java
@@ -26,7 +26,6 @@ import org.apache.doris.manager.agent.task.Task;
import org.apache.doris.manager.agent.task.TaskDesc;
import org.apache.doris.manager.agent.task.TaskHandlerFactory;
import org.apache.doris.manager.agent.task.TaskHook;
-import org.apache.doris.manager.agent.task.TaskLruLog;
import org.apache.doris.manager.common.domain.CommandType;
import org.apache.doris.manager.common.domain.ServiceRole;
import org.apache.doris.manager.common.domain.WriteBeConfCommandRequestBody;
@@ -54,7 +53,7 @@ public class WriteBeConfCommand extends BeCommand {
desc.setCreateStorageDir(requestBody.isCreateStorageDir());
WriteBeConfTaskHook hook = new WriteBeConfTaskHook();
- return new Task<TaskDesc>(desc, new TaskLruLog(), hook) {
+ return new Task<TaskDesc>(desc, hook) {
@Override
protected int execute() throws IOException {
try (FileOutputStream fos = new FileOutputStream(ServiceContext.getServiceMap().get(ServiceRole.BE).getConfigFilePath());
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/WriteFeConfCommand.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/WriteFeConfCommand.java
index 63a2b59..3b2c6d2 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/WriteFeConfCommand.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/command/WriteFeConfCommand.java
@@ -26,7 +26,6 @@ import org.apache.doris.manager.agent.task.Task;
import org.apache.doris.manager.agent.task.TaskDesc;
import org.apache.doris.manager.agent.task.TaskHandlerFactory;
import org.apache.doris.manager.agent.task.TaskHook;
-import org.apache.doris.manager.agent.task.TaskLruLog;
import org.apache.doris.manager.common.domain.CommandType;
import org.apache.doris.manager.common.domain.ServiceRole;
import org.apache.doris.manager.common.domain.WriteFeConfCommandRequestBody;
@@ -55,7 +54,7 @@ public class WriteFeConfCommand extends FeCommand {
WriteFeConfTaskHook hook = new WriteFeConfTaskHook();
- return new Task<TaskDesc>(desc, new TaskLruLog(), hook) {
+ return new Task<TaskDesc>(desc, hook) {
@Override
protected int execute() throws IOException {
try (FileOutputStream fos = new FileOutputStream(ServiceContext.getServiceMap().get(ServiceRole.FE).getConfigFilePath());
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/common/AgentConstants.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/common/AgentConstants.java
index ddcf8d1..c727145 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/common/AgentConstants.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/common/AgentConstants.java
@@ -19,7 +19,10 @@ package org.apache.doris.manager.agent.common;
public class AgentConstants {
public static final int COMMAND_LOG_PAGE_SIZE = 1000;
- public static final int COMMAND_EXECUTE_UNHEALTH_CODE = -1;
+ public static final int COMMAND_EXECUTE_SUCCESS_CODE = 0;
+ public static final int COMMAND_EXECUTE_UNHEALTH_CODE = -10;
+ public static final int COMMAND_EXECUTE_TIMEOUT_CODE = -11;
+ public static final int COMMAND_EXECUTE_TIMEOUT_MILSECOND = 600000;
public static final int FE_HTTP_PORT_DEFAULT = 8030;
public static final int BE_HTTP_PORT_DEFAULT = 8040;
@@ -38,7 +41,11 @@ public class AgentConstants {
public static final String BE_DEFAULT_STORAGE_DIR_RELATIVE_PATH = "/storage";
public static final String FE_CONFIG_FILE_RELATIVE_PATH = "/conf/fe.conf";
public static final String BE_CONFIG_FILE_RELATIVE_PATH = "/conf/be.conf";
+ public static final String LOG_FILE_RELATIVE_PATH = "/log/";
+ public static final String TASK_LOG_FILE_RELATIVE_PATH = "/log/task.log";
public static final int TASK_ERROR_CODE_DEFAULT = -501;
public static final int TASK_ERROR_CODE_EXCEPTION = -502;
+
+ public static final String LOG_TYPE_TASK = "TASK_LOG";
}
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/register/AgentContext.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/register/AgentContext.java
index 3538feb..95d17a5 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/register/AgentContext.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/register/AgentContext.java
@@ -45,7 +45,7 @@ public class AgentContext {
}
private static boolean registerToServer(String agentIp, Integer agentPort, String agentServer, String agentInstallDir) {
- String requestUrl = "http://" + agentServer + "/server/register";
+ String requestUrl = "http://" + agentServer + "/api/server/register";
Map<String, Object> map = new HashMap<>();
map.put("host", agentIp);
map.put("port", agentPort);
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/register/AgentHeartbeat.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/register/AgentHeartbeat.java
index bdaabd3..880adce 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/register/AgentHeartbeat.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/register/AgentHeartbeat.java
@@ -46,7 +46,7 @@ public class AgentHeartbeat {
}
private static boolean heartbeat() {
- String requestUrl = "http://" + AgentContext.getAgentServer() + "/server/heartbeat";
+ String requestUrl = "http://" + AgentContext.getAgentServer() + "/api/server/heartbeat";
Map<String, Object> map = new HashMap<>();
map.put("host", AgentContext.getAgentIp());
map.put("port", AgentContext.getAgentPort());
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/service/BeService.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/service/BeService.java
index b6f66c6..ae85962 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/service/BeService.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/service/BeService.java
@@ -45,6 +45,24 @@ public class BeService extends Service {
doLoad();
}
+ private static List<String> parseStorageDirs(String storageRootPath) {
+ ArrayList<String> list = new ArrayList<>();
+ String[] splitArr = storageRootPath.split(";");
+ for (String split : splitArr) {
+ if (split.trim().length() <= 0) {
+ continue;
+ }
+
+ int lastIndex = split.lastIndexOf(".");
+ if (lastIndex == -1) {
+ list.add(split.substring(0));
+ } else {
+ list.add(split.substring(0, lastIndex));
+ }
+ }
+ return list;
+ }
+
@Override
public void doLoad() {
String httpPortStr = getConfig().getProperty(AgentConstants.BE_CONFIG_KEY_HTTP_PORT);
@@ -117,22 +135,4 @@ public class BeService extends Service {
}
}
}
-
- private static List<String> parseStorageDirs(String storageRootPath) {
- ArrayList<String> list = new ArrayList<>();
- String[] splitArr = storageRootPath.split(";");
- for (String split : splitArr) {
- if (split.trim().length() <= 0) {
- continue;
- }
-
- int lastIndex = split.lastIndexOf(".");
- if (lastIndex == -1) {
- list.add(split.substring(0));
- } else {
- list.add(split.substring(0, lastIndex));
- }
- }
- return list;
- }
}
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/service/ServiceContext.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/service/ServiceContext.java
index 906cd33..e2dcf85 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/service/ServiceContext.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/service/ServiceContext.java
@@ -76,7 +76,7 @@ public class ServiceContext {
}
private static boolean registerToServer(String serverIpPort, String agentIp, ServiceRole role, String serviceInstallDir) {
- String requestUrl = "http://" + serverIpPort + "/agent/register";
+ String requestUrl = "http://" + serverIpPort + "/api/agent/register";
Map<String, Object> map = new HashMap<>();
map.put("host", agentIp);
map.put("role", role.name());
@@ -96,7 +96,7 @@ public class ServiceContext {
}
private static List<AgentRoleRegister> queryServiceList(String serverIpPort, String agentIp) {
- String requestUrl = "http://" + serverIpPort + "/server/agentRole";
+ String requestUrl = "http://" + serverIpPort + "/api/server/agentRole";
Map<String, Object> map = new HashMap<>();
map.put("host", agentIp);
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/AbsAsyncTaskHandler.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/AbsAsyncTaskHandler.java
index e1df06f..95ed2e6 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/AbsAsyncTaskHandler.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/AbsAsyncTaskHandler.java
@@ -17,6 +17,8 @@
package org.apache.doris.manager.agent.task;
+import org.apache.doris.manager.common.domain.TaskState;
+
import java.util.concurrent.ExecutorService;
public abstract class AbsAsyncTaskHandler extends TaskHandler {
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/ScriptTask.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/ScriptTask.java
index 0b7dfcc..53e8ee5 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/ScriptTask.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/ScriptTask.java
@@ -17,18 +17,24 @@
package org.apache.doris.manager.agent.task;
+import org.apache.doris.manager.agent.common.AgentConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
public class ScriptTask extends Task<ScriptTaskDesc> {
+ private static final Logger TASKLOG = LoggerFactory.getLogger(AgentConstants.LOG_TYPE_TASK);
+
public ScriptTask(ScriptTaskDesc taskDesc) {
- super(taskDesc, new TaskLruLog());
+ super(taskDesc);
}
public ScriptTask(ScriptTaskDesc taskDesc, TaskHook taskHook) {
- super(taskDesc, new TaskLruLog(), taskHook);
+ super(taskDesc, taskHook);
}
@Override
@@ -38,13 +44,15 @@ public class ScriptTask extends Task<ScriptTaskDesc> {
BufferedReader stdInput = new BufferedReader(new InputStreamReader(proc.getInputStream()));
BufferedReader stdError = new BufferedReader(new InputStreamReader(proc.getErrorStream()));
+ TASKLOG.info("scriptCmd:{}", taskDesc.getScriptCmd());
+
String s = null;
while ((s = stdInput.readLine()) != null) {
- getTasklog().appendStdLog(s);
+ TASKLOG.info(s);
}
while ((s = stdError.readLine()) != null) {
- getTasklog().appendErrLog(s);
+ TASKLOG.info(s);
}
int exitVal = proc.waitFor();
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/Task.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/Task.java
index b8e585f..78788a8 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/Task.java
+++ b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/Task.java
@@ -18,18 +18,22 @@
package org.apache.doris.manager.agent.task;
import org.apache.doris.manager.agent.common.AgentConstants;
+import org.apache.doris.manager.common.domain.TaskResult;
+import org.apache.doris.manager.common.domain.TaskState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Date;
import java.util.UUID;
public abstract class Task<T extends TaskDesc> {
+ private static final Logger TASKLOG = LoggerFactory.getLogger(AgentConstants.LOG_TYPE_TASK);
+
protected T taskDesc;
private TaskResult taskResult;
- private ITaskLog tasklog;
-
private TaskHook taskHook;
protected Task(T taskDesc) {
@@ -38,21 +42,11 @@ public abstract class Task<T extends TaskDesc> {
this.taskResult.setTaskId(UUID.randomUUID().toString().replace("-", ""));
}
- public Task(T taskDesc, ITaskLog taskLruLog) {
- this(taskDesc);
- this.tasklog = taskLruLog;
- }
-
- public Task(T taskDesc, ITaskLog taskLruLog, TaskHook taskHook) {
+ public Task(T taskDesc, TaskHook taskHook) {
this(taskDesc);
- this.tasklog = taskLruLog;
this.taskHook = taskHook;
}
- public ITaskLog getTasklog() {
- return tasklog;
- }
-
public String getTaskId() {
return this.taskResult.getTaskId();
}
@@ -73,9 +67,7 @@ public abstract class Task<T extends TaskDesc> {
} catch (Throwable e) {
code = AgentConstants.TASK_ERROR_CODE_EXCEPTION;
e.printStackTrace();
- if (tasklog != null) {
- tasklog.appendErrLog(e.getMessage());
- }
+ TASKLOG.info(e.getMessage());
} finally {
this.taskResult.setEndTime(new Date());
this.taskResult.setRetCode(code);
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/TaskLruLog.java b/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/TaskLruLog.java
deleted file mode 100644
index 4329281..0000000
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/TaskLruLog.java
+++ /dev/null
@@ -1,92 +0,0 @@
-// 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.doris.manager.agent.task;
-
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.doris.manager.agent.common.AgentConstants;
-
-import java.util.ArrayList;
-import java.util.List;
-
-public class TaskLruLog implements ITaskLog {
- private LRU<Integer, String> stdlogs;
- private LRU<Integer, String> errlogs;
-
- private int currentStdPage = 0;
- private int currentErrPage = 0;
-
- public TaskLruLog() {
- this(AgentConstants.TASK_LOG_ROW_MAX_COUNT, AgentConstants.TASK_LOG_ROW_MAX_COUNT);
- }
-
- public TaskLruLog(int maxStdCacheSize, int maxErrCacheSize) {
- this.stdlogs = new LRU<>(maxStdCacheSize);
- this.errlogs = new LRU<>(maxErrCacheSize);
- }
-
- @Override
- public ImmutablePair<Integer, List<String>> stdLog(int offset, int size) {
- int count = 0;
- ArrayList<String> list = new ArrayList<>();
- String temp = null;
- if (offset > stdlogs.size()) {
- return new ImmutablePair(stdlogs.size() - 1 > 0 ? stdlogs.size() - 1 : 0, list);
- }
-
- while (count < size && offset + count < stdlogs.size()) {
- if ((temp = stdlogs.get(offset + count)) != null) {
- list.add(temp);
- }
- count++;
- }
- ImmutablePair<Integer, List<String>> pair = new ImmutablePair(offset + count, list);
- return pair;
- }
-
- @Override
- public List<String> errLog(int offset, int size) {
- int count = 0;
- ArrayList<String> list = new ArrayList<>();
- String temp = null;
- while (count < size) {
- if ((temp = errlogs.get(offset + count)) != null) {
- list.add(temp);
- }
- count++;
- }
- return list;
- }
-
- public List<String> allStdLog() {
- return stdLog(0, stdlogs.size()).getValue();
- }
-
- public List<String> allErrLog() {
- return errLog(0, errlogs.size());
- }
-
- @Override
- public void appendStdLog(String log) {
- stdlogs.put(currentStdPage++, log);
- }
-
- @Override
- public void appendErrLog(String log) {
- errlogs.put(currentErrPage++, log);
- }
-}
diff --git a/manager/dm-agent/src/main/resources/logback-spring.xml b/manager/dm-agent/src/main/resources/logback-spring.xml
index 78e0e5d..0bac9d4 100644
--- a/manager/dm-agent/src/main/resources/logback-spring.xml
+++ b/manager/dm-agent/src/main/resources/logback-spring.xml
@@ -36,8 +36,25 @@ under the License.
</encoder>
</appender>
+ <appender name="TASK_LOG_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+ <file>${LOG_HOME}/task.log</file>
+ <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+ <FileNamePattern>${LOG_HOME}/task.%d{yyyy-MM-dd}.%i.log</FileNamePattern>
+ <MaxHistory>30</MaxHistory>
+ <maxFileSize>50MB</maxFileSize>
+ <totalSizeCap>10GB</totalSizeCap>
+ </rollingPolicy>
+ <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
+ <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
+ </encoder>
+ </appender>
+
<root level="INFO">
<appender-ref ref="STDOUT"/>
<appender-ref ref="FILE"/>
</root>
+
+ <logger name="TASK_LOG" additivity="false" level="INFO">
+ <appender-ref ref="TASK_LOG_FILE"/>
+ </logger>
</configuration>
\ No newline at end of file
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/AgentCommon.java b/manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/CommandResult.java
similarity index 69%
rename from manager/dm-server/src/main/java/org/apache/doris/stack/req/AgentCommon.java
rename to manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/CommandResult.java
index 410af84..5503dad 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/AgentCommon.java
+++ b/manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/CommandResult.java
@@ -15,27 +15,20 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.req;
+package org.apache.doris.manager.common.domain;
-public class AgentCommon {
+public class CommandResult {
+ private TaskResult taskResult;
- private String host;
-
- private Integer port;
-
- public String getHost() {
- return host;
- }
-
- public void setHost(String host) {
- this.host = host;
+ public CommandResult(TaskResult taskResult) {
+ this.taskResult = taskResult;
}
- public Integer getPort() {
- return port;
+ public TaskResult getTaskResult() {
+ return taskResult;
}
- public void setPort(Integer port) {
- this.port = port;
+ public void setTaskResult(TaskResult taskResult) {
+ this.taskResult = taskResult;
}
}
diff --git a/manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/RResult.java b/manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/RResult.java
index 2fc15f7..d76636e 100644
--- a/manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/RResult.java
+++ b/manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/RResult.java
@@ -83,4 +83,8 @@ public class RResult extends HashMap<String, Object> {
public int getCode() {
return (int) this.get("code");
}
+
+ public boolean isSuccess() {
+ return getCode() == CODE_SUCCESS;
+ }
}
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/TaskResult.java b/manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/TaskResult.java
similarity index 97%
rename from manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/TaskResult.java
rename to manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/TaskResult.java
index 2c73ca3..52b9e50 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/TaskResult.java
+++ b/manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/TaskResult.java
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.manager.agent.task;
+package org.apache.doris.manager.common.domain;
import java.util.Date;
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/TaskState.java b/manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/TaskState.java
similarity index 85%
rename from manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/TaskState.java
rename to manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/TaskState.java
index 198a2b4..71769c4 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/TaskState.java
+++ b/manager/dm-common/src/main/java/org/apache/doris/manager/common/domain/TaskState.java
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.manager.agent.task;
+package org.apache.doris.manager.common.domain;
public enum TaskState {
// Task has been queued for execution by the driver
@@ -27,5 +27,9 @@ public enum TaskState {
// Task has completed
FINISHED,
// Task state is unkown
- UNKNOWN
+ UNKNOWN;
+
+ public boolean typeIsRunning() {
+ return this == INITIALIZED || this == QUEUED || this == RUNNING;
+ }
}
diff --git a/manager/dm-server/pom.xml b/manager/dm-server/pom.xml
index 067015d..dec0b43 100644
--- a/manager/dm-server/pom.xml
+++ b/manager/dm-server/pom.xml
@@ -26,30 +26,14 @@
<groupId>org.apache.doris</groupId>
<artifactId>general</artifactId>
</dependency>
-
- <dependency>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
- <version>2.5</version>
- </dependency>
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- <version>3.9</version>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>manager</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpclient</artifactId>
- <version>4.5.6</version>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <version>2.6</version>
</dependency>
-
</dependencies>
-
- <build>
- <resources>
- <resource>
- <directory>src/main/resources</directory>
- </resource>
- </resources>
- </build>
</project>
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentHeatbeatRunner.java b/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentHeatbeatRunner.java
index 3b9f8f6..c707da7 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentHeatbeatRunner.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentHeatbeatRunner.java
@@ -21,7 +21,6 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.doris.stack.constants.AgentStatus;
import org.apache.doris.stack.dao.AgentRepository;
import org.apache.doris.stack.entity.AgentEntity;
-import org.apache.doris.stack.service.ServerProcess;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
@@ -44,9 +43,6 @@ public class AgentHeatbeatRunner implements ApplicationRunner {
private static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
@Autowired
- private ServerProcess serverProcess;
-
- @Autowired
private AgentRepository agentRepository;
@Autowired
@@ -61,7 +57,7 @@ public class AgentHeatbeatRunner implements ApplicationRunner {
log.error("heartbeat check fail:", ex);
ex.printStackTrace();
}
- }, 0, HEALTH_TIME, TimeUnit.MILLISECONDS);
+ }, HEALTH_TIME, HEALTH_TIME, TimeUnit.MILLISECONDS);
}
/**
@@ -69,7 +65,7 @@ public class AgentHeatbeatRunner implements ApplicationRunner {
*/
private void heartbeatCheck() {
long currTime = System.currentTimeMillis();
- List<AgentEntity> agents = serverProcess.agentList();
+ List<AgentEntity> agents = agentRepository.findAll();
for (AgentEntity agent : agents) {
Date lastReportedTime = agent.getLastReportedTime();
long diff = HEALTH_TIME + 1;
@@ -77,7 +73,7 @@ public class AgentHeatbeatRunner implements ApplicationRunner {
diff = currTime - lastReportedTime.getTime();
}
if (diff > HEALTH_TIME) {
- agent.setStatus(AgentStatus.STOP.name());
+ agent.setStatus(AgentStatus.STOP);
agentRepository.save(agent);
agentCache.putAgent(agent);
log.warn("agent {} is unhealthly", agent.getHost());
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentRest.java b/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentRest.java
index ca5e746..442e27d 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentRest.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentRest.java
@@ -35,9 +35,9 @@ public class AgentRest {
private static final String AGENT_COMMAND_EXEC = "http://%s:%s/command/execute";
private static final String AGENT_TASK_INFO = "http://%s:%s/command/result?taskId={taskId}";
- private static final String AGENT_TASK_STD_LOG = "http://%s:%s/command/stdlog?taskId={taskId}&offset={offset}";
- private static final String AGENT_TASK_ERR_LOG = "http://%s:%s/command/errlog?taskId={taskId}&offset={offset}";
private static final String SERVICE_CONFIG = "http://%s:%s/service/config?serviceRole={serviceRole}";
+ private static final String SERVICE_LOG = "http://%s:%s/log?type={type}";
+ private static final String AGENT_TASK_LOG = "http://%s:%s/log/task?taskId={taskId}";
@Autowired
private RestTemplate restTemplate;
@@ -48,20 +48,26 @@ public class AgentRest {
return result;
}
- public RResult taskInfo(String host, Integer port, Map<String, Object> param) {
- String restUrl = String.format(AGENT_TASK_INFO, host, port);
+ public RResult serverLog(String host, Integer port, String type) {
+ Map<String, Object> param = Maps.newHashMap();
+ param.put("type", type);
+ String restUrl = String.format(SERVICE_LOG, host, port);
RResult result = restTemplate.getForObject(restUrl, RResult.class, param);
return result;
}
- public RResult taskStdLog(String host, Integer port, Map<String, Object> param) {
- String restUrl = String.format(AGENT_TASK_STD_LOG, host, port);
+ public RResult taskInfo(String host, Integer port, String taskId) {
+ Map<String, Object> param = Maps.newHashMap();
+ param.put("taskId", taskId);
+ String restUrl = String.format(AGENT_TASK_INFO, host, port);
RResult result = restTemplate.getForObject(restUrl, RResult.class, param);
return result;
}
- public RResult taskErrLog(String host, Integer port, Map<String, Object> param) {
- String restUrl = String.format(AGENT_TASK_ERR_LOG, host, port);
+ public RResult taskLog(String host, Integer port, String taskId) {
+ Map<String, Object> param = Maps.newHashMap();
+ param.put("taskId", taskId);
+ String restUrl = String.format(AGENT_TASK_LOG, host, port);
RResult result = restTemplate.getForObject(restUrl, RResult.class, param);
return result;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentHeatbeatRunner.java b/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentTaskStatusRunner.java
similarity index 53%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentHeatbeatRunner.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentTaskStatusRunner.java
index 3b9f8f6..6beb838 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentHeatbeatRunner.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/agent/AgentTaskStatusRunner.java
@@ -18,70 +18,51 @@
package org.apache.doris.stack.agent;
import lombok.extern.slf4j.Slf4j;
-import org.apache.doris.stack.constants.AgentStatus;
-import org.apache.doris.stack.dao.AgentRepository;
-import org.apache.doris.stack.entity.AgentEntity;
-import org.apache.doris.stack.service.ServerProcess;
+import org.apache.doris.stack.dao.ProcessInstanceRepository;
+import org.apache.doris.stack.entity.ProcessInstanceEntity;
+import org.apache.doris.stack.service.ProcessTask;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
-import java.util.Date;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
- * agent status check
+ * agent task status check
**/
@Component
@Slf4j
-public class AgentHeatbeatRunner implements ApplicationRunner {
+public class AgentTaskStatusRunner implements ApplicationRunner {
- private static final long HEALTH_TIME = 60 * 1000L;
+ private static final long REFRESH_TIME = 60 * 1000L;
private static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
@Autowired
- private ServerProcess serverProcess;
+ private ProcessTask processTask;
@Autowired
- private AgentRepository agentRepository;
-
- @Autowired
- private AgentCache agentCache;
+ private ProcessInstanceRepository processInstanceRepository;
@Override
public void run(ApplicationArguments args) throws Exception {
this.scheduler.scheduleWithFixedDelay(() -> {
try {
- heartbeatCheck();
+ refreshTaskStatus();
} catch (Exception ex) {
- log.error("heartbeat check fail:", ex);
+ log.error("refresh agent task fail:", ex);
ex.printStackTrace();
}
- }, 0, HEALTH_TIME, TimeUnit.MILLISECONDS);
+ }, REFRESH_TIME, REFRESH_TIME, TimeUnit.MILLISECONDS);
}
- /**
- * The last heartbeat time exceeds the HEALTH_TIME and is recognized as unhealthy
- */
- private void heartbeatCheck() {
- long currTime = System.currentTimeMillis();
- List<AgentEntity> agents = serverProcess.agentList();
- for (AgentEntity agent : agents) {
- Date lastReportedTime = agent.getLastReportedTime();
- long diff = HEALTH_TIME + 1;
- if (lastReportedTime != null) {
- diff = currTime - lastReportedTime.getTime();
- }
- if (diff > HEALTH_TIME) {
- agent.setStatus(AgentStatus.STOP.name());
- agentRepository.save(agent);
- agentCache.putAgent(agent);
- log.warn("agent {} is unhealthly", agent.getHost());
- }
+ private void refreshTaskStatus() {
+ List<ProcessInstanceEntity> processEntities = processInstanceRepository.queryProcessList();
+ for (ProcessInstanceEntity process : processEntities) {
+ processTask.refreshAgentTaskStatus(process.getId());
}
}
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/bean/SpringApplicationContext.java b/manager/dm-server/src/main/java/org/apache/doris/stack/bean/SpringApplicationContext.java
new file mode 100644
index 0000000..bbc7716
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/bean/SpringApplicationContext.java
@@ -0,0 +1,46 @@
+// 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.doris.stack.bean;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.stereotype.Component;
+
+@Component
+public class SpringApplicationContext implements ApplicationContextAware {
+
+ private static ApplicationContext applicationContext;
+
+ public static <T> T getBean(Class<T> requiredType) {
+ return applicationContext.getBean(requiredType);
+ }
+
+ @Override
+ public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+ SpringApplicationContext.applicationContext = applicationContext;
+ }
+
+ /**
+ * Close this application context, destroying all beans in its bean factory.
+ */
+ public void close() {
+ ((AbstractApplicationContext) applicationContext).close();
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/component/AgentComponent.java b/manager/dm-server/src/main/java/org/apache/doris/stack/component/AgentComponent.java
index b6641d1..a1c99e8 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/component/AgentComponent.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/component/AgentComponent.java
@@ -21,7 +21,6 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.doris.stack.constants.AgentStatus;
import org.apache.doris.stack.dao.AgentRepository;
import org.apache.doris.stack.entity.AgentEntity;
-import org.apache.doris.stack.req.AgentRegister;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -42,6 +41,10 @@ public class AgentComponent {
return agentRepository.queryAgentNodes(hosts);
}
+ public List<AgentEntity> queryAgentNodes(int clusterId) {
+ return agentRepository.queryAgentNodesByClusterId(clusterId);
+ }
+
public AgentEntity agentInfo(String host) {
List<AgentEntity> agentEntities = agentRepository.agentInfo(host);
if (agentEntities != null && !agentEntities.isEmpty()) {
@@ -51,19 +54,18 @@ public class AgentComponent {
}
}
- public int refreshAgentStatus(String host, Integer port) {
+ public boolean refreshAgentStatus(String host, Integer port) {
AgentEntity agentInfo = agentInfo(host);
if (agentInfo == null) {
- return 0;
+ return false;
}
- agentInfo.setStatus(AgentStatus.RUNNING.name());
+ agentInfo.setStatus(AgentStatus.RUNNING);
agentInfo.setLastReportedTime(new Date());
agentRepository.save(agentInfo);
- return 1;
+ return true;
}
- public AgentEntity registerAgent(AgentRegister agent) {
- AgentEntity agentEntity = new AgentEntity(agent.getHost(), agent.getPort(), agent.getInstallDir(), AgentStatus.RUNNING.name());
- return agentRepository.save(agentEntity);
+ public AgentEntity saveAgent(AgentEntity agent) {
+ return agentRepository.save(agent);
}
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/component/AgentRoleComponent.java b/manager/dm-server/src/main/java/org/apache/doris/stack/component/AgentRoleComponent.java
index e708ee5..c0fe5c9 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/component/AgentRoleComponent.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/component/AgentRoleComponent.java
@@ -20,7 +20,6 @@ package org.apache.doris.stack.component;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
-import org.apache.doris.manager.common.domain.AgentRoleRegister;
import org.apache.doris.stack.dao.AgentRoleRepository;
import org.apache.doris.stack.entity.AgentRoleEntity;
import org.springframework.beans.factory.annotation.Autowired;
@@ -35,15 +34,19 @@ public class AgentRoleComponent {
@Autowired
private AgentRoleRepository agentRoleRepository;
- public List<AgentRoleEntity> queryAgentRoles() {
+ public List<AgentRoleEntity> queryAllAgentRoles() {
return agentRoleRepository.findAll();
}
- public List<AgentRoleEntity> queryAgentByRole(String role) {
+ public List<AgentRoleEntity> queryAgentRoles(int clusterId) {
+ return agentRoleRepository.queryAgentRoles(clusterId);
+ }
+
+ public List<AgentRoleEntity> queryAgentByRole(String role, int clusterId) {
if (StringUtils.isBlank(role)) {
- return Lists.newArrayList();
+ return agentRoleRepository.queryAgentRoles(clusterId);
}
- return agentRoleRepository.queryAgentByRole(role);
+ return agentRoleRepository.queryAgentByRole(role, clusterId);
}
public List<AgentRoleEntity> queryAgentByHost(String host) {
@@ -53,8 +56,11 @@ public class AgentRoleComponent {
return agentRoleRepository.queryAgentByHost(host);
}
- public AgentRoleEntity registerAgentRole(AgentRoleRegister agentReg) {
- AgentRoleEntity agentRoleEntity = new AgentRoleEntity(agentReg.getHost(), agentReg.getRole(), agentReg.getInstallDir());
+ public AgentRoleEntity queryByHostRole(String host, String role) {
+ return agentRoleRepository.queryByHostRole(host, role);
+ }
+
+ public AgentRoleEntity saveAgentRole(AgentRoleEntity agentRoleEntity) {
return agentRoleRepository.save(agentRoleEntity);
}
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/component/ProcessInstanceComponent.java b/manager/dm-server/src/main/java/org/apache/doris/stack/component/ProcessInstanceComponent.java
new file mode 100644
index 0000000..cee7e8e
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/component/ProcessInstanceComponent.java
@@ -0,0 +1,82 @@
+// 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.doris.stack.component;
+
+import com.google.common.base.Preconditions;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.doris.stack.constants.ProcessTypeEnum;
+import org.apache.doris.stack.dao.ProcessInstanceRepository;
+import org.apache.doris.stack.entity.ProcessInstanceEntity;
+import org.apache.doris.stack.exceptions.ServerException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.Optional;
+
+@Component
+@Slf4j
+public class ProcessInstanceComponent {
+
+ @Autowired
+ private ProcessInstanceRepository processInstanceRepository;
+
+ /**
+ * save process and return id
+ */
+ public int saveProcess(ProcessInstanceEntity processInstance) {
+ checkHasUnfinishProcess(processInstance.getUserId(), -1);
+ return processInstanceRepository.save(processInstance).getId();
+ }
+
+ public void checkHasUnfinishProcess(int userId, int processId) {
+ //query whether there is a process currently being installed
+ ProcessInstanceEntity processEntity = processInstanceRepository.queryProcessByuserId(userId);
+ if (processEntity != null && processEntity.getId() != processId) {
+ throw new ServerException("You already have an installation in the current environment!");
+ }
+ }
+
+ public ProcessInstanceEntity refreshProcess(int processId, ProcessTypeEnum processType) {
+ ProcessInstanceEntity processInstance = queryProcessById(processId);
+ Preconditions.checkArgument(processInstance != null, "install process is not exist");
+ processInstance.setProcessType(processType);
+ return processInstanceRepository.save(processInstance);
+ }
+
+ public ProcessInstanceEntity queryProcessByuserId(int userId) {
+ return processInstanceRepository.queryProcessByuserId(userId);
+ }
+
+ /**
+ * query process by id
+ */
+ public ProcessInstanceEntity queryProcessById(int processId) {
+ Optional<ProcessInstanceEntity> optional = processInstanceRepository.findById(processId);
+ if (optional.isPresent()) {
+ return optional.get();
+ }
+ return null;
+ }
+
+ /**
+ * update process status
+ */
+ public void updateProcess(ProcessInstanceEntity processInstance) {
+ processInstanceRepository.save(processInstance);
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/component/TaskInstanceComponent.java b/manager/dm-server/src/main/java/org/apache/doris/stack/component/TaskInstanceComponent.java
new file mode 100644
index 0000000..7626226
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/component/TaskInstanceComponent.java
@@ -0,0 +1,132 @@
+// 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.doris.stack.component;
+
+import com.alibaba.fastjson.JSON;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.doris.manager.common.domain.CommandResult;
+import org.apache.doris.manager.common.domain.RResult;
+import org.apache.doris.manager.common.domain.TaskResult;
+import org.apache.doris.stack.constants.ExecutionStatus;
+import org.apache.doris.stack.constants.Flag;
+import org.apache.doris.stack.constants.ProcessTypeEnum;
+import org.apache.doris.stack.constants.TaskTypeEnum;
+import org.apache.doris.stack.dao.TaskInstanceRepository;
+import org.apache.doris.stack.entity.TaskInstanceEntity;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.Date;
+import java.util.List;
+import java.util.Optional;
+
+@Component
+@Slf4j
+public class TaskInstanceComponent {
+
+ @Autowired
+ private TaskInstanceRepository taskInstanceRepository;
+
+ /**
+ * The same host, the same tasktype, can only have one in a install process
+ */
+ public boolean checkTaskRunning(int processId, String host, ProcessTypeEnum processType, TaskTypeEnum taskType) {
+ TaskInstanceEntity taskEntity = taskInstanceRepository.queryTask(processId, host, processType, taskType);
+ if (taskEntity == null) {
+ return false;
+ } else if (taskEntity.getStatus().typeIsRunning()) {
+ log.warn("task {} already running in host {}", taskType.name(), host);
+ return true;
+ } else {
+ taskInstanceRepository.deleteById(taskEntity.getId());
+ return false;
+ }
+ }
+
+ /**
+ * If the same task is already running on the host, skip it
+ */
+ public TaskInstanceEntity saveTask(int processId, String host, ProcessTypeEnum processType, TaskTypeEnum taskType, ExecutionStatus status) {
+ if (!checkTaskRunning(processId, host, processType, taskType)) {
+ return taskInstanceRepository.save(new TaskInstanceEntity(processId, host, processType, taskType, status));
+ } else {
+ return null;
+ }
+ }
+
+ /**
+ * refresh task status
+ */
+ public TaskInstanceEntity refreshTask(TaskInstanceEntity taskInstance, RResult result) {
+ if (result == null || result.getData() == null) {
+ taskInstance.setStatus(ExecutionStatus.FAILURE);
+ } else {
+ CommandResult commandResult = JSON.parseObject(JSON.toJSONString(result.getData()), CommandResult.class);
+ if (commandResult == null) {
+ taskInstance.setStatus(ExecutionStatus.FAILURE);
+ } else {
+ TaskResult taskResult = commandResult.getTaskResult();
+ if (taskResult == null) {
+ taskInstance.setStatus(ExecutionStatus.FAILURE);
+ } else {
+ taskInstance.setExecutorId(taskResult.getTaskId());
+ if (taskResult.getTaskState().typeIsRunning()) {
+ taskInstance.setStatus(ExecutionStatus.RUNNING);
+ } else if (taskResult.getRetCode() != null && taskResult.getRetCode() == 0) {
+ taskInstance.setStatus(ExecutionStatus.SUCCESS);
+ taskInstance.setFinish(Flag.YES);
+ taskInstance.setEndTime(new Date());
+ } else {
+ taskInstance.setStatus(ExecutionStatus.FAILURE);
+ taskInstance.setEndTime(new Date());
+ }
+ }
+ }
+ }
+ return taskInstanceRepository.save(taskInstance);
+ }
+
+ /**
+ * Check whether the parent task is successful
+ */
+ public boolean checkParentTaskSuccess(int processId, ProcessTypeEnum processType) {
+ ProcessTypeEnum parent = ProcessTypeEnum.findParent(processType);
+ if (parent == null) {
+ return true;
+ }
+ List<TaskInstanceEntity> taskInstanceEntities = taskInstanceRepository.queryTasksByProcessStep(processId, parent);
+ for (TaskInstanceEntity task : taskInstanceEntities) {
+ if (Flag.NO.equals(task.getFinish())) {
+ log.info("task {} is unsuccess", task.getTaskType());
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * query task by id
+ */
+ public TaskInstanceEntity queryTaskById(int taskId) {
+ Optional<TaskInstanceEntity> optional = taskInstanceRepository.findById(taskId);
+ if (optional.isPresent()) {
+ return optional.get();
+ }
+ return null;
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/constants/AgentStatus.java b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/AgentStatus.java
index ab8948d..4c6a354 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/constants/AgentStatus.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/AgentStatus.java
@@ -18,6 +18,8 @@
package org.apache.doris.stack.constants;
public enum AgentStatus {
+ INIT,
+ REGISTER,
RUNNING,
STOP;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/constants/CmdTypeEnum.java b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/CmdTypeEnum.java
index c75b18d..5f61bef 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/constants/CmdTypeEnum.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/CmdTypeEnum.java
@@ -19,8 +19,7 @@ package org.apache.doris.stack.constants;
public enum CmdTypeEnum {
START,
- STOP,
- RESTART;
+ STOP;
public static CmdTypeEnum findByName(String name) {
for (CmdTypeEnum type : CmdTypeEnum.values()) {
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/constants/Constants.java b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/Constants.java
index 6584f95..fcfb390 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/constants/Constants.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/Constants.java
@@ -19,11 +19,12 @@ package org.apache.doris.stack.constants;
public class Constants {
- public static final String KEY_DORIS_AGENT_START_SCRIPT = "agent/bin/agent_start.sh";
+ public static final String KEY_DORIS_AGENT_START_SCRIPT = "bin/agent_start.sh";
public static final String KEY_FE_QUERY_PORT = "query_port";
public static final String KEY_BE_HEARTBEAT_PORT = "heartbeat_service_port";
public static final String DORIS_DEFAULT_QUERY_USER = "root";
public static final String DORIS_DEFAULT_QUERY_PASSWORD = "";
public static final Integer DORIS_DEFAULT_FE_QUERY_PORT = 9030;
+ public static final Integer DORIS_DEFAULT_FE_HTTP_PORT = 8030;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/constants/ExecutionStatus.java b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/ExecutionStatus.java
new file mode 100644
index 0000000..266b2dd
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/ExecutionStatus.java
@@ -0,0 +1,88 @@
+// 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.doris.stack.constants;
+
+/**
+ * running status for workflow and task nodes
+ */
+public enum ExecutionStatus {
+
+ /**
+ * status：
+ * 0 submit success
+ * 1 running
+ * 2 failure
+ * 3 success
+ */
+ SUBMITTED(0, "submit success"),
+ RUNNING(1, "running"),
+ FAILURE(2, "failure"),
+ SUCCESS(3, "success");
+
+ private final int code;
+ private final String desc;
+
+ ExecutionStatus(int code, String desc) {
+ this.code = code;
+ this.desc = desc;
+ }
+
+ /**
+ * status is success
+ *
+ * @return status
+ */
+ public boolean typeIsSuccess() {
+ return this == SUCCESS;
+ }
+
+ /**
+ * status is failure
+ *
+ * @return status
+ */
+ public boolean typeIsFailure() {
+ return this == FAILURE;
+ }
+
+ /**
+ * status is finished
+ *
+ * @return status
+ */
+ public boolean typeIsFinished() {
+ return typeIsSuccess() || typeIsFailure();
+ }
+
+ /**
+ * status is running
+ *
+ * @return status
+ */
+ public boolean typeIsRunning() {
+ return this == RUNNING || this == SUBMITTED;
+ }
+
+ public int getCode() {
+ return code;
+ }
+
+ public String getDesc() {
+ return desc;
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/TaskInfoReq.java b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/Flag.java
similarity index 68%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/req/TaskInfoReq.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/constants/Flag.java
index 06c6fc9..c20957a 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/TaskInfoReq.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/Flag.java
@@ -15,27 +15,29 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.req;
-
-public class TaskInfoReq {
-
- private String host;
-
- private String taskId;
-
- public String getHost() {
- return host;
- }
-
- public void setHost(String host) {
- this.host = host;
+package org.apache.doris.stack.constants;
+
+public enum Flag {
+ /**
+ * 0 no
+ * 1 yes
+ */
+ NO(0, "no"),
+ YES(1, "yes");
+
+ private final int code;
+ private final String desc;
+
+ Flag(int code, String desc) {
+ this.code = code;
+ this.desc = desc;
}
- public String getTaskId() {
- return taskId;
+ public int getCode() {
+ return code;
}
- public void setTaskId(String taskId) {
- this.taskId = taskId;
+ public String getDesc() {
+ return desc;
}
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/constants/ProcessTypeEnum.java b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/ProcessTypeEnum.java
new file mode 100644
index 0000000..069a3e1
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/ProcessTypeEnum.java
@@ -0,0 +1,73 @@
+// 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.doris.stack.constants;
+
+/**
+ * process step type
+ **/
+public enum ProcessTypeEnum {
+ INSTALL_AGENT(0, "install agent"),
+ INSTALL_SERVICE(1, "install service"),
+ DEPLOY_CONFIG(2, "deploy config"),
+ START_SERVICE(3, "start service"),
+ BUILD_CLUSTER(4, "build cluster");
+
+ private int code;
+ private String desc;
+
+ ProcessTypeEnum(int code, String desc) {
+ this.code = code;
+ this.desc = desc;
+ }
+
+ public static ProcessTypeEnum findByName(String name) {
+ for (ProcessTypeEnum type : ProcessTypeEnum.values()) {
+ if (type.name().equals(name)) {
+ return type;
+ }
+ }
+ return null;
+ }
+
+ public static ProcessTypeEnum findByCode(int code) {
+ for (ProcessTypeEnum type : ProcessTypeEnum.values()) {
+ if (type.code == code) {
+ return type;
+ }
+ }
+ return null;
+ }
+
+ public static ProcessTypeEnum findParent(ProcessTypeEnum processType) {
+ if (processType != null) {
+ int parentCode = processType.getCode() - 1;
+ return findByCode(parentCode);
+ } else {
+ return processType;
+ }
+ }
+
+ public int getCode() {
+ return code;
+ }
+
+ public String getDesc() {
+ return desc;
+ }
+
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/constants/CmdTypeEnum.java b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/TaskTypeEnum.java
similarity index 71%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/constants/CmdTypeEnum.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/constants/TaskTypeEnum.java
index c75b18d..2a86e1b 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/constants/CmdTypeEnum.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/constants/TaskTypeEnum.java
@@ -17,17 +17,26 @@
package org.apache.doris.stack.constants;
-public enum CmdTypeEnum {
- START,
- STOP,
- RESTART;
+/**
+ * task type
+ **/
+public enum TaskTypeEnum {
+ INSTALL_AGENT,
+ INSTALL_FE,
+ INSTALL_BE,
+ DEPLOY_FE_CONFIG,
+ DEPLOY_BE_CONFIG,
+ START_FE,
+ START_BE,
+ STOP_FE,
+ STOP_BE,
+ JOIN_BE;
- public static CmdTypeEnum findByName(String name) {
- for (CmdTypeEnum type : CmdTypeEnum.values()) {
- if (type.name().equals(name)) {
- return type;
- }
- }
- return null;
+ /**
+ * agent side task
+ */
+ public boolean agentTask() {
+ return this != INSTALL_AGENT && this != JOIN_BE;
}
+
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/controller/AgentController.java b/manager/dm-server/src/main/java/org/apache/doris/stack/controller/AgentController.java
index 20de71f..7b65dbb 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/controller/AgentController.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/controller/AgentController.java
@@ -18,84 +18,105 @@
package org.apache.doris.stack.controller;
import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
import org.apache.doris.manager.common.domain.AgentRoleRegister;
import org.apache.doris.manager.common.domain.RResult;
-import org.apache.doris.stack.req.DorisExecReq;
-import org.apache.doris.stack.req.DorisInstallReq;
-import org.apache.doris.stack.req.TaskInfoReq;
-import org.apache.doris.stack.req.TaskLogReq;
-import org.apache.doris.stack.service.ServerAgent;
+import org.apache.doris.stack.model.request.BeJoinReq;
+import org.apache.doris.stack.model.request.DeployConfigReq;
+import org.apache.doris.stack.model.request.DorisExecReq;
+import org.apache.doris.stack.model.request.DorisInstallReq;
+import org.apache.doris.stack.model.request.DorisStartReq;
+import org.apache.doris.stack.service.AgentProcess;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
import java.util.List;
@Api(tags = "Agent API")
@RestController
-@RequestMapping("/agent")
+@RequestMapping("/api/agent")
public class AgentController {
@Autowired
- private ServerAgent serverAgent;
+ private AgentProcess agentProcess;
/**
- * install doris
+ * install doris service
*/
- @RequestMapping(value = "/installDoris", method = RequestMethod.POST)
- public RResult install(@RequestBody DorisInstallReq installReq) {
- return RResult.success(serverAgent.install(installReq));
+ @ApiOperation(value = "install doris service")
+ @RequestMapping(value = "/installService", method = RequestMethod.POST)
+ public RResult installService(HttpServletRequest request, HttpServletResponse response,
+ @RequestBody DorisInstallReq installReq) throws Exception {
+ agentProcess.installService(request, response, installReq);
+ return RResult.success();
}
/**
- * request agent:start stop fe/be
+ * deploy config
*/
- @RequestMapping(value = "/execute", method = RequestMethod.POST)
- public RResult execute(@RequestBody DorisExecReq dorisExec) {
- return RResult.success(serverAgent.execute(dorisExec));
+ @ApiOperation(value = "deploy doris config")
+ @RequestMapping(value = "/deployConfig", method = RequestMethod.POST)
+ public RResult deployConfig(HttpServletRequest request, HttpServletResponse response,
+ @RequestBody DeployConfigReq deployConfigReq) throws Exception {
+ agentProcess.deployConfig(request, response, deployConfigReq);
+ return RResult.success();
}
/**
- * request task detail
+ * Start the service when installing the cluster
*/
- @RequestMapping(value = "/task", method = RequestMethod.POST)
- public RResult taskInfo(@RequestBody TaskInfoReq taskInfo) {
- return serverAgent.taskInfo(taskInfo);
+ @ApiOperation(value = "Start the service when installing the cluster")
+ @RequestMapping(value = "/startService", method = RequestMethod.POST)
+ public RResult startService(HttpServletRequest request, HttpServletResponse response,
+ @RequestBody DorisStartReq dorisStart) throws Exception {
+ agentProcess.startService(request, response, dorisStart);
+ return RResult.success();
}
/**
- * request task stdout log
+ * join be to cluster
*/
- @RequestMapping(value = "/stdlog", method = RequestMethod.POST)
- public RResult taskStdlog(@RequestBody TaskLogReq taskInfo) {
- return serverAgent.taskStdlog(taskInfo);
+ @ApiOperation(value = "join be to cluster")
+ @RequestMapping(value = "/joinBe", method = RequestMethod.POST)
+ public RResult joinBe(HttpServletRequest request, HttpServletResponse response,
+ @RequestBody BeJoinReq beJoinReq) throws Exception {
+ agentProcess.joinBe(request, response, beJoinReq);
+ return RResult.success();
}
/**
- * request task error log
+ * register role service (be/fe)
*/
- @RequestMapping(value = "/errlog", method = RequestMethod.POST)
- public RResult taskErrlog(@RequestBody TaskLogReq taskInfo) {
- return serverAgent.taskErrlog(taskInfo);
+ @ApiOperation(value = "register role service (be/fe)")
+ @RequestMapping(value = "/register", method = RequestMethod.POST)
+ public RResult register(@RequestBody AgentRoleRegister agentReg) {
+ boolean register = agentProcess.register(agentReg);
+ return RResult.success(register);
}
/**
- * join be to cluster
+ * execute command: START/STOP
*/
- @RequestMapping(value = "/joinBe", method = RequestMethod.POST)
- public RResult joinBe(@RequestBody List<String> hosts) {
- serverAgent.joinBe(hosts);
- return RResult.success();
+ @ApiOperation(value = "execute command:START/STOP")
+ @RequestMapping(value = "/execute", method = RequestMethod.POST)
+ public RResult execute(@RequestBody DorisExecReq dorisExec) {
+ List<Integer> taskIds = agentProcess.execute(dorisExec);
+ return RResult.success(taskIds);
}
/**
- * register role service (be/fe)
+ * query log
+ * type:fe.log/fe.out/be.log/be.out
*/
- @RequestMapping(value = "/register", method = RequestMethod.POST)
- public RResult register(@RequestBody AgentRoleRegister agentReg) {
- boolean register = serverAgent.register(agentReg);
- return RResult.success(register);
+ @ApiOperation(value = "query service log,type:fe.log|fe.out|be.log|be.out")
+ @RequestMapping(value = "/log", method = RequestMethod.GET)
+ public RResult log(@RequestParam String host, @RequestParam String type) {
+ return RResult.success(agentProcess.log(host, type));
}
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/controller/ProcessTaskController.java b/manager/dm-server/src/main/java/org/apache/doris/stack/controller/ProcessTaskController.java
new file mode 100644
index 0000000..5c58c8f
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/controller/ProcessTaskController.java
@@ -0,0 +1,116 @@
+// 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.doris.stack.controller;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import org.apache.doris.manager.common.domain.RResult;
+import org.apache.doris.stack.entity.ProcessInstanceEntity;
+import org.apache.doris.stack.entity.TaskInstanceEntity;
+import org.apache.doris.stack.service.ProcessTask;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.util.List;
+
+/**
+ * process task controller
+ **/
+@Api(tags = "Process Task API")
+@RestController
+@RequestMapping("/api/process")
+public class ProcessTaskController {
+
+ @Autowired
+ private ProcessTask processTask;
+
+ /**
+ * query user history installation progress
+ */
+ @ApiOperation(value = "query user history installation progress")
+ @RequestMapping(value = "/historyProgress", method = RequestMethod.GET)
+ public RResult historyProgress(HttpServletRequest request, HttpServletResponse response) throws Exception {
+ ProcessInstanceEntity process = processTask.historyProgress(request, response);
+ return RResult.success(process);
+ }
+
+ /**
+ * Installation progress of the current process
+ */
+ @ApiOperation(value = "query user history installation progress")
+ @RequestMapping(value = "/{processId}/progress", method = RequestMethod.GET)
+ public RResult processProgress(HttpServletRequest request, HttpServletResponse response,
+ @PathVariable(value = "processId") int processId) {
+ List<TaskInstanceEntity> tasks = processTask.processProgress(request, response, processId);
+ return RResult.success(tasks);
+ }
+
+ /**
+ * Query the installation status of tasks in the current installation process
+ */
+ @ApiOperation(value = "Query the installation status of tasks in the current installation process")
+ @RequestMapping(value = "/{processId}/currentTasks", method = RequestMethod.GET)
+ public RResult taskProgress(HttpServletRequest request, HttpServletResponse response,
+ @PathVariable(value = "processId") int processId) {
+ List<TaskInstanceEntity> tasks = processTask.taskProgress(request, response, processId);
+ return RResult.success(tasks);
+ }
+
+ /**
+ * After the installation is complete, call the interface
+ */
+ @ApiOperation(value = "After the installation is complete, call the interface")
+ @RequestMapping(value = "/installComplete/{processId}", method = RequestMethod.POST)
+ public RResult installComplete(HttpServletRequest request, HttpServletResponse response,
+ @PathVariable(value = "processId") int processId) throws Exception {
+ processTask.installComplete(request, response, processId);
+ return RResult.success();
+ }
+
+ /**
+ * Skip task when task fails
+ */
+ @ApiOperation(value = "Skip task when task fails")
+ @RequestMapping(value = "/task/skip/{taskId}", method = RequestMethod.POST)
+ public RResult skipTask(@PathVariable(value = "taskId") int taskId) throws Exception {
+ processTask.skipTask(taskId);
+ return RResult.success();
+ }
+
+ /**
+ * request task info
+ */
+ @RequestMapping(value = "/task/info/{taskId}", method = RequestMethod.GET)
+ public RResult taskInfo(@PathVariable int taskId) {
+ return RResult.success(processTask.taskInfo(taskId));
+ }
+
+ /**
+ * request task stdout log
+ */
+ @RequestMapping(value = "/task/log/{taskId}", method = RequestMethod.GET)
+ public RResult taskLog(@PathVariable int taskId) {
+ return RResult.success(processTask.taskLog(taskId));
+ }
+
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/controller/ServerController.java b/manager/dm-server/src/main/java/org/apache/doris/stack/controller/ServerController.java
index 6419995..7183f93 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/controller/ServerController.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/controller/ServerController.java
@@ -18,11 +18,12 @@
package org.apache.doris.stack.controller;
import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.doris.manager.common.domain.RResult;
-import org.apache.doris.stack.req.AgentCommon;
-import org.apache.doris.stack.req.AgentRegister;
-import org.apache.doris.stack.req.SshInfo;
+import org.apache.doris.stack.model.request.AgentCommon;
+import org.apache.doris.stack.model.request.AgentInstallReq;
+import org.apache.doris.stack.model.request.AgentRegister;
import org.apache.doris.stack.service.ServerProcess;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestBody;
@@ -31,9 +32,12 @@ import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
@Api(tags = "Server API")
@RestController
-@RequestMapping("/server")
+@RequestMapping("/api/server")
@Slf4j
public class ServerController {
@@ -41,26 +45,38 @@ public class ServerController {
private ServerProcess serverProcess;
/**
- * install and start agent
+ * install and start agent and return processId
*/
+ @ApiOperation(value = "install agent and return processId")
@RequestMapping(value = "/installAgent", method = RequestMethod.POST)
- public RResult installAgent(@RequestBody SshInfo sshInfo) {
- serverProcess.initAgent(sshInfo);
- serverProcess.startAgent(sshInfo);
- return RResult.success();
+ public RResult installAgent(HttpServletRequest request, HttpServletResponse response,
+ @RequestBody AgentInstallReq agentInstallReq) throws Exception {
+ int processId = serverProcess.installAgent(request, response, agentInstallReq);
+ return RResult.success(processId);
}
/**
* agent info list
*/
- @RequestMapping(value = "/agentList", method = RequestMethod.POST)
- public RResult agentList() {
- return RResult.success(serverProcess.agentList());
+ @ApiOperation(value = "agent info list")
+ @RequestMapping(value = "/agentList", method = RequestMethod.GET)
+ public RResult agentList(@RequestParam int clusterId) {
+ return RResult.success(serverProcess.agentList(clusterId));
+ }
+
+ /**
+ * role info list
+ */
+ @ApiOperation(value = "agent role info list")
+ @RequestMapping(value = "/roleList", method = RequestMethod.GET)
+ public RResult roleList(@RequestParam int clusterId) {
+ return RResult.success(serverProcess.roleList(clusterId));
}
/**
* agent role info
*/
+ @ApiOperation(value = "agent role info")
@RequestMapping(value = "/agentRole", method = RequestMethod.GET)
public RResult agentRole(@RequestParam String host) {
return RResult.success(serverProcess.agentRole(host));
@@ -69,6 +85,7 @@ public class ServerController {
/**
* heatbeat report api
*/
+ @ApiOperation(value = "agent heartbeat report")
@RequestMapping(value = "/heartbeat", method = RequestMethod.POST)
public RResult heartbeat(@RequestBody AgentCommon agent) {
log.info("{} heartbeat.", agent.getHost());
@@ -79,6 +96,7 @@ public class ServerController {
/**
* register agent
*/
+ @ApiOperation(value = "agent register")
@RequestMapping(value = "/register", method = RequestMethod.POST)
public RResult register(@RequestBody AgentRegister agent) {
log.info("{} register.", agent.getHost());
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRepository.java b/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRepository.java
index d24d786..f080dd6 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRepository.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRepository.java
@@ -24,13 +24,6 @@ import org.springframework.data.repository.query.Param;
import java.util.List;
-/**
- * Copyright (C) 2020 Baidu, Inc. All Rights Reserved.
- *
- * @Author: songchuanyuan@baidu.com
- * @Description：
- * @Date: 2021/8/18
- */
public interface AgentRepository extends JpaRepository<AgentEntity, Integer> {
@Query("select f from AgentEntity f where f.host in (:hosts)")
@@ -39,4 +32,6 @@ public interface AgentRepository extends JpaRepository<AgentEntity, Integer> {
@Query("select f from AgentEntity f where f.host = :host ")
List<AgentEntity> agentInfo(@Param("host") String host);
+ @Query("select f from AgentEntity f where f.clusterId = :clusterId")
+ List<AgentEntity> queryAgentNodesByClusterId(@Param("clusterId") int clusterId);
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRoleRepository.java b/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRoleRepository.java
index 47b7f52..c36dc86 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRoleRepository.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRoleRepository.java
@@ -24,19 +24,17 @@ import org.springframework.data.repository.query.Param;
import java.util.List;
-/**
- * Copyright (C) 2020 Baidu, Inc. All Rights Reserved.
- *
- * @Author: songchuanyuan@baidu.com
- * @Description：
- * @Date: 2021/8/18
- */
-public interface AgentRoleRepository extends JpaRepository<AgentRoleEntity, String> {
-
- @Query("select f from AgentRoleEntity f where f.role = :role")
- List<AgentRoleEntity> queryAgentByRole(@Param("role") String role);
+public interface AgentRoleRepository extends JpaRepository<AgentRoleEntity, Integer> {
+
+ @Query("select f from AgentRoleEntity f where f.role = :role and f.clusterId = :clusterId")
+ List<AgentRoleEntity> queryAgentByRole(@Param("role") String role, @Param("clusterId") int clusterId);
@Query("select f from AgentRoleEntity f where f.host = :host")
List<AgentRoleEntity> queryAgentByHost(@Param("host") String host);
+ @Query("select f from AgentRoleEntity f where f.host = :host and f.role = :role")
+ AgentRoleEntity queryByHostRole(@Param("host") String host, @Param("role") String role);
+
+ @Query("select f from AgentRoleEntity f where f.clusterId = :clusterId")
+ List<AgentRoleEntity> queryAgentRoles(@Param("clusterId") int clusterId);
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRepository.java b/manager/dm-server/src/main/java/org/apache/doris/stack/dao/ProcessInstanceRepository.java
similarity index 65%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRepository.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/dao/ProcessInstanceRepository.java
index d24d786..3c8b35c 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRepository.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/dao/ProcessInstanceRepository.java
@@ -17,26 +17,18 @@
package org.apache.doris.stack.dao;
-import org.apache.doris.stack.entity.AgentEntity;
+import org.apache.doris.stack.entity.ProcessInstanceEntity;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;
import java.util.List;
-/**
- * Copyright (C) 2020 Baidu, Inc. All Rights Reserved.
- *
- * @Author: songchuanyuan@baidu.com
- * @Description：
- * @Date: 2021/8/18
- */
-public interface AgentRepository extends JpaRepository<AgentEntity, Integer> {
+public interface ProcessInstanceRepository extends JpaRepository<ProcessInstanceEntity, Integer> {
- @Query("select f from AgentEntity f where f.host in (:hosts)")
- List<AgentEntity> queryAgentNodes(@Param("hosts") List<String> hosts);
-
- @Query("select f from AgentEntity f where f.host = :host ")
- List<AgentEntity> agentInfo(@Param("host") String host);
+ @Query("select f from ProcessInstanceEntity f where f.userId = :userId and finish = 0")
+ ProcessInstanceEntity queryProcessByuserId(@Param("userId") int userId);
+ @Query("select f from ProcessInstanceEntity f where finish = 0")
+ List<ProcessInstanceEntity> queryProcessList();
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRepository.java b/manager/dm-server/src/main/java/org/apache/doris/stack/dao/TaskInstanceRepository.java
similarity index 50%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRepository.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/dao/TaskInstanceRepository.java
index d24d786..3b76def 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/dao/AgentRepository.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/dao/TaskInstanceRepository.java
@@ -17,26 +17,23 @@
package org.apache.doris.stack.dao;
-import org.apache.doris.stack.entity.AgentEntity;
+import org.apache.doris.stack.constants.ProcessTypeEnum;
+import org.apache.doris.stack.constants.TaskTypeEnum;
+import org.apache.doris.stack.entity.TaskInstanceEntity;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;
import java.util.List;
-/**
- * Copyright (C) 2020 Baidu, Inc. All Rights Reserved.
- *
- * @Author: songchuanyuan@baidu.com
- * @Description：
- * @Date: 2021/8/18
- */
-public interface AgentRepository extends JpaRepository<AgentEntity, Integer> {
+public interface TaskInstanceRepository extends JpaRepository<TaskInstanceEntity, Integer> {
- @Query("select f from AgentEntity f where f.host in (:hosts)")
- List<AgentEntity> queryAgentNodes(@Param("hosts") List<String> hosts);
+ @Query("select f from TaskInstanceEntity f where f.processId = :processId")
+ List<TaskInstanceEntity> queryTasksByProcessId(@Param("processId") int processId);
- @Query("select f from AgentEntity f where f.host = :host ")
- List<AgentEntity> agentInfo(@Param("host") String host);
+ @Query("select f from TaskInstanceEntity f where f.processId = :processId and f.processType = :processType")
+ List<TaskInstanceEntity> queryTasksByProcessStep(@Param("processId") int processId, @Param("processType") ProcessTypeEnum processType);
+ @Query("select f from TaskInstanceEntity f where f.processId = :processId and host = :host and processType = :processType and taskType = :taskType")
+ TaskInstanceEntity queryTask(@Param("processId") int processId, @Param("host") String host, @Param("processType") ProcessTypeEnum processType, @Param("taskType") TaskTypeEnum taskType);
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentEntity.java b/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentEntity.java
index e6dc50d..8a31567 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentEntity.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentEntity.java
@@ -20,9 +20,12 @@ package org.apache.doris.stack.entity;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
+import org.apache.doris.stack.constants.AgentStatus;
import javax.persistence.Column;
import javax.persistence.Entity;
+import javax.persistence.EnumType;
+import javax.persistence.Enumerated;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
@@ -33,7 +36,7 @@ import java.util.Date;
* agent entity
**/
@Entity
-@Table(name = "t_agent")
+@Table(name = "agent")
@Data
@NoArgsConstructor
@AllArgsConstructor
@@ -43,14 +46,20 @@ public class AgentEntity {
@GeneratedValue(strategy = GenerationType.IDENTITY)
private int id;
+ @Column(name = "host")
private String host;
- private Integer port;
+ @Column(name = "port")
+ private int port;
+
+ @Column(name = "cluster_id")
+ private int clusterId;
@Column(name = "install_dir")
private String installDir;
- private String status;
+ @Enumerated(EnumType.STRING)
+ private AgentStatus status;
@Column(name = "register_time")
private Date registerTime;
@@ -58,11 +67,18 @@ public class AgentEntity {
@Column(name = "last_reported_time")
private Date lastReportedTime;
- public AgentEntity(String host, Integer port, String installDir, String status) {
+ public AgentEntity(String host, int port, String installDir, AgentStatus status) {
this.host = host;
this.port = port;
this.installDir = installDir;
this.status = status;
this.registerTime = new Date();
}
+
+ public AgentEntity(String host, String installDir, AgentStatus status, int clusterId) {
+ this.host = host;
+ this.installDir = installDir;
+ this.status = status;
+ this.clusterId = clusterId;
+ }
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java b/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
index 989e26f..37af083 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
@@ -20,26 +20,50 @@ package org.apache.doris.stack.entity;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
+import org.apache.doris.stack.constants.Flag;
import javax.persistence.Column;
import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.Table;
@Entity
-@Table(name = "t_agent_role")
+@Table(name = "agent_role")
@Data
@NoArgsConstructor
@AllArgsConstructor
public class AgentRoleEntity {
@Id
+ @GeneratedValue(strategy = GenerationType.IDENTITY)
+ private int id;
+
@Column(length = 150)
private String host;
+ @Column(name = "cluster_id")
+ private int clusterId;
+
+ @Column(name = "role")
private String role;
+ //FOLLOWER / OBSERVER
+ @Column(name = "fe_node_type")
+ private String feNodeType;
+
@Column(name = "install_dir")
private String installDir;
+ @Column(name = "register")
+ private Flag register;
+
+ public AgentRoleEntity(String host, String role, String feNodeType, String installDir, Flag register) {
+ this.host = host;
+ this.role = role;
+ this.feNodeType = feNodeType;
+ this.installDir = installDir;
+ this.register = register;
+ }
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentEntity.java b/manager/dm-server/src/main/java/org/apache/doris/stack/entity/ProcessInstanceEntity.java
similarity index 51%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentEntity.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/entity/ProcessInstanceEntity.java
index e6dc50d..07d384f 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentEntity.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/entity/ProcessInstanceEntity.java
@@ -20,9 +20,13 @@ package org.apache.doris.stack.entity;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
+import org.apache.doris.stack.constants.Flag;
+import org.apache.doris.stack.constants.ProcessTypeEnum;
import javax.persistence.Column;
import javax.persistence.Entity;
+import javax.persistence.EnumType;
+import javax.persistence.Enumerated;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
@@ -30,39 +34,53 @@ import javax.persistence.Table;
import java.util.Date;
/**
- * agent entity
+ * process instance entity
**/
@Entity
-@Table(name = "t_agent")
+@Table(name = "process_instance")
@Data
@NoArgsConstructor
@AllArgsConstructor
-public class AgentEntity {
+public class ProcessInstanceEntity {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private int id;
- private String host;
+ @Column(name = "cluster_id", nullable = false)
+ private int clusterId;
- private Integer port;
+ @Column(name = "user_id", nullable = false)
+ private int userId;
- @Column(name = "install_dir")
- private String installDir;
+ @Enumerated(EnumType.STRING)
+ @Column(name = "process_type", nullable = false)
+ private ProcessTypeEnum processType;
+
+ @Column(name = "create_time", nullable = false)
+ private Date createTime;
- private String status;
+ @Column(name = "update_time", nullable = false)
+ private Date updateTime;
- @Column(name = "register_time")
- private Date registerTime;
+ @Enumerated(EnumType.ORDINAL)
+ @Column(name = "finish", nullable = false)
+ private Flag finish;
- @Column(name = "last_reported_time")
- private Date lastReportedTime;
+ @Column(name = "package_url", length = 1024)
+ private String packageUrl;
+
+ @Column(name = "install_dir", length = 1024)
+ private String installDir;
- public AgentEntity(String host, Integer port, String installDir, String status) {
- this.host = host;
- this.port = port;
+ public ProcessInstanceEntity(int clusterId, int userId, ProcessTypeEnum processType, String packageUrl, String installDir) {
+ this.clusterId = clusterId;
+ this.userId = userId;
+ this.processType = processType;
this.installDir = installDir;
- this.status = status;
- this.registerTime = new Date();
+ this.packageUrl = packageUrl;
+ this.createTime = new Date();
+ this.updateTime = new Date();
+ this.finish = Flag.NO;
}
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/TaskInstanceEntity.java b/manager/dm-server/src/main/java/org/apache/doris/stack/entity/TaskInstanceEntity.java
new file mode 100644
index 0000000..0a547a4
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/entity/TaskInstanceEntity.java
@@ -0,0 +1,109 @@
+// 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.doris.stack.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.doris.stack.constants.ExecutionStatus;
+import org.apache.doris.stack.constants.Flag;
+import org.apache.doris.stack.constants.ProcessTypeEnum;
+import org.apache.doris.stack.constants.TaskTypeEnum;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.EnumType;
+import javax.persistence.Enumerated;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
+import javax.persistence.Id;
+import javax.persistence.Table;
+import java.util.Date;
+
+/**
+ * task instance entity
+ **/
+@Entity
+@Table(name = "task_instance")
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class TaskInstanceEntity {
+
+ @Id
+ @GeneratedValue(strategy = GenerationType.IDENTITY)
+ private int id;
+
+ @Column(name = "process_id")
+ private int processId;
+
+ @Column(name = "host", nullable = false)
+ private String host;
+
+ @Enumerated(EnumType.STRING)
+ @Column(name = "process_type")
+ private ProcessTypeEnum processType;
+
+ @Enumerated(EnumType.STRING)
+ @Column(name = "task_type", nullable = false)
+ private TaskTypeEnum taskType;
+
+ @Enumerated(EnumType.STRING)
+ @Column(name = "status", nullable = false)
+ private ExecutionStatus status;
+
+ @Column(name = "start_time", nullable = false)
+ private Date startTime;
+
+ @Column(name = "end_time")
+ private Date endTime;
+
+ @Column(name = "executor_id")
+ private String executorId;
+
+ @Column(name = "result", length = 1024)
+ private String result;
+
+ @Enumerated(EnumType.ORDINAL)
+ @Column(name = "finish", nullable = false)
+ private Flag finish;
+
+ public TaskInstanceEntity(int processId, String host, ProcessTypeEnum processType, TaskTypeEnum taskType, ExecutionStatus status) {
+ this.processId = processId;
+ this.host = host;
+ this.processType = processType;
+ this.taskType = taskType;
+ this.status = status;
+ this.startTime = new Date();
+ this.finish = Flag.NO;
+ }
+
+ public TaskInstanceEntity(int processId, String host, ProcessTypeEnum processType) {
+ this.processId = processId;
+ this.host = host;
+ this.processType = processType;
+ this.startTime = new Date();
+ this.finish = Flag.NO;
+ }
+
+ public TaskInstanceEntity(String host) {
+ this.host = host;
+ this.startTime = new Date();
+ this.finish = Flag.NO;
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/AgentRegister.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/AgentInstall.java
similarity index 63%
rename from manager/dm-server/src/main/java/org/apache/doris/stack/req/AgentRegister.java
rename to manager/dm-server/src/main/java/org/apache/doris/stack/model/AgentInstall.java
index 1063d94..66413df 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/AgentRegister.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/AgentInstall.java
@@ -15,37 +15,33 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.req;
+package org.apache.doris.stack.model;
-public class AgentRegister {
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.doris.stack.model.request.AgentInstallReq;
+import org.springframework.beans.BeanUtils;
- private String host;
-
- private Integer port;
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class AgentInstall {
- private String installDir;
+ private String host;
- public String getHost() {
- return host;
- }
+ private String user;
- public void setHost(String host) {
- this.host = host;
- }
+ private int sshPort;
- public Integer getPort() {
- return port;
- }
+ private String sshKey;
- public void setPort(Integer port) {
- this.port = port;
- }
+ private String installDir;
- public String getInstallDir() {
- return installDir;
- }
+ private int clusterId;
- public void setInstallDir(String installDir) {
- this.installDir = installDir;
+ public AgentInstall(String host, AgentInstallReq installReq) {
+ this.host = host;
+ BeanUtils.copyProperties(installReq, this);
}
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/BeJoin.java
similarity index 73%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/model/BeJoin.java
index cf1270c..5c88a5e 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/BeJoin.java
@@ -15,22 +15,25 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.req;
+package org.apache.doris.stack.model;
-import java.util.List;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
/**
* be join cluster req
**/
-public class BeJoinReq {
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class BeJoin {
- private List<AgentCommon> list;
+ private String feHost;
- public List<AgentCommon> getList() {
- return list;
- }
+ private int feQueryPort;
- public void setList(List<AgentCommon> list) {
- this.list = list;
- }
+ private String beHost;
+
+ private int agentPort;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/DeployConfig.java
similarity index 74%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/model/DeployConfig.java
index 989e26f..6e91a30 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/DeployConfig.java
@@ -15,31 +15,23 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.entity;
+package org.apache.doris.stack.model;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.Id;
-import javax.persistence.Table;
-
-@Entity
-@Table(name = "t_agent_role")
+/**
+ * deploy Config
+ **/
@Data
-@NoArgsConstructor
@AllArgsConstructor
-public class AgentRoleEntity {
+@NoArgsConstructor
+public class DeployConfig {
- @Id
- @Column(length = 150)
private String host;
private String role;
- @Column(name = "install_dir")
- private String installDir;
-
+ private String conf;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/DorisStart.java
similarity index 74%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/model/DorisStart.java
index 989e26f..1f78ea4 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/DorisStart.java
@@ -15,31 +15,21 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.entity;
+package org.apache.doris.stack.model;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.Id;
-import javax.persistence.Table;
-
-@Entity
-@Table(name = "t_agent_role")
@Data
-@NoArgsConstructor
@AllArgsConstructor
-public class AgentRoleEntity {
+@NoArgsConstructor
+public class DorisStart {
- @Id
- @Column(length = 150)
private String host;
-
+ /**
+ * BE/FE
+ */
private String role;
- @Column(name = "install_dir")
- private String installDir;
-
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/InstallInfo.java
similarity index 74%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/model/InstallInfo.java
index 989e26f..93cb526 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/InstallInfo.java
@@ -15,31 +15,22 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.entity;
+package org.apache.doris.stack.model;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.Id;
-import javax.persistence.Table;
-
-@Entity
-@Table(name = "t_agent_role")
@Data
-@NoArgsConstructor
@AllArgsConstructor
-public class AgentRoleEntity {
-
- @Id
- @Column(length = 150)
+@NoArgsConstructor
+public class InstallInfo {
private String host;
-
+ /**
+ * be / fe
+ */
private String role;
- @Column(name = "install_dir")
- private String installDir;
-
+ // FOLLOWER / OBSERVER
+ private String feNodeType;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/TaskLogReq.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/AgentCommon.java
similarity index 75%
rename from manager/dm-server/src/main/java/org/apache/doris/stack/req/TaskLogReq.java
rename to manager/dm-server/src/main/java/org/apache/doris/stack/model/request/AgentCommon.java
index da44425..a454590 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/TaskLogReq.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/AgentCommon.java
@@ -15,16 +15,18 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.req;
+package org.apache.doris.stack.model.request;
-public class TaskLogReq extends TaskInfoReq {
- private int offset;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
- public int getOffset() {
- return offset;
- }
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class AgentCommon {
- public void setOffset(int offset) {
- this.offset = offset;
- }
+ private String host;
+
+ private Integer port;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/AgentInstallReq.java
similarity index 73%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/model/request/AgentInstallReq.java
index 989e26f..29f81a5 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/AgentInstallReq.java
@@ -15,31 +15,30 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.entity;
+package org.apache.doris.stack.model.request;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.Id;
-import javax.persistence.Table;
+import java.util.List;
-@Entity
-@Table(name = "t_agent_role")
@Data
-@NoArgsConstructor
@AllArgsConstructor
-public class AgentRoleEntity {
+@NoArgsConstructor
+public class AgentInstallReq {
+
+ private int clusterId;
+
+ private List<String> hosts;
+
+ private String user;
- @Id
- @Column(length = 150)
- private String host;
+ private int sshPort;
- private String role;
+ private String sshKey;
- @Column(name = "install_dir")
private String installDir;
+ private String packageUrl;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/AgentRegister.java
similarity index 75%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/model/request/AgentRegister.java
index 989e26f..0b533b5 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/AgentRegister.java
@@ -15,31 +15,20 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.entity;
+package org.apache.doris.stack.model.request;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.Id;
-import javax.persistence.Table;
-
-@Entity
-@Table(name = "t_agent_role")
@Data
-@NoArgsConstructor
@AllArgsConstructor
-public class AgentRoleEntity {
+@NoArgsConstructor
+public class AgentRegister {
- @Id
- @Column(length = 150)
private String host;
- private String role;
+ private int port;
- @Column(name = "install_dir")
private String installDir;
-
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/BeJoinReq.java
similarity index 78%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/model/request/BeJoinReq.java
index cf1270c..299248d 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/BeJoinReq.java
@@ -15,22 +15,23 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.req;
+package org.apache.doris.stack.model.request;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
import java.util.List;
/**
* be join cluster req
**/
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
public class BeJoinReq {
- private List<AgentCommon> list;
-
- public List<AgentCommon> getList() {
- return list;
- }
+ private int processId;
- public void setList(List<AgentCommon> list) {
- this.list = list;
- }
+ private List<String> hosts;
}
diff --git a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/ITaskLog.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DeployConfigReq.java
similarity index 68%
rename from manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/ITaskLog.java
rename to manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DeployConfigReq.java
index a196778..54cfa95 100644
--- a/manager/dm-agent/src/main/java/org/apache/doris/manager/agent/task/ITaskLog.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DeployConfigReq.java
@@ -15,23 +15,23 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.manager.agent.task;
+package org.apache.doris.stack.model.request;
-import org.apache.commons.lang3.tuple.ImmutablePair;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.doris.stack.model.DeployConfig;
import java.util.List;
-public interface ITaskLog {
- ImmutablePair<Integer, List<String>> stdLog(int offset, int size);
+/**
+ * deploy config req
+ **/
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class DeployConfigReq {
- List<String> errLog(int offset, int size);
-
- void appendStdLog(String log);
-
- void appendErrLog(String log);
-
- List<String> allStdLog();
-
- List<String> allErrLog();
+ List<DeployConfig> deployConfigs;
+ private int processId;
}
-
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DorisExecReq.java
similarity index 72%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DorisExecReq.java
index cf1270c..f3b36c3 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DorisExecReq.java
@@ -15,22 +15,24 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.req;
+package org.apache.doris.stack.model.request;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
import java.util.List;
/**
- * be join cluster req
+ * doris exec command req
**/
-public class BeJoinReq {
-
- private List<AgentCommon> list;
+@AllArgsConstructor
+@NoArgsConstructor
+@Data
+public class DorisExecReq {
- public List<AgentCommon> getList() {
- return list;
- }
+ private String command;
- public void setList(List<AgentCommon> list) {
- this.list = list;
- }
+ //agent role ids
+ private List<Integer> roles;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/DorisInstallReq.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DorisInstallReq.java
similarity index 78%
rename from manager/dm-server/src/main/java/org/apache/doris/stack/req/DorisInstallReq.java
rename to manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DorisInstallReq.java
index 8d505b6..b6860d7 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/DorisInstallReq.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DorisInstallReq.java
@@ -15,19 +15,21 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.req;
+package org.apache.doris.stack.model.request;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.doris.stack.model.InstallInfo;
import java.util.List;
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
public class DorisInstallReq {
- private List<InstallInfo> installInfos;
-
- public List<InstallInfo> getInstallInfos() {
- return installInfos;
- }
+ private int processId;
- public void setInstallInfos(List<InstallInfo> installInfos) {
- this.installInfos = installInfos;
- }
+ private List<InstallInfo> installInfos;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DorisStartReq.java
similarity index 71%
rename from manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java
rename to manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DorisStartReq.java
index cf1270c..8d1b9dc 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/BeJoinReq.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/model/request/DorisStartReq.java
@@ -15,22 +15,21 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.req;
+package org.apache.doris.stack.model.request;
-import java.util.List;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.doris.stack.model.DorisStart;
-/**
- * be join cluster req
- **/
-public class BeJoinReq {
+import java.util.List;
- private List<AgentCommon> list;
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class DorisStartReq {
- public List<AgentCommon> getList() {
- return list;
- }
+ private int processId;
- public void setList(List<AgentCommon> list) {
- this.list = list;
- }
+ private List<DorisStart> dorisStarts;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/DeployConfig.java b/manager/dm-server/src/main/java/org/apache/doris/stack/req/DeployConfig.java
deleted file mode 100644
index cc36061..0000000
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/DeployConfig.java
+++ /dev/null
@@ -1,56 +0,0 @@
-// 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.doris.stack.req;
-
-/**
- * deploy Config req
- **/
-public class DeployConfig {
- private String host;
- private String role;
- private String conf;
-
- public DeployConfig(String host, String conf) {
- this.host = host;
- this.conf = conf;
- }
-
- public String getHost() {
- return host;
- }
-
- public void setHost(String host) {
- this.host = host;
- }
-
- public String getRole() {
- return role;
- }
-
- public void setRole(String role) {
- this.role = role;
- }
-
- public String getConf() {
- return conf;
- }
-
- public void setConf(String conf) {
- this.conf = conf;
- }
-}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/DorisExec.java b/manager/dm-server/src/main/java/org/apache/doris/stack/req/DorisExec.java
deleted file mode 100644
index 7075b8b..0000000
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/DorisExec.java
+++ /dev/null
@@ -1,51 +0,0 @@
-// 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.doris.stack.req;
-
-public class DorisExec {
-
- private String host;
- //FE、BE
- private String role;
-
- private boolean master;
-
- public String getHost() {
- return host;
- }
-
- public void setHost(String host) {
- this.host = host;
- }
-
- public String getRole() {
- return role;
- }
-
- public void setRole(String role) {
- this.role = role;
- }
-
- public boolean isMaster() {
- return master;
- }
-
- public void setMaster(boolean master) {
- this.master = master;
- }
-}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/DorisExecReq.java b/manager/dm-server/src/main/java/org/apache/doris/stack/req/DorisExecReq.java
deleted file mode 100644
index d289a3b..0000000
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/DorisExecReq.java
+++ /dev/null
@@ -1,44 +0,0 @@
-// 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.doris.stack.req;
-
-import java.util.List;
-
-public class DorisExecReq {
-
- //START STOP RESTART
- private String command;
-
- private List<DorisExec> dorisExecs;
-
- public String getCommand() {
- return command;
- }
-
- public void setCommand(String command) {
- this.command = command;
- }
-
- public List<DorisExec> getDorisExecs() {
- return dorisExecs;
- }
-
- public void setDorisExecs(List<DorisExec> dorisExecs) {
- this.dorisExecs = dorisExecs;
- }
-}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/InstallInfo.java b/manager/dm-server/src/main/java/org/apache/doris/stack/req/InstallInfo.java
deleted file mode 100644
index 39fc8c9..0000000
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/InstallInfo.java
+++ /dev/null
@@ -1,94 +0,0 @@
-// 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.doris.stack.req;
-
-public class InstallInfo {
- //host
- private String host;
-
- //role be fe
- private String role;
-
- //package url
- private String packageUrl;
-
- private boolean mkFeMetadir;
-
- private boolean mkBeStorageDir;
-
- private String installDir;
-
- //fe.conf be.conf
- private String conf;
-
- public String getHost() {
- return host;
- }
-
- public void setHost(String host) {
- this.host = host;
- }
-
- public String getRole() {
- return role;
- }
-
- public void setRole(String role) {
- this.role = role;
- }
-
- public String getPackageUrl() {
- return packageUrl;
- }
-
- public void setPackageUrl(String packageUrl) {
- this.packageUrl = packageUrl;
- }
-
- public boolean isMkFeMetadir() {
- return mkFeMetadir;
- }
-
- public void setMkFeMetadir(boolean mkFeMetadir) {
- this.mkFeMetadir = mkFeMetadir;
- }
-
- public String getInstallDir() {
- return installDir;
- }
-
- public void setInstallDir(String installDir) {
- this.installDir = installDir;
- }
-
- public boolean isMkBeStorageDir() {
- return mkBeStorageDir;
- }
-
- public void setMkBeStorageDir(boolean mkBeStorageDir) {
- this.mkBeStorageDir = mkBeStorageDir;
- }
-
- public String getConf() {
- return conf;
- }
-
- public void setConf(String conf) {
- this.conf = conf;
- }
-}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/SshInfo.java b/manager/dm-server/src/main/java/org/apache/doris/stack/req/SshInfo.java
deleted file mode 100644
index 5252255..0000000
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/SshInfo.java
+++ /dev/null
@@ -1,73 +0,0 @@
-// 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.doris.stack.req;
-
-import java.util.List;
-
-public class SshInfo {
-
- private List<String> hosts;
-
- private String user;
-
- private Integer sshPort;
-
- private String sshKey;
-
- private String installDir;
-
- public List<String> getHosts() {
- return hosts;
- }
-
- public void setHosts(List<String> hosts) {
- this.hosts = hosts;
- }
-
- public String getUser() {
- return user;
- }
-
- public void setUser(String user) {
- this.user = user;
- }
-
- public Integer getSshPort() {
- return sshPort;
- }
-
- public void setSshPort(Integer sshPort) {
- this.sshPort = sshPort;
- }
-
- public String getSshKey() {
- return sshKey;
- }
-
- public void setSshKey(String sshKey) {
- this.sshKey = sshKey;
- }
-
- public String getInstallDir() {
- return installDir;
- }
-
- public void setInstallDir(String installDir) {
- this.installDir = installDir;
- }
-}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java b/manager/dm-server/src/main/java/org/apache/doris/stack/runner/TaskContext.java
similarity index 70%
copy from manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
copy to manager/dm-server/src/main/java/org/apache/doris/stack/runner/TaskContext.java
index 989e26f..e813e53 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/entity/AgentRoleEntity.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/runner/TaskContext.java
@@ -15,31 +15,25 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.entity;
+package org.apache.doris.stack.runner;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
+import org.apache.doris.stack.constants.TaskTypeEnum;
+import org.apache.doris.stack.entity.TaskInstanceEntity;
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.Id;
-import javax.persistence.Table;
-
-@Entity
-@Table(name = "t_agent_role")
+/**
+ * task context
+ **/
@Data
-@NoArgsConstructor
@AllArgsConstructor
-public class AgentRoleEntity {
-
- @Id
- @Column(length = 150)
- private String host;
+@NoArgsConstructor
+public class TaskContext {
- private String role;
+ private TaskTypeEnum taskType;
- @Column(name = "install_dir")
- private String installDir;
+ private TaskInstanceEntity taskInstance;
+ private Object requestParams;
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/runner/TaskExecCallback.java b/manager/dm-server/src/main/java/org/apache/doris/stack/runner/TaskExecCallback.java
new file mode 100644
index 0000000..47ac74d
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/runner/TaskExecCallback.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.doris.stack.runner;
+
+import com.google.common.util.concurrent.FutureCallback;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.doris.stack.bean.SpringApplicationContext;
+import org.apache.doris.stack.constants.ExecutionStatus;
+import org.apache.doris.stack.constants.Flag;
+import org.apache.doris.stack.dao.TaskInstanceRepository;
+import org.apache.doris.stack.entity.TaskInstanceEntity;
+
+import java.util.Date;
+
+/**
+ * task call back update status
+ **/
+@Slf4j
+public class TaskExecCallback implements FutureCallback<Object> {
+
+ private TaskContext taskContext;
+
+ private TaskInstanceRepository taskInstanceRepository;
+
+ public TaskExecCallback(TaskContext taskContext) {
+ this.taskContext = taskContext;
+ this.taskInstanceRepository = SpringApplicationContext.getBean(TaskInstanceRepository.class);
+ }
+
+ @Override
+ public void onSuccess(Object result) {
+ //change task status
+ TaskInstanceEntity taskInstance = taskContext.getTaskInstance();
+ taskInstance.setEndTime(new Date());
+ taskInstance.setResult(String.valueOf(result));
+ taskInstance.setStatus(ExecutionStatus.SUCCESS);
+ taskInstance.setFinish(Flag.YES);
+ taskInstanceRepository.save(taskInstance);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ TaskInstanceEntity taskInstance = taskContext.getTaskInstance();
+ log.error("task {} in host {} execute error:", taskInstance.getTaskType().name(), taskInstance.getHost(), throwable);
+ taskInstance.setEndTime(new Date());
+ taskInstance.setResult(throwable.getMessage());
+ taskInstance.setStatus(ExecutionStatus.FAILURE);
+ taskInstanceRepository.save(taskInstance);
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/runner/TaskExecuteThread.java b/manager/dm-server/src/main/java/org/apache/doris/stack/runner/TaskExecuteThread.java
new file mode 100644
index 0000000..7478e79
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/runner/TaskExecuteThread.java
@@ -0,0 +1,76 @@
+// 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.doris.stack.runner;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.doris.stack.bean.SpringApplicationContext;
+import org.apache.doris.stack.constants.ExecutionStatus;
+import org.apache.doris.stack.constants.TaskTypeEnum;
+import org.apache.doris.stack.dao.TaskInstanceRepository;
+import org.apache.doris.stack.entity.TaskInstanceEntity;
+import org.apache.doris.stack.task.AbstractTask;
+import org.apache.doris.stack.task.InstallAgentTask;
+import org.apache.doris.stack.task.JoinBeTask;
+
+import java.util.concurrent.Callable;
+
+/**
+ * task execute thread
+ **/
+@Slf4j
+public class TaskExecuteThread implements Callable<Object> {
+
+ private AbstractTask task;
+
+ private TaskContext taskContext;
+
+ private TaskInstanceRepository taskInstanceRepository;
+
+ public TaskExecuteThread(TaskContext taskContext) {
+ this.taskContext = taskContext;
+ this.taskInstanceRepository = SpringApplicationContext.getBean(TaskInstanceRepository.class);
+ }
+
+ @Override
+ public Object call() throws Exception {
+ updateStatus();
+ task = initTask();
+ task.init();
+ task.handle();
+ return "SUCCESS";
+ }
+
+ private void updateStatus() {
+ TaskInstanceEntity taskInstance = taskContext.getTaskInstance();
+ taskInstance.setStatus(ExecutionStatus.RUNNING);
+ taskInstanceRepository.save(taskInstance);
+ }
+
+ public AbstractTask initTask() {
+ TaskTypeEnum taskType = taskContext.getTaskType();
+ switch (taskType) {
+ case INSTALL_AGENT:
+ return new InstallAgentTask(taskContext);
+ case JOIN_BE:
+ return new JoinBeTask(taskContext);
+ default:
+ log.error("not support task type: {}", taskType.name());
+ throw new IllegalArgumentException("not support task type");
+ }
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/service/ServerAgent.java b/manager/dm-server/src/main/java/org/apache/doris/stack/service/AgentProcess.java
similarity index 50%
rename from manager/dm-server/src/main/java/org/apache/doris/stack/service/ServerAgent.java
rename to manager/dm-server/src/main/java/org/apache/doris/stack/service/AgentProcess.java
index 7bedfa9..4b6f0dd 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/service/ServerAgent.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/service/AgentProcess.java
@@ -18,42 +18,41 @@
package org.apache.doris.stack.service;
import org.apache.doris.manager.common.domain.AgentRoleRegister;
-import org.apache.doris.manager.common.domain.RResult;
-import org.apache.doris.stack.req.DorisExecReq;
-import org.apache.doris.stack.req.DorisInstallReq;
-import org.apache.doris.stack.req.TaskInfoReq;
-import org.apache.doris.stack.req.TaskLogReq;
-
+import org.apache.doris.stack.model.request.BeJoinReq;
+import org.apache.doris.stack.model.request.DeployConfigReq;
+import org.apache.doris.stack.model.request.DorisExecReq;
+import org.apache.doris.stack.model.request.DorisInstallReq;
+import org.apache.doris.stack.model.request.DorisStartReq;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
import java.util.List;
/**
* server agent
**/
-public interface ServerAgent {
+public interface AgentProcess {
/**
* install doris
*/
- List<Object> install(DorisInstallReq installReq);
+ void installService(HttpServletRequest request, HttpServletResponse response, DorisInstallReq installReq) throws Exception;
/**
- * request agent rest api
+ * deploy config
*/
- List<Object> execute(DorisExecReq dorisExec);
+ void deployConfig(HttpServletRequest request, HttpServletResponse response, DeployConfigReq deployConfigReq) throws Exception;
/**
- * fetch task info
+ * start service
*/
- RResult taskInfo(TaskInfoReq taskInfo);
+ void startService(HttpServletRequest request, HttpServletResponse response, DorisStartReq dorisStart) throws Exception;
- /**
- * fetch log
- */
- RResult taskStdlog(TaskLogReq taskInfo);
+ void joinBe(HttpServletRequest request, HttpServletResponse response, BeJoinReq beJoinReq) throws Exception;
- RResult taskErrlog(TaskLogReq taskInfo);
+ boolean register(AgentRoleRegister agentReg);
- void joinBe(List<String> hosts);
+ List<Integer> execute(DorisExecReq dorisExec);
- boolean register(AgentRoleRegister agentReg);
+ Object log(String host, String type);
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/service/ProcessTask.java b/manager/dm-server/src/main/java/org/apache/doris/stack/service/ProcessTask.java
new file mode 100644
index 0000000..c784eaf
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/service/ProcessTask.java
@@ -0,0 +1,57 @@
+// 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.doris.stack.service;
+
+import org.apache.doris.stack.entity.ProcessInstanceEntity;
+import org.apache.doris.stack.entity.TaskInstanceEntity;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.util.List;
+
+public interface ProcessTask {
+
+ /**
+ * query user history installation progress
+ * null means that nothing currently being installed
+ */
+ ProcessInstanceEntity historyProgress(HttpServletRequest request, HttpServletResponse response) throws Exception;
+
+ List<TaskInstanceEntity> processProgress(HttpServletRequest request, HttpServletResponse response, int processId);
+
+ List<TaskInstanceEntity> taskProgress(HttpServletRequest request, HttpServletResponse response, int processId);
+
+ void installComplete(HttpServletRequest request, HttpServletResponse response, int processId) throws Exception;
+
+ void skipTask(int taskId);
+
+ /**
+ * Refresh the task status on the agent side again
+ */
+ void refreshAgentTaskStatus(int processId);
+
+ /**
+ * fetch task info
+ */
+ Object taskInfo(int taskId);
+
+ /**
+ * fetch log
+ */
+ Object taskLog(int taskId);
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/service/ServerProcess.java b/manager/dm-server/src/main/java/org/apache/doris/stack/service/ServerProcess.java
index 3d7dabc..7080cb7 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/service/ServerProcess.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/service/ServerProcess.java
@@ -19,9 +19,11 @@ package org.apache.doris.stack.service;
import org.apache.doris.stack.entity.AgentEntity;
import org.apache.doris.stack.entity.AgentRoleEntity;
-import org.apache.doris.stack.req.AgentRegister;
-import org.apache.doris.stack.req.SshInfo;
+import org.apache.doris.stack.model.request.AgentInstallReq;
+import org.apache.doris.stack.model.request.AgentRegister;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
import java.util.List;
/**
@@ -29,17 +31,17 @@ import java.util.List;
*/
public interface ServerProcess {
- void initAgent(SshInfo sshInfo);
-
/**
* install agent
*/
- void startAgent(SshInfo sshInfo);
+ int installAgent(HttpServletRequest request, HttpServletResponse response, AgentInstallReq agentInstallReq) throws Exception;
/**
* agent list
*/
- List<AgentEntity> agentList();
+ List<AgentEntity> agentList(int clusterId);
+
+ List<AgentRoleEntity> roleList(int clusterId);
List<AgentRoleEntity> agentRole(String host);
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/AgentProcessImpl.java b/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/AgentProcessImpl.java
new file mode 100644
index 0000000..7b931a2
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/AgentProcessImpl.java
@@ -0,0 +1,482 @@
+// 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.doris.stack.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ObjectUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.doris.manager.common.domain.AgentRoleRegister;
+import org.apache.doris.manager.common.domain.BeInstallCommandRequestBody;
+import org.apache.doris.manager.common.domain.CommandRequest;
+import org.apache.doris.manager.common.domain.CommandResult;
+import org.apache.doris.manager.common.domain.CommandType;
+import org.apache.doris.manager.common.domain.FeInstallCommandRequestBody;
+import org.apache.doris.manager.common.domain.FeStartCommandRequestBody;
+import org.apache.doris.manager.common.domain.RResult;
+import org.apache.doris.manager.common.domain.ServiceRole;
+import org.apache.doris.manager.common.domain.TaskResult;
+import org.apache.doris.manager.common.domain.WriteBeConfCommandRequestBody;
+import org.apache.doris.manager.common.domain.WriteFeConfCommandRequestBody;
+import org.apache.doris.stack.agent.AgentCache;
+import org.apache.doris.stack.agent.AgentRest;
+import org.apache.doris.stack.component.AgentRoleComponent;
+import org.apache.doris.stack.component.ProcessInstanceComponent;
+import org.apache.doris.stack.component.TaskInstanceComponent;
+import org.apache.doris.stack.constants.AgentStatus;
+import org.apache.doris.stack.constants.CmdTypeEnum;
+import org.apache.doris.stack.constants.Constants;
+import org.apache.doris.stack.constants.ExecutionStatus;
+import org.apache.doris.stack.constants.Flag;
+import org.apache.doris.stack.constants.ProcessTypeEnum;
+import org.apache.doris.stack.constants.TaskTypeEnum;
+import org.apache.doris.stack.entity.AgentEntity;
+import org.apache.doris.stack.entity.AgentRoleEntity;
+import org.apache.doris.stack.entity.ProcessInstanceEntity;
+import org.apache.doris.stack.entity.TaskInstanceEntity;
+import org.apache.doris.stack.exceptions.ServerException;
+import org.apache.doris.stack.model.BeJoin;
+import org.apache.doris.stack.model.DeployConfig;
+import org.apache.doris.stack.model.DorisStart;
+import org.apache.doris.stack.model.InstallInfo;
+import org.apache.doris.stack.model.request.BeJoinReq;
+import org.apache.doris.stack.model.request.DeployConfigReq;
+import org.apache.doris.stack.model.request.DorisExecReq;
+import org.apache.doris.stack.model.request.DorisInstallReq;
+import org.apache.doris.stack.model.request.DorisStartReq;
+import org.apache.doris.stack.runner.TaskContext;
+import org.apache.doris.stack.runner.TaskExecCallback;
+import org.apache.doris.stack.runner.TaskExecuteThread;
+import org.apache.doris.stack.service.AgentProcess;
+import org.apache.doris.stack.service.user.AuthenticationService;
+import org.apache.doris.stack.util.JdbcUtil;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.File;
+import java.io.UnsupportedEncodingException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Base64;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
+
+/**
+ * server agent
+ **/
+@Service
+@Slf4j
+public class AgentProcessImpl implements AgentProcess {
+
+ /**
+ * thread executor service
+ */
+ private final ListeningExecutorService taskExecService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
+
+ @Autowired
+ private AgentRest agentRest;
+
+ @Autowired
+ private AgentCache agentCache;
+
+ @Autowired
+ private AgentRoleComponent agentRoleComponent;
+
+ @Autowired
+ private ProcessInstanceComponent processInstanceComponent;
+
+ @Autowired
+ private TaskInstanceComponent taskInstanceComponent;
+
+ @Autowired
+ private AuthenticationService authenticationService;
+
+ @Override
+ @Transactional
+ public void installService(HttpServletRequest request, HttpServletResponse response,
+ DorisInstallReq installReq) throws Exception {
+ int userId = authenticationService.checkAllUserAuthWithCookie(request, response);
+ processInstanceComponent.checkHasUnfinishProcess(userId, installReq.getProcessId());
+ boolean success = taskInstanceComponent.checkParentTaskSuccess(installReq.getProcessId(), ProcessTypeEnum.INSTALL_SERVICE);
+ Preconditions.checkArgument(success, "The agent is not installed successfully and the service cannot be installed");
+
+ ProcessInstanceEntity process = processInstanceComponent.refreshProcess(installReq.getProcessId(), ProcessTypeEnum.INSTALL_SERVICE);
+ //Installed host and service
+ List<String> agentRoleList = agentRoleComponent.queryAgentRoles(process.getClusterId()).stream()
+ .map(m -> (m.getHost() + "-" + m.getRole()))
+ .collect(Collectors.toList());
+ List<InstallInfo> installInfos = installReq.getInstallInfos();
+ if (installInfos == null) {
+ throw new ServerException("Please specify the host configuration to be installed");
+ }
+ for (InstallInfo install : installInfos) {
+ String key = install.getHost() + "-" + install.getRole();
+ if (agentRoleList.contains(key)) {
+ log.warn("agent {} already install doris {}", install.getHost(), install.getRole());
+ continue;
+ }
+ String installDir = process.getInstallDir();
+ if (!installDir.endsWith(File.separator)) {
+ installDir = installDir + File.separator;
+ }
+ installDir = installDir + install.getRole().toLowerCase();
+ installDoris(process.getId(), install, process.getPackageUrl(), installDir);
+
+ agentRoleComponent.saveAgentRole(new AgentRoleEntity(install.getHost(), install.getRole(), install.getFeNodeType(), installDir, Flag.NO));
+ log.info("agent {} installing doris {}", install.getHost(), install.getRole());
+ }
+ }
+
+ private void installDoris(int processId, InstallInfo install, String packageUrl, String installDir) {
+ CommandRequest creq = new CommandRequest();
+ TaskInstanceEntity installService = new TaskInstanceEntity(processId, install.getHost(), ProcessTypeEnum.INSTALL_SERVICE);
+ if (ServiceRole.FE.name().equals(install.getRole())) {
+ FeInstallCommandRequestBody feBody = new FeInstallCommandRequestBody();
+ feBody.setMkFeMetadir(true);
+ feBody.setPackageUrl(packageUrl);
+ feBody.setInstallDir(installDir);
+ creq.setCommandType(CommandType.INSTALL_FE.name());
+ creq.setBody(JSON.toJSONString(feBody));
+ installService.setTaskType(TaskTypeEnum.INSTALL_FE);
+ } else if (ServiceRole.BE.name().equals(install.getRole())) {
+ BeInstallCommandRequestBody beBody = new BeInstallCommandRequestBody();
+ beBody.setMkBeStorageDir(true);
+ beBody.setPackageUrl(packageUrl);
+ beBody.setInstallDir(installDir);
+ creq.setCommandType(CommandType.INSTALL_BE.name());
+ creq.setBody(JSON.toJSONString(beBody));
+ installService.setTaskType(TaskTypeEnum.INSTALL_BE);
+ } else {
+ throw new ServerException("The service installation is not currently supported");
+ }
+ handleAgentTask(installService, install.getHost(), creq);
+ }
+
+ private void handleAgentTask(TaskInstanceEntity installService, String host, CommandRequest creq) {
+ boolean isRunning = taskInstanceComponent.checkTaskRunning(installService.getProcessId(), installService.getHost(), installService.getProcessType(), installService.getTaskType());
+ if (isRunning) {
+ return;
+ }
+ RResult result = agentRest.commandExec(host, agentPort(host), creq);
+ refreshAgentResult(host, agentPort(host), result);
+ taskInstanceComponent.refreshTask(installService, result);
+ }
+
+ @Override
+ public void deployConfig(HttpServletRequest request, HttpServletResponse response, DeployConfigReq deployConfigReq) throws Exception {
+ int userId = authenticationService.checkAllUserAuthWithCookie(request, response);
+ processInstanceComponent.checkHasUnfinishProcess(userId, deployConfigReq.getProcessId());
+ boolean success = taskInstanceComponent.checkParentTaskSuccess(deployConfigReq.getProcessId(), ProcessTypeEnum.DEPLOY_CONFIG);
+ Preconditions.checkArgument(success, "doris is not installed successfully and the configuration cannot be delivered");
+
+ ProcessInstanceEntity process = processInstanceComponent.refreshProcess(deployConfigReq.getProcessId(), ProcessTypeEnum.DEPLOY_CONFIG);
+ List<DeployConfig> deployConfigs = deployConfigReq.getDeployConfigs();
+ for (DeployConfig config : deployConfigs) {
+ deployConf(process.getId(), config);
+ log.info("agent {} deploy {} conf", config.getHost(), config.getRole());
+ }
+ }
+
+ private void deployConf(int processId, DeployConfig deployConf) {
+ if (StringUtils.isBlank(deployConf.getConf())) {
+ return;
+ }
+ Base64.Encoder encoder = Base64.getEncoder();
+ try {
+ deployConf.setConf(new String(encoder.encode(deployConf.getConf().getBytes()), "UTF-8"));
+ } catch (UnsupportedEncodingException e) {
+ log.error("conf {} can not encoding:", deployConf.getConf(), e);
+ }
+ CommandRequest creq = new CommandRequest();
+ TaskInstanceEntity deployTask = new TaskInstanceEntity(processId, deployConf.getHost(), ProcessTypeEnum.DEPLOY_CONFIG);
+ if (ServiceRole.FE.name().equals(deployConf.getRole())) {
+ WriteFeConfCommandRequestBody feConf = new WriteFeConfCommandRequestBody();
+ feConf.setContent(deployConf.getConf());
+ feConf.setCreateMetaDir(true);
+ creq.setBody(JSON.toJSONString(feConf));
+ creq.setCommandType(CommandType.WRITE_FE_CONF.name());
+ deployTask.setTaskType(TaskTypeEnum.DEPLOY_FE_CONFIG);
+ } else if (ServiceRole.BE.name().equals(deployConf.getRole())) {
+ WriteBeConfCommandRequestBody beConf = new WriteBeConfCommandRequestBody();
+ beConf.setContent(deployConf.getConf());
+ beConf.setCreateStorageDir(true);
+ creq.setBody(JSON.toJSONString(beConf));
+ creq.setCommandType(CommandType.WRITE_BE_CONF.name());
+ deployTask.setTaskType(TaskTypeEnum.DEPLOY_BE_CONFIG);
+ }
+ handleAgentTask(deployTask, deployConf.getHost(), creq);
+ }
+
+ @Override
+ public void startService(HttpServletRequest request, HttpServletResponse response, DorisStartReq dorisStart) throws Exception {
+ int userId = authenticationService.checkAllUserAuthWithCookie(request, response);
+ processInstanceComponent.checkHasUnfinishProcess(userId, dorisStart.getProcessId());
+ boolean success = taskInstanceComponent.checkParentTaskSuccess(dorisStart.getProcessId(), ProcessTypeEnum.START_SERVICE);
+ Preconditions.checkArgument(success, "The configuration was not successfully delivered and the service could not be started");
+
+ ProcessInstanceEntity process = processInstanceComponent.refreshProcess(dorisStart.getProcessId(), ProcessTypeEnum.START_SERVICE);
+
+ String leaderFe = getLeaderFeHostPort(process.getClusterId(), request);
+ List<DorisStart> dorisStarts = dorisStart.getDorisStarts();
+ for (DorisStart start : dorisStarts) {
+ CommandType commandType = transAgentCmd(CmdTypeEnum.START, ServiceRole.findByName(start.getRole()));
+ if (commandType == null) {
+ log.error("not support command {} {}", CmdTypeEnum.START, start.getRole());
+ continue;
+ }
+ CommandRequest creq = new CommandRequest();
+ TaskInstanceEntity execTask = new TaskInstanceEntity(process.getId(), start.getHost(), ProcessTypeEnum.START_SERVICE);
+ switch (commandType) {
+ case START_FE:
+ FeStartCommandRequestBody feBody = new FeStartCommandRequestBody();
+ if (StringUtils.isNotBlank(leaderFe)) {
+ feBody.setHelpHostPort(leaderFe);
+ }
+ creq.setBody(JSON.toJSONString(feBody));
+ execTask.setTaskType(TaskTypeEnum.START_FE);
+ break;
+ case START_BE:
+ execTask.setTaskType(TaskTypeEnum.START_BE);
+ break;
+ default:
+ log.error("not support command: {}", commandType.name());
+ break;
+ }
+ creq.setCommandType(commandType.name());
+
+ handleAgentTask(execTask, start.getHost(), creq);
+ log.info("agent {} starting {} ", start.getHost(), start.getRole());
+ }
+ }
+
+ /**
+ * get fe http port
+ **/
+ public Integer getFeQueryPort(String host, Integer port) {
+ Properties feConf = agentRest.roleConfig(host, port, ServiceRole.FE.name());
+ try {
+ Integer httpPort = Integer.valueOf(feConf.getProperty(Constants.KEY_FE_QUERY_PORT));
+ return httpPort;
+ } catch (NumberFormatException e) {
+ log.warn("get fe http port fail,return default port 8030");
+ return Constants.DORIS_DEFAULT_FE_HTTP_PORT;
+ }
+ }
+
+ /**
+ * get alive agent
+ */
+ public AgentEntity getAliveAgent(int cluserId) {
+ List<AgentRoleEntity> agentRoleEntities = agentRoleComponent.queryAgentByRole(ServiceRole.FE.name(), cluserId);
+ AgentEntity aliveAgent = null;
+ for (AgentRoleEntity agentRole : agentRoleEntities) {
+ aliveAgent = agentCache.agentInfo(agentRole.getHost());
+ if (AgentStatus.RUNNING.equals(aliveAgent.getStatus())) {
+ break;
+ }
+ }
+ Preconditions.checkNotNull(aliveAgent, "no agent alive");
+ return aliveAgent;
+ }
+
+ /**
+ * query leader fe host editLogPort
+ */
+ public String getLeaderFeHostPort(int clusterId, HttpServletRequest request) {
+ AgentEntity aliveAgent = getAliveAgent(clusterId);
+ Integer queryPort = getFeQueryPort(aliveAgent.getHost(), aliveAgent.getPort());
+ //query leader fe
+ Connection conn = null;
+ PreparedStatement stmt = null;
+ ResultSet rs = null;
+ String leaderFe = null;
+ try {
+ conn = JdbcUtil.getConnection(aliveAgent.getHost(), queryPort);
+ stmt = conn.prepareStatement("SHOW FRONTENDS");
+ rs = stmt.executeQuery();
+ while (rs.next()) {
+ boolean isMaster = rs.getBoolean("IsMaster");
+ if (isMaster) {
+ String ip = rs.getString("IP");
+ String editLogPort = rs.getString("EditLogPort");
+ leaderFe = ip + ":" + editLogPort;
+ break;
+ }
+ }
+ } catch (SQLException e) {
+ log.error("query show frontends fail", e);
+ } finally {
+ JdbcUtil.closeStmt(stmt);
+ JdbcUtil.closeRs(rs);
+ JdbcUtil.closeConn(conn);
+ }
+ if (StringUtils.isBlank(leaderFe)) {
+ log.error("can not get leader fe");
+ }
+ return leaderFe;
+ }
+
+ /**
+ * trans server command to agent command
+ */
+ public CommandType transAgentCmd(CmdTypeEnum cmdType, ServiceRole role) {
+ Preconditions.checkNotNull(cmdType, "unrecognized cmd type " + cmdType);
+ Preconditions.checkNotNull(role, "unrecognized role " + role);
+ String cmd = cmdType.name() + "_" + role.name();
+ return CommandType.findByName(cmd);
+ }
+
+ private int agentPort(String host) {
+ AgentEntity agent = agentCache.agentInfo(host);
+ if (agent == null) {
+ throw new ServerException("query agent port fail");
+ }
+ return agent.getPort();
+ }
+
+ @Override
+ public void joinBe(HttpServletRequest request, HttpServletResponse response, BeJoinReq beJoinReq) throws Exception {
+ int userId = authenticationService.checkAllUserAuthWithCookie(request, response);
+ processInstanceComponent.checkHasUnfinishProcess(userId, beJoinReq.getProcessId());
+ boolean success = taskInstanceComponent.checkParentTaskSuccess(beJoinReq.getProcessId(), ProcessTypeEnum.BUILD_CLUSTER);
+ Preconditions.checkArgument(success, "The service has not been started and completed, and the component cannot be clustered");
+ ProcessInstanceEntity process = processInstanceComponent.refreshProcess(beJoinReq.getProcessId(), ProcessTypeEnum.BUILD_CLUSTER);
+
+ for (String be : beJoinReq.getHosts()) {
+ addBeToCluster(process.getId(), be, process.getClusterId());
+ }
+ }
+
+ private void addBeToCluster(int processId, String be, int clusterId) {
+ int agentPort = agentPort(be);
+ AgentEntity aliveAgent = getAliveAgent(clusterId);
+ Integer queryPort = getFeQueryPort(aliveAgent.getHost(), aliveAgent.getPort());
+ TaskInstanceEntity joinBeTask = taskInstanceComponent.saveTask(processId, be, ProcessTypeEnum.BUILD_CLUSTER, TaskTypeEnum.JOIN_BE, ExecutionStatus.SUBMITTED);
+ if (joinBeTask == null) {
+ return;
+ }
+ TaskContext taskContext = new TaskContext(TaskTypeEnum.JOIN_BE, joinBeTask, new BeJoin(aliveAgent.getHost(), queryPort, be, agentPort));
+ ListenableFuture<Object> submit = taskExecService.submit(new TaskExecuteThread(taskContext));
+ Futures.addCallback(submit, new TaskExecCallback(taskContext));
+ }
+
+ @Override
+ public boolean register(AgentRoleRegister agentReg) {
+ AgentRoleEntity agent = agentRoleComponent.queryByHostRole(agentReg.getHost(), agentReg.getRole());
+ if (agent == null) {
+ log.error("can not find agent {} role {}", agentReg.getHost(), agentReg.getRole());
+ throw new ServerException("can not register " + agentReg.getHost() + " role " + agentReg.getRole());
+ } else if (Flag.NO.equals(agent.getRegister())) {
+ agent.setRegister(Flag.YES);
+ } else {
+ log.info("agent {} role {} already register.", agentReg.getHost(), agentReg.getRole());
+ }
+
+ AgentRoleEntity agentRoleEntity = agentRoleComponent.saveAgentRole(agent);
+ return agentRoleEntity != null;
+ }
+
+ @Override
+ public List<Integer> execute(DorisExecReq dorisExec) {
+ List<Integer> taskIds = Lists.newArrayList();
+ List<Integer> roles = dorisExec.getRoles();
+ Preconditions.checkArgument(ObjectUtils.isNotEmpty(roles), "roles can not empty");
+ List<AgentRoleEntity> agentRoleEntities = agentRoleComponent.queryAllAgentRoles();
+ for (AgentRoleEntity agentRole : agentRoleEntities) {
+ if (roles.contains(agentRole.getId())) {
+ int taskId = execute(agentRole.getHost(), agentRole.getRole(), dorisExec.getCommand());
+ taskIds.add(taskId);
+ }
+ }
+ return taskIds;
+ }
+
+ /**
+ * execute a command return taskId
+ */
+ private int execute(String host, String roleName, String commandName) {
+ CommandRequest creq = new CommandRequest();
+ TaskInstanceEntity execTask = new TaskInstanceEntity(host);
+ CommandType command = CommandType.findByName(commandName + "_" + roleName);
+ switch (command) {
+ case START_FE:
+ execTask.setTaskType(TaskTypeEnum.START_FE);
+ break;
+ case START_BE:
+ execTask.setTaskType(TaskTypeEnum.START_BE);
+ break;
+ case STOP_FE:
+ execTask.setTaskType(TaskTypeEnum.STOP_FE);
+ break;
+ case STOP_BE:
+ execTask.setTaskType(TaskTypeEnum.STOP_BE);
+ break;
+ default:
+ log.error("not support command: {}", command.name());
+ break;
+ }
+
+ creq.setCommandType(command.name());
+ RResult result = agentRest.commandExec(host, agentPort(host), creq);
+ refreshAgentResult(execTask.getHost(), agentPort(host), result);
+ TaskInstanceEntity taskInstanceEntity = taskInstanceComponent.refreshTask(execTask, result);
+ log.info("agent {} {} {} ", host, execTask.getTaskType().name(), roleName);
+ return taskInstanceEntity.getId();
+ }
+
+ /**
+ * Re-fetch result after requesting the agent interface each time
+ */
+ private RResult refreshAgentResult(String host, Integer port, RResult result) {
+ if (result == null || result.getData() == null) {
+ return null;
+ }
+ CommandResult commandResult = JSON.parseObject(JSON.toJSONString(result.getData()), CommandResult.class);
+ if (commandResult == null) {
+ return null;
+ }
+ TaskResult taskResult = commandResult.getTaskResult();
+ if (taskResult == null) {
+ return null;
+ }
+ return agentRest.taskInfo(host, port, taskResult.getTaskId());
+ }
+
+ @Override
+ public Object log(String host, String type) {
+ RResult rResult = agentRest.serverLog(host, agentPort(host), type);
+ if (rResult != null && rResult.isSuccess()) {
+ return rResult.getData();
+ } else {
+ return "fetch log fail";
+ }
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/ProcessTaskImpl.java b/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/ProcessTaskImpl.java
new file mode 100644
index 0000000..86bb518
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/ProcessTaskImpl.java
@@ -0,0 +1,164 @@
+// 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.doris.stack.service.impl;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.doris.manager.common.domain.RResult;
+import org.apache.doris.stack.agent.AgentCache;
+import org.apache.doris.stack.agent.AgentRest;
+import org.apache.doris.stack.component.ProcessInstanceComponent;
+import org.apache.doris.stack.component.TaskInstanceComponent;
+import org.apache.doris.stack.constants.Flag;
+import org.apache.doris.stack.dao.TaskInstanceRepository;
+import org.apache.doris.stack.entity.AgentEntity;
+import org.apache.doris.stack.entity.ProcessInstanceEntity;
+import org.apache.doris.stack.entity.TaskInstanceEntity;
+import org.apache.doris.stack.exceptions.ServerException;
+import org.apache.doris.stack.service.ProcessTask;
+import org.apache.doris.stack.service.user.AuthenticationService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * agent task service
+ **/
+@Service
+@Slf4j
+public class ProcessTaskImpl implements ProcessTask {
+
+ @Autowired
+ private AgentRest agentRest;
+
+ @Autowired
+ private AgentCache agentCache;
+
+ @Autowired
+ private ProcessInstanceComponent processInstanceComponent;
+
+ @Autowired
+ private TaskInstanceComponent taskInstanceComponent;
+
+ @Autowired
+ private TaskInstanceRepository taskInstanceRepository;
+
+ @Autowired
+ private AuthenticationService authenticationService;
+
+ @Override
+ public ProcessInstanceEntity historyProgress(HttpServletRequest request, HttpServletResponse response) throws Exception {
+ int userId = authenticationService.checkAllUserAuthWithCookie(request, response);
+ ProcessInstanceEntity processEntity = processInstanceComponent.queryProcessByuserId(userId);
+ return processEntity;
+ }
+
+ @Override
+ public List<TaskInstanceEntity> processProgress(HttpServletRequest request, HttpServletResponse response, int processId) {
+ ProcessInstanceEntity processInstance = processInstanceComponent.queryProcessById(processId);
+ if (processInstance == null) {
+ log.error("The process {} does not exist", processId);
+ throw new ServerException("The process does not exist");
+ }
+ refreshAgentTaskStatus(processId);
+ List<TaskInstanceEntity> taskEntities = taskInstanceRepository.queryTasksByProcessId(processId);
+ return taskEntities;
+ }
+
+ @Override
+ public List<TaskInstanceEntity> taskProgress(HttpServletRequest request, HttpServletResponse response, int processId) {
+ ProcessInstanceEntity processEntity = processInstanceComponent.queryProcessById(processId);
+ Preconditions.checkArgument(processEntity != null, "can not find processId " + processId);
+ refreshAgentTaskStatus(processId);
+ return taskInstanceRepository.queryTasksByProcessStep(processId, processEntity.getProcessType());
+ }
+
+ @Override
+ public void refreshAgentTaskStatus(int processId) {
+ List<TaskInstanceEntity> taskEntities = taskInstanceRepository.queryTasksByProcessId(processId);
+ for (TaskInstanceEntity task : taskEntities) {
+ if (task.getTaskType().agentTask()
+ && task.getStatus().typeIsRunning()
+ && StringUtils.isNotBlank(task.getExecutorId())) {
+ RResult rResult = agentRest.taskInfo(task.getHost(), agentPort(task.getHost()), task.getExecutorId());
+ taskInstanceComponent.refreshTask(task, rResult);
+ }
+ }
+ }
+
+ @Override
+ public void installComplete(HttpServletRequest request, HttpServletResponse response, int processId) throws Exception {
+ ProcessInstanceEntity processInstance = processInstanceComponent.queryProcessById(processId);
+ if (processInstance != null) {
+ processInstance.setFinish(Flag.YES);
+ processInstance.setUpdateTime(new Date());
+ processInstanceComponent.updateProcess(processInstance);
+ }
+ }
+
+ @Override
+ public void skipTask(int taskId) {
+ TaskInstanceEntity taskEntity = taskInstanceComponent.queryTaskById(taskId);
+ Preconditions.checkNotNull("taskId {} not exist", taskId);
+ if (taskEntity.getStatus().typeIsFailure()) {
+ taskEntity.setFinish(Flag.YES);
+ taskInstanceRepository.save(taskEntity);
+ }
+ }
+
+ @Override
+ public Object taskInfo(int taskId) {
+ TaskInstanceEntity taskEntity = taskInstanceComponent.queryTaskById(taskId);
+ Preconditions.checkNotNull("taskId {} not exist", taskId);
+ if (taskEntity.getTaskType().agentTask()) {
+ RResult result = agentRest.taskInfo(taskEntity.getHost(), agentPort(taskEntity.getHost()), taskEntity.getExecutorId());
+ return taskInstanceComponent.refreshTask(taskEntity, result);
+ } else {
+ return taskEntity;
+ }
+ }
+
+ @Override
+ public Object taskLog(int taskId) {
+ TaskInstanceEntity taskEntity = taskInstanceComponent.queryTaskById(taskId);
+ Preconditions.checkNotNull("taskId {} not exist", taskId);
+ if (taskEntity.getTaskType().agentTask()) {
+ RResult result = agentRest.taskLog(taskEntity.getHost(), agentPort(taskEntity.getHost()), taskEntity.getExecutorId());
+ return result.getData();
+ } else {
+ Map<String, Object> result = Maps.newHashMap();
+ result.put("log", taskEntity.getResult());
+ return result;
+ }
+ }
+
+ private int agentPort(String host) {
+ AgentEntity agent = agentCache.agentInfo(host);
+ if (agent == null) {
+ throw new ServerException("query agent port fail");
+ }
+ return agent.getPort();
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/ServerAgentImpl.java b/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/ServerAgentImpl.java
deleted file mode 100644
index 182d51f..0000000
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/ServerAgentImpl.java
+++ /dev/null
@@ -1,313 +0,0 @@
-// 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.doris.stack.service.impl;
-
-import com.alibaba.fastjson.JSON;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.doris.manager.common.domain.AgentRoleRegister;
-import org.apache.doris.manager.common.domain.BeInstallCommandRequestBody;
-import org.apache.doris.manager.common.domain.CommandRequest;
-import org.apache.doris.manager.common.domain.CommandType;
-import org.apache.doris.manager.common.domain.FeInstallCommandRequestBody;
-import org.apache.doris.manager.common.domain.FeStartCommandRequestBody;
-import org.apache.doris.manager.common.domain.RResult;
-import org.apache.doris.manager.common.domain.ServiceRole;
-import org.apache.doris.manager.common.domain.WriteBeConfCommandRequestBody;
-import org.apache.doris.manager.common.domain.WriteFeConfCommandRequestBody;
-import org.apache.doris.stack.agent.AgentCache;
-import org.apache.doris.stack.agent.AgentRest;
-import org.apache.doris.stack.component.AgentComponent;
-import org.apache.doris.stack.component.AgentRoleComponent;
-import org.apache.doris.stack.constants.AgentStatus;
-import org.apache.doris.stack.constants.CmdTypeEnum;
-import org.apache.doris.stack.constants.Constants;
-import org.apache.doris.stack.entity.AgentEntity;
-import org.apache.doris.stack.entity.AgentRoleEntity;
-import org.apache.doris.stack.exceptions.JdbcException;
-import org.apache.doris.stack.exceptions.ServerException;
-import org.apache.doris.stack.req.DeployConfig;
-import org.apache.doris.stack.req.DorisExec;
-import org.apache.doris.stack.req.DorisExecReq;
-import org.apache.doris.stack.req.DorisInstallReq;
-import org.apache.doris.stack.req.InstallInfo;
-import org.apache.doris.stack.req.TaskInfoReq;
-import org.apache.doris.stack.req.TaskLogReq;
-import org.apache.doris.stack.service.ServerAgent;
-import org.apache.doris.stack.util.JdbcUtil;
-import org.apache.doris.stack.util.Preconditions;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
-
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.stream.Collectors;
-
-/**
- * server agent
- **/
-@Service
-@Slf4j
-public class ServerAgentImpl implements ServerAgent {
-
- @Autowired
- private AgentRest agentRest;
-
- @Autowired
- private AgentCache agentCache;
-
- @Autowired
- private AgentRoleComponent agentRoleComponent;
-
- @Autowired
- private AgentComponent agentComponent;
-
- @Override
- @Transactional
- public List<Object> install(DorisInstallReq installReq) {
- List<String> agentRoleList = agentRoleComponent.queryAgentRoles().stream()
- .map(m -> (m.getHost() + "-" + m.getRole()))
- .collect(Collectors.toList());
- List<Object> results = new ArrayList<>();
- List<InstallInfo> installInfos = installReq.getInstallInfos();
- for (InstallInfo install : installInfos) {
- String key = install.getHost() + "-" + install.getRole();
- if (agentRoleList.contains(key)) {
- log.warn("agent {} already install doris {}", install.getHost(), install.getRole());
- continue;
- }
- RResult result = installDoris(install);
- agentRoleComponent.registerAgentRole(new AgentRoleRegister(install.getHost(), install.getRole(), install.getInstallDir()));
- results.add(result.getData());
- }
- return results;
- }
-
- private RResult deployConf(DeployConfig deployConf) {
- CommandRequest creq = new CommandRequest();
- if (ServiceRole.FE.name().equals(deployConf.getRole())) {
- WriteFeConfCommandRequestBody feConf = new WriteFeConfCommandRequestBody();
- feConf.setContent(deployConf.getConf());
- creq.setBody(JSON.toJSONString(feConf));
- creq.setCommandType(CommandType.WRITE_FE_CONF.name());
- } else if (ServiceRole.BE.name().equals(deployConf.getRole())) {
- WriteBeConfCommandRequestBody beConf = new WriteBeConfCommandRequestBody();
- beConf.setContent(deployConf.getConf());
- creq.setBody(JSON.toJSONString(beConf));
- creq.setCommandType(CommandType.WRITE_BE_CONF.name());
- }
- RResult result = agentRest.commandExec(deployConf.getHost(), agentPort(deployConf.getHost()), creq);
- return result;
- }
-
- private RResult installDoris(InstallInfo install) {
- CommandRequest creq = new CommandRequest();
- if (ServiceRole.FE.name().equals(install.getRole())) {
- FeInstallCommandRequestBody feBody = new FeInstallCommandRequestBody();
- feBody.setMkFeMetadir(install.isMkFeMetadir());
- feBody.setPackageUrl(install.getPackageUrl());
- feBody.setInstallDir(install.getInstallDir());
- creq.setCommandType(CommandType.INSTALL_FE.name());
- creq.setBody(JSON.toJSONString(feBody));
- } else if (ServiceRole.BE.name().equals(install.getRole())) {
- BeInstallCommandRequestBody beBody = new BeInstallCommandRequestBody();
- beBody.setMkBeStorageDir(install.isMkBeStorageDir());
- beBody.setInstallDir(install.getInstallDir());
- beBody.setPackageUrl(install.getPackageUrl());
- creq.setCommandType(CommandType.INSTALL_BE.name());
- creq.setBody(JSON.toJSONString(beBody));
- }
- RResult result = agentRest.commandExec(install.getHost(), agentPort(install.getHost()), creq);
- return result;
- }
-
- @Override
- public List<Object> execute(DorisExecReq dorisExec) {
- List<Object> results = new ArrayList<>();
- CmdTypeEnum cmdType = CmdTypeEnum.findByName(dorisExec.getCommand());
- List<DorisExec> dorisExecs = dorisExec.getDorisExecs();
- for (DorisExec exec : dorisExecs) {
- CommandType commandType = transAgentCmd(cmdType, ServiceRole.findByName(exec.getRole()));
- CommandRequest creq = new CommandRequest();
- if (CommandType.START_FE.equals(commandType) && !exec.isMaster()) {
- FeStartCommandRequestBody feBody = new FeStartCommandRequestBody();
- feBody.setHelpHostPort(getLeaderFeHostPort());
- creq.setBody(JSON.toJSONString(feBody));
- }
- creq.setCommandType(commandType.name());
- RResult result = agentRest.commandExec(exec.getHost(), agentPort(exec.getHost()), creq);
- Object data = result.getData();
- results.add(data);
- }
- return results;
- }
-
- /**
- * get fe jdbc port
- **/
- public Integer getFeQueryPort(String host, Integer port) {
- Properties feConf = agentRest.roleConfig(host, port, ServiceRole.FE.name());
- try {
- Integer jdbcPort = Integer.valueOf(feConf.getProperty(Constants.KEY_FE_QUERY_PORT));
- return jdbcPort;
- } catch (NumberFormatException e) {
- log.warn("get fe query port fail,return default port 9030");
- return Constants.DORIS_DEFAULT_FE_QUERY_PORT;
- }
- }
-
- /**
- * get alive agent
- */
- public AgentEntity getAliveAgent() {
- List<AgentRoleEntity> agentRoleEntities = agentRoleComponent.queryAgentByRole(ServiceRole.FE.name());
- AgentEntity aliveAgent = null;
- for (AgentRoleEntity agentRole : agentRoleEntities) {
- aliveAgent = agentCache.agentInfo(agentRole.getHost());
- if (AgentStatus.RUNNING.name().equals(aliveAgent.getStatus())) {
- break;
- }
- }
- Preconditions.checkNotNull(aliveAgent, "no agent alive");
- return aliveAgent;
- }
-
- /**
- * query leader fe host editLogPort
- */
- public String getLeaderFeHostPort() {
- AgentEntity aliveAgent = getAliveAgent();
- Integer jdbcPort = getFeQueryPort(aliveAgent.getHost(), aliveAgent.getPort());
- //query leader fe
- Connection conn = null;
- PreparedStatement stmt = null;
- ResultSet rs = null;
- String leaderFe = null;
- try {
- conn = JdbcUtil.getConnection(aliveAgent.getHost(), jdbcPort);
- stmt = conn.prepareStatement("SHOW PROC '/frontends'");
- rs = stmt.executeQuery();
- while (rs.next()) {
- boolean isMaster = rs.getBoolean("IsMaster");
- if (isMaster) {
- String ip = rs.getString("IP");
- String editLogPort = rs.getString("EditLogPort");
- leaderFe = ip + ":" + editLogPort;
- break;
- }
- }
- } catch (SQLException e) {
- log.error("query show frontends fail", e);
- } finally {
- JdbcUtil.closeConn(conn);
- JdbcUtil.closeStmt(stmt);
- JdbcUtil.closeRs(rs);
- }
- Preconditions.checkArgument(StringUtils.isNotBlank(leaderFe), "can not get leader fe info");
- return leaderFe;
- }
-
- /**
- * trans server command to agent command
- */
- public CommandType transAgentCmd(CmdTypeEnum cmdType, ServiceRole role) {
- Preconditions.checkNotNull(cmdType, "unrecognized cmd type " + cmdType);
- Preconditions.checkNotNull(role, "unrecognized role " + role);
- String cmd = cmdType.name() + "_" + role.name();
- return CommandType.findByName(cmd);
- }
-
- @Override
- public RResult taskInfo(TaskInfoReq taskInfo) {
- Map<String, Object> param = new HashMap<>();
- param.put("taskId", taskInfo.getTaskId());
- RResult result = agentRest.taskInfo(taskInfo.getHost(), agentPort(taskInfo.getHost()), param);
- return result;
- }
-
- @Override
- public RResult taskStdlog(TaskLogReq taskInfo) {
- Map<String, Object> param = new HashMap<>();
- param.put("taskId", taskInfo.getTaskId());
- param.put("offset", taskInfo.getOffset());
- RResult result = agentRest.taskStdLog(taskInfo.getHost(), agentPort(taskInfo.getHost()), param);
- return result;
- }
-
- @Override
- public RResult taskErrlog(TaskLogReq taskInfo) {
- Map<String, Object> param = new HashMap<>();
- param.put("taskId", taskInfo.getTaskId());
- param.put("offset", taskInfo.getOffset());
- RResult result = agentRest.taskErrLog(taskInfo.getHost(), agentPort(taskInfo.getHost()), param);
- return result;
- }
-
- private Integer agentPort(String host) {
- AgentEntity agent = agentCache.agentInfo(host);
- if (agent == null) {
- throw new ServerException("query agent port fail");
- }
- return agent.getPort();
- }
-
- @Override
- public void joinBe(List<String> hosts) {
- AgentEntity aliveAgent = getAliveAgent();
- Integer jdbcPort = getFeQueryPort(aliveAgent.getHost(), aliveAgent.getPort());
- Connection conn = null;
- try {
- conn = JdbcUtil.getConnection(aliveAgent.getHost(), jdbcPort);
- } catch (SQLException e) {
- throw new JdbcException("Failed to get fe's jdbc connection");
- }
- List<Boolean> result = new ArrayList<>();
- for (String be : hosts) {
- //query be's doris port
- Properties beConf = agentRest.roleConfig(be, agentPort(be), ServiceRole.BE.name());
- String port = beConf.getProperty(Constants.KEY_BE_HEARTBEAT_PORT);
- String addBeSqlFormat = "ALTER SYSTEM ADD BACKEND %s:%s";
- String beSql = String.format(addBeSqlFormat, be, port);
- boolean flag = JdbcUtil.execute(conn, beSql);
- if (!flag) {
- log.error("add be node fail:{}", beSql);
- }
- result.add(flag);
- }
- JdbcUtil.closeConn(conn);
- RResult.success(result);
- }
-
- @Override
- public boolean register(AgentRoleRegister agentReg) {
- AgentEntity agent = agentComponent.agentInfo(agentReg.getHost());
- if (agent == null) {
- throw new ServerException("can not find " + agentReg.getHost() + " agent");
- }
- AgentRoleEntity agentRoleEntity = agentRoleComponent.registerAgentRole(agentReg);
- return agentRoleEntity != null;
- }
-}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/ServerProcessImpl.java b/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/ServerProcessImpl.java
index 0a33138..3e773f8 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/ServerProcessImpl.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/service/impl/ServerProcessImpl.java
@@ -17,38 +17,42 @@
package org.apache.doris.stack.service.impl;
-import com.google.common.collect.Lists;
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
import org.apache.doris.stack.agent.AgentCache;
import org.apache.doris.stack.component.AgentComponent;
import org.apache.doris.stack.component.AgentRoleComponent;
-import org.apache.doris.stack.constant.EnvironmentDefine;
-import org.apache.doris.stack.constants.Constants;
+import org.apache.doris.stack.component.ProcessInstanceComponent;
+import org.apache.doris.stack.component.TaskInstanceComponent;
+import org.apache.doris.stack.constants.AgentStatus;
+import org.apache.doris.stack.constants.ExecutionStatus;
+import org.apache.doris.stack.constants.ProcessTypeEnum;
+import org.apache.doris.stack.constants.TaskTypeEnum;
import org.apache.doris.stack.entity.AgentEntity;
import org.apache.doris.stack.entity.AgentRoleEntity;
-import org.apache.doris.stack.exceptions.ServerException;
-import org.apache.doris.stack.req.AgentRegister;
-import org.apache.doris.stack.req.SshInfo;
+import org.apache.doris.stack.entity.ProcessInstanceEntity;
+import org.apache.doris.stack.entity.TaskInstanceEntity;
+import org.apache.doris.stack.model.AgentInstall;
+import org.apache.doris.stack.model.request.AgentInstallReq;
+import org.apache.doris.stack.model.request.AgentRegister;
+import org.apache.doris.stack.runner.TaskContext;
+import org.apache.doris.stack.runner.TaskExecCallback;
+import org.apache.doris.stack.runner.TaskExecuteThread;
import org.apache.doris.stack.service.ServerProcess;
-import org.apache.doris.stack.shell.SCP;
-import org.apache.doris.stack.shell.SSH;
-import org.apache.doris.stack.util.Preconditions;
+import org.apache.doris.stack.service.user.AuthenticationService;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.system.ApplicationHome;
import org.springframework.stereotype.Service;
-import java.io.File;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.nio.charset.Charset;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.nio.file.attribute.PosixFilePermission;
-import java.util.HashSet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.util.Date;
import java.util.List;
-import java.util.Set;
+import java.util.concurrent.Executors;
/**
* server
@@ -57,7 +61,10 @@ import java.util.Set;
@Slf4j
public class ServerProcessImpl implements ServerProcess {
- private static final String AGENT_START_SCRIPT = Constants.KEY_DORIS_AGENT_START_SCRIPT;
+ /**
+ * thread executor service
+ */
+ private final ListeningExecutorService taskExecService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
@Autowired
private AgentComponent agentComponent;
@@ -66,49 +73,56 @@ public class ServerProcessImpl implements ServerProcess {
private AgentRoleComponent agentRoleComponent;
@Autowired
+ private ProcessInstanceComponent processInstanceComponent;
+
+ @Autowired
+ private TaskInstanceComponent taskInstanceComponent;
+
+ @Autowired
private AgentCache agentCache;
- @Override
- public void initAgent(SshInfo sshInfo) {
- ApplicationHome applicationHome = new ApplicationHome(ServerProcessImpl.class);
- String dorisManagerHome = applicationHome.getSource().getParentFile().getParentFile().toString();
- log.info("doris manager home : {}", dorisManagerHome);
- String agentHome = dorisManagerHome + File.separator + "agent";
- Preconditions.checkNotNull(sshInfo.getHosts(), "hosts is empty");
- File sshKeyFile = buildSshKeyFile();
- writeSshKeyFile(sshInfo.getSshKey(), sshKeyFile);
- scpFile(sshInfo, agentHome, sshInfo.getInstallDir());
- }
+ @Autowired
+ private AuthenticationService authenticationService;
@Override
- public void startAgent(SshInfo sshInfo) {
- String command = "sh " + sshInfo.getInstallDir() + File.separator + AGENT_START_SCRIPT + " --server " + getServerAddr() + " --agent %s";
- List<String> hosts = sshInfo.getHosts();
- for (String host : hosts) {
- File sshKeyFile = buildSshKeyFile();
- String cmd = String.format(command, host);
- SSH ssh = new SSH(sshInfo.getUser(), sshInfo.getSshPort(),
- sshKeyFile.getAbsolutePath(), host, cmd);
- Integer run = ssh.run();
- if (run != 0) {
- throw new ServerException("agent start failed");
- } else {
- log.info("agent start success");
+ public int installAgent(HttpServletRequest request, HttpServletResponse response, AgentInstallReq installReq) throws Exception {
+ Preconditions.checkArgument(StringUtils.isNotBlank(installReq.getInstallDir()), "agent install dir not empty!");
+ int userId = authenticationService.checkAllUserAuthWithCookie(request, response);
+ ProcessInstanceEntity processInstance = new ProcessInstanceEntity(installReq.getClusterId(), userId, ProcessTypeEnum.INSTALL_AGENT, installReq.getPackageUrl(), installReq.getInstallDir());
+ int processId = processInstanceComponent.saveProcess(processInstance);
+ //install agent for per host
+ for (String host : installReq.getHosts()) {
+ TaskInstanceEntity installAgent = taskInstanceComponent.saveTask(processId, host, ProcessTypeEnum.INSTALL_AGENT, TaskTypeEnum.INSTALL_AGENT, ExecutionStatus.SUBMITTED);
+ if (installAgent == null) {
+ continue;
}
+ TaskContext taskContext = new TaskContext(TaskTypeEnum.INSTALL_AGENT, installAgent, new AgentInstall(host, installReq));
+ ListenableFuture<Object> submit = taskExecService.submit(new TaskExecuteThread(taskContext));
+ Futures.addCallback(submit, new TaskExecCallback(taskContext));
+
+ //save agent
+ agentComponent.saveAgent(new AgentEntity(host, installReq.getInstallDir(), AgentStatus.INIT, installReq.getClusterId()));
+ log.info("host {} installing agent.", host);
}
+ return processId;
}
@Override
- public List<AgentEntity> agentList() {
- List<AgentEntity> agentEntities = agentComponent.queryAgentNodes(Lists.newArrayList());
+ public List<AgentEntity> agentList(int clusterId) {
+ List<AgentEntity> agentEntities = agentComponent.queryAgentNodes(clusterId);
return agentEntities;
}
@Override
+ public List<AgentRoleEntity> roleList(int clusterId) {
+ List<AgentRoleEntity> agentRoleEntities = agentRoleComponent.queryAgentRoles(clusterId);
+ return agentRoleEntities;
+ }
+
+ @Override
public List<AgentRoleEntity> agentRole(String host) {
List<AgentRoleEntity> agentRoles = agentRoleComponent.queryAgentByHost(host);
return agentRoles;
-
}
@Override
@@ -119,85 +133,19 @@ public class ServerProcessImpl implements ServerProcess {
@Override
public boolean register(AgentRegister agent) {
AgentEntity agentEntity = agentComponent.agentInfo(agent.getHost());
- if (agentEntity != null) {
+ if (agentEntity == null) {
+ agentEntity = new AgentEntity(agent.getHost(), agent.getPort(), agent.getInstallDir(), AgentStatus.REGISTER);
+ } else if (AgentStatus.INIT.equals(agentEntity.getStatus())) {
+ agentEntity.setStatus(AgentStatus.REGISTER);
+ agentEntity.setPort(agent.getPort());
+ agentEntity.setInstallDir(agent.getInstallDir());
+ agentEntity.setRegisterTime(new Date());
+ } else {
log.warn("agent already register");
return true;
}
- AgentEntity newAgentEntity = agentComponent.registerAgent(agent);
- agentCache.putAgent(newAgentEntity);
+ agentCache.putAgent(agentComponent.saveAgent(agentEntity));
log.info("agent register success");
return true;
}
-
- /**
- * scp agent package
- */
- private void scpFile(SshInfo sshDesc, String localPath, String remotePath) {
- List<String> hosts = sshDesc.getHosts();
- String checkFileExistCmd = "if test -e " + remotePath + "; then echo ok; else mkdir -p " + remotePath + " ;fi";
- File sshKeyFile = buildSshKeyFile();
- for (String host : hosts) {
- //check remote dir exist
- SSH ssh = new SSH(sshDesc.getUser(), sshDesc.getSshPort(),
- sshKeyFile.getAbsolutePath(), host, checkFileExistCmd);
- if (ssh.run() != 0) {
- throw new ServerException("scp create remote dir failed");
- }
- SCP scp = new SCP(sshDesc.getUser(), sshDesc.getSshPort(),
- sshKeyFile.getAbsolutePath(), host, localPath, remotePath);
- Integer run = scp.run();
- if (run != 0) {
- log.error("scp agent package failed:{} to {}", localPath, remotePath);
- throw new ServerException("scp agent package failed");
- }
- }
- }
-
- /**
- * sshkey trans to file
- */
- private void writeSshKeyFile(String sshKey, File sshKeyFile) {
- try {
- if (sshKeyFile.exists()) {
- sshKeyFile.delete();
- }
- FileUtils.writeStringToFile(sshKeyFile, sshKey, Charset.defaultCharset());
- } catch (IOException e) {
- log.error("build sshKey file failed:", e);
- throw new ServerException("build sshKey file failed");
- }
- }
-
- /**
- * build sshkeyfile per request
- */
- private File buildSshKeyFile() {
- File sshKeyFile = new File("conf", "sshkey");
-
- // chmod ssh key file permission to 600
- try {
- Set<PosixFilePermission> permission = new HashSet<>();
- permission.add(PosixFilePermission.OWNER_READ);
- permission.add(PosixFilePermission.OWNER_WRITE);
- Files.setPosixFilePermissions(Paths.get(sshKeyFile.getAbsolutePath()), permission);
- } catch (IOException e) {
- log.error("set ssh key file permission fail");
- }
- return sshKeyFile;
- }
-
- /**
- * get server address
- */
- private String getServerAddr() {
- String host = null;
- try {
- host = InetAddress.getLocalHost().getHostAddress();
- } catch (UnknownHostException e) {
- throw new ServerException("get server ip fail");
- }
- String port = System.getenv(EnvironmentDefine.STUDIO_PORT_ENV);
- return host + ":" + port;
- }
-
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/shell/SCP.java b/manager/dm-server/src/main/java/org/apache/doris/stack/shell/SCP.java
index 0b907bd..b266ce4 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/shell/SCP.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/shell/SCP.java
@@ -23,13 +23,13 @@ package org.apache.doris.stack.shell;
public class SCP extends BaseCommand {
private String user;
- private Integer sshPort;
+ private int sshPort;
private String sshKeyFile;
private String host;
private String localPath;
private String remotePath;
- public SCP(String user, Integer sshPort, String sshKeyFile, String host, String localPath, String remotePath) {
+ public SCP(String user, int sshPort, String sshKeyFile, String host, String localPath, String remotePath) {
this.user = user;
this.sshPort = sshPort;
this.sshKeyFile = sshKeyFile;
@@ -45,7 +45,7 @@ public class SCP extends BaseCommand {
"-o", "ConnectTimeOut=60",
"-o", "StrictHostKeyChecking=no",
"-o", "BatchMode=yes",
- "-P", this.sshPort.toString(),
+ "-P", String.valueOf(this.sshPort),
"-i", this.sshKeyFile,
this.localPath,
this.user + "@" + this.host + ":" + this.remotePath
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/shell/SSH.java b/manager/dm-server/src/main/java/org/apache/doris/stack/shell/SSH.java
index c0e7e18..938c13e 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/shell/SSH.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/shell/SSH.java
@@ -24,7 +24,7 @@ public class SSH extends BaseCommand {
private String user;
- private Integer sshPort;
+ private int sshPort;
private String sshKeyFile;
@@ -32,7 +32,7 @@ public class SSH extends BaseCommand {
private String command;
- public SSH(String user, Integer sshPort, String sshKeyFile, String host, String command) {
+ public SSH(String user, int sshPort, String sshKeyFile, String host, String command) {
this.user = user;
this.sshPort = sshPort;
this.sshKeyFile = sshKeyFile;
@@ -47,7 +47,7 @@ public class SSH extends BaseCommand {
"-o", "BatchMode=yes",
"-tt",
"-i", this.sshKeyFile,
- "-p", this.sshPort.toString(),
+ "-p", String.valueOf(this.sshPort),
this.user + "@" + this.host, this.command
};
this.resultCommand = command;
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/req/TaskInfoReq.java b/manager/dm-server/src/main/java/org/apache/doris/stack/task/AbstractTask.java
similarity index 68%
rename from manager/dm-server/src/main/java/org/apache/doris/stack/req/TaskInfoReq.java
rename to manager/dm-server/src/main/java/org/apache/doris/stack/task/AbstractTask.java
index 06c6fc9..8cbd0e5 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/req/TaskInfoReq.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/task/AbstractTask.java
@@ -15,27 +15,27 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.stack.req;
+package org.apache.doris.stack.task;
-public class TaskInfoReq {
+import org.apache.doris.stack.runner.TaskContext;
- private String host;
+public abstract class AbstractTask {
- private String taskId;
+ protected TaskContext taskContext;
- public String getHost() {
- return host;
+ public AbstractTask(TaskContext taskContext) {
+ this.taskContext = taskContext;
}
- public void setHost(String host) {
- this.host = host;
+ /**
+ * init task
+ */
+ public void init() {
}
- public String getTaskId() {
- return taskId;
- }
+ /**
+ * task handle
+ */
+ public abstract void handle();
- public void setTaskId(String taskId) {
- this.taskId = taskId;
- }
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/task/InstallAgentTask.java b/manager/dm-server/src/main/java/org/apache/doris/stack/task/InstallAgentTask.java
new file mode 100644
index 0000000..d9b7835
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/task/InstallAgentTask.java
@@ -0,0 +1,162 @@
+// 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.doris.stack.task;
+
+import com.google.common.base.Preconditions;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.io.FileUtils;
+import org.apache.doris.stack.constant.EnvironmentDefine;
+import org.apache.doris.stack.constants.Constants;
+import org.apache.doris.stack.exceptions.ServerException;
+import org.apache.doris.stack.model.AgentInstall;
+import org.apache.doris.stack.runner.TaskContext;
+import org.apache.doris.stack.service.impl.ServerProcessImpl;
+import org.apache.doris.stack.shell.SCP;
+import org.apache.doris.stack.shell.SSH;
+import org.springframework.boot.system.ApplicationHome;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.nio.charset.Charset;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.nio.file.attribute.PosixFilePermission;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * intall agent task
+ **/
+@Slf4j
+public class InstallAgentTask extends AbstractTask {
+
+ private static final String AGENT_START_SCRIPT = Constants.KEY_DORIS_AGENT_START_SCRIPT;
+
+ public InstallAgentTask(TaskContext taskContext) {
+ super(taskContext);
+ }
+
+ @Override
+ public void handle() {
+ AgentInstall requestParams = (AgentInstall) taskContext.getRequestParams();
+ distAgentPackage(requestParams);
+ startAgent(requestParams);
+ }
+
+ /**
+ * scp agent to host
+ */
+ private void distAgentPackage(AgentInstall agentInstall) {
+ ApplicationHome applicationHome = new ApplicationHome(ServerProcessImpl.class);
+ String dorisManagerHome = applicationHome.getSource().getParentFile().getParentFile().getParentFile().toString();
+ log.info("doris manager home : {}", dorisManagerHome);
+ String agentHome = dorisManagerHome + File.separator + "agent";
+ Preconditions.checkNotNull(agentInstall.getHost(), "host is empty");
+ File sshKeyFile = buildSshKeyFile();
+ writeSshKeyFile(agentInstall.getSshKey(), sshKeyFile);
+ scpFile(agentInstall, agentHome, agentInstall.getInstallDir());
+ }
+
+ /**
+ * start agent
+ */
+ private void startAgent(AgentInstall agentInstall) {
+ String agentHome = agentInstall.getInstallDir() + File.separator + "agent";
+ String command = "cd %s && sh %s --server %s --agent %s";
+ File sshKeyFile = buildSshKeyFile();
+ String cmd = String.format(command, agentHome, AGENT_START_SCRIPT, getServerAddr(), agentInstall.getHost());
+ SSH ssh = new SSH(agentInstall.getUser(), agentInstall.getSshPort(),
+ sshKeyFile.getAbsolutePath(), agentInstall.getHost(), cmd);
+ Integer run = ssh.run();
+ if (run != 0) {
+ throw new ServerException("agent start failed");
+ } else {
+ log.info("agent start success");
+ }
+ }
+
+ /**
+ * scp agent package
+ */
+ private void scpFile(AgentInstall agentInstall, String localPath, String remotePath) {
+ String checkFileExistCmd = "if test -e " + remotePath + "; then echo ok; else mkdir -p " + remotePath + " ;fi";
+ File sshKeyFile = buildSshKeyFile();
+ //check remote dir exist
+ SSH ssh = new SSH(agentInstall.getUser(), agentInstall.getSshPort(),
+ sshKeyFile.getAbsolutePath(), agentInstall.getHost(), checkFileExistCmd);
+ if (ssh.run() != 0) {
+ throw new ServerException("scp create remote dir failed");
+ }
+ SCP scp = new SCP(agentInstall.getUser(), agentInstall.getSshPort(),
+ sshKeyFile.getAbsolutePath(), agentInstall.getHost(), localPath, remotePath);
+ Integer run = scp.run();
+ if (run != 0) {
+ log.error("scp agent package failed:{} to {}", localPath, remotePath);
+ throw new ServerException("scp agent package failed");
+ }
+ }
+
+ /**
+ * sshkey trans to file
+ */
+ private void writeSshKeyFile(String sshKey, File sshKeyFile) {
+ try {
+ if (sshKeyFile.exists()) {
+ sshKeyFile.delete();
+ }
+ FileUtils.writeStringToFile(sshKeyFile, sshKey, Charset.defaultCharset());
+ } catch (IOException e) {
+ log.error("build sshKey file failed:", e);
+ throw new ServerException("build sshKey file failed");
+ }
+ }
+
+ /**
+ * build sshkeyfile per request
+ */
+ private File buildSshKeyFile() {
+ File sshKeyFile = new File("conf", "sshkey");
+
+ // chmod ssh key file permission to 600
+ try {
+ Set<PosixFilePermission> permission = new HashSet<>();
+ permission.add(PosixFilePermission.OWNER_READ);
+ permission.add(PosixFilePermission.OWNER_WRITE);
+ Files.setPosixFilePermissions(Paths.get(sshKeyFile.getAbsolutePath()), permission);
+ } catch (IOException e) {
+ log.error("set ssh key file permission fail");
+ }
+ return sshKeyFile;
+ }
+
+ /**
+ * get server address
+ */
+ private String getServerAddr() {
+ String host = null;
+ try {
+ host = InetAddress.getLocalHost().getHostAddress();
+ } catch (UnknownHostException e) {
+ throw new ServerException("get server ip fail");
+ }
+ String port = System.getenv(EnvironmentDefine.STUDIO_PORT_ENV);
+ return host + ":" + port;
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/task/JoinBeTask.java b/manager/dm-server/src/main/java/org/apache/doris/stack/task/JoinBeTask.java
new file mode 100644
index 0000000..5ffec40
--- /dev/null
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/task/JoinBeTask.java
@@ -0,0 +1,71 @@
+// 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.doris.stack.task;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.doris.manager.common.domain.ServiceRole;
+import org.apache.doris.stack.agent.AgentRest;
+import org.apache.doris.stack.bean.SpringApplicationContext;
+import org.apache.doris.stack.constants.Constants;
+import org.apache.doris.stack.exceptions.JdbcException;
+import org.apache.doris.stack.exceptions.ServerException;
+import org.apache.doris.stack.model.BeJoin;
+import org.apache.doris.stack.runner.TaskContext;
+import org.apache.doris.stack.util.JdbcUtil;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.Properties;
+
+/**
+ * join be to cluster
+ **/
+@Slf4j
+public class JoinBeTask extends AbstractTask {
+
+ private AgentRest agentRest;
+
+ public JoinBeTask(TaskContext taskContext) {
+ super(taskContext);
+ this.agentRest = SpringApplicationContext.getBean(AgentRest.class);
+ }
+
+ @Override
+ public void handle() {
+ BeJoin requestParams = (BeJoin) taskContext.getRequestParams();
+ String backendHost = requestParams.getBeHost();
+ Properties beConf = agentRest.roleConfig(backendHost, requestParams.getAgentPort(), ServiceRole.BE.name());
+ String beHeatPort = beConf.getProperty(Constants.KEY_BE_HEARTBEAT_PORT);
+ Connection conn = null;
+ try {
+ conn = JdbcUtil.getConnection(requestParams.getFeHost(), requestParams.getFeQueryPort());
+ } catch (SQLException e) {
+ log.error("get connection fail:", e);
+ throw new JdbcException("Failed to get fe's jdbc connection");
+ }
+
+ String addBeSqlFormat = "ALTER SYSTEM ADD BACKEND \"%s:%s\"";
+ String addBe = String.format(addBeSqlFormat, requestParams.getBeHost(), beHeatPort);
+ try {
+ JdbcUtil.execute(conn, addBe);
+ } catch (SQLException e) {
+ log.error("Failed to add backend:{}:{}", requestParams.getBeHost(), beHeatPort, e);
+ throw new ServerException(e.getMessage());
+ }
+ }
+}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/util/JdbcUtil.java b/manager/dm-server/src/main/java/org/apache/doris/stack/util/JdbcUtil.java
index b8deff8..fe7fd0a 100644
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/util/JdbcUtil.java
+++ b/manager/dm-server/src/main/java/org/apache/doris/stack/util/JdbcUtil.java
@@ -51,19 +51,15 @@ public class JdbcUtil {
return DriverManager.getConnection(url);
}
- public static boolean execute(Connection conn, String sql) {
- boolean flag = false;
+ public static boolean execute(Connection conn, String sql) throws SQLException {
PreparedStatement stmt = null;
try {
stmt = conn.prepareStatement(sql);
- flag = stmt.execute();
- } catch (SQLException e) {
- log.error("sql execute error:{}", sql, e);
+ return stmt.execute();
} finally {
closeConn(conn);
closeStmt(stmt);
}
- return flag;
}
public static void closeConn(Connection conn) {
@@ -72,7 +68,7 @@ public class JdbcUtil {
conn.close();
}
} catch (SQLException e) {
- log.error("close connection error:", e);
+ log.warn("close connection error:", e);
}
}
@@ -82,7 +78,7 @@ public class JdbcUtil {
stmt.close();
}
} catch (SQLException e) {
- log.error("close statement error:", e);
+ log.warn("close statement error:", e);
}
}
@@ -92,7 +88,7 @@ public class JdbcUtil {
rs.close();
}
} catch (SQLException e) {
- log.error("close resultset error:", e);
+ log.warn("close resultset error:", e);
}
}
}
diff --git a/manager/dm-server/src/main/java/org/apache/doris/stack/util/Preconditions.java b/manager/dm-server/src/main/java/org/apache/doris/stack/util/Preconditions.java
deleted file mode 100644
index ea9cf89..0000000
--- a/manager/dm-server/src/main/java/org/apache/doris/stack/util/Preconditions.java
+++ /dev/null
@@ -1,47 +0,0 @@
-// 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.doris.stack.util;
-
-public class Preconditions {
-
- public static <T> T checkNotNull(T reference) {
- if (reference == null) {
- throw new NullPointerException();
- }
- return reference;
- }
-
- public static <T> T checkNotNull(T reference, String errorMessage) {
- if (reference == null) {
- throw new NullPointerException(String.valueOf(errorMessage));
- }
- return reference;
- }
-
- public static void checkArgument(boolean condition) {
- if (!condition) {
- throw new IllegalArgumentException();
- }
- }
-
- public static void checkArgument(boolean condition, Object errorMessage) {
- if (!condition) {
- throw new IllegalArgumentException(String.valueOf(errorMessage));
- }
- }
-}
diff --git a/manager/dm-server/src/main/resources/dm-server.sql b/manager/dm-server/src/main/resources/dm-server.sql
deleted file mode 100644
index 2af665e..0000000
--- a/manager/dm-server/src/main/resources/dm-server.sql
+++ /dev/null
@@ -1,19 +0,0 @@
-CREATE TABLE `t_agent` (
- `id` int(11) NOT NULL AUTO_INCREMENT,
- `host` varchar(128) DEFAULT NULL COMMENT 'agent host',
- `port` int(8) DEFAULT NULL COMMENT 'agent port',
- `install_dir` varchar(256) DEFAULT NULL COMMENT 'agent install dir',
- `status` varchar(256) DEFAULT NULL COMMENT 'agent status',
- `register_time` timestamp NULL COMMENT 'register_time',
- `last_reported_time` timestamp NULL COMMENT 'last report time'
- PRIMARY KEY (`id`)
- UNIQUE KEY `agent_index` (`host`) USING BTREE
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-
-
-CREATE TABLE `t_agent_role` (
- `host` varchar(128) DEFAULT NULL COMMENT 'host',
- `role` varchar(8) DEFAULT NULL COMMENT 'BE/FE',
- `install_dir` varchar(256) DEFAULT NULL COMMENT 'doris install dir',
- UNIQUE KEY `host_index` (`host`,`role`) USING BTREE
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
\ No newline at end of file
diff --git "a/manager/doc/Doris Manager Server \346\216\245\345\217\243\346\226\207\346\241\243.md" "b/manager/doc/Doris Manager Server \346\216\245\345\217\243\346\226\207\346\241\243.md"
index 555fe38..580429e 100644
--- "a/manager/doc/Doris Manager Server \346\216\245\345\217\243\346\226\207\346\241\243.md"
+++ "b/manager/doc/Doris Manager Server \346\216\245\345\217\243\346\226\207\346\241\243.md"
@@ -1,13 +1,8 @@
[TOC]
---
-## Doris Manager API文档
-### Doris Manager 接口文档
-
-Direct access after local service startup:http://localhost:port/swagger-ui.html#/
-
-### Doris Manager Server 接口文档
+### DorisManager-Agent Server 接口文档
#### 1.安装Agent
@@ -17,7 +12,7 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
**URL**
-> /server/installAgent
+> /api/server/installAgent
**支持格式**
@@ -31,25 +26,32 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
> |参数|必选|类型|说明|
> |:----- |:-------|:-----|----- |
-> |hosts |ture |List|机器列表 |
+> |clusterId|true|String|集群ID ,新建集群后返回的id|
+> |hosts |true |List|机器列表 |
> |user |true |String |ssh 用户|
> |port |true |int |ssh 端口|
> |sshKey |true |String |ssh 私钥|
+> |packageUrl |true |String |doris编译包下载地址，只支持http方式。|
+> |installDir |true |String |安装路径|
**返回字段**
> |返回字段|字段类型|说明 |
-|:----- |:------|:----------------------------- |
-|msg |String |调用信息 |
-|code |String | 结果状态。0：正常 |
+> |:----- |:------|:----------------------------- |
+> |msg |String |调用信息 |
+> |code |String | 结果状态。0：正常 |
+> |data |Int | 当前安装的流程ID |
**接口示例**
-> 地址：http://localhost:9601/server/installAgent
+> 地址：http://localhost:9601/api/server/installAgent
> 请求参数：
``` json
{
+ "clusterId":"1",
+ "packageUrl":"https://palo-cloud-repo-bd.bd.bcebos.com/baidu-doris-release/PALO-0.15.1-rc03-no-avx2-binary.tar.gz",
+ "installDir":"/usr/local/doris",
"hosts":["10.10.10.11"],
"user":"root",
"sshPort":22,
@@ -60,10 +62,13 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
``` json
{
"msg": "success",
- "code": 0
+ "code": 0,
+ "data":1
}
```
+
+
#### 2.查看Agent列表
**接口功能**
@@ -72,7 +77,7 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
**URL**
-> /server/agentList
+> /api/server/agentList
**支持格式**
@@ -80,28 +85,29 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
**HTTP请求方式**
-> POST
+> GET
**请求参数**
-
-无
+> |返回字段|字段类型|说明 |
+> |:----- |:------|:----------------------------- |
+> |clusterId |int |集群id |
**返回字段**
> |返回字段|字段类型|说明 |
-|:----- |:------|:----------------------------- |
-|msg |String |调用信息 |
-|code |String | 结果状态。0：正常 |
-|data.id |int | agentId |
-|data.host |String |agent host |
-|data.port |String | agent port |
-|data.status |String | agent状态 ：RUNNING STOP|
-|data.registerTime |Date | agent注册时间 |
-|data.lastReportedTime |Date | agent最后上报时间 |
+> |:----- |:------|:----------------------------- |
+> |msg |String |调用信息 |
+> |code |String | 结果状态。0：正常 |
+> |data.id |int | agentId |
+> |data.host |String |agent host |
+> |data.port |String | agent port |
+> |data.status |String | agent状态 ：RUNNING STOP|
+> |data.registerTime |Date | agent注册时间 |
+> |data.lastReportedTime |Date | agent最后上报时间 |
**接口示例**
-> 地址：http://localhost:9601/server/agentList
+> 地址：http://localhost:9601/api/server/agentList
> 请求参数：无
@@ -131,7 +137,7 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
**URL**
-> /agent/installDoris
+> /api/agent/installService
**支持格式**
@@ -145,41 +151,36 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
> |参数|必选|类型|说明|
> |:----- |:-------|:-----|----- |
-> |host |ture | String |指定安装doris的机器 |
-> |role |true |String |doris角色：FE、BE|
-> |packageUrl |true |String |doris编译包下载地址，只支持http方式。|
-> |mkFeMetadir |false |boolean |安装FE时是否新建 元数据目录|
-> |mkBeStorageDir |false |boolean |安装BE时是否新建 数据存储目录|
-> |installDir |true |String |安装路径|
+> |processId|true|int|当前安装的流程ID，接口1返回的结果|
+> |installInfos.host |ture | String |指定安装doris的机器 |
+> |installInfos.role |true |String |doris角色：FE、BE|
+> |installInfos.feNodeType |false |String |角色为FE时:FOLLOWER / OBSERVER|
**返回字段**
+
> |返回字段|字段类型|说明 |
> |:----- |:------|:----------------------------- |
> |msg |String |调用信息 |
> |code |String | 结果状态。0：正常 |
-> |data.taskResult.taskId |String | 任务ID |
-> |data.taskResult.submitTime |Date | 任务提交时间 |
-> |data.taskResult.startTime |Date | 任务开始时间 |
-> |data.taskResult.endTime |Date | 任务终止时间 |
-> |data.taskResult.taskState |String | 任务状态 |
-> |data.taskResult.retCode |int | 任务返回状态。0：正常 |
-> |data.stdlogs |List | 任务输出日志 |
-> |data.errlogs |List | 任务错误日志 |
-
+>
**接口示例**
-> 地址：http://localhost:9601/agent/installDoris
+> 地址：http://localhost:9601/api/agent/installService
> 请求参数：
``` json
{
+ "processId":1,
"installInfos":[{
- "host":"10.10.10.11",
+ "host":"10.220.147.155",
"role":"FE",
- "packageUrl":"http://10.10.10.11/fileupload/doris-fe.tar.gz",
- "mkFeMetadir":true,
- "installDir":"/usr/local/doris-fe"
+ "feNodeType":"FOLLOWER"
+ },
+ {
+ "host":"10.220.147.155",
+ "role":"BE"
+
}]
}
```
@@ -187,33 +188,80 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
``` json
{
"msg": "success",
- "code": 0,
- "data": [
- {
- "taskResult": {
- "taskId": "a3b966cc10794f4c8b9b2c0dc1fa9b26",
- "submitTime": "2021-08-12T02:37:11.757+00:00",
- "startTime": "2021-08-12T02:37:11.759+00:00",
- "endTime": null,
- "taskState": "RUNNING",
- "retCode": null
- },
- "stdlogs": [],
- "errlogs": []
- }
- ]
+ "code": 0
+}
+```
+
+#### 4.分发配置
+
+**接口功能**
+
+> 分发FE、BE的配置
+
+**URL**
+
+> /api/agent/deployConfig
+
+**支持格式**
+
+> JSON
+
+**HTTP请求方式**
+
+> POST
+
+**请求参数**
+
+> |参数|必选|类型|说明|
+> |:----- |:-------|:-----|----- |
+> |processId|true|int|当前安装的流程ID，接口1返回的结果|
+> |deployConfigs.host |true |String |指定的机器|
+> |deployConfigs.role |true |String |doris角色：FE、BE|
+> |deployConfigs.conf |true |String |配置文件内容|
+
+**返回字段**
+
+> |返回字段|字段类型|说明 |
+> |:----- |:------|:----------------------------- |
+> |msg |String |调用信息 |
+> |code |String | 结果状态。0：正常 |
+
+**接口示例**
+
+> 地址：http://localhost:9601/api/agent/deployConfig
+
+> 请求参数：
+``` json
+{
+ "processId":"1",
+ "deployConfigs":[{
+ "host":"10.220.147.155",
+ "role":"FE",
+ "conf":"LOG_DIR = ${DORIS_HOME}/log\nDATE = `date +%Y%m%d-%H%M%S`\nJAVA_OPTS=\"-Xmx4096m -XX:+UseMembar -XX:SurvivorRatio=8 -XX:MaxTenuringThreshold=7 -XX:+PrintGCDateStamps -XX:+PrintGCDetails -XX:+UseConcMarkSweepGC -XX:+UseParNewGC -XX:+CMSClassUnloadingEnabled -XX:-CMSParallelRemarkEnabled -XX:CMSInitiatingOccupancyFraction=80 -XX:SoftRefLRUPolicyMSPerMB=0 -Xloggc:$DORIS_HOME/log/fe.gc.log.$DATE\"\nJAVA_OPTS_FOR_JDK_9=\"-Xmx4096m -XX:SurvivorRatio=8 -XX:MaxTenuringThreshold=7 [...]
+ },{
+ "host":"10.220.147.155",
+ "role":"BE",
+ "conf":"PPROF_TMPDIR=\"$DORIS_HOME/log/\"\nsys_log_level = INFO\nbe_port = 19060\nbe_rpc_port = 19070\nwebserver_port = 18040\nheartbeat_service_port = 19050\nbrpc_port = 18060\nstorage_root_path = /usr/local/doris/be/storage\n"
+ }]
+}
+```
+> 返回参数：
+``` json
+{
+ "msg": "success",
+ "code": 0
}
```
-#### 4.启停Doris
+#### 5.启动Doris
**接口功能**
-> 对指定机器的doris进行启动、停止
+> 启动指定机器的doris
**URL**
-> /agent/execute
+> /api/agent/startService
**支持格式**
@@ -227,9 +275,9 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
> |参数|必选|类型|说明|
> |:----- |:-------|:-----|----- |
-> |command |ture | String | 执行命令: START STOP |
-> |dorisExecs.host |true |String |指定的机器列表|
-> |dorisExecs.role |true |String |doris角色：FE、BE|
+> |processId|true|int|当前安装的流程ID，接口1返回的结果|
+> |dorisStarts.host |true |String |指定机器|
+> |dorisStarts.role |true |String |doris角色：FE、BE|
**返回字段**
@@ -237,26 +285,21 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
> |:----- |:------|:----------------------------- |
> |msg |String |调用信息 |
> |code |String | 结果状态。0：正常 |
-> |data.taskResult.taskId |String | 任务ID |
-> |data.taskResult.submitTime |Date | 任务提交时间 |
-> |data.taskResult.startTime |Date | 任务开始时间 |
-> |data.taskResult.endTime |Date | 任务终止时间 |
-> |data.taskResult.taskState |String | 任务状态 |
-> |data.taskResult.retCode |int | 任务返回状态。0：正常 |
-> |data.stdlogs |List | 任务输出日志 |
-> |data.errlogs |List | 任务错误日志 |
**接口示例**
-> 地址：http://localhost:9601/agent/execute
+> 地址：http://localhost:9601/api/agent/startService
> 请求参数：
``` json
{
- "command":"START",
- "dorisExecs":[{
- "host":"10.10.10.11",
+ "processId":"1",
+ "dorisStarts":[{
+ "host":"10.220.147.155",
"role":"FE"
+ },{
+ "host":"10.220.147.155",
+ "role":"BE"
}]
}
```
@@ -264,33 +307,19 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
``` json
{
"msg": "success",
- "code": 0,
- "data": [
- {
- "taskResult": {
- "taskId": "a3b966cc10794f4c8b9b2c0dc1fa9b26",
- "submitTime": "2021-08-12T02:37:11.757+00:00",
- "startTime": "2021-08-12T02:37:11.759+00:00",
- "endTime": null,
- "taskState": "RUNNING",
- "retCode": null
- },
- "stdlogs": [],
- "errlogs": []
- }
- ]
+ "code": 0
}
```
-#### 5.查看任务执行状态
+#### 5.启动组件集群
**接口功能**
-> 查看任务执行状态
+> 组件集群，将BE加入集群
**URL**
-> /agent/task
+> /api/agent/joinBe
**支持格式**
@@ -304,8 +333,8 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
> |参数|必选|类型|说明|
> |:----- |:-------|:-----|----- |
-> |host |ture | String |agent 机器host |
-> |taskId |true |String |任务id|
+> |processId|true|int|当前安装的流程ID，接口1返回的结果|
+> |hosts |true |String |加入机器列表|
**返回字段**
@@ -313,68 +342,35 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
> |:----- |:------|:----------------------------- |
> |msg |String |调用信息 |
> |code |String | 结果状态。0：正常 |
-> |data.taskResult.taskId |String | 任务ID |
-> |data.taskResult.submitTime |Date | 任务提交时间 |
-> |data.taskResult.startTime |Date | 任务开始时间 |
-> |data.taskResult.endTime |Date | 任务终止时间 |
-> |data.taskResult.taskState |String | 任务状态 |
-> |data.taskResult.retCode |int | 任务返回状态。0：正常 |
-> |data.stdlogs |List | 任务输出日志 |
-> |data.errlogs |List | 任务错误日志 |
**接口示例**
-> 地址：http://localhost:9601/agent/task
+> 地址：http://localhost:9601/api/agent/joinBe
> 请求参数：
``` json
{
- "host":"10.10.10.11",
- "taskId":"a3b966cc10794f4c8b9b2c0dc1fa9b26"
+ "processId":1,
+ "hosts":["10.220.147.155"]
}
```
> 返回参数：
``` json
{
"msg": "success",
- "code": 0,
- "data": {
- "taskResult": {
- "taskId": "a3b966cc10794f4c8b9b2c0dc1fa9b26",
- "submitTime": "2021-08-12T02:37:11.757+00:00",
- "startTime": "2021-08-12T02:37:11.759+00:00",
- "endTime": "2021-08-12T02:37:14.759+00:00",
- "taskState": "FINISHED",
- "retCode": 0
- },
- "stdlogs": [
- "bin/",
- "bin/start_fe.sh",
- "bin/stop_fe.sh",
- "conf/",
- "conf/fe.conf",
- "lib/",
- "lib/RoaringBitmap-0.8.13.jar",
- "lib/activation-1.1.1.jar",
- "lib/aircompressor-0.10.jar",
- "lib/amqp-client-5.7.3.jar",
- "lib/annotations-2.15.45.jar"
- .....
- ],
- "errlogs": []
- }
+ "code": 0
}
```
-#### 6.查看任务输出日志
+#### 6.完成安装
**接口功能**
-> 查看任务输出日志
+> 完成安装，完成安装后调用
**URL**
-> /agent/stdlog
+> /api/process/installComplete/{processId}
**支持格式**
@@ -388,9 +384,8 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
> |参数|必选|类型|说明|
> |:----- |:-------|:-----|----- |
-> |host |ture | String |agent 机器host |
-> |taskId |true |String |任务id|
-> |offset |false |int |日志偏移量|
+> |processId|true|int|当前安装的流程ID，接口1返回的结果|
+
**返回字段**
@@ -398,46 +393,87 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
> |:----- |:------|:----------------------------- |
> |msg |String |调用信息 |
> |code |String | 结果状态。0：正常 |
-> |data.key |String | 下一页起始偏移量 |
-> |data.value |List | 日志 |
**接口示例**
-> 地址：http://localhost:9601/agent/stdlog
+> 地址：http://localhost:9601/api/process/installComplete/1
> 请求参数：
``` json
{
- "host":"10.10.10.11",
- "taskId":"a3b966cc10794f4c8b9b2c0dc1fa9b26",
- "offset":10,
+ "processId":1,
+ "hosts":["10.220.147.155"]
}
```
> 返回参数：
``` json
{
"msg": "success",
+ "code": 0
+}
+```
+
+#### 7.查询该用户当前安装的流程
+
+**接口功能**
+
+> 查询该用户当前安装的流程
+
+**URL**
+
+> /api/process/historyProgress
+
+**支持格式**
+
+> JSON
+
+**HTTP请求方式**
+
+> POST
+
+**请求参数**
+
+**返回字段**
+
+> |返回字段|字段类型|说明 |
+> |:----- |:------|:----------------------------- |
+> |msg |String |调用信息 |
+> |code |String | 结果状态。0：正常 |
+> |data.id |String | 流程id |
+> |data.processType |String | 流程进度 |
+> |data.finish |String | 流程完成标志 |
+
+**接口示例**
+
+> 地址：http://localhost:9601/api/process/historyProgress
+
+> 返回参数：
+``` json
+{
+ "msg": "success",
"code": 0,
"data": {
- "key": 366,
- "value": [
- "lib/annotations-2.15.45.jar",
- "lib/antlr4-runtime-4.7.jar",
- .....
- ]
+ "id": 1,
+ "clusterId": 0,
+ "userId": 0,
+ "processType": "START_SERVICE",
+ "createTime": "2021-11-02T01:29:58.000+00:00",
+ "updateTime": "2021-11-02T01:29:58.000+00:00",
+ "finish": "NO"
}
}
```
-#### 7.查看任务错误日志
+
+#### 8.查看当前流程当前安装任务状态
**接口功能**
-> 查看任务错误日志
+> 查看当前流程当前安装任务状态
**URL**
-> /agent/errlog
+> /api/process/{processId}/currentTasks
**支持格式**
@@ -451,9 +487,7 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
> |参数|必选|类型|说明|
> |:----- |:-------|:-----|----- |
-> |host |ture | String |agent 机器host |
-> |taskId |true |String |任务id|
-> |offset |false |int |日志偏移量|
+> |processId |ture | String |流程ID，接口1返回 |
**返回字段**
@@ -461,32 +495,240 @@ Direct access after local service startup:http://localhost:port/swagger-ui.html#
> |:----- |:------|:----------------------------- |
> |msg |String |调用信息 |
> |code |String | 结果状态。0：正常 |
-> |data.key |String | 下一页起始偏移量 |
-> |data.value |List | 日志 |
+> |data.id |int | 任务ID |
+> |data.processId |int | 流程id |
+> |data.host |String | 任务运行host |
+> |data.processType |String | 流程类型 |
+> |data.taskType |String | 任务类型 |
+> |data.status |String | 任务执行状态 |
+> |data.startTime |Date | 任务开始时间 |
+> |data.endTime |Date | 任务终止时间 |
+> |data.finish |int | 任务执行完成标志 |
**接口示例**
-> 地址：http://localhost:9601/agent/errlog
+> 地址：http://localhost:9601/api/process/1/currentTasks
> 请求参数：
+无
+> 返回参数：
``` json
{
- "host":"10.10.10.11",
- "taskId":"a3b966cc10794f4c8b9b2c0dc1fa9b26",
- "offset":10,
+ "msg": "success",
+ "code": 0,
+ "data": [
+ {
+ "id": 6,
+ "processId": 1,
+ "host": "10.220.147.155",
+ "processType": "START_SERVICE",
+ "taskType": "START_FE",
+ "status": "SUCCESS",
+ "startTime": "2021-11-02T01:33:08.000+00:00",
+ "endTime": "2021-11-02T01:33:20.000+00:00",
+ "executorId": "377bb55156774cb7a72804fbba207e94",
+ "result": null,
+ "finish": "YES"
+ },
+ {
+ "id": 7,
+ "processId": 1,
+ "host": "10.220.147.155",
+ "processType": "START_SERVICE",
+ "taskType": "START_BE",
+ "status": "SUCCESS",
+ "startTime": "2021-11-02T01:33:08.000+00:00",
+ "endTime": "2021-11-02T01:33:20.000+00:00",
+ "executorId": "f052ba23ad9d4428900d328a7979d7a2",
+ "result": null,
+ "finish": "YES"
+ }
+ ]
}
```
+
+#### 9.查看任务执行信息
+
+**接口功能**
+
+> 查看任务执行信息
+
+**URL**
+
+> /api/process/task/info/{taskId}
+
+**支持格式**
+
+> JSON
+
+**HTTP请求方式**
+
+> GET
+
+**请求参数**
+
+> |参数|必选|类型|说明|
+> |:----- |:-------|:-----|----- |
+> |taskId |true |String |任务id|
+
+**返回字段**
+
+> |返回字段|字段类型|说明 |
+> |:----- |:------|:----------------------------- |
+> |msg |String |调用信息 |
+> |code |String | 结果状态。0：正常 |
+> |data.id |int | 任务ID |
+> |data.processId |int | 任务所属流程ID |
+> |data.host |Date | 任务执行host |
+> |data.processType |String | 安装类型 |
+> |data.taskType |String | 任务类型 |
+> |data.status |int | 任务返回状态。 |
+> |data.startTime |Date | 任务开始时间 |
+> |data.endTime |Date | 任务结束时间 |
+> |data.result |String | 任务返回结果 |
+> |data.finish |String | 任务是否执行成功。YES NO|
+
+**接口示例**
+
+> 地址：http://localhost:9601/api/process/task/info/9
+
+```
> 返回参数：
-``` json
+``` json
+{
+ "msg": "success",
+ "code": 0,
+ "data": {
+ "id": 1,
+ "processId": 1,
+ "host": "10.220.147.155",
+ "processType": "INSTALL_AGENT",
+ "taskType": "INSTALL_AGENT",
+ "status": "SUCCESS",
+ "startTime": "2021-11-05T10:33:23.000+00:00",
+ "endTime": "2021-11-05T10:33:26.000+00:00",
+ "executorId": null,
+ "result": "SUCCESS",
+ "finish": "YES"
+ }
+}
+```
+
+
+#### 10.查看任务执行日志
+
+**接口功能**
+
+> 查看任务最近1MB日志
+
+**URL**
+
+> /api/process/task/log/{taskId}
+
+**支持格式**
+
+> JSON
+
+**HTTP请求方式**
+
+> GET
+
+**请求参数**
+
+> |参数|必选|类型|说明|
+> |:----- |:-------|:-----|----- |
+> |taskId |true |String |任务id|
+
+**返回字段**
+
+> |返回字段|字段类型|说明 |
+> |:----- |:------|:----------------------------- |
+> |msg |String |调用信息 |
+> |code |String | 结果状态。0：正常 |
+> |data.logPath |String | 任务日志路径 |
+> |data.log |String | 日志内容 |
+
+**接口示例**
+
+> 地址：http://localhost:9601/api/process/task/log/9
+
+```
+> 返回参数：
+``` json
{
"msg": "success",
"code": 0,
"data": {
- "key": 366,
- "value": [
- .....
- ]
+ "logPath": "/usr/local/doris/agent/log/task.log",
+ "log": "2021-11-05 19:01:38.434 [pool-2-thread-1] INFO TASK_LOG - scriptCmd:/bin/sh /usr/local/doris/agent/bin/install_fe.sh --installDir /usr"
}
}
```
+#### 11.查看Agent上安装的角色列表
+
+**接口功能**
+
+> 查看Agent上安装的角色列表
+
+**URL**
+
+> /api/server/roleList
+
+**支持格式**
+
+> JSON
+
+**HTTP请求方式**
+
+> GET
+
+**请求参数**
+> |返回字段|字段类型|说明 |
+> |:----- |:------|:----------------------------- |
+> |clusterId |int |集群id |
+
+**返回字段**
+
+> |返回字段|字段类型|说明 |
+> |:----- |:------|:----------------------------- |
+> |msg |String |调用信息 |
+> |code |String | 结果状态。0：正常 |
+> |data.host |String |agent host |
+> |data.role |String |安装角色 FE BE |
+> |data.feNodeType |String | 角色类型 FOLLOWer OBserveer|
+> |data.register |String | 安装后是否注册成功 |
+
+**接口示例**
+
+> 地址：http://localhost:9601/api/server/agentList
+
+> 请求参数：无
+
+> 返回参数：
+``` json
+{
+ "msg": "success",
+ "code": 0,
+ "data": [
+ {
+ "id": 1,
+ "host": "10.220.147.155",
+ "clusterId": 0,
+ "role": "FE",
+ "feNodeType": null,
+ "installDir": "/usr/local/doris/fe",
+ "register": "YES"
+ },
+ {
+ "id": 2,
+ "host": "10.220.147.155",
+ "clusterId": 0,
+ "role": "BE",
+ "feNodeType": null,
+ "installDir": "/usr/local/doris/be",
+ "register": "YES"
+ }
+ ]
+}
+```
\ No newline at end of file
diff --git a/manager/manager-bin/agent/bin/agent_start.sh b/manager/manager-bin/agent/bin/agent_start.sh
index ff3bbfa..c75a796 100644
--- a/manager/manager-bin/agent/bin/agent_start.sh
+++ b/manager/manager-bin/agent/bin/agent_start.sh
@@ -16,6 +16,9 @@
# specific language governing permissions and limitations
# under the License.
+source /etc/profile
+source ~/.bash_profile
+
curdir=`dirname "$0"`
curdir=`cd "$curdir"; pwd`
diff --git a/manager/manager-bin/agent/bin/agent_stop.sh b/manager/manager-bin/agent/bin/agent_stop.sh
index e651584..237f686 100644
--- a/manager/manager-bin/agent/bin/agent_stop.sh
+++ b/manager/manager-bin/agent/bin/agent_stop.sh
@@ -16,6 +16,9 @@
# specific language governing permissions and limitations
# under the License.
+source /etc/profile
+source ~/.bash_profile
+
curdir=`dirname "$0"`
curdir=`cd "$curdir"; pwd`
diff --git a/manager/manager-bin/agent/bin/install_be.sh b/manager/manager-bin/agent/bin/install_be.sh
index d0e970a..b8f6adb 100644
--- a/manager/manager-bin/agent/bin/install_be.sh
+++ b/manager/manager-bin/agent/bin/install_be.sh
@@ -16,6 +16,9 @@
# specific language governing permissions and limitations
# under the License.
+source /etc/profile
+source ~/.bash_profile
+
curdir=`dirname "$0"`
curdir=`cd "$curdir"; pwd`
@@ -29,17 +32,17 @@ OPTS=$(getopt \
eval set -- "$OPTS"
URL=
-DORIS_HOME=
+DORIS_BE_HOME=
while true; do
case "$1" in
--url) URL=$2 ; shift 2;;
- --installDir) DORIS_HOME=$2 ; shift 2;;
+ --installDir) DORIS_BE_HOME=$2 ; shift 2;;
--) shift ; break ;;
*) echo "Internal error" ; exit 1 ;;
esac
done
-if [ x"$DORIS_HOME" == x"" ]; then
+if [ x"$DORIS_BE_HOME" == x"" ]; then
echo "--installDir can not empty!"
exit 1
fi
@@ -49,12 +52,27 @@ if [ x"$URL" == x"" ]; then
exit 1
fi
-if [ ! -d $DORIS_HOME ]; then
- mkdir -p $DORIS_HOME
+if [ ! -d $DORIS_BE_HOME ]; then
+ mkdir -p $DORIS_BE_HOME
fi
+
+cd $DORIS_BE_HOME
+rm -rf doris-be.tar.gz
+if [ $? -ne 0 ] ;then exit 1;fi
+
wget $URL -O doris-be.tar.gz --quiet
if [ $? -ne 0 ] ;then exit 1;fi
-tar -zxvf doris-be.tar.gz -C $DORIS_HOME
+rm -rf $DORIS_BE_HOME/betmp
+mkdir -p $DORIS_BE_HOME/betmp
+if [ $? -ne 0 ] ;then exit 1;fi
+
+tar -zxvf doris-be.tar.gz --strip-components=1 -C $DORIS_BE_HOME/betmp
+if [ $? -ne 0 ] ;then exit 1;fi
+
+mv $DORIS_BE_HOME/betmp/be/* $DORIS_BE_HOME/
+if [ $? -ne 0 ] ;then exit 1;fi
+
+rm -rf $DORIS_BE_HOME/betmp
if [ $? -ne 0 ] ;then exit 1;fi
\ No newline at end of file
diff --git a/manager/manager-bin/agent/bin/install_fe.sh b/manager/manager-bin/agent/bin/install_fe.sh
index ec54d52..b60c598 100644
--- a/manager/manager-bin/agent/bin/install_fe.sh
+++ b/manager/manager-bin/agent/bin/install_fe.sh
@@ -16,6 +16,9 @@
# specific language governing permissions and limitations
# under the License.
+source /etc/profile
+source ~/.bash_profile
+
curdir=`dirname "$0"`
curdir=`cd "$curdir"; pwd`
@@ -29,17 +32,17 @@ OPTS=$(getopt \
eval set -- "$OPTS"
URL=
-DORIS_HOME=
+DORIS_FE_HOME=
while true; do
case "$1" in
--url) URL=$2 ; shift 2;;
- --installDir) DORIS_HOME=$2 ; shift 2;;
+ --installDir) DORIS_FE_HOME=$2 ; shift 2;;
--) shift ; break ;;
*) echo "Internal error" ; exit 1 ;;
esac
done
-if [ x"$DORIS_HOME" == x"" ]; then
+if [ x"$DORIS_FE_HOME" == x"" ]; then
echo "--installDir can not empty!"
exit 1
fi
@@ -49,12 +52,26 @@ if [ x"$URL" == x"" ]; then
exit 1
fi
-if [ ! -d $DORIS_HOME ]; then
- mkdir -p $DORIS_HOME
+if [ ! -d $DORIS_FE_HOME ]; then
+ mkdir -p $DORIS_FE_HOME
fi
+cd $DORIS_FE_HOME
+rm -rf doris-fe.tar.gz
+if [ $? -ne 0 ] ;then exit 1;fi
+
wget $URL -O doris-fe.tar.gz --quiet
if [ $? -ne 0 ] ;then exit 1;fi
-tar -zxvf doris-fe.tar.gz -C $DORIS_HOME
+rm -rf $DORIS_FE_HOME/fetmp
+mkdir -p $DORIS_FE_HOME/fetmp
+if [ $? -ne 0 ] ;then exit 1;fi
+
+tar -zxvf doris-fe.tar.gz --strip-components=1 -C $DORIS_FE_HOME/fetmp
+if [ $? -ne 0 ] ;then exit 1;fi
+
+mv $DORIS_FE_HOME/fetmp/fe/* $DORIS_FE_HOME/
+if [ $? -ne 0 ] ;then exit 1;fi
+
+rm -rf $DORIS_FE_HOME/fetmp
if [ $? -ne 0 ] ;then exit 1;fi
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org