[DOIK2] 쿠버네티스 오퍼레이터 @Kafka

 

- cloudnet에서 주관하는 쿠버네티스 데이터베이스 오퍼레이터 스터디 2기 내용입니다

- 목적: 다양한 db 오퍼레이터 실습으로 eks 환경에서 db 배포 및 운영을 위한 다양한 db 오퍼레이터 실습

이번주에는 Kubernates 환경에서 Kafka를 배포하고 관련 내용을 조금 정리해 보았다.

 

 

0. Broker란?

publish/subscribe messaging 플랫폼은 스트리밍해야 하는 이벤트가 여러 개일 때 유용하다. 위 사진과 같은 경우 각각을 연결한다면 최대 7*4=28개의 connection 이 필요하지만 중앙의 Pub/Sub 서버로 모아서 뿌려주면 11개의 connection만 필요하다. 

 

메시징 플랫폼은 이벤트 브로커와 메시지 브로커로 나뉜다. 메시지 브로커보다 이벤트 브로커가 지원하는 기능이 많다.

 

- 메시지 브로커(대표: RabbitMQ): publisher가 생산한 메시지를 저장했다가, consumer가 가져갈 수 있게 한다. 가져가고 난 이후에는 데이터가 빠르게 삭제된다(스트리밍)

- 이벤트 브로커(대표: kafka): 메시지 브로커+publisher가 생산한 이벤트를 저장해서 가져가고 난 이후에도 데이터와 이벤트를 홀드하고 있다. 따라서 장애 발생시 처음부터 다시가 아니라 장애 발생시점부터 다시 처리를 시작할 수 있다.

 

1. Kafka란?

카프카를 처음 접하는 눈높이에서 쓴 내용입니다. 잘못된 부분이 있다면 고칠 수 있게 알려주시면 감사하겠습니다 :)

용어 정리

- Zookeeper: 메타데이터를 관리하는 코디네이터 역할의 애플리케이션. 의존성을 없애고 Kraft로 갈아탄다고는 하는데... 아직까지는 불안정한 듯

- 프로듀서: 카프카로 메시지를 보내는 역할의 클라이언트

- 컨슈머: 카프카에서 메시지를 받는 역할의 클라이언트

- 토픽: DB의 테이블과 비슷한 개념

- 파티션: DB의 샤딩과 비슷한 개념(늘릴 수는 있지만 줄일 수는 없음에 주의!)

- 메시지/레코드: 프로듀서->브로커, 브로커-> 컨슈머 전달되는 데이터 조각의 단위

- 클러스터: 브로커들의 모음

 

카프카의 장점

- 쓰기 특화: 토픽 하나를 여러개로 나눠 병렬 처리가 가능하다. 또한 나뉜 파티션의 수만큼 컨슈머를 연결할 수 있다

- 확장성: 브로커 스케일 아웃이 가능하다

- 영속성: 타 메시징 플랫폼과 다르게 전송받은 데이터를 메모리가 아닌 파일 시스템에 저장한다

- 고가용성: 팔로워 파티션이 리더 파티션의 데이터를 계속 복제하고 있기 때문에 장애시에도 지속적인 처리가 가능하다

 

2. 카프카의 복제와 병렬처리

- 카프카의 데이터 복제는 파티션 단위로 이루어진다

- 팔로워 파티션은 리더 파티션의 오프셋과 자신의 오프셋을 비교하여 차이가 나면 복제한다

- ISR(In-Sync Replicas): 리더-팔로워 파티션이 모두 싱크된 상태

 

3. KafKa Strimizi

- 쿠버네티스 환경에서 카프카 운영 관리에 도움을 주는 오퍼레이터

- 제공 기능: 카프카 클러스터/구성요소 배포 및 관리, 카프카 접속설정 및 업그레이드, 브로커 관리, 토픽 및 유저 생성 관리

https://strimzi.io/docs/operators/latest/quickstart.html#key-features-operators_str

- 이 그림의 모든 걸 Strimizi로 해결할 수 있다

 

 

 

4. 실습

Strimizi Operator 설치

- 0.38.0버전 사용

# kafka를 설치할 namespace 생성
kubectl create namespace kafka

#helm repo에서 strimzi 를 받아온다
helm repo add strimzi https://strimzi.io/charts/

# 차트를 설치하면 오퍼레이터 파드가 설치된다
helm install kafka-operator strimzi/strimzi-kafka-operator --version 0.38.0 --namespace kafka

# helm chart 내용 확인
helm show values strimzi/strimzi-kafka-operator

 

- helm show values로 차트의 내용을 확인하면 아래 사진과 같이 strimzi가 어떤 컴포넌트로 구성되어 있는지, 리소스가 얼마나 할당되어 있는지 등을 확인할 수 있다.

 

- 그리고 받아온 chart의 내용을 바탕으로 strimzi를 설치하면 성공한다.

- 이제 배포가 잘 되었는지 확인해보자.

# 배포 확인
kubectl get deploy,pod -n kafka

# 오퍼레이터 호환 카프카 버전 확인
kubectl describe deploy -n kafka | grep KAFKA_IMAGES: -A3

# 배포한 리소스 확인
kubectl get crd | grep strimzi

# CRD 상세정보 확인
kubectl describe crd kafkas.kafka.strimzi.io
kubectl describe crd kafkatopics.kafka.strimzi.io

 

- 배포 이후 kafka namespace 안의 deploy, pod 상태를  확인하면 cluster operator 1개, pod 1개가 돌고 있음을 볼 수 있다.

- 우리가 방금 사용한 0.38.0 버전 strimzi는 3.5.0, 3.5.1, 3.6.0 버전 kafka와 호환됨을 확인할 수 있다. 참고로 -A3 옵션은 해당 내용 중 3줄을 표시하라는 옵션이다(기본 1개만 반환)

 

- 배포한 CRD를 확인해보면 다양한 기능이 있다. 각 CRD는 각각의 제공 기능이다. 중요한 것 몇개만 소개해보면

 

kafkabridge:카프카 클러스터와 외부 https client application을 연결해줌

kafkaconnectors: DB 등의 외부 시스템과 카프카를 쉽게 연결해줌

kafkamirrormaker: 다중 클러스터 환경에서 클러스터 간 복제 구성 (mirrormaker 2.0은 양방향 토픽 replication을 지원한다)

 

 

- 그러면 이제 카프카 클러스터를 배포해보자. 지금까지 한 건 뭐냐고? 카프카 클러스터를 배포하기 위한 오퍼레이터 strimzi를 배포한 것이고, 이제 본격적으로 카프카 클러스터를 배포한다. 

#.yaml 다운로드
curl -s -O https://raw.githubusercontent.com/gasida/DOIK/main/strimzi/kafka-1.yaml

# 내용 확인
cat kafka-1.yaml | yh

# 배포
kubectl apply -f kafka-1.yaml -n kafka

 

- 스터디장 가시다님이 올려주신 .yaml파일을 받아와서 방금 만든 kafka namespace에 적용한다. 3개의 브로커로 구성된 카프카와 이를 컨트롤하기 위한 주키퍼 3개로 구성되어 있다.

# 배포된 전체 리소스 확인
kubectl get-all --namespace kafka

# 배포된 리소스 확인
kubectl get kafka -n kafka
kubectl get cm,secret -n kafka

# 배포된 pod가 statefulset이 아니라 strimzipodset으로 생성되었음을 확인
kubectl get strimzipodsets -n kafka

# 배포된 pod중 kafka, zookeeper, mycluster 가 다 있는지 확인
kubectl get pod -n kafka -l app.kubernetes.io/name=kafka
kubectl get pod -n kafka -l app.kubernetes.io/name=zookeeper
kubectl get pod -n kafka -l app.kubernetes.io/instance=my-cluster

# config map 확인
kubectl get cm -n kafka

# 배포된 저장소 볼륨 확인
kubectl get pvc,pv -n kafka

 

- 전체 네임스페이스를 확인하면 오래전에 생성된 것 -> 최근에 생성된 순서로 정렬된다. 순서를 보면 클러스터 -> 주키퍼 -> pod -> secret -> service -> replicaset -> strimzipodset -> endpoint 순으로 생성됨을 볼 수 있다. 

 

 

kafka에 배포된 config map과 secret이 어덯게 구성되어 있는지 확인할 수 있다.

 

- 이렇게 endpoint를 확인하면 어싸인한 서비스가 다 잘 생성되었음을, 그리고 접속할 때 사용할 엔드포인트를 확인할 수 있다.

 

테스트용 파드 생성을 위해 마찬가지로 가시다님께서 올려주신 .yaml 파일을 다운받아 적용한다. 

# yaml download
curl -s -O https://raw.githubusercontent.com/gasida/DOIK/main/strimzi/myclient.yaml

# 데몬셋으로 파드 배포
VERSION=3.6 envsubst < myclient.yaml | kubectl apply -f -

# client 가 제공하는 도구들 확인
kubectl exec -it ds/myclient -- ls /opt/bitnami/kafka/bin

# 카프카 파드의 도메인명을 변수로 지정(자주 사용하므로)
SVCDNS=my-cluster-kafka-bootstrap.kafka.svc:9092
echo "export SVCDNS=my-cluster-kafka-bootstrap.kafka.svc:9092" >> /etc/profile

# 브로커 정보 확인
kubectl exec -it ds/myclient -- kafka-broker-api-versions.sh --bootstrap-server $SVCDNS

 

이 bash file들이 모두 클라이언트가 제공하는 정보들이다.

확인할 수 있는 브로커의 상태정보이다. 

 

토픽 생성 및 메시지 주고받기

토픽 생성

# .yaml 파일 다운로드 및 내용 확인
curl -s -O https://raw.githubusercontent.com/gasida/DOIK/main/3/mytopic.yaml
cat mytopic.yaml | yh

# .yaml 파일 적용해서 상호 통신 실습을 위해 토픽 2개 생성
# mytopic1
TOPICNAME=mytopic1 envsubst < mytopic.yaml | kubectl apply -f - -n kafka
# mytopic2
kubectl exec -it ds/myclient -- kafka-topics.sh --create --bootstrap-server $SVCDNS --topic mytopic2 --partitions 1 --replication-factor 3 --config retention.ms=172800000

# 토픽 생성 확인
kubectl get kafkatopics -n kafka


kubectl exec -it ds/myclient -- kafka-topics.sh --bootstrap-server $SVCDNS --list | grep mytopic

# 토픽 상세정보 확인
kubectl exec -it ds/myclient -- kafka-topics.sh --bootstrap-server $SVCDNS --topic mytopic1 --describe

 

- 마찬가지로 가시다님께서 토픽 생성 .yaml 파일을 배포해 주셨다. 파일을 다운로드받고 내용을 확인하면 1개의 파티션에 3개의 복제 set(replica set) 으로 클러스터를 구성하게 되어 있다. 

- 다운로드받은 mytopic.yaml 파일을 MYTOPIC1 환경에 설치한다. 그리고 해당 파일을 앞에서 만들어둔 kafka namespace에서 mytopic 이라는 cluster가 생성됨을 확인한다.

- --describe 명령어를 이용해서 mytopic의 레플리카가 3개이고 이 중 2개에서 insync replica를 2개로 설정했음을 확인한다.

토픽에 메시지 주고받기

# 모니터링 창 띄우기
watch -d kubectl get kafkatopics -n kafka

# mytopic1에 데이터 넣기 (다 넣고 ctrl+D로 입력 종료)
kubectl exec -it ds/myclient -- kafka-console-producer.sh --bootstrap-server $SVCDNS --topic mytopic1

# mytopic1에 있는 데이터 확인
kubectl exec -it ds/myclient -- kafka-console-consumer.sh --bootstrap-server $SVCDNS --topic mytopic1 --from-beginning

 

- 모니터링 창을 띄우면 topic1, topic2, strimizi operator, consumer를 확인할 수 있다.

- kafka-console-producer.sh를 통해서 kafka 에 데이터를 삽입하고, kafka-console-consumer.sh를 통해서 kafka에 삽입되어 있는 데이터를 가져갈 수 있다. 

 

- 이외에도 다양한 API가 존재하고 특정 컨슈머 그룹을 묶어서 관리하거나 메시지 키를 통해서 정해진 파티션에 데이터를 전달하는 방식으로 메시지의 전달 순서를 보장하고, 데이터를 분산할 수 있다.

 

 

 

References

- https://www.youtube.com/watch?v=3F4XwgCfQc8

- https://www.youtube.com/watch?v=c6pJDCdnhbI

- https://www.youtube.com/watch?v=catN_YhV6To&list=PL3Re5Ri5rZmksx3uuv7gU7Mg6fm69y7wh

- https://www.youtube.com/watch?v=H_DaPyUOeTo

- CloudNet 배포 스터디 자료 (감사합니다!)