Prometheus 监控之 kafka

男娘i 2022-04-11 10:30 874阅读 0赞

### 初探 ###

默认情况下, Kafka metrics 所有的 metric 都可以通过 JMX 获取，暴露kafka metrics 支持两种方式  
1.在 Kafka Broker 外部, 作为一个独立进程, 通过 JMX 的 RMI 接口读取数据. 这种方式的好处是有任何调整不需要重启 Kafka Broker 进程, 缺点是多维护了一个独立的进程。  
2.在 Kafka Broker 进程内部读取 JMX 数据, 这样解析数据的逻辑就在 Kafka Broker 进程内部, 如果有任何调整, 需要重启 Broker。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTM0NDAx_size_16_color_FFFFFF_t_70]

### 选择暴露 kafka-metric 方式 ###

第一种需要外部多维护一个程序，而且还要考虑之后各种版本升级，实现起来比较繁琐，还好的是github上有许多优秀的开源kafka\_exporter 下载过来直接启动就好了。简单介绍下  
git项目地址：[https://github.com/danielqsj/kafka\_exporter][https_github.com_danielqsj_kafka_exporter]  
下载地址： [https://github.com/danielqsj/kafka\_exporter/releases/download/v1.2.0/kafka\_exporter-1.2.0.linux-amd64.tar.gz][https_github.com_danielqsj_kafka_exporter_releases_download_v1.2.0_kafka_exporter-1.2.0.linux-amd64.tar.gz]  
启动

kafka_exporter --kafka.server=kafka:9092 [--kafka.server=another-server ...]

Grafana画图也有许多优秀的开源dashboard

第二种是读取 JMX 的数据. Prometheus 官方的组件 jmx\_exporter 把两种实现都提供了:

*  jmx\_prometheus\_httpserver 通过独立进程读取 JMX 的数据
 *  jmx\_prometheus\_javaagent 使用 Java Agent 方式, 尽量无侵入(仅需在 java 命令行中使用 -javaagent 参数)的启动 in-process library, 读取 JMX 数据.  
    Prometheus 采用了 PULL 方式, Prometheus 主动抓取 metrics 数据, 而不是靠客户端主动 PUSH 数据, 因此 jmx\_prometheus 都是通过暴露 HTTP 端口的方式暴露 metrics 数据, 方便 Prometheus 抓取数据.

### 选择方案2 ###

我们这里选择第二种jmx\_prometheus\_javaagent 方式收集kafka指标

##### 部署流程： #####

###### 下载jmx\_prometheus\_javaagent和kafka.yml ######

wget https://raw.githubusercontent.com/prometheus/jmx_exporter/master/example_configs/kafka-0-8-2.yml
    wget https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.6/jmx_prometheus_javaagent-0.6.jar

###### 打开 [kafka-server-start.sh][] 文件 ######

添加几行代码：

export JMX_PORT="9999"
    export KAFKA_OPTS="-javaagent:/path/jmx_prometheus_javaagent-0.6.jar=9991:/path/kafka-0-8-2.yml"

然后重启kafka。  
访问 [http://localhost:9991/metrics][http_localhost_9991_metrics] 可以看到各种指标了。

### 监控指标 ###

部分监控指标解释，不一定准确，请参考。还有参考 [monitoring kafafka][] 有详细的指标信息

<table> 
 <thead> 
  <tr> 
   <th>指标</th> 
   <th>解释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>kafka_server_replicafetchermanager_maxlag</td> 
   <td>Max</td> 
  </tr> 
  <tr> 
   <td>kafka_server_replicamanager_isrexpands_total</td> 
   <td>ISR expansion rate 扩大率(ISR是in-sync replicas的简写)</td> 
  </tr> 
  <tr> 
   <td>kafka_server_replicamanager_isrshrinks_total</td> 
   <td>ISR shrink rate 收缩率</td> 
  </tr> 
  <tr> 
   <td>kafka_server_replicamanager_underreplicatedpartitions</td> 
   <td># of under replicated partitions (|ISR| &lt; |all replicas|)</td> 
  </tr> 
  <tr> 
   <td>kafka_network_requestmetrics_responsesendtimems</td> 
   <td>Time to send the response Produce</td> 
  </tr> 
  <tr> 
   <td>kafka_network_socketserver_networkprocessoravgidlepercent</td> 
   <td>The average fraction of time the network processors are idle</td> 
  </tr> 
  <tr> 
   <td>kafka_network_requestmetrics_responsesendtimems</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>kafka_network_requestmetrics_requestqueuetimems</td> 
   <td>Time the request waiting in the request queue Produce</td> 
  </tr> 
  <tr> 
   <td>kafka_network_requestmetrics_remotetimems</td> 
   <td>Time the request waits for the follower Produce</td> 
  </tr> 
  <tr> 
   <td>kafka_network_requestmetrics_localtimems</td> 
   <td>Time the request being processed at the leader Produce</td> 
  </tr> 
  <tr> 
   <td>kafka_log_logflushstats_logflushrateandtimems_count</td> 
   <td>Log flush latency</td> 
  </tr> 
  <tr> 
   <td>kafka_server_replicafetchermanager_minfetchrate</td> 
   <td>Max lag in messages btw follower and leader replicas &gt; 4000</td> 
  </tr> 
  <tr> 
   <td>kafka_controller_controllerstats_uncleanleaderelectionspersec</td> 
   <td>Unclean leader election has occurred last 15m</td> 
  </tr> 
  <tr> 
   <td>kafka_server_replicamanager_underreplicatedpartitions</td> 
   <td>Under replicated partitions</td> 
  </tr> 
  <tr> 
   <td>kafka_controller_kafkacontroller_activecontrollercount</td> 
   <td>活跃的 Controller 的数量</td> 
  </tr> 
  <tr> 
   <td>kafka_controller_controllerstats_uncleanleaderelectionspersec</td> 
   <td>争议的 leader 选举次数</td> 
  </tr> 
  <tr> 
   <td>kafka_controller_controllerstats_controlledshutdownrateandtimems</td> 
   <td>将ISR中处于关闭状态的副本从集合中去除掉，返回一个新的ISR集合，然后选取第一个副本作为leader，然后令当前AR作为接收LeaderAndIsr请求的副本。</td> 
  </tr> 
  <tr> 
   <td>kafka_controller_kafkacontroller_offlinepartitionscount</td> 
   <td>从活着的ISR中选择一个broker作为leader，如果ISR中没有活着的副本，则从assignedReplicas中选择一个副本作为leader，leader选举成功后注册到Zookeeper中，并更新所有的缓存。</td> 
  </tr> 
  <tr> 
   <td>broker指标</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>kafka_server_brokertopicmetrics_messagesin_total</td> 
   <td>所有topic消息(进出)流量 消息写入总量</td> 
  </tr> 
  <tr> 
   <td>kafka_server_brokertopicmetrics_bytesrejected_total</td> 
   <td>扔掉的流量</td> 
  </tr> 
  <tr> 
   <td>kafka_server_brokertopicmetrics_failedfetchrequests_total</td> 
   <td>当前机器fetch请求失败的数量</td> 
  </tr> 
  <tr> 
   <td>kafka_server_brokertopicmetrics_bytesout_total</td> 
   <td>输出的流量</td> 
  </tr> 
  <tr> 
   <td>kafka_server_brokertopicmetrics_bytesin_total</td> 
   <td>输入的流量</td> 
  </tr> 
  <tr> 
   <td>kafka_server_brokertopicmetrics_failedproducerequests_total</td> 
   <td>当前机器produce请求失败的数量</td> 
  </tr> 
  <tr> 
   <td>kafka_server_replicamanager_partitioncount</td> 
   <td>该broker上的partition的数量</td> 
  </tr> 
  <tr> 
   <td>kafka_server_replicamanager_leadercount</td> 
   <td>Leader的replica的数量</td> 
  </tr> 
  <tr> 
   <td>kafka_network_requestmetrics_totaltimems{FetchConsumer\FetchFollower\Produce}</td> 
   <td>一个请求FetchConsumer\FetchFollower\Produce耗费的所有时间</td> 
  </tr> 
 </tbody> 
</table>

### 预警指标分析 ###

kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions \*\*  
含义: 正在复制的 Partition 的数量.  
建议报警阈值: > 0 就建议报警. 但如果 Kafka 集群正在 reassign partition 时, 这个值也会 >0

kafka.controller:type=KafkaController,name=OfflinePartitionsCount  
含义: 没有 Leader 的 Partition 的数量. 处于这个状态的 Partition 是不可读也不可写  
建议报警阈值: >0 一旦出现就报警.

kafka.controller:type=KafkaController,name=ActiveControllerCount  
含义: 活跃的 Controller 的数量.  
建议报警阈值: != 0 就赶紧报警

kafka.server:type=ReplicaManager,name=PartitionCount  
含义: 集群中 Partition 的总数  
建议报警阈值: 感觉这个报警不可控.

kafka\_controller\_controllerstats\_leaderelectionrateandtimems  
含义: Leader election rate 领导人选举率

UncleanLeaderElectionsPerSec  
含义: Unclean leader election rate 争议的 leader 选举次数

描述：所有的topic的消息速率(消息数/秒)  
Mbean名：“kafka.server”:name=“AllTopicsMessagesInPerSec”,type=“BrokerTopicMetrics”  
正常的值：

描述：所有的topic的流入数据速率(字节/秒)  
Mbean名：“kafka.server”:name=“AllTopicsBytesInPerSec”,type=“BrokerTopicMetrics”  
正常的值：

描述：producer或Fetch-consumer或Fetch-follower的请求速率(请求次数/秒)  
Mbean名：“kafka.network”:name="\{Produce|Fetch-consumer|Fetch-follower\}-RequestsPerSec",type=“RequestMetrics”  
正常的值：

描述：所有的topic的流出数据速率(字节/秒)  
Mbean名： “kafka.server”:name=“AllTopicsBytesOutPerSec”,type=“BrokerTopicMetrics”  
正常的值：

描述：刷日志的速率和耗时  
Mbean名： “kafka.log”:name=“LogFlushRateAndTimeMs”,type=“LogFlushStats”  
正常的值：

描述：正在做复制的partition的数量(|ISR| < |all replicas|)  
Mbean名：“kafka.server”:name=“UnderReplicatedPartitions”,type=“ReplicaManager”  
正常的值：0

描述：当前的broker是否为controller  
Mbean名：“kafka.controller”:name=“ActiveControllerCount”,type=“KafkaController”  
正常的值：在集群中只有一个broker的这个值为1

描述：选举leader的速率  
Mbean名：“kafka.controller”:name=“LeaderElectionRateAndTimeMs”,type=“ControllerStats”  
正常的值：如果有broker挂了，此值非0

描述：Unclean的leader选举速率  
Mbean名：“kafka.controller”:name=“UncleanLeaderElectionsPerSec”,type=“ControllerStats”  
正常的值：0

描述：该broker上的partition的数量  
Mbean名： “kafka.server”:name=“PartitionCount”,type=“ReplicaManager”  
正常的值：应在各个broker中平均分布

描述：Leader的replica的数量  
Mbean名： “kafka.server”:name=“LeaderCount”,type=“ReplicaManager”  
正常的值：应在各个broker中平均分布

描述：ISR的收缩(shrink)速率  
Mbean名：“kafka.server”:name=“ISRShrinksPerSec”,type=“ReplicaManager”  
正常的值：如果一个broker挂掉了，一些partition的ISR会收缩。当那个broker重新起来时，一旦它的replica完全跟上，ISR会扩大(expand)。除此之外，正常情况下，此值和下面的扩大速率都是0。

描述：ISR的扩大(expansion)速率  
Mbean名： “kafka.server”:name=“ISRExpandsPerSec”,type=“ReplicaManager”  
正常的值：参见ISR的收缩(shrink)速率

描述：follower落后leader replica的最大的消息数量  
Mbean名：“kafka.server”:name="(\[-.\\w\]+)-MaxLag",type=“ReplicaFetcherManager”  
正常的值：小于replica.lag.max.messages

描述：每个follower replica落后的消息速率  
Mbean名：“kafka.server”:name="(\[-.\\w\]+)-ConsumerLag",type=“FetcherLagMetrics”  
正常的值：小于replica.lag.max.messages

描述：等待producer purgatory的请求数  
Mbean名：“kafka.server”:name=“PurgatorySize”,type=“ProducerRequestPurgatory”  
正常的值：如果ack=-1，应为非0值

描述：等待fetch purgatory的请求数  
Mbean名：“kafka.server”:name=“PurgatorySize”,type=“FetchRequestPurgatory”  
正常的值：依赖于consumer的fetch.wait.max.ms的设置

描述：一个请求(producer,Fetch-Consumer,Fetch-Follower)耗费的所有时间  
Mbean名：“kafka.network”:name="\{Produce|Fetch-Consumer|Fetch-Follower\}-TotalTimeMs",type=“RequestMetrics”  
正常的值：包括了queue, local, remote和response send time

描述：请求(producer,Fetch-Consumer,Fetch-Follower)在请求队列中的等待时间  
Mbean名：“kafka.network”:name="\{Produce|Fetch-Consumer|Fetch-Follower\}-QueueTimeMs",type=“RequestMetrics”  
正常的值：

描述：请求(producer,Fetch-Consumer,Fetch-Follower)在leader处理请求花的时间  
Mbean名：“kafka.network”:name="\{Produce|Fetch-Consumer|Fetch-Follower\}-LocalTimeMs",type=“RequestMetrics”  
正常的值：

描述：请求(producer,Fetch-Consumer,Fetch-Follower)等待follower花费的时间  
Mbean名：“kafka.network”:name="\{Produce|Fetch-Consumer|Fetch-Follower\}-RemoteTimeMs",type=“RequestMetrics”  
正常的值：producer的ack=-1时，非0才正常

描述：发送响应花费的时间  
Mbean名：“kafka.network”:name="\{Produce|Fetch-Consumer|Fetch-Follower\}-ResponseSendTimeMs",type=“RequestMetrics”  
正常的值：

描述：consumer落后producer的消息数量  
Mbean名：“kafka.consumer”:name="(\[-.\\w\]+)-MaxLag",type=“ConsumerFetcherManager”  
正常的值：  
建议对GC耗时和其他参数和诸如系统CPU，I/O时间等等进行监控。在client端，建议对"消息数量/字节数"的速率(全局的和对于每一个topic)，请求的"速率/大小/耗时"进行监控。还有consumer端，所有partition的最大的落后情况和最小的fetch请求的速率。consumer为了能跟上，最大落后数量需要少于一个threshold并且最小fetch速率需要大于0.

### Grafana画图 ###

json文件链接:[https://pan.baidu.com/s/1H6MesKpqi80R14OF5k7auQ][https_pan.baidu.com_s_1H6MesKpqi80R14OF5k7auQ] 密码:kiox

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTM0NDAx_size_16_color_FFFFFF_t_70 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTM0NDAx_size_16_color_FFFFFF_t_70]: /images/20220411/861f4141401c42dbbf28b7fd5f4d2c73.png
[https_github.com_danielqsj_kafka_exporter]: https://github.com/danielqsj/kafka_exporter
[https_github.com_danielqsj_kafka_exporter_releases_download_v1.2.0_kafka_exporter-1.2.0.linux-amd64.tar.gz]: https://github.com/danielqsj/kafka_exporter/releases/download/v1.2.0/kafka_exporter-1.2.0.linux-amd64.tar.gz
[kafka-server-start.sh]: http://kafka-server-start.sh
[http_localhost_9991_metrics]: http://localhost:9991/metrics
[monitoring kafafka]: https://docs.confluent.io/current/kafka/monitoring.html#producer-global-request-metrics
[https_pan.baidu.com_s_1H6MesKpqi80R14OF5k7auQ]: https://pan.baidu.com/s/1H6MesKpqi80R14OF5k7auQ
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1OTM0NDAx_size_16_color_FFFFFF_t_70 1]: /images/20220411/c78728943256449ab0ff0000ac7490c6.png