Kafka является фактическим хранилищем событий и распределенным брокером сообщений для больших систем с микросервисной архитектурой. Kubernetes является отраслевым стандартом для оркестровки контейнерных сервисов. Для многих организаций развертывание Kafka на Kubernetes – это подход, не требующий больших затрат, который вписывается в их архитектурную стратегию.

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

Краткое описание технологий

Начнем с краткого обзора Kubernetes и Kafka.

Kubernetes

Компания Google начала разработку того, что в итоге стало Kubernetes (k8s), в 2003 году. В то время проект был известен как Borg. В 2014 году Google выпустил k8s как проект с открытым исходным кодом на Github, и k8s быстро обрел партнеров в лице Microsoft, Red Hat и Docker. Со временем Kubernetes стали использовать все больше и больше компаний, включая сам GitHub и популярную игру Pokémon GO. В 2022 году мы видим, что использование k8s растет в пространстве AI/ML и с растущим акцентом на безопасность.

Внедрение k8s в жизненный цикл облачной разработки обеспечило несколько ключевых преимуществ:

• развертывание с нулевым временем простоя
• Масштабируемость
• Неизменяемая инфраструктура
• Самовосстанавливающиеся системы

Многие из этих преимуществ обусловлены использованием в k8s декларативной конфигурации. Для того чтобы изменить конфигурацию инфраструктуры, ресурсы должны быть уничтожены и восстановлены, что обеспечивает неизменяемость. Кроме того, если k8s обнаруживает ресурсы, которые отклонились от заявленной спецификации, он пытается восстановить состояние системы, чтобы она снова соответствовала этой спецификации.

Для разработчиков использование k8s означает, что наконец-то покончено с удручающими полуночными развертываниями, когда приходится бросать все дела, чтобы масштабировать сервисы или исправлять производственные среды непосредственно вручную.

Apache Kafka

Kafka – это инструмент распределенной обработки потоков с открытым исходным кодом. Kafka позволяет нескольким “производителям” добавлять сообщения (пары ключ-значение) в “темы”. Эти сообщения упорядочиваются в каждой теме в виде очереди. “Потребители” подписываются на тему и могут получать сообщения в порядке их поступления в очередь.

Kafka размещается на сервере, который обычно называется “брокер”. В разных регионах может быть множество различных брокеров Kafka. Помимо брокеров Kafka, еще одна служба под названием Zookeeper синхронизирует работу различных брокеров и помогает координировать темы и сообщения.

Гениальность Kafka заключается в том, что он может обрабатывать сотни тысяч сообщений в секунду, будучи распределенным и при относительно низкой стоимости одного мегабайта.

Kafka часто занимает место сродни “центральной нервной системе” в архитектуре микросервисов. Сообщения передаются между производителями и потребителями, которые в действительности являются сервисами внутри вашего облака. Один и тот же сервис может быть и потребителем, и производителем сообщений из одной или разных тем внутри Kafka.

В качестве примера можно привести создание нового пользователя в вашем приложении. Служба User публикует сообщение в теме “Provision User”. Служба Email потребляет это сообщение о новом пользователе и затем отправляет ему приветственное письмо. Службам User и Email не нужно было напрямую передавать сообщения друг другу, но их соответствующие задания выполнялись асинхронно.

Развертывание Kafka с помощью Kubernetes

Для нашего мини-проекта мы установим Kubernetes и Kafka нейтральным для облака способом, используя Minikube, который позволяет нам запустить весь кластер k8s на одной машине. Мы установили следующие приложения:

• Minikube версии v1.25.2
• Клиент Kubectl версии 1.23.3

Установка Minikube

Установив Minikube, мы можем запустить его с помощью команды minikube start. После этого мы можем увидеть состояние:

$ minikube status

minikube

type: Control Plane
host: Running
kubelet: Running
apiserver: Running
kubeconfig: Configured

Инструкции по настройке Kubernetes для работы в выбранном вами облачном провайдере можно найти в документации по каждому провайдеру (например, AWS, GCP или Azure), но приведенные ниже конфигурационные файлы YAML должны работать для всех провайдеров, с небольшими корректировками для IP-адресов и связанных с ними полей.

Определение пространства имен Kafka

Сначала мы определим пространство имен для развертывания всех ресурсов Kafka, используя файл с именем 00-namespace.yaml:

apiVersion: v1
kind: Namespace
metadata:
  name: "kafka"
  labels:
    name: "kafka"

Мы применим этот файл с помощью kubectl apply -f 00-namespace.yaml.

Мы можем проверить, что пространство имен было создано правильно, запустив kubectl get namespaces, проверив, что Kafka является пространством имен, присутствующим в Minikube.

Развертывание Zookeeper

Далее мы развернем Zookeeper в нашем пространстве имен k8s. Мы создаем файл 01-zookeeper.yaml со следующим содержимым:

apiVersion: v1
kind: Service
metadata:
  labels:
    app: zookeeper-service
  name: zookeeper-service
  namespace: kafka
spec:
  type: NodePort
  ports:
    - name: zookeeper-port
      port: 2181
      nodePort: 30181
      targetPort: 2181
  selector:
    app: zookeeper
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: zookeeper
  name: zookeeper
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: zookeeper
  template:
    metadata:
      labels:
        app: zookeeper
    spec:
      containers:
        - image: wurstmeister/zookeeper
          imagePullPolicy: IfNotPresent
          name: zookeeper
          ports:
            - containerPort: 2181

В этом YAML-файле создано два ресурса. Первый – это служба zookeeper-service, которая будет использовать развертывание, созданное во втором ресурсе zookeeper. Развертывание использует образ Docker wurstmeister/zookeeper для фактического бинарного файла Zookeeper. Сервис открывает это развертывание через порт во внутренней сети k8s. В данном случае мы используем стандартный порт Zookeeper 2181, который также открывает контейнер Docker.

Мы применяем этот файл с помощью следующей команды: kubectl apply -f 01-zookeeper.yaml.

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

$ kubectl get services -n kafka
NAME               TYPE      CLUSTER-IP     PORT(S)         AGE
zookeeper-service  NodePort  10.100.69.243  2181:30181/TCP  3m4s

Мы видим внутренний IP-адрес Zookeeper (10.100.69.243), который нам понадобится, чтобы указать брокеру, где его слушать.

Развертывание брокера Kafka

Последний шаг – развертывание брокера Kafka. Мы создадим файл 02-kafka.yaml со следующим содержимым, где заменим <ZOOKEEPER-INTERNAL-IP> на CLUSTER-IP из предыдущего шага для Zookeeper. Если не выполнить этот шаг, брокер не будет развернут.

apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka-broker
  name: kafka-service
  namespace: kafka
spec:
  ports:
  - port: 9092
  selector:
    app: kafka-broker
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: kafka-broker
  name: kafka-broker
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka-broker
  template:
    metadata:
      labels:
        app: kafka-broker
    spec:
      hostname: kafka-broker
      containers:
      - env:
        - name: KAFKA_BROKER_ID
          value: "1"
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: <ZOOKEEPER-INTERNAL-IP>:2181
        - name: KAFKA_LISTENERS
          value: PLAINTEXT://:9092
        - name: KAFKA_ADVERTISED_LISTENERS
          value: PLAINTEXT://kafka-broker:9092
        image: wurstmeister/kafka
        imagePullPolicy: IfNotPresent
        name: kafka-broker
        ports:
        - containerPort: 9092

И снова мы создаем два ресурса – сервис и развертывание – для одного брокера Kafka. Мы запускаем kubectl apply -f 02-kafka.yaml. Мы убеждаемся в этом, увидев стручки в нашем пространстве имен:

$ kubectl get pods -n kafka
NAME                            READY   STATUS    RESTARTS   AGE
kafka-broker-5c55f544d4-hrgnv   1/1     Running   0          48s
zookeeper-55b668879d-xc8vd      1/1     Running   0          35m

Капсуле Kafka Broker может потребоваться минута, чтобы перейти из состояния ContainerCreating в состояние Running.

Обратите внимание на строку в 02-kafka.yaml, где мы указываем значение для KAFKA_ADVERTISED_LISTENERS. Чтобы гарантировать, что Zookeeper и Kafka смогут общаться, используя это имя хоста (kafka-broker), нам нужно добавить следующую запись в файл /etc/hosts на нашей локальной машине:

127.0.0.1 kafka-broker

Тестирование тем Kafka

Чтобы проверить, что мы можем отправлять и получать сообщения из темы в Kafka, нам нужно открыть порт для Kafka, чтобы сделать его доступным с localhost. Выполните следующую команду, чтобы открыть порт:

$ kubectl port-forward kafka-broker-5c55f544d4-hrgnv 9092 -n kafka
Forwarding from 127.0.0.1:9092 -> 9092
Forwarding from [::1]:9092 -> 9092

Приведенная выше команда kafka-broker-5c55f544d4-hrgnv ссылается на k8s pod, который мы видели выше, когда перечисляли pods в нашем пространстве имен kafka. Эта команда делает порт 9092 этого стручка доступным вне кластера Minikube k8s по адресу localhost:9092.

Чтобы легко отправлять и получать сообщения из Kafka, мы будем использовать инструмент командной строки под названием KCat (ранее Kafkacat). Чтобы создать сообщение и тему с именем test, мы выполним следующую команду:

$ echo "hello world!" | kafkacat -P -b localhost:9092 -t test

Команда должна выполниться без ошибок, что указывает на то, что производители нормально взаимодействуют с Kafka в k8s. Как посмотреть, какие сообщения находятся в очереди с именем test? Выполним следующую команду:

$ kafkacat -C -b localhost:9092 -t test

hello world!
% Reached end of topic test [0] at offset 1

Мы сделали это! Мы успешно развернули Kafka с Kubernetes!

Следующие шаги

Kafka является ключевым игроком для организаций, которые заинтересованы в реализации архитектур и систем, управляемых событиями в реальном времени. Развертывание Kafka с Kubernetes – это отличное начало, но организациям также необходимо выяснить, как обеспечить бесперебойную и безопасную работу Kafka с существующими экосистемами API. Им понадобятся инструменты для управления и обеспечения безопасности на протяжении всего жизненного цикла API.

Среди поставщиков я наткнулся на Gravitee, одно из ведущих решений, которое особенно сосредоточено на помощи организациям в управлении, обеспечении безопасности, регулировании и создании экосистем API – независимо от того, на основе каких протоколов, сервисов или стилей они строятся. Gravitee даже имеет коннектор Kafka, который принимает данные, открывая конечные точки, преобразующие запросы в сообщения, которые затем могут быть опубликованы в вашей теме Kafka. Он также может передавать события Kafka потребителям с помощью удобных для работы в Интернете протоколов, таких как Websocket.

Заключение

В этой статье мы рассказали о том, как Kafka помогает хореографировать микросервисные архитектуры, являясь центральной нервной системой, передающей сообщения к различным сервисам и от них. Для развертывания Kafka мы рассмотрели Kubernetes, мощную платформу оркестровки контейнеров, которую можно запускать локально (с помощью Minikube) или в производственных средах облачных провайдеров. Наконец, мы продемонстрировали, как использовать Minikube для создания локального кластера Kubernetes, развертывания Kafka, а затем проверили успешное развертывание и конфигурацию с помощью KCat.

Удачного развертывания!