kubernetes 환경에서 Apache Kafka cluster를 구성해보자 (Operator 패턴)

Kubernetes 환경에서 Apache Kafka cluster 구성하기

​

생각해보니 정작 Kafka에 대한 소개를 한 적 없이 그동안 블로그에 썼다는 언급만 계속 해왔다.

조만간 자세한 내용을 정리해서 올리겠음

​

​

​

​

Apache Kafka on Kubernetes - strimzi Operator

Kafka를 Strimzi operator를 사용해서 운영하는 방법을 찾아서, Helm으로 Apache kafka를 구성해봤다

​

https://strimzi.io/

​

 

왜 Strimzi은 Operator 패턴으로 Kafka를 배포하는가?

1. Operator 패턴이 Kafka Cluster를 배포하기 편리하게 구성
2. Kafka의 가용성 확보에 필요한 Zookeeper 또한 Operator가 관리해준다는 편리함
3. Kafka Bridge, Exporter 등의 다른 구성요소도 쉽게 추가

 

• cluster operator: kafaka cluster, zookeeper cluster 등 컴퍼넌트를 배포하고 관리
• entity operator: user operator와 topic operator를 관리
  • topic operator: topic생성, 삭제 등 topic관리
  • user operator: kafaka 유저 관리
• zookeeper cluster
  • 카프카의 메타데이터 관리 및 브로커의 정상 상태 점검
• kafka cluster
  • 카프카 클러스터(여러 대 브로커 구성) 구성

​

strimzi-kafka 컴포넌트 설치 (Helm Chart 방식)

helm repo add strimzi https://strimzi.io/charts/
helm install kafka-operator strimzi/strimzi-kafka-operator --namespace kafka --create-namespace

 

​

Operator 패턴

kubernetes 공식 문서에 따르면 Operator 패턴을 다음과 같이 소개하고 있다.

​

Operators are software extensions to Kubernetes that make use of custom resources to manage applications and their components. Operators follow Kubernetes principles, notably the control loop​

​

 

그러니까 사용자 정의 리소스를 사용해서 애플리케이션 및 해당 컴포넌트를 관리하는 역할을 한다는거다.

 

https://docs.kubermatic.com/kubecarrier/v0.3/references/kubernetes-operators/

 

 

Operator : 사용자 정의 리소스(custom resource)의 컨트롤러 역할을 하는 k8s API의 클라이언트

• kubernetes 자체가 제공하는 것 이상의 작업을 자동화하기 위해서 코드를 작성
• 컨트롤러를 하나 이상의 custom resource에 연결하여 동작을 확장

 

 

오퍼레이터는 지속적으로 custom resource를 주시하다 이벤트가 발생히면 그에 맞춰 실제 k8s 오브젝트를 조절하는 역할을 한다.

​

Statefulset으로 이루어진 DB cluster나 kafka cluster 등의 복잡한 요소들을 하나의 custom resource로 추상화시켜서 간편하고 안전하게 관리할 수 있다.

​

 

 

Kafka Cluster 생성하기

Ephemeral cluster (임시 Kafka 클러스터)

• 개발 및 테스트 목적으로 Pod가 다운되면 삭제

Persistent cluster (영구 Kafka 클러스터)

• PV를 획득해서 자동으로 Volume 프로비저닝을 진행

​

The following examples show some common types of persistent volumes:

• If your Kubernetes cluster runs on Amazon AWS, Kubernetes can provision Amazon EBS volumes
• If your Kubernetes cluster runs on Microsoft Azure, Kubernetes can provision Azure Disk Storage volumes
• If your Kubernetes cluster runs on Google Cloud, Kubernetes can provision Persistent Disk volumes
• If your Kubernetes cluster runs on bare metal, Kubernetes can provision local persistent volumes

​

다음의 Kafka CR를 kubectl apply -f *.yaml -n kafka로 생성해주자.

default 네임스페이스에 해도 되긴하는데, 나는 구분하기 위해서 kafka ns에 죄다 담았다.

 

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    version: 3.7.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      inter.broker.protocol.version: "3.7"
    storage:
      type: ephemeral
  zookeeper:
    replicas: 3
    storage:
      type: ephemeral
  entityOperator:
    topicOperator: {}
    userOperator: {}

​

• kubectl get pod -n kafka : kafka 네임스페이스의 모든 파드들을 나열

​

• kubectl get strimzi : kafkatopic와 같은 kafka의 CR를 나열

​

• kubectl exec my-cluster-kafka-0 -it -- bin/kafka-broker-api-versions.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 : Kafka Broker의 설정을 확인할 수 있다.

id=0,1,2로 명명된 my-cluster-kafka-0,1,2 Broker들의 정보를 볼 수 있다.

​

​

Kafka Topic 생성하기

Operator에서 생성한 CR인 KafkaTopic을 이용해서 Topic을 생성하자.

​

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: common
  labels:
    strimzi.io/cluster: "my-cluster"
spec:
  partitions: 1
  replicas: 3
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824
    min.insync.replicas: 2

kubectl get kafkatopic 명령어로 정상적으로 생성됐는지 확인하자.

​

​

Producer & Conusmer 생성과 Message 주고받기

편리하게도, 스크립트 파일중에서 Producer와 Consumer 역할을 대신 해줄 녀석들(producer.sh, consumer.sh)이 존재한다.​ 원래엿으면 각 애플리케이션이 필요했음.

​

​

Producer부터 생성해보자.

 

새 터미널을 열어서 아래의 명령어를 입력

kubectl exec my-cluster-kafka-0 -n kafka -it -- bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic common

해당 Producer 역할의 터미널은 Message를 입력할 수 있음

​

​

다음으로 Consumer를 생성한다.

kubectl exec my-cluster-kafka-0 -it -- bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic mytopic --from-beginning

​

Consumer Group 생성하기

• 기존 명령어 끝에 --group mygroup 플래그를 추가

​

​

​

kafka cluster가 잘 작동한다.

Produce한 메시지가 잘 소비되는 모습

​

​

​

​

---

​

(해결) kafka-cluster.yaml 적용시 오류

ubuntu:~/environment/k8s-resource $ kubectl apply -f kafka.yaml
Error from server (BadRequest): error when creating "kafka.yaml": 
     Kafka cannot be handled as a Kafka: strict decoding error: unknown field "zookeeper"

 

화난다 이 삽질!!!

https://github.com/strimzi/strimzi-kafka-operator/tree/0.41.0/examples

strimzi-kafka-operator/examples at 0.41.0 · strimzi/strimzi-kafka-operator

직접 짜다가 실패한거 같아서 제공하는 Ephemeral cluster를 공식 문서에서 찾아냈다.

얘네를 리소스들에 kafka namespace 붙여서 생성했다.

​

​

​

​

(해결) Producer-console 명령어 호출시 LEADER_NOT_AVAILABLE 오류

[Producer clientId=console-producer] Error while fetching metadata with correlation id 7 
                  : {common=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

LEADER_NOT_AVAILABLE 오류는 Kafka Broker가 요청된 Topic의 Leader Partition을 제공할 수 없다는 뜻이다.

​

보통 Broker가 사용가능하지 않거나, 토픽이 생성되지 않은 경우 생기는건데

난 순서가 엇갈린거같음. kafka namespace 안에서 topic 생성해줘서 해결했다.

​

​

​

(해결) Consumer-console 명령어 호출시 no such file or directory 오류

error: Internal error occurred: error executing command in container: 
failed to exec in container: failed to start exec <container>: OCI runtime exec failed: 

exec failed: unable to start container process: exec: "bin/kafka-consumer-producer.sh": 
                   stat bin/kafka-consumer-producer.sh: no such file or directory: unknown

 

별거 없는 오타였다.