「布道师系列文章」宝兰德徐清康解析 Kafka 和 AutoMQ 的监控

作者｜北京宝兰德公司解决方案总监徐清康

01

前言

当我们使用一个软件的时候，经常都会问这个软件怎么监控、监控他的哪些指标？Kafka 的监控挺长时间都是一个老大难的问题，社区在监控方面一直没有投入太大的精力。如果要实现一个全面的 Kafka 监控框架，至少应该囊括 Kafka 所在主机资源、JVM（毕竟 Kafka 的 Broker 就是一个 Java 进程）、Kafka 集群本身等的监控，监控 Kafka 集群时还需要关注其客户端程序的性能。本文关注的重点在于 Kafka 和 AutoMQ 集群的监控，对于主机监控和 JVM 监控大家应该已经非常熟悉了。为了更好的说明，先对所涉及的验证环境进行简要介绍，其中包含依赖组件 ZooKeeper、Kafka/AutoMQ 集群自身、CMAK 监控服务。

02

Kafka 监控

为了简单起见，ZooKeeper 只采用了单实例运行。Kafka Broker 则使用了 3 台主机搭建了真实的集群。CMAK 全称为 Cluster Manager for Apache Kafka，也就是以前的 Kafka Manager 改名而来，由 Yahoo 奉献的管理 Kafka 的开源软件。笔者最开始看到这个 CMAK 时老联想起跨平台编译工具“CMake”～_～。

1. 给 Kafka 开启 JMX 监控

Kafka 的官方文档中提到了可以通过 JMX 方式获取 Kafka 的 metrics 信息，可参照 https://kafka.apache.org/documentation/#monitoring。聊到 JMX，其全称为“Java Management Extensions”，它是一种用于监控和管理 Java 应用程序的框架。曾经非常流行，包含著名的 WebLogic 在内的众多 JavaEE 应用服务器中间件，其管理和监控端都可以通过 JMX 技术进行中间件配置管理和监控。很多使用 Zabbix 监控软件对应用服务器中间件监控的场景，也是通过 JMX 方式进行的。JMX 技术的几个核心要点：

1. 在 JVM 体系里边，需要被管理或者监控的对象，最后都叫做一个 MBean（也就是 Managed Bean），我们不需要去管 MBean 有标准 MBean、动态 MBean、模型 MBean，只需要知道本质其实就是一个普通的 Java 对象即可。

2. 每个 MBean 通过一个 ObjectName 进行标识，ObjectName 通常是多个键值对组合而成。

3. 需要管理和监控的项，则以 Attribute 进行暴露。

以 Kafka 为例，如果需要获取一个主题中入队的消息数量统计，则需要查看 ObjecyName 为“kafka.server：type=BrokerTopicMetrics，name= MessagesInPerSec，topic=mainTopic1”的 MBean，该 MBean 会有“Count”、“OneMinuteRate”、“FiveMinuteRate”、“FifteenMinuteRate”等属性分别表示“消息总数”、“过去 1 分钟入队消息速率”、“过去 5 分钟入队消息速率”、““过去 15 分钟入队消息速率”。Kafka 默认是不开启 JMX 的，因此需要在启动之前通过设置 JMX_PORT 环境变量让 Kafka 开启 JMX。

#!/bin/sh
KAFKA_HOME=/home/xuqingkang/kafka-3.7.0
export KAFKA_HEAP_OPTS=" -Xms6g -Xmx6g"
export JMX_PORT="19009"
export KAFKA_JVM_PERFORMANCE_OPTS=" -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true"$KAFKA_HOME/bin/kafka-server-start.sh -daemon 
$KAFKA_HOME/config/server.properties

环境变量中包含 JMX_PORT 时，Kafka 启动脚本通过-D 设置 JMX 相关的几个属性，如"-Dcom.sun.management.jmxremote"，便开启了 JMX 支持。

>cat kafka-run-class.sh
# JMX settings
if [ -z "$KAFKA_JMX_OPTS" ]; then
  KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false  -Dcom.sun.management.jmxremote.ssl=false "
fi
# JMX port to use
if [  $JMX_PORT ]; then
   KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.port=$JMX_PORT "
   if ! echo "$KAFKA_JMX_OPTS" | grep -qF -- '-Dcom.sun.management.jmxremote.rmi.port=' ; then
    # If unset, set the RMI port to address issues with monitoring Kafka running in containers
    KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.rmi.port=$JMX_PORT"
  fi
fi

当 Kafka 的 Broker 开启了 JMX，使用 JDK 自带的 JConsole 工具就可以连接。打开 jconsole，通过远程的方式指定 ip 地址和端口即可。

JConsole 上正常连接后就可以在 MBean Tab 页，查看 Kafka 众多暴露的 MBean 从而查看到监控数据。如下为获取 Kafka Broker 状态的 MBean。

2. 基于 JMX 自实现 Java 客户端获取 Kafka 监控数据

在某些环境上，可能无法使用 jconsole 可视化界面，那么可以自实现 Java 类，获取 Kafka 监控信息，自实现的 Java 类不需要依赖任何 JDK 之外的 jar 包就可以运行，还是非常的方便。Java 代码的主要逻辑为获取MBeanServer 连接，通过 MBeanServer 连接进行 MBean 的查询以及属性获取。如下代码样例主要是查询 Kafka Broker 之上的所有 LogEndOffset MBean，获取各主题（Topic）分区的 LogEndOffset。

import javax.management.*;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import java.io.IOException;
import java.net.MalformedURLException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * Created by qingkang.xu on 2024/6/2.
 */
 public class KafkaJMXMonitor {

     private static MBeanServerConnection conn;
     
     //设置Kafka Broker的JMX连接信息，包括主机ip和JMX_PORT
     private static String ipAndPort = "192.168.32.170:19009";
    
     public static void main(String[] args) {
         // 1、初始化，获取Kafka JMX MbeanServer连接
         if(!init()){
              return;
         }                
         
         // 2、通过正则表达式，获取Kafka下所有主题分区的LogEndOffset MBean
         Set objectNames = null;
         try {            
             // Kafka每个主题所有分区，每个分区对应一个MBean, topic和partition设为"*"模糊查询。
             ObjectName logOffsetObjName = new ObjectName(               
                "kafka.log:type=Log,name=LogEndOffset,topic=*,partition=*");
              objectNames = conn.queryNames(logOffsetObjName,null);
         } catch (MalformedObjectNameException e) {
            e.printStackTrace();
            return;
         } catch (IOException e) {
            e.printStackTrace();
            return;
         }  
         
         if(objectNames == null){ 
             return; 
         }  
         
         //  3、获取每一个主题分区对应的LogEndOffset MBean信息，得到LogEndOffset        
         for(ObjectName objName:objectNames){
             String topicName = objName.getKeyProperty("topic");
             //  __consumer_offsets为kafka存储客户端offset的专用主题，忽略            
             if("__consumer_offsets".equals(topicName)){
                 continue; 
             }                        
             
             int partId = Integer.parseInt(objName.getKeyProperty("partition"));            
             try{ 
                 Object val = conn.getAttribute(objName,"Value");
                 if(val !=null){
                     System.out.println("{topicName:" + topicName + ",partition:" + partId + ",LogEndOffset:" + val);
                  } 
             }catch (Exception e) {
               e.printStackTrace();
               return; 
             }
          }  
      }
      
    //  初始化JMX MBeanServer连接
    public static boolean init(){
        String jmxURL = "service:jmx:rmi:///jndi/rmi://" +ipAndPort+ "/jmxrmi";  
        System.out.println("Init JMX, jmxUrl: {" + jmxURL + "}, and begin to connect it");
        try {
             JMXServiceURL serviceURL = new  JMXServiceURL(jmxURL);          
             JMXConnector connector = JMXConnectorFactory.connect(serviceURL,null); 
             conn = connector.getMBeanServerConnection(); 
             if(conn == null){
                 System.err.println("Get JMX Connection Return Null!");
                 return  false;
              }
          } catch (MalformedURLException e) {
             e.printStackTrace();
             return false; 
          } catch (IOException e) {
              e.printStackTrace();
              return false;
          }        
          return true;
    }
}

如上代码编译运行的结果类似如下：从结果可以看出该 Broker 上有 3 个主题，每个主题有两个分区，各分区的 LogEndOffset 也正常的获取到。

[xuqingkang@rhel75-170 jmx-client]$ javac KafkaJMXMonitor.java
[xuqingkang@rhel75-170 jmx-client]$ java -cp . KafkaJMXMonitor
Init JMX, jmxUrl: {service:jmx:rmi:///jndi/rmi://192.168.32.170:19009/jmxrmi}, and begin to connect it
{topicName:mainTopic1,partition:0,LogEndOffset:23
{topicName:mainTopic2,partition:2,LogEndOffset:0
{topicName:mainTopic2,partition:0,LogEndOffset:8
{topicName:mainTopic3,partition:0,LogEndOffset:0
{topicName:mainTopic3,partition:2,LogEndOffset:0
{topicName:mainTopic1,partition:1,LogEndOffset:6

3. CMAK 管理出场
如上的监控在界面体验上都不是很方便，到 CMAK 出场的时候了，通过如下几步便可以实现一个直观的 B/S 架构的 Kafka 监控。1、通过 git 从 github 获取 CMAK：

git clone https://github.com/yahoo/CMAK.git

2、编译 CMAK 和配置CMAK 的编译构建使用 sbt 即可，sbt 是专门用于构建 Scala 项目的编译构建工具，类似于大家熟知的 Maven 和 Gradle 等。但在编译构建之前，按照官方要求需要的 JDK 版本为 11 以上。因此在真正运行 sbt 之前需要将 11 版本以上的 JDK 路径进行正确的设置，参照如下命令：

PATH=/home/xuqingkang/jdk-14.0.2/bin:$PATH 
JAVA_HOME=/home/xuqingkang/jdk-14.0.2 
/home/xuqingkang/CMAK/sbt -java-home /home/xuqingkang/jdk-14.0.2 clean dist

如果一切顺利，最后将会看到"Your package is ready in"的提示告知最后编译好的 CMAK 所在路径，如下所示：

......
[info] Compiling 140 Scala sources and 2 Java sources to /home/xuqingkang/CMAK/target/scala-2.12/classes ...
[info] LESS compiling on 1 source(s)
model contains 662 documentable templates
[info] Main Scala API documentation successful.
[success] All package validations passed
[info] Your package is ready in /home/xuqingkang/CMAK/target/universal/cmak-3.0.0.7.zip
[success] Total time: 114 s (01:54), completed 2024年6月2日 上午10:39:58

使用 unzip 命令对 target/universal/cmak-3.0.0.7.zip 进行解压，解压之后修改 conf/application.conf 文件，将 cmak.zkhosts 修改为实际的 zookeeper 地址。

cmak.zkhosts="192.168.32.170:2181"

最后，就可以使用 cmak 的脚本直接启动了，因为笔者环境中有多个版本的 JDK，因此特意使用"-java-home"选项指定了 jdk 14 版本的 JDK，以满足官方要求的 11 以上。

nohup ./bin/cmak -java-home /home/xuqingkang/jdk-14.0.2/ &

3、CMAK 运行和基本使用CMAK 默认使用的端口是 9000，也可以在启动的时候使用“-Dhttp.port”选项进行修改，比如“cmak -Dhttp.port=8080”。在启动完成之后，便可以使用浏览器进行访问。当看到 CMAK 界面后，第一步应该做的事情就是在“Cluster”菜单中点击“Add Cluster”将需要管理的 Kafka 集群信息录入。

在添加 Kafka 集群的时候，最重要的信息就是设置“Cluster Zookeeper Hosts”选项，保持和真正的 Kafka 集群的 ZooKeeper 一致即可。对于“Enable JMX Polling”选项，是获取到“Combined Metric”指标必须具备选择的，他也需要 Kafka Broker 启动的时候设置 JMX_PORT 环境变量，使 Kafka 开启 JMX 监控。

如果设置正确，接下来就可以在集群清单中选择创建的 kafka 集群进行监控。

03

AutoMQ 的监控

AutoMQ 的官网给出的监控指引中，对 Metrics 集成做了比较详细的说明。如下来自 AutoMQ 官网的图中，在大家所熟知的 Grafana+Promethues 组合之间增加了 OT Receiver 和 OT Collector 这两个角色，实际上是对 OpenTelemetry 的支持。Grafana+OpenTelemetry+Promethues 的监控架构。引入的 OT Collector 和 OT Receiver 主要是让 Kafka/AutoMQ 服务作为被监控对象，可以将自己的 Metrics 信息直接以推送（Push）的方式上报。而且 OT Collector 可以完成数据过滤、聚合、转换以及将数据导出到多个后端等动作，这是当前构建云原生大规模监控的主流架构。

参照下文中几个步骤，便可以搭建好 Grafana、OTel Collector、Promethues 服务，并且将 Kafka/AutoMQ 的 Controller、Broker 接入到监控中。

1. 以 Docker 方式运行 Promethues 等核心监控服务

AutoMQ 的源码中（链接 https://github.com/AutoMQ/automq\)，docker/telemetry/目录下有对应构建监控服务的脚本以及监控服务配置样例。可以直接运行 "install.sh start"，该脚本主要是以 docker-compose 的方式将 Grafana、Promethues、OTel Collector 这些服务以 Docker 容器的方式运行。如下是核心代码和配置的摘录：

>cat install.sh //直接调用docker compose启动容器
start_containers() {
  docker compose -f ./docker-compose.yaml up -d
  echo "Done."
}

docker-compose.yaml 中则包含了 grafana、promethues 等各个服务的容器启动配置。

>cat docker-compose.yaml
version: '3'
services:
  grafana:
    image: grafana/grafana-enterprise
    container_name: grafana
    ......
    extra_hosts:
      - "host.docker.internal:host-gateway"
  prometheus:
    image: prom/prometheus
    ports: 
      - 9090:9090
    ......
    extra_hosts:
      - "host.docker.internal:host-gateway"
  alertmanager:
    image: prom/alertmanager
    ports:
      - "9087:9087"
    ......
    extra_hosts:
      - "host.docker.internal:host-gateway"
  otel-collector:
    image: otel/opentelemetry-collector-contrib
    ......
      - 8890:8890 # Prometheus exporter metrics
      - 13133:13133 # health_check extension
      - 4317:4317 # OTLP gRPC receiver
      - 4318:4318 # OTLP http receiver 
    extra_hosts: 
      - "host.docker.internal:host-gateway"

运行"install.sh start"之后可以在当前主机中使用 docker ps 查看监控服务对应的容器是否正常运行。一切正常的话，"http://ip:3000"应该已经可以打开 grafana 的界面了。

>docker ps
5af530eebd6c   grafana/grafana-enterprise                 "/run.sh"                33 hours ago     Up 33 hours       0.0.0.0:3000->3000/tcp, :::3000->3000/tcp                                                                                                                                                                                                                        grafana
21bbd335c5a3   prom/prometheus                            "/bin/prometheus --s…"   33 hours ago     Up 33 hours       0.0.0.0:9090->9090/tcp, :::9090->9090/tcp                                                                                                                                                                                                                        telemetry-prometheus-1
1914f31ef125   otel/opentelemetry-collector-contrib       "/otelcol-contrib --…"   33 hours ago     Up 33 hours       0.0.0.0:1888->1888/tcp, :::1888->1888/tcp, 0.0.0.0:4317-4318->4317-4318/tcp, :::4317-4318->4317-4318/tcp, 0.0.0.0:8888->8888/tcp, :::8888->8888/tcp, 0.0.0.0:8890->8890/tcp, :::8890->8890/tcp, 0.0.0.0:13133->13133/tcp, :::13133->13133/tcp, 55678-55679/tcp   telemetry-otel-collector-1
00cf4d65a6a1   phpmyadmin/phpmyadmin                      "/docker-entrypoint.…"   8 months ago     Up 3 days         0.0.0.0:19001->80/tcp, :::19001->80/tcp                                                                                                                                                                                                                          7218907947dc48c1

2. 修改 AutoMQ 的 Broker 和 Controller 配置以接入监控。

不得不说，这里本人踩了不少坑，最后发现其实很简单。核心是在启动 Broker 和 Contoller 之前，需要在其配置文件中增加 metrics 配置以使得 Broker 和 Controller 能将监控数据以 Push 推送的方式发到 OTel Collector。注意 otlp.endpoint 需改为您实际环境中的 IP 地址即可，因为在 Docker 容器中也是可以直接访问宿主机网络的，这里使用了宿主机的 IP 地址。

s3.telemetry.metrics.enable=true
s3.telemetry.metrics.exporter.type=otlp
s3.telemetry.exporter.otlp.endpoint=http://10.0.4.14:4317

以容器的方式运行，推荐在官方镜像的基础之上对关键配置文件做变更即可，重新构建“automqinc/automq”镜像，具体步骤如下：1）、构建目录，其中 config 目录下的配置文件全是从官方镜像拷贝出来的，当然也可以从 github 源码中直接拷贝。

[root@txcloud-centos8-1 addOTelinDocker]# tree
.
├── config
│   └── kraft
│       ├── broker.properties
│       ├── controller.properties
│       └── server.properties
├── Dockerfile
└── makeDocker.sh

2）、3 个 properties 文件中增加"s3.telemetry..."配置

############################# Server Basics #####
# The role of this server. Setting this puts us in KRaft modeprocess.roles=broker
# The node id associated with this instance's roles
node.id=2

# The connect string for the controller quorum
controller.quorum.voters=1@localhost:9093
############################# Socket Server Settings #############################
#     listeners = PLAINTEXT://your.host.name:9092listeners=PLAINTEXT://localhost:9092
# The Prometheus HTTP server host and port, if exporter type is set to prometheus
# s3.metrics.exporter.prom.host=127.0.0.1
# s3.metrics.exporter.prom.port=9090
# The OTel Collector endpoint, if exporter type is set to otlp or tracing is enabled
# s3.telemetry.exporter.otlp.endpoint=http://${your_host_name}:4317
s3.telemetry.metrics.enable=true
s3.telemetry.metrics.exporter.type=otlp
s3.telemetry.exporter.otlp.endpoint=http://10.0.4.14:4317

3）、完成 Dockerfile 和 makeDocker.sh 脚本，逻辑比较简单，就是在官方镜像基础之上实现配置文件覆盖，构建新镜像。直接运行 makeDocker.sh 脚本在本地构建好新镜像。

[root@txcloud-centos8-1 addOTelinDocker]# cat Dockerfile 
# pull base image
# --------------
FROM automqinc/automq:latest

# Maintainer
# --------------
MAINTAINER support  

COPY config/kraft/server.properties /opt/kafka/kafka/config/kraft/server.properties
COPY config/kraft/broker.properties /opt/kafka/kafka/config/kraft/broker.properties
COPY config/kraft/controller.properties /opt/kafka/kafka/config/kraft/controller.properties

[root@txcloud-centos8-1 addOTelinDocker]# cat makeDocker.sh 
#!/bin/sh
docker build --force-rm=true --no-cache=true -t automqinc/automq:latest -f Dockerfile .

3、启动 AutoMQ Broker/Controller 等服务同样也是需要参照 AutoMQ 官方的文档，在本地以 Docker 的方式快速拉起 Broker 和 Controller 等服务。注意，因为上一步骤已经在本地构建了“automqinc/automq”镜像，因此 docker-compose 会使用它。

curl https://download.automq.com/community_edition/standalone_deployment/install_run.sh | bash

AutoMQ 官方提供的 install_run.sh 脚本，其核心逻辑是从官网下载 docker-compose.yaml 文件后用 docker-compose 拉起 Broker 和 Controller，如下是关键代码的摘录：

curl -O https://download.automq.com/community_edition/standalone_deployment/docker-compose.yaml
if [ ! -f "docker-compose.yaml" ]; then
    echo "[ERROR] Docker compose yaml file not exist."
        exit 4
fi
# Check if the current operating system is Linux
if [[ "$(uname)" == "Linux" ]]; then
    echo "Please enter your password for sudo:"
    sudo /usr/local/bin/docker-compose -f docker-compose.yaml up -d || exit 5
else
    docker-compose -f docker-compose.yaml up -d || exit 5
fi

docker-compose.yaml 文件则指示了 Broker 和 Controller 这些核心的服务，怎么启动其 Docker 容器。

version: "3.8"
services:
  ......
  controller:    image: automqinc/automq:latest
  ......
  command:
    - bash
    - -c
    - |
      /opt/kafka/scripts/start.sh up --process.roles controller --node.id 0 --controller.quorum.voters 0@controller:9093 --s3.bucket automq-data --s3.endpoint http://10.6.0.2:4566 --s3.region us-east-1
  networks:
    - automq_net    
  depends_on:
    - localstack
    - aws-cli
  broker1:
    image: automqinc/automq:latest
    ......
    command:
      - bash
      - -c
      - |
        /opt/kafka/scripts/start.sh up --process.roles broker --node.id 1 --controller.quorum.voters 0@controller:9093 --s3.bucket automq-data --s3.endpoint http://10.6.0.2:4566 --s3.region us-east-1

  broker2:
    ...雷同broker1

最后，通过浏览器访问 Grafana，即可看到 AutoMQ 的监控数据，包括 Controller 数量、Broker 数量、Partition 数量、每秒收发字节数、主题、消费组等等。

END

关于我们

我们是来自 Apache RocketMQ 和 Linux LVS 项目的核心团队，曾经见证并应对过消息队列基础设施在大型互联网公司和云计算公司的挑战。现在我们基于对象存储优先、存算分离、多云原生等技术理念，重新设计并实现了 Apache Kafka 和 Apache RocketMQ，带来高达 10 倍的成本优势和百倍的弹性效率提升。

🌟 GitHub 地址：https://github.com/AutoMQ/automq
💻 官网：https://www.automq.com