[카프카 핵심 가이드] 카프카 모니터링하기, 스트림 처리

카프카 모니터링

카프카 브로커 지표

카프카 클러스터에서 발생하는 문제는 보통 하드웨어나 운영체제 수준의 이슈가 아닌 이상 대부분 카프카 클러스터 안에서 요청이 몰려서 발생한다. 카프카는 클러스터 내 데이터를 브로커 간에 최대한 고르게 분포시키려 하지만, 데이터 접근하는 클라이언트가 고르게 분포되지는 않는다. 따라서 외부 툴을 사용해서 클러스터 균형을 유지하는 것을 권장한다. 

 

클러스터 균형을 유지하는 툴에는 크루즈 컨트롤(Cruise Control)이 있는데, 이 애플리케이션은 클러스터를 계속 모니터링하다가 내부 파티션을 리밸런싱해준다. 클러스터의 균형이 잡히더라도, 다수의 브로커가 요청에 대해 높은 지연을 가지거나 요청 핸들러 풀의 유휴 비율이 높다면, 브로커가 처리할 수 있는 트래픽의 한계에 다다른 것으로 과적재에 대해서도 모니터링을 해야한다.

지표 이름	JMX MBBean	설명
불완전 복제 파티션 수	kafka.server:type=ReplicaManager, name=UnderReplicatedPartitions	0 이상 Integer - 브로커 단위 집계 - 단일 브로커가 리더 레플리카를 가지고 있는 파티션 중 팔로워 레플리카가 따라오지 못하고 있는 파티션 수를 나타낸다.
활성 컨트롤러 수	kafka.controller:type=KafkaController, name=ActiveControllerCount	0 또는 1 - 특정 브로커가 현재 클러스터의 컨트롤러 역할을 맞는지 나타냄 - 만약, 1개가 아닌 0개 또는 2개 이상의 브로커가 컨트롤러라고 하면 문제가 있는 것이다.
컨트롤러 큐 크기	kafka.controller:type=ControllerEventManager, name=EventQueueSize	0 이상 Integer - 현재 컨트롤러에서 브로커의 처리를 기다리고 있는 요청 수 - 순간적으로 튈 수 있지만, 계속해서 증가하거나 높아진 상태로 유지된다면 컨트롤러에 문제 발생한 것이다.
요청 핸들러 유휴 비율	kafka.server:type=KafkaRequestHandlerPool, name=RequestHandlerAvgIdlePercent	0 이상 1 이하 Float - 요청 핸들러 스레드는 메시지를 디스크에 쓰거나 읽어 오는 것을 포함한 클라이언트 요청 그 자체 처리를 담당함 - 따라서, 브로커에 많은 부하가 있으면 이 스레드 풀에 영햐을 줌
전 토픽 바이트 인입 지표	kafka.server:type=BrokenTopicMetrics, name=BytesInPerSec	초당 인입률(rate)는 Double, 개수(count)는 Integer - 브로커가 프로듀서 클라이언트로부터 얼마나 많은 메시지 트래픽을 받는지에 대한 측정값 - 이 지표의 변화 추이를 살펴보면 클러스터를 언제 확장할 지 등을 결정하는데 도움을 줌
전 토픽 바이트 유출	kafka.server:type=BrokerTopicMetrics, name=BytesOutPerSec	초당 유출률(rate)은 Double, 개수(count)는 Integer - 컨슈머가 메시지를 읽는 속도를 보여줌 - 유출 바이트 속도는 레플리카에 의해 발생하는 트래픽도 포함되므로, 지표 볼 때 유의해야 함   - 예) 토픽 복제 팩터가 2이고, 컨슈머 클라이언트가 없어도 인입 속도와 유출 속도가 같음
초당 메시지 수	kafka.server:type=BrokerTopicMetrics, name=MessagesInPerSec	초당 개수(rate)는 Double, 개수(count)는 Integer - 메시지 인입 속도는 메시키 크기와 상관 없이, 초당 들어오는 메시지 수를 의미함 - 프로듀서 트래픽만큼 트래픽의 성장을 보여주는 지표라 유용함
오프라인 파티션 수	kafka.controller:type=KafkaController, name=OfflinePartitionsCount	0 이상 Integer - 현재 리더가 없는 파티션 개수를 보여줌

 

브로커 지표는 전체적인 지표 외에도 각 토픽 또는 파티션별 지표도 존재한다. 토픽별 지표는 브로커 지표와 매우 비슷하며, 유일한 차이점은 토픽 이름을 지정해야 한다는 점이다. 파티션별 지표는 토픽별 지표에 비해 덜 유용하다.

종류	이름	JMX MBean
토픽지표	초당 바이트 인입	kafka.server:type=BrokerTopicMetrics, name=BytesInPerSec, topic=TOPICNAME
	초당 바이트 유출	kafka.server:type=BrokerTopicMetrics, name=BytesOutPerSec, topic=TOPICNAME
	초당 실패한 읽기 요청 개수	kafka.server:type=BrokerTopicMetrics, name=FailedFetchRequestsPerSec, topic=TOPICNAME
	초당 실패한 쓰기 요청 개수	kafka.server:type=BrokerTopicMetrics, name=FailedProduceRequestsPerSec, topic=TOPICNAME
	초당 인입 메시지 수	kafka.server:type=BrokerTopicMetrics, name=MessagesInPerSec, topic=TOPICNAME
	초당 읽기 요청 수	kafka.server:type=BrokerTopicMetrics, name=TotalFetchRequestsPerSec, topic=TOPICNAME
	초당 쓰기 요청 수	kafka.server:type=BrokerTopicMetrics, name=TotalProduceRequestsPerSec, topic=TOPICNAME
파티션별 지표	파티션 크기	kafka.log:type=Log, name=Size, topic=TOPICNAME, partition=0
	로그 세그먼트 개수	kafka.log:type=Log, name=NumLogSegments, topic=TOPICNAME, partition=0
	로그 끝 오프셋	kafka.log:type=Log, name=LogEndOffset, topic=TOPICNAME, partition=0
	로그 시작 오프셋	kafka.log:type=Log, name=LogStartOffset, topic=TOPICNAME, partition=0

참고로, 로그 끝 오프셋과 로그 시작 오프셋은 해당 파티션에 속한 메시지 중 가장 높은 오프셋과 낮은 오프셋이지만 이 둘 간의 차이가 해당 파티션 메시지 수를 의미하지는 않는다. 로그 압착으로 인해 동일한 키를 가진 최신의 메시지만 남아 파티션에서 제거된 '사라진' 오프셋들이 있을 수 있기 때문이다.

 

카프카 브로커에서 제공하는 지표 외에도 JVM을 포함한 서버의 측정값 역시 모니터링이 필요하다. GC 사이클에 소요된 시간이나 GC 사이클 수 등을 모니터링할 수 있다. 추가로, MaxFileDescriptorCount와 OpenFileDescriptorCount 지표는 운영체제 정보이지만, JVM이 제공해주는 정보이며 이 지표 역시 모니터링하면 좋다.

• MaxFileDescriptorCount: JVM이 열 수 있는 최대 파일 디스크립터의 수
• OpenFileDescriptorCount: 현재 열려 있는 FD 수로, 모든 로그 세그먼트와 네트워크 연결별로 FD가 열려서 이 값은 빠르게 늘어날 수 있어서 유의해야 함

카프카 프로듀서 지표

프로듀서 종합 지표는 메시지 배치 크기부터 메모리 버퍼 활용까지 모든 속성을 제공한다. 이 지표들은 디버깅에도 사용되지만, 정기적으로 사용이 필요한 것들도 있어서 상시 모니터링 및 알람 설정이 되어있어야 한다.

• record-error-rate: 항상 0이어야 하는 지표
  • 0보다 크다면 프로듀서가 브로커로 메시지 보내는 도중에 누수가 발생하는 것이라 알람을 걸어야 한다.
• request-latency-avg: 브로커가 쓰기 요청을 받을 때까지 걸린 평균 시간
  • 정상 작동 상태에서 이 지표의 기준점을 찾고 그 값보다 큰 값을 경보 제한값으로 설정하면 된다.
  • 요청에 대한 지연이 증가하는 것은 쓰기 요청이 점점 더 느려지고 있다는 것을 의미하며, 네트워크 문제일 수도 있지만 브로커에 문제가 발생했을 수 있기에 알람을 걸어야 한다.
• record-queue-time-avg: 애플리케이션이 메시지를 전송한 뒤 실제로 카프카에 쓰여지기 전까지 프로듀서에서 대기하는 평균 시간
  • 프로듀서는 batch.size만큼 메시지가 채워지거나 마지막 배치가 전송된 이후 linger.ms 시간이 경과되어야 브로커에 메시지를 전송한다.
  • 이 시간을 보면서 batch.size나 linger.ms 값을 튜닝해볼 수 있다.

프로듀서도 브로커처럼 토픽별 지표도 있지만, 단일 프로듀서가 여러 토픽에 메시지를 전송할 때에만 유용하다. 

카프카 컨슈머 지표

• fetch-latency-avg: 브로커로 읽기 요청을 보내는 데 걸리는 시간
• bytes-consumed-rate, records-consumed-rate: 얼마나 많은 메시지 트래픽을 처리중인지 확인할 수 있는 지표

컨슈머는 컨슈머 랙을 반드시 모니터링해야 한다. 하지만, 카프카 컨슈머 지표에서 주는 records-lag-max 속성은 모든 파티션 중 가장 뒤쳐진 파티션의 랙을 보여주기 때문에 정확성이 떨어진다. 따라서 컨슈머 랙은 외부 랙 모니터링을 이용해야 한다.

쿼터

카프카는 하나의 클라이언트가 전체 클러스터를 독차지하는 것을 방지하고자 클라이언트 요청을 스로틀링할 수 있다. 이는 프로듀서와 컨슈머 모두 설정 가능하며, 쿼터 기능을 활성화하지 않았더라도 미리 지표를 모니터링하면 좋다.

랙 모니터링

컨슈머 클라이언트에도 랙 지표가 있지만, 가장 랙이 큰 파티션 기준으로 단일 값만 제공하기 때문에 컨슈머가 전체적으로 얼마만큼 뒤쳐져있는지 정확히 보기 어렵다. 

 

컨슈머 랙 모니터링을 하는 주요 방법은 브로커의 파티션 상태와 컨슈머 상태를 모두 보면서 쓰여진 오프셋과 커밋한 오프셋을 추적하는 외부 프로세스를 두는 것이다. 컨슈머 그룹을 모니터링할 때 발생하는 복잡성을 줄이는 방법 중 하나는 버로우(Burrow)를 사용하는 것이다. 버로우는 링크드인에서 개발된 컨슈머 상태 모니터링용 오픈소스 애플리케이션으로 클러스터 내 모든 컨슈머 그룹의 랙 정보를 가져온 뒤 각 그룹이 제대로 작동하는지를 계산해서 보여준다. 버로우를 사용하는 것은 단일 클러스터 뿐 아니라 다중 클러스터 환경에서도 모든 컨슈머를 모니터링하는 쉬운 방법이 될 수 있으며, 이미 사용중인 모니터링 시스템과도 연동하기 쉽다.

종단 모니터링

카프카 클러스터가 제대로 작동하는지를 보기 위해서 권장하는 또 다른 외부 모니터링은 카프카 클러스터의 작동 상태에 대한 클러스터 관점을 제공하는 종단 모니터링 시스템이다. 링크드인 카프카 팀에서는 오픈소스로 Xinfra Monitor를 제공하며, 이 툴은 클러스터의 모든 브로커에 걸쳐 있는 토픽에 계속해서 데이터를 읽고 쓰면서 브로커별 모니터링을 수행한다.

스트림 처리

데이터 스트림(이벤트 스트림)이란 무한히 늘어나는 데이터세트를 추상화한 것이다. 무한하다는 의미는 끝없이 늘어난다는 의미로 시간이 흐름에 따라 새로운 레코드가 계속해서 추가되므로 데이터세트가 무한하다고 표현한다.

이벤트 스트림 특징

• 이벤트 스트림에는 순서가 있다
  • 이벤트는 그 자체로 다른 이벤트 전/후에 발생했다는 의미를 가진다.
  • 데이터베이스의 테이블 레코드와 다른 점이다. (DB 조회 시 order by 옵션이 있긴 하지만, 이는 관계형DB의 모델의 일부는 아니며 편의에 의해 추가한 컬럼을 기준으로 정렬한 것)
• 데이터 레코드는 불변하다
  • 이벤트는 한 번 발생한 뒤 고칠 수 없다.
  • 데이터베이스도 audit 테이블을 구성할 수 있긴 하지만, 기본적으로 테이블에서는 같은 레코드의 값을 바꾸는 개념이다.
  • 데이터베이스의 redo log와 유사한 개념으로 볼 수 있다.
• 이벤트 스트림은 재생이 가능하다
  • 과거에 발생한 로우 스트림을 그대로 재생할 수 있다. 

프로그래밍 패러다임

• 요청-응답
  • 응답 시간이 1 ms초 미만 ~ x ms초 수준인 패러다임으로 가장 지연이 적은 패러다임이다.
  • 보통 블로킹 방식이라 애플리케이션 요청을 보낸 뒤 처리 시스템이 응답할 때까지 대기하는 게 보통이다.
• 배치 처리
  • 배치 처리는 지연이 크지만, 처리량 역시 크다.
  • 데이터베이스 세계에서 데이터 웨어하우스나 비즈니스 인텔리전스 시스템이 이러한 부류에 속한다.
    • 하루에 한 번 대량의 배치 단위로 적재되고, 리포트 생성 후 새로 배치가 일어나기 전까지는 이전 리포트를 보게 됨
• 스트림 처리
  • 연속적이고 논블로킹하게 작동한다.
  • 스트림 처리는 요청-응답 방식과 배치 처리 사이의 격차를 매꿔준다.

스트림 처리 개념

• 토폴로지
  • 스트림 처리 애플리케이션은 하나 이상의 처리 토폴로지를 포함한다.
  • 하나의 처리 토폴로지는 하나 이상의 소스 스트림, 스트림 프로세서, 하나 이상의 싱크 스트림이 서로 연결된 것이다.
• 시간
  • 카프카 스트림즈는 TimestampExtractor 인터페이스를 사용해서 각각의 이벤트에 시간을 부여햔다. 
  • 시간을 다룰 때는 시간대에 주의해야 한다.
  • 이벤트 시간
    • 다루고자 하는 이벤트가 발생하여 레코드가 생성된 시점이다.
    • 예를 들어, 상품이 팔린 시각/사용자가 웹 페이지를 조회한 시각 등이 될 수 있다.
  • 로그 추가 시간
    • 이벤트가 카프카 브로커에 전달되어 저장된 시점이다.
  • 처리 시간
    • 스트림 처리 애플리케이션이 연산을 수행하기 위해 이벤트를 받은 시각이다.

상태

각각의 이벤트를 따로따로 처리한다면 스트림 프로세싱은 매우 간단하다. 하지만, 스트림 처리는 다수의 이벤트가 포함되는 작업을 하고자 할 수 있고 이때는 많은 정보를 추적하기 위해 상태를 가져야 한다.

• 로컬 혹은 내부 상태
  • 특정 인스턴스에서만 사용할 수 있는 상태다.
  • 주로 애플리케이션에 포함되어 구동되는 내장형 인메모리 데이터베이스를 사용해 유지 관리된다.
  • 장점은 엄청 빠르다는 것이며, 단점은 사용 가능한 메모리 크기에 제한을 받는다는 것이다.
• 외부 상태
  • 카산드라와 같은 NoSQL 시스템을 사용해 저장되는 상태다.
  • 외부 상태의 장점은 크기에 제한이 없고, 여러 애플리케이션에서 접근이 가능하다는 점이다. 단점은, 다른 시스템을 추가하는 데 따른 지연 증가, 복잡도 증가, 가용성 문제가 발생할 수 있다는 것이다.

스트림 처리 디자인 패턴

단일 이벤트 처리 

각각의 이벤트를 개별적으로 처리하는 방식으로, 맵/필터 패턴으로 알려져있다. 맵 단계에서는 이벤트를 변환하고, 리듀스 단계에서는 집계한다고 보면 된다.이 패턴에서 스트림 처리 애플리케이션은 스트림의 이벤트를 읽어와서 이벤트를 수정한 뒤, 수정된 이벤트를 다른 스트림에 쓴다.

로컬 상태와 스트림 처리

정보의 집계에 초점을 맞춘 스트림 처리 애플리케이션은 스트림의 상태를 유지할 필요가 있다. 예를 들어, 주식의 일별 최저가와 평균가를 계산하기 위해서는 최소값과 총합 레코드 수를 저장해야 한다.

로컬 상태를 보유할 때 고려해야 할 사항은 아래와 같다.

• 메모리 사용
  • 로컬 상태는 애플리케이션 인스턴스가 사용 가능한 메모리 안에 들어갈 수 있는게 이상적이다.
• 영속성
  • 애플리케이션 인스턴스가 종료되어도 상태가 유실되지 않고, 재실행되거나 다른 인스턴스에 의해 대체되었을 때 복구될 수 있어야 한다.
  • 카프카 스트림즈는 내장된 RocksDB를 사용함으로써 로컬 상태를 인메모리 방식으로 저장함과 동시에 재시작 시 빠르게 복구 가능하도록 디스크에 데이터를 영속적으로 저장해야 한다.
• 리밸런싱
  • 파티션은 이따금 서로 다른 컨슈머에 다시 할당될 수 있다. 재할당 시, 파티션을 상실한 애플리케이션 인스턴스는 마지막 상태를 저장해 해당 파티션을 할당받은 인스턴스가 재할당 이전 상태를 복구시킬 수 있도록 해야 한다.

다단계 처리/리파티셔닝

그룹별 집계를 할 때는 로컬 상태를 사용하면 충분하다.하지만, 그룹별이 아닌 모든 정보를 사용해 집계를 해야할 때는 로컬 상태만으로는 해결이 되지 않고 두 단계로 접근을 해야 한다. 

 

우선 그룹별 집계를 한 뒤, 하나의 파티션만 가진 새로운 토픽에 결과를 쓴다. 그 후 이 새로운 토픽을 하나의 애플리케이션 인스턴스에서 읽어서 집계를 하는 것이다.

스트림-테이블 조인

스트림 처리를 할 때 외부 데이터를 스트림과 조인해야 할 수도 있다. 예를 들어 거래 내역을 데이터베이스에 저장된 규칙을 사용해 검증하거나 사용자 클릭 내역을 클릭한 사용자 정보와 합쳐 확장하는 상황등이 있다.

위 구조처럼 클릭 이벤트가 발생해 스트림으로 들어올 때 데이터베이스를 찾아서 검증하고 새로운 토픽에 데이터를 쓸 수도 있지만, 외부 검색이 각각의 레코드 처리하는 데 있어서 상당한 지연을 발생시킨다는 문제가 있다.

성능과 가용성을 만족하기 위해서는 스트림 처리 애플리케이션 내 데이터베이스에 저장된 데이터를 캐싱할 필요가 있다. 하지만, 캐시를 관리하는 비용이 크기 때문에 아래와 같은 방식을 해볼 수 있다.

데이터베이스 테이블에 변경이 일어나면, 그 변경을 이벤트 스트림에 담고, 스트림 처리 작업은 이 스트림을 통해 로컬 캐시를 업데이트할 수 있다.데이터베이스의 변경 내역을 이벤트 스트림으로 받아오는 것을 CDC(change data capture)라고 하며, 카프카 커넥트에는 CDC를 수행해 데이터베이스 테이블을 변경 이벤트 스트림으로 변환할 수 있는 커넥터가 여럿 있다.

스트리밍 조인

스트림과 테이블이 아닌, 두 개으 이벤트 스트림을 조인해야 할 경우도 있다. 테이블과 스트림을 조인하는 것과 다르게 스트림과 스트림 조인할 때에는 같은 시간 윈도우에 발생한 이벤트끼리 맞춰야 한다.

비순차 이벤트

IOT 환경에서는 잠시 WiFi 신호가 끊긴 모바일 장치가 재접속할 때 과거 이벤트를 한번에 전송하는 등 비순차 이벤트가 발생할 수 있다. 스트림 애플리케이션은 이러한 상황을 처리할 수 있어야 한다. 비순차 이벤트 대응을 하기 위해서는 아래와 같은 일을 해야 한다.

• 이벤트가 순서를 벗어났음을 알아차릴 수 있어야 한다.
• 비순차 이벤트의 순서를 복구할 수 있는 시간 영역을 정의해야 한다.
• 순서를 복구하기 위해 이벤트를 묶을 수 있어야 한다.
• 결과를 변경할 수 있어야 한다.