Queues: очереди Kafka и RabbitMQ

«Queues» — вид для брокеров сообщений Kafka и RabbitMQ. Он показывает, кто перестал читать сообщения и почему, разбирает очереди мёртвых писем и даёт прочитать сами сообщения, а для RabbitMQ ещё и хранит историю очередей. В очередь мёртвых писем (DLQ) попадают сообщения, которые не удалось обработать. К брокеру приложение подключается с вашего компьютера.

Агенту для работы с брокерами нужен модуль «Queues». Модуль — набор инструментов, инструкций и навыков агента для одной области, подробнее — на странице Модули. Вам самим модуль не нужен: вид работает и без него. Агент с модулем может читать то же самое, но только на том уровне доступа, который вы ему дадите: по умолчанию брокеры для агента закрыты.

#Как открыть

• В полосе заголовка окна выберите «Другие инструменты» → «Queues». Кнопку «Queues» можно вынести в саму полосу: это делается в «Настройки → Расположение панелей».
• Или наберите в поле чата слэш-команду /queues (её синоним — /kafka).

Слева в виде — список подключений, справа — вкладки выбранного подключения. Красная отметка на кнопке «Queues» значит, что есть алерты, а отметка ожидания — что во вкладке «Запросы» запрос агента ждёт вашего решения. Отличить их можно по подсказке: наведите курсор на кнопку, и она покажет число алертов («Queues: алертов — N») или запросов агента («Queues: агент просит — N»). Чтобы вернуться на шаг назад внутри вида, нажмите «←» в строке подключения, ⌘ [ (на Windows и Linux — Ctrl [) или Alt ←. Вид можно открепить в отдельное окно. Кнопка «Агент» показывает разговор с агентом рядом с видом.

#Как подключить брокер

1. Нажмите «Добавить подключение» (или «+» над списком подключений).
2. Заполните «Имя» и выберите «Брокер».
3. Kafka: в поле «Адреса» укажите host:port, несколько адресов — через запятую. В «Вход» выберите «без входа», «SASL PLAIN», «SASL SCRAM-SHA-256» или «SASL SCRAM-SHA-512»; для вариантов с SASL заполните ещё «Пользователь» и пароль. Нужен TLS — включите «TLS» и укажите «Файл CA (PEM)». Если сообщения закодированы по схемам, добавьте «Schema registry»: подойдёт любой registry, совместимый с Confluent.
4. RabbitMQ: в «Где брокер» выберите «по адресу» и укажите «Адрес management». Это адрес веб-интерфейса плагина management, обычно http(s)://host:15672; порт AMQP 5672 сюда не подходит. Или выберите «в Kubernetes» и заполните «Context», «Namespace», «Service» и «Порт» — приложение само откроет проброс порта к этому сервису. Затем заполните «Пользователь», пароль и «vhost». Пустой «vhost» значит все vhost, которые видит учётная запись.
5. В «Откуда пароль» выберите «ввести здесь (связка ключей)» или «из хранилища секретов». Во втором случае укажите ссылку: vault://путь#ключ, env://ИМЯ или k8s://namespace/secret/ключ. Значение по ссылке приложение читает при подключении и нигде не сохраняет.
6. Нажмите «Проверить». Если всё в порядке, появится «Отвечает: …». Затем нажмите «Сохранить».

В блоке «Дополнительно» есть ещё настройки:

• «Окружение» — prod, stage и т. п. Политика организации может быть для окружения строже: запретить в нём действия над брокером и задать для агента более строгое маскирование;
• «Следить, пока приложение открыто» — по умолчанию выключено. Если включить, приложение раз в минуту ищет остановленных потребителей, отставание и рост DLQ, даже когда вид закрыт;
• «Проект для инцидентов и схем» — если поле пустое, приложение берёт проект, открытый в окне;
• для Kafka — «Ещё топики DLQ», для RabbitMQ — сбор истории, адрес Prometheus и пороги «Когда отмечать проблему».

#Kafka

У подключения Kafka такие вкладки: «Topics», «Consumer groups», «DLQ», «Сообщения», «Запросы», «Алерты», «Журнал».

• «Topics» — партиции, число сообщений, размер, входящий поток, отставание. Нажмите на топик, чтобы открыть его сообщения.
• «Consumer groups» — карточки групп, проблемные стоят сверху. На каждой карточке короткий вердикт: «Читает», «Отстаёт», «Остановился», «Партиция стоит», «Никто не читает», «Коммитов ещё нет». Если раскрыть карточку, видны партиции и участники группы.
• «DLQ» — топики с окончаниями .dlq, -dlq, _dlq, .DLT, -dlt, dead-letter, deadletter и те, что вы указали сами. Блок «Почему не обработались» собирает причины из заголовков сообщений.
• «Сообщения» — до 500 сообщений: последние, с начала, с офсета или с момента времени. Фильтровать можно по ключу, заголовку и тексту. Чтение ничего не забирает из топика и не сдвигает офсеты потребителей.

Тело сообщения раскодируется по схеме из registry (Avro, Protobuf, JSON Schema). Без registry приложение ищет схему в проекте: в файле .koder/queues.json вида {"topics": {"orders": {"value": "schemas/order.avsc"}}} или по имени <топик>.avsc, <топик>-value.avsc, <топик>.proto в папках schemas/, schema/, avro/, proto/. Схемы нет — тело показывается как JSON, текст или байты.

#RabbitMQ

У подключения RabbitMQ такие вкладки: «Queues», «История», «Consumers», «Топология», «Сообщения», «Трассировка», «Запросы», «Алерты», «Журнал».

• «Queues» — очереди с числами ready и unacked, скоростями и вердиктом: «Читается», «Никто не читает», «Консьюмер завис», «Цикл redelivery», «Растёт».
• «История» — графики сообщений, скоростей и консьюмеров за час, сутки, неделю или свой период и список «Что произошло». Сам RabbitMQ историю не хранит, поэтому её собирает приложение, пока оно открыто, и хранит 7 дней. Если указан Prometheus, приложение берёт историю из него.
• «Consumers» — кто какую очередь читает, prefetch, режим подтверждения и какие консьюмеры зависли.
• «Топология» — граф exchange → привязка → очередь → консьюмер → сервис.
• «Сообщения» — «Прочитать (get)», до 100 сообщений. Такое чтение оставляет след: RabbitMQ возвращает сообщения в очередь с флагом redelivered, и порядок может измениться. Приложение предупреждает об этом перед каждым чтением. Unacked-сообщения прочитать нельзя.
• «Трассировка» — вы задаёте время и объём, а трассировка ловит сообщения очереди, exchange или routing key и потом выключается сама. Пойманные сообщения хранятся только на вашем компьютере.

#Действия над брокером

Для Kafka есть действия «Повторная отправка из DLQ», «Сдвиг офсетов группы», «Очистка топика» и «Тестовое сообщение». Для RabbitMQ — повторная отправка из DLQ, очистка очереди, тестовое сообщение и закрытие соединения клиента.

По умолчанию ни одно действие не разрешено, и кнопок для них в виде нет. Чтобы разрешить действия, откройте «Настройки → Безопасность агента → Подключения», раскройте «Ещё: брокеры сообщений, браузер, Jira, скрытие данных…» и в блоке «Брокеры сообщений» найдите нужное подключение. В строке «Разрешённые здесь действия» у него отдельный флажок на каждое действие этого брокера, Kafka или RabbitMQ: отметьте те, что хотите разрешить. Действие, которое запрещает политика организации, отметить нельзя.

Каждое действие проходит три шага: форма → «Показать план» (куда, сколько сообщений и первые из них) → «Выполнить». Перед выполнением приложение заново читает брокер, и если он заметно изменился, вы увидите новый план. Чтобы очистить топик или очередь, закрыть соединение или сильно сдвинуть офсеты назад, нужно ввести имя объекта. К тестовым сообщениям приложение добавляет заголовок x-koder-test: 1. Каждое выполненное действие попадает во вкладку «Журнал».

#Алерты

Алерты бывают такие: «Потребитель остановился», «Большое отставание», «Растёт DLQ», «Цикл redelivery», «Очередь растёт». Приложение проверяет их, пока вид открыт, а у подключений со «Следить» — ещё и раз в минуту в фоне, пока приложение запущено. «Queues» — не система мониторинга: закрыли приложение, и алертов нет. Если окно не в фокусе, придёт системное уведомление. Если подключение открыто агенту хотя бы на уровне «Метаданные», алерт попадает ещё и в инциденты проекта.

#Агент и брокеры

#Как включить

1. Включите модуль «Queues» в «Настройки → Модули» или кнопкой «Включить для агента». Кнопка стоит на плашке под шапкой вида «Queues»: плашка видна, пока модуль выключен и есть хотя бы одно подключение. Модуль начинает работать, как только появится хотя бы одно подключение. Шаг можно пропустить: когда на шаге 2 вы выберете подключению уровень доступа выше «Закрыто», приложение само включит модуль, если организация его не запретила.
2. Откройте подключение агенту: «Настройки → Безопасность агента → Подключения» → «Ещё: …» → «Брокеры сообщений».
3. Изменения действуют в новых чатах.

У каждого подключения свой уровень доступа:

Там же доступ можно сузить списками vhost, очередей, топиков и exchange: то, что вы закрыли, агент не увидит даже по имени. Подробнее об уровнях — на странице «Доступ агента к базам, кластерам и очередям».

#Что может агент

• Читать списки топиков и очередей с вердиктами, отставание групп, DLQ, историю и топологию RabbitMQ.
• Читать сообщения Kafka. На уровне «Сообщения» подтверждение не нужно: чтение брокер не меняет.
• Читать сообщения RabbitMQ и включать трассировку — только после того, как вы нажмёте «Разрешить», и так на каждый вызов. Варианта «Больше не спрашивать» здесь нет, разрешить с телефона тоже нельзя.
• Менять брокер агент не может. На уровне «Сообщения и запросы» он оставляет запрос с планом во вкладке «Запросы», а вы нажимаете там «Посмотреть и решить…» или «Отклонить» с причиной. Запрос ждёт час и пропадает, если перезапустить приложение. Просить агент может только о действиях, которые разрешены на подключении.

Когда агент обращается к брокеру, вид «Queues» сам выходит в центр окна, но не чаще раза в 30 секунд и не тогда, когда вы печатаете.

#Маскирование

Всё, что получает агент, — тела, ключи, заголовки, причины из DLQ, тексты алертов — приложение маскирует, прежде чем отдать ему. «Стандартный набор» скрывает почту, телефоны, ФИО, паспорта, СНИЛС, ИНН, номера карт, адреса, токены и пароли. «Только структура» оставляет ключи и типы и прячет значения. Свои правила на JSONPath можно добавить в блоке «Маскирование для агента». Сами вы в виде видите сообщения без маскирования.

#Ограничения

• Брокер должен быть доступен с вашего компьютера: приложение подключается к нему оттуда.
• Если проект на сервере по SSH, приложение всё равно опрашивает брокер с этого компьютера, и инструменты агента работают. Схемы из репозитория и инциденты для такого проекта не используются.
• История RabbitMQ есть только с того момента, как включён сбор, и только за время, когда приложение было открыто, — если приложение не берёт её из Prometheus.
• Stream-очереди RabbitMQ прочитать нельзя. Число сообщений в compacted-топиках Kafka приблизительное.

#Какие данные куда уходят

Приложение само соединяется с брокером, registry, Prometheus, Vault и Kubernetes — с вашего компьютера и под вашими учётными данными. Подключения хранятся в ~/.koder/ade/queues.json, пароли — в системной связке ключей. То, что прочитал агент, уже в маскированном виде попадает в запрос к модели и идёт к ней через сервер DevMachine. Подробнее — на странице «Данные и приватность».

#Что дальше

• Доступ агента к базам, кластерам и очередям
• Инциденты
• Kubernetes
• Модули