<?php

namespace Enqueue\Monitoring;

use Enqueue\Client\Config;

use Enqueue\Dsn\Dsn;

use InfluxDB\Client;

use InfluxDB\Database;

use InfluxDB\Driver\QueryDriverInterface;

use InfluxDB\Exception as InfluxDBException;

use InfluxDB\Point;

class InfluxDbStorage implements StatsStorage

{

/**

* @var array

*/

private $config;

/**

* @var Client

*/

private $client;

/**

* @var Database

*/

private $database;

/**

* The config could be an array, string DSN, null or instance of InfluxDB\Client. In case of null it will attempt to connect to localhost.

*

* $config = [

* 'dsn' => 'influxdb://127.0.0.1:8086',

* 'host' => '127.0.0.1',

* 'port' => '8086',

* 'user' => '',

* 'password' => '',

* 'db' => 'enqueue',

* 'measurementSentMessages' => 'sent-messages',

* 'measurementConsumedMessages' => 'consumed-messages',

* 'measurementConsumers' => 'consumers',

* 'client' => null, # Client instance. Null by default.

* 'retentionPolicy' => null,

* ]

*

* or

*

* influxdb://127.0.0.1:8086?user=Jon&password=secret

*

* @param array|string|null $config

*/

public function __construct($config = 'influxdb:')

{

if (false == class_exists(Client::class)) {

throw new \LogicException('Seems client library is not installed. Please install "influxdb/influxdb-php"');

}

if (empty($config)) {

$config = [];

} elseif (is_string($config)) {

$config = self::parseDsn($config);

} elseif (is_array($config)) {

$config = empty($config['dsn']) ? $config : self::parseDsn($config['dsn']);

} elseif ($config instanceof Client) {

// Passing Client instead of array config is deprecated because it prevents setting any configuration values

// and causes library to use defaults.

@trigger_error(

sprintf('Passing %s as %s argument is deprecated. Pass it as "client" array property or use createWithClient instead',

Client::class,

__METHOD__

), \E_USER_DEPRECATED);

$this->client = $config;

$config = [];

} else {

throw new \LogicException('The config must be either an array of options, a DSN string or null');

}

$config = array_replace([

'host' => '127.0.0.1',

'port' => '8086',

'user' => '',

'password' => '',

'db' => 'enqueue',

'measurementSentMessages' => 'sent-messages',

'measurementConsumedMessages' => 'consumed-messages',

'measurementConsumers' => 'consumers',

'client' => null,

'retentionPolicy' => null,

], $config);

if (null !== $config['client']) {

if (!$config['client'] instanceof Client) {

throw new \InvalidArgumentException(sprintf('%s configuration property is expected to be an instance of %s class. %s was passed instead.', 'client', Client::class, gettype($config['client'])));

}

$this->client = $config['client'];

}

$this->config = $config;

}

/**

* @param string $config

*/

public static function createWithClient(Client $client, $config = 'influxdb:'): self

{

if (is_string($config)) {

$config = self::parseDsn($config);

}

$config['client'] = $client;

return new self($config);

}

public function pushConsumerStats(ConsumerStats $stats): void

{

$points = [];

foreach ($stats->getQueues() as $queue) {

$tags = [

'queue' => $queue,

'consumerId' => $stats->getConsumerId(),

];

$values = [

'startedAtMs' => $stats->getStartedAtMs(),

'started' => $stats->isStarted(),

'finished' => $stats->isFinished(),

'failed' => $stats->isFailed(),

'received' => $stats->getReceived(),

'acknowledged' => $stats->getAcknowledged(),

'rejected' => $stats->getRejected(),

'requeued' => $stats->getRequeued(),

'memoryUsage' => $stats->getMemoryUsage(),

'systemLoad' => $stats->getSystemLoad(),

];

if ($stats->getFinishedAtMs()) {

$values['finishedAtMs'] = $stats->getFinishedAtMs();

}

$points[] = new Point($this->config['measurementConsumers'], null, $tags, $values, $stats->getTimestampMs());

}

$this->doWrite($points);

}

public function pushConsumedMessageStats(ConsumedMessageStats $stats): void

{

$tags = [

'queue' => $stats->getQueue(),

'status' => $stats->getStatus(),

];

$properties = $stats->getProperties();

if (false === empty($properties[Config::TOPIC])) {

$tags['topic'] = $properties[Config::TOPIC];

}

if (false === empty($properties[Config::COMMAND])) {

$tags['command'] = $properties[Config::COMMAND];

}

$values = [

'receivedAt' => $stats->getReceivedAtMs(),

'processedAt' => $stats->getTimestampMs(),

'redelivered' => $stats->isRedelivered(),

];

if (ConsumedMessageStats::STATUS_FAILED === $stats->getStatus()) {

$values['failed'] = 1;

}

$runtime = $stats->getTimestampMs() - $stats->getReceivedAtMs();

$points = [

new Point($this->config['measurementConsumedMessages'], $runtime, $tags, $values, $stats->getTimestampMs()),

];

$this->doWrite($points);

}

public function pushSentMessageStats(SentMessageStats $stats): void

{

$tags = [

'destination' => $stats->getDestination(),

];

$properties = $stats->getProperties();

if (false === empty($properties[Config::TOPIC])) {

$tags['topic'] = $properties[Config::TOPIC];

}

if (false === empty($properties[Config::COMMAND])) {

$tags['command'] = $properties[Config::COMMAND];

}

$points = [

new Point($this->config['measurementSentMessages'], 1, $tags, [], $stats->getTimestampMs()),

];

$this->doWrite($points);

}

private function doWrite(array $points): void

{

if (null === $this->client) {

$this->client = new Client(

$this->config['host'],

$this->config['port'],

$this->config['user'],

$this->config['password']

);

}

if ($this->client->getDriver() instanceof QueryDriverInterface) {

if (null === $this->database) {

$this->database = $this->client->selectDB($this->config['db']);

$this->database->create();

}

$this->database->writePoints($points, Database::PRECISION_MILLISECONDS, $this->config['retentionPolicy']);

} else {

// Code below mirrors what `writePoints` method of Database does.

try {

$parameters = [

'url' => sprintf('write?db=%s&precision=%s', $this->config['db'], Database::PRECISION_MILLISECONDS),

'database' => $this->config['db'],

'method' => 'post',

];

if (null !== $this->config['retentionPolicy']) {

$parameters['url'] .= sprintf('&rp=%s', $this->config['retentionPolicy']);

}

$this->client->write($parameters, $points);

} catch (\Exception $e) {

throw new InfluxDBException($e->getMessage(), $e->getCode());

}

}

}

private static function parseDsn(string $dsn): array

{

$dsn = Dsn::parseFirst($dsn);

if (false === in_array($dsn->getSchemeProtocol(), ['influxdb'], true)) {

throw new \LogicException(sprintf('The given scheme protocol "%s" is not supported. It must be "influxdb"', $dsn->getSchemeProtocol()));

}

return array_filter(array_replace($dsn->getQuery(), [

'host' => $dsn->getHost(),

'port' => $dsn->getPort(),

'user' => $dsn->getUser(),

'password' => $dsn->getPassword(),

'db' => $dsn->getString('db'),

'measurementSentMessages' => $dsn->getString('measurementSentMessages'),

'measurementConsumedMessages' => $dsn->getString('measurementConsumedMessages'),

'measurementConsumers' => $dsn->getString('measurementConsumers'),

'retentionPolicy' => $dsn->getString('retentionPolicy'),

]), function ($value) { return null !== $value; });

}

}