原文：https://mp.weixin.qq.com/s/9A2bN37Zv2S-IAE-jA0nCw
原文主要讲述使用阿里云自带的云服务进行监控Kafka，本文保留了原文的理论知识，然后使用原生的kafka-exporter进行Kafka的监控。

1、什么是Kafka

1.1、起源

Kafka 是由 Linkedin 公司开发，并捐赠给 Apache 软件基金会的分布式发布订阅消息系统，Kafka 的目的是通过 Hadoop 的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

图片.png

Kafka 的诞生是为了解决 Linkedin 的数据管道问题，用作 LinkedIn 的活动流（Activity Stream）和运营数据处理管道（Pipeline）的基础。起初 Linkedin 采用 ActiveMQ 进行数据交换，但当时的 ActiveMQ 无法满足 Linkedin 对数据传递系统的要求，经常出现消息阻塞或者服务无法正常访问等问题。Linkedin 决定研发自己的消息队列，Linkedin 时任首席架构师 Jay Kreps 便开始组建团队进行消息队列的研发。

1.2、特性

相较于其他消息队列产品，Kafka 存在以下特性：

持久性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失；
高吞吐：Kafka 每秒可以处理百万条消息；
可扩展：Kafka 集群支持热扩展；
容错性：允许集群中节点失败（若副本数量为 n,则允许 n-1 个节点失败）；
高并发：支持数千个客户端同时读写。

与此同时，区别于其他消息队列产品，Kafka 不使用 AMQP 或任何其他预先存在的协议进行通信，使用基于 TCP 的自定义二进制协议。并具有强大的排序语义和持久性保证。

1.3、应用场景

基于以上的特性，Kafka 通过实时的处理大量数据以满足各种需求场景：

大数据领域：如网站行为分析、日志聚合、应用监控、流式数据处理、在线和离线数据分析等领域。
数据集成：将消息导入 ODPS、OSS、RDS、Hadoop、HBase 等离线数据仓库。
流计算集成：与 StreamComput e、E-MapReduce、Spark、Storm 等流计算引擎集成。

1.4、技术架构

一个消息队列 Kafka 版集群包括 Producer、Kafka Broker、Consumer Group、Zookeeper。

图片.png

Producer：消息发布者，也称为消息生产者，通过 Push 模式向 Broker 发送消息。发送的消息可以是网站的页面访问、服务器日志，也可以是 CPU 和内存相关的系统资源信息。
Broker：用于存储消息的服务器。Broker 支持水平扩展。Broker 节点的数量越多，集群吞吐率越高。
Consumer Group：Consumer 被称为消息订阅者或消息消费者，负责向服务器读取消息并进行消费。Consumer Group 指一类 Consumer，这类 Consumer 通常接收并消费同一类消息，且消息消费逻辑一致。通过 Pull 模式从 Broker 订阅并消费消息。
Zookeeper：管理集群配置、选举 Leader 分区，并在 Consumer Group 发生变化时进行负载均衡。其中值得一提的是，如果没有 ZooKeeper 就无法完成 Kafka 部署。ZooKeeper 是将所有东西粘合在一起的粘合剂
发布/订阅模型：Kafka 采用发布/订阅模型，Consumer Group 和 Topic 的对应关系是 N : N，即一个 Consumer Group 可以同时订阅多个 Topic，一个 Topic 也可以被多个 Consumer Group 同时订阅。虽然一个Topic可以被多个 Consumer Group 同时订阅，但该 Topic 只能被同一个 Consumer Group 内的任意一个 Consumer 消费。

2、关键指标

如果使用自建 Kafka，那么需要关注的指标就非常多，主要包含以下四个方向：Broker、Producer、Consumer、Zookeeper。

图片.png

2.1、Broker指标

由于所有消息都必须通过 Broker 才能被使用，因此，对 Broker 进行监控并预警非常重要。Broker 指标关注：Kafka-emitted 指标、Host-level 指标、JVM 垃圾收集指标。

Broker - Kafka-emitted 指标

未复制的分区数：UnderReplicatedPartitions（可用性）

kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

在运行正常集群中，同步副本（ISR）数量应等于副本总数。如果分区副本远远落后于 Leader，则从 ISR 池中删除这个 follower。如果代理不可用，则 UnderReplicatedPartitions 指标急剧增加。
Tips：UnderReplicatedPartitions 较长时间内大于零，需要进行排查。

同步副本（ISR）池缩小/扩展的速率：IsrShrinksPerSec / IsrExpandsPerSec（可用性）

kafka.server:type=ReplicaManager,name=IsrShrinksPerSec

Tips：如果某副本在一段时间内未联系 Leader 或者 follower 的 offset 远远落后于 Leader，则将其从 ISR 池中删除。因此，需要关注 IsrShrinksPerSec / IsrExpandsPerSec 的相关波动。IsrShrinksPerSec 增加，不应该造成 IsrExpandsPerSec 增加。在扩展 Brokers 集群或删除分区等特殊情况以外，特定分区同步副本（ISR）数量应保持相对稳定。

离线分区数（仅控制器）：OfflinePartitionsCount（可用性）

kafka.controller:type=KafkaController,name=OfflinePartitionsCount

顾名思义，主要统计没有活跃 Leader 的分区数。
Tips：由于所有读写操作仅在分区引导程序上执行，因此该指标出现非零值，就需要进行关注，防止服务中断。

集群中活动控制器的数量：ActiveControllerCount（可用性）

kafka.server:type=ReplicaManager,name=IsrShrinksPerSec

Tips：所有 brokers 中 ActiveControllerCount 总和始终等于 1，如出现波动应及时告警。Kafka 集群中启动的第一个节点将自动成为Controller且只有一个。Kafka 集群中的Controller负责维护分区 Leader 列表，并协调 Leader 变更（比如某分区 leader 不可用）。

每秒 UncleanLeader 选举次数：UncleanLeaderElectionsPerSec（可用性）

kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec

在可用性和一致性之间，Kafka 默选了可用性。当 Kafka Brokers 的分区 Leader 不可用时，就会发生 unclean 的 leader 选举。当作为分区 Leader 的代理脱机时，将从该分区的 ISR 集中选举出新的 Leader。
Tips：UncleanLeaderElectionsPerSec 代表着数据丢失，因此需要进行告警。

特定请求（生产/提取）用时：TotalTimeMs（性能）

kafka.network:type=RequestMetrics,name=TotalTimeMs,request={Produce|FetchConsumer|FetchFollower}

TotalTimeMs 作为一个指标族，用来衡量服务请求（包括生产请求，获取消费者请求或获取跟随者请求）的用时，其中涵盖在请求队列中等待所花费的时间 Queue，处理所花费的时间 Local，等待消费者响应所花费的时间 Remote（仅当时requests.required.acks=-1）发送回复的时间 Response。
Tips：正常情况下 TotalTimeMs 应该近似静态且只有非常小的波动。如果发现异常，需要检查各个队列、本地、远程和响应值，定位导致速度下降的确切请求段。

传入/传出字节率：BytesInPerSec / BytesOutPerSec（性能）

kafka.server:type=ReplicaManager,name=IsrShrinksPerSec

Tips：我们可以考虑是否启用消息的端到端压缩等优化措施。磁盘吞吐量、网络吞吐量都可能成为 Kafka 的性能瓶颈。比如跨数据中心发送消息且 Topic 数量众多，或副本恰好是 Leader。

每秒请求数：RequestsPerSec（性能）

kafka.network:type=RequestMetrics,name=RequestsPerSec,request={Produce|FetchConsumer|FetchFollower},version=([0-9]+)

通过 RequestsPerSec，了解 Producer、Consumer、Followers 的请求率，确保 Kafka 的高效通信。
Tips：请求率会随着 Producer发送更多流量或集群扩展而增加，从而增加需要提取消息的 Consumer 或 Followers。如果 RequestsPerSec 持续高企，需要考虑增加 Producer、Consumer、Followers。通过减少请求数量来提高吞吐量，减少非必要开销。

Broker - Host 基础指标 & JVM 垃圾收集指标

除了主机级别的相关指标，由于 Kafka 是由 Scala 编写且运行在 JVM 上，需要依赖 Java 的垃圾回收机制来释放内存，并随着集群活跃度提升，垃圾回收频率不断提升。

消耗磁盘空间消耗与可用磁盘空间：Disk usage（可用性）
由于 Kafka 将所有数据持久保存到磁盘，因此需要监视 Kafka 可用磁盘空间量。
页面缓存读取与磁盘读取的比率：Page cache reads ratio（性能）
类似于数据库 cache-hit ratio 缓存命中率，该指标越高读取速度越快，性能越好。如果副本追上了 Leader（如产生新代理），则该指标短暂下降。
CPU 使用率：CPU usage（性能）
CPU 很少是性能问题根因。但如果发生 CPU 使用率暴涨，最好还是检查一下。
网络字节发送/接收（性能）
代理托管其他网络服务情况下。网络使用率过高可能是性能下降的先兆。
JVM 执行垃圾回收进程总数：CollectionCount（性能）

java.lang:type=GarbageCollector,name=G1 (Young|Old) Generation

YoungGarbageCollector 相对经常发生。在执行时所有应用线程都会暂停，因此该指标的波动会造成 Kafka 性能的波动。

JVM 执行垃圾收集进程用时：CollectionTime（性能）

java.lang:type=GarbageCollector,name=G1 (Young|Old) Generation

OldGarbageCollector 释放老堆栈中未使用的内存，虽然也会暂停应用线程，但只是间歇运行。如果该动作的耗时或者发生频次过高，需要考虑是否有相应的内存支撑。

2.2、Producer指标

Producer 将消息推送到 Broker 进行消费。如果 Producer 失败，Consumer 将没有新消息。因此，我们需要监测以下指标，保障稳定的传入数据流。

每秒收到的平均响应数: Response rate（性能）

kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)

对于 Producer，响应率表示从 Brokers 收到的响应率。收到数据后，Brokers 对 Producer 做出响应。结合 request.required.acks 实际配置，“收到”具备不同含义，比如：Leader 已将消息写入磁盘，Leader 已从所有副本收到确认已将数据写入磁盘。在收到确认之前，Producer 数据不可用于消费。

每秒发送的平均请求数: Request rate（性能）

kafka.network:type=RequestMetrics,name=RequestsPerSec,request={Produce|FetchConsumer|FetchFollower},version=([0-9]+)

请求速率指 Producer 将数据发送给 Brokers 的速率。速率走势是保障服务可用性的重要指标。

平均请求等待时长: Request latency average（性能）

kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.\w]+),node-id=([0-9]+)

从调用 KafkaProducer.send()到 Producer 收到来自 Broker 的响应之间的时长。Producer 的 linger.ms 值确定在发送消息批之前将等待的最长时间，这允许它累积大量消息，再在单个请求中发送它们。如果增加 linger.ms 提高 Kafka 吞吐量，则应关注请求延迟，确保不会超过限制。

每秒平均传出/传入字节数：Outgoing byte rate（性能）

kafka.producer:type=producer-metrics,client-id=([-.w]+)

了解 Producer 效率，并定位可能的传输延迟原因。

I / O 线程等待的平均时长: I/O wait time（性能）

kafka.[producer|consumer|connect]:type=[producer|consumer|connect]-metrics,client-id=([-.\w]+)

每个分区每个请求发送的平均字节数：Batch size（性能）

kafka.producer:type=producer-metrics,client-id=([-.w]+)

为了提升网络资源使用率，Producer 尝试在发送消息前将消息分组。Producer 将等待累积由 batch.size 定义的数据量，等待时长受 linger.ms 约束。

2.3、Consumer指标

Consumer 在此分区上滞后于 Producer 的消息数：Records lag/Records lag max（性能）

kafka.consumer:type=consumer-fetch-manager-metrics,partition="{partition}",topic="{topic}",client-id="{client-id}"

该指标用来记录 Consumer 当前的日志偏移量和 Producer 的当前日志偏移量之间的计算差。如果 Consumer 是处理实时数据，则始终较高的滞后值可能表示使用者过载，在这种情况下，配置更多使用者和将 Topic 划分到更多分区中提高吞吐量并减少滞后。

特定 Topic 每秒平均消耗的字节数: bytes consumed rate（性能）

kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{client-id}",topic="{topic}"

特定 Topic 每秒平均消耗的记录数: records consumed rate（性能）

kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{client-id}",topic="{topic}"

Consumer 每秒获取的请求数: fetch rate（性能）

kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{client-id}",topic="{topic}"

该指标可以直观反映 Consumer 的整体状况。接近零值的获取率表明 Consumer 存在问题。如果出现指标下降，则可能是 Consumer 消费消息失败。

相关指标可以参考 Kafka 官方文档，指标名称、指标定义、Mean name 在实际操作过程中以文档中最新版本为准。

3、搭建Kafka

Kafka部署到Kubernetes集群内部，安装方式使用Helm。

（1）添加helm repo

helm repo add stable http://mirror.azure.cn/kubernetes/charts
helm repo update
helm search repo kafka
helm pull incubator/kafka:0.21.5

（2）修改helm value.yaml配置文件

replicas: 3
image: "confluentinc/cp-kafka"
imageTag: "5.0.1"  # Confluent image for Kafka 2.0.0
imagePullPolicy: "IfNotPresent"
resources:
   limits:
     cpu: 500m
     memory: 2048Mi
   requests:
     cpu: 100m
     memory: 1024Mi
kafkaHeapOptions: "-Xmx1G -Xms1G"
securityContext: {}
updateStrategy:
  type: "OnDelete"
podManagementPolicy: OrderedReady
logSubPath: "logs"
nodeSelector: {}
readinessProbe:
  initialDelaySeconds: 30
  periodSeconds: 10
  timeoutSeconds: 5
  successThreshold: 1
  failureThreshold: 3
terminationGracePeriodSeconds: 60
tolerations: []
headless:
  port: 9092
external:
  enabled: false
  type: NodePort
  dns:
    useInternal: false
    useExternal: true
  distinct: false
  servicePort: 19092
  firstListenerPort: 31090
  domain: cluster.local
  loadBalancerIP: []
  loadBalancerSourceRanges: []
  init:
    image: "lwolf/kubectl_deployer"
    imageTag: "0.4"
    imagePullPolicy: "IfNotPresent"
podDisruptionBudget:
  maxUnavailable: 1 
persistence:
  enabled: true
  size: "20Gi"
  mountPath: "/opt/kafka/data"
  storageClass: "alicloud-disk-efficiency"
testsEnabled: true
zookeeper:
  enabled: true
  resources: ~
  env:
    ZK_HEAP_SIZE: "1G"
  persistence:
    enabled: false
  image:
    PullPolicy: "IfNotPresent"
  port: 2181

（3）执行安装

helm install kafka -n middleware .

4、监控体系

搭建Prometheus，通过Kafka-Exporter进行指标获取。

（1）下载Kafka-Exporter

helm repo add gkarthiks https://gkarthiks.github.io/helm-charts
helm pull gkarthiks/prometheus-kafka-exporter

（2）修改templates下的servicemonitor.yaml文件（由于原有的配置有问题，导致prometheus无法正常抓取指标）

{{- if .Values.prometheus.serviceMonitor.enabled }}
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: {{ include "prometheus-kafka-exporter.fullname" . }}
  {{- if .Values.prometheus.serviceMonitor.namespace }}
  namespace: {{ .Values.prometheus.serviceMonitor.namespace }}
  {{- end }}
  labels:
    app.kubernetes.io/name: {{ include "prometheus-kafka-exporter.name" . }}
    helm.sh/chart: {{ include "prometheus-kafka-exporter.chart" . }}
    app.kubernetes.io/instance: {{ .Release.Name }}
    app.kubernetes.io/managed-by: {{ .Release.Service }}
    {{- if .Values.prometheus.serviceMonitor.additionalLabels }}
{{ toYaml .Values.prometheus.serviceMonitor.additionalLabels | indent 4 -}}
    {{- end }}
spec:
  jobLabel: jobLabel
  selector:
    matchLabels:
      app: {{ template "prometheus-kafka-exporter.name" . }}
      release: {{ .Release.Name }}
  namespaceSelector:
    matchNames:
    - {{ .Release.Namespace }}
  endpoints:
  - port: http
    interval: {{ .Values.prometheus.serviceMonitor.interval }}
    {{- if .Values.prometheus.serviceMonitor.scrapeTimeout }}
    scrapeTimeout: {{ .Values.prometheus.serviceMonitor.scrapeTimeout }}
    {{- end }}
{{- end }}

（2）配置Helm value.yaml，修改后端的kafka地址

kafkaServer:
  - kafka:9092

（3）安装

helm install kafka-exporter -n middleware .

安装完成后在Prometheus Dashboard上看到监控的Target。
图片.png

（4）导入dashboard，[https://grafana.com/grafana/dashboards/7589]
图片.png

目录CONTENT

在K8S中部署以及监控Kafka

1、什么是Kafka

1.1、起源

1.2、特性

1.3、应用场景

1.4、技术架构

2、关键指标

2.1、Broker指标

2.2、Producer指标

2.3、Consumer指标

3、搭建Kafka

4、监控体系

评论区