Kafka Consumer Lag 모니터링, Burrow를 알아보자 (1)

카프카 모니터링을 적용 하다보면 브로커는 JMX만으로 충분하지만 Producer나 Consumer는 까다롭습니다.

https://docs.confluent.io/current/kafka/monitoring.html#server-metrics

Monitoring Kafka — Confluent Platform

위 링크에는 JMX를 이용해서 Producer, Consumer metric을 볼 수 있다고 나오지만, 현실적으로는 어려움이 있습니다.
(Kafka Client를 JVM 언어로 한정시키기, Producer/Consumer가 새롭게 뜨거나 없어질 때마다 JMX 모니터링을 위한 추가 작업)

 

지금까지 Kafka를 운영하면서 Broker 모니터링만 했는데 시간이 지나면서 Kafka를 서비스에 이용하는 사례도 생기고

활용 빈도가 높아졌습니다.

앞에서 얘기한 Client 모니터링이 필요해져서 Burrow를 리서치했고 현재는 적용 중에 있습니다.

 

리서치하면서 알게 된 Burrow의 동작원리를 간단히 살펴보고 Consumer 리스트 조회시 오래전 종료한 Consumer도 보여지는
이슈(?)에 대해서 알아보겠습니다.

 

들어가기 전에

토픽 별 메시지 인입량을 Producer 모니터링으로(?) 생각한 이유는 Producer 프로세스 상태는 관심 없고, 정상 동작한다면

트래픽이 올라가면 해당 토픽으로 인입되는 메시지가 증가할 것이고 트래픽이 감소하면 인입량이 감소할 것으로 생각했습니다.
동일하게 Consumer Lag을 Consumer을 Consumer 모니터링으로 생각한 이유는 Consumer 프로세스의 상태는 관심 없고,

정상 동작한다면 Consumer Lag이 허용범위 내에서 일정하게 유지할 것으로 생각했습니다.

---

Burrow의 구조

Kafka Consumer V0.9 이상부터는 offset 정보를 __consumer_offsets 토픽에 produce 합니다.

즉 해당 토픽을 consume 하고 있으면 어떤 Consumer group id가 어느 offset 까지 처리했는지 알 수 있습니다.

 

Burrow는 __consumer_offsets 토픽을 consume 해서 다양한 정보를 가공 및 저장하고

HTTP로 조회가 가능하도록 만든 프로그램입니다.

※ __consumer_offsets을 처음 보셨다면 카프카 미니 밋업 때 발표했던 자료를 참고해주세요.

※ Consumer에서 MaxLag 지표를 제공하지만 Burrow에서는 사용하지 않는 이유: Why Not MaxLag?

 

Burrow module!!

Burrow 1.1이 되면서 기능별로 module이 나눠졌습니다.

core/internal 밑에 폴더별로 coordinator.go가 존재하고 기능에 맞게 Start, Stop, Configure method가 구현되었습니다.

Burrow가 실행되면
kafka cluster, consumer, evaluator, zookeeper, storage, notifier, httpserver 총 7개의 모듈이 초기화됩니다.

 

해당 글에서는 Topic 리스트와 토픽 별 offset을 가공하는 cluster module과 __consumer_offsets 토픽을 consume 하면서
Consumer group id 별 offset, Lag을 가공하는 consumer module을 알아보겠습니다.

cluster module에는 metadataTicker(Topic 리스트 가공)와 offsetTicker(토픽 별 offset 가공)가 존재해서 일정 간격으로 

메타 데이터 조회 한 후 storage module을 이용하여 메모리에 저장합니다.

※ default: meta refresh 120 초, offset refresh 30 초

 

consumer module은 __consumer_offsets 토픽을 consume 하면서 fetch 한 메시지를

storage module을 이용하여 메모리에 저장합니다.

fetch 한 메시지에는 Cluster 정보, Topic 명, Partition 번호, Consumer Group Id 명, Timestamp, Offset이 있고
이 정보를 이용하여 Consumer 리스트, Consumer Detail 데이터를 가공합니다.

 

---

 

이미 종료된 Consumer Group이 조회되는 혼돈의 카오스.

페북 글, (저도 작년에 테스트했는데 멘붕 와서 Burrow 지웠던 기억이...)

 

 

문제

Consumer 리스트를 조회하면 오래전 종료한 consumer group id가 보인다.

오래전 종료한 consumer group id의 detail 정보를 조회하면 404가 보이고 다시 Consumer 리스트를 조회하면

오래전 종료한 Consume group id는 안 보인다.

 

원인

__consumer_offsets 토픽을 consume 해서 fetch한 각 메시지는 다음과 같은 로직이 적용됩니다.
if 문에 안 걸린다면 Burrow memory에 저장됩니다.

// inmemory.go
func addConsumerOffset(...) {
  ...
  if request.Timestamp < ((time.Now().Unix() - module.expireGroup) * 1000) {
	requestLogger.Debug("dropped", zap.String("reason", "old offset"))
	return
  }
  ...
  //해당 Kafka 클러스터의 consumer group 정보 갱신
}

계속 동작하고 있는 Consumer는 위 로직이 적용되어 Burrow 메모리에 반영되겠지만 종료된 Consumer의 경우 삭제하는 로직은
Consumer detail 조회하는곳에 있습니다!!!!

/v3/kafka/{cluster name}/consumer/{consumer group name} API가 호출되면 아래 로직이 적용되는데

이때 expireGroup과 lastCommit을 비교해서 오래됐다고 판단되면 해당 Consumer 정보를 메모리에서 삭제합니다.

// inmemory.go
func fetchConsumer(...) {
  ...
  if ((time.Now().Unix() - module.expireGroup) * 1000) > consumerMap.lastCommit {
    // 오래된 Consumer라고 판단해서 해당 consumer 정보 삭제!!
  }
  ...
}

 

즉 Consumer detail API가 한 번도 호출되지 않는다면 오래전에 종료했더라도 해당 Consumer는 Burrow에서 지워지지 않아서
Consumer 리스트 조회시 보여집니다.

 

burrow.toml -> consumer 설정에
"consumer.[consumer name].start-latest=true"을 추가하면 Burrow 시작될 때 __consumer_offsets 토픽의 가장 최신 것부터

가져와서 잠깐 동안은 현재 접속 중인 Consumer 리스트 조회가 되지만 시간이 지나면 동일한 현상이 일어납니다.

(이미 종료된 Consumer인데 Consumer 리스트 조회 시 보이는 현상)

 

Wiki 첫 줄에 "Burrow is a monitoring tool for keeping track of consumer lag in Apache Kafka."를 곱씹어 봅니다.

 

제가 생각한 차선책은 일정 주기마다(예로 30분) Consumer 리스트를 조회하고 각 Consumer의  detail API를

호출해서 expire-group이 지난 Consumer는 삭제해줍니다.  그리고 사용자에게 Consumer 리스트를 노출할 땐

"최근 30분 이내 접근한(?) Consumer 정보입니다." 라는 멘트를 적어주면 괜찮지 않을까요? ㅎ

 

맨 처음 Burrow를 접했을 때 제가 바랬던 결과와 Burrow에서 내뱉는 결과가 달라서 멘붕이 왔었습니다.(페북 글과 동일)
이번에 조사하면서 왜 그랬는지 알게 되니 과거 Burrow를 욕했던 제가 부끄럽네요....

 

이번 포스트가 Burrow 검토하는 분에게 도움이 되었으면 좋겠습니다.