Kafka 的灵魂伴侣 Logi-KafkaManger(3) 之运维管控 -- 集群列表

2023-09-16
江西
本文字数：7774 字
阅读完需：约 26 分钟

提示：本文可能已过期,请点击原文查看：Kafka的灵魂伴侣Logi-KafkaManger(3)之运维管控--集群列表

推荐一款非常好用的 kafka 管理平台,kafka 的灵魂伴侣 滴滴开源Logi-KafkaManager 一站式Kafka监控与管控平台

1 技术交流

有想进滴滴 LogI 开源用户群的加我个人微信: jjdlmn_进群(备注：进群)群里面主要交流 kakfa、es、agent、LogI-kafka-manager、等等相关技术;群内有专人解答你的问题对~ 相关技术领域的解答人员都有; 你问的问题都会得到回应

有想进滴滴 LogI 开源用户群的加我个人微信: jjdlmn_ 进群(备注：进群)群里面主要交流 kakfa、es、agent、以及其他技术群内有专人解答疑问,你所问的都能得到回应

技术交流
接入集群
物理集群列表集群概览 Broker 信息 Topic 信息 (TODO 页面跳转)磁盘信息 (TODO 页面跳转)partition 信息消费者信息 Region 信息逻辑集群信息 Controller 信息限流信息

项目地址: didi/Logi-KafkaManager: 一站式Apache Kafka集群指标监控与运维管控平台

前面的文章简单介绍了如何接入集群,以及 Topic 的申请和配额申请,这个时候我们还不是很了解Logi-KafkaManager究竟有哪些优点,如何去管理众多的 kafka 集群;

今天这篇文章,我们就来详细的了解一下;运维人员如何去了解和管控我们所有的集群

Part1 运维管控

运维管控这个菜单栏目下面主要是供运维人员来管理所有集群的;

2 接入集群

Kafka的灵魂伴侣Logi-KafkaManger一之集群的接入及相关概念讲解

3 物理集群列表

列出了所有物理集群,点击一个物理集群进去看详细信息;

如果没有信息请检查一下是否正确开启了 JMX; ==> JMX-连接失败问题解决

集群概览

实时流量

指标说明

在这里插入图片描述

因为我发送和消费过消息, 为了不让之前的数据干扰; 我们重新把 Broker 重启一下,Jmx 的数据就会清 0 了; 历史数据清楚就去数据库中把_metrics结尾的表数据全部清空;

执行下面的代码,验证一下实时流量的指标准不准确;下面的代码表示的是: 60S 秒发送 60 条消息; 每条消息大小 1 个字节; 但是在这 60S 内只发送一次消息; 因为将linger.ms=60000, 设置为 60 秒后发送;那么期望中的实时指标是;

    @Test    void contextLoads() {
            Properties props = new Properties();            props.put("bootstrap.servers", "xxxxxxx");            props.put("acks", "all");            props.put("retries", 0);            props.put("batch.size", 16384000);            props.put("linger.ms", 60000);            props.put("buffer.memory", 335544320);            props.put("client.id", "appId_000001_cn.Test2");            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");            Producer<String, String> producer = new KafkaProducer<>(props);            for(int i = 0; i < 60; i++){                //将一个消息设置大一点                byte[] log = new byte[1024];                String slog = new String(log);                producer.send(new ProducerRecord<String, String>("Test2",0, Integer.toString(i),  slog));            }        try {            Thread.sleep(62000);        } catch (InterruptedException e) {            e.printStackTrace();        }        producer.close();    }

复制代码

messagesIn:每秒发送到kafka的消息条数 = 1条byteIn:每秒发送到kafka的字节数 = 1字节totalProduceReques:每秒总共发送的请求数 = 1/60=0.017 这里是请求数量,因为 60s 内实际上只发送了一次请求;

执行代码然后看结果

基本上是符合我们预期的,实时流量数据还是准确的;

除了上面几个指标,我们应该还要关注下面几个异常指标，正常情况下他们都是 0; 如果不为 0 的情况说明可能就有异常了,运维同学就应该去查查异常日志了;byteRejected(B/s) 每秒被拒绝的字节数 failedFetchRequest 每秒拉取失败的请求数 failedProduceRequest 每秒发送失败的请求数 messageIn/totalProduceRequest 消息条数/总请求数也可以关注一下; 假如他们的结果=1; 说明没有批量发送,一条消息就发送了一个请求了

历史流量

指标说明

历史数据都存放在_metrics结尾的表中;

Broker 信息

在这里插入图片描述

上面左边部分是对所有 Broker 峰值使用率的看板, 可以通过这个图简单了解一下 Broker 的峰值情况, 那么这个使用率是怎么计算的,计算的到底准不准确,得需要去源码里面看看, 这个图我们可以作为一个参考值来了解;

副本状态图, 可以理解为在 ISR 中的是同步；不在 ISR 中的是未同步;

我们现在把其中一台 Broker 1 关机模拟 Broker 宕机等异常情况; 可以看到变成了下面这样子;

图中可以看到, 1 的状态为未使用, 0,2 两台 broker 的副本状态都变成了未同步;

副本状态:失效副本分区的个数 大于 0 则这个副本状态就展示未同步; 失效副本分区的个数 UnderReplicatedPartitions 是通过 JMX 访问kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions获取到的值;如果获取的 UnderReplicatedPartitions 值大于 0，有可能是某个 Broker 的问题，也有可能引申到整个集群的问题，也许还要引入其他一些信息、指标等配合找出问题之所在。

注意：如果 Kafka 集群正在做分区迁移（kafka-reassign-partitions.sh）的时候，这个值也会大于 0。

更多关于失效副本分区数异常问题排查请看 失效副本的诊断及预警

理解了副本状态的意思,那上图我们就可以理解了; 之所以 Broker[0,2] 都显示未同步,是因为 Broker 2 中含有[0,2]的副本; Broker2 宕机了,失效副本分区的个数就大于 0 了

删除操作:当 Broker 下线的时候,可以执行删除操作, 一般是当你把这个 Broker 移除集群的时候你就可以去删除掉他, 不过删除之后,如果重新加入到集群还是会被添加回来的; 如果仅仅只是 Broker 宕机就不要删除了;

Leader Rebalance

想要知道这个功能用来干什么, 那么我们得先了解一个概念 leader 均衡机制;

Leader 均衡机制（auto.leader.rebalance.enable=true）

当一个 broker 停止或崩溃时，这个 broker 中所有分区的 leader 将转移给其他副本。这意味着在默认情况下，当这个 broker 重新启动之后，它的所有分区都将仅作为 follower，不再用于客户端的读写操作。
为了避免这种不平衡，Kafka 有一个首选副本的概念。如果一个分区的副本列表是 1，5，9，节点 1 将优先作为其他两个副本 5 和 9 的 leader，因为它较早存在于副本中。你可以通过运行以下命令让 Kafka 集群尝试恢复已恢复正常的副本的 leader 地位：。不会导致负载不均衡和资源浪费，这就是 leader 的均衡机制
# kafka版本 <= 2.4
> bin/kafka-preferred-replica-election.sh --zookeeper zk_host:port/chroot
# kafka新版本
> bin/kafka-preferred-replica-election.sh --bootstrap-server broker_host:port
kafka平衡leader

在配置文件 conf/ server.properties 中配置开启（默认就是开启）auto.leader.rebalance.enable = true与其相关的配置还有leader.imbalance.check.interval.seconds partition 检查重新 rebalance 的周期时间 ; 默认 300 秒;leader.imbalance.per.broker.percentage 标识每个 Broker 失去平衡的比率，如果超过改比率，则执行重新选举 Broker 的 leader；默认比例是 10%;这个比率的算法是：broker 不平衡率=非优先副本的 leader 个数/总分区数，假如一个 topic 有 3 个分区[0,1,2],并且 3 个副本 ,正常情况下,[0,1,2]分别都为一个 leader 副本; 这个时候 0/3=0%;

上面几个配置都是 && 的关系; 同时满足才能触发再平衡;

调优建议：考虑到 leader 重选举的代价比较大，可能会带来性能影响，也可能会引发客户端的阻塞，生产环境建议设置为 false。或者周期设置长一点,比如一天一次;

那么如果我们关闭了均衡机制 , 或者周期时间比较长, 也就有可能造成上面说的问题, 那么 Kafka-manager 就提供了一个手动再平衡的操作;

假如有一台 Broker 宕机了, 等它重启之后, 并且等它副本同步完成之后(为了副本同步与再平衡错开一下), 运维管理人员 就可以操作一下这个 Leader Rebalance ;手动触发一下再平衡;

举个栗子🌰

首先将 broker 的自动均衡关闭auto.leader.rebalance.enable = false ; 并且逐个重启
查看一下某个 Topic 在各个 broker 的 Leader 分布情况 ;我们这里看看 TEST3 这个 TOPIC 的情况;

Broker-0

Broker-1

Broker-2

在逐个启动完成的时候他们的 Leader 分布情况如下;

因为 Broker-2 是我启的第一台; 所以所有分区的 Leader 都集中在这一台机器上; 而后面启动的 Broker 都没有分配到 Leader;这样的情况明显不合理; 所以我们需要执行一次再均衡;

手动执行再均衡策略；下拉选中的 Broker; 这里选择 Broker 的作用是选择这台 Broker 上的所有 Topic 来进行再均衡

再均衡之后再看看 Leader 情况

可以看到均衡之后的结果,Broker-0 分配了 2 个 Leader ; 自动恢复到了之前的分配情况;

PS: Leader Rebalance 时候选择的 Broker 的作用是针对该 Broker 下面的所有 Topic 来进行再均衡; 假如你 3 台 Broker 上的 Topic 都一样,那选哪个 Broker 都一样

Broker 详情

基本信息

展示了当前 Broker 的基本信息和 实时流量 历史流量 ; 注意这里的流量信息展示的是当前这一台 Broker 的流量; 集群概览那里展示的是整个物理集群的所有流量信息(Brokers 之和);

监控信息

按照时间轴展示多个指标信息,当然指标也是当前选中的 Broker 的指标信息;

Topic 信息 (TODO 页面跳转)

展示当前 Broker 下有哪些 Topic；更为详细的介绍情况 TODO....

磁盘信息 (TODO 页面跳转)

展示当前 Broker 的一些磁盘信息; 但是此功能需要接入滴滴的 kafka-gatway 组件才可以生效; 目前该组件为企业服务,暂未有开源计划; 更为详细的介绍请看 TODO....

partition 信息

展示当前 Broker 的 partition 信息, 列出当前 Broker 所有 Topic 的 Leader 和副本以及未同步副本情况;

在上面的 Leader Rebalance 模块中,其实已经说明讲解了一部分这里的信息情况;

例如 Broker-0 宕机了,可以看到那些在 Broker-0 中存在对应副本的 Topic, 清晰的展示了哪些副本是没有同步的; 像下面的 TEST2 在 Broker-0 中不含有副本,所有它的状态是已同步;

Topic 分析

当前 Broker 的 Topic 基本信息,其实这里的信息在 最左边的基本信息里面有了不过这里展示的是最近一分钟的数据,而且把所有 Topic 的数据列出来展示对比;

我们模拟一下批量发送消息,给 TEST2 TEST3 的 TOPIC 发个 1 万条消息

bin/kafka-producer-perf-test.sh --topic TEST3 --num-records 10000 --record-size 100 --throughput 100  --producer-props bootstrap.servers=xxx:9092,xxx:9092,xxx:9092
bin/kafka-producer-perf-test.sh --topic TEST2 --num-records 10000 --record-size 100 --throughput 100  --producer-props bootstrap.servers=xxx:9092,xxx:9092,xxx:9092

复制代码

看看展示的数据

通过这个数据可以看到当前 Broker 下最近一分钟的 Topic 活动状态; 可以看到哪个 Topic 比较活跃; 图中的百分比应该算的有问题,去提一个 BUG;

消费者信息

在这里插入图片描述

展示当前 Broker 下的所有消费组信息, Location 表示数据是从 Broker 上获取的(老版本存放在 ZK 中); 注意刚启动的时候这里可能为空,一分钟之后执行获取 Consumer 的任务才会获取到

Region 信息

Region 列表

展示的是当前物理集群下划分的所有 Region;

我们主要看上面的几个参数预估容量: 很多人对这个数值比较疑惑, 也不知道怎来的; 我们找找源码就知道它是怎么来的了;此数值计算的是当前 Region 下面能够承受的最大流量值 ;比如上面的表示最大支持 360M/s; 但是这个值其实是一个非常模糊的预估值,是需要运维管理人员 去设置的,如果没有设置默认的就是每台 Broker 最大流量值是 120M/s;运维管理人员 需要对自己的 Broker 能够承受的峰值流量有个数; 然后设置完成可以直观的了解到此 Region 是否能够承受住峰值流量;

实际流量: 从历史数据中计算一下实际的峰值流量;

预估流量: 实际流量+ 新申请的 Topic 的预估流量;解释一下; 我们新申请的 Topic，这个时候还没有流量进来, 但是我们要给这个新申请的 Topic 预留一个 Buffer; 我们在申请 Topic 的时候不是有让填写一个预估峰值流量么; 但是当前代码里面实际流量=预估流量; 待优化

那么如何修改 Broker 能够承受的峰值流量呢?

点击 运维管控 ->平台管理->平台配置 填写如下信息

配制键: REGION_CAPACITY_CONFIG配置值 Json 串; 它是一个 array


[{ "clusterId": 4, //物理集群的ID "duration": 10, //持续时间,为了最大值最小值对实际流量产生的紊乱，才有这么一个值,具体含义就不分析了,默认值就是10 "latestTimeUnitMs": 604800000,// 表示计算的是最近多少天内的数据;比如这个默认值是7天;7 * 24 * 60 * 60 * 1000L "maxCapacityUnitB": 125829120 //预估容量;默认值是120 * 1024 * 1024 ;也就是120M; 针对的是单台Broker}, { "clusterId": 5, "duration": 10, "latestTimeUnitMs": 604800000, "maxCapacityUnitB": 125829120}]

复制代码

PS: 上面的配置每个都是针对物理集群下面的所有 Broker; 比如我设置的 clusterId=4 的物理集群的 maxCapacityUnitB=125829120;(120M),那么这个物理集群下面的所有 Region 下面的 Broker;给的预估容量都是 120M

上面的计算是每隔 12 小时才会计算一次;

针对这一块,后续社区应该会做优化改造,或者让预估容量可以自动计算平台配置那里也不方便; 或者社区也会做修改

逻辑集群信息

逻辑集群列表

展示当前物理集群的所有逻辑集群信息;

创建逻辑集群讲解请看 【KafkaManager 二】集群的接入及相关概念讲解

Controller 信息

展示 Controller 的变更历史和设置候选 Controller 关于 Controller

控制器组件（Controller），是 Apache Kafka 的核心组件。它的主要作用是在 Apache ZooKeeper 的帮助下管理和协调整个 Kafka 集群。集群中任意一台 Broker 都能充当控制器的角色，但是，在运行过程中，只能有一个 Broker 成为控制器，行使其管理和协调的职责

更为详细内容请参考Kafka的Controller Broker是什么

在这里插入图片描述

设置了候选 Controller 之后 : Controller 将会优先从选中的 Broker 中选举 ; 这个功能使用的场景可能是你知道哪几台 Broker 比较空闲 , 想让他们承担 Controller 的责任;

限流信息

这里展示的是当前物理集群中此时此刻正在被限流的所有 Topic 信息;

还记得我们上一篇文章有也有讲过限流的相关么 【KafkaManager 三】kafka针对Topic粒度的配额管理(限流)

那里是查看当前的 Topic 是否被限流了

关于 kafka 的配额限流 kafka中的配额管理（限速）机制

那么我们这里的限流信息怎么看呢?什么时候出来呢？

那我们来制造一个 Producer 发生限流的场景;

1.设置一个限流配置

  // 添加限流信息sh bin/kafka-configs.sh --bootstrap-server broker1:9092 --alter --add-config 'producer_byte_rate=100,consumer_byte_rate=100' --entity-type clients --entity-name appId_000001_cn.Test2

复制代码

上面的命令的意思是在broker1:9092上添加一个针对客户端clientName = appId_000001_cn.Test2 加上一个限流配置;生产者 producer 的速率是 100b/s ; 消费组 consumer 的速率是 100b/s ;

不放心我们也可以去 zk 上看看是不是配置成功了

2.生产消息

@Test    void contextLoads() throws InterruptedException {            Properties props = new Properties();            props.put("bootstrap.servers", "broker1:9092");            props.put("acks", "all");            props.put("retries", 0);            props.put("batch.size", 110);            props.put("linger.ms", 0);            props.put("client.id", "appId_000001_cn.Test2");            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");            Producer<String, String> producer = new KafkaProducer<>(props);            for(int i = 0; i < 1000; i++){                //将一个消息设置大一点                byte[] log = new byte[100];                String slog = new String(log);                producer.send(new ProducerRecord<String, String>("Test2",0, Integer.toString(i),  slog));                System.out.println("i="+i);            }        try {            Thread.sleep(62000);        } catch (InterruptedException e) {            e.printStackTrace();        }        producer.close();    }

复制代码

上面的代码表示每次发送 100b 的消息出去,并且是立即发送; 因为我们设置的限流速度是 100b/s；那么妥妥的就被限流了嘛;

注意：客户端 id 设置的为 appId_000001_cn.Test2 ;跟我们上面针对的客户端限流名称一样才会生效;

执行代码之后我们再看看效果;

限流的 Topic 就展示出来了,当然这个展示的是当前限流的；等它不限流了就会消息;

PS：这里有个要注意的地方就是,这里展示的是针对单个 Topic 的限流信息; 我们知道 kafka 当前是不支持针对 Topic 这一维度来进行限流配置的; 当然想要自己实现针对 Topic 限流也很简单,只需要让每个 Topic 的 client.id 不一样；然后针对每个 topic 的 client.id 做限流配置就行; 看上面我设置的客户端是 appId_000001_cn.Test2 这样的格式; 但是自己这样去做非常麻烦；不建议自己去做; 上篇文章有讲过 【KafkaManager 三】kafka针对Topic粒度的配额管理(限流)

如果只是开源版本的话这一块功能还是用不上了(自己做麻烦主要是)

不过滴滴的kafka-gateway 是支持这个功能的; 但是kafka-gateway 是滴滴的商业服务,暂未开源; kafka-gateway 在原生的 kafka 上做了很多的增强; 想要使用kafka-gateway的开源联系滴滴官方