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 'didi:master' into master

Browse Source
This commit is contained in:
lucasun
2022-01-18 14:07:44 +08:00
committed by GitHub
parent 859cf74bd6 a7f8c3ced3
commit 7372617b14
60 changed files with 950 additions and 506 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
container/helm/templates/configmap.yaml
View File

@@ -67,7 +67,16 @@ data:
# ldap settings
ldap:
enabled: false
authUserRegistration: false
url: ldap://127.0.0.1:389/
basedn: dc=tsign,dc=cn
factory: com.sun.jndi.ldap.LdapCtxFactory
filter: sAMAccountName
security:
authentication: simple
principal: cn=admin,dc=tsign,dc=cn
credentials: admin
auth-user-registration: false
auth-user-registration-role: normal
kcm:
enabled: false

1
distribution/conf/application.yml
View File

@@ -15,6 +15,7 @@ server:
spring:
application:
name: kafkamanager
version: @project.version@
profiles:
active: dev
datasource:

98
distribution/conf/application.yml.example
View File

@@ -15,6 +15,7 @@ server:
spring:
application:
name: kafkamanager
version: @project.version@
profiles:
active: dev
datasource:
@@ -26,7 +27,6 @@ spring:
main:
allow-bean-definition-overriding: true
servlet:
multipart:
max-file-size: 100MB
@@ -36,28 +36,32 @@ logging:
config: classpath:logback-spring.xml
custom:
idc: cn # 部署的数据中心, 忽略该配置, 后续会进行删除
jmx:
max-conn: 10 # 2.3版本配置不在这个地方生效
idc: cn
store-metrics-task:
community:
broker-metrics-enabled: true # 社区部分broker metrics信息收集开关, 关闭之后metrics信息将不会进行收集及写DB
topic-metrics-enabled: true # 社区部分topic的metrics信息收集开关, 关闭之后metrics信息将不会进行收集及写DB
didi:
app-topic-metrics-enabled: false # 滴滴埋入的指标, 社区AK不存在该指标，因此默认关闭
topic-request-time-metrics-enabled: false # 滴滴埋入的指标, 社区AK不存在该指标，因此默认关闭
topic-throttled-metrics-enabled: false # 滴滴埋入的指标, 社区AK不存在该指标，因此默认关闭
topic-metrics-enabled: true
didi: # 滴滴Kafka特有的指标
app-topic-metrics-enabled: false
topic-request-time-metrics-enabled: false
topic-throttled-metrics-enabled: false
# 任务相关的开关
# 任务相关的配置
task:
op:
sync-topic-enabled: false # 未落盘的Topic定期同步到DB中
order-auto-exec: # 工单自动化审批线程的开关
topic-enabled: false # Topic工单自动化审批开关, false:关闭自动化审批, true:开启
app-enabled: false # App工单自动化审批开关, false:关闭自动化审批, true:开启
sync-topic-enabled: false # 未落盘的Topic定期同步到DB中
order-auto-exec: # 工单自动化审批线程的开关
topic-enabled: false # Topic工单自动化审批开关, false:关闭自动化审批, true:开启
app-enabled: false # App工单自动化审批开关, false:关闭自动化审批, true:开启
metrics:
delete-metrics:
delete-limit-size: 1000
collect: # 收集指标
broker-metrics-enabled: true # 收集Broker指标
sink: # 上报指标
cluster-metrics: # 上报cluster指标
sink-db-enabled: true # 上报到db
broker-metrics: # 上报broker指标
sink-db-enabled: true # 上报到db
delete: # 删除指标
delete-limit-size: 1000 # 单次删除的批大小
cluster-metrics-save-days: 14 # 集群指标保存天数
broker-metrics-save-days: 14 # Broker指标保存天数
topic-metrics-save-days: 7 # Topic指标保存天数
@@ -65,8 +69,25 @@ task:
topic-throttled-metrics-save-days: 7 # Topic限流指标保存天数
app-topic-metrics-save-days: 7 # App+Topic指标保存天数
# ldap相关的配置
thread-pool:
collect-metrics:
thread-num: 256 # 收集指标线程池大小
queue-size: 5000 # 收集指标线程池的queue大小
api-call:
thread-num: 16 # api服务线程池大小
queue-size: 5000 # api服务线程池的queue大小
client-pool:
kafka-consumer:
min-idle-client-num: 24 # 最小空闲客户端数
max-idle-client-num: 24 # 最大空闲客户端数
max-total-client-num: 24 # 最大客户端数
borrow-timeout-unit-ms: 3000 # 租借超时时间，单位毫秒
account:
jump-login:
gateway-api: false # 网关接口
third-part-api: false # 第三方接口
ldap:
enabled: false
url: ldap://127.0.0.1:389/
@@ -80,28 +101,20 @@ account:
auth-user-registration: true
auth-user-registration-role: normal
# 集群升级部署相关的功能，需要配合夜莺及S3进行使用
kcm:
enabled: false
s3:
kcm: # 集群安装部署，仅安装broker
enabled: false # 是否开启
s3: # s3 存储服务
endpoint: s3.didiyunapi.com
access-key: 1234567890
secret-key: 0987654321
bucket: logi-kafka
n9e:
base-url: http://127.0.0.1:8004
user-token: 12345678
timeout: 300
account: root
script-file: kcm_script.sh
# 监控告警相关的功能，需要配合夜莺进行使用
# enabled: 表示是否开启监控告警的功能, true: 开启, false: 不开启
# n9e.nid: 夜莺的节点ID
# n9e.user-token: 用户的密钥，在夜莺的个人设置中
# n9e.mon.base-url: 监控地址
# n9e.sink.base-url: 数据上报地址
# n9e.rdb.base-url: 用户资源中心地址
n9e: # 夜莺
base-url: http://127.0.0.1:8004 # 夜莺job服务地址
user-token: 12345678 # 用户的token
timeout: 300 # 当台操作的超时时间
account: root # 操作时使用的账号
script-file: kcm_script.sh # 脚本，已内置好，在源码的kcm模块内，此处配置无需修改
logikm-url: http://127.0.0.1:8080 # logikm部署地址，部署时kcm_script.sh会调用logikm检查部署中的一些状态
monitor:
enabled: false
@@ -115,10 +128,9 @@ monitor:
rdb:
base-url: http://127.0.0.1:8000 # 夜莺v4版本，默认端口统一调整为了8000
notify: # 通知的功能
kafka: # 默认通知发送到kafka的指定Topic中
cluster-id: 95 # Topic的集群ID
topic-name: didi-kafka-notify # Topic名称
order: # 部署的KM的地址
detail-url: http://127.0.0.1
notify:
kafka:
cluster-id: 95
topic-name: didi-kafka-notify
order:
detail-url: http://127.0.0.1

6
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/ApiPrefix.java
View File

@@ -20,12 +20,6 @@ public class ApiPrefix {
// open
public static final String API_V1_THIRD_PART_PREFIX = API_V1_PREFIX + "third-part/";
// 开放给OP的接口, 后续对 应的接口的集群都需要是物理集群
public static final String API_V1_THIRD_PART_OP_PREFIX = API_V1_THIRD_PART_PREFIX + "op/";
// 开放给Normal的接口, 后续对应的接口的集群，都需要是逻辑集群
public static final String API_V1_THIRD_PART_NORMAL_PREFIX = API_V1_THIRD_PART_PREFIX + "normal/";
// gateway
public static final String GATEWAY_API_V1_PREFIX = "/gateway" + API_V1_PREFIX;

4
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/constant/KafkaConstant.java
View File

@@ -17,6 +17,10 @@ public class KafkaConstant {
public static final String RETENTION_MS_KEY = "retention.ms";
public static final String EXTERNAL_KEY = "EXTERNAL";
public static final String INTERNAL_KEY = "INTERNAL";
private KafkaConstant() {
}
}

18
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/entity/ao/common/IpPortData.java Normal file
View File

@@ -0,0 +1,18 @@
package com.xiaojukeji.kafka.manager.common.entity.ao.common;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
@Data
@NoArgsConstructor
@AllArgsConstructor
public class IpPortData implements Serializable {
private static final long serialVersionUID = -428897032994630685L;
private String ip;
private String port;
}

33
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/events/metrics/BaseMetricsCollectedEvent.java Normal file
View File

@@ -0,0 +1,33 @@
package com.xiaojukeji.kafka.manager.common.events.metrics;
import org.springframework.context.ApplicationEvent;
/**
* @author zengqiao
* @date 22/01/17
*/
public class BaseMetricsCollectedEvent extends ApplicationEvent {
/**
* 物理集群ID
*/
protected final Long physicalClusterId;
/**
* 收集时间，依据业务需要来设置，可以设置任务开始时间，也可以设置任务结束时间
*/
protected final Long collectTime;
public BaseMetricsCollectedEvent(Object source, Long physicalClusterId, Long collectTime) {
super(source);
this.physicalClusterId = physicalClusterId;
this.collectTime = collectTime;
}
public Long getPhysicalClusterId() {
return physicalClusterId;
}
public Long getCollectTime() {
return collectTime;
}
}

22
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/events/metrics/BatchBrokerMetricsCollectedEvent.java Normal file
View File

@@ -0,0 +1,22 @@
package com.xiaojukeji.kafka.manager.common.events.metrics;
import com.xiaojukeji.kafka.manager.common.entity.metrics.BrokerMetrics;
import java.util.List;
/**
* @author zengqiao
* @date 20/8/31
*/
public class BatchBrokerMetricsCollectedEvent extends BaseMetricsCollectedEvent {
private final List<BrokerMetrics> metricsList;
public BatchBrokerMetricsCollectedEvent(Object source, Long physicalClusterId, Long collectTime, List<BrokerMetrics> metricsList) {
super(source, physicalClusterId, collectTime);
this.metricsList = metricsList;
}
public List<BrokerMetrics> getMetricsList() {
return metricsList;
}
}

55
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/utils/BackoffUtils.java
View File

@@ -1,9 +1,18 @@
package com.xiaojukeji.kafka.manager.common.utils;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class BackoffUtils {
private BackoffUtils() {
}
/**
* 需要进行回退的事件信息
* <回退事件名，回退结束时间>
*/
private static final Map<String, Long> NEED_BACK_OFF_EVENT_MAP = new ConcurrentHashMap<>();
public static void backoff(long timeUnitMs) {
if (timeUnitMs <= 0) {
return;
@@ -17,4 +26,50 @@ public class BackoffUtils {
// ignore
}
}
/**
* 记录回退设置
* @param backoffEventKey 回退事件key
* @param backoffTimeUnitMs 回退时间(ms)
*/
public static void putNeedBackoffEvent(String backoffEventKey, Long backoffTimeUnitMs) {
if (backoffEventKey == null || backoffTimeUnitMs == null || backoffTimeUnitMs <= 0) {
return;
}
NEED_BACK_OFF_EVENT_MAP.put(backoffEventKey, backoffTimeUnitMs + System.currentTimeMillis());
}
/**
* 移除回退设置
* @param backoffEventKey 回退事件key
*/
public static void removeNeedBackoffEvent(String backoffEventKey) {
NEED_BACK_OFF_EVENT_MAP.remove(backoffEventKey);
}
/**
* 检查是否需要回退
* @param backoffEventKey 回退事件key
* @return
*/
public static boolean isNeedBackoff(String backoffEventKey) {
Long backoffEventEndTimeUnitMs = NEED_BACK_OFF_EVENT_MAP.get(backoffEventKey);
if (backoffEventEndTimeUnitMs == null) {
return false;
}
if (backoffEventEndTimeUnitMs > System.currentTimeMillis()) {
return true;
}
// 移除
try {
NEED_BACK_OFF_EVENT_MAP.remove(backoffEventKey, backoffEventEndTimeUnitMs);
} catch (Exception e) {
// 如果key不存在，这里可能出现NPE，不过不管什么异常都可以忽略
}
return false;
}
}

50
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/utils/jmx/JmxConfig.java
View File

@@ -1,5 +1,10 @@
package com.xiaojukeji.kafka.manager.common.utils.jmx;
import lombok.Data;
import lombok.ToString;
@Data
@ToString
public class JmxConfig {
/**
* 单台最大连接数
@@ -21,45 +26,8 @@ public class JmxConfig {
*/
private Boolean openSSL;
public Integer getMaxConn() {
return maxConn;
}
public void setMaxConn(Integer maxConn) {
this.maxConn = maxConn;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this.password = password;
}
public Boolean isOpenSSL() {
return openSSL;
}
public void setOpenSSL(Boolean openSSL) {
this.openSSL = openSSL;
}
@Override
public String toString() {
return "JmxConfig{" +
"maxConn=" + maxConn +
", username='" + username + '\'' +
", password='" + password + '\'' +
", openSSL=" + openSSL +
'}';
}
/**
* 连接重试回退事件
*/
private Long retryConnectBackoffTimeUnitMs;
}

80
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/utils/jmx/JmxConnectorWrap.java
View File

@@ -18,6 +18,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
/**
* JMXConnector包装类
@@ -41,6 +42,8 @@ public class JmxConnectorWrap {
private JmxConfig jmxConfig;
private final ReentrantLock modifyJMXConnectorLock = new ReentrantLock();
public JmxConnectorWrap(Long physicalClusterId, Integer brokerId, String host, int port, JmxConfig jmxConfig) {
this.physicalClusterId = physicalClusterId;
this.brokerId = brokerId;
@@ -51,7 +54,12 @@ public class JmxConnectorWrap {
this.jmxConfig = new JmxConfig();
}
if (ValidateUtils.isNullOrLessThanZero(this.jmxConfig.getMaxConn())) {
this.jmxConfig.setMaxConn(1);
// 默认设置20
this.jmxConfig.setMaxConn(20);
}
if (ValidateUtils.isNullOrLessThanZero(this.jmxConfig.getRetryConnectBackoffTimeUnitMs())) {
// 默认回退10分钟
this.jmxConfig.setRetryConnectBackoffTimeUnitMs(10 * 60 * 1000L);
}
this.atomicInteger = new AtomicInteger(this.jmxConfig.getMaxConn());
}
@@ -63,17 +71,40 @@ public class JmxConnectorWrap {
if (port == -1) {
return false;
}
return createJmxConnector();
return safeCreateJmxConnector();
}
public synchronized void close() {
public void close() {
this.closeJmxConnect();
}
public void closeJmxConnect() {
if (jmxConnector == null) {
return;
}
try {
modifyJMXConnectorLock.lock();
// 移除设置的backoff事件
BackoffUtils.removeNeedBackoffEvent(buildConnectJmxFailedBackoffEventKey(physicalClusterId, brokerId));
jmxConnector.close();
} catch (IOException e) {
LOGGER.warn("close JmxConnector exception, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port, e);
} catch (Exception e) {
LOGGER.error("close JmxConnector exception, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port, e);
} finally {
jmxConnector = null;
modifyJMXConnectorLock.unlock();
}
}
private boolean safeCreateJmxConnector() {
try {
modifyJMXConnectorLock.lock();
return createJmxConnector();
} finally {
modifyJMXConnectorLock.unlock();
}
}
@@ -81,6 +112,12 @@ public class JmxConnectorWrap {
if (jmxConnector != null) {
return true;
}
if (BackoffUtils.isNeedBackoff(buildConnectJmxFailedBackoffEventKey(physicalClusterId, brokerId))) {
// 被设置了需要进行回退，则本次不进行创建
return false;
}
String jmxUrl = String.format("service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi", host, port);
try {
Map<String, Object> environment = new HashMap<String, Object>();
@@ -88,7 +125,9 @@ public class JmxConnectorWrap {
// fixed by riyuetianmu
environment.put(JMXConnector.CREDENTIALS, new String[]{this.jmxConfig.getUsername(), this.jmxConfig.getPassword()});
}
if (jmxConfig.isOpenSSL() != null && this.jmxConfig.isOpenSSL()) {
if (jmxConfig.getOpenSSL() != null && this.jmxConfig.getOpenSSL()) {
// 开启ssl
environment.put(Context.SECURITY_PROTOCOL, "ssl");
SslRMIClientSocketFactory clientSocketFactory = new SslRMIClientSocketFactory();
environment.put(RMIConnectorServer.RMI_CLIENT_SOCKET_FACTORY_ATTRIBUTE, clientSocketFactory);
@@ -96,13 +135,17 @@ public class JmxConnectorWrap {
}
jmxConnector = JMXConnectorFactory.connect(new JMXServiceURL(jmxUrl), environment);
LOGGER.info("JMX connect success, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port);
LOGGER.info("connect JMX success, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port);
return true;
} catch (MalformedURLException e) {
LOGGER.error("JMX url exception, physicalClusterId:{} brokerId:{} host:{} port:{} jmxUrl:{}", physicalClusterId, brokerId, host, port, jmxUrl, e);
LOGGER.error("connect JMX failed, JMX url exception, physicalClusterId:{} brokerId:{} host:{} port:{} jmxUrl:{}.", physicalClusterId, brokerId, host, port, jmxUrl, e);
} catch (Exception e) {
LOGGER.error("JMX connect exception, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port, e);
LOGGER.error("connect JMX failed, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port, e);
}
// 设置连接backoff
BackoffUtils.putNeedBackoffEvent(buildConnectJmxFailedBackoffEventKey(physicalClusterId, brokerId), this.jmxConfig.getRetryConnectBackoffTimeUnitMs());
return false;
}
@@ -116,6 +159,11 @@ public class JmxConnectorWrap {
acquire();
MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
return mBeanServerConnection.getAttribute(name, attribute);
} catch (IOException ioe) {
// io错误，则重置连接
this.closeJmxConnect();
throw ioe;
} finally {
atomicInteger.incrementAndGet();
}
@@ -131,6 +179,11 @@ public class JmxConnectorWrap {
acquire();
MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
return mBeanServerConnection.getAttributes(name, attributes);
} catch (IOException ioe) {
// io错误，则重置连接
this.closeJmxConnect();
throw ioe;
} finally {
atomicInteger.incrementAndGet();
}
@@ -143,6 +196,11 @@ public class JmxConnectorWrap {
acquire();
MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
return mBeanServerConnection.queryNames(name, query);
} catch (IOException ioe) {
// io错误，则重置连接
this.closeJmxConnect();
throw ioe;
} finally {
atomicInteger.incrementAndGet();
}
@@ -165,4 +223,8 @@ public class JmxConnectorWrap {
}
}
}
private static String buildConnectJmxFailedBackoffEventKey(Long physicalClusterId, Integer brokerId) {
return "CONNECT_JMX_FAILED_BACK_OFF_EVENT_PHY_" + physicalClusterId + "_BROKER_" + brokerId;
}
}

162
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/zookeeper/znode/brokers/BrokerMetadata.java
View File

@@ -1,6 +1,17 @@
package com.xiaojukeji.kafka.manager.common.zookeeper.znode.brokers;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.xiaojukeji.kafka.manager.common.constant.KafkaConstant;
import com.xiaojukeji.kafka.manager.common.entity.ao.common.IpPortData;
import com.xiaojukeji.kafka.manager.common.utils.NumberUtils;
import lombok.Data;
import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @author zengqiao
@@ -10,7 +21,7 @@ import java.util.List;
* 节点结构:
* {
* "listener_security_protocol_map":{"SASL_PLAINTEXT":"SASL_PLAINTEXT"},
* "endpoints":["SASL_PLAINTEXT://10.179.162.202:9093"],
* "endpoints":["SASL_PLAINTEXT://127.0.0.1:9093"],
* "jmx_port":9999,
* "host":null,
* "timestamp":"1546632983233",
@@ -18,22 +29,48 @@ import java.util.List;
* "version":4,
* "rack": "CY"
* }
*
* {
* "listener_security_protocol_map":{"SASL_PLAINTEXT":"SASL_PLAINTEXT","PLAINTEXT":"PLAINTEXT"},
* "endpoints":["SASL_PLAINTEXT://127.0.0.1:9093","PLAINTEXT://127.0.0.1:9092"],
* "jmx_port":8099,
* "host":"127.0.0.1",
* "timestamp":"1628833925822",
* "port":9092,
* "version":4
* }
*
* {
* "listener_security_protocol_map":{"EXTERNAL":"SASL_PLAINTEXT","INTERNAL":"SASL_PLAINTEXT"},
* "endpoints":["EXTERNAL://127.0.0.1:7092","INTERNAL://127.0.0.1:7093"],
* "jmx_port":8099,
* "host":null,
* "timestamp":"1627289710439",
* "port":-1,
* "version":4
* }
*
*/
public class BrokerMetadata implements Cloneable {
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class BrokerMetadata implements Serializable {
private static final long serialVersionUID = 3918113492423375809L;
private long clusterId;
private int brokerId;
private List<String> endpoints;
// <EXTERNAL|INTERNAL, <ip, port>>
private Map<String, IpPortData> endpointMap;
private String host;
private int port;
/*
* ZK上对应的字段就是这个名字, 不要进行修改
*/
private int jmx_port;
@JsonProperty("jmx_port")
private int jmxPort;
private String version;
@@ -41,91 +78,54 @@ public class BrokerMetadata implements Cloneable {
private String rack;
public long getClusterId() {
return clusterId;
@JsonIgnore
public String getExternalHost() {
if (!endpointMap.containsKey(KafkaConstant.EXTERNAL_KEY)) {
return null;
}
return endpointMap.get(KafkaConstant.EXTERNAL_KEY).getIp();
}
public void setClusterId(long clusterId) {
this.clusterId = clusterId;
@JsonIgnore
public String getInternalHost() {
if (!endpointMap.containsKey(KafkaConstant.INTERNAL_KEY)) {
return null;
}
return endpointMap.get(KafkaConstant.INTERNAL_KEY).getIp();
}
public int getBrokerId() {
return brokerId;
}
public static void parseAndUpdateBrokerMetadata(BrokerMetadata brokerMetadata) {
brokerMetadata.setEndpointMap(new HashMap<>());
public void setBrokerId(int brokerId) {
this.brokerId = brokerId;
}
if (brokerMetadata.getEndpoints().isEmpty()) {
return;
}
public List<String> getEndpoints() {
return endpoints;
}
// example EXTERNAL://10.179.162.202:7092
for (String endpoint: brokerMetadata.getEndpoints()) {
int idx1 = endpoint.indexOf("://");
int idx2 = endpoint.lastIndexOf(":");
if (idx1 == -1 || idx2 == -1 || idx1 == idx2) {
continue;
}
public void setEndpoints(List<String> endpoints) {
this.endpoints = endpoints;
}
String brokerHost = endpoint.substring(idx1 + "://".length(), idx2);
String brokerPort = endpoint.substring(idx2 + 1);
public String getHost() {
return host;
}
brokerMetadata.getEndpointMap().put(endpoint.substring(0, idx1), new IpPortData(brokerHost, brokerPort));
public void setHost(String host) {
this.host = host;
}
if (KafkaConstant.EXTERNAL_KEY.equals(endpoint.substring(0, idx1))) {
// 优先使用external的地址进行展示
brokerMetadata.setHost(brokerHost);
brokerMetadata.setPort(NumberUtils.string2Integer(brokerPort));
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
public int getJmxPort() {
return jmx_port;
}
public void setJmxPort(int jmxPort) {
this.jmx_port = jmxPort;
}
public String getVersion() {
return version;
}
public void setVersion(String version) {
this.version = version;
}
public long getTimestamp() {
return timestamp;
}
public void setTimestamp(long timestamp) {
this.timestamp = timestamp;
}
public String getRack() {
return rack;
}
public void setRack(String rack) {
this.rack = rack;
}
@Override
public String toString() {
return "BrokerMetadata{" +
"clusterId=" + clusterId +
", brokerId=" + brokerId +
", endpoints=" + endpoints +
", host='" + host + '\'' +
", port=" + port +
", jmxPort=" + jmx_port +
", version='" + version + '\'' +
", timestamp=" + timestamp +
", rack='" + rack + '\'' +
'}';
if (null == brokerMetadata.getHost()) {
brokerMetadata.setHost(brokerHost);
brokerMetadata.setPort(NumberUtils.string2Integer(brokerPort));
}
}
}
}

13
kafka-manager-console/package.json
View File

@@ -3,7 +3,8 @@
"version": "2.5.0",
"description": "",
"scripts": {
"start": "webpack-dev-server",
"prestart": "npm install --save-dev webpack-dev-server",
"start": "webpack serve",
"daily-build": "cross-env NODE_ENV=production webpack",
"pre-build": "cross-env NODE_ENV=production webpack",
"prod-build": "cross-env NODE_ENV=production webpack",
@@ -13,18 +14,19 @@
"license": "ISC",
"devDependencies": {
"@hot-loader/react-dom": "^16.8.6",
"@types/echarts": "^4.4.1",
"@types/events": "^3.0.0",
"@types/lodash.debounce": "^4.0.6",
"@types/react": "^16.8.8",
"@types/react-dom": "^16.8.2",
"@types/react-router-dom": "^4.3.1",
"@types/spark-md5": "^3.0.2",
"@webpack-cli/serve": "^1.6.0",
"antd": "^3.26.15",
"clean-webpack-plugin": "^3.0.0",
"clipboard": "^2.0.8",
"cross-env": "^7.0.2",
"css-loader": "^2.1.0",
"echarts": "^4.5.0",
"echarts": "^5.2.1",
"file-loader": "^5.0.2",
"html-webpack-plugin": "^3.2.0",
"increase-memory-limit": "^1.0.7",
@@ -50,11 +52,10 @@
"typescript": "^3.3.3333",
"url-loader": "^4.1.1",
"webpack": "^4.29.6",
"webpack-cli": "^3.2.3",
"webpack-dev-server": "^3.2.1",
"webpack-cli": "^4.9.1",
"xlsx": "^0.16.1"
},
"dependencies": {
"format-to-json": "^1.0.4"
}
}
}

29
kafka-manager-console/src/component/chart/bar-chart.tsx
View File

@@ -1,14 +1,29 @@
import * as React from 'react';
import { Spin, notification } from 'component/antd';
import echarts, { EChartOption } from 'echarts/lib/echarts';
import * as echarts from 'echarts/core';
// 引入柱状图
import 'echarts/lib/chart/bar';
import { BarChart } from 'echarts/charts';
// 引入提示框和标题组件
import 'echarts/lib/component/tooltip';
import 'echarts/lib/component/title';
import 'echarts/lib/component/legend';
import {
TitleComponent,
TooltipComponent,
LegendComponent,
GridComponent,
} from 'echarts/components';
import { CanvasRenderer } from 'echarts/renderers';
import { EChartsOption } from 'echarts';
// 注册必须的组件
echarts.use([
TitleComponent,
LegendComponent,
TooltipComponent,
BarChart,
GridComponent,
CanvasRenderer,
]);
interface IChartProps {
getChartData: any;
@@ -38,7 +53,7 @@ export class BarChartComponet extends React.Component<IChartProps> {
this.chart.resize();
}
public isHasData = (data: EChartOption) => {
public isHasData = (data: any) => {
const noData = !(data.series && data.series.length);
this.setState({ noData });
return !noData;
@@ -54,7 +69,7 @@ export class BarChartComponet extends React.Component<IChartProps> {
const chartOptions = getChartData();
if ((typeof chartOptions.then) === 'function') {
return chartOptions.then((data: EChartOption) => {
return chartOptions.then((data: EChartsOption) => {
this.setState({ loading: false });
if (this.isHasData(data)) {

43
kafka-manager-console/src/component/chart/date-picker-chart.tsx
View File

@@ -3,16 +3,34 @@ import { DatePicker, notification, Spin } from 'component/antd';
import moment, { Moment } from 'moment';
import { timeStampStr } from 'constants/strategy';
import { disabledDate } from 'lib/utils';
import echarts from 'echarts';
import * as echarts from 'echarts/core';
// 引入柱状图和折线图
import 'echarts/lib/chart/bar';
import 'echarts/lib/chart/line';
// 引入柱状图
import { BarChart, LineChart } from 'echarts/charts';
// 引入提示框和标题组件
import 'echarts/lib/component/tooltip';
import 'echarts/lib/component/title';
import 'echarts/lib/component/legend';
import {
TitleComponent,
TooltipComponent,
LegendComponent,
GridComponent,
MarkLineComponent,
DatasetComponent,
} from 'echarts/components';
import { CanvasRenderer } from 'echarts/renderers';
// 注册必须的组件
echarts.use([
TitleComponent,
LegendComponent,
TooltipComponent,
GridComponent,
BarChart,
LineChart,
CanvasRenderer,
DatasetComponent,
MarkLineComponent,
]);
import './index.less';
const { RangePicker } = DatePicker;
@@ -61,11 +79,12 @@ export class ChartWithDatePicker extends React.Component<IChartProps> {
const noData = options.series.length ? false : true;
this.setState({ noData });
options.tooltip.formatter = (params: any) => {
var res =
"<div style='margin-bottom:5px;padding:0 12px;width:100%;height:24px;line-height:24px;border-radius:3px;'><p>" +
let res =
'<div style=\'margin-bottom:5px;padding:0 12px;width:100%;height:24px;line-height:24px;border-radius:3px;\'><p>' +
params[0].data.time +
" </p></div>";
for (var i = 0; i < params.length; i++) {
' </p></div>';
// tslint:disable-next-line:prefer-for-of
for (let i = 0; i < params.length; i++) {
res += `<div key=${params[i].seriesName} style="color: #fff;padding:0 12px;line-height: 24px">
<span style="display:inline-block;margin-right:5px;border-radius:50%;width:10px;height:10px;background-color:${[
params[i].color,
@@ -75,7 +94,7 @@ export class ChartWithDatePicker extends React.Component<IChartProps> {
</div>`;
}
return res;
}
};
this.chart.setOption(options, true);
}

28
kafka-manager-console/src/component/chart/doughnut-chart.tsx
View File

@@ -1,13 +1,27 @@
import * as React from 'react';
import { Spin } from 'component/antd';
import echarts from 'echarts/lib/echarts';
// 引入饼状图
import 'echarts/lib/chart/pie';
// 引入提示框和标题组件
import 'echarts/lib/component/tooltip';
import 'echarts/lib/component/title';
import 'echarts/lib/component/legend';
import * as echarts from 'echarts/core';
// 引入饼图
import { PieChart } from 'echarts/charts';
// 引入提示框和标题组件
import {
TitleComponent,
TooltipComponent,
LegendComponent,
GridComponent,
} from 'echarts/components';
import { CanvasRenderer } from 'echarts/renderers';
// 注册必须的组件
echarts.use([
PieChart,
TitleComponent,
LegendComponent,
TooltipComponent,
GridComponent,
CanvasRenderer,
]);
interface IPieProps {
getChartData: any;
}

46
kafka-manager-console/src/component/chart/line-chart.tsx
View File

@@ -1,25 +1,45 @@
import React from 'react';
import echarts, { EChartOption } from 'echarts/lib/echarts';
import 'echarts/lib/chart/pie';
import 'echarts/lib/chart/line';
import 'echarts/lib/component/legend';
import 'echarts/lib/component/tooltip';
import 'echarts/lib/component/title';
import 'echarts/lib/component/axis';
import * as echarts from 'echarts/core';
import './index.less';
// 引入柱状图
import { PieChart, LineChart } from 'echarts/charts';
// 引入提示框和标题组件
import {
TitleComponent,
TooltipComponent,
LegendComponent,
GridComponent,
ToolboxComponent,
DatasetComponent,
} from 'echarts/components';
import { CanvasRenderer } from 'echarts/renderers';
// 注册必须的组件
echarts.use([
PieChart,
LineChart,
ToolboxComponent,
TitleComponent,
LegendComponent,
TooltipComponent,
GridComponent,
DatasetComponent,
CanvasRenderer,
]);
export interface IEchartsProps {
width?: number;
height?: number;
options?: EChartOption;
options?: any;
}
export const hasData = (options: EChartOption) => {
export const hasData = (options: any) => {
if (options && options.series && options.series.length) return true;
return false;
};
export default class LineChart extends React.Component<IEchartsProps> {
export default class LineCharts extends React.Component<IEchartsProps> {
public id = null as HTMLDivElement;
public myChart = null as echarts.ECharts;
@@ -27,7 +47,7 @@ export default class LineChart extends React.Component<IEchartsProps> {
public componentDidMount() {
const { options } = this.props;
this.myChart = echarts.init(this.id);
this.myChart.setOption(options);
this.myChart.setOption(options, true);
window.addEventListener('resize', this.resize);
}
@@ -41,7 +61,7 @@ export default class LineChart extends React.Component<IEchartsProps> {
public refresh = () => {
const { options } = this.props;
this.myChart.setOption(options);
this.myChart.setOption(options, true);
}
public resize = () => {
@@ -50,6 +70,6 @@ export default class LineChart extends React.Component<IEchartsProps> {
public render() {
const { height, width } = this.props;
return <div ref={id => this.id = id} style={{width: `${width}px`, height: `${height}px`}} />;
return <div ref={id => this.id = id} style={{ width: `${width}px`, height: `${height}px` }} />;
}
}

3
kafka-manager-console/src/container/admin/data-curve/config.ts
View File

@@ -1,4 +1,3 @@
import { EChartOption } from 'echarts/lib/echarts';
import moment from 'moment';
import { ICurve } from 'container/common-curve/config';
import { adminMonitor } from 'store/admin-monitor';
@@ -124,7 +123,7 @@ export interface ICurveType {
type: curveType;
title: string;
curves: ICurve[];
parser: (option: ICurve, data: any[]) => EChartOption;
parser: (option: ICurve, data: any[]) => any;
}
export const byteTypeCurves: ICurveType[] = [

10
kafka-manager-console/src/container/admin/data-curve/parser.ts
View File

@@ -1,5 +1,5 @@
import moment from 'moment';
import { EChartOption } from 'echarts';
import { EChartsOption } from 'echarts';
import { ICurve, ILineData, baseLineLegend, baseLineGrid, baseAxisStyle, noAxis, UNIT_HEIGHT } from 'container/common-curve/config';
import { IClusterMetrics, ISeriesOption } from 'types/base-type';
import { timeFormat } from 'constants/strategy';
@@ -48,20 +48,20 @@ export const getBaseOptions = (option: ICurve, data: ILineData[]) => {
return Number(i.value);
}),
}],
} as EChartOption;
} as EChartsOption;
};
export const parseLine = (option: ICurve, data: ILineData[]): EChartOption => {
export const parseLine = (option: ICurve, data: ILineData[]): EChartsOption => {
return Object.assign({}, getBaseOptions(option, data), {
legend: {
...baseLineLegend,
bottom: '0',
align: 'auto',
},
}) as EChartOption;
}) as EChartsOption;
};
export const parseBrokerMetricOption = (option: ICurve, data: IClusterMetrics[]): EChartOption => {
export const parseBrokerMetricOption = (option: ICurve, data: IClusterMetrics[]): EChartsOption => {
let name;
let series: ISeriesOption[];
data = data || [];

6
kafka-manager-console/src/container/alarm/alarm-detail/history-detail.tsx
View File

@@ -6,7 +6,7 @@ import { alarm } from 'store/alarm';
import { observer } from 'mobx-react';
import { handlePageBack } from 'lib/utils';
import LineChart, { hasData } from 'component/chart/line-chart';
import { EChartOption } from 'echarts';
import { EChartsOption } from 'echarts';
import { timeFormat } from 'constants/strategy';
import Url from 'lib/url-parser';
import moment = require('moment');
@@ -40,7 +40,7 @@ export class HistoryDetail extends React.Component {
return <div className="no-data-info" style={{ ...style }} key="loading"><Spin /></div>;
}
public renderEchart = (options: EChartOption, loading = false) => {
public renderEchart = (options: EChartsOption, loading = false) => {
const data = hasData(options);
if (loading) return this.renderLoading(400);
if (!data) return this.renderNoData(400);
@@ -51,7 +51,7 @@ export class HistoryDetail extends React.Component {
}
public renderHistoricalTraffic(metric: IMonitorMetric) {
const option = this.getChartOption() as EChartOption;
const option = this.getChartOption() as EChartsOption;
return (
<>

11
kafka-manager-console/src/container/common-curve/config.ts
View File

@@ -1,5 +1,4 @@
import { EChartOption } from 'echarts/lib/echarts';
import moment from 'moment';
import { EChartsOption } from 'echarts';
export interface ILineData {
value: number;
@@ -9,7 +8,7 @@ export interface ICurve {
title?: string;
path: string;
colors: string[];
parser?: (option: ICurve, data: ILineData) => EChartOption;
parser?: (option: ICurve, data: ILineData) => EChartsOption;
message?: string;
unit?: string;
api?: any;
@@ -69,13 +68,13 @@ export const noAxis = {
},
};
export const getHight = (options: EChartOption) => {
let grid = options ? options.grid as EChartOption.Grid : null;
export const getHight = (options: any) => {
let grid = options ? options.grid : null;
if (!options || !grid) grid = baseLineGrid;
return Number(grid.height) + getLegendHight(options) + Number(grid.top) + LEGEND_PADDING + UNIT_HEIGHT;
};
export const getLegendHight = (options: EChartOption) => {
export const getLegendHight = (options: any) => {
if (!options) return 0;
if (options.legend.show === false) return 0;
const legendHight = options.legend.textStyle.lineHeight + defaultLegendPadding;

10
kafka-manager-console/src/container/common-curve/index.tsx
View File

@@ -1,4 +1,4 @@
import { EChartOption } from 'echarts';
import { EChartsOption } from 'echarts';
import { observer } from 'mobx-react';
import React from 'react';
import { curveInfo } from 'store/curve-info';
@@ -10,7 +10,7 @@ import LineChart, { hasData } from 'component/chart/line-chart';
export interface ICommonCurveProps {
options: ICurve;
parser?: (option: ICurve, data: any[]) => EChartOption;
parser?: (option: ICurve, data: any[]) => any;
}
@observer
@@ -41,7 +41,7 @@ export class CommonCurve extends React.Component<ICommonCurveProps> {
fullScreen.show(this.renderCurve(options, loading, true));
}
public renderOpBtns = (options: EChartOption, expand = false) => {
public renderOpBtns = (options: EChartsOption, expand = false) => {
const data = hasData(options);
return (
<div className="charts-op" key="op">
@@ -85,7 +85,7 @@ export class CommonCurve extends React.Component<ICommonCurveProps> {
return <div className="no-data-info" style={{ ...style }} key="loading"><Spin /></div>;
}
public renderEchart = (options: EChartOption, loading = false) => {
public renderEchart = (options: EChartsOption, loading = false) => {
const height = getHight(options);
const data = hasData(options);
@@ -94,7 +94,7 @@ export class CommonCurve extends React.Component<ICommonCurveProps> {
return <LineChart height={height} options={options} key="chart" />;
}
public renderCurve = (options: EChartOption, loading: boolean, expand = false) => {
public renderCurve = (options: any, loading: boolean, expand = false) => {
const data = hasData(options);
return (
<div className="common-chart-wrapper" >

2
kafka-manager-console/src/container/header/index.tsx
View File

@@ -7,7 +7,7 @@ import { urlPrefix } from 'constants/left-menu';
import { region, IRegionIdcs } from 'store/region';
import logoUrl from '../../assets/image/kafka-logo.png';
import userIcon from '../../assets/image/normal.png';
import weChat from '../../assets/image/wechat.png';
import weChat from '../../assets/image/weChat.png';
import { users } from 'store/users';
import { observer } from 'mobx-react';
import { Link } from 'react-router-dom';

8
kafka-manager-console/src/store/curve-info.ts
View File

@@ -1,6 +1,6 @@
import { observable, action } from 'mobx';
import moment = require('moment');
import { EChartOption } from 'echarts/lib/echarts';
import { EChartsOption } from 'echarts';
import { ICurve } from 'container/common-curve/config';
import { curveKeys, PERIOD_RADIO_MAP } from 'container/admin/data-curve/config';
import { timeFormat } from 'constants/strategy';
@@ -13,7 +13,7 @@ class CurveInfo {
public timeRange: [moment.Moment, moment.Moment] = PERIOD_RADIO_MAP.get(this.periodKey).dateRange;
@observable
public curveData: { [key: string]: EChartOption } = {};
public curveData: { [key: string]: EChartsOption } = {};
@observable
public curveLoading: { [key: string]: boolean } = {};
@@ -25,7 +25,7 @@ class CurveInfo {
public currentOperator: string;
@action.bound
public setCurveData(key: curveKeys | string, data: EChartOption) {
public setCurveData(key: curveKeys | string, data: EChartsOption) {
this.curveData[key] = data;
}
@@ -59,7 +59,7 @@ class CurveInfo {
public getCommonCurveData = (
options: ICurve,
parser: (option: ICurve, data: any[]) => EChartOption,
parser: (option: ICurve, data: any[]) => EChartsOption,
reload?: boolean) => {
const { path } = options;
this.setCurveData(path, null);

4
kafka-manager-console/webpack.config.js
View File

@@ -122,11 +122,11 @@ module.exports = {
},
},
devServer: {
contentBase: outPath,
// contentBase: outPath,
host: '127.0.0.1',
port: 1025,
hot: true,
disableHostCheck: true,
// disableHostCheck: true,
historyApiFallback: true,
proxy: {
'/api/v1/': {

27
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/cache/KafkaClientPool.java vendored
View File

@@ -14,6 +14,8 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.util.Map;
import java.util.Properties;
@@ -25,9 +27,22 @@ import java.util.concurrent.locks.ReentrantLock;
* @author zengqiao
* @date 19/12/24
*/
@Service
public class KafkaClientPool {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaClientPool.class);
@Value(value = "${client-pool.kafka-consumer.min-idle-client-num:24}")
private Integer kafkaConsumerMinIdleClientNum;
@Value(value = "${client-pool.kafka-consumer.max-idle-client-num:24}")
private Integer kafkaConsumerMaxIdleClientNum;
@Value(value = "${client-pool.kafka-consumer.max-total-client-num:24}")
private Integer kafkaConsumerMaxTotalClientNum;
@Value(value = "${client-pool.kafka-consumer.borrow-timeout-unit-ms:3000}")
private Integer kafkaConsumerBorrowTimeoutUnitMs;
/**
* AdminClient
*/
@@ -84,7 +99,7 @@ public class KafkaClientPool {
return true;
}
private static void initKafkaConsumerPool(ClusterDO clusterDO) {
private void initKafkaConsumerPool(ClusterDO clusterDO) {
lock.lock();
try {
GenericObjectPool<KafkaConsumer<String, String>> objectPool = KAFKA_CONSUMER_POOL.get(clusterDO.getId());
@@ -92,9 +107,9 @@ public class KafkaClientPool {
return;
}
GenericObjectPoolConfig<KafkaConsumer<String, String>> config = new GenericObjectPoolConfig<>();
config.setMaxIdle(24);
config.setMinIdle(24);
config.setMaxTotal(24);
config.setMaxIdle(kafkaConsumerMaxIdleClientNum);
config.setMinIdle(kafkaConsumerMinIdleClientNum);
config.setMaxTotal(kafkaConsumerMaxTotalClientNum);
KAFKA_CONSUMER_POOL.put(clusterDO.getId(), new GenericObjectPool<>(new KafkaConsumerFactory(clusterDO), config));
} catch (Exception e) {
LOGGER.error("create kafka consumer pool failed, clusterDO:{}.", clusterDO, e);
@@ -118,7 +133,7 @@ public class KafkaClientPool {
}
}
public static KafkaConsumer<String, String> borrowKafkaConsumerClient(ClusterDO clusterDO) {
public KafkaConsumer<String, String> borrowKafkaConsumerClient(ClusterDO clusterDO) {
if (ValidateUtils.isNull(clusterDO)) {
return null;
}
@@ -132,7 +147,7 @@ public class KafkaClientPool {
}
try {
return objectPool.borrowObject(3000);
return objectPool.borrowObject(kafkaConsumerBorrowTimeoutUnitMs);
} catch (Exception e) {
LOGGER.error("borrow kafka consumer client failed, clusterDO:{}.", clusterDO, e);
}

5
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/cache/PhysicalClusterMetadataManager.java vendored
View File

@@ -50,6 +50,9 @@ public class PhysicalClusterMetadataManager {
@Autowired
private ClusterService clusterService;
@Autowired
private ThreadPool threadPool;
private static final Map<Long, ClusterDO> CLUSTER_MAP = new ConcurrentHashMap<>();
private static final Map<Long, ControllerData> CONTROLLER_DATA_MAP = new ConcurrentHashMap<>();
@@ -125,7 +128,7 @@ public class PhysicalClusterMetadataManager {
zkConfig.watchChildren(ZkPathUtil.BROKER_IDS_ROOT, brokerListener);
//增加Topic监控
TopicStateListener topicListener = new TopicStateListener(clusterDO.getId(), zkConfig);
TopicStateListener topicListener = new TopicStateListener(clusterDO.getId(), zkConfig, threadPool);
topicListener.init();
zkConfig.watchChildren(ZkPathUtil.BROKER_TOPICS_ROOT, topicListener);

68
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/cache/ThreadPool.java vendored
View File

@@ -1,37 +1,63 @@
package com.xiaojukeji.kafka.manager.service.cache;
import com.xiaojukeji.kafka.manager.common.utils.factory.DefaultThreadFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.util.concurrent.*;
import javax.annotation.PostConstruct;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @author zengqiao
* @date 20/8/24
*/
@Service
public class ThreadPool {
private static final ExecutorService COLLECT_METRICS_THREAD_POOL = new ThreadPoolExecutor(
256,
256,
120L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new DefaultThreadFactory("Collect-Metrics-Thread")
);
private static final ExecutorService API_CALL_THREAD_POOL = new ThreadPoolExecutor(
16,
16,
120L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new DefaultThreadFactory("Api-Call-Thread")
);
@Value(value = "${thread-pool.collect-metrics.thread-num:256}")
private Integer collectMetricsThreadNum;
public static void submitCollectMetricsTask(Runnable collectMetricsTask) {
COLLECT_METRICS_THREAD_POOL.submit(collectMetricsTask);
@Value(value = "${thread-pool.collect-metrics.queue-size:10000}")
private Integer collectMetricsQueueSize;
@Value(value = "${thread-pool.api-call.thread-num:16}")
private Integer apiCallThreadNum;
@Value(value = "${thread-pool.api-call.queue-size:10000}")
private Integer apiCallQueueSize;
private ThreadPoolExecutor collectMetricsThreadPool;
private ThreadPoolExecutor apiCallThreadPool;
@PostConstruct
public void init() {
collectMetricsThreadPool = new ThreadPoolExecutor(
collectMetricsThreadNum,
collectMetricsThreadNum,
120L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(collectMetricsQueueSize),
new DefaultThreadFactory("TaskThreadPool")
);
apiCallThreadPool = new ThreadPoolExecutor(
apiCallThreadNum,
apiCallThreadNum,
120L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(apiCallQueueSize),
new DefaultThreadFactory("ApiThreadPool")
);
}
public static void submitApiCallTask(Runnable apiCallTask) {
API_CALL_THREAD_POOL.submit(apiCallTask);
public void submitCollectMetricsTask(Long clusterId, Runnable collectMetricsTask) {
collectMetricsThreadPool.submit(collectMetricsTask);
}
public void submitApiCallTask(Long clusterId, Runnable apiCallTask) {
apiCallThreadPool.submit(apiCallTask);
}
}

5
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/BrokerServiceImpl.java
View File

@@ -61,6 +61,9 @@ public class BrokerServiceImpl implements BrokerService {
@Autowired
private PhysicalClusterMetadataManager physicalClusterMetadataManager;
@Autowired
private ThreadPool threadPool;
@Override
public ClusterBrokerStatus getClusterBrokerStatus(Long clusterId) {
// 副本同步状态
@@ -201,7 +204,7 @@ public class BrokerServiceImpl implements BrokerService {
return getBrokerMetricsFromJmx(clusterId, brokerId, metricsCode);
}
});
ThreadPool.submitApiCallTask(taskList[i]);
threadPool.submitApiCallTask(clusterId, taskList[i]);
}
List<BrokerMetrics> metricsList = new ArrayList<>(brokerIdSet.size());
for (int i = 0; i < brokerIdList.size(); i++) {

7
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/JmxServiceImpl.java
View File

@@ -39,6 +39,9 @@ public class JmxServiceImpl implements JmxService {
@Autowired
private PhysicalClusterMetadataManager physicalClusterMetadataManager;
@Autowired
private ThreadPool threadPool;
@Override
public BrokerMetrics getBrokerMetrics(Long clusterId, Integer brokerId, Integer metricsCode) {
if (clusterId == null || brokerId == null || metricsCode == null) {
@@ -98,7 +101,7 @@ public class JmxServiceImpl implements JmxService {
);
}
});
ThreadPool.submitCollectMetricsTask(taskList[i]);
threadPool.submitCollectMetricsTask(clusterId, taskList[i]);
}
List<TopicMetrics> metricsList = new ArrayList<>();
@@ -303,7 +306,7 @@ public class JmxServiceImpl implements JmxService {
return metricsList;
}
});
ThreadPool.submitCollectMetricsTask(taskList[i]);
threadPool.submitCollectMetricsTask(clusterId, taskList[i]);
}
Map<String, TopicMetrics> metricsMap = new HashMap<>();

7
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/TopicServiceImpl.java
View File

@@ -87,6 +87,9 @@ public class TopicServiceImpl implements TopicService {
@Autowired
private AbstractHealthScoreStrategy healthScoreStrategy;
@Autowired
private KafkaClientPool kafkaClientPool;
@Override
public List<TopicMetricsDO> getTopicMetricsFromDB(Long clusterId, String topicName, Date startTime, Date endTime) {
try {
@@ -340,7 +343,7 @@ public class TopicServiceImpl implements TopicService {
Map<TopicPartition, Long> topicPartitionLongMap = new HashMap<>();
KafkaConsumer kafkaConsumer = null;
try {
kafkaConsumer = KafkaClientPool.borrowKafkaConsumerClient(clusterDO);
kafkaConsumer = kafkaClientPool.borrowKafkaConsumerClient(clusterDO);
if ((offsetPosEnum.getCode() & OffsetPosEnum.END.getCode()) > 0) {
topicPartitionLongMap = kafkaConsumer.endOffsets(topicPartitionList);
} else if ((offsetPosEnum.getCode() & OffsetPosEnum.BEGINNING.getCode()) > 0) {
@@ -541,7 +544,7 @@ public class TopicServiceImpl implements TopicService {
List<PartitionOffsetDTO> partitionOffsetDTOList = new ArrayList<>();
try {
kafkaConsumer = KafkaClientPool.borrowKafkaConsumerClient(clusterDO);
kafkaConsumer = kafkaClientPool.borrowKafkaConsumerClient(clusterDO);
Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestampMap = kafkaConsumer.offsetsForTimes(timestampsToSearch);
if (offsetAndTimestampMap == null) {
return new ArrayList<>();

5
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/strategy/healthscore/DidiHealthScoreStrategy.java
View File

@@ -45,6 +45,9 @@ public class DidiHealthScoreStrategy extends AbstractHealthScoreStrategy {
@Autowired
private JmxService jmxService;
@Autowired
private ThreadPool threadPool;
@Override
public Integer calBrokerHealthScore(Long clusterId, Integer brokerId) {
BrokerMetadata brokerMetadata = PhysicalClusterMetadataManager.getBrokerMetadata(clusterId, brokerId);
@@ -125,7 +128,7 @@ public class DidiHealthScoreStrategy extends AbstractHealthScoreStrategy {
return calBrokerHealthScore(clusterId, brokerId);
}
});
ThreadPool.submitApiCallTask(taskList[i]);
threadPool.submitApiCallTask(clusterId, taskList[i]);
}
Integer topicHealthScore = HEALTH_SCORE_HEALTHY;

5
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/utils/ConfigUtils.java
View File

@@ -18,6 +18,9 @@ public class ConfigUtils {
@Value(value = "${custom.idc:cn}")
private String idc;
@Value(value = "${spring.profiles.active}")
@Value(value = "${spring.profiles.active:dev}")
private String kafkaManagerEnv;
@Value(value = "${spring.application.version:unknown}")
private String applicationVersion;
}

11
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/zookeeper/BrokerStateListener.java
View File

@@ -74,15 +74,10 @@ public class BrokerStateListener implements StateChangeListener {
BrokerMetadata brokerMetadata = null;
try {
brokerMetadata = zkConfig.get(ZkPathUtil.getBrokerIdNodePath(brokerId), BrokerMetadata.class);
if (!brokerMetadata.getEndpoints().isEmpty()) {
String endpoint = brokerMetadata.getEndpoints().get(0);
int idx = endpoint.indexOf("://");
endpoint = endpoint.substring(idx + "://".length());
idx = endpoint.indexOf(":");
brokerMetadata.setHost(endpoint.substring(0, idx));
brokerMetadata.setPort(Integer.parseInt(endpoint.substring(idx + 1)));
}
// 解析并更新本次存储的broker元信息
BrokerMetadata.parseAndUpdateBrokerMetadata(brokerMetadata);
brokerMetadata.setClusterId(clusterId);
brokerMetadata.setBrokerId(brokerId);
PhysicalClusterMetadataManager.putBrokerMetadata(clusterId, brokerId, brokerMetadata, jmxConfig);

8
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/zookeeper/TopicStateListener.java
View File

@@ -10,6 +10,7 @@ import com.xiaojukeji.kafka.manager.service.cache.ThreadPool;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import java.util.HashSet;
import java.util.List;
@@ -28,9 +29,12 @@ public class TopicStateListener implements StateChangeListener {
private ZkConfigImpl zkConfig;
public TopicStateListener(Long clusterId, ZkConfigImpl zkConfig) {
private ThreadPool threadPool;
public TopicStateListener(Long clusterId, ZkConfigImpl zkConfig, ThreadPool threadPool) {
this.clusterId = clusterId;
this.zkConfig = zkConfig;
this.threadPool = threadPool;
}
@Override
@@ -47,7 +51,7 @@ public class TopicStateListener implements StateChangeListener {
return null;
}
});
ThreadPool.submitCollectMetricsTask(taskList[i]);
threadPool.submitCollectMetricsTask(clusterId, taskList[i]);
}
} catch (Exception e) {
LOGGER.error("init topics metadata failed, clusterId:{}.", clusterId, e);

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

@@ -31,16 +31,16 @@ public class BaseSessionSignOn extends AbstractSingleSignOn {
private LdapAuthentication ldapAuthentication;
//是否开启ldap验证
@Value(value = "${account.ldap.enabled:}")
@Value(value = "${account.ldap.enabled:false}")
private Boolean accountLdapEnabled;
//ldap自动注册的默认角色。请注意：它通常来说都是低权限角色
@Value(value = "${account.ldap.auth-user-registration-role:}")
@Value(value = "${account.ldap.auth-user-registration-role:normal}")
private String authUserRegistrationRole;
//ldap自动注册是否开启
@Value(value = "${account.ldap.auth-user-registration:}")
private boolean authUserRegistration;
@Value(value = "${account.ldap.auth-user-registration:false}")
private Boolean authUserRegistration;
@Override
public Result<String> loginAndGetLdap(HttpServletRequest request, HttpServletResponse response, LoginDTO dto) {

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

@@ -14,6 +14,7 @@ import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import javax.servlet.http.Cookie;
@@ -27,7 +28,13 @@ import javax.servlet.http.HttpSession;
*/
@Service("loginService")
public class LoginServiceImpl implements LoginService {
private final static Logger LOGGER = LoggerFactory.getLogger(LoginServiceImpl.class);
private static final Logger LOGGER = LoggerFactory.getLogger(LoginServiceImpl.class);
@Value(value = "${account.jump-login.gateway-api:false}")
private Boolean jumpLoginGatewayApi;
@Value(value = "${account.jump-login.third-part-api:false}")
private Boolean jumpLoginThirdPartApi;
@Autowired
private AccountService accountService;
@@ -75,8 +82,10 @@ public class LoginServiceImpl implements LoginService {
return false;
}
if (classRequestMappingValue.equals(ApiPrefix.API_V1_SSO_PREFIX)) {
// 白名单接口直接true
if (classRequestMappingValue.equals(ApiPrefix.API_V1_SSO_PREFIX) ||
(jumpLoginGatewayApi != null && jumpLoginGatewayApi && classRequestMappingValue.equals(ApiPrefix.GATEWAY_API_V1_PREFIX)) ||
(jumpLoginThirdPartApi != null && jumpLoginThirdPartApi && classRequestMappingValue.equals(ApiPrefix.API_V1_THIRD_PART_PREFIX))) {
// 登录接口 or 允许跳过且是跳过类型的接口，则直接跳过登录
return true;
}

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

@@ -37,21 +37,24 @@ import java.util.Map;
public class N9e extends AbstractAgent {
private static final Logger LOGGER = LoggerFactory.getLogger(N9e.class);
@Value("${kcm.n9e.base-url}")
@Value("${kcm.n9e.base-url:}")
private String baseUrl;
@Value("${kcm.n9e.user-token}")
@Value("${kcm.n9e.user-token:12345678}")
private String userToken;
@Value("${kcm.n9e.account}")
@Value("${kcm.n9e.account:root}")
private String account;
@Value("${kcm.n9e.timeout}")
@Value("${kcm.n9e.timeout:300}")
private Integer timeout;
@Value("${kcm.n9e.script-file}")
@Value("${kcm.n9e.script-file:kcm_script.sh}")
private String scriptFile;
@Value("${kcm.n9e.logikm-url:}")
private String logiKMUrl;
private String script;
private static final String CREATE_TASK_URI = "/api/job-ce/tasks";
@@ -219,7 +222,8 @@ public class N9e extends AbstractAgent {
sb.append(creationTaskData.getKafkaPackageUrl()).append(",,");
sb.append(creationTaskData.getServerPropertiesName().replace(KafkaFileEnum.SERVER_CONFIG.getSuffix(), "")).append(",,");
sb.append(creationTaskData.getServerPropertiesMd5()).append(",,");
sb.append(creationTaskData.getServerPropertiesUrl());
sb.append(creationTaskData.getServerPropertiesUrl()).append(",,");
sb.append(this.logiKMUrl);
N9eCreationTask n9eCreationTask = new N9eCreationTask();
n9eCreationTask.setTitle(Constant.TASK_TITLE_PREFIX + "-集群ID:" + creationTaskData.getClusterId());

12
kafka-manager-extends/kafka-manager-kcm/src/main/resources/kcm_script.sh
View File

@@ -18,12 +18,13 @@ p_kafka_server_properties_name=${7} #server配置名
p_kafka_server_properties_md5=${8} #server配置MD5
p_kafka_server_properties_url=${9} #server配置文件下载地址
p_kafka_manager_url=${10} #LogiKM地址
#----------------------------------------配置信息------------------------------------------------------#
g_base_dir='/home'
g_cluster_task_dir=${g_base_dir}"/kafka_cluster_task/task_${p_task_id}" #部署升级路径
g_rollback_version=${g_cluster_task_dir}"/rollback_version" #回滚版本
g_new_kafka_package_name='' #最终的包名
g_kafka_manager_addr='' #kafka-manager地址
g_local_ip=`ifconfig -a|grep inet|grep -v 127.0.0.1|grep -v inet6|awk '{print $2}'|tr -d "addr:"`
g_hostname=${g_local_ip}
@@ -47,7 +48,7 @@ function dchat_alarm() {
# 检查并初始化环境
function check_and_init_env() {
if [ -z "${p_task_id}" -o -z "${p_cluster_task_type}" -o -z "${p_kafka_package_url}" -o -z "${p_cluster_id}" -o -z "${p_kafka_package_name}" -o -z "${p_kafka_package_md5}" -o -z "${p_kafka_server_properties_name}" -o -z "${p_kafka_server_properties_md5}" ]; then
if [ -z "${p_task_id}" -o -z "${p_cluster_task_type}" -o -z "${p_kafka_package_url}" -o -z "${p_cluster_id}" -o -z "${p_kafka_package_name}" -o -z "${p_kafka_package_md5}" -o -z "${p_kafka_server_properties_name}" -o -z "${p_kafka_server_properties_md5}" -o -z "${p_kafka_manager_url}" ]; then
ECHO_LOG "存在为空的参数不合法, 退出集群任务"
dchat_alarm "存在为空的参数不合法, 退出集群任务"
exit 1
@@ -72,11 +73,11 @@ function check_and_init_env() {
# 检查并等待集群所有的副本处于同步的状态
function check_and_wait_broker_stabled() {
under_replication_count=`curl -s -G -d "hostname="${g_hostname} ${g_kafka_manager_addr}/api/v1/third-part/${p_cluster_id}/broker-stabled | python -m json.tool | grep true |wc -l`
under_replication_count=`curl -s -G -d "hostname="${g_hostname} ${p_kafka_manager_url}/api/v1/third-part/${p_cluster_id}/broker-stabled | python -m json.tool | grep true |wc -l`
while [ "$under_replication_count" -ne 1 ]; do
ECHO_LOG "存在${under_replication_count}个副本未同步, sleep 10s"
sleep 10
under_replication_count=`curl -s -G -d "hostname="${g_hostname} ${g_kafka_manager_addr}/api/v1/third-part/${p_cluster_id}/broker-stabled | python -m json.tool | grep true |wc -l`
under_replication_count=`curl -s -G -d "hostname="${g_hostname} ${p_kafka_manager_url}/api/v1/third-part/${p_cluster_id}/broker-stabled | python -m json.tool | grep true |wc -l`
done
ECHO_LOG "集群副本都已经处于同步的状态, 可以进行集群升级"
}
@@ -324,6 +325,7 @@ ECHO_LOG " p_kafka_package_name=${p_kafka_package_name}"
ECHO_LOG " p_kafka_package_md5=${p_kafka_package_md5}"
ECHO_LOG " p_kafka_server_properties_name=${p_kafka_server_properties_name}"
ECHO_LOG " p_kafka_server_properties_md5=${p_kafka_server_properties_md5}"
ECHO_LOG " p_kafka_manager_url=${p_kafka_manager_url}"
@@ -342,7 +344,7 @@ fi
ECHO_LOG "停kafka服务"
stop_kafka_server
ECHO_LOG "停5秒, 确保"
ECHO_LOG "再停5秒, 确保端口已释放"
sleep 5
if [ "${p_cluster_task_type}" == "0" ];then

5
kafka-manager-extends/kafka-manager-openapi/src/main/java/com/xiaojukeji/kafka/manager/openapi/impl/ThirdPartServiceImpl.java
View File

@@ -42,6 +42,9 @@ public class ThirdPartServiceImpl implements ThirdPartService {
@Autowired
private ConsumerService consumerService;
@Autowired
private KafkaClientPool kafkaClientPool;
@Override
public Result<ConsumeHealthEnum> checkConsumeHealth(Long clusterId,
String topicName,
@@ -109,7 +112,7 @@ public class ThirdPartServiceImpl implements ThirdPartService {
Long timestamp) {
KafkaConsumer kafkaConsumer = null;
try {
kafkaConsumer = KafkaClientPool.borrowKafkaConsumerClient(clusterDO);
kafkaConsumer = kafkaClientPool.borrowKafkaConsumerClient(clusterDO);
if (ValidateUtils.isNull(kafkaConsumer)) {
return null;
}

3
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/component/AbstractScheduledTask.java
View File

@@ -1,7 +1,6 @@
package com.xiaojukeji.kafka.manager.task.component;
import com.google.common.collect.Lists;
import com.xiaojukeji.kafka.manager.common.constant.LogConstant;
import com.xiaojukeji.kafka.manager.common.utils.factory.DefaultThreadFactory;
import com.xiaojukeji.kafka.manager.common.utils.JsonUtils;
import com.xiaojukeji.kafka.manager.common.utils.NetUtils;
@@ -29,7 +28,7 @@ import java.util.concurrent.*;
* @date 20/8/10
*/
public abstract class AbstractScheduledTask<E extends Comparable> implements SchedulingConfigurer {
private final static Logger LOGGER = LoggerFactory.getLogger(LogConstant.SCHEDULED_TASK_LOGGER);
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractScheduledTask.class);
@Autowired
private HeartbeatDao heartbeatDao;

8
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/component/BaseBizTask.java
View File

@@ -1,6 +1,5 @@
package com.xiaojukeji.kafka.manager.task.component;
import com.xiaojukeji.kafka.manager.common.constant.LogConstant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -9,11 +8,11 @@ import org.slf4j.LoggerFactory;
* @date 20/8/10
*/
public class BaseBizTask<E extends Comparable> implements Runnable {
private final static Logger LOGGER = LoggerFactory.getLogger(LogConstant.SCHEDULED_TASK_LOGGER);
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractScheduledTask.class);
private E task;
private final E task;
private AbstractScheduledTask scheduledTask;
private final AbstractScheduledTask scheduledTask;
public BaseBizTask(E task, AbstractScheduledTask scheduledTask) {
this.task = task;
@@ -30,6 +29,7 @@ public class BaseBizTask<E extends Comparable> implements Runnable {
} catch (Throwable t) {
LOGGER.error("scheduled task scheduleName:{} execute failed, task:{}", scheduledTask.getScheduledName(), task, t);
}
LOGGER.info("scheduled task scheduleName:{} finished, cost-time:{}ms.", scheduledTask.getScheduledName(), System.currentTimeMillis() - startTime);
}
}

93
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/dispatch/metrics/collect/CollectAndPublishBrokerMetrics.java Normal file
View File

@@ -0,0 +1,93 @@
package com.xiaojukeji.kafka.manager.task.dispatch.metrics.collect;
import com.xiaojukeji.kafka.manager.common.constant.KafkaMetricsCollections;
import com.xiaojukeji.kafka.manager.common.entity.metrics.BrokerMetrics;
import com.xiaojukeji.kafka.manager.common.events.metrics.BatchBrokerMetricsCollectedEvent;
import com.xiaojukeji.kafka.manager.common.utils.SpringTool;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.common.utils.jmx.JmxConstant;
import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
import com.xiaojukeji.kafka.manager.service.cache.PhysicalClusterMetadataManager;
import com.xiaojukeji.kafka.manager.service.service.ClusterService;
import com.xiaojukeji.kafka.manager.service.service.JmxService;
import com.xiaojukeji.kafka.manager.service.strategy.AbstractHealthScoreStrategy;
import com.xiaojukeji.kafka.manager.task.component.AbstractScheduledTask;
import com.xiaojukeji.kafka.manager.task.component.CustomScheduled;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import java.util.ArrayList;
import java.util.List;
/**
* Broker指标信息收集
* @author zengqiao
* @date 20/5/7
*/
@CustomScheduled(name = "collectAndPublishBrokerMetrics", cron = "21 0/1 * * * ?", threadNum = 2)
@ConditionalOnProperty(prefix = "task.metrics.collect", name = "broker-metrics-enabled", havingValue = "true", matchIfMissing = true)
public class CollectAndPublishBrokerMetrics extends AbstractScheduledTask<ClusterDO> {
private static final Logger LOGGER = LoggerFactory.getLogger(CollectAndPublishBrokerMetrics.class);
@Autowired
private JmxService jmxService;
@Autowired
private ClusterService clusterService;
@Autowired
private AbstractHealthScoreStrategy healthScoreStrategy;
@Override
protected List<ClusterDO> listAllTasks() {
return clusterService.list();
}
@Override
public void processTask(ClusterDO clusterDO) {
long startTime = System.currentTimeMillis();
try {
SpringTool.publish(new BatchBrokerMetricsCollectedEvent(
this,
clusterDO.getId(),
startTime,
this.getBrokerMetrics(clusterDO.getId()))
);
} catch (Exception e) {
LOGGER.error("collect broker-metrics failed, physicalClusterId:{}.", clusterDO.getId(), e);
}
LOGGER.info("collect broker-metrics finished, physicalClusterId:{} costTime:{}", clusterDO.getId(), System.currentTimeMillis() - startTime);
}
private List<BrokerMetrics> getBrokerMetrics(Long clusterId) {
List<BrokerMetrics> metricsList = new ArrayList<>();
for (Integer brokerId: PhysicalClusterMetadataManager.getBrokerIdList(clusterId)) {
BrokerMetrics metrics = jmxService.getBrokerMetrics(
clusterId,
brokerId,
KafkaMetricsCollections.BROKER_TO_DB_METRICS
);
if (ValidateUtils.isNull(metrics)) {
continue;
}
metrics.getMetricsMap().put(
JmxConstant.HEALTH_SCORE,
healthScoreStrategy.calBrokerHealthScore(clusterId, brokerId, metrics)
);
metricsList.add(metrics);
}
if (ValidateUtils.isEmptyList(metricsList)) {
return new ArrayList<>();
}
return metricsList;
}
}

5
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/dispatch/metrics/collect/CollectAndPublishCGData.java
View File

@@ -44,6 +44,9 @@ public class CollectAndPublishCGData extends AbstractScheduledTask<ClusterDO> {
@Autowired
private ConsumerService consumerService;
@Autowired
private ThreadPool threadPool;
@Override
protected List<ClusterDO> listAllTasks() {
return clusterService.list();
@@ -82,7 +85,7 @@ public class CollectAndPublishCGData extends AbstractScheduledTask<ClusterDO> {
return getTopicConsumerMetrics(clusterDO, topicName, startTimeUnitMs);
}
});
ThreadPool.submitCollectMetricsTask(taskList[i]);
threadPool.submitCollectMetricsTask(clusterDO.getId(), taskList[i]);
}
List<ConsumerMetrics> consumerMetricsList = new ArrayList<>();

14
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/dispatch/metrics/delete/DeleteMetrics.java
View File

@@ -42,25 +42,25 @@ public class DeleteMetrics extends AbstractScheduledTask<EmptyEntry> {
@Autowired
private TopicThrottledMetricsDao topicThrottledMetricsDao;
@Value(value = "${task.metrics.delete-metrics.delete-limit-size:1000}")
@Value(value = "${task.metrics.delete.delete-limit-size:1000}")
private Integer deleteLimitSize;
@Value(value = "${task.metrics.delete-metrics.cluster-metrics-save-days:14}")
@Value(value = "${task.metrics.delete.cluster-metrics-save-days:14}")
private Integer clusterMetricsSaveDays;
@Value(value = "${task.metrics.delete-metrics.broker-metrics-save-days:14}")
@Value(value = "${task.metrics.delete.broker-metrics-save-days:14}")
private Integer brokerMetricsSaveDays;
@Value(value = "${task.metrics.delete-metrics.topic-metrics-save-days:7}")
@Value(value = "${task.metrics.delete.topic-metrics-save-days:7}")
private Integer topicMetricsSaveDays;
@Value(value = "${task.metrics.delete-metrics.topic-request-time-metrics-save-days:7}")
@Value(value = "${task.metrics.delete.topic-request-time-metrics-save-days:7}")
private Integer topicRequestTimeMetricsSaveDays;
@Value(value = "${task.metrics.delete-metrics.topic-throttled-metrics-save-days:7}")
@Value(value = "${task.metrics.delete.topic-throttled-metrics-save-days:7}")
private Integer topicThrottledMetricsSaveDays;
@Value(value = "${task.metrics.delete-metrics.app-topic-metrics-save-days:7}")
@Value(value = "${task.metrics.delete.app-topic-metrics-save-days:7}")
private Integer appTopicMetricsSaveDays;
@Override

146
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/dispatch/metrics/store/StoreBrokerMetrics.java
View File

@@ -1,146 +0,0 @@
package com.xiaojukeji.kafka.manager.task.dispatch.metrics.store;
import com.xiaojukeji.kafka.manager.common.constant.Constant;
import com.xiaojukeji.kafka.manager.common.constant.KafkaMetricsCollections;
import com.xiaojukeji.kafka.manager.common.constant.LogConstant;
import com.xiaojukeji.kafka.manager.common.entity.metrics.BaseMetrics;
import com.xiaojukeji.kafka.manager.common.entity.metrics.BrokerMetrics;
import com.xiaojukeji.kafka.manager.common.entity.metrics.ClusterMetrics;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.common.utils.jmx.JmxConstant;
import com.xiaojukeji.kafka.manager.common.zookeeper.znode.brokers.TopicMetadata;
import com.xiaojukeji.kafka.manager.dao.BrokerMetricsDao;
import com.xiaojukeji.kafka.manager.dao.ClusterMetricsDao;
import com.xiaojukeji.kafka.manager.common.entity.pojo.BrokerMetricsDO;
import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterDO;
import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterMetricsDO;
import com.xiaojukeji.kafka.manager.service.cache.PhysicalClusterMetadataManager;
import com.xiaojukeji.kafka.manager.service.service.ClusterService;
import com.xiaojukeji.kafka.manager.service.service.JmxService;
import com.xiaojukeji.kafka.manager.service.strategy.AbstractHealthScoreStrategy;
import com.xiaojukeji.kafka.manager.service.utils.MetricsConvertUtils;
import com.xiaojukeji.kafka.manager.task.component.AbstractScheduledTask;
import com.xiaojukeji.kafka.manager.task.component.CustomScheduled;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
/**
* Broker指标信息存DB, Broker流量, 集群流量
* @author zengqiao
* @date 20/5/7
*/
@CustomScheduled(name = "storeBrokerMetrics", cron = "21 0/1 * * * ?", threadNum = 2)
@ConditionalOnProperty(prefix = "custom.store-metrics-task.community", name = "broker-metrics-enabled", havingValue = "true", matchIfMissing = true)
public class StoreBrokerMetrics extends AbstractScheduledTask<ClusterDO> {
private static final Logger LOGGER = LoggerFactory.getLogger(LogConstant.SCHEDULED_TASK_LOGGER);
@Autowired
private JmxService jmxService;
@Autowired
private ClusterService clusterService;
@Autowired
private BrokerMetricsDao brokerMetricsDao;
@Autowired
private ClusterMetricsDao clusterMetricsDao;
@Autowired
private AbstractHealthScoreStrategy healthScoreStrategy;
@Override
protected List<ClusterDO> listAllTasks() {
return clusterService.list();
}
@Override
public void processTask(ClusterDO clusterDO) {
long startTime = System.currentTimeMillis();
List<ClusterMetrics> clusterMetricsList = new ArrayList<>();
try {
List<BrokerMetrics> brokerMetricsList = getAndBatchAddMetrics(startTime, clusterDO.getId());
clusterMetricsList.add(supplyAndConvert2ClusterMetrics(
clusterDO.getId(),
MetricsConvertUtils.merge2BaseMetricsByAdd(brokerMetricsList))
);
} catch (Exception t) {
LOGGER.error("collect failed, clusterId:{}.", clusterDO.getId(), t);
}
long endTime = System.currentTimeMillis();
LOGGER.info("collect finish, clusterId:{} costTime:{}", clusterDO.getId(), endTime - startTime);
List<ClusterMetricsDO> doList = MetricsConvertUtils.convertAndUpdateCreateTime2ClusterMetricsDOList(
startTime,
clusterMetricsList
);
if (ValidateUtils.isEmptyList(doList)) {
return;
}
clusterMetricsDao.batchAdd(doList);
}
private List<BrokerMetrics> getAndBatchAddMetrics(Long startTime, Long clusterId) {
List<BrokerMetrics> metricsList = new ArrayList<>();
for (Integer brokerId: PhysicalClusterMetadataManager.getBrokerIdList(clusterId)) {
BrokerMetrics metrics = jmxService.getBrokerMetrics(
clusterId,
brokerId,
KafkaMetricsCollections.BROKER_TO_DB_METRICS
);
if (ValidateUtils.isNull(metrics)) {
continue;
}
metrics.getMetricsMap().put(
JmxConstant.HEALTH_SCORE,
healthScoreStrategy.calBrokerHealthScore(clusterId, brokerId, metrics)
);
metricsList.add(metrics);
}
if (ValidateUtils.isEmptyList(metricsList)) {
return new ArrayList<>();
}
List<BrokerMetricsDO> doList =
MetricsConvertUtils.convertAndUpdateCreateTime2BrokerMetricsDOList(startTime, metricsList);
int i = 0;
do {
List<BrokerMetricsDO> subDOList = doList.subList(i, Math.min(i + Constant.BATCH_INSERT_SIZE, doList.size()));
if (ValidateUtils.isEmptyList(subDOList)) {
break;
}
brokerMetricsDao.batchAdd(subDOList);
i += Constant.BATCH_INSERT_SIZE;
} while (i < doList.size());
return metricsList;
}
private ClusterMetrics supplyAndConvert2ClusterMetrics(Long clusterId, BaseMetrics baseMetrics) {
ClusterMetrics metrics = new ClusterMetrics(clusterId);
Map<String, Object> metricsMap = metrics.getMetricsMap();
metricsMap.putAll(baseMetrics.getMetricsMap());
metricsMap.put(JmxConstant.TOPIC_NUM, PhysicalClusterMetadataManager.getTopicNameList(clusterId).size());
metricsMap.put(JmxConstant.BROKER_NUM, PhysicalClusterMetadataManager.getBrokerIdList(clusterId).size());
Integer partitionNum = 0;
for (String topicName : PhysicalClusterMetadataManager.getTopicNameList(clusterId)) {
TopicMetadata topicMetaData = PhysicalClusterMetadataManager.getTopicMetadata(clusterId, topicName);
if (ValidateUtils.isNull(topicMetaData)) {
continue;
}
partitionNum += topicMetaData.getPartitionNum();
}
metricsMap.put(JmxConstant.PARTITION_NUM, partitionNum);
return metrics;
}
}

55
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/sink/db/SinkBrokerMetrics2DB.java Normal file
View File

@@ -0,0 +1,55 @@
package com.xiaojukeji.kafka.manager.task.listener.sink.db;
import com.xiaojukeji.kafka.manager.common.constant.Constant;
import com.xiaojukeji.kafka.manager.common.entity.metrics.BrokerMetrics;
import com.xiaojukeji.kafka.manager.common.entity.pojo.BrokerMetricsDO;
import com.xiaojukeji.kafka.manager.common.events.metrics.BatchBrokerMetricsCollectedEvent;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.dao.BrokerMetricsDao;
import com.xiaojukeji.kafka.manager.service.utils.MetricsConvertUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* @author zengqiao
* @date 22/01/17
*/
@Component
@ConditionalOnProperty(prefix = "task.metrics.sink.broker-metrics", name = "sink-db-enabled", havingValue = "true", matchIfMissing = true)
public class SinkBrokerMetrics2DB implements ApplicationListener<BatchBrokerMetricsCollectedEvent> {
private static final Logger logger = LoggerFactory.getLogger(SinkBrokerMetrics2DB.class);
@Autowired
private BrokerMetricsDao metricsDao;
@Override
public void onApplicationEvent(BatchBrokerMetricsCollectedEvent event) {
logger.debug("sink broker-metrics to db start, event:{}.", event);
List<BrokerMetrics> metricsList = event.getMetricsList();
if (ValidateUtils.isEmptyList(metricsList)) {
logger.warn("sink broker-metrics to db finished, without need sink, event:{}.", event);
return;
}
List<BrokerMetricsDO> doList = MetricsConvertUtils.convertAndUpdateCreateTime2BrokerMetricsDOList(event.getCollectTime(), metricsList);
int i = 0;
while (i < doList.size()) {
List<BrokerMetricsDO> subDOList = doList.subList(i, Math.min(i + Constant.BATCH_INSERT_SIZE, doList.size()));
if (ValidateUtils.isEmptyList(subDOList)) {
break;
}
metricsDao.batchAdd(subDOList);
i += Constant.BATCH_INSERT_SIZE;
}
logger.debug("sink broker-metrics to db finished, event:{}.", event);
}
}

80
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/sink/db/SinkClusterMetrics2DB.java Normal file
View File

@@ -0,0 +1,80 @@
package com.xiaojukeji.kafka.manager.task.listener.sink.db;
import com.xiaojukeji.kafka.manager.common.entity.metrics.BaseMetrics;
import com.xiaojukeji.kafka.manager.common.entity.metrics.BrokerMetrics;
import com.xiaojukeji.kafka.manager.common.entity.metrics.ClusterMetrics;
import com.xiaojukeji.kafka.manager.common.entity.pojo.ClusterMetricsDO;
import com.xiaojukeji.kafka.manager.common.events.metrics.BatchBrokerMetricsCollectedEvent;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.common.utils.jmx.JmxConstant;
import com.xiaojukeji.kafka.manager.common.zookeeper.znode.brokers.TopicMetadata;
import com.xiaojukeji.kafka.manager.dao.ClusterMetricsDao;
import com.xiaojukeji.kafka.manager.service.cache.PhysicalClusterMetadataManager;
import com.xiaojukeji.kafka.manager.service.utils.MetricsConvertUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
/**
* @author zengqiao
* @date 22/01/17
*/
@Component
@ConditionalOnProperty(prefix = "task.metrics.sink.cluster-metrics", name = "sink-db-enabled", havingValue = "true", matchIfMissing = true)
public class SinkClusterMetrics2DB implements ApplicationListener<BatchBrokerMetricsCollectedEvent> {
private static final Logger logger = LoggerFactory.getLogger(SinkClusterMetrics2DB.class);
@Autowired
private ClusterMetricsDao clusterMetricsDao;
@Override
public void onApplicationEvent(BatchBrokerMetricsCollectedEvent event) {
logger.debug("sink cluster-metrics to db start, event:{}.", event);
List<BrokerMetrics> metricsList = event.getMetricsList();
if (ValidateUtils.isEmptyList(metricsList)) {
logger.warn("sink cluster-metrics to db finished, without need sink, event:{}.", event);
return;
}
List<ClusterMetricsDO> doList = MetricsConvertUtils.convertAndUpdateCreateTime2ClusterMetricsDOList(
event.getCollectTime(),
// 合并broker-metrics为cluster-metrics
Arrays.asList(supplyAndConvert2ClusterMetrics(event.getPhysicalClusterId(), MetricsConvertUtils.merge2BaseMetricsByAdd(event.getMetricsList())))
);
if (ValidateUtils.isEmptyList(doList)) {
logger.warn("sink cluster-metrics to db finished, without need sink, event:{}.", event);
return;
}
clusterMetricsDao.batchAdd(doList);
logger.debug("sink cluster-metrics to db finished, event:{}.", event);
}
private ClusterMetrics supplyAndConvert2ClusterMetrics(Long clusterId, BaseMetrics baseMetrics) {
ClusterMetrics metrics = new ClusterMetrics(clusterId);
Map<String, Object> metricsMap = metrics.getMetricsMap();
metricsMap.putAll(baseMetrics.getMetricsMap());
metricsMap.put(JmxConstant.TOPIC_NUM, PhysicalClusterMetadataManager.getTopicNameList(clusterId).size());
metricsMap.put(JmxConstant.BROKER_NUM, PhysicalClusterMetadataManager.getBrokerIdList(clusterId).size());
Integer partitionNum = 0;
for (String topicName : PhysicalClusterMetadataManager.getTopicNameList(clusterId)) {
TopicMetadata topicMetaData = PhysicalClusterMetadataManager.getTopicMetadata(clusterId, topicName);
if (ValidateUtils.isNull(topicMetaData)) {
continue;
}
partitionNum += topicMetaData.getPartitionNum();
}
metricsMap.put(JmxConstant.PARTITION_NUM, partitionNum);
return metrics;
}
}

2
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/StoreCommunityTopicMetrics2DB.java → kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/sink/db/StoreCommunityTopicMetrics2DB.java
View File

@@ -1,4 +1,4 @@
package com.xiaojukeji.kafka.manager.task.listener;
package com.xiaojukeji.kafka.manager.task.listener.sink.db;
import com.xiaojukeji.kafka.manager.common.constant.Constant;
import com.xiaojukeji.kafka.manager.common.constant.LogConstant;

2
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/StoreTopicThrottledMetrics2DB.java → kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/sink/db/StoreTopicThrottledMetrics2DB.java
View File

@@ -1,4 +1,4 @@
package com.xiaojukeji.kafka.manager.task.listener;
package com.xiaojukeji.kafka.manager.task.listener.sink.db;
import com.xiaojukeji.kafka.manager.common.bizenum.KafkaClientEnum;
import com.xiaojukeji.kafka.manager.common.constant.Constant;

2
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/SinkCommunityTopicMetrics2Kafka.java → kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/sink/kafka/SinkCommunityTopicMetrics2Kafka.java
View File

@@ -1,4 +1,4 @@
package com.xiaojukeji.kafka.manager.task.listener;
package com.xiaojukeji.kafka.manager.task.listener.sink.kafka;
import com.xiaojukeji.kafka.manager.common.constant.ConfigConstant;
import com.xiaojukeji.kafka.manager.common.constant.LogConstant;

2
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/SinkConsumerMetrics2Kafka.java → kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/sink/kafka/SinkConsumerMetrics2Kafka.java
View File

@@ -1,4 +1,4 @@
package com.xiaojukeji.kafka.manager.task.listener;
package com.xiaojukeji.kafka.manager.task.listener.sink.kafka;
import com.xiaojukeji.kafka.manager.common.constant.ConfigConstant;
import com.xiaojukeji.kafka.manager.common.constant.LogConstant;

2
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/SinkCommunityTopicMetrics2Monitor.java → kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/sink/monitor/SinkCommunityTopicMetrics2Monitor.java
View File

@@ -1,4 +1,4 @@
package com.xiaojukeji.kafka.manager.task.listener;
package com.xiaojukeji.kafka.manager.task.listener.sink.monitor;
import com.xiaojukeji.kafka.manager.monitor.common.entry.bizenum.MonitorMetricNameEnum;
import com.xiaojukeji.kafka.manager.common.constant.LogConstant;

2
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/SinkConsumerMetrics2Monitor.java → kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/sink/monitor/SinkConsumerMetrics2Monitor.java
View File

@@ -1,4 +1,4 @@
package com.xiaojukeji.kafka.manager.task.listener;
package com.xiaojukeji.kafka.manager.task.listener.sink.monitor;
import com.xiaojukeji.kafka.manager.monitor.common.entry.bizenum.MonitorMetricNameEnum;
import com.xiaojukeji.kafka.manager.common.constant.LogConstant;

2
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/SinkTopicThrottledMetrics2Monitor.java → kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/listener/sink/monitor/SinkTopicThrottledMetrics2Monitor.java
View File

@@ -1,4 +1,4 @@
package com.xiaojukeji.kafka.manager.task.listener;
package com.xiaojukeji.kafka.manager.task.listener.sink.monitor;
import com.xiaojukeji.kafka.manager.common.bizenum.KafkaClientEnum;
import com.xiaojukeji.kafka.manager.monitor.common.MonitorSinkConstant;

5
kafka-manager-task/src/main/java/com/xiaojukeji/kafka/manager/task/schedule/metadata/FlushZKConsumerGroupMetadata.java
View File

@@ -32,6 +32,9 @@ public class FlushZKConsumerGroupMetadata {
@Autowired
private ClusterService clusterService;
@Autowired
private ThreadPool threadPool;
@Scheduled(cron="35 0/1 * * * ?")
public void schedule() {
List<ClusterDO> doList = clusterService.list();
@@ -95,7 +98,7 @@ public class FlushZKConsumerGroupMetadata {
return new ArrayList<>();
}
});
ThreadPool.submitCollectMetricsTask(taskList[i]);
threadPool.submitCollectMetricsTask(clusterId, taskList[i]);
}
Map<String, Set<String>> topicNameConsumerGroupMap = new HashMap<>();

6
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/api/versionone/thirdpart/ThirdPartBrokerController.java
View File

@@ -32,7 +32,7 @@ import java.util.stream.Collectors;
*/
@Api(tags = "开放接口-Broker相关接口(REST)")
@RestController
@RequestMapping(ApiPrefix.API_V1_THIRD_PART_OP_PREFIX)
@RequestMapping(ApiPrefix.API_V1_THIRD_PART_PREFIX)
public class ThirdPartBrokerController {
@Autowired
private BrokerService brokerService;
@@ -44,7 +44,7 @@ public class ThirdPartBrokerController {
private ClusterService clusterService;
@ApiOperation(value = "Broker信息概览", notes = "")
@RequestMapping(value = "{clusterId}/brokers/{brokerId}/overview", method = RequestMethod.GET)
@GetMapping(value = "{clusterId}/brokers/{brokerId}/overview")
@ResponseBody
public Result<ThirdPartBrokerOverviewVO> getBrokerOverview(@PathVariable Long clusterId,
@PathVariable Integer brokerId) {
@@ -70,7 +70,7 @@ public class ThirdPartBrokerController {
}
@ApiOperation(value = "BrokerRegion信息", notes = "所有集群的")
@RequestMapping(value = "broker-regions", method = RequestMethod.GET)
@GetMapping(value = "broker-regions")
@ResponseBody
public Result<List<BrokerRegionVO>> getBrokerRegions() {
List<ClusterDO> clusterDOList = clusterService.list();

7
kafka-manager-web/src/main/java/com/xiaojukeji/kafka/manager/web/config/SwaggerConfig.java
View File

@@ -1,5 +1,7 @@
package com.xiaojukeji.kafka.manager.web.config;
import com.xiaojukeji.kafka.manager.service.utils.ConfigUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.config.annotation.*;
@@ -20,6 +22,9 @@ import springfox.documentation.swagger2.annotations.EnableSwagger2;
@EnableWebMvc
@EnableSwagger2
public class SwaggerConfig implements WebMvcConfigurer {
@Autowired
private ConfigUtils configUtils;
@Override
public void addResourceHandlers(ResourceHandlerRegistry registry) {
registry.addResourceHandler("swagger-ui.html").addResourceLocations("classpath:/META-INF/resources/");
@@ -41,7 +46,7 @@ public class SwaggerConfig implements WebMvcConfigurer {
return new ApiInfoBuilder()
.title("LogiKM接口文档")
.description("欢迎使用滴滴LogiKM")
.version("2.5.0")
.version(configUtils.getApplicationVersion())
.build();
}

58
kafka-manager-web/src/main/resources/application.yml
View File

@@ -9,6 +9,7 @@ server:
spring:
application:
name: kafkamanager
version: @project.version@
profiles:
active: dev
datasource:
@@ -32,7 +33,6 @@ custom:
idc: cn
store-metrics-task:
community:
broker-metrics-enabled: true
topic-metrics-enabled: true
didi: # 滴滴Kafka特有的指标
app-topic-metrics-enabled: false
@@ -42,13 +42,20 @@ custom:
# 任务相关的配置
task:
op:
sync-topic-enabled: false # 未落盘的Topic定期同步到DB中
order-auto-exec: # 工单自动化审批线程的开关
topic-enabled: false # Topic工单自动化审批开关, false:关闭自动化审批, true:开启
app-enabled: false # App工单自动化审批开关, false:关闭自动化审批, true:开启
sync-topic-enabled: false # 未落盘的Topic定期同步到DB中
order-auto-exec: # 工单自动化审批线程的开关
topic-enabled: false # Topic工单自动化审批开关, false:关闭自动化审批, true:开启
app-enabled: false # App工单自动化审批开关, false:关闭自动化审批, true:开启
metrics:
delete-metrics:
delete-limit-size: 1000
collect: # 收集指标
broker-metrics-enabled: true # 收集Broker指标
sink: # 上报指标
cluster-metrics: # 上报cluster指标
sink-db-enabled: true # 上报到db
broker-metrics: # 上报broker指标
sink-db-enabled: true # 上报到db
delete: # 删除指标
delete-limit-size: 1000 # 单次删除的批大小
cluster-metrics-save-days: 14 # 集群指标保存天数
broker-metrics-save-days: 14 # Broker指标保存天数
topic-metrics-save-days: 7 # Topic指标保存天数
@@ -56,7 +63,25 @@ task:
topic-throttled-metrics-save-days: 7 # Topic限流指标保存天数
app-topic-metrics-save-days: 7 # App+Topic指标保存天数
thread-pool:
collect-metrics:
thread-num: 256 # 收集指标线程池大小
queue-size: 5000 # 收集指标线程池的queue大小
api-call:
thread-num: 16 # api服务线程池大小
queue-size: 5000 # api服务线程池的queue大小
client-pool:
kafka-consumer:
min-idle-client-num: 24 # 最小空闲客户端数
max-idle-client-num: 24 # 最大空闲客户端数
max-total-client-num: 24 # 最大客户端数
borrow-timeout-unit-ms: 3000 # 租借超时时间，单位毫秒
account:
jump-login:
gateway-api: false # 网关接口
third-part-api: false # 第三方接口
ldap:
enabled: false
url: ldap://127.0.0.1:389/
@@ -70,19 +95,20 @@ account:
auth-user-registration: true
auth-user-registration-role: normal
kcm:
enabled: false
s3:
kcm: # 集群安装部署，仅安装broker
enabled: false # 是否开启
s3: # s3 存储服务
endpoint: s3.didiyunapi.com
access-key: 1234567890
secret-key: 0987654321
bucket: logi-kafka
n9e:
base-url: http://127.0.0.1:8004
user-token: 12345678
timeout: 300
account: root
script-file: kcm_script.sh
n9e: # 夜莺
base-url: http://127.0.0.1:8004 # 夜莺job服务地址
user-token: 12345678 # 用户的token
timeout: 300 # 当台操作的超时时间
account: root # 操作时使用的账号
script-file: kcm_script.sh # 脚本，已内置好，在源码的kcm模块内，此处配置无需修改
logikm-url: http://127.0.0.1:8080 # logikm部署地址，部署时kcm_script.sh会调用logikm检查部署中的一些状态
monitor:
enabled: false

5
pom.xml
View File

@@ -65,6 +65,11 @@
<artifactId>swagger-annotations</artifactId>
<version>${swagger.version}</version>
</dependency>
<dependency>
<groupId>io.swagger</groupId>
<artifactId>swagger-models</artifactId>
<version>${swagger.version}</version>
</dependency>
<!-- javax -->
<dependency>