MessagePlugin#

Плагин предназначен для:

1. Формирования логов трассировки при следующих событиях:

   • поступление сообщения на брокер/уход сообщения с брокера (при вычитке сообщения consumer);

   • перемещение сообщения в expire, dead-letter очереди;

   • перемещение сообщения с одного брокера на другой в кластерном соединении при его дистрибуции;

   • перемещение или удаление сообщения вручную (через веб-клиент SMBX);

   • перемещение сообщения с одного брокера на другой в кластерном соединении при live-only scaleDown.

2. Формирования заголовка в сообщении о времени, когда сообщение покидает брокер(при его вычитке клиентом consumer).

Для вывода логов необходима настройка logger - ru.sbt.ss.artemis.plugin.message.MessagePlugin.

Важно: При использовании плагина на протоколе AMQP, настройки TIMESTAMP_HEADERS_MODE в конфигурационном файле messagePlugin.conf в логи выбрасываются исключения типа WARNING, например:

2024-06-04 14:21:16.250 [Thread-3 (ActiveMQ-server-org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl$6@4012d5bc)] WARN  org.apache.activemq.artemis.core.server  - AMQ222215: Global Address Size has negative and inconsistent value as -64
2024-06-04 14:21:16.250 [Thread-3 (ActiveMQ-server-org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl$6@4012d5bc)] WARN  org.apache.activemq.artemis.core.server  - AMQ222214: Destination test-bridge has an inconsistent and negative address size=-64.
2024-06-04 14:21:18.263 [Thread-2 (ActiveMQ-server-org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl$6@4012d5bc)] WARN  org.apache.activemq.artemis.core.server  - AMQ222215: Global Address Size has negative and inconsistent value as -64

Возможное решение проблемы: отключить настройку TIMESTAMP_HEADERS_MODE и перезапустить брокер, чтобы настройка отключилась.

Подключение плагина#

Для подключения плагина к SMBX необходимо:

• сформировать конфигурационный файл messagePlugin.conf, в котором должны быть указаны настройки плагина;

• в broker_instance_dir/etc/broker.xml необходимо добавить настройки плагина в разделе broker-plugins;

• в classpath SMBX (например, broker_instance_dir/lib) положить logger-plugin-*.jar-файл.

Настройки плагина#

Для корректной работы плагина, необходимо на брокере в разделе address-settings установить параметру enable-ingress-timestamp значение true.

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

<address-settings>
    ...
    <address-setting match="#">
        ...
        <enable-ingress-timestamp>true</enable-ingress-timestamp>
        ...
    </address-setting>
    ...
</address-settings>

Настройки плагина, связанные с формированием логов трассировки#

* - идентификатор сообщения может быть указан в заголовке сообщения при его отправке на брокер. Если ни в одном из заголовков идентификатор не был найден,
то идентификатор ищется в поле userId внутреннего формата сообщения SMBX. В случае, если и в userId отсутствует идентификатор, то идентификатор сообщения:UNDEFINED. ** - в лог будут выведены только заголовки, которые присутствуют в сообщении.
*** - по умолчанию в лог трассировки не выводятся сообщения для следующих адресов: activemq.management, activemq.notifications и адресов с префиксом: $.artemis.internal., notif.. Если настройка не указана или значением настройки является пустая строка, то фильтрация по BLACK_LIST_ADDRESSES отсутствует. Настройка поддерживает wildcard-синтаксис для адресов.
**** - если настройка не указана или значением настройки является пустая строка, то фильтрация по WHITE_LIST_ADDRESSES отсутствует. Настройка поддерживает wildcard-синтаксис для адресов.
***** - если настройка не указана, то используется имя заголовка durable.

Настройки плагина по формированию заголовка сообщения, в котором указывается время ухода сообщения с брокера#

* - формирование заголовка происходит на этапе вычитки сообщения из очереди клиентом-consumer

Пример конфигурационного файла плагина messagePlugin.conf#

  messagePlugin: {
        MESSAGE_ID_HEADER_NAMES: "headerName_1, headerName_2"     // необязательная настройка
        HEADER_NAMES_FOR_LOG: "headerName"                        // необязательная настройка
        BLACK_LIST_ADDRESSES: "address_1, address_2, address 3"   // необязательная настройка
        WHITE_LIST_ADDRESSES: "address_21, address_22, address_33"// необязательная настройка
        DURABLE_HEADER_NAME: "_HDR_DURABLE_"                      // необязательная настройка
       
        TIMESTAMP_HEADERS_MODE: "ON_FOR_ALL"                      // необязательная настройка 
        FROM_SERVER_TIMESTAMP_HEADER_NAME: "TIMESTAMP_FROM_SERVER"// Обязательная настройка для режимов ON_FOR_ALL/ON_IF_EXIST

        // Параметры предназначены для настройки по формированию latency-метрики(функционал находится на стадии разработки)
        // Необходимо либо BROKER_LATENCY_METRIC_MODE=OFF, либо эти настройки вообще не указывать
        BROKER_LATENCY_METRIC_MODE: "OFF"                         // необязательная настройка
        BROKER_LATENCY_METRIC_COORDINATES: "address.example.#"    // необязательная настройка
  }

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

Пример настройки плагина в broker.xml#

<broker-plugins>
    
    <broker-plugin class-name="ru.sbt.ss.artemis.plugin.message.MessagePlugin">
        <property key="CONFIG_PATH" value="path_to_messagePlugin_configuration" />  <!-- ОБЯЗАТЕЛЬНЫЙ путь до конфигурационного файла /-->
        <property key="RELOAD_CHECK_PERIOD_MS" value="period_in_ms" />              <!-- необязательная периодичность проверки планировщиком обнаружения изменений в файле конфигурации в миллисекундах /-->
    </broker-plugin>
</broker-plugins>

Формат логов при трассировке сообщений#

Примеры логов трассировки#

• сообщение поступает на брокер

{
  "text": "MESSAGE_ARRIVED_AT_THE_SERVER",
  "brokerHost": "broker_host",
  "brokerId": "347be2fe-b3fb-11ec-82dc-acde48001122",
  "msgIdHeaderNames": [
    "MESSAGE_ID"
  ],
  "headers": [
    "HEADER_1=Header_value_1",
    "HEADER_2=Header_value_2"
  ],
  "address": "test_address",
  "user": "myUser",
  "msgId": "123450",
  "artemisMsgId": 12,
  "msgBodyBytesSize": 17,
  "durable": "false",
  "timestamp": 1649065204737
}

• сообщение покидает брокер при его вычитке consumer

{
  "text": "MESSAGE_LEFT_THE_SERVER",
  "brokerHost": "broker_host_test_localhost",
  "brokerId": "06b2fd19-bb3f-11ec-b6a9-7a73e0a59c62",
  "msgIdHeaderNames": [
    "MESSAGE_ID_TEST"
  ],
  "headers": [
    "HEADER_TEST_2=null",
    "HEADER_TEST_1=Header_value_test"
  ],
  "address": "queues.test_1.Q1",
  "user": "user_consumer",
  "msgId": "125656567",
  "artemisMsgId": 16,
  "msgBodyBytesSize": 435,
  "durable": "true",
  "timestamp": 1649863991986
}

• сообщение покидает брокер при дистрибуции сообщения в кластере

{
  "text": "MESSAGE_LEFT_THE_SERVER_DURING_DISTRIBUTION",
  "brokerHost": "broker_host",
  "brokerId": "347be2fe-b3fb-11ec-82dc-acde48001122",
  "msgIdHeaderNames": [
    "RqUID"
  ],
  "headers": [
    "HEADER_1=Header_value_1",
    "HEADER_2=null"
  ],
  "address": "test_address", 
  "distributionAddress": "localhost/127.0.0.1:9092",
  "msgId": "32435",
  "artemisMsgId": 243,
  "msgBodyBytesSize": 17,
  "durable": "true",
  "timestamp": 1649065204737
}

• сообщение поступает в expiry-очередь на брокере

{
  "text": "MESSAGE_ARRIVED_AT_THE_EXPIRED_QUEUE",
  "brokerHost": "broker_host",
  "brokerId": "347be2fe-b3fb-11ec-82dc-acde48001122",
  "msgIdHeaderNames": [
    "MESSAGE_ID_TEST"
  ],
  "headers": [],
  "address": "ExpiryQueue",
  "msgId": "5465676",
  "artemisMsgId": 12,
  "msgBodyBytesSize": 34,
  "durable": "false",
  "timestamp": 1649065204859
}

• сообщение поступает в dead-letter-очередь на брокере

{
  "text": "MESSAGE_ARRIVED_AT_THE_DEAD_LETTER_QUEUE",
  "brokerHost": "broker_host",
  "brokerId": "36499e25-b3fb-11ec-82dc-acde48001122",
  "msgIdHeaderNames": [
    "MESSAGE_ID_TEST"
  ],
  "headers": [
    "HEADER_TEST_1=Header_value_test"
  ],
  "address": "DLQ",
  "msgId": "ID:123450",
  "artemisMsgId": 12,
  "msgBodyBytesSize": 1024,
  "durable": "true",
  "timestamp": 1649065207104
}

• удаление сообщения при истечении времени жизни (expiry-очередь не настроена)

{
  "text": "MESSAGE_ARRIVED_AT_THE_EXPIRED_QUEUE",
  "brokerHost": "broker_host",
  "brokerId": "3794e2dc-b3fb-11ec-82dc-acde48001122",
  "msgIdHeaderNames": [
    "MESSAGE_ID_TEST"
  ],
  "headers": [
    "HEADER_TEST_1=Header_value_test"
  ],
  "address": "UNDEFINED_EXPIRY_ADDRESS",
  "msgId": "ID:123450",
  "artemisMsgId": 9,
  "msgBodyBytesSize": 1876,
  "durable": "false",
  "timestamp": 1649065209282
}

• удаление или перемещение сообщения через веб-интерфейс SMBX

{
   "text":"MESSAGE_DELETED_OR_MOVED",
   "brokerHost":"broker_host_test_localhost",
   "brokerId":"d816950d-7f90-11ed-8782-3a07a8ec4565",
   "msgIdHeaderNames":[
      "MESSAGE_ID_TEST"
   ],
   "headers":[
      "HEADER_TEST_1=Header_value_test"
   ],
   "address":"queues.test_1.Q1",
   "msgId":"ID:123450",
   "artemisMsgId":12,
   "msgBodyBytesSize":17,
   "durable":"false",
   "timestamp":1671449560622
}

• перемещение сообщения при scale-down

{
   "text":"MESSAGE_SCALED_DOWN",
   "brokerHost":"broker_host_test_localhost",
   "brokerId":"d816950d-7f90-11ed-8782-3a07a8ec4565",
   "msgIdHeaderNames":[
      "MESSAGE_ID_TEST"
   ],
   "headers":[
      "HEADER_TEST_1=Header_value_test"
   ],
   "address":"queues.test_1.Q1",
   "distributionAddress": "localhost/127.0.0.1:9092",
   "msgId":"ID:123450",
   "artemisMsgId":12,
   "msgBodyBytesSize":17,
   "durable":"false",
   "timestamp":1671449560622
}