[k8s] Kafka & Strimzi Operator

안녕하세요. 최근 Database Operator In Kubernetes study(=DOIK)를 진행하며 해당 내용을 보다 깊이 이해하고 공유하기 위해 작성한 글입니다.

전편 [k8s] Operator & MySQL Operator 에 이어 Kafka & Strimzi Operator에 대해 알아보겠습니다.

• [참고] Apache Kafka and Stream Processing O’Reilly 도서

Kafka 란?

비동기 방식의 대표 스트리밍 플랫폼이자 분산 데이터 스트리밍 플랫폼입니다. 많이 쓰이고 있는 메시지 브로커(Rabbit MQ)와 이벤트 브로커(Kafka) 는 어떠한 공통점과 차이점이 있을까요?

• 공통점 : 메시지 브로커(Rabbitmq)와 이벤트 브로커(Kafka) 모두 pub/sub 기반의 메시지 큐 서비스를 제공합니다.
• 차이점 : RabbitMQ의 경우 kafka에 비해 컨슈머와 메시지 브로커간의 결합도가 높기 때문에 트래픽이 작으면서 비즈니스가 추후 확장되지 않을 확률이 높다면 RabbitMQ를 사용하는 것이 좋습니다. 하지만 대규모 트래픽이 예상되고, 추후 확장이 예상된다면 kafka를 선택하는 것이 바람직해보입니다.[참고]

Kafka 구성요소

Kafka component interaction

• Kafka Connect : 데이터베이스 같은 외부 시스템과 카프카를 쉽게 연결해줌
• Kafka Cluster : 여러 대의 브로커를 구성한 클러스터를 의미함
• Kafka MirrorMaker : 다중 클러스터 환경에서 클러스터 간 리플리케이션(미러링)을 구성해줌
• Kakfa Bridge : 카프카 클러스터와 internal & external HTTP client applications 를 연동해줌
• Kafka Exporter : 모니터링을 위해 추가 Kafka 메트릭 데이터를 추출함
• Zookeeper : 카프카의 메타데이터 관리 및 브로커의 정상 상태 점검 health check 담당

Kafka brokers and topics

• Broker : 서버 또는 노드라고도 하는 Broker는 메시지의 저장 및 전달을 조정합니다.
• Cluster : Broker 인스턴스 그룹입니다.
• Topic : Topic은 데이터 저장을 위한 대상을 제공합니다. 각 Topic은 하나 이상의 파티션으로 분할됩니다.
• Partition : Topic 파티션의 수는 Topic 파티션 수로 정의됩니다
• Partition leader : Topic에 대한 모든 생산자 요청을 처리합니다.
• Partition follower : Partition leader의 파티션 데이터를 복제하고 선택적으로 소비자 요청을 처리합니다.

Kafka 기본동작

Kafka는 데이터를 받아서 전달하는 데이터 버스(data bus)의 역할을 합니다. Kafka에 데이터 (메시지)를 만들어서 주는 쪽은 프로듀서(producer)라 부르고, 데이터를 빼내서 소비하는 쪽은 컨슈머(consumer)라 합니다.

Producers and consumers

• Producer : 카프카로 메시지를 보내는 역할을 하는 클라이언트를 총칭
• Consumer : 카프카에서 메시지를 꺼내가는 역할을 하는 클라이언트를 총칭

Strimzi

쿠버네티스 환경에서 카프카 Kafka 운영 관리에 도움을 주는 Operator 입니다. Strimzi는 아래와 같은 기능을 제공합니다.

• 카프카 클러스터/구성요소 배포 및 관리
• 카프카 접속 설정
• 카프카 업그레이드
• 브로커 brokers 관리
• 토픽 topic 과 유저 user 생성 및 관리

Operators within the Strimzi architecture

위 그림과 같이 Strimzi가 제공하는 Operator로 클러스터 Operator, 엔티티 Entity Operator, 토픽 Entity, 유저 Entity가 있음을 알 수 있습니다.

Strimzi 설치 & 카프카 클러스터 생성

# (옵션)kube-ops-view 
$> kubectl apply -k DOIK/kubeopsview/ 
$> NODEPIP=$(curl -s ipinfo.io/ip) 
$> NODEPORT=$(kubectl get svc -n kube-system kube-ops-view -o jsonpath="{.spec.ports[0].nodePort}") 
$> echo -e "Kube Ops View URL = http://$NODEPIP:$NODEPORT"  # 네임스페이스 생성 
$> kubectl create namespace kafka  # Repo 추가 
$> helm repo add strimzi https://strimzi.io/charts/ 
$> helm show values strimzi/strimzi-kafka-operator  # 차트 설치 : 마스터노드에 오퍼레이터 파드 설치 
$> printf 'tolerations: [{key: node-role.kubernetes.io/master, operator: Exists, effect: NoSchedule}]\n' | \ helm install kafka-operator strimzi/strimzi-kafka-operator --version 0.29.0 --namespace kafka \   --set nodeSelector."kubernetes\.io/hostname"=k8s-m --values /dev/stdin  # 배포한 리소스 확인 : Operator 디플로이먼트(파드) 
$> kubectl get deploy,pod -n kafka  # 오퍼레이터가 지원하는 카프카 버전 확인 : 3.0.0 ~ 3.2.0 
$> kubectl describe deploy -n kafka | grep KAFKA_IMAGES: -A4# 배포한 리소스 확인 : CRDs - 각각이 제공 기능으로 봐도됨! 
$> kubectl get crd

테스트용 파드 생성 후 Kafka 클러스터 정보 확인

# myclient 파드 3대 배포 : envsubst 활용 
$> cat ~/DOIK/3/myclient.yaml 
$> for ((i=1; i<=3; i++)); do PODNAME=myclient$i envsubst < ~/DOIK/3/myclient.yaml | kubectl apply -f - ; done# myclient 파드들 확인 
$> kubectl get pod -l app=myclient# Kafka client 에서 제공되는 kafka 관련 도구들 확인 
$> kubectl exec -it myclient1 -- ls /opt/bitnami/kafka/bin# 카프카 파드의 SVC 도메인이름을 변수에 지정 
$> SVCDNS=my-cluster-kafka-bootstrap.kafka.svc:9092# 브로커 정보 
$> kubectl exec -it myclient1 -- kafka-broker-api-versions.sh --bootstrap-server $SVCDNS

Topic 생성

 

# 토픽 Topic 생성 (kubectl native) : 파티션 1개 리플리케이션 3개, envsubst 활용 
$> cat ~/DOIK/3/mytopic.yaml 
$> TOPICNAME=mytopic1 envsubst < ~/DOIK/3/mytopic.yaml | kubectl apply -f - -n kafka# 토픽 생성 확인 (kubectl native) 
$> kubectl get kafkatopics -n kafka

메시지 주고 받기

# 토픽에 데이터 넣어보기 
$> kubectl exec -it myclient1 -- kafka-console-producer.sh --bootstrap-server $SVCDNS --topic mytopic1 
> hello 

CTRL+D 로 빠져나오기# 토픽 데이터 확인 
$> kubectl exec -it myclient2 -- kafka-console-consumer.sh --bootstrap-server $SVCDNS --topic mytopic1 --from-beginning 
hello 

CTRL+C 로 빠져나오기# 토픽에 데이터(메시지키+메시지값) 넣어보기 
$> kubectl exec -it myclient1 -- kafka-console-producer.sh --bootstrap-server $SVCDNS --topic mytopic1 --property "parse.key=true" --property "key.separator=:" 
>key1:doik1 

CTRL+D 로 빠져나오기# 토픽에 데이터(메시지키+메시지값) 확인 
$> kubectl exec -it myclient2 -- kafka-console-consumer.sh --bootstrap-server $SVCDNS --topic mytopic1 --property print.key=true --property key.separator="-" --from-beginning 
null-hello 
key1-doik1 

CTRL+C 로 빠져나오기

 

blog migration project

written in 2022.6.12

https://medium.com/techblog-hayleyshim/k8s-kafka-strimzi-operator-17a2166f36f9