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

Настояшее Руководство содержит описание и сценарии эксплуатации и администрирования 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 двухэтапная проверка не работает и система выдаст предупреждение об этом.

{note:warning} Режим без двухэтапной проверки строго не рекомендуется для синхронных очередей. {/note}

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

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

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

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

Пример 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. По умолчанию – basic.
• poll_max_batch – максимальное количество сообщений, которое одно ядро вернет за один запрос подписки. По умолчанию – 100.
• poll_min_batch – минимальное количество сообщений, которое одно ядро вернет за один запрос подписки. По умолчанию – 10.
• 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) доступен режим статического шардирования — это режим ручной настройки карты распределения бакетов по наборам реплик в кластере. Настройка задается с помощью глобальной конфигурации кластера, после чего необходимо выполнить начальную загрузку. В результате в наборах реплик кластера создаются бакеты в соответствии с заданной картой.

{note:warning} После загрузки кластера изменять конфигурацию нельзя. {/note}

Конфигурация статического шардирования осуществляется посредством настройки роли 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:

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

где:

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

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

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

 
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:  file: log.jsonl  format: json  level: info

Основные параметры обеспечивают настройку адресов и портов подключения модуля 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

где:

• 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.

Секция producer модуля API обеспечивает настройку параметров публикации.

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

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

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

Секция consumer модуля API обеспечивает настройку параметров подписки на очередь сообщений.

 
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:  file: log.jsonl  format: json  level: info

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

• file - имя файла записи журнала (может содержать абсолютный или относительный путь, по умолчанию /dev/stderr, либо в файл server.json, при запуске модуля API в фоновом режиме с опцией -d);
• format - формат вывода сообщений, принимаемые значения:
  • text (по умолчанию);
  • json;
• level - уровень журналирования, принимаемые значения:
  • debug;
  • info (по умолчанию);
  • warn;
  • error;
  • dpanic(паника в режиме разработки);
  • panic(паника);
  • fatal.

Для резервного копирования данных 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: количество кортежей, число перезапусков, время работы.