Datastream

• Datastream - описание соединения между двумя системами, источник и получатель.

• Brooklin позволяет создавать любое количество Datastream для организации различных подключений источника и получателя.

• Для обеспечения масштабируемости Brooklin партиционирует данные в разрезе Datastream либо представляет их как поток с единственной партицией.

• Каждый блок партиционированных данных разбивается на несколько DatastreamTasks с ограниченным количеством партиций для обработки в целях повышения увеличения полосы пропускания.

Connector

• Connector — абстрактное представление модулей для чтения данных.

• Существует несколько реализаций Connector для чтения различных источников.

• Каждый Connector связан с AssignmentStrategy, которая определяет, как разбиваются задачи в разрезе Datastreams и как задачи распределяются между запущенными приложениями Brooklin.

TransportProvider

• TransportProvider — абстрактное представление модулей для записи данных.

• Существует несколько реализаций TransportProvider для записи данных в различные системы.

Coordinator

• Brooklin Coordinator — модуль, отвечающий за управление различными Connector.

• Объект Coordinator может быть только один в каждом запущенном приложении Brooklin.

• В кластере Brooklin Coordinator допускается только один Leader, остальные являются ведомыми.

• Brooklin использует Zookeeper для голосования в выборе Coordinator-лидера. 

• Leader отвечает за распределение задач между Coordinators. 

• Каждый Leader Coordinator связан с AssignmentStrategy для назначения задач Connector в разрезе Datastreams и DatastreamTasks.

• Пример AssignmentStrategy — LoadbalancingStrategy-распределения DatastreamTasks между всеми Coordinators.

Структура

• Каждый кластер Brooklin имеет одну znode верхнего уровня.

• /dms – определения  Datastreams (JSON).

• /connectors

  • sub-znode для каждого зарегистрированного типа Connector;

  • /connectors/<connector-type>/ определения DatastreamTasks, обрабатываемые этим типом Connector, одна znode на  DatastreamTask;

  • два подуровня  /connectors/<connector-type>/<datastreamtask> znode: config and state;

  • config хранит назначения  DatastreamTasks;

  • state хранит информацию о статусе Connectors (offsets/checkpoints).

• /liveinstances

  • каждый экземпляр Brooklin создаёт инкрементальную эфемерную znodes для организации leader Coordinator; 

  • значение каждой sub-znode — это имя хоста соответствующего Brooklin-приложения.

• /instances

  • каждое приложение Brooklin создаёт персистентную znode. Znode представляется в виде имени машины и уникального номера в соответствии с /liveinstances;

  • так же создаются две sub-znode — assignments and errors;

  • assignments всех DatastreamTasks назначенных приложению Brooklin;

  • errors-сообщения об ошибках каждого приложения.

Процесс работы

• Leader Coordinator наблюдает за изменениями в the /dms znode — создание, удаление и изменение Datastreams. 

• Каждый Coorindator наблюдает за изменениями соответствующей ему znode /instances для получения уведомления об изменениях в дочерних znode.

• Leader Coordinator назначает DatastreamTasks приложениям Brooklin путём изменения в assignment znode-ветки /instances/<instance-name>. 

Метрики

Метрики и доступны через JMX exporter (JmxReporter) или файлы CSV (CsvReporter).

brooklin.server.coordinator.cluster=brooklin

brooklin.server.coordinator.zkAddress=bmmzkp1:2181,bmmzkp2:2181,bmmzkp3:2181

brooklin.server.httpPort=32311

brooklin.server.connectorNames=dev_plaintext,stg_plaintext,dev_ssl,stg_ssl

brooklin.server.transportProviderNames=dev_plaintext,stg_plaintext,dev_ssl,stg_ssl

############################# dev_plaintext #############################

brooklin.server.connector.dev_plaintext.factoryClassName=com.linkedin.datastream.connectors.kafka.mirrormaker.KafkaMirrorMakerConnectorFactory

brooklin.server.connector.dev_plaintext.assignmentStrategyFactory=com.linkedin.datastream.server.assignment.BroadcastStrategyFactory

brooklin.server.transportProvider.dev_plaintext.factoryClassName=com.linkedin.datastream.kafka.KafkaTransportProviderAdminFactory

brooklin.server.transportProvider.dev_plaintext.bootstrap.servers=devkafkabroker1:9092

brooklin.server.transportProvider.dev_plaintext.client.id=dev_plaintext-producer

############################# stg_plaintext #############################

brooklin.server.connector.stg_plaintext.factoryClassName=com.linkedin.datastream.connectors.kafka.mirrormaker.KafkaMirrorMakerConnectorFactory

brooklin.server.connector.stg_plaintext.assignmentStrategyFactory=com.linkedin.datastream.server.assignment.BroadcastStrategyFactory

brooklin.server.transportProvider.stg_plaintext.factoryClassName=com.linkedin.datastream.kafka.KafkaTransportProviderAdminFactory

brooklin.server.transportProvider.stg_plaintext.bootstrap.servers=stgkafkabroker1:9092

brooklin.server.transportProvider.stg_plaintext.client.id=stg_plaintext-producer

############################# dev_ssl #############################

brooklin.server.connector.dev_ssl.factoryClassName=com.linkedin.datastream.connectors.kafka.mirrormaker.KafkaMirrorMakerConnectorFactory

brooklin.server.connector.dev_ssl.assignmentStrategyFactory=com.linkedin.datastream.server.assignment.BroadcastStrategyFactory

brooklin.server.connector.dev_ssl.consumer.security.protocol=ssl

brooklin.server.connector.dev_ssl.consumer.ssl.keystore.type=JKS

brooklin.server.connector.dev_ssl.consumer.ssl.truststore.type=JKS

brooklin.server.connector.dev_ssl.consumer.ssl.keystore.location=/opt/brooklin/config/dev_ssl.keystore.jks

brooklin.server.connector.dev_ssl.consumer.ssl.keystore.password=password

brooklin.server.connector.dev_ssl.consumer.ssl.truststore.location=/opt/brooklin/config/dev_ssl.keystore.jks

brooklin.server.connector.dev_ssl.consumer.ssl.truststore.password=password

brooklin.server.transportProvider.dev_ssl.factoryClassName=com.linkedin.datastream.kafka.KafkaTransportProviderAdminFactory

brooklin.server.transportProvider.dev_ssl.bootstrap.servers= devkafkabroker1:9093

brooklin.server.transportProvider.dev_ssl.client.id=dev_ssl-producer

brooklin.server.transportProvider.dev_ssl.security.protocol=ssl

brooklin.server.transportProvider.dev_ssl.ssl.keystore.type=JKS

brooklin.server.transportProvider.dev_ssl.ssl.truststore.type=JKS

brooklin.server.transportProvider.dev_ssl.ssl.keystore.location=/opt/brooklin/config/dev_ssl.keystore.jks

brooklin.server.transportProvider.dev_ssl.ssl.keystore.password=password

brooklin.server.transportProvider.dev_ssl.ssl.truststore.location=/opt/brooklin/config/dev_ssl.keystore.jks

brooklin.server.transportProvider.dev_ssl.ssl.truststore.password=password

############################# stg_ssl #############################

brooklin.server.connector.stg_ssl.factoryClassName=com.linkedin.datastream.connectors.kafka.mirrormaker.KafkaMirrorMakerConnectorFactory

brooklin.server.connector.stg_ssl.assignmentStrategyFactory=com.linkedin.datastream.server.assignment.BroadcastStrategyFactory

brooklin.server.connector.stg_ssl.consumer.security.protocol=ssl

brooklin.server.connector.stg_ssl.consumer.ssl.keystore.type=JKS

brooklin.server.connector.stg_ssl.consumer.ssl.truststore.type=JKS

brooklin.server.connector.stg_ssl.consumer.ssl.keystore.location=/opt/brooklin/config/stg_ssl.keystore.jks

brooklin.server.connector.stg_ssl.consumer.ssl.keystore.password=password

brooklin.server.connector.stg_ssl.consumer.ssl.truststore.location=/opt/brooklin/config/stg_ssl.keystore.jks

brooklin.server.connector.stg_ssl.consumer.ssl.truststore.password=password

#brooklin.server.connector.stg_ssl.consumer.ssl.key.password=

brooklin.server.transportProvider.stg_ssl.factoryClassName=com.linkedin.datastream.kafka.KafkaTransportProviderAdminFactory

brooklin.server.transportProvider.stg_ssl.bootstrap.servers=s stgkafkabroker1:9093

brooklin.server.transportProvider.stg_ssl.client.id=stg_ssl-producer

brooklin.server.transportProvider.stg_ssl.security.protocol=ssl

brooklin.server.transportProvider.stg_ssl.ssl.keystore.type=JKS

brooklin.server.transportProvider.stg_ssl.ssl.truststore.type=JKS

brooklin.server.transportProvider.stg_ssl.ssl.keystore.location=/opt/brooklin/config/stg_ssl.keystore.jks

brooklin.server.transportProvider.stg_ssl.ssl.keystore.password=password

brooklin.server.transportProvider.stg_ssl.ssl.truststore.location=/opt/brooklin/config/stg_ssl.keystore.jks

brooklin.server.transportProvider.stg_ssl.ssl.truststore.password=password

# Просмотр datastreams 

curl -s http://brooklin:32311/datastream | jq

# Создание datastream

curl -X POST -H "Content-Type: application/json" -d @opendatacollecting http://brooklin:32311/datastream

cat <<EOF > opendatacollecting
{
 "metadata": {
  "owner": "test-user", - владелец задачи
  "group.id": "bmm_opendatacollecting ", - название консьюмер группы
  "system.reuseExistingDestination": "false",
  "system.IsConnectorManagedDestination": "true",
  "system.IsUserManagedDestination": "true",
  "system.destination.identityPartitioningEnabled": "false", - точное соответствие данных в партициях при зеркалировании
  "system.auto.offset.reset": "latest" – c какого места начинаем читать топик
 },
 "transportProviderName": "stg_ssl", - название транспортного провайдера
 "destination": {
     "connectionString": "kafkassl://stgkafkabroker1:9093,stgkafkabroker2:9093/opendatacollecting_product" –тип подключения (SSL),список брокеров, имя топика destination
 },
 "name": "TICKET-1374", - название datastream
 "connectorName": "dev_ssl", - название коннектора
 "source": {
     "connectionString": "kafkassl://devkafkabroker1:9093,devkafkabroker2:9093/opendatacollecting " – тип подключения (SSL),список брокеров, имя топика source
 }
}
EOF

# Пауза datastream

curl -X POST http://brooklin:32311/datastream/TICKET-1374?action=pause

# Запуск datastream

curl -X POST http://brooklin:32311/datastream/TICKET-1374?action=resume

# Удаление datastream

curl -X DELETE http://brooklin:32311/datastream/TICKET-1374 

# Пауза консьюма партиций (1-19) from 20 (остается консьюм 0 партиции) задача должна быть в статусе READY

curl -X POST -H "Content-Type: application/json" -d @stopopendatacollecting http:// brooklin:32311/datastream/TICKET-1374?action=pauseSourcePartitions
 
cat <<EOF > stopopendatacollecting
{
 "sourcePartitions" : {
  "opendatacollecting ": "1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19"
 }
}
EOF

# Запуск консьюма остановленных партиций

curl -X POST -H "Content-Type: application/json" -d  http:/brooklin:32311/datastream/TICKET-1374?action=resumeSourcePartitions

rules:
 # Base pod metrics
 - pattern: 'java.lang<type=OperatingSystem><>(FreePhysicalMemorySize|TotalPhysicalMemorySize|SystemCpuLoad|ProcessCpuLoad|OpenFileDescriptorCount|AvailableProcessors|TotalStartedThreadCount)'
  name: java_lang_OperatingSystem_$1
  type: GAUGE

 # Consumer metrics by tasks
 - pattern: 'kafka.consumer<type=consumer-metrics, client-id=(.+)><>(connection-count|consumer-metadata-request-rate|failed-authentication-rate|incoming-byte-rate|network-io-rate|outgoing-byte-rate|request-rate|response-rate)'
  name: kafka_consumer_$2
  labels:
   clientid: "$1"

 # Brooklin metrics
 - pattern: metrics<name=Coordinator.isLeader><>Value
  name: brooklin_isleader
 - pattern: metrics<name=Coordinator.numRebalances><>FiveMinuteRate
  name: brooklin_numRebalances_FiveMinuteRate
 # devkafka.numDatastreamTasks
 - pattern: metrics<name=(\w+).numDatastreamTasks><>Value
  name: brooklin_numDatastreamTasks
  labels:
   connector: "$1"
 # Kafka Connector Metrics
 # devkafka.KafkaConnectorTask.dev-mirroring-2.numTopics
 - pattern: metrics<name=(\w+).KafkaConnectorTask.(.+).numTopics><>Value
  name: brooklin_KafkaConnectorTask_numTopics
  labels:
   connector: "$1"
   task: "$2"
 # devkafka.KafkaConnector.numTaskRestarts
 - pattern: metrics<name=(\w+).KafkaConnector.numTaskRestarts><>Value
  name: brooklin_KafkaConnector_numTaskRestarts
  labels:
   connector: "$1"
 - pattern: metrics<name=(\w+).KafkaConnectorTask.(.+).errorRate><>(FifteenMinuteRate)
  name: brooklin_KafkaConnectorTask_errorrate_FiveMinuteRate
  labels:
   connector: "$1"
   task: "$2"
 # devkafka.KafkaConnectorTask.aggregate.numPartitions
 # devkafka.KafkaConnectorTask.dev-mirroring-2.numPartitions
 - pattern: metrics<name=(\w+).KafkaConnectorTask.(.+).numPartitions><>Value
  name: brooklin_KafkaConnectorTask_numPartitions
  labels:
   connector: "$1"
   task: "$2"
 # devkafka.KafkaConnectorTask.aggregate.rebalanceRate Count
 # devkafka.KafkaConnectorTask.dev-mirroring-2.rebalanceRate
 - pattern: metrics<name=(\w+).KafkaConnectorTask.(.+).rebalanceRate><>Count
  name: brooklin_KafkaConnectorTask_rebalanceRate
  labels:
   connector: "$1"
   task: "$2"
 # devkafka.KafkaConnectorTask.aggregate.stuckPartitions
 - pattern: metrics<name=(\w+).KafkaConnectorTask.(.+).stuckPartitions><>Value
  name: brooklin_KafkaConnectorTask_stuckPartitions
  labels:
   connector: "$1"
   task: "$2"
 # devkafka.KafkaConnectorTask.dev-mirroring-2.eventsByteProcessedRate
 - pattern: metrics<name=(\w+).KafkaConnectorTask.(.+).eventsByteProcessedRate><>Count
  name: brooklin_KafkaConnectorTask_eventsByteProcessedRate
  labels:
   connector: "$1"
   task: "$2"
 # devkafka.KafkaConnectorTask.dev-mirroring-2.numPolls
 - pattern: metrics<name=(\w+).KafkaConnectorTask.(.+).numPolls><>FiveMinuteRate
  name: brooklin_KafkaConnectorTask_numPolls_FiveMinuteRate
  labels:
   connector: "$1"
   task: "$2"
 # Kafka Mirror Maker Connector metrics
 - pattern: metrics<name=(\w+).KafkaMirrorMakerConnectorTask.(.+).numTopics><>Value
  name: brooklin_KafkaMirrorMakerConnectorTask_numTopics
  labels:
   connector: "$1"
   task: "$2"
 # metrics:name=devkafka.KafkaMirrorMakerConnector.numTaskRestarts
 - pattern: metrics<name=(\w+).KafkaMirrorMakerConnector.numTaskRestarts><>Value
  name: brooklin_KafkaMirrorMakerConnector_numTaskRestarts
  labels:
   connector: "$1"
 - pattern: metrics<name=(\w+).KafkaMirrorMakerConnectorTask.(.+).errorRate><>(FifteenMinuteRate)
  name: brooklin_KafkaMirrorMakerConnectorTask_errorrate_FiveMinuteRate
  labels:
   connector: "$1"
   task: "$2"
 - pattern: metrics<name=(\w+).KafkaMirrorMakerConnectorTask.(.+).numPartitions><>Value
  name: brooklin_KafkaMirrorMakerConnectorTask_numPartitions
  labels:
   connector: "$1"
   task: "$2"
 - pattern: metrics<name=(\w+).KafkaMirrorMakerConnectorTask.(.+).rebalanceRate><>Count
  name: brooklin_KafkaMirrorMakerConnectorTask_rebalanceRate
  labels:
   connector: "$1"
   task: "$2"
 - pattern: metrics<name=(\w+).KafkaMirrorMakerConnectorTask.(.+).stuckPartitions><>Value
  name: brooklin_KafkaMirrorMakerConnectorTask_stuckPartitions
  labels:
   connector: "$1"
   task: "$2"
 - pattern: metrics<name=(\w+).KafkaMirrorMakerConnectorTask.(.+).eventsByteProcessedRate><>Count
  name: brooklin_KafkaMirrorMakerConnectorTask_eventsByteProcessedRate
  labels:
   connector: "$1"
   task: "$2"
 - pattern: metrics<name=(\w+).KafkaMirrorMakerConnectorTask.(.+).numPolls><>FiveMinuteRate
  name: brooklin_KafkaMirrorMakerConnectorTask_numPolls_FiveMinuteRate
  labels:
   connector: "$1"
   task: "$2"
 # Kafka Mirror Maker Connector metrics
 - pattern: metrics<name=(\w+).KafkaMirrorMakerConnectorTask.(.+).eventsProcessedRate><>Count
  name: brooklin_KafkaMirrorMakerConnectorTask_eventsProcessedRate
  labels:
   connector: "$1"
   task: "$2"
 # Producer metrics
 - pattern: metrics<name=EventProducer.(.+).eventProduceRate><>Count
  name: brooklin_EventProducer_eventProduceRate
  labels:
   connector: "$1"
 - pattern: metrics<name=EventProducer.(.+).totalEventsProduced><>Count
  name: brooklin_EventProducer_totalEventsProduced
  labels:
   connector: "$1"
 # Topic metrics
 - pattern: metrics<name=KafkaTransportProvider.(.+).eventByteWriteRate><>Count
  name: brooklin_KafkaTransportProvider_eventByteWriteRate
  labels:
   topic: "$1"
 - pattern: metrics<name=KafkaTransportProvider.(.+).eventWriteRate><>Count
  name: brooklin_KafkaTransportProvider_eventWriteRate
  labels:
   topic: "$1"