8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2025-12-24 11:52:08 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into v2.2.1_ldap

Browse Source
This commit is contained in:
李民
2021-02-10 10:00:32 +08:00
committed by GitHub
parent e7349161f3 ec7f801929
commit f70f4348b3
147 changed files with 3080 additions and 875 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/AccountService.java
View File

@@ -2,6 +2,7 @@ package com.xiaojukeji.kafka.manager.account;
import com.xiaojukeji.kafka.manager.account.common.EnterpriseStaff;
import com.xiaojukeji.kafka.manager.common.bizenum.AccountRoleEnum;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.entity.ao.account.Account;
import com.xiaojukeji.kafka.manager.common.entity.pojo.AccountDO;
@@ -25,14 +26,14 @@ public interface AccountService {
* @param username 用户名
* @return
*/
AccountDO getAccountDO(String username);
Result<AccountDO> getAccountDO(String username);
/**
* 删除用户
* @param username 用户名
* @return
*/
ResultStatus deleteByName(String username);
ResultStatus deleteByName(String username, String operator);
/**
* 更新账号

3
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/LoginService.java
View File

@@ -1,5 +1,6 @@
package com.xiaojukeji.kafka.manager.account;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ao.account.Account;
import com.xiaojukeji.kafka.manager.common.entity.dto.normal.LoginDTO;
@@ -11,7 +12,7 @@ import javax.servlet.http.HttpServletResponse;
* @date 20/8/20
*/
public interface LoginService {
Account login(HttpServletRequest request, HttpServletResponse response, LoginDTO dto);
Result<Account> login(HttpServletRequest request, HttpServletResponse response, LoginDTO dto);
void logout(HttpServletRequest request, HttpServletResponse response, Boolean needJump2LoginPage);

3
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/component/AbstractSingleSignOn.java
View File

@@ -1,5 +1,6 @@
package com.xiaojukeji.kafka.manager.account.component;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.dto.normal.LoginDTO;
import javax.servlet.http.HttpServletRequest;
@@ -18,7 +19,7 @@ public abstract class AbstractSingleSignOn {
protected static final String HEADER_REDIRECT_KEY = "location";
public abstract String loginAndGetLdap(HttpServletRequest request, HttpServletResponse response, LoginDTO dto);
public abstract Result<String> loginAndGetLdap(HttpServletRequest request, HttpServletResponse response, LoginDTO dto);
public abstract void logout(HttpServletRequest request, HttpServletResponse response, Boolean needJump2LoginPage);

9
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/component/account/BaseEnterpriseStaffService.java
View File

@@ -41,7 +41,14 @@ public class BaseEnterpriseStaffService extends AbstractEnterpriseStaffService {
@Override
public List<EnterpriseStaff> searchEnterpriseStaffByKeyWord(String keyWord) {
try {
List<AccountDO> doList = accountDao.searchByNamePrefix(keyWord);
List<AccountDO> doList = null;
if (ValidateUtils.isBlank(keyWord)) {
// 当用户没有任何输入的时候, 返回全部的用户
doList = accountDao.list();
} else {
doList = accountDao.searchByNamePrefix(keyWord);
}
if (ValidateUtils.isEmptyList(doList)) {
return new ArrayList<>();
}

29
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/component/sso/BaseSessionSignOn.java
View File

@@ -4,6 +4,7 @@ import com.xiaojukeji.kafka.manager.account.AccountService;
import com.xiaojukeji.kafka.manager.account.component.AbstractSingleSignOn;
import com.xiaojukeji.kafka.manager.common.bizenum.AccountRoleEnum;
import com.xiaojukeji.kafka.manager.common.constant.LoginConstant;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.dto.normal.LoginDTO;
import com.xiaojukeji.kafka.manager.common.entity.pojo.AccountDO;
import com.xiaojukeji.kafka.manager.common.utils.EncryptUtil;
@@ -41,42 +42,44 @@ public class BaseSessionSignOn extends AbstractSingleSignOn {
private boolean authUserRegistration;
@Override
public String loginAndGetLdap(HttpServletRequest request, HttpServletResponse response, LoginDTO dto) {
public Result<String> loginAndGetLdap(HttpServletRequest request, HttpServletResponse response, LoginDTO dto) {
if (ValidateUtils.isBlank(dto.getUsername()) || ValidateUtils.isNull(dto.getPassword())) {
return null;
}
AccountDO accountDO = accountService.getAccountDO(dto.getUsername());
Result<AccountDO> accountResult = accountService.getAccountDO(dto.getUsername());
//modifier limin
//判断是否激活了LDAP验证。若激活并且数据库无此用户则自动注册
if(ldapEnabled){
//验证账密
//去LDAP验证账密
if(!ldapAuthentication.authenricate(dto.getUsername(),dto.getPassword())){
return null;
}
if(accountDO==null && authUserRegistration){
if(ValidateUtils.isNull(accountResult) && authUserRegistration){
//自动注册
accountDO = new AccountDO();
AccountDO accountDO = new AccountDO();
accountDO.setUsername(dto.getUsername());
accountDO.setRole(AccountRoleEnum.getUserRoleEnum(authUserRegistrationRole).getRole());
accountDO.setPassword(EncryptUtil.md5(dto.getPassword()));
accountService.createAccount(accountDO);
return dto.getUsername();
}
return dto.getUsername();
return Result.buildSuc(dto.getUsername());
}
if (ValidateUtils.isNull(accountDO)) {
return null;
if (ValidateUtils.isNull(accountResult) || accountResult.failed()) {
return new Result<>(accountResult.getCode(), accountResult.getMessage());
}
if (!accountDO.getPassword().equals(EncryptUtil.md5(dto.getPassword()))) {
return null;
if (ValidateUtils.isNull(accountResult.getData())) {
return Result.buildFailure("username illegal");
}
return dto.getUsername();
if (!accountResult.getData().getPassword().equals(EncryptUtil.md5(dto.getPassword()))) {
return Result.buildFailure("password illegal");
}
return Result.buildSuc(accountResult.getData().getUsername());
}
@Override

23
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/impl/AccountServiceImpl.java
View File

@@ -6,7 +6,10 @@ import com.xiaojukeji.kafka.manager.account.AccountService;
import com.xiaojukeji.kafka.manager.account.common.EnterpriseStaff;
import com.xiaojukeji.kafka.manager.account.component.AbstractEnterpriseStaffService;
import com.xiaojukeji.kafka.manager.common.bizenum.AccountRoleEnum;
import com.xiaojukeji.kafka.manager.common.bizenum.ModuleEnum;
import com.xiaojukeji.kafka.manager.common.bizenum.OperateEnum;
import com.xiaojukeji.kafka.manager.common.constant.Constant;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.entity.ao.account.Account;
import com.xiaojukeji.kafka.manager.common.entity.pojo.AccountDO;
@@ -14,6 +17,7 @@ import com.xiaojukeji.kafka.manager.common.utils.EncryptUtil;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.dao.AccountDao;
import com.xiaojukeji.kafka.manager.service.service.ConfigService;
import com.xiaojukeji.kafka.manager.service.service.OperateRecordService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -47,6 +51,9 @@ public class AccountServiceImpl implements AccountService {
@Autowired
private AbstractEnterpriseStaffService enterpriseStaffService;
@Autowired
private OperateRecordService operateRecordService;
/**
* 用户组织信息
* <username, Staff>
@@ -81,9 +88,12 @@ public class AccountServiceImpl implements AccountService {
}
@Override
public ResultStatus deleteByName(String username) {
public ResultStatus deleteByName(String username, String operator) {
try {
if (accountDao.deleteByName(username) > 0) {
Map<String, String> content = new HashMap<>();
content.put("username", username);
operateRecordService.insert(operator, ModuleEnum.AUTHORITY, username, OperateEnum.DELETE, content);
return ResultStatus.SUCCESS;
}
} catch (Exception e) {
@@ -101,7 +111,7 @@ public class AccountServiceImpl implements AccountService {
return ResultStatus.ACCOUNT_NOT_EXIST;
}
if (!ValidateUtils.isNull(accountDO.getPassword())) {
if (!ValidateUtils.isBlank(accountDO.getPassword())) {
accountDO.setPassword(EncryptUtil.md5(accountDO.getPassword()));
} else {
accountDO.setPassword(oldAccountDO.getPassword());
@@ -117,8 +127,13 @@ public class AccountServiceImpl implements AccountService {
}
@Override
public AccountDO getAccountDO(String username) {
return accountDao.getByName(username);
public Result<AccountDO> getAccountDO(String username) {
try {
return Result.buildSuc(accountDao.getByName(username));
} catch (Exception e) {
LOGGER.warn("class=AccountServiceImpl||method=getAccountDO||username={}||errMsg={}||msg=get account fail", username, e.getMessage());
}
return Result.buildFrom(ResultStatus.MYSQL_ERROR);
}
@Override

13
kafka-manager-extends/kafka-manager-account/src/main/java/com/xiaojukeji/kafka/manager/account/impl/LoginServiceImpl.java
View File

@@ -6,6 +6,7 @@ import com.xiaojukeji.kafka.manager.account.LoginService;
import com.xiaojukeji.kafka.manager.common.bizenum.AccountRoleEnum;
import com.xiaojukeji.kafka.manager.common.constant.ApiPrefix;
import com.xiaojukeji.kafka.manager.common.constant.LoginConstant;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ao.account.Account;
import com.xiaojukeji.kafka.manager.common.entity.dto.normal.LoginDTO;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
@@ -34,15 +35,15 @@ public class LoginServiceImpl implements LoginService {
private AbstractSingleSignOn singleSignOn;
@Override
public Account login(HttpServletRequest request, HttpServletResponse response, LoginDTO loginDTO) {
String username = singleSignOn.loginAndGetLdap(request, response, loginDTO);
if (ValidateUtils.isBlank(username)) {
public Result<Account> login(HttpServletRequest request, HttpServletResponse response, LoginDTO loginDTO) {
Result<String> userResult = singleSignOn.loginAndGetLdap(request, response, loginDTO);
if (ValidateUtils.isNull(userResult) || userResult.failed()) {
logout(request, response, false);
return null;
return new Result<>(userResult.getCode(), userResult.getMessage());
}
Account account = accountService.getAccountFromCache(username);
Account account = accountService.getAccountFromCache(userResult.getData());
initLoginContext(request, response, account);
return account;
return Result.buildSuc(account);
}
private void initLoginContext(HttpServletRequest request, HttpServletResponse response, Account account) {

12
kafka-manager-extends/kafka-manager-bpm/src/main/java/com/xiaojukeji/kafka/manager/bpm/common/entry/apply/gateway/OrderExtensionAddGatewayConfigDTO.java
View File

@@ -18,6 +18,9 @@ public class OrderExtensionAddGatewayConfigDTO {
@ApiModelProperty(value = "值")
private String value;
@ApiModelProperty(value = "描述说明")
private String description;
public String getType() {
return type;
}
@@ -42,12 +45,21 @@ public class OrderExtensionAddGatewayConfigDTO {
this.value = value;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
@Override
public String toString() {
return "OrderExtensionAddGatewayConfigDTO{" +
"type='" + type + '\'' +
", name='" + name + '\'' +
", value='" + value + '\'' +
", description='" + description + '\'' +
'}';
}

12
kafka-manager-extends/kafka-manager-bpm/src/main/java/com/xiaojukeji/kafka/manager/bpm/common/entry/apply/gateway/OrderExtensionModifyGatewayConfigDTO.java
View File

@@ -23,6 +23,9 @@ public class OrderExtensionModifyGatewayConfigDTO {
@ApiModelProperty(value = "值")
private String value;
@ApiModelProperty(value = "描述说明")
private String description;
public Long getId() {
return id;
}
@@ -55,6 +58,14 @@ public class OrderExtensionModifyGatewayConfigDTO {
this.value = value;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
@Override
public String toString() {
return "OrderExtensionModifyGatewayConfigDTO{" +
@@ -62,6 +73,7 @@ public class OrderExtensionModifyGatewayConfigDTO {
", type='" + type + '\'' +
", name='" + name + '\'' +
", value='" + value + '\'' +
", description='" + description + '\'' +
'}';
}

2
kafka-manager-extends/kafka-manager-bpm/src/main/java/com/xiaojukeji/kafka/manager/bpm/order/impl/ApplyAppOrder.java
View File

@@ -87,6 +87,6 @@ public class ApplyAppOrder extends AbstractAppOrder {
appDO.setDescription(orderDO.getDescription());
appDO.generateAppIdAndPassword(orderDO.getId(), configUtils.getIdc());
appDO.setType(0);
return appService.addApp(appDO);
return appService.addApp(appDO, userName);
}
}

2
kafka-manager-extends/kafka-manager-bpm/src/main/java/com/xiaojukeji/kafka/manager/bpm/order/impl/ApplyAuthorityOrder.java
View File

@@ -95,7 +95,7 @@ public class ApplyAuthorityOrder extends AbstractAuthorityOrder {
}
TopicDO topicDO = topicManagerService.getByTopicName(physicalClusterId, orderExtensionDTO.getTopicName());
if (ValidateUtils.isNull(topicDO)) {
return ResultStatus.TOPIC_NOT_EXIST;
return ResultStatus.TOPIC_BIZ_DATA_NOT_EXIST;
}
AppDO appDO = appService.getByAppId(topicDO.getAppId());
if (!appDO.getPrincipals().contains(userName)) {

5
kafka-manager-extends/kafka-manager-kcm/pom.xml
View File

@@ -68,5 +68,10 @@
<artifactId>spring-test</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>io.minio</groupId>
<artifactId>minio</artifactId>
</dependency>
</dependencies>
</project>

3
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/KafkaFileService.java
View File

@@ -4,6 +4,7 @@ import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.entity.dto.normal.KafkaFileDTO;
import com.xiaojukeji.kafka.manager.common.entity.pojo.KafkaFileDO;
import org.springframework.web.multipart.MultipartFile;
import java.util.List;
@@ -24,7 +25,7 @@ public interface KafkaFileService {
KafkaFileDO getFileByFileName(String fileName);
Result<String> downloadKafkaConfigFile(Long fileId);
Result<MultipartFile> downloadKafkaFile(Long fileId);
String getDownloadBaseUrl();
}

18
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/common/Constant.java Normal file
View File

@@ -0,0 +1,18 @@
package com.xiaojukeji.kafka.manager.kcm.common;
public class Constant {
/**
*
*/
public static final String TASK_TITLE_PREFIX = "Logi-Kafka";
/**
* 并发度，顺序执行
*/
public static final Integer AGENT_TASK_BATCH = 1;
/**
* 失败的容忍度为0
*/
public static final Integer AGENT_TASK_TOLERANCE = 0;
}

39
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/common/bizenum/ClusterTaskActionEnum.java
View File

@@ -6,34 +6,35 @@ package com.xiaojukeji.kafka.manager.kcm.common.bizenum;
* @date 20/4/26
*/
public enum ClusterTaskActionEnum {
START(0, "start"),
PAUSE(1, "pause"),
IGNORE(2, "ignore"),
CANCEL(3, "cancel"),
ROLLBACK(4, "rollback"),
UNKNOWN("unknown"),
START("start"),
PAUSE("pause"),
IGNORE("ignore"),
CANCEL("cancel"),
REDO("redo"),
KILL("kill"),
ROLLBACK("rollback"),
;
private Integer code;
private String message;
private String action;
ClusterTaskActionEnum(Integer code, String message) {
this.code = code;
this.message = message;
ClusterTaskActionEnum(String action) {
this.action = action;
}
public Integer getCode() {
return code;
}
public String getMessage() {
return message;
public String getAction() {
return action;
}
@Override
public String toString() {
return "TaskActionEnum{" +
"code=" + code +
", message='" + message + '\'' +
return "ClusterTaskActionEnum{" +
"action='" + action + '\'' +
'}';
}
}

24
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/common/entry/ao/ClusterTaskLog.java Normal file
View File

@@ -0,0 +1,24 @@
package com.xiaojukeji.kafka.manager.kcm.common.entry.ao;
public class ClusterTaskLog {
private String stdout;
public ClusterTaskLog(String stdout) {
this.stdout = stdout;
}
public String getStdout() {
return stdout;
}
public void setStdout(String stdout) {
this.stdout = stdout;
}
@Override
public String toString() {
return "AgentOperationTaskLog{" +
"stdout='" + stdout + '\'' +
'}';
}
}

4
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/common/entry/ao/CreationTaskData.java
View File

@@ -1,5 +1,7 @@
package com.xiaojukeji.kafka.manager.kcm.common.entry.ao;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import java.util.List;
/**
@@ -119,7 +121,7 @@ public class CreationTaskData {
@Override
public String toString() {
return "CreationTaskDTO{" +
return "CreationTaskData{" +
"uuid='" + uuid + '\'' +
", clusterId=" + clusterId +
", hostList=" + hostList +

75
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/component/agent/AbstractAgent.java
View File

@@ -1,9 +1,18 @@
package com.xiaojukeji.kafka.manager.kcm.component.agent;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskActionEnum;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskStateEnum;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskSubStateEnum;
import com.xiaojukeji.kafka.manager.kcm.common.entry.ao.ClusterTaskLog;
import com.xiaojukeji.kafka.manager.kcm.common.entry.ao.CreationTaskData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.Map;
@@ -13,33 +22,79 @@ import java.util.Map;
* @date 20/4/26
*/
public abstract class AbstractAgent {
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAgent.class);
/**
* 创建任务
* @param creationTaskData 创建任务参数
* @return 任务ID
*/
public abstract Long createTask(CreationTaskData dto);
public abstract Result<Long> createTask(CreationTaskData creationTaskData);
/**
* 任务动作
* 执行任务
* @param taskId 任务ID
* @param actionEnum 执行动作
* @return true:触发成功, false:触发失败
*/
public abstract Boolean actionTask(Long taskId, String action);
public abstract boolean actionTask(Long taskId, ClusterTaskActionEnum actionEnum);
/**
* 任务动作
* 执行任务
* @param taskId 任务ID
* @param actionEnum 执行动作
* @param hostname 具体主机
* @return true:触发成功, false:触发失败
*/
public abstract Boolean actionHostTask(Long taskId, String action, String hostname);
public abstract boolean actionHostTask(Long taskId, ClusterTaskActionEnum actionEnum, String hostname);
/**
* 获取任务状态
* 获取任务运行的状态[阻塞, 执行中, 完成等]
* @param taskId 任务ID
* @return 任务状态
*/
public abstract ClusterTaskStateEnum getTaskState(Long agentTaskId);
public abstract Result<ClusterTaskStateEnum> getTaskExecuteState(Long taskId);
/**
* 获取任务结果
* @param taskId 任务ID
* @return 任务结果
*/
public abstract Map<String, ClusterTaskSubStateEnum> getTaskResult(Long taskId);
public abstract Result<Map<String, ClusterTaskSubStateEnum>> getTaskResult(Long taskId);
/**
* 获取任务日志
* 获取任务执行日志
* @param taskId 任务ID
* @param hostname 具体主机
* @return 机器运行日志
*/
public abstract String getTaskLog(Long agentTaskId, String hostname);
public abstract Result<ClusterTaskLog> getTaskLog(Long taskId, String hostname);
protected static String readScriptInJarFile(String fileName) {
InputStream inputStream = AbstractAgent.class.getClassLoader().getResourceAsStream(fileName);
if (inputStream == null) {
LOGGER.error("class=AbstractAgent||method=readScriptInJarFile||fileName={}||msg=read script failed", fileName);
return "";
}
try {
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
String line = null;
StringBuilder sb = new StringBuilder();
while ((line = bufferedReader.readLine()) != null) {
sb.append(line).append("\n");
}
return sb.toString();
} catch (Exception e) {
LOGGER.error("class=AbstractAgent||method=readScriptInJarFile||fileName={}||errMsg={}||msg=read script failed", fileName, e.getMessage());
} finally {
try {
inputStream.close();
} catch (IOException e) {
LOGGER.error("class=AbstractAgent||method=readScriptInJarFile||fileName={}||errMsg={}||msg=close reading script failed", fileName, e.getMessage());
}
}
return "";
}
}

222
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/component/agent/n9e/N9e.java
View File

@@ -1,8 +1,11 @@
package com.xiaojukeji.kafka.manager.kcm.component.agent.n9e;
import com.alibaba.fastjson.JSON;
import com.xiaojukeji.kafka.manager.common.bizenum.KafkaFileEnum;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.kcm.common.Constant;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskActionEnum;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskTypeEnum;
import com.xiaojukeji.kafka.manager.kcm.common.entry.ao.ClusterTaskLog;
import com.xiaojukeji.kafka.manager.kcm.common.entry.ao.CreationTaskData;
import com.xiaojukeji.kafka.manager.common.utils.HttpUtils;
import com.xiaojukeji.kafka.manager.common.utils.JsonUtils;
@@ -11,20 +14,17 @@ import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskStateEnum;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskSubStateEnum;
import com.xiaojukeji.kafka.manager.kcm.component.agent.AbstractAgent;
import com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry.N9eCreationTask;
import com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry.N9eResult;
import com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry.N9eTaskResultDTO;
import com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry.N9eTaskStatusEnum;
import com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry.N9eTaskStdoutDTO;
import com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry.N9eTaskResult;
import com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry.N9eTaskStdoutLog;
import com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry.bizenum.N9eTaskStatusEnum;
import org.springframework.beans.factory.annotation.Value;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -54,16 +54,6 @@ public class N9e extends AbstractAgent {
private String script;
/**
* 并发度，顺序执行
*/
private static final Integer BATCH = 1;
/**
* 失败的容忍度为0
*/
private static final Integer TOLERANCE = 0;
private static final String CREATE_TASK_URI = "/api/job-ce/tasks";
private static final String ACTION_TASK_URI = "/api/job-ce/task/{taskId}/action";
@@ -82,143 +72,134 @@ public class N9e extends AbstractAgent {
}
@Override
public Long createTask(CreationTaskData creationTaskData) {
Map<String, Object> param = buildCreateTaskParam(creationTaskData);
public Result<Long> createTask(CreationTaskData creationTaskData) {
String content = JsonUtils.toJSONString(buildCreateTaskParam(creationTaskData));
String response = null;
try {
response = HttpUtils.postForString(
baseUrl + CREATE_TASK_URI,
JsonUtils.toJSONString(param),
buildHeader()
);
N9eResult zr = JSON.parseObject(response, N9eResult.class);
if (!ValidateUtils.isBlank(zr.getErr())) {
LOGGER.warn("class=N9e||method=createTask||param={}||errMsg={}||msg=call create task fail", JsonUtils.toJSONString(param),zr.getErr());
return null;
response = HttpUtils.postForString(baseUrl + CREATE_TASK_URI, content, buildHeader());
N9eResult nr = JsonUtils.stringToObj(response, N9eResult.class);
if (!ValidateUtils.isBlank(nr.getErr())) {
LOGGER.error("class=N9e||method=createTask||param={}||response={}||msg=call create task failed", content, response);
return Result.buildFailure(nr.getErr());
}
return Long.valueOf(zr.getDat().toString());
return Result.buildSuc(Long.valueOf(nr.getDat().toString()));
} catch (Exception e) {
LOGGER.error("create task failed, req:{}.", creationTaskData, e);
LOGGER.error("class=N9e||method=createTask||param={}||response={}||errMsg={}||msg=call create task failed", content, response, e.getMessage());
}
return null;
return Result.buildFailure("create n9e task failed");
}
@Override
public Boolean actionTask(Long taskId, String action) {
public boolean actionTask(Long taskId, ClusterTaskActionEnum actionEnum) {
Map<String, Object> param = new HashMap<>(1);
param.put("action", action);
param.put("action", actionEnum.getAction());
String response = null;
try {
response = HttpUtils.putForString(
baseUrl + ACTION_TASK_URI.replace("{taskId}", taskId.toString()),
JSON.toJSONString(param),
buildHeader()
);
N9eResult zr = JSON.parseObject(response, N9eResult.class);
if (ValidateUtils.isBlank(zr.getErr())) {
response = HttpUtils.putForString(baseUrl + ACTION_TASK_URI.replace("{taskId}", String.valueOf(taskId)), JsonUtils.toJSONString(param), buildHeader());
N9eResult nr = JsonUtils.stringToObj(response, N9eResult.class);
if (ValidateUtils.isBlank(nr.getErr())) {
return true;
}
LOGGER.warn("class=N9e||method=actionTask||param={}||errMsg={}||msg=call action task fail", JSON.toJSONString(param),zr.getErr());
LOGGER.error("class=N9e||method=actionTask||param={}||response={}||msg=call action task fail", JsonUtils.toJSONString(param), response);
return false;
} catch (Exception e) {
LOGGER.error("action task failed, taskId:{}, action:{}.", taskId, action, e);
LOGGER.error("class=N9e||method=actionTask||param={}||response={}||errMsg={}||msg=call action task fail", JsonUtils.toJSONString(param), response, e.getMessage());
}
return false;
}
@Override
public Boolean actionHostTask(Long taskId, String action, String hostname) {
Map<String, Object> param = new HashMap<>(2);
param.put("action", action);
param.put("hostname", hostname);
public boolean actionHostTask(Long taskId, ClusterTaskActionEnum actionEnum, String hostname) {
Map<String, Object> params = new HashMap<>(2);
params.put("action", actionEnum.getAction());
params.put("hostname", hostname);
String response = null;
try {
response = HttpUtils.putForString(
baseUrl + ACTION_HOST_TASK_URI.replace("{taskId}", taskId.toString()),
JSON.toJSONString(param),
buildHeader()
);
N9eResult zr = JSON.parseObject(response, N9eResult.class);
if (ValidateUtils.isBlank(zr.getErr())) {
response = HttpUtils.putForString(baseUrl + ACTION_HOST_TASK_URI.replace("{taskId}", String.valueOf(taskId)), JsonUtils.toJSONString(params), buildHeader());
N9eResult nr = JsonUtils.stringToObj(response, N9eResult.class);
if (ValidateUtils.isBlank(nr.getErr())) {
return true;
}
LOGGER.warn("class=N9e||method=actionHostTask||param={}||errMsg={}||msg=call action host task fail", JSON.toJSONString(param),zr.getErr());
LOGGER.error("class=N9e||method=actionHostTask||params={}||response={}||msg=call action host task fail", JsonUtils.toJSONString(params), response);
return false;
} catch (Exception e) {
LOGGER.error("action task failed, taskId:{} action:{} hostname:{}.", taskId, action, hostname, e);
LOGGER.error("class=N9e||method=actionHostTask||params={}||response={}||errMsg={}||msg=call action host task fail", JsonUtils.toJSONString(params), response, e.getMessage());
}
return false;
}
@Override
public ClusterTaskStateEnum getTaskState(Long agentTaskId) {
public Result<ClusterTaskStateEnum> getTaskExecuteState(Long taskId) {
String response = null;
try {
// 获取任务的state
response = HttpUtils.get(
baseUrl + TASK_STATE_URI.replace("{taskId}", agentTaskId.toString()), null
);
N9eResult n9eResult = JSON.parseObject(response, N9eResult.class);
if (!ValidateUtils.isBlank(n9eResult.getErr())) {
LOGGER.error("get response result failed, agentTaskId:{} response:{}.", agentTaskId, response);
return null;
response = HttpUtils.get(baseUrl + TASK_STATE_URI.replace("{taskId}", String.valueOf(taskId)), null);
N9eResult nr = JsonUtils.stringToObj(response, N9eResult.class);
if (!ValidateUtils.isBlank(nr.getErr())) {
return Result.buildFailure(nr.getErr());
}
String state = JSON.parseObject(JSON.toJSONString(n9eResult.getDat()), String.class);
String state = JsonUtils.stringToObj(JsonUtils.toJSONString(nr.getDat()), String.class);
N9eTaskStatusEnum n9eTaskStatusEnum = N9eTaskStatusEnum.getByMessage(state);
if (ValidateUtils.isNull(n9eTaskStatusEnum)) {
LOGGER.error("get task status failed, agentTaskId:{} state:{}.", agentTaskId, state);
return null;
LOGGER.error("class=N9e||method=getTaskExecuteState||taskId={}||response={}||msg=get task state failed", taskId, response);
return Result.buildFailure("unknown state, state:" + state);
}
return n9eTaskStatusEnum.getStatus();
return Result.buildSuc(n9eTaskStatusEnum.getStatus());
} catch (Exception e) {
LOGGER.error("get task status failed, agentTaskId:{} response:{}.", agentTaskId, response, e);
LOGGER.error("class=N9e||method=getTaskExecuteState||taskId={}||response={}||errMsg={}||msg=get task state failed", taskId, response, e.getMessage());
}
return null;
return Result.buildFailure("get task state failed");
}
@Override
public Map<String, ClusterTaskSubStateEnum> getTaskResult(Long agentTaskId) {
public Result<Map<String, ClusterTaskSubStateEnum>> getTaskResult(Long taskId) {
String response = null;
try {
// 获取子任务的state
response = HttpUtils.get(baseUrl + TASK_SUB_STATE_URI.replace("{taskId}", agentTaskId.toString()), null);
N9eResult n9eResult = JSON.parseObject(response, N9eResult.class);
response = HttpUtils.get(baseUrl + TASK_SUB_STATE_URI.replace("{taskId}", String.valueOf(taskId)), null);
N9eResult nr = JsonUtils.stringToObj(response, N9eResult.class);
if (!ValidateUtils.isBlank(nr.getErr())) {
LOGGER.error("class=N9e||method=getTaskResult||taskId={}||response={}||msg=get task result failed", taskId, response);
return Result.buildFailure(nr.getErr());
}
N9eTaskResultDTO n9eTaskResultDTO =
JSON.parseObject(JSON.toJSONString(n9eResult.getDat()), N9eTaskResultDTO.class);
return n9eTaskResultDTO.convert2HostnameStatusMap();
return Result.buildSuc(JsonUtils.stringToObj(JsonUtils.toJSONString(nr.getDat()), N9eTaskResult.class).convert2HostnameStatusMap());
} catch (Exception e) {
LOGGER.error("get task result failed, agentTaskId:{} response:{}.", agentTaskId, response, e);
LOGGER.error("class=N9e||method=getTaskResult||taskId={}||response={}||errMsg={}||msg=get task result failed", taskId, response, e.getMessage());
}
return null;
return Result.buildFailure("get task result failed");
}
@Override
public String getTaskLog(Long agentTaskId, String hostname) {
public Result<ClusterTaskLog> getTaskLog(Long taskId, String hostname) {
Map<String, String> params = new HashMap<>(1);
params.put("hostname", hostname);
String response = null;
try {
Map<String, String> params = new HashMap<>(1);
params.put("hostname", hostname);
response = HttpUtils.get(baseUrl + TASK_STD_LOG_URI.replace("{taskId}", String.valueOf(taskId)), params);
N9eResult nr = JsonUtils.stringToObj(response, N9eResult.class);
if (!ValidateUtils.isBlank(nr.getErr())) {
LOGGER.error("class=N9e||method=getTaskLog||taskId={}||response={}||msg=get task log failed", taskId, response);
return Result.buildFailure(nr.getErr());
}
response = HttpUtils.get(baseUrl + TASK_STD_LOG_URI.replace("{taskId}", agentTaskId.toString()), params);
N9eResult n9eResult = JSON.parseObject(response, N9eResult.class);
if (!ValidateUtils.isBlank(n9eResult.getErr())) {
LOGGER.error("get task log failed, agentTaskId:{} response:{}.", agentTaskId, response);
return null;
}
List<N9eTaskStdoutDTO> dtoList =
JSON.parseArray(JSON.toJSONString(n9eResult.getDat()), N9eTaskStdoutDTO.class);
List<N9eTaskStdoutLog> dtoList = JsonUtils.stringToArrObj(JsonUtils.toJSONString(nr.getDat()), N9eTaskStdoutLog.class);
if (ValidateUtils.isEmptyList(dtoList)) {
return "";
return Result.buildSuc(new ClusterTaskLog(""));
}
return dtoList.get(0).getStdout();
return Result.buildSuc(new ClusterTaskLog(dtoList.get(0).getStdout()));
} catch (Exception e) {
LOGGER.error("get task log failed, agentTaskId:{}.", agentTaskId, e);
LOGGER.error("class=N9e||method=getTaskLog||taskId={}||response={}||errMsg={}||msg=get task log failed", taskId, response, e.getMessage());
}
return null;
return Result.buildFailure("get task log failed");
}
private Map<String, String> buildHeader() {
@@ -228,7 +209,7 @@ public class N9e extends AbstractAgent {
return headers;
}
private Map<String, Object> buildCreateTaskParam(CreationTaskData creationTaskData) {
private N9eCreationTask buildCreateTaskParam(CreationTaskData creationTaskData) {
StringBuilder sb = new StringBuilder();
sb.append(creationTaskData.getUuid()).append(",,");
sb.append(creationTaskData.getClusterId()).append(",,");
@@ -240,46 +221,17 @@ public class N9e extends AbstractAgent {
sb.append(creationTaskData.getServerPropertiesMd5()).append(",,");
sb.append(creationTaskData.getServerPropertiesUrl());
Map<String, Object> params = new HashMap<>(10);
params.put("title", String.format("集群ID=%d-升级部署", creationTaskData.getClusterId()));
params.put("batch", BATCH);
params.put("tolerance", TOLERANCE);
params.put("timeout", timeout);
params.put("pause", ListUtils.strList2String(creationTaskData.getPauseList()));
params.put("script", this.script);
params.put("args", sb.toString());
params.put("account", account);
params.put("action", "pause");
params.put("hosts", creationTaskData.getHostList());
return params;
}
private static String readScriptInJarFile(String fileName) {
InputStream inputStream = N9e.class.getClassLoader().getResourceAsStream(fileName);
if (inputStream == null) {
LOGGER.error("read kcm script failed, filename:{}", fileName);
return "";
}
try {
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
String line = null;
StringBuilder stringBuilder = new StringBuilder("");
while ((line = bufferedReader.readLine()) != null) {
stringBuilder.append(line);
stringBuilder.append("\n");
}
return stringBuilder.toString();
} catch (IOException e) {
LOGGER.error("read kcm script failed, filename:{}", fileName, e);
return "";
} finally {
try {
inputStream.close();
} catch (IOException e) {
LOGGER.error("close reading kcm script failed, filename:{}", fileName, e);
}
}
N9eCreationTask n9eCreationTask = new N9eCreationTask();
n9eCreationTask.setTitle(Constant.TASK_TITLE_PREFIX + "-集群ID:" + creationTaskData.getClusterId());
n9eCreationTask.setBatch(Constant.AGENT_TASK_BATCH);
n9eCreationTask.setTolerance(Constant.AGENT_TASK_TOLERANCE);
n9eCreationTask.setTimeout(this.timeout);
n9eCreationTask.setPause(ListUtils.strList2String(creationTaskData.getPauseList()));
n9eCreationTask.setScript(this.script);
n9eCreationTask.setArgs(sb.toString());
n9eCreationTask.setAccount(this.account);
n9eCreationTask.setAction(ClusterTaskActionEnum.PAUSE.getAction());
n9eCreationTask.setHosts(creationTaskData.getHostList());
return n9eCreationTask;
}
}

151
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/component/agent/n9e/entry/N9eCreationTask.java Normal file
View File

@@ -0,0 +1,151 @@
package com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry;
import java.util.List;
public class N9eCreationTask {
/**
* 任务标题
*/
private String title;
/**
* 并发度, =2则表示两台并发执行
*/
private Integer batch;
/**
* 错误容忍度, 达到容忍度之上时, 任务会被暂停并不可以继续执行
*/
private Integer tolerance;
/**
* 单台任务的超时时间(秒)
*/
private Integer timeout;
/**
* 暂停点, 格式: host1,host2,host3
*/
private String pause;
/**
* 任务执行对应的脚本
*/
private String script;
/**
* 任务参数
*/
private String args;
/**
* 使用的账号
*/
private String account;
/**
* 动作
*/
private String action;
/**
* 操作的主机列表
*/
private List<String> hosts;
public String getTitle() {
return title;
}
public void setTitle(String title) {
this.title = title;
}
public Integer getBatch() {
return batch;
}
public void setBatch(Integer batch) {
this.batch = batch;
}
public Integer getTolerance() {
return tolerance;
}
public void setTolerance(Integer tolerance) {
this.tolerance = tolerance;
}
public Integer getTimeout() {
return timeout;
}
public void setTimeout(Integer timeout) {
this.timeout = timeout;
}
public String getPause() {
return pause;
}
public void setPause(String pause) {
this.pause = pause;
}
public String getScript() {
return script;
}
public void setScript(String script) {
this.script = script;
}
public String getArgs() {
return args;
}
public void setArgs(String args) {
this.args = args;
}
public String getAccount() {
return account;
}
public void setAccount(String account) {
this.account = account;
}
public String getAction() {
return action;
}
public void setAction(String action) {
this.action = action;
}
public List<String> getHosts() {
return hosts;
}
public void setHosts(List<String> hosts) {
this.hosts = hosts;
}
@Override
public String toString() {
return "N9eCreationTask{" +
"title='" + title + '\'' +
", batch=" + batch +
", tolerance=" + tolerance +
", timeout=" + timeout +
", pause='" + pause + '\'' +
", script='" + script + '\'' +
", args='" + args + '\'' +
", account='" + account + '\'' +
", action='" + action + '\'' +
", hosts=" + hosts +
'}';
}
}

2
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/component/agent/n9e/entry/N9eTaskResultDTO.java → kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/component/agent/n9e/entry/N9eTaskResult.java
View File

@@ -12,7 +12,7 @@ import java.util.Map;
* @author zengqiao
* @date 20/9/7
*/
public class N9eTaskResultDTO {
public class N9eTaskResult {
private List<String> waiting;
private List<String> running;

35
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/component/agent/n9e/entry/N9eTaskStdoutLog.java Normal file
View File

@@ -0,0 +1,35 @@
package com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry;
/**
* @author zengqiao
* @date 20/9/7
*/
public class N9eTaskStdoutLog {
private String host;
private String stdout;
public String getHost() {
return host;
}
public void setHost(String host) {
this.host = host;
}
public String getStdout() {
return stdout;
}
public void setStdout(String stdout) {
this.stdout = stdout;
}
@Override
public String toString() {
return "N9eTaskStdoutDTO{" +
"host='" + host + '\'' +
", stdout='" + stdout + '\'' +
'}';
}
}

59
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/component/agent/n9e/entry/bizenum/N9eTaskStatusEnum.java Normal file
View File

@@ -0,0 +1,59 @@
package com.xiaojukeji.kafka.manager.kcm.component.agent.n9e.entry.bizenum;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskStateEnum;
/**
* @author zengqiao
* @date 20/9/3
*/
public enum N9eTaskStatusEnum {
DONE(0, "done", ClusterTaskStateEnum.FINISHED),
PAUSE(1, "pause", ClusterTaskStateEnum.BLOCKED),
START(2, "start", ClusterTaskStateEnum.RUNNING),
;
private Integer code;
private String message;
private ClusterTaskStateEnum status;
N9eTaskStatusEnum(Integer code, String message, ClusterTaskStateEnum status) {
this.code = code;
this.message = message;
this.status = status;
}
public Integer getCode() {
return code;
}
public void setCode(Integer code) {
this.code = code;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
public ClusterTaskStateEnum getStatus() {
return status;
}
public void setStatus(ClusterTaskStateEnum status) {
this.status = status;
}
public static N9eTaskStatusEnum getByMessage(String message) {
for (N9eTaskStatusEnum elem: N9eTaskStatusEnum.values()) {
if (elem.message.equals(message)) {
return elem;
}
}
return null;
}
}

11
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/component/storage/AbstractStorageService.java
View File

@@ -10,13 +10,20 @@ import org.springframework.web.multipart.MultipartFile;
public abstract class AbstractStorageService {
/**
* 上传
* @param fileName 文件名
* @param fileMd5 文件md5
* @param uploadFile 文件
* @return 上传结果
*/
public abstract boolean upload(String fileName, String fileMd5, MultipartFile uploadFile);
/**
* 下载
* 下载文件
* @param fileName 文件名
* @param fileMd5 文件md5
* @return 文件
*/
public abstract Result<String> download(String fileName, String fileMd5);
public abstract Result<MultipartFile> download(String fileName, String fileMd5);
/**
* 下载base地址

33
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/component/storage/local/Local.java
View File

@@ -1,33 +0,0 @@
package com.xiaojukeji.kafka.manager.kcm.component.storage.local;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import com.xiaojukeji.kafka.manager.kcm.component.storage.AbstractStorageService;
import org.springframework.web.multipart.MultipartFile;
/**
* @author zengqiao
* @date 20/9/17
*/
@Service("storageService")
public class Local extends AbstractStorageService {
@Value("${kcm.storage.base-url}")
private String baseUrl;
@Override
public boolean upload(String fileName, String fileMd5, MultipartFile uploadFile) {
return false;
}
@Override
public Result<String> download(String fileName, String fileMd5) {
return Result.buildFrom(ResultStatus.DOWNLOAD_FILE_FAIL);
}
@Override
public String getDownloadBaseUrl() {
return baseUrl;
}
}

128
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/component/storage/s3/S3Service.java Normal file
View File

@@ -0,0 +1,128 @@
package com.xiaojukeji.kafka.manager.kcm.component.storage.s3;
import com.xiaojukeji.kafka.manager.common.entity.Result;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.kcm.component.storage.AbstractStorageService;
import io.minio.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.mock.web.MockMultipartFile;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.PostConstruct;
import java.io.IOException;
import java.io.InputStream;
@Service("storageService")
public class S3Service extends AbstractStorageService {
private final static Logger LOGGER = LoggerFactory.getLogger(S3Service.class);
@Value("${kcm.s3.endpoint:}")
private String endpoint;
@Value("${kcm.s3.access-key:}")
private String accessKey;
@Value("${kcm.s3.secret-key:}")
private String secretKey;
@Value("${kcm.s3.bucket:}")
private String bucket;
private MinioClient minioClient;
@PostConstruct
public void init() {
try {
if (ValidateUtils.anyBlank(this.endpoint, this.accessKey, this.secretKey, this.bucket)) {
// without config s3
return;
}
minioClient = new MinioClient(endpoint, accessKey, secretKey);
} catch (Exception e) {
LOGGER.error("class=S3Service||method=init||fields={}||errMsg={}", this.toString(), e.getMessage());
}
}
@Override
public boolean upload(String fileName, String fileMd5, MultipartFile uploadFile) {
InputStream inputStream = null;
try {
if (!createBucketIfNotExist()) {
return false;
}
inputStream = uploadFile.getInputStream();
minioClient.putObject(PutObjectArgs.builder()
.bucket(this.bucket)
.object(fileName)
.stream(inputStream, inputStream.available(), -1)
.build()
);
return true;
} catch (Exception e) {
LOGGER.error("class=S3Service||method=upload||fileName={}||errMsg={}||msg=upload failed", fileName, e.getMessage());
} finally {
if (inputStream != null) {
try {
inputStream.close();
} catch (IOException e) {
; // ignore
}
}
}
return false;
}
@Override
public Result<MultipartFile> download(String fileName, String fileMd5) {
try {
final ObjectStat stat = minioClient.statObject(this.bucket, fileName);
InputStream is = minioClient.getObject(this.bucket, fileName);
return Result.buildSuc(new MockMultipartFile(fileName, fileName, stat.contentType(), is));
} catch (Exception e) {
LOGGER.error("class=S3Service||method=download||fileName={}||errMsg={}||msg=download failed", fileName, e.getMessage());
}
return Result.buildFrom(ResultStatus.STORAGE_DOWNLOAD_FILE_FAILED);
}
@Override
public String getDownloadBaseUrl() {
if (this.endpoint.startsWith("http://")) {
return this.endpoint + "/" + this.bucket;
}
return "http://" + this.endpoint + "/" + this.bucket;
}
private boolean createBucketIfNotExist() {
try {
boolean found = minioClient.bucketExists(BucketExistsArgs.builder().bucket(this.bucket).build());
if (!found) {
minioClient.makeBucket(MakeBucketArgs.builder().bucket(this.bucket).build());
}
LOGGER.info("class=S3Service||method=createBucketIfNotExist||bucket={}||msg=check and create bucket success", this.bucket);
return true;
} catch (Exception e) {
LOGGER.error("class=S3Service||method=createBucketIfNotExist||bucket={}||errMsg={}||msg=create bucket failed", this.bucket, e.getMessage());
}
return false;
}
@Override
public String toString() {
return "S3Service{" +
"endpoint='" + endpoint + '\'' +
", accessKey='" + accessKey + '\'' +
", secretKey='" + secretKey + '\'' +
", bucket='" + bucket + '\'' +
'}';
}
}

108
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/impl/ClusterTaskServiceImpl.java
View File

@@ -6,6 +6,7 @@ import com.xiaojukeji.kafka.manager.kcm.ClusterTaskService;
import com.xiaojukeji.kafka.manager.kcm.common.Converters;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskActionEnum;
import com.xiaojukeji.kafka.manager.kcm.common.entry.ClusterTaskConstant;
import com.xiaojukeji.kafka.manager.kcm.common.entry.ao.ClusterTaskLog;
import com.xiaojukeji.kafka.manager.kcm.common.entry.ao.ClusterTaskSubStatus;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskStateEnum;
import com.xiaojukeji.kafka.manager.kcm.common.bizenum.ClusterTaskSubStateEnum;
@@ -34,7 +35,7 @@ import java.util.*;
*/
@Service("clusterTaskService")
public class ClusterTaskServiceImpl implements ClusterTaskService {
private final static Logger LOGGER = LoggerFactory.getLogger(ClusterTaskServiceImpl.class);
private static final Logger LOGGER = LoggerFactory.getLogger(ClusterTaskServiceImpl.class);
@Autowired
private AbstractAgent abstractAgent;
@@ -63,13 +64,13 @@ public class ClusterTaskServiceImpl implements ClusterTaskService {
}
// 创建任务
Long agentTaskId = abstractAgent.createTask(dtoResult.getData());
if (ValidateUtils.isNull(agentTaskId)) {
Result<Long> createResult = abstractAgent.createTask(dtoResult.getData());
if (ValidateUtils.isNull(createResult) || createResult.failed()) {
return Result.buildFrom(ResultStatus.CALL_CLUSTER_TASK_AGENT_FAILED);
}
try {
if (clusterTaskDao.insert(Converters.convert2ClusterTaskDO(agentTaskId, dtoResult.getData(), operator)) > 0) {
if (clusterTaskDao.insert(Converters.convert2ClusterTaskDO(createResult.getData(), dtoResult.getData(), operator)) > 0) {
return Result.buildFrom(ResultStatus.SUCCESS);
}
} catch (Exception e) {
@@ -87,45 +88,44 @@ public class ClusterTaskServiceImpl implements ClusterTaskService {
Long agentTaskId = getActiveAgentTaskId(clusterTaskDO);
Boolean rollback = inRollback(clusterTaskDO);
ClusterTaskStateEnum stateEnum = abstractAgent.getTaskState(agentTaskId);
if (ClusterTaskActionEnum.START.getMessage().equals(action)
&& ClusterTaskStateEnum.BLOCKED.equals(stateEnum)) {
Result<ClusterTaskStateEnum> stateEnumResult = abstractAgent.getTaskExecuteState(agentTaskId);
if (ValidateUtils.isNull(stateEnumResult) || stateEnumResult.failed()) {
return ResultStatus.CALL_CLUSTER_TASK_AGENT_FAILED;
}
if (ClusterTaskActionEnum.START.getAction().equals(action) && ClusterTaskStateEnum.BLOCKED.equals(stateEnumResult.getData())) {
// 暂停状态, 可以执行开始
return actionTaskExceptRollbackAction(agentTaskId, action, "");
return actionTaskExceptRollbackAction(agentTaskId, ClusterTaskActionEnum.START, "");
}
if (ClusterTaskActionEnum.PAUSE.getMessage().equals(action)
&& ClusterTaskStateEnum.RUNNING.equals(stateEnum)) {
if (ClusterTaskActionEnum.PAUSE.getAction().equals(action) && ClusterTaskStateEnum.RUNNING.equals(stateEnumResult.getData())) {
// 运行状态, 可以执行暂停
return actionTaskExceptRollbackAction(agentTaskId, action, "");
return actionTaskExceptRollbackAction(agentTaskId, ClusterTaskActionEnum.PAUSE, "");
}
if (ClusterTaskActionEnum.IGNORE.getMessage().equals(action)
|| ClusterTaskActionEnum.CANCEL.getMessage().equals(action)) {
if (ClusterTaskActionEnum.IGNORE.getAction().equals(action)) {
// 忽略 & 取消随时都可以操作
return actionTaskExceptRollbackAction(agentTaskId, action, hostname);
return actionTaskExceptRollbackAction(agentTaskId, ClusterTaskActionEnum.IGNORE, hostname);
}
if ((!ClusterTaskStateEnum.FINISHED.equals(stateEnum) || !rollback)
&& ClusterTaskActionEnum.ROLLBACK.getMessage().equals(action)) {
if (ClusterTaskActionEnum.CANCEL.getAction().equals(action)) {
// 忽略 & 取消随时都可以操作
return actionTaskExceptRollbackAction(agentTaskId, ClusterTaskActionEnum.CANCEL, hostname);
}
if ((!ClusterTaskStateEnum.FINISHED.equals(stateEnumResult.getData()) || !rollback)
&& ClusterTaskActionEnum.ROLLBACK.getAction().equals(action)) {
// 暂未操作完时可以回滚, 回滚所有操作过的机器到上一个版本
return actionTaskRollback(clusterTaskDO);
}
return ResultStatus.OPERATION_FAILED;
}
private ResultStatus actionTaskExceptRollbackAction(Long agentId, String action, String hostname) {
private ResultStatus actionTaskExceptRollbackAction(Long agentId, ClusterTaskActionEnum actionEnum, String hostname) {
if (!ValidateUtils.isBlank(hostname)) {
return actionHostTaskExceptRollbackAction(agentId, action, hostname);
return actionHostTaskExceptRollbackAction(agentId, actionEnum, hostname);
}
if (abstractAgent.actionTask(agentId, action)) {
return ResultStatus.SUCCESS;
}
return ResultStatus.OPERATION_FAILED;
return abstractAgent.actionTask(agentId, actionEnum)? ResultStatus.SUCCESS: ResultStatus.OPERATION_FAILED;
}
private ResultStatus actionHostTaskExceptRollbackAction(Long agentId, String action, String hostname) {
if (abstractAgent.actionHostTask(agentId, action, hostname)) {
return ResultStatus.SUCCESS;
}
return ResultStatus.OPERATION_FAILED;
private ResultStatus actionHostTaskExceptRollbackAction(Long agentId, ClusterTaskActionEnum actionEnum, String hostname) {
return abstractAgent.actionHostTask(agentId, actionEnum, hostname)? ResultStatus.SUCCESS: ResultStatus.OPERATION_FAILED;
}
private ResultStatus actionTaskRollback(ClusterTaskDO clusterTaskDO) {
@@ -133,9 +133,9 @@ public class ClusterTaskServiceImpl implements ClusterTaskService {
return ResultStatus.OPERATION_FORBIDDEN;
}
Map<String, ClusterTaskSubStateEnum> subStatusEnumMap =
Result<Map<String, ClusterTaskSubStateEnum>> subStatusEnumMapResult =
abstractAgent.getTaskResult(clusterTaskDO.getAgentTaskId());
if (ValidateUtils.isNull(subStatusEnumMap)) {
if (ValidateUtils.isNull(subStatusEnumMapResult) || subStatusEnumMapResult.failed()) {
return ResultStatus.CALL_CLUSTER_TASK_AGENT_FAILED;
}
@@ -143,7 +143,7 @@ public class ClusterTaskServiceImpl implements ClusterTaskService {
List<String> rollbackHostList = new ArrayList<>();
List<String> rollbackPauseHostList = new ArrayList<>();
for (String host: ListUtils.string2StrList(clusterTaskDO.getHostList())) {
ClusterTaskSubStateEnum subStateEnum = subStatusEnumMap.get(host);
ClusterTaskSubStateEnum subStateEnum = subStatusEnumMapResult.getData().get(host);
if (ValidateUtils.isNull(subStateEnum)) {
// 机器对应的任务查询失败
return ResultStatus.OPERATION_FAILED;
@@ -166,17 +166,17 @@ public class ClusterTaskServiceImpl implements ClusterTaskService {
clusterTaskDO.setRollbackPauseHostList(ListUtils.strList2String(rollbackPauseHostList));
// 创建任务
Long agentTaskId = abstractAgent.createTask(Converters.convert2CreationTaskData(clusterTaskDO));
if (ValidateUtils.isNull(agentTaskId)) {
Result<Long> createResult = abstractAgent.createTask(Converters.convert2CreationTaskData(clusterTaskDO));
if (ValidateUtils.isNull(createResult) || createResult.failed()) {
return ResultStatus.CALL_CLUSTER_TASK_AGENT_FAILED;
}
try {
clusterTaskDO.setAgentRollbackTaskId(agentTaskId);
clusterTaskDO.setAgentRollbackTaskId(createResult.getData());
if (clusterTaskDao.updateRollback(clusterTaskDO) <= 0) {
return ResultStatus.MYSQL_ERROR;
}
abstractAgent.actionTask(clusterTaskDO.getAgentTaskId(), ClusterTaskActionEnum.CANCEL.getMessage());
abstractAgent.actionTask(clusterTaskDO.getAgentTaskId(), ClusterTaskActionEnum.CANCEL);
return ResultStatus.SUCCESS;
} catch (Exception e) {
LOGGER.error("create cluster task failed, clusterTaskDO:{}.", clusterTaskDO, e);
@@ -191,11 +191,11 @@ public class ClusterTaskServiceImpl implements ClusterTaskService {
return Result.buildFrom(ResultStatus.TASK_NOT_EXIST);
}
String stdoutLog = abstractAgent.getTaskLog(getActiveAgentTaskId(clusterTaskDO, hostname), hostname);
if (ValidateUtils.isNull(stdoutLog)) {
Result<ClusterTaskLog> stdoutLogResult = abstractAgent.getTaskLog(getActiveAgentTaskId(clusterTaskDO, hostname), hostname);
if (ValidateUtils.isNull(stdoutLogResult) || stdoutLogResult.failed()) {
return Result.buildFrom(ResultStatus.CALL_CLUSTER_TASK_AGENT_FAILED);
}
return new Result<>(stdoutLog);
return new Result<>(stdoutLogResult.getData().getStdout());
}
@Override
@@ -205,24 +205,33 @@ public class ClusterTaskServiceImpl implements ClusterTaskService {
return Result.buildFrom(ResultStatus.TASK_NOT_EXIST);
}
Result<ClusterTaskStateEnum> statusEnumResult = abstractAgent.getTaskExecuteState(getActiveAgentTaskId(clusterTaskDO));
if (ValidateUtils.isNull(statusEnumResult) || statusEnumResult.failed()) {
return new Result<>(statusEnumResult.getCode(), statusEnumResult.getMessage());
}
return new Result<>(new ClusterTaskStatus(
clusterTaskDO.getId(),
clusterTaskDO.getClusterId(),
inRollback(clusterTaskDO),
abstractAgent.getTaskState(getActiveAgentTaskId(clusterTaskDO)),
statusEnumResult.getData(),
getTaskSubStatus(clusterTaskDO)
));
}
@Override
public ClusterTaskStateEnum getTaskState(Long agentTaskId) {
return abstractAgent.getTaskState(agentTaskId);
Result<ClusterTaskStateEnum> statusEnumResult = abstractAgent.getTaskExecuteState(agentTaskId);
if (ValidateUtils.isNull(statusEnumResult) || statusEnumResult.failed()) {
return null;
}
return statusEnumResult.getData();
}
private List<ClusterTaskSubStatus> getTaskSubStatus(ClusterTaskDO clusterTaskDO) {
Map<String, ClusterTaskSubStateEnum> statusMap = this.getClusterTaskSubState(clusterTaskDO);
if (ValidateUtils.isNull(statusMap)) {
return null;
return Collections.emptyList();
}
List<String> pauseList = ListUtils.string2StrList(clusterTaskDO.getPauseHostList());
@@ -242,20 +251,22 @@ public class ClusterTaskServiceImpl implements ClusterTaskService {
}
private Map<String, ClusterTaskSubStateEnum> getClusterTaskSubState(ClusterTaskDO clusterTaskDO) {
Map<String, ClusterTaskSubStateEnum> statusMap = abstractAgent.getTaskResult(clusterTaskDO.getAgentTaskId());
if (ValidateUtils.isNull(statusMap)) {
Result<Map<String, ClusterTaskSubStateEnum>> statusMapResult = abstractAgent.getTaskResult(clusterTaskDO.getAgentTaskId());
if (ValidateUtils.isNull(statusMapResult) || statusMapResult.failed()) {
return null;
}
Map<String, ClusterTaskSubStateEnum> statusMap = statusMapResult.getData();
if (!inRollback(clusterTaskDO)) {
return statusMap;
}
Map<String, ClusterTaskSubStateEnum> rollbackStatusMap =
Result<Map<String, ClusterTaskSubStateEnum>> rollbackStatusMapResult =
abstractAgent.getTaskResult(clusterTaskDO.getAgentRollbackTaskId());
if (ValidateUtils.isNull(rollbackStatusMap)) {
if (ValidateUtils.isNull(rollbackStatusMapResult) || rollbackStatusMapResult.failed()) {
return null;
}
statusMap.putAll(rollbackStatusMap);
statusMap.putAll(rollbackStatusMapResult.getData());
return statusMap;
}
@@ -276,7 +287,7 @@ public class ClusterTaskServiceImpl implements ClusterTaskService {
} catch (Exception e) {
LOGGER.error("get all cluster task failed.");
}
return null;
return Collections.emptyList();
}
@Override
@@ -302,9 +313,6 @@ public class ClusterTaskServiceImpl implements ClusterTaskService {
}
private boolean inRollback(ClusterTaskDO clusterTaskDO) {
if (ClusterTaskConstant.INVALID_AGENT_TASK_ID.equals(clusterTaskDO.getAgentRollbackTaskId())) {
return false;
}
return true;
return !ClusterTaskConstant.INVALID_AGENT_TASK_ID.equals(clusterTaskDO.getAgentRollbackTaskId());
}
}

9
kafka-manager-extends/kafka-manager-kcm/src/main/java/com/xiaojukeji/kafka/manager/kcm/impl/KafkaFileServiceImpl.java
View File

@@ -15,6 +15,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;
import java.util.ArrayList;
import java.util.List;
@@ -52,7 +53,7 @@ public class KafkaFileServiceImpl implements KafkaFileService {
kafkaFileDTO.getUploadFile())
) {
kafkaFileDao.deleteById(kafkaFileDO.getId());
return ResultStatus.UPLOAD_FILE_FAIL;
return ResultStatus.STORAGE_UPLOAD_FILE_FAILED;
}
return ResultStatus.SUCCESS;
} catch (DuplicateKeyException e) {
@@ -113,7 +114,7 @@ public class KafkaFileServiceImpl implements KafkaFileService {
if (kafkaFileDao.updateById(kafkaFileDO) <= 0) {
return ResultStatus.MYSQL_ERROR;
}
return ResultStatus.UPLOAD_FILE_FAIL;
return ResultStatus.STORAGE_UPLOAD_FILE_FAILED;
} catch (Exception e) {
LOGGER.error("rollback modify kafka file failed, kafkaFileDTO:{}.", kafkaFileDTO, e);
}
@@ -163,13 +164,13 @@ public class KafkaFileServiceImpl implements KafkaFileService {
}
@Override
public Result<String> downloadKafkaConfigFile(Long fileId) {
public Result<MultipartFile> downloadKafkaFile(Long fileId) {
KafkaFileDO kafkaFileDO = kafkaFileDao.getById(fileId);
if (ValidateUtils.isNull(kafkaFileDO)) {
return Result.buildFrom(ResultStatus.RESOURCE_NOT_EXIST);
}
if (KafkaFileEnum.PACKAGE.getCode().equals(kafkaFileDO.getFileType())) {
return Result.buildFrom(ResultStatus.FILE_TYPE_NOT_SUPPORT);
return Result.buildFrom(ResultStatus.STORAGE_FILE_TYPE_NOT_SUPPORT);
}
return storageService.download(kafkaFileDO.getFileName(), kafkaFileDO.getFileMd5());