MaxRokatansky Apr 5 2023 at 15:46

Приложения в Kubernetes: быстрый запуск Kafka с KRaft

6 min

4.9K

OTUS corporate blogApache*Kubernetes*

Tutorial

Автор статьи: Рустем Галиев

IBM Senior DevOps Engineer & Integration Architect

Привет Хабр!

Сегодня у нас будет возможность установить Kafka с одной из самых простых конфигураций. Эта установка не оптимизирована для производственных сред, но идеально подходит для быстрой и локальной разработки.

Apache Kafka — это фреймворк, реализующий программную шину, использующую потоковую обработку. Это программная платформа с открытым исходным кодом, разработанная Apache Software Foundation и написанная на Scala и Java. Проект направлен на предоставление унифицированной платформы с высокой пропускной способностью и малой задержкой для обработки потоков данных в реальном времени.
Иными же словами распределенная система обмена сообщениями между серверными приложениями в режиме реального времени.

Эта установка Kafka использует проект ранней версии под названием Apache Kafka Raft (KRaft). Это проект, в котором устранена зависимость от ZooKeeper.

Сперва нам будет полезно иметь реестр на этапах билда, пуша и деплоймента. Нет необходимости передавать частные образы через Интернет. Вместо этого мы сохраняем все в локальном реестре.

Установим реестр

Существует множество вариантов создания реестра образов контейнеров. Мы предпочитаем чистое решение Kubernetes и поэтому устанавливаем реестр через Docker Registry Helm Chart.

Добавим репозиторий для установки диаграммы Helm:

helm repo add twuni https://helm.twun.io && helm repo list
Установите чарт для частного реестра контейнеров:

helm install registry twuni/docker-registry \
  --version 2.1.0 \
  --namespace kube-system \
  --set service.type=NodePort \
  --set service.nodePort=31500

Реестр теперь доступен как услуга. Его можно вывести:

kubectl get service --namespace kube-system

Не забудьте назначить переменную среды общему расположению реестра:

У меня было

export REGISTRY=4f809e0c89d3406b8e4ccc59da3d2223-2887247877-31500-kira01.environments.vmoops.com

Пройдет несколько секунд, прежде чем развертывание реестра сообщит, что оно доступно:

kubectl get deployments registry-docker-registry --namespace kube-system

Как только реестр станет доступен, проверим содержимое пустого реестра:

curl $REGISTRY/v2/_catalog | jq -c

Вы увидите этот ответ реестра с ожидаемым пустым массивом: {"repositories":[]}

Запустим Кафка брокеры

Kafka — это распределенная система, в которой реализованы основные функции системы публикации-подписки. На каждом хосте в кластере Kafka работает сервер, называемый брокером, который хранит сообщения, отправленные в топики, и обслуживает запросы консьюмеров. В настоящее время Kafka использует ZooKeeper для отслеживания состояния брокеров в кластере Kafka и ведения списка топиков и сообщений Kafka.

На сегодня мы будем использовать ранний доступ и, возможно, будущую реализацию Kafka, использующую KRaft. Вместо того, чтобы полагаться на ZooKeeper, управление метаданными реализовано в ядре Kafka в виде набора контроллеров кворума. Как и ZooKeeper, они основаны на алгоритме консенсуса Raft, поэтому реализация является надежной и отказоустойчивой, и она обещает улучшить производительность и безопасность Kafka. Конфигурация KRaft также хорошо подходит для быстрой разработки.

В настоящее время нет официального образа контейнера, поддерживающего реализацию KRaft. Однако IBM предоставляет два файла — Dockerfile и entrypoint.sh, которые были загружены в домашний каталог нашей ВМ.

Dockerfile:

FROM openjdk:11

ENV KAFKA_VERSION=3.3.2
ENV SCALA_VERSION=2.13
ENV KAFKA_HOME=/opt/kafka
ENV PATH=${PATH}:${KAFKA_HOME}/bin

LABEL name="kafka" version=${KAFKA_VERSION}

RUN wget -O /tmp/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz https://downloads.apache.org/kafka/${KAFKA_VERSION}/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz \
 && tar xfz /tmp/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz -C /opt \
 && rm /tmp/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz \
 && ln -s /opt/kafka_${SCALA_VERSION}-${KAFKA_VERSION} ${KAFKA_HOME} \
 && rm -rf /tmp/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz

COPY ./entrypoint.sh /
RUN ["chmod", "+x", "/entrypoint.sh"]
ENTRYPOINT ["/entrypoint.sh"]

entrypoint.sh:

NODE_ID=${HOSTNAME:6}
LISTENERS="PLAINTEXT://:9092,CONTROLLER://:9093"
ADVERTISED_LISTENERS="PLAINTEXT://kafka-$NODE_ID.$SERVICE.$NAMESPACE.svc.cluster.local:9092"

CONTROLLER_QUORUM_VOTERS=""
for i in $( seq 0 $REPLICAS); do
	if [[ $i != $REPLICAS ]]; then
    	CONTROLLER_QUORUM_VOTERS="$CONTROLLER_QUORUM_VOTERS$i@kafka-$i.$SERVICE.$NAMESPACE.svc.cluster.local:9093,"
	else
    	CONTROLLER_QUORUM_VOTERS=${CONTROLLER_QUORUM_VOTERS::-1}
	fi
done

mkdir -p $SHARE_DIR/$NODE_ID

if [[ ! -f "$SHARE_DIR/cluster_id" && "$NODE_ID" = "0" ]]; then
	CLUSTER_ID=$(kafka-storage.sh random-uuid)
	echo $CLUSTER_ID > $SHARE_DIR/cluster_id
else
	CLUSTER_ID=$(cat $SHARE_DIR/cluster_id)
fi

sed -e "s+^node.id=.*+node.id=$NODE_ID+" \
-e "s+^controller.quorum.voters=.*+controller.quorum.voters=$CONTROLLER_QUORUM_VOTERS+" \
-e "s+^listeners=.*+listeners=$LISTENERS+" \
-e "s+^advertised.listeners=.*+advertised.listeners=$ADVERTISED_LISTENERS+" \
-e "s+^log.dirs=.*+log.dirs=$SHARE_DIR/$NODE_ID+" \
/opt/kafka/config/kraft/server.properties > server.properties.updated \
&& mv server.properties.updated /opt/kafka/config/kraft/server.properties

kafka-storage.sh format -t $CLUSTER_ID -c /opt/kafka/config/kraft/server.properties

exec kafka-server-start.sh /opt/kafka/config/kraft/server.properties

Из этих двух файлов можно создать надежный образ Kafka на основе KRaft. Создадим образ KRaft:

docker build -t $REGISTRY/kafka-kraft .
Создание образа контейнера займет несколько минут. Отправим образ KRaft в реестр образов контейнеров в локальном кластере Kubernetes, который был настроен ранее:

docker push $REGISTRY/kafka-kraft
IBM также предоставляет манифест StatefulSet для запуска KRaft в Kubernetes kafka.yaml:

apiVersion: v1
kind: Namespace
metadata:
  name: kafka-kraft
---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: kafka-pv-volume
  labels:
	type: local
spec:
  storageClassName: manual
  capacity:
	storage: 10Gi
  accessModes:
	- ReadWriteOnce
  hostPath:
	path: '/mnt/data'
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: kafka-pv-claim
  namespace: kafka-kraft
spec:
  storageClassName: manual
  accessModes:
	- ReadWriteOnce
  resources:
	requests:
  	storage: 3Gi
---
apiVersion: v1
kind: Service
metadata:
  name: kafka-svc
  labels:
	app: kafka-app
  namespace: kafka-kraft
spec:
  clusterIP: None
  ports:
	- name: '9092'
  	port: 9092
  	protocol: TCP
  	targetPort: 9092
  selector:
	app: kafka-app
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
  labels:
	app: kafka-app
  namespace: kafka-kraft
spec:
  serviceName: kafka-svc
  replicas: 3
  selector:
	matchLabels:
  	app: kafka-app
  template:
	metadata:
  	labels:
    	app: kafka-app
	spec:
  	volumes:
    	- name: kafka-storage
      	persistentVolumeClaim:
        	claimName: kafka-pv-claim
  	containers:
    	- name: kafka-container
      	image: $REGISTRY/kafka-kraft
      	ports:
        	- containerPort: 9092
        	- containerPort: 9093
      	env:
        	- name: REPLICAS
          	value: '3'
        	- name: SERVICE
          	value: kafka-svc
        	- name: NAMESPACE
          	value: kafka-kraft
        	- name: SHARE_DIR
          	value: /mnt/kafka
      	volumeMounts:
        	- name: kafka-storage
          	mountPath: /mnt/kafka

Один из манифестов — это объявление PersistentVolume для хранения данных Kafka. Для этого создадим локальную директорию:

mkdir /mnt/kafka
Имея KRaft в реестре образов контейнеров, мы можем запустить KRaft:

envsubst < kafka.yaml | kubectl apply -f -

envsubst используется для ввода значения $REGISTRY в YAML. KRaft устанавливается в пространство имен kafka-kraft; переключим контекст на это пространство имен, чтобы последующие команды предполагали этот контекст:

kubectl config set-context --current --namespace=kafka-kraft

Проверим статус установки:

kubectl get services,statefulsets,pods,pv,pvc

Когда хотя бы первый брокер Kafka запущен, проверьте его логи:

kubectl logs kafka-0

Если вы видите ошибки в логах, такие как Error connecting to node (Ошибка подключения к узлу), в настоящее время являются нормальной активностью, поскольку брокеры пытаются соединиться друг с другом при запуске.
Через несколько секунд поды будут сообщать о том, что они запущены, а StatefulSet сообщит о готовности 3/3:

kubectl get services,statefulsets,pods | grep -z 'Running\|3/3\|kafka-svc\|9092'

Обратите внимание, что служба Kafka доступна в кластере по адресу kafka-svc:9092.

В завершение хочу пригласить вас на бесплатный урок, где мои коллеги из OTUS расскажут как устроен мониторинг кластера, его компоненты и приложения в кластере. Вы изучите различные подходы мониторинга, подходы к мониторингу как приложения так и компонентов кластера, основные метрики Kubernetes. Также узнаете про кластеризацию/федерацию Prometheus, дополнительные хранилища метрик для prometheus (victoria metrics; thanos, cortex).

Зарегистрироваться на бесплатный вебинар

Tags:

Hubs:

Приложения в Kubernetes: быстрый запуск Kafka с KRaft

Автор статьи: Рустем Галиев

Установим реестр

Запустим Кафка брокеры

Articles

Information