8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2026-01-02 02:02:13 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

合并3.3.0企业版改动

Browse Source
This commit is contained in:
zengqiao
2023-02-24 17:49:26 +08:00
parent cca7246281
commit a82d7f594e
137 changed files with 591 additions and 1082 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
km-common/pom.xml
View File

@@ -22,12 +22,6 @@
</properties>
<dependencies>
<dependency>
<groupId>com.xiaojukeji.kafka</groupId>
<artifactId>km-rebalance</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>

2
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/constant/ESConstant.java
View File

@@ -36,8 +36,6 @@ public class ESConstant {
public static final Integer DEFAULT_RETRY_TIME = 3;
public static final String PARTITION_INDEX = "ks_kafka_partition_metric";
/**
* 获取Topic-Latest指标时，单次允许的Topic数
*/

26
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/dto/ClusterBalanceIntervalDTO.java
View File

@@ -1,26 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.dto;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceIntervalDTO {
@NotBlank(message = "clusterBalanceIntervalDTO.type不允许为空")
@ApiModelProperty("均衡维度:cpu,disk,bytesIn,bytesOut")
private String type;
@NotNull(message = "clusterBalanceIntervalDTO.intervalPercent不允许为空")
@ApiModelProperty("平衡区间百分比")
private Double intervalPercent;
@NotNull(message = "clusterBalanceIntervalDTO.priority不允许为空")
@ApiModelProperty("优先级")
private Integer priority;
}

20
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/dto/ClusterBalanceOverviewDTO.java
View File

@@ -1,20 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.dto;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.bean.dto.pagination.PaginationBaseDTO;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.util.Map;
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceOverviewDTO extends PaginationBaseDTO {
@ApiModelProperty("host")
private String host;
@ApiModelProperty("key:disk,bytesOut,bytesIn value:均衡状态 0：已均衡；2：未均衡")
private Map<String, Integer> stateParam;
}

43
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/dto/ClusterBalancePreviewDTO.java
View File

@@ -1,43 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.dto;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.bean.dto.BaseDTO;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.util.List;
/**
* @author zengqiao
* @date 22/02/24
*/
@Data
@EnterpriseLoadReBalance
public class ClusterBalancePreviewDTO extends BaseDTO {
@ApiModelProperty("集群id")
private Long clusterId;
@ApiModelProperty("均衡节点")
private List<Integer> brokers;
@ApiModelProperty("topic黑名单")
private List<String> topicBlackList;
@ApiModelProperty("均衡区间详情")
private List<ClusterBalanceIntervalDTO> clusterBalanceIntervalList;
@ApiModelProperty("指标计算周期，单位分钟")
private Integer metricCalculationPeriod;
@ApiModelProperty("任务并行数")
private Integer parallelNum;
@ApiModelProperty("执行策略， 1：优先最大副本，2：优先最小副本")
private Integer executionStrategy;
@ApiModelProperty("限流值")
private Long throttleUnitB;
}

66
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/dto/ClusterBalanceStrategyDTO.java
View File

@@ -1,66 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.dto;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.bean.dto.BaseDTO;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
import java.util.List;
/**
* @author zengqiao
* @date 22/02/24
*/
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceStrategyDTO extends BaseDTO {
@ApiModelProperty("是否是周期性任务")
private boolean scheduleJob;
@NotBlank(message = "scheduleCron不允许为空")
@ApiModelProperty("如果是周期任务，那么任务的周期cron表达式")
private String scheduleCron;
@NotNull(message = "status不允许为空")
@ApiModelProperty("周期任务状态：0:不开启，1：开启")
private Integer status;
@NotNull(message = "clusterId不允许为空")
@ApiModelProperty("集群id")
private Long clusterId;
@ApiModelProperty("均衡节点")
private List<Integer> brokers;
@ApiModelProperty("topic黑名单")
private List<String> topicBlackList;
@NotNull(message = "clusterBalanceIntervalDTO不允许为空")
@ApiModelProperty("均衡区间详情")
private List<ClusterBalanceIntervalDTO> clusterBalanceIntervalList;
@NotNull(message = "metricCalculationPeriod不允许为空")
@ApiModelProperty("指标计算周期，单位秒")
private Integer metricCalculationPeriod;
@NotNull(message = "parallelNum不允许为空")
@ApiModelProperty("任务并行数（0代表不限）")
private Integer parallelNum;
@NotNull(message = "executionStrategy不允许为空")
@ApiModelProperty("执行策略， 1：优先最大副本，2：优先最小副本")
private Integer executionStrategy;
@Min(value = 1, message = "throttleUnitB不允许小于1")
@ApiModelProperty("限流值")
private Long throttleUnitB;
@ApiModelProperty("备注说明")
private String description;
}

27
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/ClusterBalanceInterval.java
View File

@@ -1,27 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@NoArgsConstructor
@AllArgsConstructor
@EnterpriseLoadReBalance
public class ClusterBalanceInterval {
/**
* 均衡维度:cpu,disk,bytesIn,bytesOut
*/
private String type;
/**
* 平衡区间百分比
*/
private Double intervalPercent;
/**
* 优先级
*/
private Integer priority;
}

41
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/ClusterBalanceItemState.java
View File

@@ -1,41 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.constant.Constant;
import com.xiaojukeji.know.streaming.km.rebalance.model.Resource;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Map;
@Data
@NoArgsConstructor
@AllArgsConstructor
@EnterpriseLoadReBalance
public class ClusterBalanceItemState {
/**
* 是否配置集群平衡:true:已配置，false:未配置
*/
private Boolean configureBalance;
/**
* 是否开启均衡:true:开启，false: 未开启
*/
private Boolean enable;
/**
* 子项是否均衡:key: disk,bytesIn,bytesOut,cpu ; value:true:已均衡，false:未均衡
* @see com.xiaojukeji.know.streaming.km.rebalance.model.Resource
*/
private Map<String, Boolean> itemState;
public Integer getResItemState(Resource res) {
if (itemState == null || !itemState.containsKey(res.resource())) {
return Constant.INVALID_CODE;
}
return itemState.get(res.resource()) ? Constant.YES: Constant.NO;
}
}

91
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/job/ClusterBalanceJobConfig.java
View File

@@ -1,91 +0,0 @@
/*
* Copyright (c) 2015, WINIT and/or its affiliates. All rights reserved. Use, Copy is subject to authorized license.
*/
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.bean.entity.BaseEntity;
import lombok.Data;
/**
* 集群均衡任务 实体类
*
* @author fengqiongfeng
* @date 2022-05-23
*/
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceJobConfig extends BaseEntity {
/**
* 序列化版本号
*/
private static final long serialVersionUID=1L;
/**
* 集群id
*/
private Long clusterId;
/**
* 均衡节点
*/
private String brokers;
/**
* topic黑名单
*/
private String topicBlackList;
/**
* 1:立即均衡，2：周期均衡
*/
private Integer type;
/**
* 任务周期
*/
private String taskCron;
/**
* 均衡区间详情
*/
private String balanceIntervalJson;
/**
* 指标计算周期，单位分钟
*/
private Integer metricCalculationPeriod;
/**
* 迁移脚本
*/
private String reassignmentJson;
/**
* 任务并行数
*/
private Integer parallelNum;
/**
* 执行策略， 1：优先最大副本，2：优先最小副本
*/
private Integer executionStrategy;
/**
* 限流值
*/
private Long throttleUnitByte;
/**
* 操作人
*/
private String creator;
/**
* 任务状态 0：未开启，1：开启
*/
private Integer status;
}

64
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/job/ClusterBalanceReassign.java
View File

@@ -1,64 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import lombok.Data;
import java.util.Date;
/**
* @author zengqiao
* @date 22/05/06
*/
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceReassign {
/**
* jobID
*/
private Long jobId;
/**
* 集群id
*/
private Long clusterId;
/**
* Topic名称
*/
private String topicName;
/**
* 分区ID
*/
private Integer partitionId;
/**
* 源BrokerId列表
*/
private String originalBrokerIds;
/**
* 目标BrokerId列表
*/
private String reassignBrokerIds;
/**
* 任务开始时间
*/
private Date startTime;
/**
* 任务完成时间
*/
private Date finishedTime;
/**
* 扩展数据
*/
private String extendData;
/**
* 任务状态
*/
private Integer status;
}

36
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/job/ClusterBalanceReassignDetail.java
View File

@@ -1,36 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.detail.ClusterBalanceDetailDataGroupByTopic;
import lombok.Data;
import java.util.Date;
import java.util.List;
/**
* @author zengqiao
* @date 22/05/06
*/
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceReassignDetail {
/**
* 限流值
*/
private Long throttleUnitB;
/**
* 开始时间
*/
private Date startTime;
/**
* 完成时间
*/
private Date finishedTime;
/**
* 详细信息
*/
private List<ClusterBalanceDetailDataGroupByTopic> reassignTopicDetailsList;
}

47
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/job/ClusterBalanceReassignExtendData.java
View File

@@ -1,47 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import lombok.Data;
/**
* @author zengqiao
* @date 22/05/06
*/
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceReassignExtendData {
/**
* 原本保存时间
*/
private Long originalRetentionTimeUnitMs;
/**
* 迁移时保存时间
*/
private Long reassignRetentionTimeUnitMs;
/**
* 需迁移LogSize
*/
private Long needReassignLogSizeUnitB;
/**
* 已完成迁移LogSize
*/
private Long finishedReassignLogSizeUnitB;
/**
* 预计剩余时长
*/
private Long remainTimeUnitMs;
/**
* 当前副本数
*/
private Integer originReplicaNum;
/**
* 新的副本数
*/
private Integer reassignReplicaNum;
}

43
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/job/content/JobClusterBalanceContent.java
View File

@@ -1,43 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.content;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.bean.entity.job.content.BaseJobCreateContent;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.dto.ClusterBalanceIntervalDTO;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import javax.validation.constraints.Min;
import java.util.List;
@Data
@EnterpriseLoadReBalance
public class JobClusterBalanceContent extends BaseJobCreateContent {
@Min(value = 1, message = "clusterId不允许为null或者小于0")
@ApiModelProperty(value = "集群ID, 默认为逻辑集群ID", example = "6")
private Long clusterId;
@Min(value = 1, message = "throttle不允许为null或者小于0")
@ApiModelProperty(value = "限流值", example = "102400000")
private Long throttleUnitB;
@ApiModelProperty("topic黑名单")
private List<String> topicBlackList;
@ApiModelProperty("均衡区间详情")
private List<ClusterBalanceIntervalDTO> clusterBalanceIntervalList;
@ApiModelProperty("指标计算周期，单位分钟")
private Integer metricCalculationPeriod;
@ApiModelProperty("任务并行数")
private Integer parallelNum;
@ApiModelProperty("执行策略， 1：优先最大副本，2：优先最小副本")
private Integer executionStrategy;
@ApiModelProperty("备注说明")
private String description;
@ApiModelProperty("是否是周期性任务")
private boolean scheduleJob;
}

79
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/job/detail/AbstractClusterBalanceDetailData.java
View File

@@ -1,79 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.detail;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import lombok.Data;
import java.util.List;
/**
* @author zengqiao
* @date 22/05/06
*/
@Data
@EnterpriseLoadReBalance
public abstract class AbstractClusterBalanceDetailData {
/**
* 物流集群ID
*/
private Long clusterPhyId;
/**
* Topic名称
*/
private String topicName;
/**
* 源Broker列表
*/
private List<Integer> originalBrokerIdList;
/**
* 目标Broker列表
*/
private List<Integer> reassignBrokerIdList;
/**
* 需迁移LogSize
*/
private Long needReassignLogSizeUnitB;
/**
* 已完成迁移LogSize
*/
private Long finishedReassignLogSizeUnitB;
/**
* 预计剩余时长
*/
private Long remainTimeUnitMs;
/**
* 当前副本数
*/
private Integer presentReplicaNum;
/**
* 新的副本数
*/
private Integer oldReplicaNum;
/**
* 新的副本数
*/
private Integer newReplicaNum;
/**
* 原本保存时间
*/
private Long originalRetentionTimeUnitMs;
/**
* 迁移时保存时间
*/
private Long reassignRetentionTimeUnitMs;
/**
* 状态
*/
private Integer status;
}

17
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/job/detail/ClusterBalanceDetailDataGroupByPartition.java
View File

@@ -1,17 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.detail;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import lombok.Data;
/**
* @author zengqiao
* @date 22/05/06
*/
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceDetailDataGroupByPartition extends AbstractClusterBalanceDetailData {
/**
* 分区ID
*/
private Integer partitionId;
}

22
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/job/detail/ClusterBalanceDetailDataGroupByTopic.java
View File

@@ -1,22 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.detail;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import lombok.Data;
import java.util.List;
/**
* @author zengqiao
* @date 22/05/06
*/
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceDetailDataGroupByTopic extends AbstractClusterBalanceDetailData {
/**
* 分区ID列表
*/
private List<Integer> partitionIdList;
private List<ClusterBalanceDetailDataGroupByPartition> reassignPartitionDetailsList;
}

76
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/entity/job/detail/ClusterBalancePlanDetail.java
View File

@@ -1,76 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.detail;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
/**
* 集群Topic信息
* @author zengqiao
* @date 22/02/23
*/
@Data
@EnterpriseLoadReBalance
@ApiModel(description = "集群均衡详细信息")
public class ClusterBalancePlanDetail implements Serializable {
@ApiModelProperty(value = "是否均衡，1：已均衡；2：未均衡")
private Integer status;
@ApiModelProperty(value = "brokerId")
private Integer brokerId;
@ApiModelProperty(value = "broker host")
private String host;
@ApiModelProperty(value = "均衡前 cpu")
private Double cpuBefore;
@ApiModelProperty(value = "均衡前 disk")
private Double diskBefore;
@ApiModelProperty(value = "均衡前 byteIn")
private Double byteInBefore;
@ApiModelProperty(value = "均衡前 byteOut")
private Double byteOutBefore;
@ApiModelProperty(value = "均衡后 cpu")
private Double cpuAfter;
@ApiModelProperty(value = "是否均衡，1：已均衡；2：未均衡")
private Integer cpuStatus;
@ApiModelProperty(value = "均衡后 disk")
private Double diskAfter;
@ApiModelProperty(value = "是否均衡，1：已均衡；2：未均衡")
private Integer diskStatus;
@ApiModelProperty(value = "均衡后 byteIn")
private Double byteInAfter;
@ApiModelProperty(value = "是否均衡，1：已均衡；2：未均衡")
private Integer byteInStatus;
@ApiModelProperty(value = "均衡后 byteOut")
private Double byteOutAfter;
@ApiModelProperty(value = "是否均衡，1：已均衡；2：未均衡")
private Integer byteOutStatus;
@ApiModelProperty(value = "均衡流入大小")
private Double inSize;
@ApiModelProperty(value = "均衡流入副本个数")
private Double inReplica;
@ApiModelProperty(value = "均衡流出大小")
private Double outSize;
@ApiModelProperty(value = "均衡流出副本个数")
private Double outReplica;
}

85
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/po/ClusterBalanceJobConfigPO.java
View File

@@ -1,85 +0,0 @@
/*
* Copyright (c) 2015, WINIT and/or its affiliates. All rights reserved. Use, Copy is subject to authorized license.
*/
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.po;
import com.baomidou.mybatisplus.annotation.TableName;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.bean.po.BasePO;
import com.xiaojukeji.know.streaming.km.common.constant.Constant;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* 集群均衡任务 实体类
*
* @author fengqiongfeng
* @date 2022-05-23
*/
@Data
@EnterpriseLoadReBalance
@NoArgsConstructor
@TableName(Constant.MYSQL_TABLE_NAME_PREFIX + "cluster_balance_job_config")
public class ClusterBalanceJobConfigPO extends BasePO {
/**
* 序列化版本号
*/
private static final long serialVersionUID=1L;
/**
* 集群id
*/
private Long clusterId;
/**
* topic黑名单
*/
private String topicBlackList;
/**
* 任务周期
*/
private String taskCron;
/**
* 均衡区间详情
*/
private String balanceIntervalJson;
/**
* 指标计算周期，单位分钟
*/
private Integer metricCalculationPeriod;
/**
* 迁移脚本
*/
private String reassignmentJson;
/**
* 任务并行数
*/
private Integer parallelNum;
/**
* 执行策略， 1：优先最大副本，2：优先最小副本
*/
private Integer executionStrategy;
/**
* 限流值
*/
private Long throttleUnitB;
/**
* 操作人
*/
private String creator;
/**
* 任务状态 0：未开启，1：开启
*/
private Integer status;
}

125
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/po/ClusterBalanceJobPO.java
View File

@@ -1,125 +0,0 @@
/*
* Copyright (c) 2015, WINIT and/or its affiliates. All rights reserved. Use, Copy is subject to authorized license.
*/
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.po;
import com.baomidou.mybatisplus.annotation.TableName;
import com.xiaojukeji.know.streaming.km.common.bean.po.BasePO;
import com.xiaojukeji.know.streaming.km.common.constant.Constant;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Date;
/**
* 集群均衡任务 实体类
*
* @author fengqiongfeng
* @date 2022-05-23
*/
@Data
@NoArgsConstructor
@TableName(Constant.MYSQL_TABLE_NAME_PREFIX + "cluster_balance_job")
public class ClusterBalanceJobPO extends BasePO {
/**
* 序列化版本号
*/
private static final long serialVersionUID=1L;
/**
* 集群id
*/
private Long clusterId;
/**
* 均衡节点
*/
private String brokers;
/**
* topic黑名单
*/
private String topicBlackList;
/**
* 1:立即均衡，2：周期均衡
*/
private Integer type;
/**
* 均衡区间详情
*/
private String balanceIntervalJson;
/**
* 指标计算周期，单位分钟
*/
private Integer metricCalculationPeriod;
/**
* 迁移脚本
*/
private String reassignmentJson;
/**
* 任务并行数
*/
private Integer parallelNum;
/**
* 执行策略， 1：优先最大副本，2：优先最小副本
*/
private Integer executionStrategy;
/**
* 限流值
*/
private Long throttleUnitB;
/**
* 总迁移大小
*/
private Double totalReassignSize;
/**
* 总迁移副本数
*/
private Integer totalReassignReplicaNum;
/**
* 移入topic
*/
private String moveInTopicList;
/**
* 节点均衡详情
*/
private String brokerBalanceDetail;
/**
* 任务状态 1：进行中，2：准备，3，成功，4：失败，5：取消
*/
private Integer status;
/**
* 操作人
*/
private String creator;
/**
* 任务开始时间
*/
private Date startTime;
/**
* 任务完成时间
*/
private Date finishedTime;
/**
* 备注说明
*/
private String description;
}

80
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/po/ClusterBalanceReassignPO.java
View File

@@ -1,80 +0,0 @@
/*
* Copyright (c) 2015, WINIT and/or its affiliates. All rights reserved. Use, Copy is subject to authorized license.
*/
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.po;
import com.baomidou.mybatisplus.annotation.TableName;
import com.xiaojukeji.know.streaming.km.common.bean.po.BasePO;
import com.xiaojukeji.know.streaming.km.common.constant.Constant;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Date;
/**
* 集群平衡迁移详情 实体类
*
* @author fengqiongfeng
* @date 2022-05-23
*/
@Data
@NoArgsConstructor
@TableName(Constant.MYSQL_TABLE_NAME_PREFIX + "cluster_balance_reassign")
public class ClusterBalanceReassignPO extends BasePO {
/**
* 序列化版本号
*/
private static final long serialVersionUID=1L;
/**
* jobID
*/
private Long jobId;
/**
* 集群id
*/
private Long clusterId;
/**
* Topic名称
*/
private String topicName;
/**
* 分区ID
*/
private Integer partitionId;
/**
* 源BrokerId列表
*/
private String originalBrokerIds;
/**
* 目标BrokerId列表
*/
private String reassignBrokerIds;
/**
* 任务开始时间
*/
private Date startTime;
/**
* 任务完成时间
*/
private Date finishedTime;
/**
* 扩展数据
*/
private String extendData;
/**
* 任务状态
*/
private Integer status;
}

24
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/vo/ClusterBalanceHistorySubVO.java
View File

@@ -1,24 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
/**
* 集群Topic信息
* @author zengqiao
* @date 22/02/23
*/
@Data
@EnterpriseLoadReBalance
@ApiModel(description = "集群均衡历史信息")
public class ClusterBalanceHistorySubVO implements Serializable {
@ApiModelProperty(value = "均衡成功节点数")
private Long successNu;
@ApiModelProperty(value = "未均衡成功节点数")
private Long failedNu;
}

34
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/vo/ClusterBalanceHistoryVO.java
View File

@@ -1,34 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
import java.util.Date;
import java.util.Map;
/**
* 集群Topic信息
* @author zengqiao
* @date 22/02/23
*/
@Data
@EnterpriseLoadReBalance
@ApiModel(description = "集群均衡历史信息")
public class ClusterBalanceHistoryVO implements Serializable {
@ApiModelProperty(value = "均衡开始执行时间")
private Date begin;
@ApiModelProperty(value = "均衡执行结束时间")
private Date end;
@ApiModelProperty(value = "均衡任务id")
private Long jobId;
@ApiModelProperty(value = "子项均衡历史信息", example = "cpu、disk")
private Map<String, ClusterBalanceHistorySubVO> sub;
}

20
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/vo/ClusterBalanceIntervalVO.java
View File

@@ -1,20 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceIntervalVO {
@ApiModelProperty("均衡维度:cpu,disk,bytesIn,bytesOut")
private String type;
@ApiModelProperty("平衡区间百分比")
private Double intervalPercent;
@ApiModelProperty("优先级")
private Integer priority;
}

55
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/vo/ClusterBalanceJobConfigVO.java
View File

@@ -1,55 +0,0 @@
/*
* Copyright (c) 2015, WINIT and/or its affiliates. All rights reserved. Use, Copy is subject to authorized license.
*/
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.ClusterBalanceInterval;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.util.List;
/**
* 集群均衡任务 实体类
*
* @author fengqiongfeng
* @date 2022-05-23
*/
@Data
@EnterpriseLoadReBalance
public class ClusterBalanceJobConfigVO {
/**
* 序列化版本号
*/
private static final long serialVersionUID=1L;
@ApiModelProperty("集群id")
private Long clusterId;
@ApiModelProperty("topic黑名单")
private List<String> topicBlackList;
@ApiModelProperty("任务周期")
private String scheduleCron;
@ApiModelProperty("均衡区间详情")
private List<ClusterBalanceInterval> clusterBalanceIntervalList;
@ApiModelProperty("指标计算周期，单位分钟")
private Integer metricCalculationPeriod;
@ApiModelProperty("任务并行数")
private Integer parallelNum;
@ApiModelProperty("执行策略， 1：优先最大副本，2：优先最小副本")
private Integer executionStrategy;
@ApiModelProperty("限流值")
private Long throttleUnitB;
@ApiModelProperty("任务状态 0：未开启，1：开启")
private Integer status;
}

32
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/vo/ClusterBalanceOverviewSubVO.java
View File

@@ -1,32 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
/**
* 集群Topic信息
* @author zengqiao
* @date 22/02/23
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
@EnterpriseLoadReBalance
@ApiModel(description = "集群均衡列表信息")
public class ClusterBalanceOverviewSubVO implements Serializable {
@ApiModelProperty(value = "平均值", example = "cpu的平均值，43.4")
private Double avg;
@ApiModelProperty(value = "规格", example = "1000")
private Double spec;
@ApiModelProperty(value = "均衡状态", example = "0:已均衡，-1:低于均衡值，1高于均衡值")
private Integer status ;
}

37
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/vo/ClusterBalanceOverviewVO.java
View File

@@ -1,37 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
import java.util.Map;
/**
* 集群Topic信息
* @author zengqiao
* @date 22/02/23
*/
@Data
@EnterpriseLoadReBalance
@ApiModel(description = "集群均衡列表信息")
public class ClusterBalanceOverviewVO implements Serializable {
@ApiModelProperty(value = "brokerId", example = "123")
private Integer brokerId;
@ApiModelProperty(value = "broker host")
private String host;
@ApiModelProperty(value = "broker 对应的 rack")
private String rack;
@ApiModelProperty(value = "leader")
private Integer leader;
@ApiModelProperty(value = "replicas")
private Integer replicas;
@ApiModelProperty(value = "子项统计详细信息", example = "cpu、disk")
private Map<String, ClusterBalanceOverviewSubVO> sub;
}

64
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/vo/ClusterBalancePlanDetailVO.java
View File

@@ -1,64 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
/**
* 集群Topic信息
* @author zengqiao
* @date 22/02/23
*/
@Data
@EnterpriseLoadReBalance
@ApiModel(description = "集群均衡历史信息")
public class ClusterBalancePlanDetailVO implements Serializable {
@ApiModelProperty(value = "是否均衡，0：已均衡；2：未均衡")
private Integer status;
@ApiModelProperty(value = "brokerId")
private Integer brokerId;
@ApiModelProperty(value = "broker host")
private String host;
@ApiModelProperty(value = "均衡前 cpu")
private Double cpuBefore;
@ApiModelProperty(value = "均衡前 disk")
private Double diskBefore;
@ApiModelProperty(value = "均衡前 byteIn")
private Double byteInBefore;
@ApiModelProperty(value = "均衡前 byteOut")
private Double byteOutBefore;
@ApiModelProperty(value = "均衡后 cpu")
private Double cpuAfter;
@ApiModelProperty(value = "均衡后 disk")
private Double diskAfter;
@ApiModelProperty(value = "均衡后 byteIn")
private Double byteInAfter;
@ApiModelProperty(value = "均衡后 byteOut")
private Double byteOutAfter;
@ApiModelProperty(value = "均衡流入大小")
private Double inSize;
@ApiModelProperty(value = "均衡流入副本个数")
private Double inReplica;
@ApiModelProperty(value = "均衡流出大小")
private Double outSize;
@ApiModelProperty(value = "均衡流出副本个数")
private Double outReplica;
}

49
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/vo/ClusterBalancePlanVO.java
View File

@@ -1,49 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
import java.util.List;
/**
* 集群Topic信息
* @author zengqiao
* @date 22/02/23
*/
@Data
@EnterpriseLoadReBalance
@ApiModel(description = "集群均衡信息")
public class ClusterBalancePlanVO implements Serializable {
@ApiModelProperty(value = "均衡计划类型，1：立即均衡；2：周期均衡")
private Integer type;
@ApiModelProperty(value = "均衡执行的节点范围")
private List<String> brokers;
@ApiModelProperty(value = "均衡执行的Topic黑名单")
private List<String> blackTopics;
@ApiModelProperty(value = "均衡执行移入的Topic名单")
private List<String> topics;
@ApiModelProperty(value = "均衡总迁移的磁盘大小，单位byte")
private Double moveSize;
@ApiModelProperty(value = "均衡总迁移的副本个数")
private Integer replicas;
@ApiModelProperty(value = "均衡阈值")
private String threshold;
@ApiModelProperty(value = "reassignment json")
private String reassignmentJson;
@ApiModelProperty(value = "均衡区间信息")
private List<ClusterBalanceIntervalVO> clusterBalanceIntervalList;
@ApiModelProperty(value = "均衡计划明细")
private List<ClusterBalancePlanDetailVO> detail;
}

32
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/vo/ClusterBalanceStateSubVO.java
View File

@@ -1,32 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@EnterpriseLoadReBalance
@AllArgsConstructor
@NoArgsConstructor
@ApiModel(description = "集群均衡状态子项的详细统计信息")
public class ClusterBalanceStateSubVO {
@ApiModelProperty(value = "平均值", example = "cpu的平均值，43.4")
private Double avg;
@ApiModelProperty(value = "周期均衡时的均衡区间", example = "cpu的均衡值")
private Double interval;
@ApiModelProperty(value = "处于周期均衡时的均衡区间的最小值以下的broker个数", example = "4")
private Long smallNu;
@ApiModelProperty(value = "处于周期均衡时的均衡区间的broker个数", example = "4")
private Long betweenNu;
@ApiModelProperty(value = "处于周期均衡时的均衡区间的最大值以上的broker个数", example = "4")
private Long bigNu;
}

32
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/bean/vo/ClusterBalanceStateVO.java
View File

@@ -1,32 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
import java.util.Date;
import java.util.Map;
/**
* 集群Topic信息
* @author zengqiao
* @date 22/02/23
*/
@Data
@EnterpriseLoadReBalance
@ApiModel(description = "集群均衡状态信息")
public class ClusterBalanceStateVO implements Serializable {
@ApiModelProperty(value = "均衡状态", example = "0:已均衡，2:未均衡")
private Integer status;
@ApiModelProperty(value = "是否开启均衡", example = "true:开启，false:未开启")
private Boolean enable;
@ApiModelProperty(value = "下次均衡开始时间")
private Date next;
@ApiModelProperty(value = "子项统计详细信息", example = "cpu、disk")
private Map<String, ClusterBalanceStateSubVO> sub;
}

476
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/converter/ClusterBalanceConverter.java
View File

@@ -1,476 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.converter;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.dto.ClusterBalanceIntervalDTO;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.dto.ClusterBalancePreviewDTO;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.dto.ClusterBalanceStrategyDTO;
import com.xiaojukeji.know.streaming.km.common.bean.entity.broker.Broker;
import com.xiaojukeji.know.streaming.km.common.bean.entity.broker.BrokerSpec;
import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.ClusterBalanceInterval;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.ClusterBalanceReassignExtendData;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.detail.ClusterBalancePlanDetail;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.content.JobClusterBalanceContent;
import com.xiaojukeji.know.streaming.km.common.bean.entity.topic.Topic;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.po.ClusterBalanceJobConfigPO;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.po.ClusterBalanceJobPO;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.po.ClusterBalanceReassignPO;
import com.xiaojukeji.know.streaming.km.common.constant.Constant;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.vo.*;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.enums.ClusterBalanceStateEnum;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.enums.ClusterBalanceTypeEnum;
import com.xiaojukeji.know.streaming.km.common.enums.job.JobStatusEnum;
import com.xiaojukeji.know.streaming.km.common.enums.job.JobTypeEnum;
import com.xiaojukeji.know.streaming.km.common.utils.CommonUtils;
import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
import com.xiaojukeji.know.streaming.km.rebalance.executor.common.*;
import com.xiaojukeji.know.streaming.km.rebalance.model.Resource;
import org.apache.kafka.clients.CommonClientConfigs;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
import static com.xiaojukeji.know.streaming.km.common.constant.ESConstant.*;
@EnterpriseLoadReBalance
public class ClusterBalanceConverter {
private ClusterBalanceConverter() {
}
public static BalanceParameter convert2BalanceParameter(ClusterBalanceJobConfigPO configPO, Map<Integer, Broker> brokerMap, Map<Integer, BrokerSpec> brokerSpecMap, ClusterPhy clusterPhy, String esUrl, List<String> topicNames) {
BalanceParameter balanceParameter = new BalanceParameter();
List<ClusterBalanceIntervalDTO> clusterBalanceIntervalDTOS = ConvertUtil.str2ObjArrayByJson(configPO.getBalanceIntervalJson(), ClusterBalanceIntervalDTO.class);
List<String> goals = new ArrayList<>();
for(ClusterBalanceIntervalDTO clusterBalanceIntervalDTO : clusterBalanceIntervalDTOS){
if (Resource.DISK.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setDiskThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.DISK.goal());
}else if (Resource.CPU.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setCpuThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
// todo cpu底层暂未实现，先不加goal
}else if (Resource.NW_IN.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setNetworkInThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.NW_IN.goal());
}else if (Resource.NW_OUT.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setNetworkOutThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.NW_OUT.goal());
}
}
balanceParameter.setGoals(goals);
balanceParameter.setCluster(clusterPhy.getId().toString());
balanceParameter.setExcludedTopics(configPO.getTopicBlackList());
balanceParameter.setEsIndexPrefix(PARTITION_INDEX + "_");
balanceParameter.setEsRestURL(esUrl);
balanceParameter.setBalanceBrokers(CommonUtils.intSet2String(brokerMap.keySet()));
balanceParameter.setHardwareEnv(convert2ListHostEnv(brokerMap, brokerSpecMap));
balanceParameter.setBeforeSeconds(configPO.getMetricCalculationPeriod());
balanceParameter.setIgnoredTopics(CommonUtils.strList2String(topicNames));
Properties kafkaConfig = new Properties();
kafkaConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterPhy.getBootstrapServers());
kafkaConfig.putAll(ConvertUtil.str2ObjByJson(clusterPhy.getClientProperties(), Properties.class));
balanceParameter.setKafkaConfig(kafkaConfig);
return balanceParameter;
}
public static BalanceParameter convert2BalanceParameter(ClusterBalanceJobPO clusterBalanceJobPO, Map<Integer, Broker> brokerMap, Map<Integer, BrokerSpec> brokerSpecMap, ClusterPhy clusterPhy, String esUrl, List<String> topicNames) {
BalanceParameter balanceParameter = new BalanceParameter();
List<ClusterBalanceIntervalDTO> clusterBalanceIntervalDTOS = ConvertUtil.str2ObjArrayByJson(clusterBalanceJobPO.getBalanceIntervalJson(), ClusterBalanceIntervalDTO.class);
List<String> goals = new ArrayList<>();
for(ClusterBalanceIntervalDTO clusterBalanceIntervalDTO : clusterBalanceIntervalDTOS){
if (Resource.DISK.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setDiskThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.DISK.goal());
}else if (Resource.CPU.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setCpuThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
// todo cpu底层暂未实现，先不加goal
}else if (Resource.NW_IN.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setNetworkInThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.NW_IN.goal());
}else if (Resource.NW_OUT.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setNetworkOutThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.NW_OUT.goal());
}
}
balanceParameter.setGoals(goals);
balanceParameter.setCluster(clusterPhy.getId().toString());
balanceParameter.setExcludedTopics(clusterBalanceJobPO.getTopicBlackList());
balanceParameter.setEsIndexPrefix(PARTITION_INDEX + "_");
balanceParameter.setEsRestURL(esUrl);
balanceParameter.setBalanceBrokers(clusterBalanceJobPO.getBrokers());
balanceParameter.setHardwareEnv(convert2ListHostEnv(brokerMap, brokerSpecMap));
balanceParameter.setBeforeSeconds(clusterBalanceJobPO.getMetricCalculationPeriod());
balanceParameter.setIgnoredTopics(CommonUtils.strList2String(topicNames));
Properties kafkaConfig = new Properties();
kafkaConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterPhy.getBootstrapServers());
kafkaConfig.putAll(ConvertUtil.str2ObjByJson(clusterPhy.getClientProperties(), Properties.class));
balanceParameter.setKafkaConfig(kafkaConfig);
return balanceParameter;
}
public static BalanceParameter convert2BalanceParameter(JobClusterBalanceContent dto, List<Broker> brokers, Map<Integer, BrokerSpec> brokerSpecMap, ClusterPhy clusterPhy, String esUrl, List<String> topicNames) {
BalanceParameter balanceParameter = new BalanceParameter();
List<ClusterBalanceIntervalDTO> clusterBalanceIntervalDTOS = dto.getClusterBalanceIntervalList().stream()
.sorted(Comparator.comparing(ClusterBalanceIntervalDTO::getPriority)).collect(Collectors.toList());
List<String> goals = new ArrayList<>();
for(ClusterBalanceIntervalDTO clusterBalanceIntervalDTO : clusterBalanceIntervalDTOS){
if (Resource.DISK.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setDiskThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.DISK.goal());
}else if (Resource.CPU.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setCpuThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
// todo cpu底层暂未实现，先不加goal
}else if (Resource.NW_IN.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setNetworkInThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.NW_IN.goal());
}else if (Resource.NW_OUT.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setNetworkOutThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.NW_OUT.goal());
}
}
Map<Integer, Broker> brokerMap = brokers.stream().collect(Collectors.toMap(Broker::getBrokerId, Function.identity()));
balanceParameter.setGoals(goals);
balanceParameter.setCluster(clusterPhy.getId().toString());
balanceParameter.setExcludedTopics(CommonUtils.strList2String(dto.getTopicBlackList()));
balanceParameter.setEsIndexPrefix(PARTITION_INDEX + "_");
balanceParameter.setEsRestURL(esUrl);
balanceParameter.setBalanceBrokers(CommonUtils.intSet2String(brokerMap.keySet()));
balanceParameter.setHardwareEnv(convert2ListHostEnv(brokerMap, brokerSpecMap));
balanceParameter.setBeforeSeconds(dto.getMetricCalculationPeriod());
balanceParameter.setIgnoredTopics(CommonUtils.strList2String(topicNames));
Properties kafkaConfig = new Properties();
kafkaConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterPhy.getBootstrapServers());
kafkaConfig.putAll(ConvertUtil.str2ObjByJson(clusterPhy.getClientProperties(), Properties.class));
balanceParameter.setKafkaConfig(kafkaConfig);
return balanceParameter;
}
public static BalanceParameter convert2BalanceParameter(ClusterBalancePreviewDTO dto, Map<Integer, Broker> brokerMap, Map<Integer, BrokerSpec> brokerSpecMap, ClusterPhy clusterPhy, String esUrl, List<String> topicNames) {
BalanceParameter balanceParameter = new BalanceParameter();
List<ClusterBalanceIntervalDTO> clusterBalanceIntervalDTOS = dto.getClusterBalanceIntervalList().stream()
.sorted(Comparator.comparing(ClusterBalanceIntervalDTO::getPriority)).collect(Collectors.toList());
List<String> goals = new ArrayList<>();
for(ClusterBalanceIntervalDTO clusterBalanceIntervalDTO : clusterBalanceIntervalDTOS){
if (Resource.DISK.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setDiskThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.DISK.goal());
}else if (Resource.CPU.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setCpuThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
// todo cpu底层暂未实现，先不加goal
}else if (Resource.NW_IN.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setNetworkInThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.NW_IN.goal());
}else if (Resource.NW_OUT.resource().equals(clusterBalanceIntervalDTO.getType())){
balanceParameter.setNetworkOutThreshold(clusterBalanceIntervalDTO.getIntervalPercent()/100);
goals.add(BalanceGoal.NW_OUT.goal());
}
}
balanceParameter.setGoals(goals);
balanceParameter.setCluster(clusterPhy.getId().toString());
balanceParameter.setExcludedTopics(CommonUtils.strList2String(dto.getTopicBlackList()));
balanceParameter.setEsIndexPrefix(PARTITION_INDEX + "_");
balanceParameter.setEsRestURL(esUrl);
balanceParameter.setBalanceBrokers(CommonUtils.intList2String(dto.getBrokers()));
balanceParameter.setHardwareEnv(convert2ListHostEnv(brokerMap, brokerSpecMap));
balanceParameter.setBeforeSeconds(dto.getMetricCalculationPeriod());
balanceParameter.setIgnoredTopics(CommonUtils.strList2String(topicNames));
Properties kafkaConfig = new Properties();
kafkaConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterPhy.getBootstrapServers());
kafkaConfig.putAll(ConvertUtil.str2ObjByJson(clusterPhy.getClientProperties(), Properties.class));
balanceParameter.setKafkaConfig(kafkaConfig);
return balanceParameter;
}
public static ClusterBalanceJobPO convert2ClusterBalanceJobPO(Long jobId, JobClusterBalanceContent jobDTO, OptimizerResult optimizerResult, List<Broker> brokers, String operator, String json) {
if (ValidateUtils.anyNull(jobDTO, optimizerResult, optimizerResult.resultJsonOverview(),
optimizerResult.resultJsonDetailed(), optimizerResult.resultDetailed(), optimizerResult.resultJsonTask())){
return null;
}
ClusterBalanceJobPO clusterBalanceJobPO = new ClusterBalanceJobPO();
clusterBalanceJobPO.setId(jobId);
clusterBalanceJobPO.setType(jobDTO.isScheduleJob()?
ClusterBalanceTypeEnum.CYCLE.getType():ClusterBalanceTypeEnum.IMMEDIATELY.getType());
clusterBalanceJobPO.setStatus(JobStatusEnum.WAITING.getStatus());
clusterBalanceJobPO.setCreator(operator);
clusterBalanceJobPO.setParallelNum(jobDTO.getParallelNum());
clusterBalanceJobPO.setThrottleUnitB(jobDTO.getThrottleUnitB());
clusterBalanceJobPO.setDescription(jobDTO.getDescription());
clusterBalanceJobPO.setBrokers(CommonUtils.intList2String(brokers.stream().map(Broker::getBrokerId).collect(Collectors.toList())));
clusterBalanceJobPO.setClusterId(jobDTO.getClusterId());
clusterBalanceJobPO.setTopicBlackList(CommonUtils.strList2String(jobDTO.getTopicBlackList()));
clusterBalanceJobPO.setMoveInTopicList(optimizerResult.resultOverview().getMoveTopics());
clusterBalanceJobPO.setExecutionStrategy(jobDTO.getExecutionStrategy());
clusterBalanceJobPO.setBalanceIntervalJson(ConvertUtil.obj2Json(jobDTO.getClusterBalanceIntervalList()));
clusterBalanceJobPO.setBrokerBalanceDetail(ConvertUtil.obj2Json(convert2ClusterBalancePlanDetail(optimizerResult.resultDetailed())));
clusterBalanceJobPO.setMetricCalculationPeriod(jobDTO.getMetricCalculationPeriod());
clusterBalanceJobPO.setReassignmentJson(json);
clusterBalanceJobPO.setTotalReassignSize(optimizerResult.resultOverview().getTotalMoveSize());
clusterBalanceJobPO.setTotalReassignReplicaNum(optimizerResult.resultOverview().getMoveReplicas());
clusterBalanceJobPO.setDescription(optimizerResult.resultJsonBalanceActionHistory());
return clusterBalanceJobPO;
}
public static ClusterBalanceReassignPO convert2ClusterBalanceReassignPO(BalanceTask balanceTask, Topic topic, Long jobId, Long clusterId) {
ClusterBalanceReassignPO reassignPO = new ClusterBalanceReassignPO();
reassignPO.setClusterId(clusterId);
reassignPO.setJobId(jobId);
reassignPO.setPartitionId(balanceTask.getPartition());
reassignPO.setOriginalBrokerIds(CommonUtils.intList2String(topic.getPartitionMap().get(balanceTask.getPartition())));
reassignPO.setReassignBrokerIds(CommonUtils.intList2String(balanceTask.getReplicas()));
reassignPO.setTopicName(balanceTask.getTopic());
ClusterBalanceReassignExtendData extendData = new ClusterBalanceReassignExtendData();
extendData.setOriginalRetentionTimeUnitMs(topic.getRetentionMs());
extendData.setReassignRetentionTimeUnitMs(topic.getRetentionMs());
extendData.setOriginReplicaNum(topic.getReplicaNum());
extendData.setReassignReplicaNum(balanceTask.getReplicas().size());
reassignPO.setExtendData(ConvertUtil.obj2Json(extendData));
reassignPO.setStatus(JobStatusEnum.WAITING.getStatus());
return reassignPO;
}
public static List<ClusterBalanceReassignPO> convert2ListClusterBalanceReassignPO(List<BalanceTask> balanceTasks, Map<String, Topic> topicMap, Long jobId, Long clusterId) {
List<ClusterBalanceReassignPO> reassignPOs = new ArrayList<>();
//生成迁移详情
Map<String, List<BalanceTask>> balanceTaskMap = balanceTasks.stream().collect(Collectors.groupingBy(BalanceTask::getTopic));
for (Map.Entry<String, List<BalanceTask>> entry : balanceTaskMap.entrySet()){
Topic topic = topicMap.get(entry.getKey());
if (topic == null || topic.getPartitionMap() == null){
continue;
}
for (BalanceTask balanceTask : entry.getValue()){
reassignPOs.add(ClusterBalanceConverter.convert2ClusterBalanceReassignPO(balanceTask, topic, jobId, clusterId));
}
}
return reassignPOs;
}
public static ClusterBalanceJobConfigPO convert2ClusterBalanceJobConfigPO(ClusterBalanceStrategyDTO dto, String operator) {
ClusterBalanceJobConfigPO jobConfigPO = new ClusterBalanceJobConfigPO();
jobConfigPO.setCreator(operator);
jobConfigPO.setParallelNum(dto.getParallelNum());
jobConfigPO.setThrottleUnitB(dto.getThrottleUnitB());
jobConfigPO.setClusterId(dto.getClusterId());
jobConfigPO.setExecutionStrategy(dto.getExecutionStrategy());
jobConfigPO.setBalanceIntervalJson(ConvertUtil.obj2Json(dto.getClusterBalanceIntervalList()));
jobConfigPO.setTaskCron(dto.getScheduleCron());
jobConfigPO.setMetricCalculationPeriod(dto.getMetricCalculationPeriod());
jobConfigPO.setStatus(dto.getStatus());
return jobConfigPO;
}
public static JobClusterBalanceContent convert2JobClusterBalanceContent(ClusterBalanceJobConfigPO configPO) {
JobClusterBalanceContent content = new JobClusterBalanceContent();
content.setType(JobTypeEnum.CLUSTER_BALANCE.getType());
content.setParallelNum(configPO.getParallelNum());
content.setThrottleUnitB(configPO.getThrottleUnitB());
content.setClusterId(configPO.getClusterId());
content.setExecutionStrategy(configPO.getExecutionStrategy());
content.setClusterBalanceIntervalList(ConvertUtil.str2ObjArrayByJson(configPO.getBalanceIntervalJson(), ClusterBalanceIntervalDTO.class));
content.setMetricCalculationPeriod(configPO.getMetricCalculationPeriod());
content.setTopicBlackList(CommonUtils.string2StrList(configPO.getTopicBlackList()));
content.setScheduleJob(Boolean.TRUE);
return content;
}
public static List<ClusterBalancePlanDetail> convert2ClusterBalancePlanDetail(Map<Integer, BalanceDetailed> detailedMap) {
List<ClusterBalancePlanDetail> details = new ArrayList<>();
for(Map.Entry<Integer, BalanceDetailed> entry : detailedMap.entrySet()){
BalanceDetailed balanceDetailed = entry.getValue();
if (balanceDetailed == null){
continue ;
}
ClusterBalancePlanDetail planDetail = new ClusterBalancePlanDetail();
planDetail.setStatus(balanceDetailed.getBalanceState()==ClusterBalanceStateEnum.BALANCE.getState()?ClusterBalanceStateEnum.BALANCE.getState():ClusterBalanceStateEnum.UNBALANCED.getState());
planDetail.setHost(balanceDetailed.getHost());
planDetail.setBrokerId(entry.getKey());
planDetail.setCpuBefore(balanceDetailed.getCurrentCPUUtilization()*Constant.ONE_HUNDRED);
planDetail.setCpuAfter(balanceDetailed.getLastCPUUtilization()*Constant.ONE_HUNDRED);
planDetail.setDiskBefore(balanceDetailed.getCurrentDiskUtilization()*Constant.ONE_HUNDRED);
planDetail.setDiskAfter(balanceDetailed.getLastDiskUtilization()*Constant.ONE_HUNDRED);
planDetail.setByteInBefore(balanceDetailed.getCurrentNetworkInUtilization()*Constant.ONE_HUNDRED);
planDetail.setByteInAfter(balanceDetailed.getLastNetworkInUtilization()*Constant.ONE_HUNDRED);
planDetail.setByteOutBefore(balanceDetailed.getCurrentNetworkOutUtilization()*Constant.ONE_HUNDRED);
planDetail.setByteOutAfter(balanceDetailed.getLastNetworkOutUtilization()*Constant.ONE_HUNDRED);
planDetail.setInReplica(balanceDetailed.getMoveInReplicas());
planDetail.setOutReplica(balanceDetailed.getMoveOutReplicas());
planDetail.setInSize(balanceDetailed.getMoveInDiskSize());
planDetail.setOutSize(balanceDetailed.getMoveOutDiskSize());
details.add(planDetail);
}
return details;
}
//更新平衡任务完成后的集群均衡状态
public static List<ClusterBalancePlanDetail> convert2ClusterBalancePlanDetail(List<ClusterBalancePlanDetail> details, Map<Integer, BrokerBalanceState> stateMap) {
details.forEach(planDetail ->{
BrokerBalanceState state = stateMap.get(planDetail.getBrokerId());
if (state == null){
return;
}
planDetail.setCpuStatus(state.getCpuBalanceState());
planDetail.setDiskStatus(state.getDiskBalanceState());
planDetail.setByteInStatus(state.getBytesInBalanceState());
planDetail.setByteOutStatus(state.getBytesOutBalanceState());
if ((state.getCpuBalanceState() == null || ClusterBalanceStateEnum.BALANCE.getState().equals(state.getCpuBalanceState()))
&& (state.getDiskBalanceState() == null || ClusterBalanceStateEnum.BALANCE.getState().equals(state.getDiskBalanceState()))
&& (state.getBytesInBalanceState() == null || ClusterBalanceStateEnum.BALANCE.getState().equals(state.getBytesInBalanceState()))
&& (state.getBytesOutBalanceState() == null || ClusterBalanceStateEnum.BALANCE.getState().equals(state.getBytesOutBalanceState()))) {
planDetail.setStatus(ClusterBalanceStateEnum.BALANCE.getState());
}else {
planDetail.setStatus(ClusterBalanceStateEnum.UNBALANCED.getState());
}
});
return details;
}
public static List<ClusterBalancePlanDetailVO> convert2ClusterBalancePlanDetailVO(List<Integer> balanceBrokerIds, Map<Integer, BalanceDetailed> detailedMap) {
List<ClusterBalancePlanDetailVO> detailVOS = new ArrayList<>();
for(Map.Entry<Integer, BalanceDetailed> entry : detailedMap.entrySet()){
BalanceDetailed value = entry.getValue();
if (value == null || !balanceBrokerIds.contains(entry.getKey())){
continue ;
}
ClusterBalancePlanDetailVO planDetailVO = new ClusterBalancePlanDetailVO();
planDetailVO.setStatus(value.getBalanceState()==ClusterBalanceStateEnum.BALANCE.getState()?ClusterBalanceStateEnum.BALANCE.getState():ClusterBalanceStateEnum.UNBALANCED.getState());
planDetailVO.setHost(value.getHost());
planDetailVO.setBrokerId(entry.getKey());
planDetailVO.setCpuBefore(value.getCurrentCPUUtilization()*Constant.ONE_HUNDRED);
planDetailVO.setCpuAfter(value.getLastCPUUtilization()*Constant.ONE_HUNDRED);
planDetailVO.setDiskBefore(value.getCurrentDiskUtilization()*Constant.ONE_HUNDRED);
planDetailVO.setDiskAfter(value.getLastDiskUtilization()*Constant.ONE_HUNDRED);
planDetailVO.setByteInBefore(value.getCurrentNetworkInUtilization()*Constant.ONE_HUNDRED);
planDetailVO.setByteInAfter(value.getLastNetworkInUtilization()*Constant.ONE_HUNDRED);
planDetailVO.setByteOutBefore(value.getCurrentNetworkOutUtilization()*Constant.ONE_HUNDRED);
planDetailVO.setByteOutAfter(value.getLastNetworkOutUtilization()*Constant.ONE_HUNDRED);
planDetailVO.setInReplica(value.getMoveInReplicas());
planDetailVO.setOutReplica(value.getMoveOutReplicas());
planDetailVO.setInSize(value.getMoveInDiskSize());
planDetailVO.setOutSize(value.getMoveOutDiskSize());
detailVOS.add(planDetailVO);
}
return detailVOS;
}
public static ClusterBalancePlanVO convert2ClusterBalancePlanVO(ClusterBalancePreviewDTO jobDTO, OptimizerResult optimizerResult, List<Broker> allBrokers) {
if (ValidateUtils.anyNull(jobDTO, optimizerResult, optimizerResult.resultJsonOverview(),
optimizerResult.resultJsonDetailed(), optimizerResult.resultDetailed(), optimizerResult.resultJsonTask())){
return null;
}
ClusterBalancePlanVO planVO = new ClusterBalancePlanVO();
planVO.setTopics(CommonUtils.string2StrList(optimizerResult.resultOverview().getMoveTopics()));
planVO.setType(ClusterBalanceTypeEnum.IMMEDIATELY.getType());
planVO.setReplicas(optimizerResult.resultOverview().getMoveReplicas());
planVO.setBlackTopics(jobDTO.getTopicBlackList());
planVO.setMoveSize(optimizerResult.resultOverview().getTotalMoveSize());
planVO.setThreshold(ConvertUtil.obj2Json(jobDTO.getClusterBalanceIntervalList()));
planVO.setBrokers(convert2HostList(allBrokers, optimizerResult.resultOverview().getNodeRange()));
planVO.setDetail(convert2ClusterBalancePlanDetailVO(jobDTO.getBrokers(), optimizerResult.resultDetailed()));
planVO.setClusterBalanceIntervalList(ConvertUtil.list2List(jobDTO.getClusterBalanceIntervalList(), ClusterBalanceIntervalVO.class));
planVO.setReassignmentJson(optimizerResult.resultJsonTask());
return planVO;
}
public static ClusterBalancePreviewDTO convert2ClusterBalancePreviewDTO(ClusterBalanceJobPO clusterBalanceJobPO) {
ClusterBalancePreviewDTO planVO = new ClusterBalancePreviewDTO();
planVO.setBrokers(CommonUtils.string2IntList(clusterBalanceJobPO.getBrokers()));
planVO.setClusterBalanceIntervalList(ConvertUtil.str2ObjArrayByJson(clusterBalanceJobPO.getBalanceIntervalJson(), ClusterBalanceIntervalDTO.class));
planVO.setClusterId(clusterBalanceJobPO.getClusterId());
planVO.setExecutionStrategy(clusterBalanceJobPO.getExecutionStrategy());
planVO.setParallelNum(clusterBalanceJobPO.getParallelNum());
planVO.setThrottleUnitB(clusterBalanceJobPO.getThrottleUnitB());
planVO.setMetricCalculationPeriod(clusterBalanceJobPO.getMetricCalculationPeriod());
planVO.setTopicBlackList(CommonUtils.string2StrList(clusterBalanceJobPO.getTopicBlackList()));
return planVO;
}
public static Map<String, ClusterBalanceOverviewSubVO> convert2MapClusterBalanceOverviewSubVO(BrokerSpec brokerSpec, BrokerBalanceState state) {
Map<String, ClusterBalanceOverviewSubVO> subVOMap = new HashMap<>();
if (brokerSpec == null){
brokerSpec = new BrokerSpec();
}
if (state == null){
state = new BrokerBalanceState();
}
Double cpuSpec = brokerSpec.getCpu()!=null?brokerSpec.getCpu()*Constant.ONE_HUNDRED:null;//转成基础单位
subVOMap.put(Resource.DISK.resource(),
new ClusterBalanceOverviewSubVO(
state.getDiskAvgResource(), brokerSpec.getDisk(),
state.getDiskBalanceState() == null || state.getDiskBalanceState().equals(ClusterBalanceStateEnum.BALANCE.getState())?state.getDiskBalanceState():ClusterBalanceStateEnum.UNBALANCED.getState()));
subVOMap.put(Resource.CPU.resource(),
new ClusterBalanceOverviewSubVO(state.getCpuAvgResource(), cpuSpec,
state.getCpuBalanceState() == null || state.getCpuBalanceState().equals(ClusterBalanceStateEnum.BALANCE.getState())?state.getCpuBalanceState():ClusterBalanceStateEnum.UNBALANCED.getState()));
subVOMap.put(Resource.NW_IN.resource(),
new ClusterBalanceOverviewSubVO(
state.getBytesInAvgResource(), brokerSpec.getFlow(),
state.getBytesInBalanceState() == null || state.getBytesInBalanceState().equals(ClusterBalanceStateEnum.BALANCE.getState())?state.getBytesInBalanceState():ClusterBalanceStateEnum.UNBALANCED.getState()));
subVOMap.put(Resource.NW_OUT.resource(),
new ClusterBalanceOverviewSubVO(
state.getBytesOutAvgResource(), brokerSpec.getFlow(),
state.getBytesOutBalanceState() == null || state.getBytesOutBalanceState().equals(ClusterBalanceStateEnum.BALANCE.getState())?state.getBytesOutBalanceState():ClusterBalanceStateEnum.UNBALANCED.getState()));
return subVOMap;
}
public static ClusterBalanceJobConfigVO convert2ClusterBalanceJobConfigVO(ClusterBalanceJobConfigPO clusterBalanceJobConfigPO){
ClusterBalanceJobConfigVO configVO = new ClusterBalanceJobConfigVO();
configVO.setScheduleCron(clusterBalanceJobConfigPO.getTaskCron());
configVO.setClusterBalanceIntervalList(ConvertUtil.str2ObjArrayByJson(clusterBalanceJobConfigPO.getBalanceIntervalJson(), ClusterBalanceInterval.class));
configVO.setClusterId(clusterBalanceJobConfigPO.getClusterId());
configVO.setExecutionStrategy(clusterBalanceJobConfigPO.getExecutionStrategy());
configVO.setParallelNum(clusterBalanceJobConfigPO.getParallelNum());
configVO.setMetricCalculationPeriod(clusterBalanceJobConfigPO.getMetricCalculationPeriod());
configVO.setThrottleUnitB(clusterBalanceJobConfigPO.getThrottleUnitB());
configVO.setTopicBlackList(CommonUtils.string2StrList(clusterBalanceJobConfigPO.getTopicBlackList()));
configVO.setStatus(clusterBalanceJobConfigPO.getStatus());
return configVO;
}
public static List<String> convert2HostList(List<Broker> allBrokers, String brokerIdStr){
if (allBrokers.isEmpty() || ValidateUtils.isBlank(brokerIdStr)){
return new ArrayList<>();
}
List<Integer> brokerIds = CommonUtils.string2IntList(brokerIdStr);
return allBrokers.stream().filter(broker -> brokerIds.contains(broker.getBrokerId()))
.map(Broker::getHost).collect(Collectors.toList());
}
private static List<HostEnv> convert2ListHostEnv(Map<Integer, Broker> brokerMap, Map<Integer, BrokerSpec> brokerSpecMap) {
List<HostEnv> hostEnvs = new ArrayList<>();
for (Map.Entry<Integer, Broker> entry : brokerMap.entrySet()) {
HostEnv hostEnv = new HostEnv();
hostEnv.setId(entry.getKey());
hostEnv.setHost(entry.getValue().getHost());
hostEnv.setRackId(entry.getValue().getRack());
BrokerSpec brokerSpec = brokerSpecMap.get(entry.getKey());
if (brokerSpec == null){
continue;
}
hostEnv.setCpu(brokerSpec.getCpu().intValue() * Constant.ONE_HUNDRED);
hostEnv.setDisk(brokerSpec.getDisk() * Constant.B_TO_GB);
hostEnv.setNetwork(brokerSpec.getFlow() * Constant.B_TO_MB);
hostEnvs.add(hostEnv);
}
return hostEnvs;
}
}

218
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/converter/ClusterBalanceReassignConverter.java
View File

@@ -1,218 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.converter;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.ClusterBalanceReassignDetail;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.ClusterBalanceReassignExtendData;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.detail.ClusterBalanceDetailDataGroupByPartition;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.entity.job.detail.ClusterBalanceDetailDataGroupByTopic;
import com.xiaojukeji.know.streaming.km.common.bean.entity.job.Job;
import com.xiaojukeji.know.streaming.km.common.bean.entity.job.JobStatus;
import com.xiaojukeji.know.streaming.km.common.bean.entity.job.detail.JobDetail;
import com.xiaojukeji.know.streaming.km.common.bean.entity.job.detail.SubJobReplicaMoveDetail;
import com.xiaojukeji.know.streaming.km.common.bean.entity.reassign.strategy.ReplaceReassignSub;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.po.ClusterBalanceJobPO;
import com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.bean.po.ClusterBalanceReassignPO;
import com.xiaojukeji.know.streaming.km.common.bean.vo.job.sub.SubJobClusterBalanceReplicaMoveVO;
import com.xiaojukeji.know.streaming.km.common.bean.vo.job.sub.SubJobPartitionDetailVO;
import com.xiaojukeji.know.streaming.km.common.bean.vo.job.sub.SubJobVO;
import com.xiaojukeji.know.streaming.km.common.enums.job.JobStatusEnum;
import com.xiaojukeji.know.streaming.km.common.utils.CommonUtils;
import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
import java.util.*;
import java.util.stream.Collectors;
@EnterpriseLoadReBalance
public class ClusterBalanceReassignConverter {
private ClusterBalanceReassignConverter() {
}
public static JobDetail convert2JobDetail(Job job, ClusterBalanceReassignDetail reassignDetail) {
JobDetail jobDetail = new JobDetail();
jobDetail.setId(job.getId());
jobDetail.setJobType(job.getJobType());
jobDetail.setJobName(job.getJobName());
jobDetail.setJobStatus(job.getJobStatus());
jobDetail.setPlanTime(job.getPlanTime());
jobDetail.setStartTime(reassignDetail.getStartTime());
jobDetail.setEndTime(reassignDetail.getFinishedTime());
jobDetail.setFlowLimit(reassignDetail.getThrottleUnitB().doubleValue());
JobStatus jobStatus = new JobStatus(reassignDetail.getReassignTopicDetailsList().stream().map(elem -> elem.getStatus()).collect(Collectors.toList()));
jobDetail.setTotal(jobStatus.getTotal());
jobDetail.setSuccess(jobStatus.getSuccess());
jobDetail.setFail(jobStatus.getFailed());
jobDetail.setDoing(jobStatus.getDoing());
List<SubJobVO> subJobDetailList = new ArrayList<>();
subJobDetailList.addAll(
ConvertUtil.list2List(convert2SubJobReplicaMoveDetailList(reassignDetail.getReassignTopicDetailsList()), SubJobClusterBalanceReplicaMoveVO.class)
);
jobDetail.setSubJobs(subJobDetailList);
return jobDetail;
}
public static ClusterBalanceReassignDetail convert2ClusterBalanceReassignDetail(ClusterBalanceJobPO jobPO, List<ClusterBalanceReassignPO> reassignPOS) {
// 按照Topic做聚合
Map<String, List<ClusterBalanceReassignPO>> topicJobPOMap = new HashMap<>();
reassignPOS.forEach(elem -> {
topicJobPOMap.putIfAbsent(elem.getTopicName(), new ArrayList<>());
topicJobPOMap.get(elem.getTopicName()).add(elem);
});
List<ClusterBalanceDetailDataGroupByTopic> reassignTopicDetailsList = new ArrayList<>();
for (Map.Entry<String, List<ClusterBalanceReassignPO>> entry: topicJobPOMap.entrySet()) {
reassignTopicDetailsList.add(convert2ClusterBalanceDetailDataGroupByTopic(entry.getValue()));
}
ClusterBalanceReassignDetail jobDetail = new ClusterBalanceReassignDetail();
jobDetail.setThrottleUnitB(jobPO.getThrottleUnitB());
jobDetail.setReassignTopicDetailsList(reassignTopicDetailsList);
jobDetail.setStartTime(jobPO.getStartTime());
if (JobStatusEnum.isFinished(jobPO.getStatus())) {
jobDetail.setFinishedTime(jobPO.getFinishedTime());
}
return jobDetail;
}
private static ClusterBalanceDetailDataGroupByTopic convert2ClusterBalanceDetailDataGroupByTopic(List<ClusterBalanceReassignPO> reassingns) {
Set<Integer> originalBrokerIdSet = new HashSet<>();
Set<Integer> reassignBrokerIdSet = new HashSet<>();
// 分区的信息
List<ClusterBalanceDetailDataGroupByPartition> partitionDetailList = new ArrayList<>();
for (ClusterBalanceReassignPO reassignPO : reassingns) {
ClusterBalanceDetailDataGroupByPartition detail = new ClusterBalanceDetailDataGroupByPartition();
detail.setPartitionId(reassignPO.getPartitionId());
detail.setClusterPhyId(reassignPO.getClusterId());
detail.setTopicName(reassignPO.getTopicName());
detail.setOriginalBrokerIdList(CommonUtils.string2IntList(reassignPO.getOriginalBrokerIds()));
detail.setReassignBrokerIdList(CommonUtils.string2IntList(reassignPO.getReassignBrokerIds()));
detail.setStatus(reassignPO.getStatus());
ClusterBalanceReassignExtendData extendData = ConvertUtil.str2ObjByJson(reassignPO.getExtendData(), ClusterBalanceReassignExtendData.class);
if (extendData != null) {
detail.setNeedReassignLogSizeUnitB(extendData.getNeedReassignLogSizeUnitB());
detail.setFinishedReassignLogSizeUnitB(extendData.getFinishedReassignLogSizeUnitB());
detail.setRemainTimeUnitMs(extendData.getRemainTimeUnitMs());
detail.setPresentReplicaNum(extendData.getOriginReplicaNum());
detail.setNewReplicaNum(extendData.getReassignReplicaNum());
detail.setOriginalRetentionTimeUnitMs(extendData.getOriginalRetentionTimeUnitMs());
detail.setReassignRetentionTimeUnitMs(extendData.getReassignRetentionTimeUnitMs());
}
originalBrokerIdSet.addAll(detail.getOriginalBrokerIdList());
reassignBrokerIdSet.addAll(detail.getReassignBrokerIdList());
partitionDetailList.add(detail);
}
// Topic的详细信息
ClusterBalanceDetailDataGroupByTopic topicDetail = new ClusterBalanceDetailDataGroupByTopic();
topicDetail.setPartitionIdList(partitionDetailList.stream().map(elem -> elem.getPartitionId()).collect(Collectors.toList()));
topicDetail.setReassignPartitionDetailsList(partitionDetailList);
topicDetail.setClusterPhyId(reassingns.get(0).getClusterId());
topicDetail.setTopicName(reassingns.get(0).getTopicName());
topicDetail.setOriginalBrokerIdList(new ArrayList<>(originalBrokerIdSet));
topicDetail.setReassignBrokerIdList(new ArrayList<>(reassignBrokerIdSet));
List<Long> needSizeList = partitionDetailList
.stream()
.filter(elem -> elem.getNeedReassignLogSizeUnitB() != null)
.map(item -> item.getNeedReassignLogSizeUnitB()).collect(Collectors.toList());
topicDetail.setNeedReassignLogSizeUnitB(needSizeList.isEmpty()? null: needSizeList.stream().reduce(Long::sum).get());
List<Long> finishedSizeList = partitionDetailList
.stream()
.filter(elem -> elem.getFinishedReassignLogSizeUnitB() != null)
.map(item -> item.getFinishedReassignLogSizeUnitB()).collect(Collectors.toList());
topicDetail.setFinishedReassignLogSizeUnitB(finishedSizeList.isEmpty()? null: finishedSizeList.stream().reduce(Long::sum).get());
List<Long> remainList = partitionDetailList
.stream()
.filter(elem -> elem.getRemainTimeUnitMs() != null)
.map(item -> item.getRemainTimeUnitMs()).collect(Collectors.toList());
topicDetail.setRemainTimeUnitMs(remainList.isEmpty()? null: remainList.stream().reduce(Long::max).get());
topicDetail.setPresentReplicaNum(partitionDetailList.get(0).getPresentReplicaNum());
topicDetail.setNewReplicaNum(partitionDetailList.get(0).getNewReplicaNum());
topicDetail.setOriginalRetentionTimeUnitMs(partitionDetailList.get(0).getOriginalRetentionTimeUnitMs());
topicDetail.setReassignRetentionTimeUnitMs(partitionDetailList.get(0).getReassignRetentionTimeUnitMs());
topicDetail.setStatus(
new JobStatus(
partitionDetailList.stream().map(elem -> elem.getStatus()).collect(Collectors.toList())
).getStatus()
);
return topicDetail;
}
public static List<SubJobPartitionDetailVO> convert2SubJobPartitionDetailVOList(ClusterBalanceDetailDataGroupByTopic detailDataGroupByTopic) {
List<SubJobPartitionDetailVO> voList = new ArrayList<>();
for (ClusterBalanceDetailDataGroupByPartition groupByPartition: detailDataGroupByTopic.getReassignPartitionDetailsList()) {
SubJobPartitionDetailVO vo = new SubJobPartitionDetailVO();
vo.setPartitionId(groupByPartition.getPartitionId());
vo.setSourceBrokerIds(groupByPartition.getOriginalBrokerIdList());
vo.setDesBrokerIds(groupByPartition.getReassignBrokerIdList());
vo.setTotalSize(groupByPartition.getNeedReassignLogSizeUnitB() != null ? groupByPartition.getNeedReassignLogSizeUnitB().doubleValue(): null);
vo.setMovedSize(groupByPartition.getFinishedReassignLogSizeUnitB() != null ? groupByPartition.getFinishedReassignLogSizeUnitB().doubleValue(): null);
vo.setStatus(groupByPartition.getStatus());
vo.setRemainTime(groupByPartition.getRemainTimeUnitMs());
voList.add(vo);
}
return voList;
}
private static List<SubJobReplicaMoveDetail> convert2SubJobReplicaMoveDetailList(List<ClusterBalanceDetailDataGroupByTopic> reassignTopicDetailsList) {
List<SubJobReplicaMoveDetail> detailList = new ArrayList<>();
for (ClusterBalanceDetailDataGroupByTopic detailDataGroupByTopic: reassignTopicDetailsList) {
SubJobReplicaMoveDetail detail = new SubJobReplicaMoveDetail();
detail.setTopicName(detailDataGroupByTopic.getTopicName());
detail.setPartitions(detailDataGroupByTopic.getPartitionIdList());
detail.setCurrentTimeSpent(detailDataGroupByTopic.getOriginalRetentionTimeUnitMs());
detail.setMoveTimeSpent(detailDataGroupByTopic.getReassignRetentionTimeUnitMs());
detail.setSourceBrokers(detailDataGroupByTopic.getOriginalBrokerIdList());
detail.setDesBrokers(detailDataGroupByTopic.getReassignBrokerIdList());
detail.setStatus(detailDataGroupByTopic.getStatus());
if (detailDataGroupByTopic.getNeedReassignLogSizeUnitB() != null) {
detail.setTotalSize(detailDataGroupByTopic.getNeedReassignLogSizeUnitB().doubleValue());
}
if (detailDataGroupByTopic.getFinishedReassignLogSizeUnitB() != null) {
detail.setMovedSize(detailDataGroupByTopic.getFinishedReassignLogSizeUnitB().doubleValue());
}
JobStatus jobStatus = new JobStatus(detailDataGroupByTopic.getReassignPartitionDetailsList().stream().map(elem -> elem.getStatus()).collect(Collectors.toList())); detail.setTotal(jobStatus.getTotal());
detail.setSuccess(jobStatus.getSuccess());
detail.setFail(jobStatus.getFailed());
detail.setDoing(jobStatus.getDoing());
detail.setRemainTime(detailDataGroupByTopic.getRemainTimeUnitMs());
detailList.add(detail);
}
return detailList;
}
public static List<ReplaceReassignSub> convert2ReplaceReassignSubList(List<ClusterBalanceReassignPO> reassignPOList) {
List<ReplaceReassignSub> voList = new ArrayList<>();
for (ClusterBalanceReassignPO reassignPO: reassignPOList) {
voList.add(convert2ReplaceReassignSub(reassignPO));
}
return voList;
}
public static ReplaceReassignSub convert2ReplaceReassignSub(ClusterBalanceReassignPO reassignPO) {
ReplaceReassignSub reassignSub = new ReplaceReassignSub();
reassignSub.setClusterPhyId(reassignPO.getClusterId());
reassignSub.setOriginalBrokerIdList(CommonUtils.string2IntList(reassignPO.getOriginalBrokerIds()));
reassignSub.setReassignBrokerIdList(CommonUtils.string2IntList(reassignPO.getReassignBrokerIds()));
reassignSub.setPartitionId(reassignPO.getPartitionId());
reassignSub.setTopicName(reassignPO.getTopicName());
return reassignSub;
}
}

31
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/enums/ClusterBalanceStateEnum.java
View File

@@ -1,31 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.enums;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import lombok.Getter;
/**
* 集群平衡状态
* @author zengqiao
* @date 22/03/08
*/
@Getter
@EnterpriseLoadReBalance
public enum ClusterBalanceStateEnum {
BELOW_BALANCE(-1, "低于均衡范围"),
BALANCE(0, "均衡范围内"),
ABOVE_BALANCE(1, "高于均衡范围"),
UNBALANCED(2, "不均衡"),
;
private final Integer state;
private final String message;
ClusterBalanceStateEnum(int state, String message) {
this.state = state;
this.message = message;
}
}

28
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/enums/ClusterBalanceTypeEnum.java
View File

@@ -1,28 +0,0 @@
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance.enums;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;
import lombok.Getter;
/**
* 集群平衡任务类型
* @author zengqiao
* @date 22/03/08
*/
@Getter
@EnterpriseLoadReBalance
public enum ClusterBalanceTypeEnum {
IMMEDIATELY(1, "立即"),
CYCLE(2, "周期"),
;
private final int type;
private final String message;
ClusterBalanceTypeEnum(int type, String message) {
this.type = type;
this.message = message;
}
}

9
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enterprise/rebalance/package-info.java
View File

@@ -1,9 +0,0 @@
/**
* Load-reBalance相关功能模块
* km-extends/km-rebalance 模块，是依据指标生成迁移 plan 的模块，是底层的一个基础功能
* 当前 package 模块是依据产品的要求，依赖 km-extends/km-rebalance 模块，构建产品实际使用功能
*/
@EnterpriseLoadReBalance
package com.xiaojukeji.know.streaming.km.common.enterprise.rebalance;
import com.xiaojukeji.know.streaming.km.common.annotations.enterprise.EnterpriseLoadReBalance;