ctx错误使用导致CPU缓慢持续增大

李天鹏2023年07月26日约 6 分钟

1.背景与问题现象

业务侧在使用内部框架的kafka consumer组件时发现有CPU缓慢增高直至跑满CPU的现象。

业务背景大概为：生产者在其他平台使用其他框架进行消息生产，每分钟约7000条数据，对应topic有30个分区，消费者使用内部框架封装的consumer（对kafka-go的v0.4.38版本做了简单封装）进行消费，消费者部署了6个pod（pod的机器规格是4u8g），每个pod有5个goroutine进行消费。

在消费过程中，发现CPU持续缓慢增加，consumer消费能力不断减小，跟不上生产者速度，导致数据消费不过来。经过大约一天时间后，对应pod的CPU已经跑满，消费能力基本为0，严重影响业务功能，只能进行pod重启，重启后积压数据被瞬间消费完成，持续观察发现一天后现象重复，CPU又被跑满。对应监控图为：

2.问题定位

2.1 观察CPU和内存

根据现有现象，pod中只跑了业务的一个进程，通过top命令发现是对应服务占用CPU过高，同时观察内存信息，发现内存占用不高，并且比较稳定。

2.2 pprof分析

单纯从top命令只能查找到是当前进程导致CPU持续增高，分析具体原因需要借助一些工具。在此使用了golang自带的pprof工具进行分析。

2.2.1 pprof数据采集

在测试环境上，在相应的代码中增加端口监听pprof采集数据。

import _ "net/http/pprof"go func() {    ip := "0.0.0.0:6060"       if err := http.ListenAndServe(ip, nil); err != nil {         fmt.Printf("start pprof failed on %s\n", ip)         os.Exit(1)     }}()

2.2.2 pprof具体分析

在运行一段时间后，CPU持续增加，由于服务在pod上，新开端口6060在安全组是被拦截的，所以不能直接通过ip:6060/debug/pprof的方式在本地浏览器进行直观查看，只能在pod中采集到数据，再下载到本地进行分析。

主要采集数据有：

cpu（CPU Profiling）：$HOST/debug/pprof/profile，默认进行30s的 CPU Profiling，得到一个分析用的profile文件；

goroutine：$HOST/debug/pprof/goroutine，查看当前所有运行的goroutines 堆栈跟踪。

使用go tool pprof http://127.0.0.1:6060/debug/pprof/profile可以生成一个profile文件（此处名称为：pprof.main.samples.cpu.003.pb ），获取后传到本地。

本地通过go tool pprof -http=:8081 pprof.main.samples.cpu.003.pb 生成火焰图，发现CPU主要耗在kafka-go的CommitMessage和FetchMessage两个方法上：

通过go tool pprof pprof.main.samples.cpu.003.pb进行分析，发现CPU主要耗在kafka-go的CommitMessage和FetchMessage两个方法上。

继续使用traces查看CommitMessage的信息，未找到具体异常：

通过list 查看CommitMessage信息，发现主要是在kafka-go的reader.go源码的ctx.Done()和errCh上。

结合使用go tool pprof http://127.0.0.1:6060/debug/pprof/goroutine查看，都是在runtime.gopark上，这个是代表协程处于休眠状态。

2.2.4 开源版本查看

由于框架仅对kafka-go做了简单封装，为此先在github上查看kafka-go开源软件是否有相应的CPU高的issue，同时查看github kafka-go releases最新版本（v0.4.39）和当前版本（v0.4.38）的changes记录，发现并没有相关内容。

2.2.5 源码分析

找到对应版本的kafka-go的源码进行分析。

func (r *Reader) CommitMessages(ctx context.Context, msgs ...Message) error {   if !r.useConsumerGroup() {      return errOnlyAvailableWithGroup   }   var errch <-chan error   creq := commitRequest{      commits: makeCommits(msgs...),   }   if r.useSyncCommits() {      ch := make(chan error, 1)      errch, creq.errch = ch, ch   }   select {   case r.commits <- creq:   case <-ctx.Done():      return ctx.Err()   case <-r.stctx.Done():      // This context is used to ensure we don't allow commits after the      // reader was closed.      return io.ErrClosedPipe   }   if !r.useSyncCommits() {      return nil   }   select {   case <-ctx.Done():      return ctx.Err()   case err := <-errch:      return err   }}

通过上述源码可以看到16-24行中的select，creq在第7行已经定义了，按说应该很快就可以入队到r.commits中，这个代码段应该可以很快执行完，但是为啥会在ctx.Done()这里耗费那么长CPU呢？于是怀疑r.commits这个chan是不是已经满了，导致creq放入不进去，因为代码中的ctx是没有设置超时时间的，一般情况下ctx.Done()肯定是卡住的。

继续分析源码，看r.commits这个chan的大小，在reader.go文件中，chan的大小是由config.QueueCapacity决定的，这个参数在没有设置时，默认是100。

2.2.5 对比分析

考虑到生产者发送的消息每分钟大约有7000条，同时r.commits的大小只有100，怀疑参数配置上问题。

于是把该参数增大到500作为对比方案一做测试。同时使用原生的kafka-go创建不同消费者组对同一个topic进行消费作为方案二进行对比测试。

测试结果发现：

方案一在运行一天后还是会出现CPU缓慢增高，直到占满CPU的情况；

方案二稳定运行一天，并且CPU利用率都在5%以下。

2.2.6 goroutine分析

在CPU高的机器上，通过命令go tool pprof http://127.0.0.1:6060/debug/pprof/goroutine?debug=1继续抓取goroutine的数据，发现在最多的时候只有400多个协程，不算非常高。

通过命令go tool pprof http://127.0.0.1:6060/debug/pprof/goroutine?debug=2查看各协程的运行情况，不存在大量长期未关闭的协程。该命令都是获取程序运行时的快照信息，在某次执行时的goroutine信息中发现：

从该信息来看，该goroutine在运行中，一直在执行context.Done()，这个不就是和之前看到的CommitMessage中的那个执行点是一致的嘛，考虑到上述goroutine中重复的context.(*valueCtx).Done数据有200多行，并且每个ctx的地址都不同，说明这个ctx并不是个单一的ctx，而是在递归叠加而成的。

2.2.7 根因定位

了解到上述情况后，分析业务代码，发现每次在获取消息之前，都会在ctx中写入traceId的value，而写入的过程是类似于：

ctx = context.WithValue(ctx, traceIdKey, traceValue)

由于传入FetchMessage和CommitMessage中的ctx正好是该ctx，而每个ctx都是一串递归的使用，ctx越来越大，链路越来越长，导致后续要拿到当前ctx时，都需要递归到最里层获取其信息，从而导致CPU会缓慢升高，最终占满CPU，而CPU被占满，对应的消费能力肯定也会下降，消息产生堆积。

3.问题复现

提供以下代码可以复现：

package mainimport (   "context"   "fmt"   "time")func main() {   ch := make(chan int, 100)   go func() {      for {         <-ch         time.Sleep(10 * time.Millisecond) // simulate business operation      }   }()   ctx := context.Background()   traceValue := 0   for {      traceValue++      ctx = context.WithValue(ctx, "traceId", traceValue) // value write to ctx      select {      case ch <- traceValue: // value write to chan      case <-ctx.Done():         fmt.Println("ctx done")         return      }   }}

在linux上执行go run main.go，通过top命令查看cpu情况，发现CPU在缓慢增长：

4. 问题处理与结果

通过上述分析之后，处理就比较简单了，如果不需要traceId值，把对应的代码删除即可，如果需要traceId，则在for循环中每次新建一个ctx来设置traceId值即可。最终稳定运行，CPU占用持续在2%以下，对应的监控图如下：

5.主要收获

通过以上问题分析，主要收获点有：

如何更好地使用pprof工具进行CPU/内存分析；
对于开源软件（不限于kafka-go），如果需要查找相关问题，最好是根据源码进行分析；
问题复现时，可以根据多种方案进行对比复现，发现不同点，缩小查找范围。