Шаг за шагом: развёртываем Kafka Connect с помощью оператора Strimzi в Kubernetes

Strimzi — это практически самый широкий оператор Kubernetes Kafka, который можно использовать для развертывания Apache Kafka, либо других его компонентов, таких как Kafka Connect, Kafka Mirror и т.д. В статье мы пошагово разберем развертывание Kafka Connect в Kubernetes. А еще затронем проблемы, с которыми можно столкнуться во время процедуры развертывания и приведем способы их решения.

Примечание: Учтите, что Strimzi основан на Apache Kafka, а не на платформе Confluent. Вот почему вам, скорее всего, потребуется добавить некоторые артефакты Confluent, например, Confluent Avro Converter, чтобы получить отдачу от этого.

Статья основана на Strimzi v0.29.0. Это значит, что вы можете установить следующие версии Kafka Connect:

• Стримзи: 0.29.0

• Apache Kafka и Kafka Connect: до 3.2

• Эквивалентная платформа слияния: 7.2.4

Примечание: вы можете преобразовать версию платформы Confluent в версию Apache Kafka и наоборот с помощью приведенной здесь таблицы.

Установка

Графический интерфейс Openshift и CLI Kubernetes

Если вы используете Openshift, перейдите в раздел Операторы > установленные операторы > Strimzi > Kafka Connect.

Шаг за шагом: Теперь вы столкнетесь с формой, содержащей конфигурации Kafka connect. Чтобы получить эквивалентный Yaml-файл формы, нажмите на Yaml View. Любое обновление представления формы применяется к представлению Yaml «на лету». Только не используйте его для непосредственного создания экземпляра. Он нужен для преобразования желаемой конфигурации в файл Yaml. После получения Yaml-файл разверните оператор с помощью команды kubectl apply. Итак, подведем итог:

• Введите конфигурацию в представлении формы.

• Нажмите на просмотр Yaml.

• Скопируйте его содержимое в файл Yaml на вашем локальном компьютере (например, kafka-connect.yaml).

• Выполнить: kubectl apply -f kafka-connect.yaml .

Вид Kafka-Connect либо развертывается, либо обновляется. Развернутые ресурсы состоят из развертывания и модулей, сервиса, конфигурационных карт и секретов.

Давайте пройдемся по минимальной конфигурации и шаг за шагом сделаем ее более продвинутой.

Минимальная конфигурация

Чтобы развернуть простую минимальную конфигурацию Kafka Connect, вы можете использовать приведенный ниже Yaml:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  namespace: <YOUR_PROJECT_NAME>
spec:
  config:
    config.storage.replication.factor: -1
    config.storage.topic: okd4-connect-cluster-configs
    group.id: okd4-connect-cluster
    offset.storage.replication.factor: -1
    offset.storage.topic: okd4-connect-cluster-offsets
    status.storage.replication.factor: -1
    status.storage.topic: okd4-connect-cluster-status
  bootstrapServers: kafka1, kafka2
  version: 3.2.0
  replicas: 1

Также для этой цели подойдет Rest API Kafka Connect на порту 8083, открытом в модуле. Предоставьте его в частной или внутренней сети, определив маршрут в OLD.

Аутентификация REST API

С помощью конфигурации, описанной здесь, добавьте аутентификацию к REST-прокси Kafka Connect. Единственное — это не сработает с оператором Strimzi. Поэтому для обеспечения безопасности в Kafka Connect, у вас есть два варианта:

• Используйте API-оператор Kafka Connector. Оператор Strimzi поможет определить тип соединителя в файле YAML. Однако в некоторых случаях это может быть непрактично, ведь необходимо обновлять, приостанавливать и останавливать соединители через REST API.

• Поместите небезопасный REST API за аутентифицированный API-шлюз, такой, как Apache APISIX. Подойти может любой другой инструмент или приложение собственной разработки.

Показатели JMX Prometheus

Чтобы предоставить метрики JMX Prometheus, полезные для наблюдения за статусами соединителей в Grafana, добавьте приведенную ниже конфигурацию:

metricsConfig:
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        key: jmx-prometheus
        name: configs
  jmxOptions: {}

Он использует предварительно определенную конфигурацию для экспорта Prometheus. Вы можете использовать эту конфигурацию:

startDelaySeconds: 0
ssl: false
lowercaseOutputName: false
lowercaseOutputLabelNames: false
rules:
- pattern : "kafka.connect<type=connect-worker-metrics>([^:]+):"
  name: "kafka_connect_connect_worker_metrics_$1"
- pattern : "kafka.connect<type=connect-metrics, client-id=([^:]+)><>([^:]+)"
  name: "kafka_connect_connect_metrics_$2"
  labels:
    client: "$1"
- pattern: "debezium.([^:]+)<type=connector-metrics, context=([^,]+), server=([^,]+), key=([^>]+)><>RowsScanned"
  name: "debezium_metrics_RowsScanned"
  labels:
    plugin: "$1"
    name: "$3"
    context: "$2"
    table: "$4"
- pattern: "debezium.([^:]+)<type=connector-metrics, context=([^,]+), server=([^>]+)>([^:]+)"
  name: "debezium_metrics_$4"
  labels:
    plugin: "$1"
    name: "$3"
    context: "$2"

Сервис для внешнего Prometheus

Если вы собираетесь развернуть Prometheus в сочетании со Strimzi для сбора показателей, следуйте инструкциям. Но помните, что в случае использования внешнего Prometheus история разворачивается по-другому.

Оператор Strimzi создает отображение портов в сервисе только для этих портов:

• 8083: Kafka Connect REST API

• 9999: порт JMX

К сожалению, это не создает сопоставление для порта 9404, HTTP-порта экспортера Prometheus. Итак, мы должны создать сервис самостоятельно:

kind: Service
apiVersion: v1
metadata:
  name: kafka-connect-jmx-prometheus
  namespace: kafka-connect
  labels:
    app.kubernetes.io/instance: kafka-connect
    app.kubernetes.io/managed-by: strimzi-cluster-operator
    app.kubernetes.io/name: kafka-connect
    app.kubernetes.io/part-of: strimzi-kafka-connect
    strimzi.io/cluster: kafka-connect
    strimzi.io/kind: KafkaConnect
spec:
  ports:
    - name: tcp-prometheus
      protocol: TCP
      port: 9404
      targetPort: 9404
  type: ClusterIP
  selector:
    strimzi.io/cluster: kafka-connect
    strimzi.io/kind: KafkaConnect
    strimzi.io/name: kafka-connect-connect
status:
  loadBalancer: {}

Примечание: этот метод работает только для развертываний с одним модулем. Вы должны определить маршрут для сервиса, и даже в случае headless сервиса маршрут возвращает один IP-модуль за раз. Следовательно, Prometheus не может очистить все показатели pods. Вот почему рекомендуется использовать Podmonitor и Prometheus в облаке.

Плагины и артефакты

Чтобы добавить плагины и артефакты, есть два способа.

Секция сборки оператора

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

Он поддерживает jar, tgz, zip, and maven. Но в случае Maven создается многоступенчатый файл Dockerfile, который проблематичен для Openshift, и он сталкивается со сбоем на стадии сборки. Следовательно, вам следует использовать только другие типы, которым не нужна стадия компиляции, например, jar, zip, tgz, и в итоге вы получите одноступенчатый файл Dockerfile.

Например, чтобы добавить плагин Debezium MySQL, вы можете использовать приведенную ниже конфигурацию:

spec:  
  build:
    output:
      image: 'kafkaconnect:1.0'
      type: imagestream
    plugins:
      - artifacts:
          - type: tgz
            url: >-
              https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/2.1.4.Final/debezium-connector-mysql-2.1.4.Final-plugin.tar.gz
        name: debezium-connector-mysql

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

Изменение изображений

Оператор может использовать желаемое вами изображение вместо изображения по умолчанию. Так вы можете добавить выбранные артефакты и плагины, создав образ вручную или с помощью CI/CD. Вы, возможно, захотите использовать именно этот способ, ведь Strimzi использует Apache Kafka image, а не платформу Confluent. То есть, в развертываниях нет совместимых пакетов, таких как Confluent Avro Converter и т.д. Поэтому вам нужно добавить их в свой образ и настроить оператора на использование вашего образа docker.

Например, если вы хотите добавить свой настроенный плагин Debezium MySQL Connector из универсальных пакетов Gitlab и Confluent Avro Converter в базовый образ, сначала используйте этот файл Dockerfile:

ARG CONFLUENT_VERSION=7.2.4

# Install confluent avro converter
FROM confluentinc/cp-kafka-connect:${CONFLUENT_VERSION} as cp
# Reassign version
ARG CONFLUENT_VERSION
RUN confluent-hub install --no-prompt confluentinc/kafka-connect-avro-converter:${CONFLUENT_VERSION}

# Copy privious artifacts to the main strimzi kafka image
FROM quay.io/strimzi/kafka:0.29.0-kafka-3.2.0
ARG GITLAB_TOKEN
ARG CI_API_V4_URL=https://gitlab.snapp.ir/api/v4
ARG CI_PROJECT_ID=3873
ARG DEBEZIUM_CONNECTOR_MYSQL_CUSTOMIZED_VERSION=1.0
USER root:root

# Copy Confluent packages from previous stage
RUN mkdir -p /opt/kafka/plugins/avro/
COPY --from=cp /usr/share/confluent-hub-components/confluentinc-kafka-connect-avro-converter/lib /opt/kafka/plugins/avro/

# Connector plugin debezium-connector-mysql
RUN 'mkdir' '-p' '/opt/kafka/plugins/debezium-connector-mysql' \
    && curl --header "${GITLAB_TOKEN}" -f -L \
    --output /opt/kafka/plugins/debezium-connector-mysql.tgz \
    ${CI_API_V4_URL}/projects/${CI_PROJECT_ID}/packages/generic/debezium-customized/${DEBEZIUM_CONNECTOR_MYSQL_CUSTOMIZED_VERSION}/debezium-connector-mysql-customized.tar.gz \
    && 'tar' 'xvfz' '/opt/kafka/plugins/debezium-connector-mysql.tgz' '-C' '/opt/kafka/plugins/debezium-connector-mysql' \
    && 'rm' '-vf' '/opt/kafka/plugins/debezium-connector-mysql.tgz'

USER 1001

Создайте образ. Отправьте его в поток изображений или любой другой репозиторий docker и настройте оператора, добавив строку ниже:

spec:  
  image: image-registry.openshift-image-registry.svc:5000/kafka-connect/kafkaconnect-customized:1.0

В зависимости от его типа используйте различные конфигурации для добавления проверки подлинности Kafka. Здесь вы можете увидеть конфигурацию для Kafka с механизмом SASL/Plaintext и scram-sha-512:

spec:
  authentication:
    passwordSecret:
      password: kafka-password
      secretName: mysecrets
    type: scram-sha-512
    username: myuser

Далее вам нужно указать пароль в секретном файле с именем my secret.

Обработка учетных данных файла

Поскольку соединителям нужны учетные данные для доступа к базам данных, вы должны определить их как секреты и получить к ним доступ с помощью переменных среды. Однако, если их слишком много, вы можете поместить все учетные данные в файл и адресовать их в соединителе с помощью модификатора $file modifier.

1. Поместите все учетные данные в качестве значения ключа с именем credentials в секретный файл.

Файл учетных данных:

USERNAME_DB_1=user1
PASSWORD_DB_1=pass1

USERNAME_DB_2=user2
PASSWORD_DB_2=pass2

Secret file:

kind: Secret
apiVersion: v1
metadata:
  name: mysecrets
  namespace: kafka-connect
data:
  credentials: <BASE64 YOUR DATA>

2. Настройте оператора с секретом в качестве тома:

spec:
  config:
    config.providers: file
    config.providers.file.class: org.apache.kafka.common.config.provider.FileConfigProvider  
  externalConfiguration:
    volumes:
      - name: database_credentials
        secret:
          items:
            - key: credentials
              path: credentials
          optional: false
          secretName: mysecrets

3. Теперь в соединителе вы можете получить доступ к PASSWORD_DB_1 с помощью приведенной ниже команды:

"${file:/opt/kafka/external-configuration/database_credentials/credentials:PASSWORD_DB_1}"

Собрать все это воедино

Если мы соберем все конфигурации вместе, у нас будет приведенная ниже конфигурация для Kafka Connect:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: kafka-connect
  namespace: kafka-connect
spec:
  authentication:
    passwordSecret:
      password: kafka-password
      secretName: mysecrets
    type: scram-sha-512
    username: myuser
  config:
    config.providers: file
    config.providers.file.class: org.apache.kafka.common.config.provider.FileConfigProvider
    config.storage.replication.factor: -1
    config.storage.topic: okd4-connect-cluster-configs
    group.id: okd4-connect-cluster
    offset.storage.replication.factor: -1
    offset.storage.topic: okd4-connect-cluster-offsets
    status.storage.replication.factor: -1
    status.storage.topic: okd4-connect-cluster-status
  bootstrapServers: 'kafka1:9092, kafka2:9092'
  metricsConfig:
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        key: jmx-prometheus
        name: configs
  resources:
    limits:
      memory: 1Gi
    requests:
      memory: 1Gi
  readinessProbe:
    failureThreshold: 10
    initialDelaySeconds: 60
    periodSeconds: 20
  jmxOptions: {}
  livenessProbe:
    failureThreshold: 10
    initialDelaySeconds: 60
    periodSeconds: 20
  image: image-registry.openshift-image-registry.svc:5000/kafka-connect/kafkaconnect-customized:1.0
  version: 3.2.0
  replicas: 2
  externalConfiguration:
    volumes:
      - name: database_credentials
        secret:
          items:
            - key: credentials
              path: credentials
          optional: false
          secretName: mysecrets

Примечание: сервис, маршрут и конфигурация сборки не указаны, это есть в статье выше.

Делаем выводы

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

Чтобы разобраться, как развертывать Kafka и использовать этот инструмент в работе, приглашаем вас на курс Apache Kafka для разработчиков. Обучение стартует 12 мая 2023. На нем разберем:

• неправильное использование Кафка и отсутствие коммитов в ней;

• ваши кейсы о проблемах при работе с Apache Kafka;

• опыт создания Data Lake на ~80 ТБ с помощью Apache Kafka;

• особенности эксплуатации kafka с retention в 99999999.

В этой статье мы рассмотрели ключевые шаги, связанные с развертыванием Kafka Connect с помощью оператора Strimzi, включая создание минимального пользовательского определения ресурсов (CRD), проблему базовой аутентификации REST API, аутентификацию Kafka, показатели JMX Prometheus, плагины и артефакты, а также обработку учетных данных файла. Выполнив эти действия, вы можете легко настроить развертывание Kafka Connect в соответствии с вашими конкретными потребностями.