kafka监控kafka manager（CMAK）部署设置

一、准备工作

1.1、服务器信息梳理

脚色IP操纵系统安装服务监控机10.45.19.20Linux CentOS 7.9CMAK3.0.0.5、ZooKeeper3.9.0、JDK11、JDK1.8被监控机 Kafka broker.id 050.50.50.101Linux CentOS 7.9Kafka、ZooKeeper（任意版本）被监控机 Kafka broker.id 150.50.50.101Linux CentOS 7.9Kafka、ZooKeeper（任意版本）被监控机 Kafka broker.id 250.50.50.101Linux CentOS 7.9Kafka、ZooKeeper（任意版本） 1.2、准备以下安装包

1. https://www.oracle.com/java/technologies/downloads/archive

复制代码

1. https://archive.apache.org/dist/zookeeper

复制代码

1. https://github.com/yahoo/CMAK

复制代码

1. jdk-11.0.26_linux-x64_bin.tar.gz
3. jdk-8u161-linux-x64.tar.gz
4. apache-zookeeper-3.9.0-bin.tar.gz
5. cmak-3.0.0.5.zip

复制代码

1.3、创建安装目录

1. mkdir /usr/local/cmak

复制代码

二、部署 JDK11 环境

1. jdk-11.0.26_linux-x64_bin.tar.gz

复制代码

1. tar zxf jdk-11.0.26_linux-x64_bin.tar.gz
2. -C /usr/local/cmak

复制代码

三、部署 ZooKeeper 存储元数据

3.1、部署 JDK1.8 环境

1. tar zxf jdk-8u161-linux-x64.tar.gz -C /usr/local/cmak

复制代码

1. vim /etc/etc/profile

复制代码

1. export PATH=$PATH:/usr/local/cmak/jdk1.8.0_161/bin

复制代码

3.2、部署 ZooKeeper

3.2.1、解压至安装目录

1. wget https://archive.apache.org/dist/zookeeper
2. /zookeeper-3.9.0/apache-zookeeper-3.9.0-bin.tar.gz

复制代码

1. tar zxf apache-zookeeper-3.9.0-bin.tar.gz -C /usr/local/cmak

复制代码

1. mv /usr/local/cmak/apache-zookeeper-3.9.0-bin /usr/local/cmak/zookeeper-3.9.0

复制代码

3.2.2、创建存储目录

1. mkdir /usr/local/cmak
2. /zookeeper-3.9.0/{data,logs}

复制代码

3.2.3、编辑设置文件

1. cp /usr/local/cmak/zookeeper-3.9.0/conf/zoo_sample.cfg /usr/local/cmak/zookeeper-3.9.0/conf/zoo.cfg

复制代码

1. vim /usr/local/cmak/zookeeper-3.9.0/conf/zoo.cfg

复制代码

1. dataDir=/usr/local/cmak/zookeeper-3.9.0/data
2. dataLogDir=/usr/local/cmak/zookeeper-3.9.0/logs
3. clientPort=3281
4. admin.serverPort=3282
5. tickTime=2000
6. initLimit=10
7. syncLimit=5

复制代码

3.2.4、启动

1. sh /usr/local/cmak/zookeeper-3.9.0/bin/zkServer.sh start

复制代码

四、部署 CMAK 监控

4.1、解压至安装目录

1. https://github.com/yahoo/CMAK
2. /releases/download/3.0.0.5/cmak-3.0.0.5.zip

复制代码

1. unzip cmak-3.0.0.6.zip -d /usr/local/cmak

复制代码

4.2、编写设置文件

1. vim /usr/local/cmak/cmak-3.0.0.6/conf/application.conf

复制代码

1. # -------------------------------------------------------------------
2. # Application 基础配置
3. # -------------------------------------------------------------------
5. # 加密会话和 Cookie 的密钥（生产环境建议使用环境变量覆盖）
6. # 默认硬编码密钥（注意：硬编码密钥不安全，建议通过环境变量设置）
7. play.crypto.secret="^<csmm5Fx4d=r2HEX8pelM3iBkFVv?k[mc;IZE<_Qoq8EkX_/7@Zt6dP05Pzea3U"
8. # 允许通过环境变量 APPLICATION_SECRET 覆盖密钥
9. play.crypto.secret=${?APPLICATION_SECRET}
11. # 会话过期时间（1小时）
12. play.http.session.maxAge="1h"
14. # 国际化支持的语言列表（当前仅英语）
15. play.i18n.langs=["en"]
17. # HTTP 请求处理器和上下文路径
18. play.http.requestHandler = "play.http.DefaultHttpRequestHandler"
19. play.http.context = "/"
21. # 应用加载器类（自定义 Kafka Manager 加载逻辑）
22. play.application.loader=loader.KafkaManagerLoader
24. # -------------------------------------------------------------------
25. # ZooKeeper 配置
26. # -------------------------------------------------------------------
28. # ZooKeeper 连接地址（默认值）
29. cmak.zkhosts="10.45.19.20:3281"
30. # 允许通过环境变量 ZK_HOSTS 覆盖 ZooKeeper 地址
31. cmak.zkhosts=${?ZK_HOSTS}
33. # -------------------------------------------------------------------
34. # 线程调度配置
35. # -------------------------------------------------------------------
37. # 固定调度器配置（确保每个 Actor 独占线程池）
38. pinned-dispatcher.type="PinnedDispatcher"
39. pinned-dispatcher.executor="thread-pool-executor"
41. # -------------------------------------------------------------------
42. # 应用功能特性开关
43. # -------------------------------------------------------------------
44. # 启用的 Kafka 管理功能列表：
45. # - 集群管理
46. # - Topic 管理
47. # - 首选副本选举
48. # - 分区重分配
49. # - 计划 Leader 选举
50. application.features=[
51.   "KMClusterManagerFeature",
52. #  "KMTopicManagerFeature",
53. #  "KMPreferredReplicaElectionFeature",
54. #  "KMReassignPartitionsFeature",
55. #  "KMScheduleLeaderElectionFeature"
56. ]
58. # -------------------------------------------------------------------
59. # Akka 配置
60. # -------------------------------------------------------------------
61. akka {
62.   # 使用 SLF4J 作为日志记录器
63.   loggers = ["akka.event.slf4j.Slf4jLogger"]
64.   # 日志级别（INFO | DEBUG | WARN | ERROR）
65. #  loglevel = "INFO"
66.   loglevel = "WARN"
67. }
69. # Akka 日志系统启动超时时间
70. akka.logger-startup-timeout = 60s
72. # -------------------------------------------------------------------
73. # 基础认证配置
74. # -------------------------------------------------------------------
76. # 是否启用基础认证（默认禁用）
77. basicAuthentication.enabled=true
78. # 允许通过环境变量 KAFKA_MANAGER_AUTH_ENABLED 覆盖
79. basicAuthentication.enabled=${?KAFKA_MANAGER_AUTH_ENABLED}
81. # LDAP 认证配置 -----------------------------------------------------
82. # 是否启用 LDAP 认证（默认禁用）
83. basicAuthentication.ldap.enabled=false
84. basicAuthentication.ldap.enabled=${?KAFKA_MANAGER_LDAP_ENABLED}
86. # LDAP 服务器地址（示例：ldap://hostname）
87. basicAuthentication.ldap.server=""
88. basicAuthentication.ldap.server=${?KAFKA_MANAGER_LDAP_SERVER}
90. # LDAP 端口（默认 389，SSL 通常为 636）
91. basicAuthentication.ldap.port=389
92. basicAuthentication.ldap.port=${?KAFKA_MANAGER_LDAP_PORT}
94. # 连接 LDAP 的用户名和密码
95. basicAuthentication.ldap.username=""
96. basicAuthentication.ldap.username=${?KAFKA_MANAGER_LDAP_USERNAME}
97. basicAuthentication.ldap.password=""
98. basicAuthentication.ldap.password=${?KAFKA_MANAGER_LDAP_PASSWORD}
100. # LDAP 搜索配置
101. basicAuthentication.ldap.search-base-dn=""
102. basicAuthentication.ldap.search-base-dn=${?KAFKA_MANAGER_LDAP_SEARCH_BASE_DN}
103. basicAuthentication.ldap.search-filter="(uid=$capturedLogin$)"
104. basicAuthentication.ldap.search-filter=${?KAFKA_MANAGER_LDAP_SEARCH_FILTER}
106. # LDAP 组过滤配置
107. basicAuthentication.ldap.group-filter=""
108. basicAuthentication.ldap.group-filter=${?KAFKA_MANAGER_LDAP_GROUP_FILTER}
110. # 连接池配置
111. basicAuthentication.ldap.connection-pool-size=10
112. basicAuthentication.ldap.connection-pool-size=${?KAFKA_MANAGER_LDAP_CONNECTION_POOL_SIZE}
114. # SSL 配置
115. basicAuthentication.ldap.ssl=false
116. basicAuthentication.ldap.ssl=${?KAFKA_MANAGER_LDAP_SSL}
117. basicAuthentication.ldap.ssl-trust-all=false
118. basicAuthentication.ldap.ssl-trust-all=${?KAFKA_MANAGER_LDAP_SSL_TRUST_ALL}
120. # 基础认证凭据（仅在启用基础认证且未启用 LDAP 时生效）
121. basicAuthentication.username="admin"
122. basicAuthentication.username=${?KAFKA_MANAGER_USERNAME}
123. basicAuthentication.password="Quk@EL!4xMyj3hnD"
124. basicAuthentication.password=${?KAFKA_MANAGER_PASSWORD}
126. # 认证领域名称和免认证路径（/api/health 用于健康检查）
127. basicAuthentication.realm="Kafka-Manager"
128. basicAuthentication.excluded=["/api/health"]
130. # -------------------------------------------------------------------
131. # Kafka 消费者配置
132. # -------------------------------------------------------------------
134. # 外部消费者配置文件路径（可选）
135. kafka-manager.consumer.properties.file=${?CONSUMER_PROPERTIES_FILE}

复制代码

1. vim /usr/local/cmak/cmak-3.0.0.6/conf/logback.xml

复制代码

1. <!-- 修改后的配置（重点关注 encoder 部分的 regex 过滤） -->
2. <configuration>
4.     <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
5.         <file>${application.home}/logs/application.log</file>
6.         <encoder>
7.            <!-- 添加敏感信息过滤 -->
8.            <pattern>%date - [%level] - from %logger in %thread %n%replace(%message){'SASL JAAS config=.*', 'SASL JAAS config=******'}%n%xException%n</pattern>
9.         </encoder>
11.         <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
12.             <fileNamePattern>${application.home}/logs/application.%d{yyyy-MM-dd}.log</fileNamePattern>
13.             <maxHistory>5</maxHistory>
14.             <totalSizeCap>5GB</totalSizeCap>
15.         </rollingPolicy>
16.     </appender>
18.     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
19.         <encoder>
20.             <!-- 添加敏感信息过滤 -->
21.             <pattern>%date - [%level] %logger{15} - %replace(%message){'SASL JAAS config=.*', 'SASL JAAS config=******'}%n%xException{10}</pattern>
22.         </encoder>
23.     </appender>
25.     <!-- 其他部分保持不变 -->
26.     <appender name="ASYNCFILE" class="ch.qos.logback.classic.AsyncAppender">
27.         <appender-ref ref="FILE" />
28.     </appender>
30.     <appender name="ASYNCSTDOUT" class="ch.qos.logback.classic.AsyncAppender">
31.         <appender-ref ref="STDOUT" />
32.     </appender>
34.     <logger name="play" level="INFO" />
35.     <logger name="application" level="INFO" />
36.     <logger name="kafka.manager" level="INFO" />
38.     <!-- Off these ones as they are annoying, and anyway we manage configuration ourself -->
39.     <logger name="com.avaje.ebean.config.PropertyMapLoader" level="OFF" />
40.     <logger name="com.avaje.ebeaninternal.server.core.XmlConfigLoader" level="OFF" />
41.     <logger name="com.avaje.ebeaninternal.server.lib.BackgroundThread" level="OFF" />
42.     <logger name="com.gargoylesoftware.htmlunit.javascript" level="OFF" />
43.     <logger name="org.apache.zookeeper" level="INFO"/>
45.     <root level="WARN">
46.         <appender-ref ref="ASYNCFILE" />
47.         <appender-ref ref="ASYNCSTDOUT" />
48.     </root>
50. </configuration>

复制代码

4.3、编写启停脚本

1. vim /usr/local/cmak/cmak-3.0.0.6/start.sh

复制代码

1. #!/bin/bash
3. JDK11_PATH="/usr/local/cmak/jdk-11.0.26"
4. CMAK_PATH="/usr/local/cmak/cmak-3.0.0.6"
5. CMAK_PORT="3283"
6. $CMAK_PATH/bin/cmak \
7.   -Dconfig.file=$CMAK_PATH/conf/application.conf \
8.   -java-home $JDK11_PATH \
9.   -Dhttp.port=$CMAK_PORT > /dev/null &

复制代码

1. vim /usr/local/cmak/cmak-3.0.0.6/stop.sh

复制代码

1. #!/bin/bash
3. CMAK_PATH="/usr/local/cmak/cmak-3.0.0.6"
4. ps aux | grep 'java' | grep 'cmak-3.0.0.6' | awk '{print $2}' | xargs -r kill -9
5. [[ -f "$CMAK_PATH/RUNNING_PID" ]] && rm -f $CMAK_PATH/RUNNING_PID

复制代码

1. chmod +x /usr/local/cmak/cmak-3.0.0.6/{start.sh,stop.sh}

复制代码

4.4、启动

1. sh /usr/local/cmak/cmak-3.0.0.6/start.sh

复制代码

五、添加集群监控

• 留意！是添加 zookeeper 集群
  

1. 10.73.251.66:2261,10.73.251.67:2261,10.73.251.68:2261

复制代码

• 认证设置格式
  

1. org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="2332***JA";

复制代码

六、Kafka 启用 JMX

6.1、添加认证文件

1. vim /usr/local/kafka_2.13-2.8.2/config/jmxremote.access

复制代码

1. monitorRole readonly
2. controlRole readwrite

复制代码

1. vim /usr/local/kafka_2.13-2.8.2/config/jmxremote.password

复制代码

1. monitorRole password123
2. controlRole password456

复制代码

1. chmod 600 /usr/local/kafka_2.13-2.8.2/config/jmxremote.password

复制代码

6.2、设置文件解释

6.2.1、jmxremote.access 文件

   jmxremote.access 文件定义了哪些脚色可以访问 JMX，及其访问级别。
  每个脚色对应差别的权限。
  通过此文件，可以设置差别的脚色及其对 JMX 毗连的权限。
  格式：

1. <role> <permission>

复制代码

• role：脚色名称（如 monitorRole，controlRole）。
  
• permission：该脚色的权限，可以是以下值之一：
  
• readonly：只读权限，允许查看监控数据，但不能进行任何操纵。
  
• readwrite：读写权限，允许查看监控数据并实验管理操纵。
  

示例：

1. monitorRole readonly
2. controlRole readwrite

复制代码

上面的示例表现：

• monitorRole 脚色具有只读权限，用户可以毗连 JMX 进行监控，但不能进行修改或控制。
  
• controlRole 脚色具有读写权限，用户不仅可以进行监控，还可以修改或控制应用程序的行为（例如重启服务等）。
  

6.2.2、jmxremote.password 文件

   jmxremote.password 文件存储了每个脚色的密码。
  这个文件与 jmxremote.access 文件配合利用，确保只有具有正确密码的用户才能毗连到 JMX 服务。
  格式：

1. <role> <password>

复制代码

• role：与 jmxremote.access 文件中脚色匹配。
  
• password：该脚色对应的密码。
  

示例：

1. monitorRole password123
2. controlRole password456

复制代码

上面的示例表现：

• monitorRole 脚色的密码是 password123。
  
• controlRole 脚色的密码是 password456。
  

6.2.3、设置与权限的关系

这两个文件的配合使得 JMX 访问可以进行基于脚色的控制和认证：
jmxremote.access 文件用于定义权限：哪些脚色能做什么操纵。
jmxremote.password 文件用于定义认证：哪些脚色有权访问，并提供密码来验证身份。
6.3、修改启动脚本

1. vim /usr/local/kafka_2.13-2.8.2/kafka_start.sh

复制代码

1. #!/bin/bash
2. kafka_path="/usr/local/kafka_2.13-2.8.2"
3. export KAFKA_OPTS=" -Djava.security.auth.login.config=$kafka_path/config/kafka_server_jaas.conf "
4. $kafka_path/bin/kafka-server-start.sh $kafka_path/config/server.properties > $kafka_path/kafka_nohup.out &

复制代码

改为：

• 留意！！！留意！！！环境变量中参数前后的空格是必须的！！！
  

1. #!/bin/bash
2. kafka_path="/usr/local/kafka_2.13-2.8.2"
4. # 设置 JMX 端口和认证
5. export JMX_PORT=9999
6. export KAFKA_JMX_OPTS=" -Dcom.sun.management.jmxremote \
7.   -Dcom.sun.management.jmxremote.port=$JMX_PORT \
8.   -Dcom.sun.management.jmxremote.rmi.port=$JMX_PORT \
9.   -Dcom.sun.management.jmxremote.authenticate=true \
10.   -Dcom.sun.management.jmxremote.ssl=false \
11.   -Djava.rmi.server.hostname=127.0.0.1 \
12.   -Dcom.sun.management.jmxremote.access.file=$kafka_path/config/jmxremote.access \
13.   -Dcom.sun.management.jmxremote.password.file=$kafka_path/config/jmxremote.password "
15. export KAFKA_OPTS=" $KAFKA_JMX_OPTS -Djava.security.auth.login.config=$kafka_path/config/kafka_server_jaas.conf "
17. # 启动 Kafka 并将日志重定向
18. $kafka_path/bin/kafka-server-start.sh $kafka_path/config/server.properties >> $kafka_path/kafka_nohup.out &

复制代码

6.4、重启 Kafka

1. netstat -lntpu | grep java | grep 9002 | awk -F '[ /]+' '{print $7}' | xargs -r kill -9

复制代码

1. netstat -lntpu | grep java | grep 9999

复制代码

七、集群监控启用 JMX