Kafka on Kubernetes multi-node

Question

Итак, моя цель - создать кластер из нескольких кафка-брокеров в распределенной форме. Но я не вижу способа заставить брокеров знать друг о друге.

Насколько я понимаю, каждый брокер нуждается в отдельном идентификаторе в своей конфигурации, который я не могу гарантировать или настроить, если я запускаю контейнеры из кубернетов?

Они также должны иметь одинаковый рекламируемый_host?

Есть ли какие-либо параметры, которые мне не хватает, которые нужно изменить для того, чтобы узлы могли обнаружить друг друга?

Было бы целесообразно выполнить такую ​​конфигурацию в конце файла Docker со сценарием? И/или общий том?

В настоящее время я пытаюсь сделать это с помощью spotify/kafka-image, который имеет предварительно сконфигурированную комбинацию zookeeper + kafka, на ванильном кубернете.

Answer 1

Мое решение для этого было использовать IP в качестве идентификатора: обрезать точки и получить уникальный идентификатор, который также доступен за пределами контейнера в других контейнерах.

С услугами вы можете получить доступ к IP-адресу нескольких контейнеров (см моего ответа здесь о том, как это сделать:. what's the best way to let kubenetes pods communicate with each other?

так что вы можете получить их идентификаторы тоже, если вы используете IP-адрес в качестве уникального идентификатора Единственная проблема в том, что идентификаторы не являются непрерывными или начинаются с 0, но Zookeeper/Кафка не возражал

EDIT 1:.

следования проблем настройки Zookeeper:

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

Эта служба должна быть запущена ПЕРЕД созданием ReplicationController (RC) модулей Zookeeper.

Сценарий запуска контейнера ZK будет необходимо:

• ожидания для службы обнаружения, чтобы заселить ZK службы с ее узлами (что занимает всего несколько секунд, пока я просто добавить спать 10 в начале моего сценария запуска, но более надежно, вы должны искать, чтобы в нем было как минимум 3 узла.)
• искать контейнеры, формирующие Сервис в службе обнаружения: это делается путем запроса API. переменная среды KUBERNETES_SERVICE_HOST доступна в каждом контейнере. Конечная точка, чтобы найти описание сервиса является то

URL="http(s)://$USERNAME:$PASSWORD@${KUBERNETES_SERVICE_HOST/api/v1/namespaces/${NAMESPACE}/endpoints/${SERVICE_NAME}"

где NAMESPACE является default, если вы не изменили его, и SERVICE_NAME будет, если вы зоопарка назвали ваш сервис зоопарка.

там вы получите описание контейнеров, образующих Сервис, с их ip в поле «ip». Вы можете сделать:

curl -s $URL | grep '\"ip\"' | awk '{print $2}' | awk -F\" '{print $2}' 

, чтобы получить список IP-адресов в службе. При том, что заполнение zoo.cfg на узле, используя идентификатор, определенный выше

Вы, возможно, потребуется USERNAME и ПАРОЛЬ достичь конечной точки на услуги, такие как Google контейнерной двигателя. Они должны быть помещены в Тайне объема (см документа здесь: http://kubernetes.io/v1.0/docs/user-guide/secrets.html)

Вы также должны использовать curl -s --insecure на Google Container Engine, если вы идете через трудность добавления сертификата CA к вашим стручкам

В основном добавьте том в контейнер и найдите значения из файла. (Вопреки тому, что док говорит, НЕ положить \ п в конце имени пользователя и пароля при кодировании base64: это просто сделать вашу жизнь более сложной при чтении тех)

EDIT 2:

Еще одна вещь, которую вам нужно сделать на узлах Kafka, - получить IP и имена хостов и поместить их в файл/etc/hosts. Кафка, кажется, нужно знать узлы от имен хостов, и они не установлены в сервисных узлах по умолчанию

EDIT 3:

После долгих проб и мысль, используя IP-адрес в качестве идентификатора не могут быть настолько велики, : это зависит от того, как вы настраиваете хранилище. для любого вида распределенного сервиса, такого как zookeeper, kafka, mongo, hdfs, вы можете использовать тип хранилища emptyDir, поэтому он находится именно на этом узле (монтируя удаленный тип хранилища, он поражает цель распространения этих сервисов!) emptyDir будет relaod с данными на одном и том же узле, поэтому представляется логичным использовать идентификатор NODE (IP-адрес узла) в качестве идентификатора, потому что тогда модуль, который перезагружается на том же узле, будет иметь данные. Это предотвращает потенциальное повреждение данных (если новый узел начинает писать в том же каталоге, который на самом деле не пуст, кто знает, что может произойти), а также с Kafka, тем, которым назначается broker.id, если изменяется идентификатор брокера , zookeeper не обновляет тему broker.id, и тема выглядит так, как будто она доступна. НО указывает на неправильный broker.id, и это беспорядок.

До сих пор я еще не нашел, как получить IP-адрес узла, но я думаю, что это можно искать в API, просматривая имена сервисных имен, а затем узел, на котором они развернуты.

EDIT 4

Чтобы получить IP-адрес узла, вы можете получить имя хоста стручок == имя из конечных точек API /API/v1/пространств имен/по умолчанию/оконечных/ , как объяснено выше. , то вы можете получить IP-адрес узла от имени стручка с /API/v1/пространства имен/по умолчанию/стручков/

PS: это вдохновляет пример в репо Kubernetes (например, для rethinkdb здесь: https://github.com/kubernetes/kubernetes/tree/master/examples/rethinkdb

Answer 2

Я сделал это с помощью докер-Compose (разница для Kubernetes бы, что бы вы передать идентификатор через ваш service.yaml и есть 2 услуги):

kafka1: 
    build: kafka-0.8.1/ 
    ports: 
    - 9092 
    links: 
    - zookeeper 
    environment: 
    - ID=1 
kafka2: 
    build: kafka-0.8.1/ 
    ports: 
    - 9092 
    links: 
    - zookeeper 
    environment: 
    - ID=2 

Config:

broker.id=${ID} 
port=9092 
advertised.host.name=${HOST} 
advertised.port=9092 
num.network.threads=3 
num.io.threads=8 
socket.send.buffer.bytes=102400 
socket.receive.buffer.bytes=102400 
socket.request.max.bytes=104857600 
log.dirs=/kafka/kafka-logs-${ID} 
num.partitions=200 
num.recovery.threads.per.data.dir=1 
log.retention.hours=168 
log.segment.bytes=1073741824 
log.retention.check.interval.ms=300000 
log.cleaner.enable=false 
zookeeper.connect=${DOCKER_ZOOKEEPER_1_PORT_2181_TCP_ADDR}:${DOCKER_ZOOKEEPER_1_PORT_2181_TCP_PORT} 
zookeeper.connection.timeout.ms=6000 

ш:

#!/bin/bash 
echo "Running config" 
export HOST=`grep $HOSTNAME /etc/hosts | awk '{print $1}'` 
export ID=${ID:?} 
perl -p -i -e 's/\$\{([^}]+)\}/defined $ENV{$1} ? $ENV{$1} : $&/eg' </broker.template> $KAFKA_HOME/config/server.properties 
echo "Done" 
echo "starting kafka with:" 
echo "$KAFKA_HOME/config/server.properties" 
echo "" 
cat $KAFKA_HOME/config/server.properties 
$KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties 

Answer 3

Посмотрите на https://github.com/CloudTrackInc/kubernetes-kafka Это позволяет начать Кафка в kubernetes и поддержке масштабирования его и авто extanding.

Answer 4

Это проявляется заметно в моих поисках, но содержит довольно устаревшую информацию. для того, чтобы обновить это с более современным решением, вы должны используйте развертывание StatefulSet, которое будет генерировать модули, которые имеют целочисленный счетчик вместо хеша в их имени, например. Кафка-контроллер-0.

Это, конечно, имя хоста, так оттуда это простой вопрос, чтобы извлечь фиксированный инвариантную брокера ID используя AWK:

hostname | awk -F'-' '{print $3}' 

Самые популярные контейнеры доступные для Кафки эти дни команды брокер ID ,