Руководство администратора

Настояшее Руководство содержит описание и сценарии эксплуатации и администрирования TQE(MQ).

• Введение
• Сценарии использования

TQE(MQ) выступает в качестве брокера, ответственного за передачу сообщений потребителям. Потребители получают только те сообщения, на которые подписаны. Такой подход снижает ненужный трафик данных, повышает эффективность обработки и обеспечивает максимально высокую пропускную способность.

Каждое сообщение состоит из нескольких полей:

• id - идентификатор сообщения. Монотонно возрастающий уникальный ключ, который генерируется мастером набора реплик. При асинхронной репликации монотонность и уникальность поддерживаются только на уровне одной реплики.
• sharding_key - ключ шардирования; Опциональное поле, если оно указано, то на его основе выбирается соответствующий целевой шард для сообщения. Если оно не указано, то система самостоятельно проставляет ключ с целью равномерной загрузки шарда. При статическом шардировании sharding_key обязателен.
• routing_key - ключ фильтрации. Опциональное поле, если оно указано, то очередь будет отвечать на запрос сообщений только сообщениями с указанным routing_key.
• deduplication_key - ключ дедупликации. Опциональное поле, если оно указано, то перед каждой вставкой очередь проверяет наличие в индексе такого deduplication_key. По умолчанию очередь отвечает ошибкой, если такой ключ уже существует.
• payload - содержимое сообщения;
• metadata - метаданные;
• timestamp - время вставки.

Система гарантирует строгий порядок сообщений в пределах одного шарда — это значит, что сообщения с одинаковым ключом шардирования (sharding_key) обрабатываются строго друг за другом.

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

Чтение. Читать сообщения можно только в пределах одного шарда. Доступны три режима навигации:

• с самого первого сообщения;
• с указанного курсора;
• с последнего сообщения в очереди.

В ответе на запрос будет содержаться поле cursor - его можно использовать для того, чтобы возобновить чтение с последнего прочитанного сообщения (например для восстановления подписки).

Если выставить таймаут ожидания polling_timeout (в миллисекундах), очередь будет ждать появления новых сообщений и ответит сразу, как только они появятся. Это избавляет от пустых опросов, когда сообщений временно нет.

Фильтрация и консистентность. При чтении можно указать routing_key, и очередь вернет только сообщения с этим ключом. Также поддерживается фильтрация по нескольким sharding_key. Если какая-то реплика отстала от лидера и получила запрос на чтение с курсора, которого у нее еще нет, она задержит ответ до момента получения актуальных данных (в пределах указанного таймаута) — так обеспечивается консистентное чтение.

Подписка на сообщения (например, через gRPC) реализуется на уровне коннектора, а не самой очереди.

Шардирование. Система поддерживает два типа шардирования:

• Стандартное шардирование tarantool, реализованное через vshard.
• Статическое шардирование — его настройка и принципы работы описаны ниже.

Отказоустойчивость. Используется стандартная процедура tarantool без дополнительных модификаций.

Журналирование. Настройка журналов ведется через параметры конфигурационного файла. См. подробнее.

Метрики. Система предоставляет набор метрик для мониторинга состояния. См. подробнее.

Сценарии использования в данном руководстве разделены согласно компонентам TQE(MQ).

Сценарии охватывают следующие роли пользователей TQE(MQ):

• инженер по эксплуатации;
• администратор.

Инженер по эксплуатации выполняет следующие сценарии:

• Проверка статусов компонентов TQE(MQ);
• Операции с метриками.

Администратор выполняет следующие сценарии:

• Настройка параметров Tarantool;
• Установка компонентов TQE(MQ);
• Конфигурация компонентов TQE(MQ);
• Резервное копирование;
• Удержание сообщений.

Проверка статуса ядра выполняется с помощью команды:

 
$ curl localhost:8081/health

В случае штатного режима работы ядра возвращаемый ответ:

 
app is OK

Проверка статуса модуля API выполняется с помощью команды:

 
$ curl localhost:18184/readyz

В случае штатного режима работы модуля API возвращаемый ответ:

 
{    "consumer-tarantool": "OK",    "producer-tarantool": "OK",    "started": "OK"}

Где:

• started обязательное значение. Указывает на статус работы модуля API. Принимает значения "OK" или текст ошибки.
• consumer-tarantool необязательное значение. Указывает на статус подключения сервиса подписки на сообщения к ядру TQE(MQ). Принимает значения "OK" или текст ошибки.
• producer-tarantool необязательное значение. Указывает на статус подключения сервиса публикации сообщений к ядру TQE(MQ). Принимает значения "OK" или текст ошибки.

Для синхронных очередей перед установкой компонентов TQE укажите значение true для параметра use_mvcc_engine в конфигурации Tarantool для включения двухэтапной проверки дубликатов сообщений:

 
database:  use_mvcc_engine: true

На первом этапе проверки система производит поиск дубликатов среди сообщений, уже подтвержденных кластером. Если дубликат найден, система ведет себя в соответствии со значением параметра deduplication_mode. Если дубликат не найден на первом этапе, то на втором этапе система производит поиск дубликатов среди сообщений, которые записаны на мастер, но еще не подтверждены кластером. При обнаружении дубликата система возвращает ошибку concurrent unconfirmed publish, retry и отклоняет весь пакет.

При use_mvcc_engine: false двухэтапная проверка не работает и система выдаст предупреждение об этом.

Установка компонентов TQE(MQ) выполняется с помощью инсталлятора Ansible Tarantool Enterprise (ATE). Подготовка ATE к работе и процедуры установки компонентов TQE(MQ) описаны в документации ATE.

Конфигурация компонентов TQE(MQ) включает в себя раздельные настройки для ядра и для модуля API. Настройки описываются в формате YAML.

Конфигурацию выполняют сначала для ядра, а затем для модуля API.

Конфигурация ядра происходит стандартными средствами Tarantool 3: с помощью обновления файла конфигурации или конфигурации в etcd/Tarantool Config Storage. Подробнее про конфигурацию в Tarantool 3 можно прочитать здесь

YAML-файл конфигурации модуля ядра называется config.yml и имеет несколько секций. Пример YAML-файла конфигурации ядра:

 
credentials:  users:    user:      roles: [super]      password: passroles_cfg:  roles.tqe-storage:    features:      metrics_enabled: true      validation_enabled: true    queues:      - name: queue      - name: another_queue        deduplication_mode: basic      - name: archive_queue        storage: disk      - name: queue_disabled_index        disabled_filters_by: [routing_key, sharding_key]  roles.tqe-router:    sharding:      routing:        core-1:            buckets:            - 1            - [2,10]        core-2:            buckets:            - [11,20]        core-3:            buckets:            - [21,1000]

Раздел настроек features отвечает за настройку параметров функций ядра. Настраивается на уровне ролей roles.tqe-storage и roles.tqe-router.

 
roles_cfg:  roles.tqe-storage:    features:      metrics_enabled: true      validation_enabled: false  roles.tqe-router:    features:      metrics_enabled: false      validation_enabled: true

Где:

• metrics_enabled - включение/выключение сбора метрик для методов API ядра. По умолчанию true;
• validation_enabled - включение/выключение валидации сообщений для методов API ядра. По умолчанию true.

Раздел настроек queues отвечает за описание используемых очередей сообщений и настройку их параметров. Настраивается на уровне ролей roles.tqe-storage и roles.tqe-router. Имеет следующую структуру:

 
roles_cfg:  roles.tqe-storage:      queues:        - name: queue1          latency: 1          disabled_filters_by: [sharding_key]          deduplication_mode: basic        - name: queue2

где:

• name - название очереди сообщений.
• latency - задержка в миллисекундах между оповещениями подписчика о новых сообщениях. По умолчанию – 1.
• deduplication_mode - режим обработки отдельных дублированных сообщений и пакетов, содержащих дублированные сообщения. Может принимать значения:
  • basic - если публикуемое сообщение является дубликатом, или пакет сообщений содержит дубликат, то публикация отменяется и система возвращает ошибку дедупликации. Это значение используется по умолчанию.
  • extended - режим расширенной дедупликации;
  • keep_latest - если сообщение является дубликатом, то оно публикуется. А ранее опубликованное сообщение удаляется;
  • keep_first - если сообщение является дубликатом, то оно не публикуется. Вместо этого возвращается id ранее опубликованного сообщения. Ошибка не возвращается.
• poll_max_batch – максимальное количество сообщений в пакете, который вернет очередь в ответ на один запрос потребителя сообщений. Сообщения, выходящие за это значение формируются в следующий пакет. Значение по умолчанию: 100.
• poll_min_batch – целевое минимальное количество сообщений, которое должна вернуть очередь в ответ на один запрос потребителя сообщений. Если количество накопленных сообщений в пакете ниже значения poll_min_batch (включая 0), но время, указанное в poll_batch_timeout истекло, то пакет будет отправлен в ответ на запрос потребителя сообщений. Значение не может превышать poll_max_batch. Значение по умолчанию: 10.
• poll_batch_timeout - время в секундах, отводимое на сбор сообщений в пакет после получения первого сообщения. Допускаются дробные значения (разделитель — точка). По истечении времени, указанного в параметре, очередь отправит пакет сообщений, даже если количество сообщений не достигло значения poll_min_batch. Значение по умолчанию: 0.500 (полсекунды).
• storage – движок хранения очереди. Допустимые значения:
  • memory - для использования memtx;
  • disk для использования vinyl. Значение по умолчанию - memory.
• poll_yield_every – период (в сообщениях), с которым обработчик запроса будет передавать управление другим обработчикам. По умолчанию – 512.
• disabled_filters_by - список отключенных фильтров. Отключение фильтра делает невозможным подписку с фильтрацией по указанному полю. Изменение этой опции после создания очереди позволяет включать фильтрацию по полю при его удалении из списка, но не позволяет отключать ее. По умолчанию - [] (фильтрация включена по всем полям). Возможные значения - routing_key, sharding_key.

Также существует другой, сокращенный вариант настройки параметров раздела queues:

 
roles_cfg:  roles.tqe-storage:    queues: ["queue1", "queue2"]

В результате такого запроса создадутся 2 очереди, queue1 и queue2, со значениями остальных параметров по умолчанию.

Расширенная дедупликация в TQE(MQ) — это режим, при котором проверяется не только уникальность deduplication_key сообщения, но и полное совпадение его содержимого. Если при одинаковом deduplication_key содержимое сообщений различается — система возвращает ошибку. Такой режим позволяет безопасно работать нескольким независимым публикаторам (основному и резервному): дублирование одинаковых сообщений допускается, а расхождение в данных обнаруживается сразу.

Для включения режима расширенной дедупликации необходимо добавить параметр deduplication_mode: extended в конфигурацию очереди в роль roles.tqe-storage:

 
roles_cfg:  roles.tqe-storage:    queues:      - name: dedup        deduplication_mode: extended

При включенном режиме расширенной дедупликации обнаружение дубликатов с разным содержимым в одиночном сообщении или при публикации в рамках пакетной отправки приводит к ошибке failed to publish messages to tarantool: storage.produce error: id (*): payload may be corrupted.

TQE(MQ) использует механизм виртуального шардирования (vshard) для организации распределения данных. Базовой атомарной единицей в vshard является бакет, логический контейнер данных с уникальным идентификатором. Если два различных сообщения принадлежат одному бакету, то они гарантированно хранятся на одном наборе реплик с ролью storage.

В TQE(MQ) доступен режим статического шардирования — это режим ручной настройки карты распределения бакетов по наборам реплик в кластере. Настройка задается с помощью глобальной конфигурации кластера, после чего необходимо выполнить начальную загрузку. В результате в наборах реплик кластера создаются бакеты в соответствии с заданной картой.

Конфигурация статического шардирования осуществляется посредством настройки роли roles.tqe-router в разделе roles_cfg:

• в разделе sharding.routing необходимо указать карту бакетов в формате replicaset-alias:bucket-range;
  • core-n - псевдоним набора реплик (replica set) из топологии;
  • buckets - диапазон обслуживаемых бакетов. Может принимать массив значений, где каждое значение - либо id одного бакета, либо диапазон бакетов "от и до".

Пример:

 
roles_cfg:  roles.tqe-router:    sharding:      routing:        core-1:            buckets:            - 1                # Можно задать идентификатор бакета точечно.            - 2            - [3,300]          # Можно указать диапазон бакетов.        core-2:            buckets:            - [301,500]        core-3:            buckets:            - [501,1000]

По окончании настройки необходимо выполнить загрузку кластера стандартной командой tarantool:

 
$ tt replicaset vshard bootstrap .

Оркестратор — это компонент, который при инициализации разбивает бакеты каждого шарда в кластере на партиции — непрерывные диапазоны бакетов. Каждая партиция уникальна в рамках кластера. Количество партиций задается при установке экземпляра TQE(MQ) и не может быть изменено без его переустановки.

Оркестратор автоматически распределяет партиции между потребителями в том случае, если потребители объединены в группы. В результате распределения один потребитель может обрабатывать несколько партиций, но каждая партиция не может обрабатывается более чем одним потребителем. Оркестратор отслеживает состав групп и при подключении или отключении потребителя от группы запускает автоматическую перебалансировку - партиции перераспределяются между новым количеством потребителей в группе. Это исключает ручные перенастройки, предотвращает дублирование сообщений и обеспечивает непрерывность обработки сообщений потребителями.

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

Оркестратор настраивается для всего кластера на уровне роли roles.tqe-orchestrator:

• partitions - количество партиций, на которые нужно разделить все шарды в кластере. Рекомендуется использовать число, на которое можно поделить общее количество бакетов кластера без остатка.
• rebalance_timeout - задержка (в секундах) между моментом, когда произошло изменение состава группы, и моментом, когда оркестратор фактически запускает перераспределение партиций.
• sync_timeout - таймаут активности потребителя (в секундах). Если Оркестратор не получает от потребителя запрос синхронизации в течение этого времени, то такой потребитель считается неактивным и исключается из группы. Исключенный потребитель может войти в группу повторно.

Пример:

 
roles:  roles.tqe-orchestrator:    partitions: 10    rebalance_timeout: 20    sync_timeout: 5

Параметры конфигурации компонента модуля API должны быть определены до его запуска и запуска TQE(MQ) во избежание проблем с подключением к очереди сообщений и аварийным завершением работы модуля API.

YAML-файл конфигурации модуля API называется config.service.yml и имеет несколько секций. Полный перечень настроек представлен в следующем примере:

 
app_name: MESSAGE_QUEUE_EE_APIapp_version: testcore_host: 0.0.0.0core_port: 18184grpc_host: 0.0.0.0grpc_port: 18182producer:  enabled: true  tarantool:    user: user    pass: pass    connections:      routers:        - "localhost:3301"consumer:  enabled: true  polling_timeout: 500ms  cursor_serilizer: 'yaml'  buffer: 12  concurrency: 2  access_mode: 'prefer_ro'  tarantool:    user: user    pass: pass    connections:      storage-1:        - "localhost:3301"log:  to: file  file: log.jsonl  format: json  level: info

Основные параметры обеспечивают настройку адресов и портов подключения модуля API:

• app_name - название приложения;
• app_version - версия приложения;
• core_host - адрес сбора метрик и проверки состояния модуля API;
• core_port - порт сбора метрик и проверки состояния модуля API;
• grpc_options - раздел для конфигурации grpc-сервера. См подробнее. Установленные значения не изменяются после запуска grpc-сервера:
  • initial_conn_window_size - начальное окно в байтах http2-соединения, по умолчанию 64kb;
  • initial_window_size - начальное окно http2-стрима в байта, по умолчанию 64kb;
  • header_table_size - размер динамической таблицы с заголовками http2, по умолчанию не задано (динамическая таблица не используется);
  • max_header_list_size - максимальный размер таблицы с заголовками http2, по умолчанию не задано;
  • max_concurrent_streams - количество одновременных http-стримов, по умолчанию 128;
  • num_stream_workers - количество обработчиков для обработки входящих запросов, по умолчанию 0 (под каждый http2-стрим создается отдельная горутина);
  • max_recv_msg_size - максимальный размер входящего сообщения в байтах, по умолчанию 4mb;
  • max_send_msg_size - максимальный размер исходящего сообщения в байтах, по умолчанию 2mb;
  • read_buffer_size - размер буфер на чтение, по умолчанию 256kb;
  • write_buffer_size - размер буфер на запись, по умолчанию 256kb;
  • shared_write_buffer - позволяет переиспользовать буфер на запись, а не создавать под каждое подключение, по умолчанию false;
• grpc_listen - набор интрефейсов, на которых обслуживает gRPC-сервер;
  • uri - адрес прослушивания, например: unix:///tmp/tqe-mq.sock или tcp://localhost:18182
• grpc_host - (устарело после введения grpc_listen) адрес модуля API;
• grpc_port - (устарело после введения grpc_listen) порт модуля API.

Пример настройки:

 
app_name: MESSAGE_QUEUE_EE_APIapp_version: 1.0core_host: 0.0.0.0core_port: 18184grpc_listen:  - uri: 'tcp://0.0.0.0:18182'grpc_host: 0.0.0.0grpc_port: 18182grpc_options:  initial_conn_window_size: 65535  initial_window_size: 65535  header_table_size: 16777216  max_header_list_size: 128  max_concurrent_streams: 128  num_stream_workers: 0  max_recv_msg_size: 4294967296  max_send_msg_size: 2147483647  read_buffer_size: 262144  write_buffer_size: 262144  shared_write_buffer: false

Параметры сетевого подключения клиента к очереди по протоколу зашифрованного канала связи TLS (Transport Layer Security) указываются в разделе настроек grpc_options файла config.service.yml:

• tls - раздел настроек подключения по протоколу зашифрованного канала связи TLS (Transport Layer Security);
  • enabled - включает и выключает подключение по TLS. Обязательный параметр. Принимает значения:
    • true - подключение по TLS включено;
    • false - подключение по TLS выключено. Это значение используется по умолчанию.
  • cert_file - абсолютный или относительный путь к файлу, который содержит публичный сертификат соответствия TLS (public_key). Параметр обязателен при enabled: true.
  • key_file - абсолютный или относительный путь к файлу, который содержит закрытый (приватный) ключ. Параметр обязателен при enabled: true.
  • ca_file - абсолютный или относительный путь к файлу, который содержит сертификаты доверенных удостоверяющих центров.
  • password - пароль для расшифровки файла закрытого ключа. Если указан одновременно с password_file, то приоритетным считается путь в password_file
  • password_file - абсолютный или относительный путь к файлу, содержащему пароль для расшифровки файла закрытого ключа. Если указан одновременно с password, то приоритетным считается путь в password_file;
  • ciphers - список алгоритмов шифрования, которые разрешены при установке защищенного соединения. Подключение выполняется только если клиент и сервер имеют хотя бы один одинаковый алгоритм в списках разрешенных. Список поддерживаемых алгоритмов представлен в таблице ниже.

Поддерживаемые алгоритмы шифрования.

Пример:

 
grpc_options:  tls:    enabled: true    cert_file: /path/to/server.crt    key_file: /path/to/server.key    ca_file: /path/to/ca.crt      password: "secret"            password_file: /path/to/passwords.txt      ciphers: 'ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256'

Подключение к маршрутизаторам для публикации сообщений в очередь бывает динамическим и статическим:

• Динамическое подключение подразумевает, что адреса маршрутизаторов и хранилищ указываются и, при необходимости, изменяются удаленной службой Tarantool configuration Storage (TcS) или распределенным хранилищем etcd.
• При статическом подключении адреса маршрутизаторов и хранилищ указываются и, при необходимости, изменяются пользователем самостоятельно.

При динамическом подключении изменения на стороне маршрутизаторов или хранилищ управляются удаленно: службой Tarantool configuration Storage (TcS) или распределенным хранилищем etcd.

Настройки подключения к TcS или etcd указываются в разделе config.

Параметры динамического подключения:

• storage - раздела для описания параметров динамического подключения к удаленной службе;
• endpoints- адреса точек подключения;
• username- логин для подключения;
• password- пароль для подключения;
• prefix - префикс для маршрутизации внутри удаленной службы;
• timeout - таймаут подключения в секундах;
• reconnect_after - интервал в секундах между повторными подключениями;
• max_reconnect_attempts - максимальное число попыток повторных подключений;

Динамическое подключение может быть осуществлено по протоколу зашифрованного канала связи TLS (Transport Layer Security). Для этого необходимо указать настройки в разделе tls:

• enabled - включает и выключает подключение по TLS. Обязательный параметр. Принимает значения:
  • true - подключение по TLS включено;
  • false - подключение по TLS выключено. Это значение используется по умолчанию.
• cert_file - абсолютный или относительный путь к файлу, который содержит публичный сертификат соответствия TLS (public_key). Параметр обязателен при enabled: true.
• key_file - абсолютный или относительный путь к файлу, который содержит закрытый (приватный) ключ. Параметр обязателен при enabled: true.
• ca_file - абсолютный или относительный путь к файлу, который содержит сертификаты доверенных удостоверяющих центров.
• password - пароль для расшифровки файла закрытого ключа. Если указан одновременно с password_file, то приоритетным считается путь в password_file
• password_file - абсолютный или относительный путь к файлу, содержащему пароль для расшифровки файла закрытого ключа. Если указан одновременно с password, то приоритетным считается путь в password_file;
• ciphers - список алгоритмов шифрования, которые разрешены при установке защищенного соединения. Подключение выполняется только если клиент и сервер имеют хотя бы один одинаковый алгоритм в списках разрешенных. См. список разрешенных алгоритмов.

Пример настройки динамического подключения с включенным протоколом зашифрованного канала связи TLS:

 
config:  storage:    endpoints:      - "127.0.0.1:4401"    username: sampleuser    password: "123456"    prefix: "/tarantool/tqe-mq"    timeout: 5s    reconnect_after: 5s    max_reconnect_attempts: 3    tls:      enabled: true      cert_file: /path/to/server.crt      key_file: /path/to/server.key      ca_file: /path/to/ca.crt        password: "secret"              password_file: /path/to/passwords.txt        ciphers: 'ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256'

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

Параметры статического подключения к маршрутизаторам:

• enabled - включает режим публикации сообщений в очередь. Возможные значения:
  • true - публикация сообщений в очередь включена;
  • false - публикация сообщений в очередь выключена;
• tarantool - раздел настроек доступа к ядру TQE(MQ):
  • user - логин доступа к очереди (указывается в разделе credentials конфигурации ядра TQE(MQ));
  • pass - пароль доступа к очереди (указывается в разделе credentials конфигурации ядра TQE(MQ));
  • connections - перечень названий типов подключений. Параметр не указывается в случае динамического подключения;

Пример настройки:

 
producer:  enabled: true  tarantool:    user: user    pass: pass    connections:      rs-1:        - "localhost:3301"

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

Параметры статического подключения к хранилищам:

• enabled - включает режим потребления сообщений из очереди. Возможные значения:
  • true - потребление сообщений из очереди включено;
  • false - потребление сообщений из очереди выключено;
• polling_timeout - время задержки между запросами на получение новых сообщений, с указанием единиц измерения времени. Пример: 500ms - 500 миллисекунд;
• cursor_serializer - тип сериализатора курсора. По умолчанию пустое значение (используется сериализатор base64). Доступные значения: json, yaml;
• buffer_size - размер внутреннего буфера потребителя в количестве сообщений. Значение по умолчанию: 8;
• concurrency - количество параллельных потоков потребления сообщений. Значение по умолчанию: 1;
• access_mode - режим взаимодействия с наборами реплик tarantool. Значение по умолчанию: prefer_rw. Подробнее о возможных значениях см. по ссылке;
• tarantool - раздел настроек доступа к очереди TQE MQ:
  • user - логин доступа к очереди (указывается в разделе credentials конфигурации ядра TQE(MQ));
  • pass - пароль доступа к очереди (указывается в разделе credentials конфигурации ядра TQE(MQ));
  • connections - раздел адресов хранилищ для подписки на сообщения. Параметры этого раздела не указываются в случае динамического подключения;
  • queues - опциональная раздел доступных очередей TQE(MQ).

Пример:

 
consumer:  enabled: true  polling_timeout: 500ms  cursor_serilizer: 'yaml'  buffer: 12  concurrency: 2  access_mode: 'prefer_ro'  tarantool:    user: user    pass: pass    queues:      queue:        connections:          storage-1:            - "localhost:3301"

Доступные параметры секции consumer:

• enabled - параметр доступности сервиса:
  • true - доступен,
  • false - нет;
• polling_timeout - время задержки между запросами новых сообщений, пример: 500ms;
• cursor_serializer - тип сериализатора курсора. По умолчанию пустое значение (используется сериализатор base64). Доступные значения: json, yaml.
• buffer_size - размер внутреннего буфера потребителя. По умолчанию: 8.
• concurrency - количество параллельных потоков потребления сообщений. По умолчанию: 1.
• access_mode - режим взаимодействия с наборами реплик tarantool. Значение по умолчанию: prefer_rw. Подробнее о возможных значениях см. по ссылке.
• tarantool - секция конфигурации доступа к очереди TQE MQ:
  • user - логин доступа к очереди (указывается в секции creds конфигурации ядра TQE(MQ));
  • pass - пароль доступа к очереди (указывается в секции creds конфигурации ядра TQE(MQ));
  • connections - секция адресов хранилищ для подписки на сообщения.
  • queues - опциональная секция доступных очередей TQE(MQ);

Параметры журналирования задаются в разделе log модуля API. Структура раздела аналогична разделу log ядра Tarantool.

Доступные параметры секции log:

• to - место записей журнала, принимаемые значения:
  • stderr - вывод событий журнала в stderr. Это значение используется по умолчанию, если модуль API запущен не в фоновом режиме;
  • file - запись в файл, указанный в параметре log.file. Это значение применяется по умолчанию, если модуль API запущен в фоновом режиме (с флагом -d).
  • syslog - отправка в системный журнал syslog, параметры подключения задаются в секции syslog;
• file - абсолютный или относительный путь до файла для записи событий журнала. Используется только при to: file. При запуске модуля API в фоновом режиме и пустом значении параметра в качестве файла записи используется server.jsonl.
• format - формат вывода событий, принимаемые значения:
  • plain - простой текстовый формат. Это значение используется по умолчанию при to: stderr и to: file.
  • json - структурированный формат JSON. Это значение используется по умолчанию при to: syslog.
• nonblock - включает неблокирующий режим записи журнала. В неблокирующем режиме данные отправляются в буфер и приложение не ждет завершения записи в журнал. В этом режиме данные могут потеряться, если буфер переполнен. Принимает значения:
  • false - неблокирующий режим выключен. Пока данные не запишутся в журнал, приложение не продолжает работу. Это значение используется по умолчанию.
  • true - неблокирующий режим включен. Значение параметра используется системой при:
    • to: file - система записывает события журнала в файл через буфер diode, который допускает потерю событий при переполнении.
    • to: syslog + server: unix - система переводит подключение на канал unixgram, который работает в режиме без установки соединения и отбрасывает события при переполнении очереди ядра. Значение параметра игнорируется системой и при загрузке конфигурации и запуске системы журналирования выводится предупреждение об этом если:
    • to: stderr - в этом случае отсутствует буфер и события журнала выводятся мгновенно;
    • to: syslog + server: udp - для подключения по udp неблокирующий режим всегда принудительно включен;
    • to: syslog + server: tcp/ server: tcp+tls - для подключения по tcp неблокирующий режим всегда принудительно выключен;
• level - уровень записи, определяющий важность события. Выбранное значение приводит к записи событий от этого значения и выше по приоритету. Принимаемые значения:
  • debug - детализированная диагностическая информация (самый низкий приоритет);
  • info - информационные сообщения о штатной работе (значение по умолчанию);
  • warn - информация о потенциально опасных событиях, при которых работа приложения продолжается;
  • error - запись информации о серьезных ошибках, которые требуют внимания;
  • dpanic - информация о панике в режиме разработки;
  • panic - критическая ошибка, при которой приложение регистрирует панику и может завершить работу;
  • fatal- фатальная ошибка, которая привела к аварийной остановке приложения;
• syslog - параметры записи событий в системный журнал syslog (используются только при to: syslog):
  • server - URL-адрес получателя событий журнала. Поддерживаемые типы подключения:
    • unix - например, unix:/dev/log. Для этой схемы тип канала регулируется параметром nonblock:
      • при nonblock: false — используется канал unix;
      • при nonblock: true используется канал unixgram, который работает в режиме без установления соединения;
    • udp - например udp://host:514;
    • tcp - например tcp://host:514;
    • tcp+tls - например tcp+tls://host:6514.
  • identity - идентификатор приложения, по умолчанию message-queue-ee;
  • facility - категория источника записи события для его сохранения в системном журнале. Принимает значения:
    • kern - ядро системы;
    • user - пользовательские процессы;
    • mail - почтовые службы;
    • daemon - фоновая служба;
    • auth - служба авторизации;
    • syslog - внутренние сообщения журнала;
    • lpr - системы печати;
    • news - новостные серверы (NNTP);
    • uucp - UUCP-подсистема;
    • cron - планировщик задач;
    • authpriv - привилегированные сообщения безопасности;
    • ftp - FTP-серверы;
    • local0...local7 - пользовательские приложения. По умолчанию используется знгачение local7.
  • tls - параметры TLS, используются только при схеме tcp+tls:
    • ca_file - абсолютный или относительный путь к файлу, который содержит сертификаты доверенных удостоверяющих центров;
    • cert_file - абсолютный или относительный путь к файлу, который содержит публичный сертификат соответствия TLS (public_key);
    • key_file - абсолютный или относительный путь к файлу, который содержит закрытый (приватный) ключ;
    • server_name - имя сервера для проверки сертификата (SNI);
    • insecure_skip_verify - отключение проверки сертификата сервера, по умолчанию false.

Пример настройки записи событий журнала в файл:

 
log:  to: file  file: log.jsonl  format: json  level: info  nonblock: false

Пример настройки отправки событий в системный журнал syslog:

 
log:  to: syslog  level: info  syslog:    server: "unix:/dev/log"                    # также: udp://host:514 | tcp://host:514 | tcp+tls://host:6514    identity: "message-queue-ee"    facility: "local7"    tls:                                       # только для схемы tcp+tls      ca_file: /etc/ssl/syslog-ca.pem      cert_file: /etc/ssl/syslog-client.pem      key_file: /etc/ssl/syslog-client.key      server_name: ""      insecure_skip_verify: false

Для резервного копирования данных TQE(MQ) необходимо выполнить процедуру бэкапа ядра (кластерного приложения) c помощью инсталлятора Ansible Tarantool Enterprise.

Так как объем памяти, доступный для хранения сообщений, ограничен местом на диске, может возникнуть потребность удаления ненужных сообщений. Для периодической очистки очереди сообщений TQE(MQ) использует модуль expirationd. Соблюдение условий для удаления сообщений контролируется функцией удержания сообщений (data retention).

Функция удержания сообщений работает в соответствии с установленной политикой (policy) очистки ненужных сообщений. Политика может быть включена (policy: cleanup) или отключена (policy: disabled). Во включенном состоянии политика удаляет сообщения в зависимости от установленных параметров:

• time - допустимое время жизни сообщения в очереди в секундах. Значение 0 или отсутствие параметра отключает проверку, сообщения могут жить в очереди "бесконечно" долго.
• length - допустимая длина очереди. Значение 0 или отсутствие параметра отключает проверку, очередь становится "бесконечно" длинной.
• consumers_required - включает очистку по прочитанным сообщениям. При включенной очистке политика сравнивает указанное значение с количеством персистентных подписчиков:
  • Если количество персистентных подписчиков превышает или совпадает со значением сonsumers_required, то сообщение будет удалено только после того, как его прочтут все персистентные подписчики.
  • Если количество персистентных подписчиков ниже значения сonsumers_required, то сообщение не будет удалено очисткой по прочитанным сообщениям.
  • При consumers_required: 0 и каком-то количестве персистентных подписчиков очистка будет ждать, пока все персистентные подписчики прочтут сообщение.
  • При consumers_required: 0 и без персистентных подписчиков очистка считает все сообщения прочитанными и удаляет их сразу.

Политика очистки ненужных сообщений и смежные параметры настраиваются в разделе retention в роли roles.tqe-storage. Раздел retention можно настраивать как на глобальном уровне, так и на уровне отдельной очереди. При наличии разницы в параметрах, приоритетной считается настройка отдельной очереди. Подробнее о настройках раздела retention смотрите в документации по expirationd.

По умолчанию конфигурация для всех очередей выглядит так:

 
roles_cfg:  roles.tqe-storage:    retention:      time: 3153600000             # Допустимое время жизни сообщений всех очередей в секундах.      length: 0                    # Проверка на максимальную длину очереди не выполняется      policy: cleanup              # Политика очистки включена.      options:                     # Конфигурационные параметры для expirationd.start.        iteration_delay: 0         # Максимальное время в секундах для fiber.sleep между итерациями.

Переопределить конфигурацию для определенной очереди можно указав раздел retention с отличными значениями параметров в разделе конфигурации конкретной очереди example_queue:

 
roles_cfg:  roles.tqe-storage:    queues:      - name: example_queue        retention:          time: 100                  # Допустимое время жизни сообщений конкретной очереди в секундах.          length: 1000               # Максимальная длина очереди указана.          policy: cleanup            # Политика очистки включена.          consumers_required: 2      # Количество постоянных подписчиков, после прочтения которыми сообщение может быть удалено.          options:            tuples_per_iteration: 512

Следующий пример конфигурации позволяет:

• Hастроить глобальное время удержания сообщений по всем очередям в 120 секунд и указать максимальную длину очереди в 1024 сообщения;
• Для очереди input:
  • Переопределить время удержания в 60 секунд;
  • Увеличить максимальную длину очереди до 2048 сообщений;
• Для вызова expirationd.start:
  • Указать опцию full_scan_time;
• Для очереди output:
  • Отключить очистку очереди от устаревших сообщений.

В подобной конфигурации раздел роли roles.tqe-storage будет выглядеть так:

 
roles_cfg:  roles.tqe-storage:    retention:      time: 120      length: 1024    queues:      - name: input        retention:          time: 60          length: 2048          options:            full_scan_time: 1024      - name: output        retention:          policy: disabled # Политика очистки отключена.

Получение сообщений в TQE(MQ) возможно только при наличии подписки.

Для получения подписки, выполните запрос SubscriptionRequest без указания consume_id.

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

Для получения персистентной подписки, выполните запрос SubscriptionRequest с указанием параметра consume_id.

С каждым сообщением персистентным подписчикам приходит значение параметра cursor, соответствующее позиции сообщения в очереди. Значение cursor критически важно при восстановлении состояния подписки после отключения.

Восстановить состояние подписки можно при переподключении. Для этого после отключения нужно выполнить запрос SubscriptionRequest с теми же значениями параметров queue, routing_key, sharding_key, consume_id, которые были использованы при инициализации подписки. В запрос обязательно нужно включить параметр cursor, который был использован непосредственно перед отключением.

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

При разрыве соединения можно повторно инициировать подписку, передав свой consume_id. Сервер автоматически восстановит состояние подписки из хранилища и возобновит чтение с последней известной позиции.

После перезапуска сервер теряет состояние в памяти. Вы должны переподключиться и отправить запрос SubscriptionRequest, включающий consume_id и cursor последнего полученного сообщения. Сервер автоматически восстановит состояние из персистентного хранилища и возобновит передачу сообщений.

При перезапуске реплики gRPC-сервер автоматически обнаруживает сбой, выбирает другую доступную реплику и возобновляет обработку запросов с последней известной позиции.

После переключения на новую мастер-реплику gRPC-сервер автоматически устанавливает соединение и проверяет состояние подписок. Для персистентных подписок их состояние полностью восстанавливается на новой реплике.

Встроенный инструментарий TQE(MQ) предоставляет метрики для оценки процесса работы с сообщениями в очередях.

Эти метрики предоставляют диагностическую информацию для использования в стороннем программном обеспечении.

Метрики модулей предоставляются в формате prometheus.

Метрики ядра TQE(MQ) соответствуют стандартному набору метрик tarantool. С перечнем метрик можно ознакомиться в справочнике метрик Tarantool.

Метрики доступны по HTTP-адресу:

 
$ curl localhost:8081/metrics

Пример частичного вывода метрик:

 
# HELP tnt_vinyl_disk_index_size Amount of index stored in files# TYPE tnt_vinyl_disk_index_size gaugetnt_vinyl_disk_index_size{alias="app"} 0# HELP tnt_read_only Is instance read only# TYPE tnt_read_only gaugetnt_read_only{alias="app"} 0# HELP tnt_vinyl_disk_data_size Amount of data stored in files# TYPE tnt_vinyl_disk_data_size gaugetnt_vinyl_disk_data_size{alias="app"} 0...

Метрики ядра предоставляют следующую диагностическую информацию, специфичную для TQE (MQ):

• mqee_tnt_duplicates_total – счетчик дубликатов
• mqee_tnt_latency_seconds - время на выполнения стадии обработки сообщения в модуле ядра:
  • point - стадия обработки сообщений, может принимать следующие значения:
    • publish-request-tnt-storage-persisted;
    • broadcast-request-tnt-storage-persisted;
  • queue - название очереди;
  • routing_key - routing_key обрабатываемоего сообщения ("" для пакетных запросов);
  • result - результат выполнения стадии:
    • success - если стадия завершилась успешно;
    • fail - если стадия завершилась с ошибкой.
• mqee_tnt_pollers_total - индикатор текущего количества поллеров:
  • queue - название очереди;
• mqee_tnt_consumer_lag_ms - отображает оставание потребителя в милисекундах:
  • queue - название очереди.

Метрики модуля API доступны на HTTP-адресе /metrics:

 
$ curl localhost:18184/metrics

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

 
# HELP go_gc_duration_seconds A summary of the pause duration of garbage collection cycles.# TYPE go_gc_duration_seconds summarygo_gc_duration_seconds{quantile="0"} 5.3002e-05go_gc_duration_seconds{quantile="0.25"} 7.574e-05go_gc_duration_seconds{quantile="0.5"} 9.047e-05go_gc_duration_seconds{quantile="0.75"} 0.000111856go_gc_duration_seconds{quantile="1"} 0.000338263go_gc_duration_seconds_sum 1.305067544go_gc_duration_seconds_count 11608...

Метрики модуля API предоставляют следующую диагностическую информацию:

• mqee_grpc_publishing_request_messages - гистограмма опубликованных сообщений в очередь:
  • grpc_method - название удаленной процедуры;
  • queue - название очереди;
  • routing_key - значение routing_key, по которому будет произведена публикация ("" для пакетных запросов);
  • result - результат публикации:
    • success - если сообщения успешно опубликованы;
    • fail - если сообщения не опубликованы.
• mqee_grpc_requests_size_bytes_total - счетчик общего размера запросов в байтах:
  • grpc_method - название удаленной процедуры;
  • queue - название очереди;
  • routing_key - routing_key обрабатываемоего сообщения ("" для пакетных запросов).
• mqee_grpc_subscribers_count - индикатор текущего количества подписчиков:
  • queue - название очереди;
  • routing_key - значение routing_key, по которому была произведена подписка.
• mqee_grpc_app_info - информация об очереди (mqee_grpc_app_info{app_name="MESSAGE_QUEUE_EE_API",app_version="test"} 1);
• mqee_grpc_gomaxprocs - индикатор текущего значения GOMAXPROCS.
• mqee_grpc_tuples_received - гистограмма количества сообщений, которые были получены от ядра или из кэша. Включает в себя mqee_grpc_tuples_received_sum (суммарное количество полученных сообщений) и mqee_grpc_tuples_received_count (число запросов на получение сообщений в ядро или кэш):
  • queue - название очереди;
  • routing_key - значение routing_key, по которому была произведена подписка.
• mqee_grpc_sent_messages - гистограмма количества сообщений, отправленных потребителям gRPC-сервером. Включает в себя mqee_grpc_sent_messages_sum (суммарное количество отправленных сообщений) и mqee_grpc_sent_messages_count (число отправленных ответов):
  • queue - название очереди;
  • routing_key - значение routing_key, по которому была произведена подписка.
• grpc_server_started_total - счетчик полученных запросов на удаленную процедуру:
  • grpc_type - тип подключения (unary, client_stream, server_stream, bidi_stream);
  • grpc_service - название gRPC-сервиса;
  • grpc_method - название процедуры:
    • Produce;
    • ProduceStream;
    • Broadcast;
    • ServerReflectionInfo;
    • Subscribe.
• grpc_server_handled_total - счетчик завершенных удаленных процедур:
  • grpc_type - тип подключения (unary, client_stream, server_stream, bidi_stream):
  • grpc_service - название gRPC-сервиса;
  • grpc_method - название процедуры (см. список в grpc_server_started_total).
  • grpc_code - код ответа gRPC-код.
• grpc_server_msg_received_total - счетчик полученных сообщений:
  • grpc_type - тип подключения(unary, client_stream, server_stream, bidi_stream);
  • grpc_service - название gRPC-сервиса;
  • grpc_method - название процедуры (см. список в grpc_server_started_total).
• grpc_server_msg_sent_total - счетчик отправленных сообщений:
  • grpc_type - тип подключения(unary, client_stream, server_stream, bidi_stream);
  • grpc_service - название gRPC-сервиса;
  • grpc_method - название процедуры (см. список в grpc_server_started_total).
• grpc_server_handling_seconds - гистограмма секунд на обработку gRPC-вызова:
  • grpc_type - тип подключения(unary, client_stream, server_stream, bidi_stream);
  • grpc_service - название gRPC-сервиса;
  • grpc_method - название процедуры (см. список в grpc_server_started_total).
• go_gc_duration_seconds - сводная информация по длительности паузы на цикл сборки мусора окружения Go. Используемый диапазон в секундах: 0, 0.25, 0.5, 0.75, 1;
• go_goroutines - индикатор текущего количества goroutine;
• go_info - информация о версии окружения Go;
• go_memstats - набор метрик для отслеживания использования памяти средой выполнения Go;
• go_threads - индикатор текущего количества потоков (threads) операционной системы, используемых средой выполнения Go.
• mqee_grpc_processing_time - время на выполнение стадии обработки сообщения в модуле API:
  • point - стадия обработки сообщений, может принимать следующие значения:
    • produce-request-received;
    • broadcast-request-received;
    • produce-request-validated;
    • broadcast-request-validated;
    • produce-request-persisted;
    • broadcast-request-persisted;
    • subscribe-notifications-sent;
  • queue - название очереди;
  • routing_key - routing_key обрабатываемоего сообщения (пустая строка, если сообщение без routing_key);
  • result - результат выполнения стадии:
    • success - если стадия завершилась успешно;
    • fail - если стадия завершилась с ошибкой.

Экспорт метрик компонентов TQE (MQ) осуществляется при помощи технологической роли roles.metrics-export.

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

• prefix - префикс для экспорта метрик в формате <prefix>.<metric_name>;
• host - IP-адрес порта сервера Graphite;
• port - номер порта сервера Graphite;
• send_interval - периодичность отправки метрик на сервер Graphite, в секундах. Необязательный параметр.

В результате настройки указанная цель будет получать все метрики компонентов TQE (MQ).

Пример настройки роли roles.metrics-export для экспорта метрик в Graphite:

 
roles_cfg:  roles.metrics-export:    graphite:    - prefix: 'tqe'      host: '127.0.0.1'      port: 2003      send_interval: 1    - prefix: 'master'      host: '127.0.0.2'      port: 4444

Также, используя роль roles.metrics-export можно подготовить предоставление метрик в формате prometheus:

 
roles_cfg:    roles.metrics-export:        http:        - listen: 8081          endpoints:          - format: prometheus            path: '/metrics'

Больше информации о роли roles.metrics-export смотрите по ссылке.

Для поддержки мониторинга TQE(MQ) можно использовать базовый дашборд Tarantool Queue Enterprise (MQ) Overview. Дашборд находится в архиве поставки проекта в директории ./monitoring. Для запуска дашборда необходимо выполнить команду:

 
docker compose -f ./monitoring/docker-compose.yml up -d

В базовый дашборд включены следующие группы панелей:

• gRPC server queue info - общее состояние очереди: количество сообщений, объем в байтах;
• Tarantool queue info - состояние очереди со стороны Tarantool: число задач в спейсе, время жизни сообщений, данные о дедупликации;
• gRPC server queue detailed info - детализация по каждой очереди: глубина, скорость приема и выдачи, задержка операций;
• gRPC server requests - интенсивность и длительность gRPC-вызовов, распределение по статусам и методам;
• gRPC server Go runtime - метрики среды исполнения Go для компонентов, реализованных на Go;
• gRPC server process info - системные метрики процесса: CPU, память и пр;
• Tarantool cluster overview - общая информация о кластере: состояние экземпляров, нагрузка, память, статус конфигурации, режим read-only и выборы лидера;
• Tarantool replication overview - информация о репликации: задержка, состояние, статистика синхронной очереди;
• Tarantool network activity - сетевая активность экземпляров: входящий/исходящий трафик, соединения, детализация запросов и потребление памяти;
• Tarantool memtx allocation overview - память memtx: аллокация кортежей и индексов, краткая инструкция по мониторингу;
• Tarantool MVCC overview - статистика менеджера транзакций движка memtx;
• Tarantool space statistics - статистика по спейсам: количество кортежей, объем данных, операции чтения/записи, фрагментация;
• Tarantool vinyl statistics - память и дисковое пространство vinyl, планировщик и статистика транзакций;
• Tarantool CPU statistics - загрузка процессора по экземплярам (пользовательское и системное время);
• Tarantool runtime overview - Метрики рантайма: память Lua и транзакций, статистика файберов, цикл событий;
• Tarantool LuaJit statistics - детальная информация о LuaJIT: аллокация объектов, сборка мусора, потребление памяти, трассы JIT;
• Tarantool operations statistics - операции над спейсами (select, update и др.) и другие вызовы (eval, call, auth, ошибки, SQL);
• Tarantool expirationd overview - cтатистика задач expirationd: количество кортежей, число перезапусков, время работы.