monitoring.md

layout	default
title	Monitoring
nav_order	95

{% include support.md %}

Monitoring.

Enqueue provides a tool for monitoring message queues. With it, you can control how many messages were sent, how many processed successfully or failed. How many consumers are working, their up time, processed messages stats, memory usage and system load. The tool could be integrated with virtually any analytics and monitoring platform. There are several integration:

• Datadog StatsD
• InfluxDB and Grafana
• WAMP (Web Application Messaging Protocol) We are working on a JS\WAMP based real-time UI tool, for more information please contact us.

contact us if need a Grafana template such as on the picture.

• Installation
• Track sent messages
• Track consumed message
• Track consumer metrics
• Consumption extension
• Enqueue Client Extension
• InfluxDB Storage
• Datadog Storage
• WAMP (Web Socket Messaging Protocol) Storage
• Symfony App

Installation

composer req enqueue/monitoring:0.9.x-dev

Track sent messages

<?phpuseEnqueue\Monitoring\SentMessageStats;
useEnqueue\Monitoring\GenericStatsStorageFactory;
$statsStorage = (newGenericStatsStorageFactory())->create('influxdb://127.0.0.1:8086?db=foo');
$statsStorage->pushSentMessageStats(newSentMessageStats(
(int) (microtime(true) * 1000), // timestamp'queue_name', // queue'aMessageId',
'aCorrelationId',
[], // headers
[] // properties
));

or, if you work with Queue Interop transport here's how you can track a message sent

<?phpuseInterop\Queue\Context;
useEnqueue\Monitoring\SentMessageStats;
useEnqueue\Monitoring\GenericStatsStorageFactory;
/** @var Context $context */$queue = $context->createQueue('foo');
$message = $context->createMessage('body');
$context->createProducer()->send($queue, $message);
$statsStorage = (newGenericStatsStorageFactory())->create('influxdb://127.0.0.1:8086?db=foo');
$statsStorage->pushSentMessageStats(newSentMessageStats(
(int) (microtime(true) * 1000),
$queue->getQueueName(),
$message->getMessageId(),
$message->getCorrelationId(),
$message->getHeaders()[],
$message->getProperties()
));

Track consumed message

<?phpuseEnqueue\Monitoring\ConsumedMessageStats;
useEnqueue\Monitoring\GenericStatsStorageFactory;
$receivedAt = (int) (microtime(true) * 1000);
// heavy processing here.$statsStorage = (newGenericStatsStorageFactory())->create('influxdb://127.0.0.1:8086?db=foo');
$statsStorage->pushConsumedMessageStats(newConsumedMessageStats(
'consumerId',
(int) (microtime(true) * 1000), // now$receivedAt,
'aQueue',
'aMessageId',
'aCorrelationId',
[], // headers
[], // propertiesfalse, // redelivered or not
ConsumedMessageStats::STATUS_ACK
));

or, if you work with Queue Interop transport here's how you can track a message sent

<?phpuseInterop\Queue\Context;
useEnqueue\Monitoring\ConsumedMessageStats;
useEnqueue\Monitoring\GenericStatsStorageFactory;
/** @var Context $context */$queue = $context->createQueue('foo');
$consumer = $context->createConsumer($queue);
$consumerId = uniqid('consumer-id', true); // we suggest using UUID hereif ($message = $consumer->receiveNoWait()) {
$receivedAt = (int) (microtime(true) * 1000);
// heavy processing here.$consumer->acknowledge($message);
$statsStorage = (newGenericStatsStorageFactory())->create('influxdb://127.0.0.1:8086?db=foo');
$statsStorage->pushConsumedMessageStats(newConsumedMessageStats(
$consumerId,
(int) (microtime(true) * 1000), // now$receivedAt,
$queue->getQueueName(),
$message->getMessageId(),
$message->getCorrelationId(),
$message->getHeaders(),
$message->getProperties(),
$message->isRedelivered(),
ConsumedMessageStats::STATUS_ACK
));
}

Track consumer metrics

Consumers are long running processes. It vital to know how many of them are running right now, how they perform, how much memory do they use and so. This example shows how you can send such metrics. Call this code from time to time between processing messages.

<?phpuseEnqueue\Monitoring\ConsumerStats;
useEnqueue\Monitoring\GenericStatsStorageFactory;
$startedAt = (int) (microtime(true) * 1000);
$statsStorage = (newGenericStatsStorageFactory())->create('influxdb://127.0.0.1:8086?db=foo');
$statsStorage->pushConsumerStats(newConsumerStats(
'consumerId',
(int) (microtime(true) * 1000), // now$startedAt,
null, // finished attrue, // is started?false, // is finished?false, // is failed
['foo'], // consume from queues123, // received messages120, // acknowledged messages1, // rejected messages1, // requeued messagesmemory_get_usage(true),
sys_getloadavg()[0]
));

Consumption extension

There is an extension ConsumerMonitoringExtension for Enqueue QueueConsumer. It could collect consumed messages and consumer stats for you.

<?phpuseEnqueue\Consumption\QueueConsumer;
useEnqueue\Consumption\ChainExtension;
useEnqueue\Monitoring\ConsumerMonitoringExtension;
useEnqueue\Monitoring\GenericStatsStorageFactory;
useInterop\Queue\Context;
/** @var Context $context */$statsStorage = (newGenericStatsStorageFactory())->create('influxdb://127.0.0.1:8086?db=foo');
$queueConsumer = newQueueConsumer($context, newChainExtension([
newConsumerMonitoringExtension($statsStorage)
]));
// bind// consume

Enqueue Client Extension

There is an extension ClientMonitoringExtension for Enqueue Client too. It could collect sent messages stats for you.

InfluxDB Storage

Install additional packages:

composer req influxdb/influxdb-php:^1.14

<?phpuseEnqueue\Monitoring\GenericStatsStorageFactory;
$statsStorage = (newGenericStatsStorageFactory())->create('influxdb://127.0.0.1:8086?db=foo');

There are available options:

* 'host' => '127.0.0.1',
* 'port' => '8086',
* 'user' => '',
* 'password' => '',
* 'db' => 'enqueue',
* 'measurementSentMessages' => 'sent-messages',
* 'measurementConsumedMessages' => 'consumed-messages',
* 'measurementConsumers' => 'consumers',
* 'client' => null,
* 'retentionPolicy' => null,

You can pass InfluxDB\Client instance in client option. Otherwise, it will be created on first use according to other options.

If your InfluxDB\Client uses driver that implements InfluxDB\Driver\QueryDriverInterface, then database will be automatically created for you if it doesn't exist. Default InfluxDB\Client will also do that.

Datadog storage

Install additional packages:

composer req datadog/php-datadogstatsd:^1.3

<?phpuseEnqueue\Monitoring\GenericStatsStorageFactory;
$statsStorage = (newGenericStatsStorageFactory())->create('datadog://127.0.0.1:8125');

For best experience please adjust units and types in metric summary.

Example dashboard:

There are available options (and all available metrics):

* 'host' => '127.0.0.1',
* 'port' => '8125',
* 'batched' => true, // performance boost
* 'global_tags' => '', // should contain keys and values
* 'metric.messages.sent' => 'enqueue.messages.sent',
* 'metric.messages.consumed' => 'enqueue.messages.consumed',
* 'metric.messages.redelivered' => 'enqueue.messages.redelivered',
* 'metric.messages.failed' => 'enqueue.messages.failed',
* 'metric.consumers.started' => 'enqueue.consumers.started',
* 'metric.consumers.finished' => 'enqueue.consumers.finished',
* 'metric.consumers.failed' => 'enqueue.consumers.failed',
* 'metric.consumers.received' => 'enqueue.consumers.received',
* 'metric.consumers.acknowledged' => 'enqueue.consumers.acknowledged',
* 'metric.consumers.rejected' => 'enqueue.consumers.rejected',
* 'metric.consumers.requeued' => 'enqueue.consumers.requeued',
* 'metric.consumers.memoryUsage' => 'enqueue.consumers.memoryUsage',

WAMP (Web Socket Messaging Protocol) Storage

Install additional packages:

composer req thruway/pawl-transport:^0.5.0 thruway/client:^0.5.0

<?phpuseEnqueue\Monitoring\GenericStatsStorageFactory;
$statsStorage = (newGenericStatsStorageFactory())->create('wamp://127.0.0.1:9090?topic=stats');

There are available options:

* 'host' => '127.0.0.1',
* 'port' => '9090',
* 'topic' => 'stats',
* 'max_retries' => 15,
* 'initial_retry_delay' => 1.5,
* 'max_retry_delay' => 300,
* 'retry_delay_growth' => 1.5,

Symfony App

You have to register some services in order to incorporate monitoring facilities into your Symfony application.

# config/packages/enqueue.yamlenqueue:
default:
transport: 'amqp://guest:guest@bar:5672/%2f'monitoring: 'influxdb://127.0.0.1:8086?db=foo'another:
transport: 'amqp://guest:guest@foo:5672/%2f'monitoring: 'wamp://127.0.0.1:9090?topic=stats'client: ~datadog:
transport: 'amqp://guest:guest@foo:5672/%2f'monitoring: 'datadog://127.0.0.1:8125?batched=false'client: ~

back to index