Compare commits

..

3 Commits

Author SHA1 Message Date
EricZeng
89405fe003 Merge pull request #434 from didi/fix_2.5.0
修复console模块关闭问题及前端文件名错误问题
2022-01-13 14:00:01 +08:00
shirenchuang
b9ea3865a5 升级到2.5版本
(cherry picked from commit 5bc6eb6774)
2022-01-13 13:47:21 +08:00
孙超
b5bd643814 修复图片名称大小写问题
(cherry picked from commit ada2718b5e)
2022-01-13 13:46:06 +08:00
147 changed files with 926 additions and 1830 deletions

View File

@@ -5,7 +5,7 @@
**一站式`Apache Kafka`集群指标监控与运维管控平台**
`LogiKM开源至今备受关注考虑到开源项目应该更贴合Apache Kafka未来发展方向经项目组慎重考虑预计22年5月份将其品牌升级成Know Streaming届时项目名称和Logo也将统一更新感谢大家一如既往的支持敬请期待`
阅读本README文档您可以了解到滴滴Logi-KafkaManager的用户群体、产品定位等信息并通过体验地址快速体验Kafka集群指标监控与运维管控的全流程。
@@ -55,56 +55,41 @@
## 2 相关文档
### 2.1 产品文档
- [滴滴LogiKM 安装手册](docs/install_guide/install_guide_cn.md)
- [滴滴LogiKM 接入集群](docs/user_guide/add_cluster/add_cluster.md)
- [滴滴LogiKM 用户使用手册](docs/user_guide/user_guide_cn.md)
- [滴滴LogiKM FAQ](docs/user_guide/faq.md)
- [滴滴Logi-KafkaManager 安装手册](docs/install_guide/install_guide_cn.md)
- [滴滴Logi-KafkaManager 接入集群](docs/user_guide/add_cluster/add_cluster.md)
- [滴滴Logi-KafkaManager 用户使用手册](docs/user_guide/user_guide_cn.md)
- [滴滴Logi-KafkaManager FAQ](docs/user_guide/faq.md)
### 2.2 社区文章
- [滴滴云官网产品介绍](https://www.didiyun.com/production/logi-KafkaManager.html)
- [7年沉淀之作--滴滴Logi日志服务套件](https://mp.weixin.qq.com/s/-KQp-Qo3WKEOc9wIR2iFnw)
- [滴滴LogiKM 一站式Kafka监控与管控平台](https://mp.weixin.qq.com/s/9qSZIkqCnU6u9nLMvOOjIQ)
- [滴滴LogiKM 开源之路](https://xie.infoq.cn/article/0223091a99e697412073c0d64)
- [滴滴LogiKM 系列视频教程](https://space.bilibili.com/442531657/channel/seriesdetail?sid=571649)
- [kafka最强最全知识图谱](https://www.szzdzhp.com/kafka/)
- [滴滴LogiKM新用户入门系列文章专栏 --石臻](https://www.szzdzhp.com/categories/LogIKM/)
- [kafka实践十五滴滴开源Kafka管控平台 LogiKM研究--A叶子叶来](https://blog.csdn.net/yezonggang/article/details/113106244)
- [滴滴Logi-KafkaManager 一站式Kafka监控与管控平台](https://mp.weixin.qq.com/s/9qSZIkqCnU6u9nLMvOOjIQ)
- [滴滴Logi-KafkaManager 开源之路](https://xie.infoq.cn/article/0223091a99e697412073c0d64)
- [滴滴Logi-KafkaManager 系列视频教程](https://mp.weixin.qq.com/s/9X7gH0tptHPtfjPPSdGO8g)
- [kafka实践十五滴滴开源Kafka管控平台 Logi-KafkaManager研究--A叶子叶来](https://blog.csdn.net/yezonggang/article/details/113106244)
- [kafka的灵魂伴侣Logi-KafkaManager系列文章专栏 --石臻](https://blog.csdn.net/u010634066/category_10977588.html)
## 3 滴滴Logi开源用户交流群
![image](https://user-images.githubusercontent.com/5287750/111266722-e531d800-8665-11eb-9242-3484da5a3099.png)
想跟各个大佬交流Kafka Es 等中间件/大数据相关技术请 加微信进群。
微信加群:添加<font color=red>mike_zhangliang</font><font color=red>danke-xie</font>的微信号备注Logi加群或关注公众号 云原生可观测性 回复 "Logi加群"
微信加群添加mike_zhangliang的微信号备注Logi加群或关注公众号 云原生可观测性 回复 "Logi加群"
## 4 知识星球
<img width="447" alt="image" src="https://user-images.githubusercontent.com/71620349/147314042-843a371a-48c0-4d9a-a65e-ca40236f3300.png">
![image](https://user-images.githubusercontent.com/51046167/140718512-5ab1b336-5c48-46c0-90bd-44b5c7e168c8.png)
<br>
<center>
✅我们正在组建国内最大最权威的
</center>
<br>
<center>
<font color=red size=5><b>【Kafka中文社区】</b></font>
</center>
在这里你可以结交各大互联网Kafka大佬以及近2000+Kafka爱好者一起实现知识共享实时掌控最新行业资讯期待您的加入中https://z.didi.cn/5gSF9
<font color=red size=5>有问必答~ </font>
<font color=red size=5>互动有礼~ </font>
PS:提问请尽量把问题一次性描述清楚,并告知环境信息情况哦~!如使用版本、操作步骤、报错/警告信息等方便大V们快速解答
✅知识星球首个【Kafka中文社区】内测期免费加入https://z.didi.cn/5gSF9
有问必答~
互动有礼~
1600+群友一起共建国内最专业的【Kafka中文社区】
PS:提问请尽量把问题一次性描述清楚,并告知环境信息情况哦~!如使用版本、操作步骤、报错/警告信息等,方便嘉宾们快速解答~
## 5 项目成员
### 5.1 内部核心人员
`iceyuhui``liuyaguang``limengmonty``zhangliangmike``xiepeng``nullhuangyiming``zengqiao``eilenexuzhe``huangjiaweihjw``zhaoyinrui``marzkonglingxu``joysunchao``石臻臻`
`iceyuhui``liuyaguang``limengmonty``zhangliangmike``nullhuangyiming``zengqiao``eilenexuzhe``huangjiaweihjw``zhaoyinrui``marzkonglingxu``joysunchao`
### 5.2 外部贡献者

View File

@@ -7,39 +7,6 @@
---
## v2.6.0
版本上线时间2022-01-24
### 能力提升
- 增加简单回退工具类
### 体验优化
- 补充周期任务说明文档
- 补充集群安装部署使用说明文档
- 升级Swagger、SpringFramework、SpringBoot、EChats版本
- 优化Task模块的日志输出
- 优化corn表达式解析失败后退出无任何日志提示问题
- Ldap用户接入时增加部门及邮箱信息等
- 对Jmx模块增加连接失败后的回退机制及错误日志优化
- 增加线程池、客户端池可配置
- 删除无用的jmx_prometheus_javaagent-0.14.0.jar
- 优化迁移任务名称
- 优化创建Region时Region容量信息不能立即被更新问题
- 引入lombok
- 更新视频教程
- 优化kcm_script.sh脚本中的LogiKM地址为可通过程序传入
- 第三方接口及网关接口,增加是否跳过登录的开关
- extends模块相关配置调整为非必须在application.yml中配置
### bug修复
- 修复批量往DB写入空指标数组时报SQL语法异常的问题
- 修复网关增加配置及修改配置时version不变化问题
- 修复集群列表页,提示框遮挡问题
- 修复对高版本Broker元信息协议解析失败的问题
- 修复Dockerfile执行时提示缺少application.yml文件的问题
- 修复逻辑集群更新时,会报空指针的问题
## v2.4.1+
版本上线时间2021-05-21

View File

@@ -19,7 +19,6 @@ ENV JAVA_OPTS="-verbose:gc \
RUN wget https://github.com/didi/Logi-KafkaManager/releases/download/v${VERSION}/kafka-manager-${VERSION}.tar.gz && \
tar xvf kafka-manager-${VERSION}.tar.gz && \
mv kafka-manager-${VERSION}/kafka-manager.jar /opt/app.jar && \
mv kafka-manager-${VERSION}/application.yml /opt/application.yml && \
rm -rf kafka-manager-${VERSION}*
EXPOSE 8080 9999

View File

@@ -55,7 +55,7 @@ data:
didi:
app-topic-metrics-enabled: false
topic-request-time-metrics-enabled: false
topic-throttled-metrics-enabled: false
topic-throttled-metrics: false
save-days: 7
# 任务相关的开关
@@ -67,16 +67,7 @@ data:
# ldap settings
ldap:
enabled: 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
authUserRegistration: false
kcm:
enabled: false

View File

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

View File

@@ -15,7 +15,6 @@ server:
spring:
application:
name: kafkamanager
version: 2.6.0
profiles:
active: dev
datasource:
@@ -27,6 +26,7 @@ spring:
main:
allow-bean-definition-overriding: true
servlet:
multipart:
max-file-size: 100MB
@@ -36,58 +36,29 @@ logging:
config: classpath:logback-spring.xml
custom:
idc: cn
idc: cn # 部署的数据中心, 忽略该配置, 后续会进行删除
jmx:
max-conn: 10 # 2.3版本配置不在这个地方生效
store-metrics-task:
community:
topic-metrics-enabled: true
didi: # 滴滴Kafka特有的指标
app-topic-metrics-enabled: false
topic-request-time-metrics-enabled: false
topic-throttled-metrics-enabled: false
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: false # 滴滴埋入的指标, 社区AK不存在该指标因此默认关闭
save-days: 7 #指标在DB中保持的天数-1表示永久保存7表示保存近7天的数据
# 任务相关的配置
# 任务相关的开关
task:
op:
sync-topic-enabled: false # 未落盘的Topic定期同步到DB中
order-auto-exec: # 工单自动化审批线程的开关
topic-enabled: false # Topic工单自动化审批开关, false:关闭自动化审批, true:开启
app-enabled: false # App工单自动化审批开关, false:关闭自动化审批, true:开启
metrics:
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指标保存天数
topic-request-time-metrics-save-days: 7 # Topic请求耗时指标保存天数
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 # 租借超时时间,单位毫秒
sync-topic-enabled: false # 未落盘的Topic定期同步到DB中
order-auto-exec: # 工单自动化审批线程的开关
topic-enabled: false # Topic工单自动化审批开关, false:关闭自动化审批, true:开启
app-enabled: false # App工单自动化审批开关, false:关闭自动化审批, true:开启
# ldap相关的配置
account:
jump-login:
gateway-api: false # 网关接口
third-part-api: false # 第三方接口
ldap:
enabled: false
url: ldap://127.0.0.1:389/
@@ -101,20 +72,28 @@ account:
auth-user-registration: true
auth-user-registration-role: normal
kcm: # 集群安装部署仅安装broker
enabled: false # 是否开启
s3: # s3 存储服务
# 集群升级部署相关的功能需要配合夜莺及S3进行使用
kcm:
enabled: false
s3:
endpoint: s3.didiyunapi.com
access-key: 1234567890
secret-key: 0987654321
bucket: logi-kafka
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检查部署中的一些状态
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: 用户资源中心地址
monitor:
enabled: false
@@ -128,9 +107,10 @@ monitor:
rdb:
base-url: http://127.0.0.1:8000 # 夜莺v4版本默认端口统一调整为了8000
notify:
kafka:
cluster-id: 95
topic-name: didi-kafka-notify
order:
detail-url: http://127.0.0.1
notify: # 通知的功能
kafka: # 默认通知发送到kafka的指定Topic中
cluster-id: 95 # Topic的集群ID
topic-name: didi-kafka-notify # Topic名称
order: # 部署的KM的地址
detail-url: http://127.0.0.1

View File

@@ -13,9 +13,6 @@ CREATE TABLE `account` (
`username` varchar(128) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL DEFAULT '' COMMENT '用户名',
`password` varchar(128) NOT NULL DEFAULT '' COMMENT '密码',
`role` tinyint(8) NOT NULL DEFAULT '0' COMMENT '角色类型, 0:普通用户 1:研发 2:运维',
`department` varchar(256) DEFAULT '' COMMENT '部门名',
`display_name` varchar(256) DEFAULT '' COMMENT '用户姓名',
`mail` varchar(256) DEFAULT '' COMMENT '邮箱',
`status` int(16) NOT NULL DEFAULT '0' COMMENT '0标识使用中-1标识已废弃',
`gmt_create` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`gmt_modify` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',

View File

@@ -39,14 +39,4 @@ ALTER TABLE `gateway_config`
ADD COLUMN `description` TEXT NULL COMMENT '描述信息' AFTER `version`;
```
### 升级至`2.6.0`版本
#### 1.mysql变更
`2.6.0`版本在`account`表增加用户姓名部门名邮箱三个字段因此需要执行下面的sql进行字段的增加。
```sql
ALTER TABLE `account`
ADD COLUMN `display_name` VARCHAR(256) NOT NULL DEFAULT '' COMMENT '用户名' AFTER `role`,
ADD COLUMN `department` VARCHAR(256) NOT NULL DEFAULT '' COMMENT '部门名' AFTER `display_name`,
ADD COLUMN `mail` VARCHAR(256) NOT NULL DEFAULT '' COMMENT '邮箱' AFTER `department`;
```

Binary file not shown.

Before

Width:  |  Height:  |  Size: 69 KiB

View File

@@ -1,89 +0,0 @@
<mxfile host="65bd71144e">
<diagram id="bhaMuW99Q1BzDTtcfRXp" name="Page-1">
<mxGraphModel dx="1138" dy="830" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="1169" pageHeight="827" math="0" shadow="0">
<root>
<mxCell id="0"/>
<mxCell id="1" parent="0"/>
<mxCell id="11" value="待部署Kafka-Broker的机器" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;labelPosition=center;verticalLabelPosition=bottom;align=center;verticalAlign=top;dashed=1;" vertex="1" parent="1">
<mxGeometry x="380" y="240" width="320" height="240" as="geometry"/>
</mxCell>
<mxCell id="24" value="" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;labelPosition=center;verticalLabelPosition=bottom;align=center;verticalAlign=top;dashed=1;fillColor=#eeeeee;strokeColor=#36393d;" vertex="1" parent="1">
<mxGeometry x="410" y="310" width="260" height="160" as="geometry"/>
</mxCell>
<mxCell id="6" style="edgeStyle=none;html=1;entryX=0;entryY=0.5;entryDx=0;entryDy=0;" edge="1" parent="1" source="2" target="3">
<mxGeometry relative="1" as="geometry"/>
</mxCell>
<mxCell id="7" value="调用夜莺接口,&lt;br&gt;创建集群安装部署任务" style="edgeLabel;html=1;align=center;verticalAlign=middle;resizable=0;points=[];" vertex="1" connectable="0" parent="6">
<mxGeometry x="-0.0875" y="1" relative="1" as="geometry">
<mxPoint x="9" y="1" as="offset"/>
</mxGeometry>
</mxCell>
<mxCell id="9" style="edgeStyle=none;html=1;" edge="1" parent="1" source="2" target="4">
<mxGeometry relative="1" as="geometry"/>
</mxCell>
<mxCell id="10" value="通过版本管理将Kafka的安装包&lt;br&gt;server配置上传到s3中" style="edgeLabel;html=1;align=center;verticalAlign=middle;resizable=0;points=[];" vertex="1" connectable="0" parent="9">
<mxGeometry x="0.0125" y="2" relative="1" as="geometry">
<mxPoint as="offset"/>
</mxGeometry>
</mxCell>
<mxCell id="2" value="LogiKM" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;fillColor=#dae8fc;strokeColor=#6c8ebf;" vertex="1" parent="1">
<mxGeometry x="40" y="100" width="120" height="40" as="geometry"/>
</mxCell>
<mxCell id="12" style="edgeStyle=none;html=1;exitX=0.5;exitY=1;exitDx=0;exitDy=0;entryX=0.5;entryY=0;entryDx=0;entryDy=0;" edge="1" parent="1" source="3" target="5">
<mxGeometry relative="1" as="geometry"/>
</mxCell>
<mxCell id="13" value="1、下发任务脚本(kcm_script.sh)&lt;br&gt;2、下发任务操作命令" style="edgeLabel;html=1;align=center;verticalAlign=middle;resizable=0;points=[];" vertex="1" connectable="0" parent="12">
<mxGeometry x="-0.0731" y="2" relative="1" as="geometry">
<mxPoint x="-2" y="-16" as="offset"/>
</mxGeometry>
</mxCell>
<mxCell id="3" value="夜莺——任务中心" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;fillColor=#cdeb8b;strokeColor=#36393d;" vertex="1" parent="1">
<mxGeometry x="480" y="100" width="120" height="40" as="geometry"/>
</mxCell>
<mxCell id="4" value="S3" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;fillColor=#ffe6cc;strokeColor=#d79b00;" vertex="1" parent="1">
<mxGeometry x="40" y="310" width="120" height="40" as="geometry"/>
</mxCell>
<mxCell id="5" value="夜莺——Agent(&lt;font color=&quot;#ff3333&quot;&gt;代理执行kcm_script.sh脚本&lt;/font&gt;)" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;fillColor=#d5e8d4;strokeColor=#82b366;" vertex="1" parent="1">
<mxGeometry x="400" y="260" width="280" height="40" as="geometry"/>
</mxCell>
<mxCell id="22" style="edgeStyle=orthogonalEdgeStyle;html=1;entryX=1;entryY=0.5;entryDx=0;entryDy=0;fontColor=#FF3333;exitX=0;exitY=0.5;exitDx=0;exitDy=0;" edge="1" parent="1" source="14" target="4">
<mxGeometry relative="1" as="geometry"/>
</mxCell>
<mxCell id="25" value="下载安装包" style="edgeLabel;html=1;align=center;verticalAlign=middle;resizable=0;points=[];fontColor=#000000;" vertex="1" connectable="0" parent="22">
<mxGeometry x="0.2226" y="-2" relative="1" as="geometry">
<mxPoint x="27" y="2" as="offset"/>
</mxGeometry>
</mxCell>
<mxCell id="14" value="执行kcm_script.sh脚本下载安装包" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;fillColor=#eeeeee;strokeColor=#36393d;" vertex="1" parent="1">
<mxGeometry x="425" y="320" width="235" height="20" as="geometry"/>
</mxCell>
<mxCell id="18" value="执行kcm_script.sh脚本安装" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;fillColor=#eeeeee;strokeColor=#36393d;" vertex="1" parent="1">
<mxGeometry x="425" y="350" width="235" height="20" as="geometry"/>
</mxCell>
<mxCell id="19" value="执行kcm_script.sh脚本检查安装结果" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;fillColor=#eeeeee;strokeColor=#36393d;" vertex="1" parent="1">
<mxGeometry x="425" y="380" width="235" height="20" as="geometry"/>
</mxCell>
<mxCell id="23" style="edgeStyle=orthogonalEdgeStyle;html=1;entryX=0.5;entryY=0;entryDx=0;entryDy=0;fontColor=#FF3333;exitX=1;exitY=0.5;exitDx=0;exitDy=0;" edge="1" parent="1" source="20" target="2">
<mxGeometry relative="1" as="geometry">
<Array as="points">
<mxPoint x="770" y="420"/>
<mxPoint x="770" y="40"/>
<mxPoint x="100" y="40"/>
</Array>
</mxGeometry>
</mxCell>
<mxCell id="26" value="检查副本同步状态" style="edgeLabel;html=1;align=center;verticalAlign=middle;resizable=0;points=[];fontColor=#000000;" vertex="1" connectable="0" parent="23">
<mxGeometry x="-0.3344" relative="1" as="geometry">
<mxPoint as="offset"/>
</mxGeometry>
</mxCell>
<mxCell id="20" value="执行kcm_script.sh脚本检查副本同步" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;fillColor=#eeeeee;strokeColor=#36393d;" vertex="1" parent="1">
<mxGeometry x="425" y="410" width="235" height="20" as="geometry"/>
</mxCell>
<mxCell id="21" value="执行kcm_script.sh脚本结束" style="rounded=0;whiteSpace=wrap;html=1;absoluteArcSize=1;arcSize=14;strokeWidth=1;fillColor=#eeeeee;strokeColor=#36393d;" vertex="1" parent="1">
<mxGeometry x="425" y="440" width="235" height="20" as="geometry"/>
</mxCell>
</root>
</mxGraphModel>
</diagram>
</mxfile>

View File

@@ -136,8 +136,7 @@ EXPIRED_TOPIC_CONFIG
配置Value
```json
{
"minExpiredDay": 30, #过期时间大于此值才显示,
"filterRegex": ".*XXX\\s+", #忽略符合此正则规则的Topic
"minExpiredDay": 30, #过期时间大于此值才显示
"ignoreClusterIdList": [ # 忽略的集群
50
]

View File

@@ -0,0 +1,27 @@
---
![kafka-manager-logo](../../assets/images/common/logo_name.png)
**一站式`Apache Kafka`集群指标监控与运维管控平台**
---
# 升级至`2.2.0`版本
`2.2.0`版本在`cluster`表及`logical_cluster`各增加了一个字段因此需要执行下面的sql进行字段的增加。
```sql
# cluster表中增加jmx_properties字段, 这个字段会用于存储jmx相关的认证以及配置信息
ALTER TABLE `cluster` ADD COLUMN `jmx_properties` TEXT NULL COMMENT 'JMX配置' AFTER `security_properties`;
# logical_cluster中增加identification字段, 同时数据和原先name数据相同, 最后增加一个唯一键.
# 此后, name字段还是表示集群名称, identification字段表示的是集群标识, 只能是字母数字及下划线组成,
# 数据上报到监控系统时, 集群这个标识采用的字段就是identification字段, 之前使用的是name字段.
ALTER TABLE `logical_cluster` ADD COLUMN `identification` VARCHAR(192) NOT NULL DEFAULT '' COMMENT '逻辑集群标识' AFTER `name`;
UPDATE `logical_cluster` SET `identification`=`name` WHERE id>=0;
ALTER TABLE `logical_cluster` ADD INDEX `uniq_identification` (`identification` ASC);
```

View File

@@ -0,0 +1,17 @@
---
![kafka-manager-logo](../../assets/images/common/logo_name.png)
**一站式`Apache Kafka`集群指标监控与运维管控平台**
---
# 升级至`2.3.0`版本
`2.3.0`版本在`gateway_config`表增加了一个描述说明的字段因此需要执行下面的sql进行字段的增加。
```sql
ALTER TABLE `gateway_config`
ADD COLUMN `description` TEXT NULL COMMENT '描述信息' AFTER `version`;
```

View File

@@ -1,39 +0,0 @@
---
![kafka-manager-logo](../assets/images/common/logo_name.png)
**一站式`Apache Kafka`集群指标监控与运维管控平台**
---
| 定时任务名称或方法名 | 所在类 | 详细说明 | cron | cron说明 | 线程数量 |
| -------------------------------------- | -------------------------------------- | ------------------------------------------ | --------------- | --------------------------------------- | -------- |
| calKafkaBill | CalKafkaTopicBill | 计算Kafka使用账单 | 0 0 1 * * | 每天凌晨1点执行一次 | 1 |
| calRegionCapacity | CalRegionCapacity | 计算Region容量 | 0 0 0/12 * * | 每隔12小时执行一次在0分钟0秒时触发 | 1 |
| calTopicStatistics | CalTopicStatistics | 定时计算Topic统计数据 | 0 0 0/4 * * ? | 每隔4小时执行一次在0分钟0秒时触发 | 5 |
| flushBrokerTable | FlushBrokerTable | 定时刷新BrokerTable数据 | 0 0 0/1 * * ? | 每隔1小时执行一次在0分钟0秒时触发 | 1 |
| flushExpiredTopic | FlushExpiredTopic | 定期更新过期Topic | 0 0 0/5 * * ? | 每隔5小时执行一次在0分钟0秒时触发 | 1 |
| syncClusterTaskState | SyncClusterTaskState | 同步更新集群任务状态 | 0 0/1 * * * ? | 每隔1分钟执行一次在每分钟的0秒时触发 | 1 |
| newCollectAndPublishCGData | CollectAndPublishCGData | 收集并发布消费者指标数据 | 30 0/1 * * * ? | 每隔1分钟执行一次在每分钟的30秒时触发 | 10 |
| collectAndPublishCommunityTopicMetrics | CollectAndPublishCommunityTopicMetrics | Topic社区指标收集 | 31 0/1 * * * ? | 每隔1分钟执行一次在每分钟的30秒时触发 | 5 |
| collectAndPublishTopicThrottledMetrics | CollectAndPublishTopicThrottledMetrics | 收集和发布Topic限流信息 | 11 0/1 * * * ? | 每隔1分钟执行一次在每分钟的11秒时触发 | 5 |
| deleteMetrics | DeleteMetrics | 定期删除Metrics信息 | 0 0/2 * * * ? | 每隔2分钟执行一次在每分钟的0秒时触发 | 1 |
| storeDiDiAppTopicMetrics | StoreDiDiAppTopicMetrics | JMX中获取appId维度的流量信息存DB | 41 0/1 * * * ? | 每隔1分钟执行一次在每分钟的41秒时触发 | 5 |
| storeDiDiTopicRequestTimeMetrics | StoreDiDiTopicRequestTimeMetrics | JMX中获取的TopicRequestTimeMetrics信息存DB | 51 0/1 * * * ? | 每隔1分钟执行一次在每分钟的51秒时触发 | 5 |
| autoHandleTopicOrder | AutoHandleTopicOrder | 定时自动处理Topic相关工单 | 0 0/1 * * * ? | 每隔1分钟执行一次在每分钟的0秒时触发 | 1 |
| automatedHandleOrder | AutomatedHandleOrder | 工单自动化审批 | 0 0/1 * * * ? | 每隔1分钟执行一次在每分钟的0秒时触发 | 1 |
| flushReassignment | FlushReassignment | 定时处理分区迁移任务 | 0 0/1 * * * ? | 每隔1分钟执行一次在每分钟的0秒时触发 | 1 |
| syncTopic2DB | SyncTopic2DB | 定期将未落盘的Topic刷新到DB中 | 0 0/2 * * * ? | 每隔2分钟执行一次在每分钟的0秒时触发 | 1 |
| sinkCommunityTopicMetrics2Monitor | SinkCommunityTopicMetrics2Monitor | 定时上报Topic监控指标 | 1 0/1 * * * ? | 每隔1分钟执行一次在每分钟的1秒时触发 | 5 |
| flush方法 | LogicalClusterMetadataManager | 定时刷新逻辑集群元数据到缓存中 | 0/30 * * * * ? | 每隔30秒执行一次 | 1 |
| flush方法 | AccountServiceImpl | 定时刷新account信息到缓存中 | 0/5 * * * * ? | 每隔5秒执行一次 | 1 |
| ipFlush方法 | HeartBeat | 定时获取管控平台所在机器IP等信息到DB | 0/10 * * * * ? | 每隔10秒执行一次 | 1 |
| flushTopicMetrics方法 | FlushTopicMetrics | 定时刷新topic指标到缓存中 | 5 0/1 * * * ? | 每隔1分钟执行一次在每分钟的5秒时触发 | 1 |
| schedule方法 | FlushBKConsumerGroupMetadata | 定时刷新broker上消费组信息到缓存中 | 15 0/1 * * * ? | 每隔1分钟执行一次在每分钟的15秒时触发 | 1 |
| flush方法 | FlushClusterMetadata | 定时刷新物理集群元信息到缓存中 | 0/30 * * * * ? | 每隔30秒执行一次 | 1 |
| flush方法 | FlushTopicProperties | 定时刷新物理集群配置到缓存中 | 25 0/1 * * * ? | 每隔1分钟执行一次在每分钟的25秒时触发 | 1 |
| schedule方法 | FlushZKConsumerGroupMetadata | 定时刷新zk上的消费组信息到缓存中 | 35 0/1 * * * ? | 每隔1分钟执行一次在每分钟的35秒时触发 | 1 |

View File

@@ -1,89 +0,0 @@
---
![kafka-manager-logo](../assets/images/common/logo_name.png)
**一站式`Apache Kafka`集群指标监控与运维管控平台**
---
# 如何使用集群安装部署功能?
[TOC]
## 1、实现原理
![KCM实现原理](./assets/kcm/kcm_principle.png)
- LogiKM上传安装包到S3服务
- LogiKM调用夜莺-Job服务接口创建执行[kcm_script.sh](https://github.com/didi/LogiKM/blob/master/kafka-manager-extends/kafka-manager-kcm/src/main/resources/kcm_script.sh)脚本的任务kcm_script.sh脚本是安装部署Kafka集群的脚本
- 夜莺将任务脚本下发到具体的机器上通过夜莺Agent执行该脚本
- kcm_script.sh脚本会进行Kafka-Broker的安装部署
---
## 2、使用方式
### 2.1、第一步:修改配置
**配置application.yml文件**
```yaml
#
kcm:
enabled: false # 是否开启将其修改为true
s3: # s3 存储服务
endpoint: s3.didiyunapi.com
access-key: 1234567890
secret-key: 0987654321
bucket: logi-kafka
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检查部署中的一些状态这里只需要填写 http://IP:PORT 就可以了
account:
jump-login:
gateway-api: false # 网关接口
third-part-api: false # 第三方接口将其修改为true即允许未登录情况下调用开放的第三方接口
```
### 2.2、第二步:检查服务
**检查s3服务**
- 测试 "运维管控-》集群运维-》版本管理" 页面的上传查看等功能是否都正常。如果存在不正常则需要查看s3的配置是否正确
- 如果都没有问题则上传Kafka的以.tgz结尾的安装包以及server.properties文件
**检查夜莺Job服务**
- 创建一个job任务机器选择需要安装Kafka集群的机器然后执行的命令是echo "Hello LogiKM",看能否被成功执行。如果不行,则需要检查夜莺的安装;
- 如果没有问题则表示夜莺和所需部署的机器之间的交互是没有问题的;
### 2.3、第三步:接入集群
在LogiKM的 “运维管控-》集群列表” 中接入需要安装部署的集群,**PS此时是允许接入一个没有任何Broker的空的Kafka集群**其中对的bootstrapServers配置搭建完成后的Kafka集群地址就可以了而ZK地址必须和集群的server.properties中的ZK地址保持一致
### 2.4、第四步:部署集群
- 打开LogiKM的 “运维管控-》集群运维-》集群任务” 页面,点击 “新建集群任务” 按钮;
- 选择集群、任务类型、包版本、server配置及填写主机列表然后点击确认即可在夜莺的Job服务中心中创建一个任务出来。**PS如果创建失败可以看一下日志我为什么创建失败**
- 随后可以点击详情及状态对任务进行操作;
### 2.5、可能问题
#### 2.5.1、问题一:任务执行超时、失败等
进入夜莺Job服务中心查看对应的任务的相关日志
- 提示安装包下载失败则需要查看对应的s3服务是否可以直接wget下载安装包如果不可以则需要对kcm_script.sh脚本进行修改
- 提示调用LogiKM失败则可以使用postman手动测试一下kcm_script.sh脚本调用LogiKM的那个接口是否有问题如果存在问题则进行相应的修改PS具体接口见kcm_script.sh脚本
## 3、备注说明
- 集群安装部署仅安装部署Kafka-Broker不安装Kafka的ZK服务
- 安装部署中有任何定制化的需求例如修改安装的目录等可以通过修改kcm_script.sh脚本实现
- kcm_script.sh脚本位置[kcm_script.sh](https://github.com/didi/LogiKM/blob/master/kafka-manager-extends/kafka-manager-kcm/src/main/resources/kcm_script.sh)

View File

@@ -51,7 +51,7 @@ custom:
didi:
app-topic-metrics-enabled: false # 滴滴埋入的指标, 社区AK不存在该指标因此默认关闭
topic-request-time-metrics-enabled: false # 滴滴埋入的指标, 社区AK不存在该指标因此默认关闭
topic-throttled-metrics-enabled: false # 滴滴埋入的指标, 社区AK不存在该指标因此默认关闭
topic-throttled-metrics: false # 滴滴埋入的指标, 社区AK不存在该指标因此默认关闭
save-days: 7 #指标在DB中保持的天数-1表示永久保存7表示保存近7天的数据
# 任务相关的开关

View File

@@ -30,7 +30,6 @@
- 18、如何在不登录的情况下调用一些需要登录的接口
- 19、为什么无法看到连接信息、耗时信息等指标
- 20、AppID鉴权、生产消费配额不起作用
- 21、如何查看周期任务说明文档
---
@@ -214,6 +213,3 @@ AppID鉴权、生产消费配额依赖于滴滴kafka-gateway通过gateway进
具体见:[滴滴Logi-KafkaManager开源版和商业版特性对比](../开源版与商业版特性对比.md)
### 20、如何查看周期任务说明文档
具体见:[周期任务说明文档](../dev_guide/周期任务说明文档.md)

View File

@@ -21,12 +21,15 @@
<java_target_version>1.8</java_target_version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<file_encoding>UTF-8</file_encoding>
<spring-version>5.1.3.RELEASE</spring-version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
<version>${spring-version}</version>
</dependency>
<!-- http -->
@@ -106,11 +109,5 @@
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>compile</scope>
</dependency>
</dependencies>
</project>

View File

@@ -20,6 +20,12 @@ 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;

View File

@@ -9,7 +9,7 @@ public class Constant {
public static final Integer MAX_AVG_BYTES_DURATION = 10;
public static final Integer BATCH_INSERT_SIZE = 30;
public static final Integer BATCH_INSERT_SIZE = 50;
public static final Integer DEFAULT_SESSION_TIMEOUT_UNIT_MS = 30000;

View File

@@ -17,10 +17,6 @@ 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() {
}
}

View File

@@ -0,0 +1,16 @@
package com.xiaojukeji.kafka.manager.common.constant;
/**
* @author zengqiao
* @date 20/8/10
*/
public class LogConstant {
public static final String COLLECTOR_METRICS_LOGGER = "COLLECTOR_METRICS_LOGGER";
public static final String API_METRICS_LOGGER = "API_METRICS_LOGGER";
public static final String SCHEDULED_TASK_LOGGER = "SCHEDULED_TASK_LOGGER";
private LogConstant() {
}
}

View File

@@ -1,18 +0,0 @@
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;
}

View File

@@ -10,8 +10,6 @@ import java.util.List;
public class TopicExpiredConfig {
private Integer minExpiredDay = 30;
private String filterRegex = "";
private List<Long> ignoreClusterIdList = new ArrayList<>();
public Integer getMinExpiredDay() {
@@ -30,19 +28,10 @@ public class TopicExpiredConfig {
this.ignoreClusterIdList = ignoreClusterIdList;
}
public String getFilterRegex() {
return filterRegex;
}
public void setFilterRegex(String filterRegex) {
this.filterRegex = filterRegex;
}
@Override
public String toString() {
return "TopicExpiredConfig{" +
"minExpiredDay=" + minExpiredDay +
", filterRegex='" + filterRegex + '\'' +
", ignoreClusterIdList=" + ignoreClusterIdList +
'}';
}

View File

@@ -21,15 +21,6 @@ public class AccountDTO {
@ApiModelProperty(value = "角色")
private Integer role;
@ApiModelProperty(value = "用户姓名")
private String displayName;
@ApiModelProperty(value = "部门")
private String department;
@ApiModelProperty(value = "邮箱")
private String mail;
public String getUsername() {
return username;
}
@@ -54,39 +45,12 @@ public class AccountDTO {
this.role = role;
}
public String getDisplayName() {
return displayName;
}
public void setDisplayName(String displayName) {
this.displayName = displayName;
}
public String getDepartment() {
return department;
}
public void setDepartment(String department) {
this.department = department;
}
public String getMail() {
return mail;
}
public void setMail(String mail) {
this.mail = mail;
}
@Override
public String toString() {
return "AccountDTO{" +
"username='" + username + '\'' +
", password='" + password + '\'' +
", role=" + role +
", displayName='" + displayName + '\'' +
", department='" + department + '\'' +
", mail='" + mail + '\'' +
'}';
}

View File

@@ -21,12 +21,6 @@ public class AccountDO {
private Integer role;
private String displayName;
private String department;
private String mail;
public String getUsername() {
return username;
}
@@ -51,43 +45,16 @@ public class AccountDO {
this.role = role;
}
public String getDisplayName() {
return displayName;
}
public void setDisplayName(String displayName) {
this.displayName = displayName;
}
public String getDepartment() {
return department;
}
public void setDepartment(String department) {
this.department = department;
}
public String getMail() {
return mail;
}
public void setMail(String mail) {
this.mail = mail;
}
@Override
public String toString() {
return "AccountDO{" +
"id=" + id +
"username='" + username + '\'' +
", password='" + password + '\'' +
", role=" + role +
", id=" + id +
", status=" + status +
", gmtCreate=" + gmtCreate +
", gmtModify=" + gmtModify +
", username='" + username + '\'' +
", password='" + password + '\'' +
", role=" + role +
", displayName='" + displayName + '\'' +
", department='" + department + '\'' +
", mail='" + mail + '\'' +
'}';
}
}

View File

@@ -1,20 +0,0 @@
package com.xiaojukeji.kafka.manager.common.events;
import com.xiaojukeji.kafka.manager.common.entity.pojo.RegionDO;
import lombok.Getter;
import org.springframework.context.ApplicationEvent;
/**
* Region创建事件
* @author zengqiao
* @date 22/01/1
*/
@Getter
public class RegionCreatedEvent extends ApplicationEvent {
private final RegionDO regionDO;
public RegionCreatedEvent(Object source, RegionDO regionDO) {
super(source);
this.regionDO = regionDO;
}
}

View File

@@ -1,33 +0,0 @@
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;
}
}

View File

@@ -1,22 +0,0 @@
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;
}
}

View File

@@ -1,18 +1,9 @@
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;
@@ -26,50 +17,4 @@ 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;
}
}

View File

@@ -1,14 +0,0 @@
package com.xiaojukeji.kafka.manager.common.utils;
/**
* @className: SplitUtils
* @description: Split string of type keyValue
* @author: Hu.Yue
* @date: 2021/8/4
**/
public class SplitUtils {
public static String keyValueSplit(String keyValue){
return keyValue.split(":\\s+")[1];
}
}

View File

@@ -1,10 +1,5 @@
package com.xiaojukeji.kafka.manager.common.utils.jmx;
import lombok.Data;
import lombok.ToString;
@Data
@ToString
public class JmxConfig {
/**
* 单台最大连接数
@@ -26,8 +21,45 @@ public class JmxConfig {
*/
private Boolean openSSL;
/**
* 连接重试回退事件
*/
private Long retryConnectBackoffTimeUnitMs;
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 +
'}';
}
}

View File

@@ -14,11 +14,11 @@ import javax.naming.Context;
import javax.rmi.ssl.SslRMIClientSocketFactory;
import java.io.IOException;
import java.net.MalformedURLException;
import java.util.Arrays;
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包装类
@@ -26,27 +26,19 @@ import java.util.concurrent.locks.ReentrantLock;
* @date 2015/11/9.
*/
public class JmxConnectorWrap {
private static final Logger LOGGER = LoggerFactory.getLogger(JmxConnectorWrap.class);
private final static Logger LOGGER = LoggerFactory.getLogger(JmxConnectorWrap.class);
private final Long physicalClusterId;
private String host;
private final Integer brokerId;
private final String host;
private final int port;
private int port;
private JMXConnector jmxConnector;
private final AtomicInteger atomicInteger;
private AtomicInteger atomicInteger;
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;
public JmxConnectorWrap(String host, int port, JmxConfig jmxConfig) {
this.host = host;
this.port = port;
this.jmxConfig = jmxConfig;
@@ -54,12 +46,7 @@ public class JmxConnectorWrap {
this.jmxConfig = new JmxConfig();
}
if (ValidateUtils.isNullOrLessThanZero(this.jmxConfig.getMaxConn())) {
// 默认设置20
this.jmxConfig.setMaxConn(20);
}
if (ValidateUtils.isNullOrLessThanZero(this.jmxConfig.getRetryConnectBackoffTimeUnitMs())) {
// 默认回退10分钟
this.jmxConfig.setRetryConnectBackoffTimeUnitMs(10 * 60 * 1000L);
this.jmxConfig.setMaxConn(1);
}
this.atomicInteger = new AtomicInteger(this.jmxConfig.getMaxConn());
}
@@ -71,40 +58,17 @@ public class JmxConnectorWrap {
if (port == -1) {
return false;
}
return safeCreateJmxConnector();
return createJmxConnector();
}
public void close() {
this.closeJmxConnect();
}
public void closeJmxConnect() {
public synchronized void close() {
if (jmxConnector == null) {
return;
}
try {
modifyJMXConnectorLock.lock();
// 移除设置的backoff事件
BackoffUtils.removeNeedBackoffEvent(buildConnectJmxFailedBackoffEventKey(physicalClusterId, brokerId));
jmxConnector.close();
} 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();
} catch (IOException e) {
LOGGER.warn("close JmxConnector exception, host:{} port:{}.", host, port, e);
}
}
@@ -112,12 +76,6 @@ 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>();
@@ -125,9 +83,7 @@ public class JmxConnectorWrap {
// fixed by riyuetianmu
environment.put(JMXConnector.CREDENTIALS, new String[]{this.jmxConfig.getUsername(), this.jmxConfig.getPassword()});
}
if (jmxConfig.getOpenSSL() != null && this.jmxConfig.getOpenSSL()) {
// 开启ssl
if (jmxConfig.isOpenSSL() != null && this.jmxConfig.isOpenSSL()) {
environment.put(Context.SECURITY_PROTOCOL, "ssl");
SslRMIClientSocketFactory clientSocketFactory = new SslRMIClientSocketFactory();
environment.put(RMIConnectorServer.RMI_CLIENT_SOCKET_FACTORY_ATTRIBUTE, clientSocketFactory);
@@ -135,17 +91,13 @@ public class JmxConnectorWrap {
}
jmxConnector = JMXConnectorFactory.connect(new JMXServiceURL(jmxUrl), environment);
LOGGER.info("connect JMX success, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port);
LOGGER.info("JMX connect success, host:{} port:{}.", host, port);
return true;
} catch (MalformedURLException e) {
LOGGER.error("connect JMX failed, JMX url exception, physicalClusterId:{} brokerId:{} host:{} port:{} jmxUrl:{}.", physicalClusterId, brokerId, host, port, jmxUrl, e);
LOGGER.error("JMX url exception, host:{} port:{} jmxUrl:{}", host, port, jmxUrl, e);
} catch (Exception e) {
LOGGER.error("connect JMX failed, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port, e);
LOGGER.error("JMX connect exception, host:{} port:{}.", host, port, e);
}
// 设置连接backoff
BackoffUtils.putNeedBackoffEvent(buildConnectJmxFailedBackoffEventKey(physicalClusterId, brokerId), this.jmxConfig.getRetryConnectBackoffTimeUnitMs());
return false;
}
@@ -159,11 +111,6 @@ public class JmxConnectorWrap {
acquire();
MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
return mBeanServerConnection.getAttribute(name, attribute);
} catch (IOException ioe) {
// io错误则重置连接
this.closeJmxConnect();
throw ioe;
} finally {
atomicInteger.incrementAndGet();
}
@@ -179,11 +126,6 @@ public class JmxConnectorWrap {
acquire();
MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
return mBeanServerConnection.getAttributes(name, attributes);
} catch (IOException ioe) {
// io错误则重置连接
this.closeJmxConnect();
throw ioe;
} finally {
atomicInteger.incrementAndGet();
}
@@ -196,11 +138,6 @@ public class JmxConnectorWrap {
acquire();
MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
return mBeanServerConnection.queryNames(name, query);
} catch (IOException ioe) {
// io错误则重置连接
this.closeJmxConnect();
throw ioe;
} finally {
atomicInteger.incrementAndGet();
}
@@ -223,8 +160,4 @@ public class JmxConnectorWrap {
}
}
}
private static String buildConnectJmxFailedBackoffEventKey(Long physicalClusterId, Integer brokerId) {
return "CONNECT_JMX_FAILED_BACK_OFF_EVENT_PHY_" + physicalClusterId + "_BROKER_" + brokerId;
}
}

View File

@@ -1,17 +1,6 @@
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
@@ -21,7 +10,7 @@ import java.util.Map;
* 节点结构:
* {
* "listener_security_protocol_map":{"SASL_PLAINTEXT":"SASL_PLAINTEXT"},
* "endpoints":["SASL_PLAINTEXT://127.0.0.1:9093"],
* "endpoints":["SASL_PLAINTEXT://10.179.162.202:9093"],
* "jmx_port":9999,
* "host":null,
* "timestamp":"1546632983233",
@@ -29,48 +18,22 @@ import java.util.Map;
* "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
* }
*
*/
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class BrokerMetadata implements Serializable {
private static final long serialVersionUID = 3918113492423375809L;
public class BrokerMetadata implements Cloneable {
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;
@JsonProperty("jmx_port")
private int jmxPort;
/*
* ZK上对应的字段就是这个名字, 不要进行修改
*/
private int jmx_port;
private String version;
@@ -78,54 +41,91 @@ public class BrokerMetadata implements Serializable {
private String rack;
@JsonIgnore
public String getExternalHost() {
if (!endpointMap.containsKey(KafkaConstant.EXTERNAL_KEY)) {
return null;
}
return endpointMap.get(KafkaConstant.EXTERNAL_KEY).getIp();
public long getClusterId() {
return clusterId;
}
@JsonIgnore
public String getInternalHost() {
if (!endpointMap.containsKey(KafkaConstant.INTERNAL_KEY)) {
return null;
}
return endpointMap.get(KafkaConstant.INTERNAL_KEY).getIp();
public void setClusterId(long clusterId) {
this.clusterId = clusterId;
}
public static void parseAndUpdateBrokerMetadata(BrokerMetadata brokerMetadata) {
brokerMetadata.setEndpointMap(new HashMap<>());
public int getBrokerId() {
return brokerId;
}
if (brokerMetadata.getEndpoints().isEmpty()) {
return;
}
public void setBrokerId(int brokerId) {
this.brokerId = brokerId;
}
// 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 List<String> getEndpoints() {
return endpoints;
}
String brokerHost = endpoint.substring(idx1 + "://".length(), idx2);
String brokerPort = endpoint.substring(idx2 + 1);
public void setEndpoints(List<String> endpoints) {
this.endpoints = endpoints;
}
brokerMetadata.getEndpointMap().put(endpoint.substring(0, idx1), new IpPortData(brokerHost, brokerPort));
public String getHost() {
return host;
}
if (KafkaConstant.EXTERNAL_KEY.equals(endpoint.substring(0, idx1))) {
// 优先使用external的地址进行展示
brokerMetadata.setHost(brokerHost);
brokerMetadata.setPort(NumberUtils.string2Integer(brokerPort));
}
public void setHost(String host) {
this.host = host;
}
if (null == brokerMetadata.getHost()) {
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 + '\'' +
'}';
}
}

View File

@@ -1,10 +1,9 @@
{
"name": "logi-kafka",
"version": "2.6.0",
"version": "2.5.0",
"description": "",
"scripts": {
"prestart": "npm install --save-dev webpack-dev-server",
"start": "webpack serve",
"start": "webpack-dev-server",
"daily-build": "cross-env NODE_ENV=production webpack",
"pre-build": "cross-env NODE_ENV=production webpack",
"prod-build": "cross-env NODE_ENV=production webpack",
@@ -14,19 +13,18 @@
"license": "ISC",
"devDependencies": {
"@hot-loader/react-dom": "^16.8.6",
"@types/events": "^3.0.0",
"@types/echarts": "^4.4.1",
"@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": "^5.2.1",
"echarts": "^4.5.0",
"file-loader": "^5.0.2",
"html-webpack-plugin": "^3.2.0",
"increase-memory-limit": "^1.0.7",
@@ -52,10 +50,11 @@
"typescript": "^3.3.3333",
"url-loader": "^4.1.1",
"webpack": "^4.29.6",
"webpack-cli": "^4.9.1",
"webpack-cli": "^3.2.3",
"webpack-dev-server": "^3.2.1",
"xlsx": "^0.16.1"
},
"dependencies": {
"format-to-json": "^1.0.4"
}
}
}

View File

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

View File

@@ -3,34 +3,16 @@ import { DatePicker, notification, Spin } from 'component/antd';
import moment, { Moment } from 'moment';
import { timeStampStr } from 'constants/strategy';
import { disabledDate } from 'lib/utils';
import * as echarts from 'echarts/core';
import echarts from 'echarts';
// 引入柱状图
import { BarChart, LineChart } from 'echarts/charts';
// 引入柱状图和折线图
import 'echarts/lib/chart/bar';
import 'echarts/lib/chart/line';
// 引入提示框和标题组件
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 'echarts/lib/component/tooltip';
import 'echarts/lib/component/title';
import 'echarts/lib/component/legend';
import './index.less';
const { RangePicker } = DatePicker;
@@ -79,12 +61,11 @@ export class ChartWithDatePicker extends React.Component<IChartProps> {
const noData = options.series.length ? false : true;
this.setState({ noData });
options.tooltip.formatter = (params: any) => {
let res =
'<div style=\'margin-bottom:5px;padding:0 12px;width:100%;height:24px;line-height:24px;border-radius:3px;\'><p>' +
var 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>';
// tslint:disable-next-line:prefer-for-of
for (let i = 0; i < params.length; i++) {
" </p></div>";
for (var 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,
@@ -94,7 +75,7 @@ export class ChartWithDatePicker extends React.Component<IChartProps> {
</div>`;
}
return res;
};
}
this.chart.setOption(options, true);
}

View File

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

View File

@@ -1,45 +1,25 @@
import React from 'react';
import * as echarts from 'echarts/core';
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 './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?: any;
options?: EChartOption;
}
export const hasData = (options: any) => {
export const hasData = (options: EChartOption) => {
if (options && options.series && options.series.length) return true;
return false;
};
export default class LineCharts extends React.Component<IEchartsProps> {
export default class LineChart extends React.Component<IEchartsProps> {
public id = null as HTMLDivElement;
public myChart = null as echarts.ECharts;
@@ -47,7 +27,7 @@ export default class LineCharts extends React.Component<IEchartsProps> {
public componentDidMount() {
const { options } = this.props;
this.myChart = echarts.init(this.id);
this.myChart.setOption(options, true);
this.myChart.setOption(options);
window.addEventListener('resize', this.resize);
}
@@ -61,7 +41,7 @@ export default class LineCharts extends React.Component<IEchartsProps> {
public refresh = () => {
const { options } = this.props;
this.myChart.setOption(options, true);
this.myChart.setOption(options);
}
public resize = () => {
@@ -70,6 +50,6 @@ export default class LineCharts 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`}} />;
}
}

View File

@@ -295,7 +295,7 @@ export class ClusterList extends SearchAndFilterContainer {
cancelText="取消"
okText="确认"
>
<Tooltip placement="bottom" title="暂停监控将无法正常监控指标信息,建议开启监控">
<Tooltip placement="left" title="暂停监控将无法正常监控指标信息,建议开启监控">
<a
className="action-button"
>

View File

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

View File

@@ -1,5 +1,5 @@
import moment from 'moment';
import { EChartsOption } from 'echarts';
import { EChartOption } 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 EChartsOption;
} as EChartOption;
};
export const parseLine = (option: ICurve, data: ILineData[]): EChartsOption => {
export const parseLine = (option: ICurve, data: ILineData[]): EChartOption => {
return Object.assign({}, getBaseOptions(option, data), {
legend: {
...baseLineLegend,
bottom: '0',
align: 'auto',
},
}) as EChartsOption;
}) as EChartOption;
};
export const parseBrokerMetricOption = (option: ICurve, data: IClusterMetrics[]): EChartsOption => {
export const parseBrokerMetricOption = (option: ICurve, data: IClusterMetrics[]): EChartOption => {
let name;
let series: ISeriesOption[];
data = data || [];

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 { EChartsOption } from 'echarts';
import { EChartOption } 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: EChartsOption, loading = false) => {
public renderEchart = (options: EChartOption, 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 EChartsOption;
const option = this.getChartOption() as EChartOption;
return (
<>

View File

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

View File

@@ -1,4 +1,4 @@
import { EChartsOption } from 'echarts';
import { EChartOption } 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[]) => any;
parser?: (option: ICurve, data: any[]) => EChartOption;
}
@observer
@@ -41,7 +41,7 @@ export class CommonCurve extends React.Component<ICommonCurveProps> {
fullScreen.show(this.renderCurve(options, loading, true));
}
public renderOpBtns = (options: EChartsOption, expand = false) => {
public renderOpBtns = (options: EChartOption, 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: EChartsOption, loading = false) => {
public renderEchart = (options: EChartOption, 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: any, loading: boolean, expand = false) => {
public renderCurve = (options: EChartOption, loading: boolean, expand = false) => {
const data = hasData(options);
return (
<div className="common-chart-wrapper" >

View File

@@ -145,7 +145,7 @@ export const Header = observer((props: IHeader) => {
<div className="left-content">
<img className="kafka-header-icon" src={logoUrl} alt="" />
<span className="kafka-header-text">LogiKM</span>
<a className='kafka-header-version' href="https://github.com/didi/Logi-KafkaManager/releases" target='_blank'>v2.6.0</a>
<a className='kafka-header-version' href="https://github.com/didi/Logi-KafkaManager/releases" target='_blank'>v2.5.0</a>
{/* 添加版本超链接 */}
</div>
<div className="mid-content">

View File

@@ -1,6 +1,6 @@
import { observable, action } from 'mobx';
import moment = require('moment');
import { EChartsOption } from 'echarts';
import { EChartOption } from 'echarts/lib/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]: EChartsOption } = {};
public curveData: { [key: string]: EChartOption } = {};
@observable
public curveLoading: { [key: string]: boolean } = {};
@@ -25,7 +25,7 @@ class CurveInfo {
public currentOperator: string;
@action.bound
public setCurveData(key: curveKeys | string, data: EChartsOption) {
public setCurveData(key: curveKeys | string, data: EChartOption) {
this.curveData[key] = data;
}
@@ -59,7 +59,7 @@ class CurveInfo {
public getCommonCurveData = (
options: ICurve,
parser: (option: ICurve, data: any[]) => EChartsOption,
parser: (option: ICurve, data: any[]) => EChartOption,
reload?: boolean) => {
const { path } = options;
this.setCurveData(path, null);

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/': {

View File

@@ -24,6 +24,7 @@
<java_target_version>1.8</java_target_version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<file_encoding>UTF-8</file_encoding>
<spring-version>5.1.3.RELEASE</spring-version>
</properties>
<dependencies>
@@ -37,10 +38,12 @@
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${spring-version}</version>
</dependency>
<!-- javax -->

View File

@@ -14,8 +14,6 @@ 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;
@@ -27,22 +25,9 @@ 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
*/
@@ -99,7 +84,7 @@ public class KafkaClientPool {
return true;
}
private void initKafkaConsumerPool(ClusterDO clusterDO) {
private static void initKafkaConsumerPool(ClusterDO clusterDO) {
lock.lock();
try {
GenericObjectPool<KafkaConsumer<String, String>> objectPool = KAFKA_CONSUMER_POOL.get(clusterDO.getId());
@@ -107,9 +92,9 @@ public class KafkaClientPool {
return;
}
GenericObjectPoolConfig<KafkaConsumer<String, String>> config = new GenericObjectPoolConfig<>();
config.setMaxIdle(kafkaConsumerMaxIdleClientNum);
config.setMinIdle(kafkaConsumerMinIdleClientNum);
config.setMaxTotal(kafkaConsumerMaxTotalClientNum);
config.setMaxIdle(24);
config.setMinIdle(24);
config.setMaxTotal(24);
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);
@@ -133,7 +118,7 @@ public class KafkaClientPool {
}
}
public KafkaConsumer<String, String> borrowKafkaConsumerClient(ClusterDO clusterDO) {
public static KafkaConsumer<String, String> borrowKafkaConsumerClient(ClusterDO clusterDO) {
if (ValidateUtils.isNull(clusterDO)) {
return null;
}
@@ -147,7 +132,7 @@ public class KafkaClientPool {
}
try {
return objectPool.borrowObject(kafkaConsumerBorrowTimeoutUnitMs);
return objectPool.borrowObject(3000);
} catch (Exception e) {
LOGGER.error("borrow kafka consumer client failed, clusterDO:{}.", clusterDO, e);
}

View File

@@ -156,9 +156,6 @@ public class LogicalClusterMetadataManager {
return logicalClusterDO.getClusterId();
}
/**
* 定时刷新逻辑集群元数据到缓存中
*/
@Scheduled(cron="0/30 * * * * ?")
public void flush() {
List<LogicalClusterDO> logicalClusterDOList = logicalClusterService.listAll();
@@ -211,8 +208,7 @@ public class LogicalClusterMetadataManager {
// 计算逻辑集群到Topic名称的映射
Set<String> topicNameSet = PhysicalClusterMetadataManager.getBrokerTopicNum(
logicalClusterDO.getClusterId(),
brokerIdSet
);
brokerIdSet);
LOGICAL_CLUSTER_ID_TOPIC_NAME_MAP.put(logicalClusterDO.getId(), topicNameSet);
// 计算Topic名称到逻辑集群的映射

View File

@@ -50,9 +50,6 @@ 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<>();
@@ -128,7 +125,7 @@ public class PhysicalClusterMetadataManager {
zkConfig.watchChildren(ZkPathUtil.BROKER_IDS_ROOT, brokerListener);
//增加Topic监控
TopicStateListener topicListener = new TopicStateListener(clusterDO.getId(), zkConfig, threadPool);
TopicStateListener topicListener = new TopicStateListener(clusterDO.getId(), zkConfig);
topicListener.init();
zkConfig.watchChildren(ZkPathUtil.BROKER_TOPICS_ROOT, topicListener);
@@ -317,7 +314,7 @@ public class PhysicalClusterMetadataManager {
metadataMap.put(brokerId, brokerMetadata);
Map<Integer, JmxConnectorWrap> jmxMap = JMX_CONNECTOR_MAP.getOrDefault(clusterId, new ConcurrentHashMap<>());
jmxMap.put(brokerId, new JmxConnectorWrap(clusterId, brokerId, brokerMetadata.getHost(), brokerMetadata.getJmxPort(), jmxConfig));
jmxMap.put(brokerId, new JmxConnectorWrap(brokerMetadata.getHost(), brokerMetadata.getJmxPort(), jmxConfig));
JMX_CONNECTOR_MAP.put(clusterId, jmxMap);
Map<Integer, KafkaVersion> versionMap = KAFKA_VERSION_MAP.getOrDefault(clusterId, new ConcurrentHashMap<>());
@@ -542,12 +539,9 @@ public class PhysicalClusterMetadataManager {
}
public static Set<String> getBrokerTopicNum(Long clusterId, Set<Integer> brokerIdSet) {
Map<String, TopicMetadata> metadataMap = TOPIC_METADATA_MAP.get(clusterId);
if (metadataMap == null) {
return new HashSet<>();
}
Set<String> topicNameSet = new HashSet<>();
Map<String, TopicMetadata> metadataMap = TOPIC_METADATA_MAP.get(clusterId);
for (String topicName: metadataMap.keySet()) {
try {
TopicMetadata tm = metadataMap.get(topicName);

View File

@@ -1,63 +1,37 @@
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 javax.annotation.PostConstruct;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
/**
* @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")
);
@Value(value = "${thread-pool.collect-metrics.thread-num:256}")
private Integer collectMetricsThreadNum;
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.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 submitCollectMetricsTask(Runnable collectMetricsTask) {
COLLECT_METRICS_THREAD_POOL.submit(collectMetricsTask);
}
public void submitCollectMetricsTask(Long clusterId, Runnable collectMetricsTask) {
collectMetricsThreadPool.submit(collectMetricsTask);
}
public void submitApiCallTask(Long clusterId, Runnable apiCallTask) {
apiCallThreadPool.submit(apiCallTask);
public static void submitApiCallTask(Runnable apiCallTask) {
API_CALL_THREAD_POOL.submit(apiCallTask);
}
}

View File

@@ -185,8 +185,7 @@ public class GatewayConfigServiceImpl implements GatewayConfigService {
List<GatewayConfigDO> gatewayConfigDOList = gatewayConfigDao.getByConfigType(gatewayConfigDO.getType());
Long version = 1L;
for (GatewayConfigDO elem: gatewayConfigDOList) {
if (elem.getVersion() >= version) {
// 大于等于的情况下,都需要+1
if (elem.getVersion() > version) {
version = elem.getVersion() + 1L;
}
}
@@ -205,7 +204,6 @@ public class GatewayConfigServiceImpl implements GatewayConfigService {
@Override
public Result deleteById(Long id) {
try {
// TODO 删除的时候不能直接删也需要变更一下version
if (gatewayConfigDao.deleteById(id) > 0) {
return Result.buildSuc();
}
@@ -234,8 +232,7 @@ public class GatewayConfigServiceImpl implements GatewayConfigService {
List<GatewayConfigDO> gatewayConfigDOList = gatewayConfigDao.getByConfigType(newGatewayConfigDO.getType());
Long version = 1L;
for (GatewayConfigDO elem: gatewayConfigDOList) {
if (elem.getVersion() >= version) {
// 大于等于的情况下,都需要+1
if (elem.getVersion() > version) {
version = elem.getVersion() + 1L;
}
}

View File

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

View File

@@ -22,7 +22,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.*;
import java.util.regex.Pattern;
/**
* @author zengqiao
@@ -241,11 +240,9 @@ public class ExpertServiceImpl implements ExpertService {
return new ArrayList<>();
}
//获取满足条件的过期Topic
List<TopicExpiredDO> filteredExpiredTopicList = new ArrayList<>();
for (TopicExpiredDO elem: expiredTopicList) {
//判定是否为忽略Cluster或者判定是否为忽略Topic名使用正则来过滤理论上不属于过期的Topic
if (config.getIgnoreClusterIdList().contains(elem.getClusterId()) || Pattern.matches(config.getFilterRegex(), elem.getTopicName())) {
if (config.getIgnoreClusterIdList().contains(elem.getClusterId())) {
continue;
}
filteredExpiredTopicList.add(elem);

View File

@@ -39,9 +39,6 @@ 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) {
@@ -101,7 +98,7 @@ public class JmxServiceImpl implements JmxService {
);
}
});
threadPool.submitCollectMetricsTask(clusterId, taskList[i]);
ThreadPool.submitCollectMetricsTask(taskList[i]);
}
List<TopicMetrics> metricsList = new ArrayList<>();
@@ -306,7 +303,7 @@ public class JmxServiceImpl implements JmxService {
return metricsList;
}
});
threadPool.submitCollectMetricsTask(clusterId, taskList[i]);
ThreadPool.submitCollectMetricsTask(taskList[i]);
}
Map<String, TopicMetrics> metricsMap = new HashMap<>();

View File

@@ -2,8 +2,6 @@ package com.xiaojukeji.kafka.manager.service.service.impl;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.entity.pojo.RegionDO;
import com.xiaojukeji.kafka.manager.common.events.RegionCreatedEvent;
import com.xiaojukeji.kafka.manager.common.utils.SpringTool;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.common.zookeeper.znode.brokers.TopicMetadata;
import com.xiaojukeji.kafka.manager.dao.RegionDao;
@@ -61,8 +59,6 @@ public class RegionServiceImpl implements RegionService {
return ResultStatus.BROKER_NOT_EXIST;
}
if (regionDao.insert(regionDO) > 0) {
// 发布region创建事件
SpringTool.publish(new RegionCreatedEvent(this, regionDO));
return ResultStatus.SUCCESS;
}
} catch (DuplicateKeyException e) {

View File

@@ -87,9 +87,6 @@ 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 {
@@ -343,7 +340,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) {
@@ -544,7 +541,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<>();

View File

@@ -45,9 +45,6 @@ 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);
@@ -128,7 +125,7 @@ public class DidiHealthScoreStrategy extends AbstractHealthScoreStrategy {
return calBrokerHealthScore(clusterId, brokerId);
}
});
threadPool.submitApiCallTask(clusterId, taskList[i]);
ThreadPool.submitApiCallTask(taskList[i]);
}
Integer topicHealthScore = HEALTH_SCORE_HEALTHY;

View File

@@ -1,6 +1,5 @@
package com.xiaojukeji.kafka.manager.service.utils;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
@@ -9,18 +8,38 @@ import org.springframework.stereotype.Service;
* @author zengqiao
* @date 20/4/26
*/
@Data
@Service("configUtils")
public class ConfigUtils {
private ConfigUtils() {
}
@Value(value = "${custom.idc:cn}")
@Value(value = "${custom.idc}")
private String idc;
@Value(value = "${spring.profiles.active:dev}")
@Value(value = "${spring.profiles.active}")
private String kafkaManagerEnv;
@Value(value = "${spring.application.version:unknown}")
private String applicationVersion;
@Value(value = "${custom.store-metrics-task.save-days}")
private Long maxMetricsSaveDays;
public String getIdc() {
return idc;
}
public void setIdc(String idc) {
this.idc = idc;
}
public String getKafkaManagerEnv() {
return kafkaManagerEnv;
}
public void setKafkaManagerEnv(String kafkaManagerEnv) {
this.kafkaManagerEnv = kafkaManagerEnv;
}
public Long getMaxMetricsSaveDays() {
return maxMetricsSaveDays;
}
public void setMaxMetricsSaveDays(Long maxMetricsSaveDays) {
this.maxMetricsSaveDays = maxMetricsSaveDays;
}
}

View File

@@ -74,10 +74,15 @@ 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(":");
// 解析并更新本次存储的broker元信息
BrokerMetadata.parseAndUpdateBrokerMetadata(brokerMetadata);
brokerMetadata.setHost(endpoint.substring(0, idx));
brokerMetadata.setPort(Integer.parseInt(endpoint.substring(idx + 1)));
}
brokerMetadata.setClusterId(clusterId);
brokerMetadata.setBrokerId(brokerId);
PhysicalClusterMetadataManager.putBrokerMetadata(clusterId, brokerId, brokerMetadata, jmxConfig);

View File

@@ -19,13 +19,13 @@ import org.springframework.dao.DuplicateKeyException;
* @date 20/5/14
*/
public class ControllerStateListener implements StateChangeListener {
private static final Logger LOGGER = LoggerFactory.getLogger(ControllerStateListener.class);
private final static Logger LOGGER = LoggerFactory.getLogger(ControllerStateListener.class);
private final Long clusterId;
private Long clusterId;
private final ZkConfigImpl zkConfig;
private ZkConfigImpl zkConfig;
private final ControllerDao controllerDao;
private ControllerDao controllerDao;
public ControllerStateListener(Long clusterId, ZkConfigImpl zkConfig, ControllerDao controllerDao) {
this.clusterId = clusterId;
@@ -35,11 +35,8 @@ public class ControllerStateListener implements StateChangeListener {
@Override
public void init() {
if (!checkNodeExist()) {
LOGGER.warn("kafka-controller data not exist, clusterId:{}.", clusterId);
return;
}
processControllerChange();
return;
}
@Override
@@ -52,21 +49,12 @@ public class ControllerStateListener implements StateChangeListener {
break;
}
} catch (Exception e) {
LOGGER.error("process controller state change failed, clusterId:{} state:{} path:{}.", clusterId, state, path, e);
LOGGER.error("process controller state change failed, clusterId:{} state:{} path:{}.",
clusterId, state, path, e);
}
}
private boolean checkNodeExist() {
try {
return zkConfig.checkPathExists(ZkPathUtil.CONTROLLER_ROOT_NODE);
} catch (Exception e) {
LOGGER.error("init kafka-controller data failed, clusterId:{}.", clusterId, e);
}
return false;
}
private void processControllerChange() {
private void processControllerChange(){
LOGGER.warn("init controllerData or controller change, clusterId:{}.", clusterId);
ControllerData controllerData = null;
try {

View File

@@ -10,7 +10,6 @@ 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;
@@ -29,12 +28,9 @@ public class TopicStateListener implements StateChangeListener {
private ZkConfigImpl zkConfig;
private ThreadPool threadPool;
public TopicStateListener(Long clusterId, ZkConfigImpl zkConfig, ThreadPool threadPool) {
public TopicStateListener(Long clusterId, ZkConfigImpl zkConfig) {
this.clusterId = clusterId;
this.zkConfig = zkConfig;
this.threadPool = threadPool;
}
@Override
@@ -51,7 +47,7 @@ public class TopicStateListener implements StateChangeListener {
return null;
}
});
threadPool.submitCollectMetricsTask(clusterId, taskList[i]);
ThreadPool.submitCollectMetricsTask(taskList[i]);
}
} catch (Exception e) {
LOGGER.error("init topics metadata failed, clusterId:{}.", clusterId, e);

View File

@@ -20,5 +20,5 @@ public interface BrokerMetricsDao {
*/
List<BrokerMetricsDO> getBrokerMetrics(Long clusterId, Integer brokerId, Date startTime, Date endTime);
int deleteBeforeTime(Date endTime, Integer limitSize);
int deleteBeforeTime(Date endTime);
}

View File

@@ -10,5 +10,5 @@ public interface ClusterMetricsDao {
List<ClusterMetricsDO> getClusterMetrics(long clusterId, Date startTime, Date endTime);
int deleteBeforeTime(Date endTime, Integer limitSize);
int deleteBeforeTime(Date endTime);
}

View File

@@ -30,5 +30,5 @@ public interface TopicAppMetricsDao {
* @param endTime
* @return
*/
int deleteBeforeTime(Date endTime, Integer limitSize);
int deleteBeforeTime(Date endTime);
}

View File

@@ -22,5 +22,5 @@ public interface TopicMetricsDao {
List<TopicMetricsDO> getLatestTopicMetrics(Long clusterId, Date afterTime);
int deleteBeforeTime(Date endTime, Integer limitSize);
int deleteBeforeTime(Date endTime);
}

View File

@@ -33,7 +33,9 @@ public interface TopicRequestMetricsDao {
* @param endTime
* @return
*/
int deleteBeforeTime(Date endTime, Integer limitSize);
int deleteBeforeTime(Date endTime);
int deleteBeforeId(Long id);
List<TopicMetricsDO> getById(Long startId, Long endId);
}

View File

@@ -32,5 +32,5 @@ public interface TopicThrottledMetricsDao {
List<TopicThrottledMetricsDO> getLatestTopicThrottledMetrics(Long clusterId, Date afterTime);
int deleteBeforeTime(Date endTime, Integer limitSize);
int deleteBeforeTime(Date endTime);
}

View File

@@ -37,10 +37,7 @@ public class BrokerMetricsImpl implements BrokerMetricsDao {
}
@Override
public int deleteBeforeTime(Date endTime, Integer limitSize) {
Map<String, Object> params = new HashMap<>(2);
params.put("endTime", endTime);
params.put("limitSize", limitSize);
return sqlSession.delete("BrokerMetricsDao.deleteBeforeTime", params);
public int deleteBeforeTime(Date endTime) {
return sqlSession.delete("BrokerMetricsDao.deleteBeforeTime", endTime);
}
}

View File

@@ -27,7 +27,7 @@ public class ClusterMetricsDaoImpl implements ClusterMetricsDao {
@Override
public List<ClusterMetricsDO> getClusterMetrics(long clusterId, Date startTime, Date endTime) {
Map<String, Object> map = new HashMap<>(3);
Map<String, Object> map = new HashMap<String, Object>(3);
map.put("clusterId", clusterId);
map.put("startTime", startTime);
map.put("endTime", endTime);
@@ -35,10 +35,7 @@ public class ClusterMetricsDaoImpl implements ClusterMetricsDao {
}
@Override
public int deleteBeforeTime(Date endTime, Integer limitSize) {
Map<String, Object> params = new HashMap<>(2);
params.put("endTime", endTime);
params.put("limitSize", limitSize);
return sqlSession.delete("ClusterMetricsDao.deleteBeforeTime", params);
public int deleteBeforeTime(Date endTime) {
return sqlSession.delete("ClusterMetricsDao.deleteBeforeTime", endTime);
}
}

View File

@@ -46,10 +46,7 @@ public class TopicAppMetricsDaoImpl implements TopicAppMetricsDao {
}
@Override
public int deleteBeforeTime(Date endTime, Integer limitSize) {
Map<String, Object> params = new HashMap<>(2);
params.put("endTime", endTime);
params.put("limitSize", limitSize);
return sqlSession.delete("TopicAppMetricsDao.deleteBeforeTime", params);
public int deleteBeforeTime(Date endTime) {
return sqlSession.delete("TopicAppMetricsDao.deleteBeforeTime", endTime);
}
}

View File

@@ -60,10 +60,7 @@ public class TopicMetricsDaoImpl implements TopicMetricsDao {
}
@Override
public int deleteBeforeTime(Date endTime, Integer limitSize) {
Map<String, Object> params = new HashMap<>(2);
params.put("endTime", endTime);
params.put("limitSize", limitSize);
return sqlSession.delete("TopicMetricsDao.deleteBeforeTime", params);
public int deleteBeforeTime(Date endTime) {
return sqlSession.delete("TopicMetricsDao.deleteBeforeTime", endTime);
}
}

View File

@@ -45,11 +45,13 @@ public class TopicRequestMetricsDaoImpl implements TopicRequestMetricsDao {
}
@Override
public int deleteBeforeTime(Date endTime, Integer limitSize) {
Map<String, Object> params = new HashMap<>();
params.put("endTime", endTime);
params.put("limitSize", limitSize);
return sqlSession.delete("TopicRequestMetricsDao.deleteBeforeTime", params);
public int deleteBeforeTime(Date endTime) {
return sqlSession.delete("TopicRequestMetricsDao.deleteBeforeTime", endTime);
}
@Override
public int deleteBeforeId(Long id) {
return sqlSession.delete("TopicRequestMetricsDao.deleteBeforeId", id);
}
@Override

View File

@@ -75,10 +75,7 @@ public class TopicThrottledMetricsDaoImpl implements TopicThrottledMetricsDao {
}
@Override
public int deleteBeforeTime(Date endTime, Integer limitSize) {
Map<String, Object> params = new HashMap<>(2);
params.put("endTime", endTime);
params.put("limitSize", limitSize);
return sqlSession.delete("TopicThrottledMetricsDao.deleteBeforeTime", params);
public int deleteBeforeTime(Date endTime) {
return sqlSession.delete("TopicThrottledMetricsDao.deleteBeforeTime", endTime);
}
}

View File

@@ -8,9 +8,6 @@
<result property="username" column="username" />
<result property="password" column="password" />
<result property="role" column="role" />
<result property="displayName" column="display_name" />
<result property="department" column="department" />
<result property="mail" column="mail" />
<result property="gmtCreate" column="gmt_create" />
<result property="gmtModify" column="gmt_modify" />
</resultMap>
@@ -18,9 +15,9 @@
<insert id="insert" parameterType="com.xiaojukeji.kafka.manager.common.entity.pojo.AccountDO">
<![CDATA[
REPLACE account
(username, password, role, display_name, department, mail)
(username, password, role)
VALUES
(#{username}, #{password}, #{role}, #{displayName}, #{department}, #{mail})
(#{username}, #{password}, #{role})
]]>
</insert>

View File

@@ -29,9 +29,9 @@
]]>
</select>
<delete id="deleteBeforeTime" parameterType="java.util.Map">
<delete id="deleteBeforeTime" parameterType="java.util.Date">
<![CDATA[
DELETE FROM broker_metrics WHERE gmt_create < #{endTime} LIMIT #{limitSize}
DELETE FROM broker_metrics WHERE gmt_create < #{endTime} LIMIT 1000
]]>
</delete>
</mapper>

View File

@@ -27,9 +27,9 @@
</foreach>
</insert>
<delete id="deleteBeforeTime" parameterType="java.util.Map">
<delete id="deleteBeforeTime" parameterType="java.util.Date">
<![CDATA[
DELETE FROM cluster_metrics WHERE gmt_create < #{endTime} LIMIT #{limitSize}
DELETE FROM cluster_metrics WHERE gmt_create < #{endTime} LIMIT 200
]]>
</delete>
</mapper>

View File

@@ -30,9 +30,9 @@
]]>
</select>
<delete id="deleteBeforeTime" parameterType="java.util.Map">
<delete id="deleteBeforeTime" parameterType="java.util.Date">
<![CDATA[
DELETE FROM topic_app_metrics WHERE gmt_create < #{endTime} LIMIT #{limitSize}
DELETE FROM topic_app_metrics WHERE gmt_create < #{endTime} LIMIT 3000
]]>
</delete>
</mapper>

View File

@@ -37,9 +37,9 @@
]]>
</select>
<delete id="deleteBeforeTime" parameterType="java.util.Map">
<delete id="deleteBeforeTime" parameterType="java.util.Date">
<![CDATA[
DELETE FROM topic_metrics WHERE gmt_create < #{endTime} LIMIT #{limitSize}
DELETE FROM topic_metrics WHERE gmt_create < #{endTime} LIMIT 3000
]]>
</delete>
</mapper>

View File

@@ -34,9 +34,15 @@
ORDER BY gmt_create ASC
</select>
<delete id="deleteBeforeTime" parameterType="java.util.Map">
<delete id="deleteBeforeTime" parameterType="java.util.Date">
<![CDATA[
DELETE FROM topic_request_time_metrics WHERE gmt_create < #{endTime} LIMIT #{limitSize}
DELETE FROM topic_request_time_metrics WHERE gmt_create < #{endTime} LIMIT 2000
]]>
</delete>
<delete id="deleteBeforeId" parameterType="java.lang.Long">
<![CDATA[
DELETE FROM topic_request_time_metrics WHERE id < #{id} LIMIT 20000
]]>
</delete>

View File

@@ -54,9 +54,9 @@
AND gmt_create > #{afterTime}
</select>
<delete id="deleteBeforeTime" parameterType="java.util.Map">
<delete id="deleteBeforeTime" parameterType="java.util.Date">
<![CDATA[
DELETE FROM topic_throttled_metrics WHERE gmt_create < #{endTime} LIMIT #{limitSize}
DELETE FROM topic_throttled_metrics WHERE gmt_create < #{endTime} LIMIT 3000
]]>
</delete>
</mapper>

View File

@@ -54,8 +54,7 @@ public class BaseEnterpriseStaffService extends AbstractEnterpriseStaffService {
}
List<EnterpriseStaff> staffList = new ArrayList<>();
for (AccountDO accountDO: doList) {
//这里对chineseName填充共识的displayNameDepartment则获取Department信息
staffList.add(new EnterpriseStaff(accountDO.getUsername(), accountDO.getDisplayName(), accountDO.getDepartment()));
staffList.add(new EnterpriseStaff(accountDO.getUsername(), accountDO.getUsername(), ""));
}
return staffList;
} catch (Exception e) {

View File

@@ -1,6 +1,5 @@
package com.xiaojukeji.kafka.manager.account.component.ldap;
import com.xiaojukeji.kafka.manager.common.utils.SplitUtils;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -15,9 +14,7 @@ import javax.naming.directory.SearchControls;
import javax.naming.directory.SearchResult;
import javax.naming.ldap.InitialLdapContext;
import javax.naming.ldap.LdapContext;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
@Component
public class LdapAuthentication {
@@ -63,11 +60,8 @@ public class LdapAuthentication {
return null;
}
private Map<String, Object> getLdapAttrsInfo(String account, LdapContext ctx) {
//存储更多的LDAP元信息
Map<String, Object> ldapAttrsInfo = new HashMap<>();
private String getUserDN(String account, LdapContext ctx) {
String userDN = "";
ldapAttrsInfo.clear();
try {
SearchControls constraints = new SearchControls();
constraints.setSearchScope(SearchControls.SUBTREE_SCOPE);
@@ -75,7 +69,7 @@ public class LdapAuthentication {
NamingEnumeration<SearchResult> en = ctx.search("", filter, constraints);
if (en == null || !en.hasMoreElements()) {
return null;
return "";
}
// maybe more than one element
while (en.hasMoreElements()) {
@@ -84,25 +78,13 @@ public class LdapAuthentication {
SearchResult si = (SearchResult) obj;
userDN += si.getName();
userDN += "," + ldapBasedn;
//携带LDAP更多元信息以填充用户元信息
ldapAttrsInfo.put("userDN", userDN);
ldapAttrsInfo.put("sAMAccountName",
SplitUtils.keyValueSplit(si.getAttributes().get("samaccountname").toString()));
ldapAttrsInfo.put("department",
SplitUtils.keyValueSplit(si.getAttributes().get("department").toString()));
ldapAttrsInfo.put("company",
SplitUtils.keyValueSplit(si.getAttributes().get("company").toString()));
ldapAttrsInfo.put("displayName",
SplitUtils.keyValueSplit(si.getAttributes().get("displayname").toString()));
ldapAttrsInfo.put("mail",
SplitUtils.keyValueSplit(si.getAttributes().get("mail").toString()));
break;
}
}
} catch (Exception e) {
LOGGER.error("class=LdapAuthentication||method=getUserDN||account={}||errMsg={}", account, e);
}
return ldapAttrsInfo;
return userDN;
}
/**
@@ -111,23 +93,23 @@ public class LdapAuthentication {
* @param password
* @return
*/
public Map<String, Object> authenticate(String account, String password) {
public boolean authenticate(String account, String password) {
LdapContext ctx = getLdapContext();
if (ValidateUtils.isNull(ctx)) {
return null;
return false;
}
try {
Map<String, Object> ldapAttrsInfo = getLdapAttrsInfo(account, ctx);
if(ValidateUtils.isNull(ldapAttrsInfo)){
return null;
String userDN = getUserDN(account, ctx);
if(ValidateUtils.isBlank(userDN)){
return false;
}
ctx.addToEnvironment(Context.SECURITY_PRINCIPAL, ldapAttrsInfo.get("userDN").toString());
ctx.addToEnvironment(Context.SECURITY_PRINCIPAL, userDN);
ctx.addToEnvironment(Context.SECURITY_CREDENTIALS, password);
ctx.reconnect(null);
return ldapAttrsInfo;
return true;
} catch (AuthenticationException e) {
LOGGER.warn("class=LdapAuthentication||method=authenticate||account={}||errMsg={}", account, e);
} catch (NamingException e) {
@@ -143,6 +125,6 @@ public class LdapAuthentication {
}
}
}
return null;
return false;
}
}

View File

@@ -17,7 +17,6 @@ import org.springframework.stereotype.Service;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.Map;
/**
* @author zengqiao
@@ -32,53 +31,43 @@ public class BaseSessionSignOn extends AbstractSingleSignOn {
private LdapAuthentication ldapAuthentication;
//是否开启ldap验证
@Value(value = "${account.ldap.enabled:false}")
@Value(value = "${account.ldap.enabled:}")
private Boolean accountLdapEnabled;
//ldap自动注册的默认角色。请注意它通常来说都是低权限角色
@Value(value = "${account.ldap.auth-user-registration-role:normal}")
@Value(value = "${account.ldap.auth-user-registration-role:}")
private String authUserRegistrationRole;
//ldap自动注册是否开启
@Value(value = "${account.ldap.auth-user-registration:false}")
private Boolean authUserRegistration;
@Value(value = "${account.ldap.auth-user-registration:}")
private boolean authUserRegistration;
@Override
public Result<String> loginAndGetLdap(HttpServletRequest request, HttpServletResponse response, LoginDTO dto) {
if (ValidateUtils.isBlank(dto.getUsername()) || ValidateUtils.isNull(dto.getPassword())) {
return Result.buildFailure("Missing parameters");
}
//先创建空对象看是在LDAP去做填充还是直接查表填充
Result<AccountDO> accountResult;
Result<AccountDO> accountResult = accountService.getAccountDO(dto.getUsername());
//判断是否激活了LDAP验证, 若激活则也可使用ldap进行认证
if(!ValidateUtils.isNull(accountLdapEnabled) && accountLdapEnabled){
//去LDAP验证账密
Map<String, Object> ldapAttrsInfo = ldapAuthentication.authenticate(dto.getUsername(),dto.getPassword());;
if(ValidateUtils.isNull(ldapAttrsInfo)){
if(!ldapAuthentication.authenticate(dto.getUsername(),dto.getPassword())){
return Result.buildFrom(ResultStatus.LDAP_AUTHENTICATION_FAILED);
}
//LDAP验证通过拿LDAP的sAMAccountName替换dto对象的值便于第一次自动注册采用LDAP值并且第二次也避免REPLACE
dto.setUsername(ldapAttrsInfo.get("sAMAccountName").toString());
accountResult = accountService.getAccountDO(dto.getUsername());
if((ValidateUtils.isNull(accountResult) || ValidateUtils.isNull(accountResult.getData())) && authUserRegistration){
//自动注册
AccountDO accountDO = new AccountDO();
accountDO.setUsername(dto.getUsername());
accountDO.setRole(AccountRoleEnum.getUserRoleEnum(authUserRegistrationRole).getRole());
accountDO.setPassword(dto.getPassword());
accountDO.setDisplayName(ldapAttrsInfo.getOrDefault("displayName", "").toString());
accountDO.setDepartment(ldapAttrsInfo.getOrDefault("department", "").toString());
accountDO.setMail(ldapAttrsInfo.getOrDefault("mail", "").toString());
accountService.createAccount(accountDO);
}
return Result.buildSuc(dto.getUsername());
}
//不走LDAP认证直接查表填充
accountResult = accountService.getAccountDO(dto.getUsername());
if (ValidateUtils.isNull(accountResult) || accountResult.failed()) {
return new Result<>(accountResult.getCode(), accountResult.getMessage());

View File

@@ -275,9 +275,6 @@ public class AccountServiceImpl implements AccountService {
return enterpriseStaffService.searchEnterpriseStaffByKeyWord(prefix);
}
/**
* 定时刷新account信息到缓存中
*/
@Scheduled(cron ="0/5 * * * * ?")
public void flush() {
try {

View File

@@ -14,7 +14,6 @@ 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;
@@ -28,13 +27,7 @@ import javax.servlet.http.HttpSession;
*/
@Service("loginService")
public class LoginServiceImpl implements LoginService {
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;
private final static Logger LOGGER = LoggerFactory.getLogger(LoginServiceImpl.class);
@Autowired
private AccountService accountService;
@@ -82,10 +75,8 @@ public class LoginServiceImpl implements LoginService {
return false;
}
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 允许跳过且是跳过类型的接口,则直接跳过登录
if (classRequestMappingValue.equals(ApiPrefix.API_V1_SSO_PREFIX)) {
// 白名单接口直接true
return true;
}

View File

@@ -28,6 +28,7 @@
<java_target_version>1.8</java_target_version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<file_encoding>UTF-8</file_encoding>
<spring-version>5.1.3.RELEASE</spring-version>
</properties>
<dependencies>
@@ -55,14 +56,17 @@
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-beans</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>

View File

@@ -37,24 +37,21 @@ 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:12345678}")
@Value("${kcm.n9e.user-token}")
private String userToken;
@Value("${kcm.n9e.account:root}")
@Value("${kcm.n9e.account}")
private String account;
@Value("${kcm.n9e.timeout:300}")
@Value("${kcm.n9e.timeout}")
private Integer timeout;
@Value("${kcm.n9e.script-file:kcm_script.sh}")
@Value("${kcm.n9e.script-file}")
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";
@@ -222,8 +219,7 @@ 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()).append(",,");
sb.append(this.logiKMUrl);
sb.append(creationTaskData.getServerPropertiesUrl());
N9eCreationTask n9eCreationTask = new N9eCreationTask();
n9eCreationTask.setTitle(Constant.TASK_TITLE_PREFIX + "-集群ID:" + creationTaskData.getClusterId());

View File

@@ -18,13 +18,12 @@ 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}
@@ -48,7 +47,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}" -o -z "${p_kafka_manager_url}" ]; 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}" ]; then
ECHO_LOG "存在为空的参数不合法, 退出集群任务"
dchat_alarm "存在为空的参数不合法, 退出集群任务"
exit 1
@@ -73,11 +72,11 @@ function check_and_init_env() {
# 检查并等待集群所有的副本处于同步的状态
function check_and_wait_broker_stabled() {
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`
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`
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} ${p_kafka_manager_url}/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} ${g_kafka_manager_addr}/api/v1/third-part/${p_cluster_id}/broker-stabled | python -m json.tool | grep true |wc -l`
done
ECHO_LOG "集群副本都已经处于同步的状态, 可以进行集群升级"
}
@@ -325,7 +324,6 @@ 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}"
@@ -344,7 +342,7 @@ fi
ECHO_LOG "停kafka服务"
stop_kafka_server
ECHO_LOG "停5秒, 确保端口已释放"
ECHO_LOG "停5秒, 确保"
sleep 5
if [ "${p_cluster_task_type}" == "0" ];then

View File

@@ -25,6 +25,7 @@
<java_target_version>1.8</java_target_version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<file_encoding>UTF-8</file_encoding>
<spring-version>5.1.3.RELEASE</spring-version>
</properties>
<dependencies>
@@ -62,10 +63,12 @@
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-beans</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${spring-version}</version>
</dependency>
</dependencies>
</project>

View File

@@ -27,19 +27,19 @@ import java.util.concurrent.TimeUnit;
public class N9eService extends AbstractMonitorService {
private static final Logger LOGGER = LoggerFactory.getLogger(N9eService.class);
@Value("${monitor.n9e.nid:}")
@Value("${monitor.n9e.nid}")
private Integer monitorN9eNid;
@Value("${monitor.n9e.user-token:}")
@Value("${monitor.n9e.user-token}")
private String monitorN9eUserToken;
@Value("${monitor.n9e.mon.base-url:}")
@Value("${monitor.n9e.mon.base-url}")
private String monitorN9eMonBaseUrl;
@Value("${monitor.n9e.sink.base-url:}")
@Value("${monitor.n9e.sink.base-url}")
private String monitorN9eSinkBaseUrl;
@Value("${monitor.n9e.rdb.base-url:}")
@Value("${monitor.n9e.rdb.base-url}")
private String monitorN9eRdbBaseUrl;
private static final Cache<String, NotifyGroup> NOTIFY_GROUP_CACHE = Caffeine.newBuilder()

View File

@@ -25,6 +25,7 @@
<java_target_version>1.8</java_target_version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<file_encoding>UTF-8</file_encoding>
<spring-version>5.1.3.RELEASE</spring-version>
</properties>
<dependencies>
@@ -47,6 +48,7 @@
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${spring-version}</version>
</dependency>
</dependencies>
</project>

View File

@@ -19,7 +19,7 @@ public class OrderPassedNotifyService implements ApplicationListener<OrderPassed
@Autowired
private AbstractNotifyService notifyService;
@Value("${notify.order.detail-url:}")
@Value("${notify.order.detail-url}")
private String orderDetailUrl;
@Async

View File

@@ -19,7 +19,7 @@ public class OrderRefusedNotifyService implements ApplicationListener<OrderRefus
@Autowired
private AbstractNotifyService notifyService;
@Value("${notify.order.detail-url:}")
@Value("${notify.order.detail-url}")
private String orderDetailUrl;
@Async

View File

@@ -24,6 +24,7 @@
<java_target_version>1.8</java_target_version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<file_encoding>UTF-8</file_encoding>
<spring-version>5.1.3.RELEASE</spring-version>
</properties>
<description>
@@ -45,6 +46,7 @@
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${spring-version}</version>
</dependency>
</dependencies>
</project>

View File

@@ -42,9 +42,6 @@ public class ThirdPartServiceImpl implements ThirdPartService {
@Autowired
private ConsumerService consumerService;
@Autowired
private KafkaClientPool kafkaClientPool;
@Override
public Result<ConsumeHealthEnum> checkConsumeHealth(Long clusterId,
String topicName,
@@ -112,7 +109,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;
}

View File

@@ -24,6 +24,7 @@
<java_target_version>1.8</java_target_version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<file_encoding>UTF-8</file_encoding>
<spring-version>5.1.3.RELEASE</spring-version>
</properties>
<dependencies>
@@ -51,6 +52,7 @@
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${spring-version}</version>
</dependency>
</dependencies>
</project>

Some files were not shown because too many files have changed in this diff Show More