Стадия жизненного цикла модуля: Preview

У модуля есть требования для установки

Kafka

Kafka — основной ресурс, с которым работает пользователь. Это namespaced ресурс, с помощью которого можно создавать инстансы брокера Kafka в Deckhouse Kubernetes Platform и привязывать их к конкретному KafkaClass.
Класс определяет допустимые размеры инстанса, перечень параметров конфигурации, доступных для переопределения, и правила валидации.

В spec ресурса Kafka есть следующие основные секции:

  • instance — требования к вычислительным ресурсам и хранилищу брокера (CPU, память, постоянный том). Должны соответствовать политикам sizing выбранного класса.
  • configuration — настройки брокера, которые вы хотите изменить. Здесь можно указывать только параметры, перечисленные в overridableConfiguration класса; остальные настройки управляются значениями по умолчанию класса.
  • tls — настройки серверного TLS-шифрования канала (cert-manager или пользовательский сертификат). См. раздел TLS.

KafkaClassName

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

spec:
  kafkaClassName: default

Instance

Описывает требования к вычислительным ресурсам и хранилищу брокера.
Должна пройти валидацию по sizingPolicy соответствующего класса:

spec:
  instance:
    memory:
      size: 1Gi
    cpu:
      cores: 1
      coreFraction: "50%"
    persistentVolumeClaim:
      size: 10Gi
      storageClassName: default

Configuration

Необязательные настройки брокера для переопределения значений по умолчанию класса.
Здесь принимаются только параметры, перечисленные в overridableConfiguration класса. Все значения проходят валидацию по правилам класса validations перед применением.

spec:
  configuration:
    logRetentionHours: 168
    logRetentionBytes: -1
    messageMaxBytes: "1Mi"
    compressionType: producer
    autoCreateTopicsEnable: false

Поля конфигурации

logRetentionHours

Тип: integer | Параметр Kafka: log.retention.hours | Мин: 1 | Пример: 168

Количество часов хранения лог-файла до его удаления.
Взаимоисключающий с logRetentionMs — задайте только одно. При указании обоих приоритет имеет logRetentionMs.

logRetentionMs

Тип: integer | Параметр Kafka: log.retention.ms | Мин: -1 | Пример: 604800000

Время хранения логов с точностью до миллисекунды. Имеет приоритет над logRetentionHours и применяется динамически без перезапуска брокера.
Установите -1, чтобы полностью отключить хранение по времени. Взаимоисключающий с logRetentionHours.

logRetentionBytes

Тип: int-or-string | Параметр Kafka: log.retention.bytes | Пример: 1Gi

Максимальный суммарный размер лога на партицию, по достижении которого удаляются старые сегменты.
Используйте -1, чтобы отключить хранение по размеру.

logSegmentBytes

Тип: int-or-string | Параметр Kafka: log.segment.bytes | Пример: 512Mi

Максимальный размер одного файла сегмента лога. По достижении лимита Kafka открывает новый сегмент.
Сообщение должно помещаться в один сегмент (messageMaxBytes ≤ logSegmentBytes).

messageMaxBytes

Тип: int-or-string | Параметр Kafka: message.max.bytes | Пример: 1Mi

Максимальный размер одного сообщения, принимаемого брокером от производителей.
Не должен превышать logSegmentBytes. Не должен превышать socketRequestMaxBytes, определённый в классе.

compressionType

Тип: string | Параметр Kafka: compression.type
Допустимые значения: producer, uncompressed, gzip, snappy, lz4, zstd

Сжатие сообщений на уровне брокера.
producer (рекомендуется) сохраняет то сжатие, которое применил производитель, избегая повторного сжатия и лишних затрат CPU.

autoCreateTopicsEnable

Тип: boolean | Параметр Kafka: auto.create.topics.enable

Управляет автоматическим созданием топиков при первом обращении производителя или потребителя.
Отключайте в production для явного управления жизненным циклом топиков.

logCleanupPolicy

Тип: string | Параметр Kafka: log.cleanup.policy
Допустимые значения: delete, compact, delete,compact

Политика очистки сегментов лога при превышении лимитов хранения.
delete удаляет старые сегменты; compact оставляет только последнее значение для каждого ключа; delete,compact применяет обе политики.

Примечание: Здесь можно указывать только поля, перечисленные в overridableConfiguration соответствующего KafkaClass. Указание поля, отсутствующего в этом списке, приведёт к ошибке валидации.

TLS

Секция spec.tls включает серверное TLS-шифрование трафика между клиентами и брокером. При включённом TLS брокер принимает клиентские подключения по протоколу SSL на порту 9092.

Предпосылки: в кластере должен быть установлен cert-manager и создан ClusterIssuer или Issuer (при использовании режима CertManager).

Режим CertManager

Оператор автоматически создаёт ресурс Certificate cert-manager, который выпускает и ротирует TLS-сертификат. Сертификат монтируется в pod брокера в формате PEM.

Необходимо указать один из параметров:

  • clusterIssuerName — имя ClusterIssuer (cluster-scoped);
  • issuerName — имя Issuer (в namespace инстанса).
spec:
  tls:
    mode: CertManager
    certManager:
      clusterIssuerName: selfsigned

Режим CustomCertificate

Используйте собственный TLS-сертификат, заранее созданный в виде Secret типа kubernetes.io/tls в namespace инстанса.

spec:
  tls:
    mode: CustomCertificate
    customCertificate:
      serverTLSSecret: my-kafka-tls-secret
      serverCASecret: my-kafka-ca-secret
Поле Описание
serverTLSSecret Имя Secret с ключами tls.crt и tls.key. Сертификат должен содержать SAN с DNS-именами сервиса Kafka.
serverCASecret Имя Secret с ключом ca.crt (CA-сертификат, которым подписан серверный сертификат).

Подключение клиентов через TLS

При включённом TLS клиенты должны подключаться по протоколу SSL. Для проверки сертификата клиенту необходим CA-сертификат.

Получение CA-сертификата (режим CertManager):

kubectl get secret d8ms-kfk-<name>-server-tls-secret -n <namespace> \
  -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt

Подключение через kcat (kafkacat):

kcat -L -b d8ms-kfk-<name>.<namespace>.svc:9092 \
  -X security.protocol=SSL \
  -X ssl.ca.location=ca.crt

Поддерживаемые версии Kafka

Единственная поддерживаемая версия Kafka — 4.2.0

Наши образы для запуска Kafka-контейнеров основаны на distroless-архитектуре.

Статус

Статус сервиса Managed Kafka отражается в ресурсе Kafka.
Структура Conditions однозначно показывает текущий статус сервиса.

Значимые типы:

  • LastValidConfigurationApplied — Агрегирующий тип, который показывает, была ли применена успешно последняя валидная конфигурация хотя бы раз.
  • ConfigurationValid — показывает, прошла ли конфигурация все валидации связанного KafkaClass.
  • ScaledToLastValidConfiguration — показывает, соответствует ли количество работающих реплик заданной конфигурации.
  • Available — показывает, работает ли брокер и принимает ли он подключения.
conditions:
    - lastTransitionTime: '2025-09-22T23:20:36Z'
      observedGeneration: 2
      status: 'True'
      type: Available
    - lastTransitionTime: '2025-09-22T14:38:04Z'
      observedGeneration: 2
      status: 'True'
      type: ConfigurationValid
    - lastTransitionTime: '2025-09-22T14:38:47Z'
      observedGeneration: 2
      status: 'True'
      type: LastValidConfigurationApplied
    - lastTransitionTime: '2025-09-22T23:20:36Z'
      observedGeneration: 2
      status: 'True'
      type: ScaledToLastValidConfiguration

Статус False говорит о проблеме на том или ином этапе либо незавершенной синхронизации состояния.
Для такого состояния будет указан reason и message с описанием.

---
    - lastTransitionTime: '2025-09-23T14:53:33Z'
      message: Syncing
      observedGeneration: 1
      reason: Syncing
      status: 'False'
      type: LastValidConfigurationApplied
    - lastTransitionTime: '2025-09-23T14:54:58Z'
      message: Not all the instances are running still waiting for 1 to become ready
      observedGeneration: 1
      reason: ScalingInProgress
      status: 'False'
      type: ScaledToLastValidConfiguration
---

Примеры использования

Базовое использование

Стандартный брокер Kafka с постоянным хранилищем и конфигурацией по умолчанию.

  1. Создайте namespace с именем kafka.

  2. Создайте ресурс Kafka c дефолтными значениями конфигурации:

    apiVersion: managed-services.deckhouse.io/v1alpha1
kind: Kafka
metadata:
  name: kafka-sample
spec:
  kafkaClassName: default
  instance:
    memory:
      size: "1Gi"
    cpu:
      cores: 2
      coreFraction: "50%"
    persistentVolumeClaim:
      size: "10Gi"
    kubectl apply -f managed-services_v1alpha1_kafka.yaml -n kafka

  3. Подождите, пока брокер будет готов и все conditions примут значение True:

    kubectl get kafka kafka-sample -n kafka -o wide -w

  4. Подключитесь к брокеру через сервис d8ms-kfk-kafka-sample на порту 9092:

    --bootstrap-server d8ms-kfk-kafka-sample:9092

Краткосрочное хранение сообщений

Брокер Kafka с уменьшенным временем хранения логов — подходит для сред разработки или тестирования.

  1. Создайте namespace с именем kafka.

  2. Создайте ресурс Kafka:

    kubectl apply -f managed-services_v1alpha1_kafka.yaml -n kafka
    apiVersion: managed-services.deckhouse.io/v1alpha1
kind: Kafka
metadata:
  name: kafka-dev
spec:
  kafkaClassName: default
  configuration:
    logRetentionHours: 24
    autoCreateTopicsEnable: true
  instance:
    memory:
      size: "1Gi"
    cpu:
      cores: 2
      coreFraction: "25%"
    persistentVolumeClaim:
      size: "5Gi"

  3. Подождите, пока брокер будет готов и все conditions примут значение True:

    kubectl get kafka kafka-dev -n kafka -o wide -w

  4. Подключитесь к брокеру через сервис d8ms-kfk-kafka-dev на порту 9092:

    --bootstrap-server d8ms-kfk-kafka-dev:9092

Брокер с TLS-шифрованием

Брокер Kafka с включённым серверным TLS через cert-manager.

  1. Убедитесь, что в кластере установлен cert-manager и создан ClusterIssuer:

    apiVersion: cert-manager.io/v1
kind: ClusterIssuer
metadata:
  name: selfsigned
spec:
  selfSigned: {}

  2. Создайте namespace и ресурс Kafka с TLS:

    apiVersion: managed-services.deckhouse.io/v1alpha1
kind: Kafka
metadata:
  name: kafka-tls
  namespace: kafka
spec:
  kafkaClassName: default
  tls:
    mode: CertManager
    certManager:
      clusterIssuerName: selfsigned
  configuration:
    logRetentionHours: 168
  instance:
    memory:
      size: "1Gi"
    cpu:
      cores: 2
      coreFraction: "50%"
    persistentVolumeClaim:
      size: "10Gi"
    kubectl apply -f kafka-tls.yaml

  3. Подождите, пока брокер будет готов:

    kubectl get kafka kafka-tls -n kafka -o wide -w

  4. Извлеките CA-сертификат:

    kubectl get secret d8ms-kfk-kafka-tls-server-tls-secret -n kafka \
  -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt

  5. Подключитесь к брокеру через SSL:

    kcat -L -b d8ms-kfk-kafka-tls.kafka.svc:9092 \
  -X security.protocol=SSL \
  -X ssl.ca.location=ca.crt

Пользовательское сжатие и размер сообщений

Брокер Kafka с включённым сжатием gzip и увеличенным максимальным размером сообщения.

  1. Создайте namespace с именем kafka.

  2. Создайте ресурс Kafka:

    kubectl apply -f managed-services_v1alpha1_kafka.yaml -n kafka
    apiVersion: managed-services.deckhouse.io/v1alpha1
kind: Kafka
metadata:
  name: kafka-compressed
spec:
  kafkaClassName: default
  configuration:
    compressionType: gzip
    messageMaxBytes: "10Mi"
    autoCreateTopicsEnable: false
  instance:
    memory:
      size: "2Gi"
    cpu:
      cores: 2
      coreFraction: "50%"
    persistentVolumeClaim:
      size: "20Gi"

  3. Подождите, пока брокер будет готов и все conditions примут значение True:

    kubectl get kafka kafka-compressed -n kafka -o wide -w

  4. Подключитесь к брокеру через сервис d8ms-kfk-kafka-compressed на порту 9092:

    --bootstrap-server d8ms-kfk-kafka-compressed:9092