Kafka的消费积压监控-Burrow

• 使用kafka, 消费生产的数据是必不可少的, 为不影响业务的正常处理, 对消费过程的积压lag的监控和报警就显得特别重要
• Kafka的lag监控工具有若干个:
  1. KafkaOffsetMonitor: 应试是很久没更新了;
  2. KafkaManager: 用于管理还可以, 监控报警的话需要自己添加少量接口实现;
  3. Burrow: 个人觉得是目前为止最好用的lag监控报警工具;

Burrow 功能简介

• Burrow的github主页(https://link.jianshu.com/?t=(https://github.com/linkedin/Burrow%29)已经对其从编译到配置使用作了很好的说明, 这里不累述;
• Burrow用Golang写成, 代码不多, 很容易读, 扩展性也很多;
• 使用Burrow作监控, 不需要预先设置lag的阈值, 他完全是基于消费过程的动态评估;
• 可以监控offset提交到broker,zk两种方式,还可以作storm的消费监控, 这部分扩展起来也很容易;
• 报警支持http, email什么的, 想要扩展个自己的短信报警什么的也是超简单, 好用的不要不要的~~~
• Burrow还贴心的提供了http接口,来获取整个集群的生产,消费等情况, 可参见wiki
• 哎呀, 我去, 又一次贴心提供了Docker镜像, 开箱即用啊~~~

Burrow 实现简介:

• 对lag情况进行报警, 当然首先需要获取各group的消费的topic的各个partition的broker offset,就是实际生产的msg的条数, 通过sarama可以轻松获取, 当然这个需要周期性不间断获取;
• 有了broker的offset, 还需要消费的commited offset, 针对kafka 0.9及以后的版本, 提交的offset可以选择保存在broker上的__consumer_offsets的内部topic上, Burrow还是通过sarama来消费__consumer_offsets这个topic来获取;
• 还有些offset是提交到zk上, Burrow也支持从zk上来获取这个 committed offsets;
• Broker offset有了, Committed offset也有了, 剩下的就是应用各种策略来评估各个group的消费情况啦, 每个group可以消费多个topic, 每个topic也有多个paritition, 针对每个partition都有一个凭估周期的概念, 一个凭估周期包括若干个凭估窗口, 每个凭估窗口都是对broker offset和committed offset的一次采样, 然后将策略应用到这个凭估周期内, 最后作出凭估, 这个策略兼顾了broker offset的变化, committed offset的变化, 具体参考wiki上的策略;
• 策略不是完美的, 我们也可以根据自身的需要增加,修改策略.

Burrow 使用中遇到的问题:

• Burrow只能监控在Burrow运行后提交过offset的group, 因为在通过sarama消费__consumer_offsets这个topic来获取committed offset时,设置了OffsetNewest,每次都是从最新开始消费, 我也尝试过改成从最旧开始消费 ,但sarama会run很多的thread起来, 撑爆了系统, 不知道是不是sarama的bug;
• 不支持topic扩展的新的partition的监控, 后来我发现最新版的Burrow里已经修了这个问题,看这里, 但是这个修复只支持了新增的partition的broker offset的获取, 并没有支持committed offset的获取,可以在addConsumerOffset里加上下面这段:

if int(offset.Partition) >= len(consumerTopicMap) {
        // The partition count must have increased. Append enough extra partitions to our slice
        for i := len(consumerTopicMap); i < partitionCount; i++ {
            consumerTopicMap = append(consumerTopicMap, nil)
        }
}

• 偶尔遇到过针对map的读写导致竞争问题:fatal error: concurrent map read and map write, 查了下是在evaluateGroup中对clusterMap.broker的读操作和在addBrokerOffset中对其写操作引发, 加锁吧~,可以拷贝一份clusterMap.broker来读.