Руководство прикладного разработчика#

Термины и определения#

Системные требования#

Для установки программного компонента Kafka Gateway (далее — Шлюза Kafka) в составе прикладного дистрибутива должно быть выполнено следующее:

1. Развернут и настроен кластер Kubernetes 1.17+ или Red Hat OpenShift 4.5.13+ в соответствии с требованиями, предъявляемыми к Платформе.

2. В кластере создан проект (namespace), в котором будет разворачиваться шлюз.

3. В проекте создана учетная запись с правами на загрузку артефактов.

4. Docker-образ шлюза размещен в целевом Docker-репозитории по ссылке, указанной в конфигурационном артефакте Deployment.

5. В проект добавлен Secret для загрузки Docker-образов из целевого Docker-репозитория.

6. Подготовлен комплект настроенных конфигурационных артефактов.

7. Брокеры Kafka настроены и готовы к работе, есть доступ к ним со стороны кластера Kubernetes или Red Hat OpenShift.

8. Созданы целевые топики сообщений, необходимый доступ к ним предоставлен.

9. В проекте имеются свободные ресурсы по лимитам и реквестам не менее, чем зарезервировано в конфигурационных артефактах.

10. При установке с использованием консоли, на рабочем месте должен быть установлен клиент Kubernetes или Red Hat OpenShift.

Минимальные ресурсы, необходимые для запуска Шлюза Kafka

При опциональном использовании совместно со Шлюзом Kafka Сайдкара компонента Сервисный прокси (входящего в состав продукта «Platform V Synapse Service Mesh») минимальные ресурсы, необходимые для запуска, следует уточнить в соответствующей документации используемого компонента.

Общие сведения#

Компонент Шлюз Kafka предназначен для организации взаимодействия компонентов Synapse с внешними АС с помощью топиков Kafka по протоколу GRPC или REST.

Подключение к брокерам Kafka осуществляется без egress-gateway Istio.

Подключение и конфигурирование#

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

2. Последовательность загрузки артефактов:

   • Config Map c application.yml

   • Secret c ssl.yml

   • Secret c ключами

   • Service Entry

   • Deployment

   • Service

3. Порядок действий при загрузке артефактов в проект описаны в руководстве по установке .

В случае использования SecMan, все конфигурации с типом Secret должны храниться в SecMan.

Карточка адаптера#

Полное описание конфигурации (/deployments/config/application.yml)

logging.file.level: <OFF/INFO/DEBUG, off - логирование в файл отключено, info - в файл логируется информация без тела сообщения, debug - в файл логируется вся информация, включая тело сообщения>
logPath: <Путь до файла с логом, по-умолчанию /opt/synapse/logs/messages>
grpc:
  server:
    server-port: <Порт, на котором будет поднят GRPC сервер>
  client:
    settings:
      default:
        hostname: <Имя сервиса в Kubernetes или Red Hat OpenShift, которое должен вызвать адаптер после вычитки сообщения из Kafka>
        port: <Порт сервиса в Kubernetes или Red Hat OpenShift>
	  signature-verification-failed-url: # вызов API при ошибке проверки подписи сообщения
        hostname: <Имя сервиса в Kubernetes или Red Hat OpenShift, которое должен вызвать адаптер после вычитки сообщения из Kafka>
        port: <Порт сервиса в Kubernetes или Red Hat OpenShift>
server:
  port: <Порт, на котором будет поднят Spring Tomcat, используется для проверки жизнеспособности приложения>
  tomcat:
    max-swallow-size: <Максимальный размер сообщения, которые может быть принять по REST API, например 40MB>
    threads:
      max: <Максимальное кол-во потоков для обработки REST вызовов>
      min-spare: <Минимальное кол-во потоков для обработки REST вызовов, которое всегда будет доступно>
logging:
  level:
    org:
      springframework: OFF/ERROR/INFO/DEBUG		 	                        # уровень логирования для springframework
    root: OFF/ERROR/INFO/DEBUG					  	                        # уровень логирования для root логгера
    com.sbt.synapse.kafka.service.api.BulkController: OFF/ERROR/INFO/DEBUG  # уровень логирования для контроллера обработки batch сообщений
    com.sbt.synapse.kafka.service.KafkaService: OFF/ERROR/INFO/DEBUG        # уровень логирования для модуля отправки сообщений в Kafka
    com.sbt.synapse.kafka.integration.utils.BulkUtils: OFF/ERROR/INFO/DEBUG # уровень логирования для обработчика batch сообщений
spring:
  main:
    banner-mode: off/console/log					# управление вывода в журнал баннера spring
  profiles: production								# используемый профиль приложения spring
  kafka:
    rest:                       
      mode: true/false                				# настройка адаптера для работы по REST
      timeout: 30000            					# таймаут в мс для REST-вызова
      max-connections: 100      					# размер пула коннектов для REST-вызовов (при чтении из топика)
      default-url: "http://localhost:1414/test"   	# URL, который будет вызван по-умолчанию
      signature-verification-failed-url: "http://localhost:1414/fail"   	# URL, который будет вызван при ошибке проверки подписи сообщения
      default-method: get/post       		    	# метод, который будет использован
      controller: /receive                          # путь, по которому будет доступен вызов контроллера API для записи в топик, по умолчанию "/". >
    ssl:
      key-store-location: <Путь до keystore, оставить пустым, если TLS не требуется>
      trust-store-location: <Путь до truststore, оставить пустым, если TLS не требуется>
      protocol: <Версия протокола SSL подключения, оставить пустым, если TLS не требуется>
    producer: <Опциональный блок, если не задан, то сообщения в топик не отправляются>
      sensitive-headers: <Перечисление через запятую заголовков, регистр которых важен при отправке kafka-сообщения. Передаваемые заголовки сравниваются по equalsIgnoreCase, и при совпадении приводятся к указанному виду.>
      batch:
        separator: <Разделитель, с помощью которого будут разделены пришедшие сообщения, по умолчанию \n>
        licenseResponsePath: <Путь до файла с ответом при проверке доступности шлюза сервисом filebeat>
        skipList: <Список для фильтрации сообщений. Если сообщение содержит хотя бы одно вхождение из списка "skipList", то оно пропускается>
          - <string, например "\"_index\"">
      bootstrap-servers: <Server:Port брокера Kafka, в который требуется помещать сообщения>
      topic: <Имя топика, в который требуется помещать сообщения>
      key-serializer: org.apache.kafka.common.serialization.StringSerializer 		# сериализатор для ключей сообщений в Kafka
      value-serializer: org.apache.kafka.common.serialization.StringSerializer      # сериализатор для сообщений в Kafka
      properties:
        # Подключить интерсептор (Producer)
        interceptor.classes: ru.sbt.ss.kafka.interceptors.KafkaSignatureProducerInterceptor
        interceptor.signature.certificate.mode: <public-key или certificate-serial, public-key - использование публичного ключа для проверки подписи на стороне потребителя (то есть в заголовках передается ключ), certificate-serial - использование серийного номера сертификата для проверки подписи (то есть в заголовках передается серийный номер)>
        # Все настройки ниже опциональны:
        # Настроить алгоритм подписи, поддерживается только алгоритм SHA256-RSA
        interceptor.signature.certificate.algorithm: SHA256WithRSA      
        # Настроить хранилище сертификатов
        # Интерсептор может использовать отдельное хранилище сертификатов:
        interceptor.signature.certificate.ssl.keystore.location = ssl/keystore.jks
        # Выбрать сертификат из хранилища, указав его alias (по умолчанию используется первый)
        interceptor.signature.certificate.key.alias =
        # Настроить имена системных заголовков
        # Имя заголовка с подписью сообщения
        interceptor.signature.header = signature
        # Префикс для имен системных заголовков
        interceptor.signature.attributes.prefix = signature.
        # Имя загловка с используемым алгоритмом подписи
        interceptor.signature.certificate.algorithm.header=algorithm
        # Имя заголовка с публичным ключом для проверки подписи
        interceptor.signature.certificate.public.key.header=public.key
    consumer: <Опциональный блок, если не задан, то сообщения из топика не вычитываются>
      transactional: <true/false - если true, то в случае ошибки доставки сообщения фиксировать ли offset топика Kafka (в этом случае адаптер будет бесконечно пробовать читать недоставленные сообщения и отправлять по grpc/rest)>
      repeat-after-error-ms: <в случае transactional=true, период повторной попытки вычитки и отправки сообщения при ошибке>
      bootstrap-servers: <Server:Port брокера Kafka, из которого требуется получать сообщения>
      topics: <Список топиков, по которым необходимо выполнять чтение>
        - topic: <Имя топика>
          group-id: <Имя consumer группы>
          enabled: <true/false - включен ли слушатель по данному топику, можно изменять в runtime>
          ratelimiter: <Настройки лимиттера, если включен, то подключается ограничитель по полученным сообщениям в секунду tpsPerSecond (если не задан tpsPerSecond, то используется tpsPerMinute) и шлюз не читает больше данного порога>
            tpsPerMinute: <Максимальное кол-во сообщений в минуту, которое может прочитать шлюз>
            tpsPerSecond: <Максимальное кол-во сообщений в секунду, которое может прочитать шлюз (приортетнее чем tpsPerMinute, если заданы оба значения)>
            enabled: <true/false - включен ли лимиттер по данному топику>
      key-serializer: org.apache.kafka.common.serialization.StringSerializer        # сериализатор для ключей сообщений в Kafka
      value-serializer: org.apache.kafka.common.serialization.StringSerializer      # сериализатор для сообщений в Kafka
      batching: <true/false - Режим чтения и отправки сообщений пачками>
      properties: <Список настроек консьюмера, ключ значение>
      	# Подключить интерсептор (Consumer)
        interceptor.classes: ru.sbt.ss.kafka.interceptors.KafkaSignatureConsumerInterceptor,com.sbt.synapse.kafka.service.SignatureValidateConsumerInterceptor
        interceptor.signature.service.class: ru.sbt.ss.kafka.interceptors.X509SignatureService
        # Все настройки ниже опциональны:
        # Настроить режим работы консюмера при ошибках (по умолчанию failOnValue и только в этом режиме поддерживается передача сообщения дальше на API)
        interceptor.signature.mode: #failOnConsume|filter|failOnValue
        # Настроить удаление системных заголовков из сообщения
        interceptor.signature.remove.headers = false
        # Настроить имена системных заголовков
        # Имя заголовка с подписью сообщения
        interceptor.signature.header = signature
        # Префикс для имен системных заголовков
        interceptor.signature.attributes.prefix = signature.
        # Имя загловка с используемым алгоритмом подписи
        interceptor.signature.certificate.algorithm.header=algorithm
        # Имя заголовка с публичным ключом для проверки подписи
        interceptor.signature.certificate.public.key.header=public.key
      filter:             # Не поддерживается для батчей
        rules:            # <Имя топика>:<условие фильтра в SPEL, если удовлетворяет условию фильтра, то сообщение отправляется, иначе - нет>
          'SrvDistributeMergePrivateClient.Request': "#headers.get('x-synapse-type') == 'CASHIN' && #headers.get('HW_ERROR') == 'true'"
          'SrvDistributeMergePrivateClient.Response': '#HW_ERROR == "true"'
    properties:
      security.protocol: <Протокол подключения, оставить пустым, если TLS не требуется>
      ssl.endpoint.identification.algorithm: "" # отключение проверки серверного сертификата
      ssl.enabled.protocols: TLSv1.2            # список протоколов разрешенных к использованию при создании подключения
transform-xml-json:
  services:
    srvupdateanddedubclient:  #название сервиса из x-synapse-operationname
      transform: true         #<Нужно ли трансформировать выполнять трансформацию JSON-XML> # не является целевым вариантом использования шлюза, трансформацию необходимо выполнять на сервисе трансформации
      unTypeFieldsXmlToJson:
        SPName: spName
      unTypeFieldsJsonToXml:
        personUpdateStatuses: PersonUpdateStatus
  encodeBodyToBase64: false   # Кодировать ли тело сообщения, сразу после получения из Kafka в base64. Работает только при получении сообщений из Kafka в одиночном режиме. Может быть полезно для работы с бинарными сообщениями. Другие трансфорамации не будут работать. По умолчанию false.
topics-routing: # правила, по которым необходимо маршрутизировать сообщения вычитанные из топика в формате <имя топика>: <маршрут>
  routes:
    <topic_name>: service1
    <topic_name2>: service2
routing:
  # правила для парсинга сообщения входящего по grpc/rest
  api:
    # правило определения маршрута
    routeName:
      valueFrom:
      # значение маршрута будет взято из тела сообщения по путь $.header.evtType или из значение заголовока x-synapse-operationname
      # приоритет отдается первому определенному
        - type: fromBody
          value: $.header.evtType
        - type: fromHeader
          value: x-synapse-operationname
    # список маршрутов по routeName, полученному в предыдущем блоке routeName.valueFrom
    # если не получилось найти соответствие routeName по маршрутам, то будет использован маршрут default
    routeList:
      - routeName: process
        # SPEL выражение, куда отправлять сообщение
        destinationExpression: ('sytester_rq_process')
        variables:
        # список переменных, на основе которых можно построить destinationExpression и заполнить x-synapse-rquid/x-synapse-operationname
          - name: OperationName
            valueFrom:
              - type: fromBody
                value: $.header.evtType
              - type: fromHeader
                value: x-synapse-operationname
              - type: fromConst
                value: const
          - name: RqUID
            valueFrom:
              - type: fromBody
                value: $.header.evtId
      - routeName: cancel
        destinationExpression: ('pprbacq-' + variables['OperationName'] + '-rq')
        variables:
          - name: OperationName
            valueFrom:
              - type: fromBody
                value: $.header.evtType
          - name: RqUID
            valueFrom:
              - type: fromBody
                value: $.header.evtId
      - routeName: default
        destinationExpression: ('sytester_rq')
        variables:
          - name: OperationName
            valueFrom:
              - type: fromBody
                value: $.header.evtType
          - name: RqUID
            valueFrom:
              - type: fromBody
                value: $.header.evtId
  #правила для парсинга сообщения вычитанного из Kafka-топика
  kafka:
    routeName:
      valueFrom:
        - type: fromBody
          value: $.header.evtType
    routeList:
      - routeName: AepPublisher
        destinationExpression: ('http://sytester-http-stub:8237/AepPublisher')
        variables:
          - name: OperationName
            valueFrom:
              - type: fromBody
                value: $.header.evtType
          - name: RqUID
            valueFrom:
              - type: fromBody
                value: $.header.evtId
          - name: Header
            valueFrom:
              - type: fromHeader
                value: test-header
          - name: Const
            valueFrom:
              - type: fromConst
                value: test-const
      - routeName: AepPublisher2
        destinationExpression: ('http://sytester-http-stub:8237/AepPublisher2')
        variables:
          - name: OperationName
            valueFrom:
              - type: fromBody
                value: $.header.evtType
          - name: RqUID
            valueFrom:
              - type: fromBody
                value: $.header.evtId
          - name: Header
            valueFrom:
              - type: fromHeader
                value: test-header
          - name: Const
            valueFrom:
              - type: fromConst
                value: test-const
      - routeName: default
        destinationExpression: ('http://sytester-http-stub:8237/default')
        variables:
          - name: OperationName
            valueFrom:
              - type: fromBody
                value: $.header.evtType
          - name: RqUID
            valueFrom:
              - type: fromBody
                value: $.header.evtId
          - name: Header
            valueFrom:
              - type: fromHeader
                value: test-header
          - name: Const
            valueFrom:
              - type: fromConst
                value: test-const
tracing:
  addPrefixForHeaders: <true/false - Нужно ли добавлять префикс x-synapse к HTTP/GRPC заголовкам, которые не имеют префикса x-, по-умолчанию true>
  generateXB3Headers: <true/false - Нужно ли генерировать XB3 заголовки>
  generate128bitTraceId: <true/false - Формировать в 128bit виде>
  tracingHeaders:   
    <Название заголовка, который необходимо добавить в GRPC/REST заголовок>: {type: <Правило получение значения (fromBody, fromConst)>, value: <Значение>}
  excludeHeaders:
  	api:
      <Названия заголовков, которые необходимо пропустить, или all для исключения всех заголовков при вызове адаптера по HTTP/GRPC>
    kafka:
      <Названия заголовков, которые необходимо пропустить, или all для исключения всех заголовков при чтении из Kafka-топика (в случае all не будет доступна фильтрация сообщений)>

**Описание ssl конфигурации (/deployments/config/ssl/ssl.yml) - должна быть определена как Secret или **

spring:
  kafka: 
    ssl:
      key-password: <Пароль от ключа клиента>
      key-store-password: <Пароль к keystore>
      trust-store-password: <Пароль к truststore>
    producer:
      properties:
        interceptor.signature.certifficate.ssl.keystore.password = <Пароль к keystore>
        interceptor.signature.certificate.ssl.key.password = <Пароль к key>
 

Миграция на текущую версию#

Для миграции на текущую версию требуется:

1. Уточнить в разделе Примечания к релизу необходимость внесения изменений в конфигурацию шлюза при переходе на текущую версию.

2. При необходимости подготовить Config map с новой конфигурацией.

3. Подготовить артефакт Deployment, в котором заменить ссылку на Docker-образ шлюза в репозитории ссылкой на Docker-образ с текущей версией.

4. Если это указано в разделе Примечания к релизу, изменить значения выделяемых ресурсов.

5. Остановить шлюз.

6. Загрузить новую конфигурацию шлюза.

7. Загрузить новый Deployment.

8. Запустить шлюз (если в Deployment указано количество реплик > 0, шлюз запустится автоматически).

Дополнительно:

Порядок действий при обновлении - см. в разделе Руководство по установке.

Остановка/запуск шлюза - см. в разделе Руководство по системному администрированию

Разработка первого приложения с использованием программного продукта#

1. Создать ConfigMap application.yml, предварительно указав:

   • подключения к брокерам Kafka:

     • spring.kafka.producer.bootstrap-servers: <hostname kafka broker>:<port>;

     • spring.kafka.producer.topic: <topic name>;

     • spring.kafka.consumer.bootstrap-servers: <hostname kafka broker>:<port>;

     • spring.kafka.consumer.topics: <topic names>;

   • порт приложения:

     • server.port: <port>;

   • сервер и клиент gRPC:

     • grpc.server.server-port: <port>;

     • grpc.client.settings.default.port: <port>;

     • grpc.client.settings.default.hostname: <hostname>;

   Пример минимальной конфигурации ConfigMap с application.yml

   apiVersion: v1
   data:
     application.yml: |-
       grpc:
         server:
           server-port: 5454
         client:
           settings:
             default:
               hostname: localhost
               port: 5556
       server:
         port: 8787
       logging:
         level:
           org:
             springframework: "INFO"
           root: "INFO"
       spring:
         main:
           banner-mode: "off"
         profiles: production
         kafka:
           producer:
             bootstrap-servers: <HOST>:<PORT>
             topic: SrvDistributeMergePrivateClient.Request
             key-serializer: org.apache.kafka.common.serialization.StringSerializer
             value-serializer: org.apache.kafka.common.serialization.StringSerializer
           consumer:
             bootstrap-servers: <HOST>:<PORT>
             topics: SrvDistributeMergePrivateClient.Response
             group-id: kafkaAdapter
             key-serializer: org.apache.kafka.common.serialization.StringSerializer
             value-serializer: org.apache.kafka.common.serialization.StringSerializer
   kind: ConfigMap
   metadata:
     name: grpc-kafka-adapter-config
   

2. Создать Service для адаптера, предварительно указав:

   • spec.ports[].name: grpc;

   • spec.ports[].port: 5454;

   • spec.ports[].targetPort: 5454.

   Service

   apiVersion: v1
   kind: Service
   metadata: {name: grpc-kafka-adapter}
   spec:
     selector: {name: grpc-kafka-adapter}
     ports:
     - {name: grpc, port: 5454, targetPort: 5454}
     type: ClusterIP
   

   Создать ConfigMap по настройке fluentbit агента, предварительно указав:

   • точки подключения к брокерам Kafka:

     • Brokers <hostname>:<port>;

     • Topics <topic>.

   • путь до файлов, которые необходимо читать:

     • Path   <path to log files>.

   fluent-bit.conf

   apiVersion: v1
   data:
     fluent-bit.conf: |-
       [SERVICE]
        Flush        1
        Daemon       Off
        Log_Level    Debug
        Parsers_File /fluent-bit/etc/parsers.conf
   
       [INPUT]
        Name   tail
        Path   <path to log files>
        Parser        custom
        Tag           custom-tag
        Buffer_Chunk_Size 400k
        Buffer_Max_Size 6MB
        Mem_Buf_Limit 6MB
   
       [OUTPUT]
        Name  kafka
        Brokers <hostname>:<port>
        Timestamp_Key fluent_time
        Topics <topic>
        rdkafka.security.protocol ssl
        rdkafka.ssl.ca.location /fluent-bit/cert/ca_cert.pem
        rdkafka.log.connection.close false
        rdkafka.request.required.acks 1
     parsers.conf: |-
       [PARSER]
        Name        custom
        Format      json
   kind: ConfigMap
   metadata:
     annotations:
     name: fluent-bit
   

   Создать Secret сертификата fluentbit, предварительно указав в блоке ca_cert.pem значение сертификата. fluent-bit-cert

   apiVersion: v1
   data:
     ca_cert.pem: >-
       LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tDQpNSUlGRFRDQ0F2V2dBd0lCQWdJUVlyUStoR2JRVHFsQktjTHVMQWo1dURBTkJna3Foa2lHOXcwQkFRVUZBREFaDQpNUmN3RlFZRFZRUURFdzVVWlhOMElGSnZiM1FnUTBFZ01qQWVGdzB4TmpBM01UVXhNVFV6TURWYUZ3MHpNakEzDQpNVFV4TWpBeU5UaGFNQmt4RnpBVkJnTlZCQU1URGxSbGMzUWdVbTl2ZENCRFFTQXlNSUlDSWpBTkJna3Foa2lHDQo5dzBCQVFFRkFBT0NBZzhBTUlJQ0NnS0NBZ0VBdS9mL1ppR04vK1RaSmhXVUsyN0N6VllGbjR6cDV4TnNVRG84DQphMktXcFVYbnNnQitDYXkweDNFK1RXRTA0R0ZlTjh5YW9wL3VIdGRITzEySmxjaWsvUmpwSnhuaDZNT3k4RnNYDQpZMEZCL3hTWEdqNjNnTkdCQkhGOTc2U2VHRWg5QmVncVVkT20yV0NNWEROcEx0OEdmZkU1ZUJuNGRXbWNwZTIwDQp6NE9SQmZoOXNiWnhya3VtVkwzKzhQaW1FQUZxam5mTmp0VXV3WWZTTThWd05QY2krZytiYXpwM2xicmxIMmRKDQpsdldLY3ZEalZKUENabEplQXJkYk05cHBvR0xzTmM0Nm8xNHZvR1RNcEY0OFJWK0Vqbm9Qdk42WHlURWZhbXV2DQpER2JoMzNoei9WY2JtdStlcFgwVXZvKy9RcG9ub0Y0Y3oyTXlTMlRvZHdRc2Nmbno0L2JjTG04TEtoQ0puMi9tDQp3VG11ZzVVbDF1TnhKdnBFRkwvSmdUa2ZZZTVtU0FJRW53U0JranBDc0krdlIzd3VhQi9zNitNK2JPSTlaV0xPDQp4clNWVkhBWXZMV1BZRkRGMGVabHNoZnFPTEZoTCtaZTJrYVhuL1gxNnFnYjdwQU4wNlhGdUR0TnpTb1lrZG4wDQoyOHdFcEJKNUVkdmRTczdOUzVPOGFaVkRId3BwamQvVVRqQlFCZGE1OENkNmlxM0VNN1oySW5KVjg3VHpXVXRJDQo5cm5Zdld0MWc0ZkhaZ1I2VGlhVmtsVzhtdDg5QzM4Uk9ELzNrNlVERy96b1F0RnhTNXA0em1HY2orREdrYm9WDQo4UktGRldxUnpYb0k3T3ZzQVhlWVVXSXMzVFVEU0QvWkFqeWhMTTJHak44TWV4dFpvaEtia1U0MW1OQ2VIR3JYDQpRSGdEeUxrQ0F3RUFBYU5STUU4d0N3WURWUjBQQkFRREFnR0dNQThHQTFVZEV3RUIvd1FGTUFNQkFmOHdIUVlEDQpWUjBPQkJZRUZKK3lUYWZyVDhzS1NRUkIweWNvckJ6THd4ZGJNQkFHQ1NzR0FRUUJnamNWQVFRREFnRUFNQTBHDQpDU3FHU0liM0RRRUJCUVVBQTRJQ0FRQkxJQTVENDJLY1Z1ZUNhWGdvRGFZK082bUdxRlFOeE1BYkVpZ1QyTk44DQpueFNiVFpjN0tuU3pCaVg2aGxUZW9BTFJWSVcrZXVKdklpZTZ6OE4ra00xNk0vNmdCZUluaTVFRHFadzZqOEhMDQpzdFkyV2R3S29VdThkcDgyWUF3Mkl4eURzamo3cUZGV3RjNVNtLzQ5VGlld2hFeDkyR1ZWcDZ5eGtlQjdHK1ZQDQpWTFNnSlZTUDc1YzNUdHR4V3NHVWZWSUt1WXVCNzQ3TldYb0Y3bkZEeXQ4TjJ5bTJqdEVmaElHOGwvRHJWMmxmDQpQb3hZZHhGWEVVcldqYlB0WHlyUTdSOEc5bm1zMWdCbU0vT1RocUFEbzRsRk44U2Q4MC9ETjV6RDIzOXBMcFExDQp3UXVPVzhGV3ozejV6bTJrNmRHLzRiTG9xSkh3QjFBU29MUW1sNUVxeENjaXd0S0lxRE1jb3ltN0ptTGhycmFIDQpIcmI3cGY5YnpOd2c3TXJZZ0VEbjFIYkhzRWR4NkJscndrVWdHWDFlQmljaVJEWENpUHJCSm16YzlrSG5EOE9ZDQpoU0F1Q21aaVcwNFhlTDM5ak8raWZtSlg4ZlVNRkJHZE1VRUR5YmFMQlhZdGlZMnZ4YzV5elBRUlVkdURhVVJKDQpSVENnN1owbERXS2ZLWUc2RDU0dGNVc2hmeENBUjBMYWs4dndBK3RoV05DN0RmU3pRLzgrajdiY0p4UW9QeWpWDQpUY3d4MDVCVnkrL3JmTW5pT3FST0M2VnozRlpWQ1lZZnAzdENhYkI5alNReXJzck41RHN6YzN0SmxjODFtakNLDQpqZUZkQlRvOW5FS2ZKcjIxd0krRXJveFhlbE9rTFF0bnRwcEpKSEE4TFh4NG5MUVlWdWFINkFiTHZQaE85R0xrDQozZz09DQotLS0tLUVORCBDRVJUSUZJQ0FURS0tLS0t
   kind: Secret
   metadata:
     name: fluent-bit-cert
   type: Opaque
   
   

3. Создать Deployment,
   В конфигурационном файле должны быть заданы параметры:

   • spec.template.spec.volumes[0].configMap.name: <Имя ConfigMap с application.yml для основного контейнера Шлюза>;

   • spec.template.spec.volumes[1].configMap.name: <Имя ConfigMap с fluent-bit.conf для контейнера журналирования>;

   • spec.template.spec.containers[0].image: <ссылка на Docker-образ Шлюза в целевом репозитории>;

   • spec.template.spec.containers[1].image: <ссылка на Docker-образ агента журналирования в целевом репозитории>.

   Deployment

   apiVersion: apps/v1
   kind: Deployment
   metadata: {name: grpc-kafka-adapter}
   spec:
     replicas: 1
     progressDeadlineSeconds: 120
     strategy:
       type: RollingUpdate
       rollingUpdate:
         maxUnavailable: 0  
     selector:
       matchLabels: {name: grpc-kafka-adapter}
     template:
       metadata:
         annotations: {sidecar.istio.io/inject: 'true'}
         labels: {app: grpc-kafka-adapter, name: grpc-kafka-adapter, version: v1}
       spec:
         terminationGracePeriodSeconds: 60
         containers:
           - env:
               - name: PROJECT_NAME
                 valueFrom:
                   fieldRef: {apiVersion: v1, fieldPath: metadata.namespace}
             securityContext:
               runAsNonRoot: true
               readOnlyRootFilesystem: true
             name: grpc-kafka-adapter
             image: <image_version>
             imagePullPolicy: Always
             ports:
               - containerPort: 5454
                 protocol: TCP
             resources:
               limits: {cpu: 400m, memory: 512Mi}
               requests: {cpu: 400m, memory: 512Mi}
             volumeMounts:
               - {mountPath: /opt/synapse/logs, name: synapselogs}
               - {mountPath: /fluent-bit/etc/, name: fluent-bit, readOnly: true}
               - {mountPath: /ssl, name: grpc-kafka-adapter-keystore, readOnly: true}
               - {mountPath: /deployments/config/, name: grpc-kafka-adapter-config, readOnly: true}
               - {mountPath: /deployments/config/ssl/, name: grpc-kafka-adapter-ssl-secret, readOnly: true}
             livenessProbe:
               httpGet: {path: /actuator/health/liveness, port: 8787}
               initialDelaySeconds: 60
               timeoutSeconds: 5
               failureThreshold: 10
             readinessProbe:
               httpGet: {path: /actuator/health/readiness, port: 8787}
               initialDelaySeconds: 60
               timeoutSeconds: 5
               failureThreshold: 3
           - image: <logging_agent_image_version>
             imagePullPolicy: Always
             securityContext:
               runAsNonRoot: true
               readOnlyRootFilesystem: true
             name: fluent-bit
             resources:
               limits: {cpu: 100m, memory: 300Mi}
               requests: {cpu: 100m, memory: 300Mi}
             volumeMounts:
               - {mountPath: /opt/synapse/logs, name: synapselogs}
               - {mountPath: /fluent-bit/etc/, name: fluent-bit, readOnly: true}
               - {mountPath: /fluent-bit/cert/, name: fluent-bit-cert, readOnly: true}
         volumes:
           - configMap:
               defaultMode: 256
               name: grpc-kafka-adapter-config
             name: grpc-kafka-adapter-config
           - configMap:
               defaultMode: 256
               name: fluent-bit
             name: fluent-bit
           - name: grpc-kafka-adapter-keystore
             secret:
               defaultMode: 256
               optional: true
               secretName: grpc-kafka-adapter-keystore
           - name: grpc-kafka-adapter-ssl-secret
             secret:
               defaultMode: 256
               optional: true
               secretName: grpc-kafka-adapter-ssl-secret
           - emptyDir: {}
             name: synapselogs
           - name: fluent-bit-cert
             secret:
               defaultMode: 256
               optional: true
               secretName: fluent-bit-cert    
   
   
   

4. Создать ServiceEntry для разрешения доступа к внешнему ресурсу - брокерам Kafka (egress не требуется), предварительно указав: spec.hosts[]: not.used; spec.ports[].protocol: TCP; spec.ports[].name: tcp; spec.ports[].number: <port>; spec.addresses: <IP>/<MASK>.

   ServiceEntry

   apiVersion: networking.istio.io/v1alpha3
   kind: ServiceEntry
   metadata:
     name: kafka-broker-dev
   spec:
     addresses:
       - <IP>/<MASK>         #диапазон ip адресов брокеров Kafka
     exportTo:
       - .
     hosts:
       - not.used
     location: MESH_EXTERNAL
     ports:
       - name: tcp
         number: <port>	  #порт брокеров Kafka
         protocol: TCP
   
   

Настройка SSL:

1. Создать JKS keystore, который будет содержать клиентский ключ и клиентский сертификат, а также Root-сертификаты от ЦА:

   # создайте из ключа и сертификата файл в формате p12, а затем конвертируйте в JKS
   openssl pkcs12 -export -in client.pem -inkey client.key -out keystore.p12 -name 00CA0001CSYNAPSEsnp99usr
   keytool -importkeystore -deststorepass <ПАРОЛЬ> -destkeypass <ПАРОЛЬ> -destkeystore keystore.jks -srckeystore keystore.p12 -srcstoretype PKCS12 -deststoretype JKS -srcstorepass <ПАРОЛЬ> -alias 00CA0001CSYNAPSEsnp99usr
   
   # добавьте ЦА сертификаты в хранилище
   keytool -keystore keystore.jks -alias "test root ca 2" -import -file ca-root.crt
   keytool -keystore keystore.jks -alias "sberbank test issuing ca 2" -import -file ca-chain.crt
   
   

2. Создать секрет, содержащий JKS keystore, предварительно указав в блоке keystore.jks значение keystore (значение после "keystore.jks: >-" хранится в base64 кодировке):
   keystore

   apiVersion: v1
   data:
     keystore.jks: >-
       /u3+7QAAAAIAAAADAAAAAQAYMDBjYTAwMDFjc3luYXBzZXNucDk5dXNyAAABasTUhmwAAAUBMIIE/TAOBgorBgEEASoCEQEBBQAEggTp5Kl/i64SYwC9uqp9VlT7ma01hwRR4+uaGfPeCR8aUVrSeQCbSHeb4Lxbt4MftZyRS8GKckGodCCCKPe2NCyu++EeqqKhXocrmMPlyxidyCupaGyN2PwtHk2NJ5mfS0ouL5i4k4ua7A2hhFspnQBRuUkczBONz168Jk7+236IkZCGPCv4okptOqxSWMIraWBJ0YK0D2cBGcNf+VYm1B5y6gQ16Bo/7a1RYFVI4B9H7MXaLWvpgUhVE/X6hZVPFumQ7VmygfTui1CPkkwCmt4lQL/oFiQTHxooN3DO7u+v6hSoO7WKF4pvnv25ZknmJnl+VhFGp3q75m3Fns9FgCqgPvKB4wN2d7+RwZ5eGHQOCm7pOoRhoK9iXmg8ZyMAs/d60gGFvmVoi/aCnm0bDDjDl2GOJ/RIPSTNLmj6DE/ZY6cg8Kqc0sNxsf7Yn8pXL6lp99pqY91jzu3LuCGscGSK7L8THSD9uOJZcMQ0eo8zNFKUuJL9oPETZZuBI6wI8toJFwv0MEcG5MJB60C0UAmFxPv5uL1olndJUF8Bsw2O7xnr/UcA7GotkvWRiLVqjBjGRv/+tl19RSTcwGFnkgEuDDxwaSBJHtDB4WBh5Ml2vBPlzLhCwW+dEZXl6zFgc8XgFnjiOqatre4PXvW3da/BTZzAWEeQPgzXRO1KDciUazvxx0uXio4selqNBESgbz/iAM6NplhSM1XU2YuZAnPEYR/fcefhMRGJ9XXi6wekE6bafZS0rCEi3z/mnPA5t1z0Lnyb24HPpMv4ZMh4TrK8eVyeV5SASJSUnYX2zoMAM01BG6IrIn5I2yQvbb0a5ZchS46OtV8v+Pg6kTJn0OubKRehA+QADaG5Q5uQ5QWt2nU+45hZ+3L48iIk1Uzdf26O8a26+FJJ2J3dic7xwfxEGeRxt1xuLHgKbKGdCucILUuLHaiIH660+ctN4vs3HBa0IfkPQl4/yDNJ1w/5liJ5ciHihDQp3FYVN8991HHeLqU3Qp5mR/1l3yY1qlW+g4Rl5/P6a5eLWS1+XTFPX2UFVobiS8lTnvj6dtqdpzOHNkCnkp4GQ2XDMSIYbyxizeBoqz32P8f9TP0bgpxFWEXz8oUN10T7ecm4Vfd7Pd7nvC6gcGa6MQyTKinhw6s9OVVOD5b7QGKKvGvy8dMONNoE0KXfk+MouTu4bnPFDhrI1InENfV9psJ4OxxxaNHyNUJ/jh4TnqEe47f982uInUQyeCKCIChvJWxPDZAkQVM5Dbf+Kv0+YvewcFkv3zbvF5W0+6VCnAsOxxbYqJyK48r4J7z6qdho7LpEidXjM6KlM91CtFmxd/sKlJQ0sGjRUY1gjLHsS1vU+giWGVCmslDEPDVw3emR9PSU7c836IBbjDP6DY+mDIzhmtlb1BjeJ7y/idNrM1JT++OLWIPKb5/mwuDUG60+u0e3PIsoC/a+ZMiDrA4+Rw2/9IDROHNoJRNCjc+ItSRtOj40T6lLIS7MjdNPJghJfehJ66IHsxSCgMfi9htZy/dcTjhD4b3sFp+kII3OZKU3BdXXVznn9oViCiXkqjW2OcNeczdLYQT4hhlhAFqsfHktEfmpw0bDHn0ufwtHmkdmV52gtvhMJpzje3vBa1b2IWr9AlcKZMes7h+XIYwNg4JORhZnU9t6dep54ubzuoXzV8lIAAAAAQAFWC41MDkAAAbgMIIG3DCCBMSgAwIBAgIKaZpdGAABAAA06TANBgkqhkiG9w0BAQsFADBjMRIwEAYKCZImiZPyLGQBGRYCcnUxFDASBgoJkiaJk/IsZAEZFgRzYnJmMRIwEAYKCZImiZPyLGQBGRYCY2ExIzAhBgNVBAMTGlNiZXJiYW5rIFRlc3QgSXNzdWluZyBDQSAyMB4XDTE5MDUwODA2MTcxMloXDTIzMDUwODA2MjcxMlowcDELMAkGA1UEBhMCUlUxLzAtBgNVBAoTJlNhdmluZ3MgQmFuayBvZiB0aGUgUnVzc2lhbiBGZWRlcmF0aW9uMQ0wCwYDVQQLEwQwMENBMSEwHwYDVQQDExgwMENBMDAwMUNTWU5BUFNFc25wOTl1c3IwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDIUmw4gt8tB8q3Q4O2snvZGBZF/wQ/ZIH22yTbZoARkwRYjuc+j5jmB5yDfjBBmzdbCWzKIUBDuq6lNyR3NWUarxnz2zBGRhX9HwOwuECVCtvb4e7FlIsh/msLOZhldyF+iLP1r0lU1Ox3Y4IC2KVYEynlsLrG+pdZMOJLfVLczFlEZXGhqWpysOlsrbdUCEmctvAtzz3ZPyod7+2UTen9mX3LGt6obaCohlKcW3kQ18dIlj0fIUyJMCV9ECTCwFYCwt3lOddmSBtI92zmLMM312v5p9iyzHnK3J/UPWjtQR3TTGWnUs1fRhP+6DvYpjzNQQdewV3qOjpsM3loyJnzAgMBAAGjggKDMIICfzAdBgNVHQ4EFgQUJgrGPrAo8fioVecAVyKrGTrgjO0wHwYDVR0jBBgwFoAUUV/qviYeTSBYvItLGq/2BYyhdTIwggFsBgNVHR8EggFjMIIBXzCCAVugggFXoIIBU4ZFaHR0cDovL3BraS5zYmVyYmFuay5ydS9wa2kvY2RwL1NiZXJiYW5rJTIwVGVzdCUyMElzc3VpbmclMjBDQSUyMDIuY3JshoHAbGRhcDovLy9DTj1TYmVyYmFuayUyMFRlc3QlMjBJc3N1aW5nJTIwQ0ElMjAyLENOPVRWLUNFUlQtVUIsQ049Q0RQLENOPVB1YmxpYyUyMEtleSUyMFNlcnZpY2VzLENOPVNlcnZpY2VzLERDPVVuYXZhaWxhYmxlQ29uZmlnRE4/Y2VydGlmaWNhdGVSZXZvY2F0aW9uTGlzdD9iYXNlP29iamVjdENsYXNzPWNSTERpc3RyaWJ1dGlvblBvaW50hkdodHRwOi8vaW50cGtpLmNhLnNicmYucnUvcGtpL2NkcC9TYmVyYmFuayUyMFRlc3QlMjBJc3N1aW5nJTIwQ0ElMjAyLmNybDCBvgYIKwYBBQUHAQEEgbEwga4wVAYIKwYBBQUHMAKGSGh0dHA6Ly9wa2kuc2JlcmJhbmsucnUvcGtpL2FpYS9TYmVyYmFuayUyMFRlc3QlMjBJc3N1aW5nJTIwQ0ElMjAyKDEpLmNydDBWBggrBgEFBQcwAoZKaHR0cDovL2ludHBraS5jYS5zYnJmLnJ1L3BraS9haWEvU2JlcmJhbmslMjBUZXN0JTIwSXNzdWluZyUyMENBJTIwMigxKS5jcnQwDAYDVR0TAQH/BAIwADANBgkqhkiG9w0BAQsFAAOCAgEAQFHEWrITBZLrwmaKelhkZHua6k8JuNrgddr2jBko9zq7S2z3sAz8n7Nt4+RA71crxk/V1wBF+NYqUzx48umKL3lWmWvIpVVpC4DQL7z+DyIG+kCdw/NlaPN+dnk2M9+2TOBBJhca2KfH7HIc4bQNlmpEs6ID+oriwnHJ1YoEWKSN9uuc0KAlnT8VPB0tLyIHd8H4QwG1xRy03YEFvp5r3jXvO8R9Y3itQDtRFLk4bOJehYyLCe5FgXJB1KVKj7ODwhT+zKYRhWyTrfbEfAxn1bofnGsAsTg29b8xv0Ow4i7PKtDd7giR1h95uorxKWfZC0blsYesS8Br5FahjLRW2Onxwedi8opqyVtzvh3dzpJiYwQ8Kj3fz6jRb+Bbt/DI2Cd0KBGN+az97wk+aQktXFCjsCNvmptC4f0JjuRA+P55g6asXfALVuoP42jFPoC+KunJ26kLVyl4rpP/V1kt7gsawzcu7XS+2kBwouukz24f6M+mtnJy+ElrTsbn0RuGo5AF/RzLFZCotOh4OX4ESztshDPtjpR9ptETz+6KlH3+ZgzqAti/5ERrgpAYiDSvJi1gEQD4kfS2z0X89GMcnSI3wJBLS7hYJwbIioVSOfK3Oig/AWzKBd2XWN3Oo7hjl0c8NyxYNdZpOWVTUIYb3CQne61SMItzCW3d1hCFJTcAAAACAA50ZXN0IHJvb3QgY2EgMgAAAWqWYEDhAAVYLjUwOQAABREwggUNMIIC9aADAgECAhBitD6EZtBOqUEpwu4sCPm4MA0GCSqGSIb3DQEBBQUAMBkxFzAVBgNVBAMTDlRlc3QgUm9vdCBDQSAyMB4XDTE2MDcxNTExNTMwNVoXDTMyMDcxNTEyMDI1OFowGTEXMBUGA1UEAxMOVGVzdCBSb290IENBIDIwggIiMA0GCSqGSIb3DQEBAQUAA4ICDwAwggIKAoICAQC79/9mIY3/5NkmFZQrbsLNVgWfjOnnE2xQOjxrYpalReeyAH4JrLTHcT5NYTTgYV43zJqin+4e10c7XYmVyKT9GOknGeHow7LwWxdjQUH/FJcaPreA0YEEcX3vpJ4YSH0F6CpR06bZYIxcM2ku3wZ98Tl4Gfh1aZyl7bTPg5EF+H2xtnGuS6ZUvf7w+KYQAWqOd82O1S7Bh9IzxXA09yL6D5trOneVuuUfZ0mW9Ypy8ONUk8JmUl4Ct1sz2mmgYuw1zjqjXi+gZMykXjxFX4SOeg+83pfJMR9qa68MZuHfeHP9Vxua756lfRS+j79CmiegXhzPYzJLZOh3BCxx+fPj9twubwsqEImfb+bBOa6DlSXW43Em+kQUv8mBOR9h7mZIAgSfBIGSOkKwj69HfC5oH+zr4z5s4j1lYs7GtJVUcBi8tY9gUMXR5mWyF+o4sWEv5l7aRpef9fXqqBvukA3TpcW4O03NKhiR2fTbzASkEnkR291Kzs1Lk7xplUMfCmmN39ROMFAF1rnwJ3qKrcQztnYiclXztPNZS0j2udi9a3WDh8dmBHpOJpWSVbya3z0LfxE4P/eTpQMb/OhC0XFLmnjOYZyP4MaRuhXxEoUVapHNegjs6+wBd5hRYizdNQNIP9kCPKEszYaM3wx7G1miEpuRTjWY0J4catdAeAPIuQIDAQABo1EwTzALBgNVHQ8EBAMCAYYwDwYDVR0TAQH/BAUwAwEB/zAdBgNVHQ4EFgQUn7JNp+tPywpJBEHTJyisHMvDF1swEAYJKwYBBAGCNxUBBAMCAQAwDQYJKoZIhvcNAQEFBQADggIBAEsgDkPjYpxW54JpeCgNpj47qYaoVA3EwBsSKBPY03yfFJtNlzsqdLMGJfqGVN6gAtFUhb564m8iJ7rPw36QzXoz/qAF4ieLkQOpnDqPwcuy1jZZ3AqhS7x2nzZgDDYjHIOyOPuoUVa1zlKb/j1OJ7CETH3YZVWnrLGR4Hsb5U9UtKAlVI/vlzdO23FawZR9Ugq5i4Hvjs1ZegXucUPK3w3bKbaO0R+EgbyX8OtXaV8+jFh3EVcRStaNs+1fKtDtHwb2eazWAGYz85OGoAOjiUU3xJ3zT8M3nMPbf2kulDXBC45bwVbPfPnObaTp0b/hsuiokfAHUBKgtCaXkSrEJyLC0oioMxyjKbsmYuGutocetvul/1vM3CDsytiAQOfUdsewR3HoGWvCRSAZfV4GJyJENcKI+sEmbNz2QecPw5iFIC4KZmJbThd4vf2M76J+Ylfx9QwUEZ0xQQPJtosFdi2Jja/FznLM9BFR24NpRElFMKDtnSUNYp8pgboPni1xSyF/EIBHQtqTy/AD62FY0LsN9LND/z6PttwnFCg/KNVNzDHTkFXL7+t8yeI6pE4LpXPcVlUJhh+ne0JpsH2NJDKuys3kOzNze0mVzzWaMIqN4V0FOj2cQp8mvbXAj4SujFd6U6QtC2e2mkkkcDwtfHictBhW5ofoBsu8+E70YuTeAAAAAgAac2JlcmJhbmsgdGVzdCBpc3N1aW5nIGNhIDIAAAFqlmAbHAAFWC41MDkAAAqcMIIKmDCCCICgAwIBAgITGAAAAAS5RQdwBbAYQAAAAAAABDANBgkqhkiG9w0BAQsFADAZMRcwFQYDVQQDEw5UZXN0IFJvb3QgQ0EgMjAeFw0xNjA3MTUxMjQ3NDNaFw0yNDA3MTUxMjU3NDNaMGMxEjAQBgoJkiaJk/IsZAEZFgJydTEUMBIGCgmSJomT8ixkARkWBHNicmYxEjAQBgoJkiaJk/IsZAEZFgJjYTEjMCEGA1UEAxMaU2JlcmJhbmsgVGVzdCBJc3N1aW5nIENBIDIwggIiMA0GCSqGSIb3DQEBAQUAA4ICDwAwggIKAoICAQDOe9PJjSMr4QGveswkep1/73kCHOmAlPiWoE8BjppW+uDibtc7yWXKipE4QgbqXNz1u3oy6WyLQ3PBA2PPclg8O425xCH8VgbOWz0FmM7vdFoMYJGLFpVq9DE1VEHI3MP3lOt5xhDU3IVaCHnMDddx8mFS2RkXNvDBLa5NBfge2UM5SslEQROJc3fuFQmLD4HRlaJe+sgOqm3buw3a5WVGbaDGQ8L/e1kmRyQa78IrHcbWFwge9aWbHaUY4e/uB1z+6EzWuaODRsjlrpXH/dvg4XLsVJGloxQcqgFhvbml3fcqlPhSWydiUHnimzIOILdPD9HBIfyOuSDpIpk5WeLXNEJe/i0EotclEIyWwWOjUoFh3ZiK6seEVbMNXWftNiKr6W3bNW+kfseSOUddASHxBz5WsZQAnigLuwwjTr+k1x7Vi7F8xTzp9JDuB2i+0eK9CSoFLjq+fHqziUmNEX2xTbIJjeJp511Wuj5iv+0imirUJJJCbRJefweO84lQpnhVvsk4wYY2AuYOm71/G/BfQ3yypwprhuLNcG5yXZ/ZTW0y/6jKmNNLlvnV56CKElOr13z427K6VK7eIwHZgeQpcUkja3zW+cblrFrwge+BduspNqO6dn/2HaP5cfoc3Un+F1nY5Whu8RxrblEecF/5ZLzc1g2JC2pzDiwTg4fQPwIDAQABo4IFjTCCBYkwEAYJKwYBBAGCNxUBBAMCAQEwIwYJKwYBBAGCNxUCBBYEFBKfeeA/4akm+B0LLAPMsvGeTDtLMB0GA1UdDgQWBBRRX+q+Jh5NIFi8i0sar/YFjKF1MjAZBgkrBgEEAYI3FAIEDB4KAFMAdQBiAEMAQTALBgNVHQ8EBAMCAYYwDwYDVR0TAQH/BAUwAwEB/zAfBgNVHSMEGDAWgBSfsk2n60/LCkkEQdMnKKwcy8MXWzCCAloGA1UdHwSCAlEwggJNMIICSaCCAkWgggJBhjdodHRwOi8vcGtpLnNiZXJiYW5rLnJ1L3BraS9jZHAvVGVzdCUyMFJvb3QlMjBDQSUyMDIuY3JshoHGbGRhcDovLy9DTj1UZXN0JTIwUm9vdCUyMENBJTIwMixDTj1UZXN0Um9vdENBMjU2LENOPUNEUCxDTj1QdWJsaWMlMjBLZXklMjBTZXJ2aWNlcyxDTj1TZXJ2aWNlcyxDTj1Db25maWd1cmF0aW9uLERDPXNpZ21hLERDPXNicmYsREM9cnU/Y2VydGlmaWNhdGVSZXZvY2F0aW9uTGlzdD9iYXNlP29iamVjdENsYXNzPWNSTERpc3RyaWJ1dGlvblBvaW50hoG9bGRhcDovLy9DTj1UZXN0JTIwUm9vdCUyMENBJTIwMixDTj1UZXN0Um9vdENBMjU2LENOPUNEUCxDTj1QdWJsaWMlMjBLZXklMjBTZXJ2aWNlcyxDTj1TZXJ2aWNlcyxDTj1Db25maWd1cmF0aW9uLERDPXNicmYsREM9cnU/Y2VydGlmaWNhdGVSZXZvY2F0aW9uTGlzdD9iYXNlP29iamVjdENsYXNzPWNSTERpc3RyaWJ1dGlvblBvaW50hj9odHRwOi8vZXh0cGtpLnNpZ21hLnNicmYucnUvQ2VydEVucm9sbC9UZXN0JTIwUm9vdCUyMENBJTIwMi5jcmyGPGh0dHA6Ly9pbnRwa2kuY2Euc2JyZi5ydS9DZXJ0RW5yb2xsL1Rlc3QlMjBSb290JTIwQ0ElMjAyLmNybDCCAncGCCsGAQUFBwEBBIICaTCCAmUwQwYIKwYBBQUHMAKGN2h0dHA6Ly9wa2kuc2JlcmJhbmsucnUvcGtpL2FpYS9UZXN0JTIwUm9vdCUyMENBJTIwMi5jcnQwgbgGCCsGAQUFBzAChoGrbGRhcDovLy9DTj1UZXN0JTIwUm9vdCUyMENBJTIwMixDTj1BSUEsQ049UHVibGljJTIwS2V5JTIwU2VydmljZXMsQ049U2VydmljZXMsQ049Q29uZmlndXJhdGlvbixEQz1zaWdtYSxEQz1zYnJmLERDPXJ1P2NBQ2VydGlmaWNhdGU/YmFzZT9vYmplY3RDbGFzcz1jZXJ0aWZpY2F0aW9uQXV0aG9yaXR5MIGvBggrBgEFBQcwAoaBomxkYXA6Ly8vQ049VGVzdCUyMFJvb3QlMjBDQSUyMDIsQ049QUlBLENOPVB1YmxpYyUyMEtleSUyMFNlcnZpY2VzLENOPVNlcnZpY2VzLENOPUNvbmZpZ3VyYXRpb24sREM9c2JyZixEQz1ydT9jQUNlcnRpZmljYXRlP2Jhc2U/b2JqZWN0Q2xhc3M9Y2VydGlmaWNhdGlvbkF1dGhvcml0eTBZBggrBgEFBQcwAoZNaHR0cDovL2V4dHBraS5zaWdtYS5zYnJmLnJ1L0NlcnRFbnJvbGwvVGVzdFJvb3RDQTI1Nl9UZXN0JTIwUm9vdCUyMENBJTIwMi5jcnQwVgYIKwYBBQUHMAKGSmh0dHA6Ly9pbnRwa2kuY2Euc2JyZi5ydS9DZXJ0RW5yb2xsL1Rlc3RSb290Q0EyNTZfVGVzdCUyMFJvb3QlMjBDQSUyMDIuY3J0MA0GCSqGSIb3DQEBCwUAA4ICAQADGKevectc4iZbVD5sHbmX1Cz6feeJql/ZPqBJyjVLLoJFELVJnwwshU8WOdakJjeOx7g2k2p+tbTw+hNpmgYh+kqqR6pTEntuigHXC3ncqYG7UlDOFAKA+6EGPpTxEAoy1cFDJo0dNFK5rOzvXD8gbCEC6jKB77bqNZN8R4jGkvOhT47tSC0PoNMwizeyKDVsVdaY7L4Dqg4XspV++ojr8k+13veME3I+REoubZlMwJUnlyBkXDwfQ636TFzbBKqwSgXqlBH0rpKIUfJ2m/Ow1S61VVQYsubbf7BRZH/eO0NwB/LqEKhHoFPrvaGj/4gxHAGJKl8ZiO93LgcPTjkWLxg/Cis+euipt+TZ/0WwXcZhnfjAzjq/vQii70054JZILb62VRZR/cmoISC6wFQ3GxkzIBnKtGPyQgCdkL+KSFYimCqV2jpecpikUvg0s4JokfVTGwFq0mtUpz3YtIjwrRp2Js+bmrCMtpoYLwEJ/5qThnxLQ/3O+0z3CklX9vrnazecwl0n1VCaec27fjFSpEglABIAmImCIc4FHuJp9ej3gwpTaCeaKdE4uGElLWzh8L3uDBi6/74gMvcBK/fPTAVc+fW/3ZXc2jNU/1Y3hN24q8DZnbtLBnnuKH1xJ7PgM2X8dRe6y6MO7VA8ERavJ9eXSr8WFuXXxVJB9bF2BfTxxF7tSmiPKEE2VYwZaWesu03a
   kind: Secret
   metadata:
     name: grpc-kafka-adapter-scm-keystore
   type: Opaque
   
   

3. Создать секрет, в котором будут указаны пароли к jks, ssl.yml должен быть определен в следующем виде:

   spring:
     kafka: 
       ssl:
         key-password: <Пароль от ключа клиента>
         key-store-password: <Пароль к keystore>
         trust-store-password: <Пароль к truststore>
    
   

   Secret ssl.yml

   apiVersion: v1
   data:
     ssl.yml: >-
       c3ByaW5nOg0KICBrYWZrYToNCiAgICBzc2w6DQogICAgICBrZXktcGFzc3dvcmQ6ICJxd2UxMjMiDQogICAgICBrZXktc3RvcmUtcGFzc3dvcmQ6ICJxd2UxMjMiDQogICAgICB0cnVzdC1zdG9yZS1wYXNzd29yZDogInF3ZTEyMyINCg==
   kind: Secret
   metadata:
     name: grpc-kafka-adapter-ssl-secret
   type: Opaque
   
   
   

4. Убедиться, что секреты монтируются в Pod, зайдя в терминал пода и посмотрев файловую систему

5. Указать в конфиге адаптера application.yml настройки ssl, которые должны содержать

   • пути до хранилищ доверенных и личных сертификатов:

     • spring.kafka.ssl.key-store-location: <path>;

     • spring.kafka.ssl.trust-store-location: <path>.

   • протокола подключения:

     • spring.kafka.ssl.protocol: TLSv1.2;

     • spring.kafka.ssl.properties.security.protocol: SSL;

     • spring.kafka.ssl.properties.ssl.enabled.protocols: TLSv1.2.

   настройки ssl

   spring:
     kafka:
       ssl:
         key-store-location: file:/ssl/keystore.jks
         trust-store-location: file:/ssl/keystore.jks
         protocol: TLSv1.2
       properties:
         security.protocol: SSL
         ssl.enabled.protocols: TLSv1.2
   

Использование программного компонента#

Шлюз Kafka предназначен для организации взаимодействия внутренних компонентов Synapse (правообладателем которых является АО «‎СберТех»‎), таких как другие шлюзы или сервисы трансформации, с внешними АС с помощью топиков, расположенных на брокерах Kafka. Для внутренних вызовов между компонентами Synapse используется протокол gRPC. Для сериализации передаваемых сообщений внутри Synapse используется ProtoBuf.

Использование компонента Шлюз Kafka позволяет исключить специфичную логику вызова топиков Kafka из внутренних компонентов Synapse.

Шлюз Kafka реализован в виде контейнеризированного Java приложения, запускаемого в среде оркестрации контейнеров.

Шлюз Kafka может использовать технологические сервисы платформы, обеспечивающие прикладное журналирование, интеграционное журналирование, а также включенного механизма оркестрации сервисов Istio.

Шлюз Kafka поставляется в виде собранного докер образа. Загрузка образа в среду исполнения и настройка его параметров для конкретной прикладной задачи производится в файлах конфигурации. Набор этих файлов и составляет прикладной дистрибутив шлюза.

Один прикладной дистрибутив (deployment) Шлюза Kafka может обеспечивать работу нескольких независимых интеграционных цепочек, при условии, что они не требуют установки взаимоисключающих значений параметров. В противном случае потребуется использовать для каждой интеграционной цепочки отдельный Deployment Шлюза Kafka с отдельным набором параметров.

Шлюз Kafka может обеспечивать взаимодействия типа Запрос-Ответ и Нотификация, как по инициативе внешней системы, так и по инициативе внутренних микросервисов.

Типовая упрощенная схема на рис. 1 показывает место Kafka шлюза в интеграционной цепочке.

Использование пользовательских метрик мониторинга#

Формирование метрик мониторинга осуществялется с помощью Spring Actuator. Метрики в формате prometheus доступны по адресу: <ip>:<spring_boot_server_port>/actuator/prometheus

Помимо стандартных метрик Spring Boot приложения доступны следующие метрики:

Управление листенерами сообщений (включение/отключение) из топиков Kafka в рантайме#

У каждого топика есть параметры spring.kafka.consumer.topics.topic.enabled (по-умолчанию true) и spring.kafka.consumer.common-enabled (по-умолчанию true), которые отвечают за работу листенеров.

spring.kafka.consumer.topics.topic.enabled необходимо указывать в локальной config map шлюза (для каждого шлюза отдельно).

spring.kafka.consumer.common-enabled необходимо указывать в глобальной config map шлюза (одна конфигурация на весь namespace), пример такой конфигурации:

kind: ConfigMap
apiVersion: v1
metadata:
  name: common-kafka-config
data:
  common-kafka-config.yml: |-
    spring:
      kafka:
        consumer:
          common-enabled: false

При старте шлюза проверяется на соответствие, что оба параметра для листенера в состоянии true, иначе листенер не будет запущен.

Таблица соответствий включения листенера:

Чтобы отключить или включить листенеры в runtime достаточно задать любому из параметров значение false/true в config map адаптера.

При переведении любого из них в false - листенеры выключаются

При переведении любого из них в true - листенеры поднимаются

Управление механизмами повторной отправки сообщений в случае ошибок#

Транзакционный режим чтения сообщения из топика кафка и отправки его по GRPC/REST#

В настройках адаптера необходимо указать spring.kafka.consumer.transactional=true.

Настройка транзакционного режима:

spring:
  kafka:
    consumer:
      transactional: <true/false - если true, то в случае ошибки доставки сообщения фиксировать ли offset топика кафки (в этом случае адаптер будет бесконечно пробовать читать недоставленные сообщения и отправлять по grpc/rest)>
      repeat-after-error-ms: <в случае transactional=true, период повторной попытки вычитки и отправки сообщения при ошибке>

Тогда в случае ошибки при попытке отправить сообщение по GRPC/REST offset в топике не будет закоммичен и спустя время, указанное в параметре repeat-after-error-ms будет произведена повторная попытка вычитать данное сообщение.

Повтор вызовов на уровне producer кафки при попытке отправить сообщения в топик#

В настройках адаптера необходимо указать acks=1/all и выставить количество ретраев.

Настройки повторов на уровне producer:

spring:
  kafka:
    producer:
      acks: 1 <поддерживаемые значения: 0 - не ждать ответа от брокеров (retries невозможны), 1 - ждать подтверждение записи от одного брокера лидера, all - ждать подтверждение записи от всех брокеров>
      retries: 10 <количество повторных попыток>

Подключение адаптера к резервному кластеру Kafka#

При подключении адаптера к резервному кластеру Kafka:

1. Адаптер читает сразу из двух кластеров;
2. Адаптер записывает сообщения в основной кластер. В случае ошибки записи в основной кластер, пытается записать в резервный кластер и исключает из балансировки основной кластер на время, указанное в параметре `spring.kafka.producer.main-cluster-exclude-time-ms` (по умолчанию `60000мс`);    
3. Соответсвенно необходимо задать следующие параметры: `spring.kafka.producer.main-cluster-exclude-time-ms`, `spring.kafka.producer.reserve-servers`, `spring.kafka.consumer.reserve-servers`.

Пример конфигурации:

kafka:
  producer:
    main-cluster-exclude-time-ms: 60000
    bootstrap-servers: host.ru:9092
    reserve-servers: host2.ru:9092
    topic: NT.SYN.ESB.CLIENTDEDUPREQUESTED
  consumer:
    bootstrap-servers: host:9092
    reserve-servers: host:9092
    topics:
    - topic: NT.COD.CLNT.CLIENTDEDUPLICATED
      group-id: kafkaAdapter

Включение синхронного режима работы#

Для того, чтобы включить синхронный режим работы адаптера, при котором происходит не только доставка сообщения в топик, но и ожидается ответ из топика, необходимо:

Добавить в блок spring.kafka.sync-mode конфигурации адаптера следующие настройки:

spring:
  kafka:
    sync-mode:
      enabled: true              # если указать true, можно не отправлять заголовок x-synapse-sync:true
      reply-topic: <reply-topic> # топик ответа
      timeout: 10 

После, при вызове API адаптера (протоколы: REST/GRPC) сообщение будет отправляться в топик producer и адаптер будет ожидать ответ из топика по умолчанию, указанного в reply-topic. Время ожидания ответа регулируется параметром timeout в секундах. Если необходимо заменить топик ответа, то при вызове API нужно передать заголовок x-synapse-reply-topic с другим названием топика. Отключить/включить синхронный режим можно также с помощью заголовка x-synapse-sync, передав в нем значения false/true соответственно.

При ответе на стороне принимающей системы должен быть скопирован заголовок запроса correlation-id для успешной корреляции запроса и ответа на стороне адаптера.

Маршрутизация сообщений с помощью адаптера по сервисам в зависимости от топика#

Для протокола GRPC необходимо добавить в блок конфигурации адаптера следующие настройки:

# сначала определяем различные маршруты для GRPC клиента
grpc:
  client:
    settings:
      default:
        hostname: <service-name>
        port: 6565
      service1:
        hostname: <service-name2>
        port: 8888
      service2:
        hostname: <service-name3>
        port: 4444 
# затем задаем правила, по которым необходимо использовать данные маршруты
topics-routing:
  routes:
    <topic_name>: service1
    <topic_name2>: service2

Приоритет выбора вызываемого сервиса, по убыванию:

1. Правило из `topics-routing`;
2. Значение заголовка `x-synapse-authority`;
3. Значение параметра grpc.client.settings.default.

Для протокола REST необходимо добавить в блок конфигурации адаптера следующие настройки:

# задаем правила, по которым необходимо вызывать REST сервисы
topics-routing:
  routes:
    <topic_name>: <rest_endpoint>       # например "http://service-name:8080/endpoint" или "http://ingress-synapse-esbfs-nt.host.ru:80/endpoint"
    <topic_name2>: <rest_endpoint2>

Приоритет выбора вызываемого сервиса, по убыванию:

1. Правило из topics-routing;
2. Значение заголовоков x-synapse-authority + x-synapse-rest-port + x-synapse-rest-path; 
3. Значение параметра spring.kafka.rest.default-url.

Маршрутизация и заполнение заголовков на основе парсинга сообщения#

Доступен парсинг json сообщений с помощью jsonpath, а также парсинг xml сообщений с помощью xpath (определение инструмента парсинга производится автоматически).

Недосупно для batch режима.

Доступны следующие парсеры:

1. fromBody - заполнение переменной на основе парсинга тела сообщения;
2. fromHeader - заполенение переменной на основе заголовка, при этом регистр не учитывается (kafka заголовка или http/grpc заголовока);
3. fromConst - заполнение переменной константой.

Все значения переменных из блока variables будут добавлены в заголовки и переданы по rest/grpc или записаны в топик kafka.

По умолчанию, всем заголовкам будет добавлен префикс x-synapse-.

Например: если в названии переменной указано operationname, тогда значение данной переменной будет добавленно в заголовок x-synapse-operationname; если в названии переменной указано rquid, тогда значение данной переменной будет добавленно в заголовок x-synapse-rquid.

Если не требуется добавлять данный префикс (или требуется передать системный заголовок), то в конфигурации требуется указать addPrefix: false.

Список системных заголовково доступен по ссылке: системные заголовки Kafka Spring API.

Пример для передачи системного заголовка kafka_messageKey, который определяет значение ключа сообщения в топике Kafka:

routing:
  api:
    routeList:
      - routeName: default
        destinationExpression: ('topicName')
        variables:
          - name: kafka_messageKey
            addPrefix: false
            valueFrom:
              - type: fromHeader
                value: x-synapse-rquid
              - type: fromBody
                value: $.rquid

Пример маршрутизации сообщений в зависимости от заголовка (топика)#

Адаптер читает два топика consumer_test и consumer_test2.

При чтении сообщения определяется routeName по значению заголовка kafka_receivedtopic, в данном заголовке передается топик, из которого было вычитано сообщение.

Затем в случае, если данное значение соответствует consumer_test, то сообщение отправляется по адресу http://host1:8080/.

Если значение соответствует consumer_test2, то сообщение отправляется по адресу http://host2:8080/.

Если значение не соответвует ни одному из определнных из routeList, оно отправляется по адресу http://default-host:8080.

Конфигурация в случае REST взаимодействия:

spring:
  kafka:
    rest:
      mode: true
    consumer:
      topics:
        - topic: consumer_test
          group-id: kafkaAdapter_group
          enabled: true
        - topic: consumer_test2
          group-id: kafkaAdapter_group
          enabled: true
routing:
  kafka:
    routeName:
      valueFrom:
        - type: fromHeader
          value: kafka_receivedtopic
    routeList:
      - routeName: consumer_test
        destinationExpression: ('http://host1:8080/')
      - routeName: consumer_test2
        destinationExpression: ('http://host2:8080/')
      - routeName: default
        destinationExpression: ('http://default-host:8080/')

Конфигурация в случае GRPC взаимодействия:

grpc:
  client:
    settings:
      default:
        hostname: localhost
        port: 8080
      service1:
        hostname: host1
        port: 8080
      service2:
        hostname: host2
        port: 8008
spring:
  kafka:
    rest:
      mode: false
    consumer:
      topics:
        - topic: consumer_test
          group-id: kafkaAdapter_group
          enabled: true
        - topic: consumer_test2
          group-id: kafkaAdapter_group
          enabled: true
routing:
  kafka:
    routeName:
      valueFrom:
        - type: fromHeader
          value: kafka_receivedtopic
    routeList:
      - routeName: consumer_test
        destinationExpression: ('service1')
      - routeName: consumer_test2
        destinationExpression: ('service2')
      - routeName: default
        destinationExpression: ('default')

Пример маршрутизации сообщений в зависимости от тега внутри JSON сообщения#

При чтении сообщения определяется routeName по значению тега evtType JSON сообщения, на основе него выполняется маршрутизация.

Пример сообщения для маршрутизации по routeName: type1 и отправки по адресу http://service-create:

{
    "evtType": "type1",
    "operName": "createRq"
} 

Пример сообщения для маршрутизации по routeName: type2 и отправки по адресу http://update-service:

{
    "evtType": "type2",
    "operName": "update"
} 

Пример конфигурации:

spring:
  kafka:
    rest:
      mode: true
    consumer:
      topics:
        - topic: consumer_test
          group-id: kafkaAdapter_group
          enabled: true
        - topic: consumer_test2
          group-id: kafkaAdapter_group
          enabled: true
routing:
  kafka:
    routeName:
      valueFrom:
        - type: fromBody
          value: $.evtType
    routeList:
      - routeName: type1
        destinationExpression: ('http://' + 'service-' + ServiceName.substring(0,ServiceName.length()-2))
        variables:
          - name: ServiceName
            valueFrom:
              - type: fromBody
                value: $.operName
      - routeName: type2
        destinationExpression: ('http://' + variables['ServiceName'] + '-service')
        variables:
          - name: ServiceName
            valueFrom:
              - type: fromBody
                value: $.operName
      - routeName: default
        destinationExpression: ('http://default-host:8080/')

Пример маршрутизации сообщений в зависимости от тега внутри XML сообщения#

При чтении сообщения определяется routeName по значению тега evtType XML сообщения, на основе него выполняется маршрутизация.

Пример сообщения для маршрутизации по routeName: type1 и отправки по адресу http://service-create:

<xml>
  <evtType>type1</evtType>
  <operName>createRq</operName>
</xml>

Пример сообщения для маршрутизации по routeName: type2 и отправки по адресу http://update-service:

<xml>
  <evtType>type2</evtType>
  <operName>update</operName>
</xml>

Пример конфигурации:

spring:
  kafka:
    rest:
      mode: true
    consumer:
      topics:
        - topic: consumer_test
          group-id: kafkaAdapter_group
          enabled: true
        - topic: consumer_test2
          group-id: kafkaAdapter_group
          enabled: true
routing:
  kafka:
    routeName:
      valueFrom:
        - type: fromBody
          value: //evtType
    routeList:
      - routeName: type1
        destinationExpression: ('http://' + 'service-' + ServiceName.substring(0,ServiceName.length()-2))
        variables:
          - name: ServiceName
            valueFrom:
              - type: fromBody
                value: //operName
      - routeName: type2
        destinationExpression: ('http://' + variables['ServiceName'] + '-service')
        variables:
          - name: ServiceName
            valueFrom:
              - type: fromBody
                value: //operName
      - routeName: default
        destinationExpression: ('http://default-host:8080/')

Исключение заголовков при чтении из топика Kafka#

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

Для выборочного исключения необходимо добавить в конфигурации адаптера в параметр tracing.excludeHeaders.kafka список заголовков.

Пример конфигации с частичным исключении заголовков:

tracing:
  excludeHeaders:
  	kafka:
	  - HEADER-TO-EXCLUDE1
	  - ...
	  - header-to-excludeN

Название заголовка вводится полностью. Обратите внимание на то, что заголовки регистрозависимы.

Для пропуска всех заголовков, кроме rquid и operationname, необходимо добавить в конфигурации адаптера в параметр tracing.excludeHeaders.kafka ключевое слово all.

При пропуске всех заголовков не будет доступен механизм фильтрации сообщений.

Пример полного исключения заголовков:

tracing:
  excludeHeaders:
  	kafka:
	  - all

Исключение заголовков при вызове адаптера по REST/gRPC#

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

Для выборочного исключения необходимо добавить в конфигурации адаптера в параметр tracing.excludeHeaders.api список заголовков.

Пример конфигации с частичным исключении заголовков:

tracing:
  excludeHeaders:
  	api:
	  - HEADER-TO-EXCLUDE1
	  - ...
	  - header-to-excludeN

Название заголовка вводится полностью. Обратите внимание на то, что заголовки регистрозависимы только для протокола gRPC. В случае REST-запроса контейнер сервлетов переводит заголовки в нижний регистр, поэтому сравнение выполняется без учёта регистра.

Для пропуска всех заголовков необходимо добавить в конфигурации адаптера в параметр tracing.excludeHeaders.api ключевое слово all.

Пример полного исключения заголовков:

tracing:
  excludeHeaders:
  	api:
	  - all

Отправка пачки сообщений (batch) в топик Kafka#

Отправка пачки сообщений (batch) в топик Kafka доступна только по REST API, контроллер /_bulk.

Первоначально строка сообщений, полученная по данному контроллеру, разделяется по параметру spring.kafka.producer.separator (по умолчанию \n).

Затем сообщения фильтруются. Если сообщение содержит хотя бы одно вхождение из списка spring.kafka.producer.skipList, то оно пропускается.

Если сообщение содержит тег agent, то оно идентфицируется, как сообщение от filebeat и в нем передается только блок message. Если сообщение не содержит тег agent, то оно идентфицируется, как сообщение от fluentbit и в нем исключется тег @timestamp

Если список skipList пуст, то сообщения не фильтруются и не модифицируются по типу filebeat/fluentbit.

Пример конфигурации:

server:
  tomcat:
    max-swallow-size: 40MB        
    threads:
      max: 200                  
      min-spare: 200            
spring:
  kafka:
    rest:
      mode: true
    producer:
      batch:
        separator: \n           
        skipList:
          - "\"_type\""
          - "\"_index\""
    properties:
      acks: 1
      linger.ms: 500
      batch.size: 524288
      buffer.memory: 104857600
      compression.type: gzip
      max.request.size: 52428800

Часто встречающиеся проблемы и пути их решения#