monitoring

Author: ASKozienko

InfluxDbStorage.php

@@ -21,28 +21,70 @@ class InfluxDbStorage implements StatsStorage
     /**
      * @var string
      */
-    private $measurMessages;
+    private $measurementMessages;
 
     /**
      * @var string
      */
-    private $measurConsumers;
+    private $measurementConsumers;
 
     /**
      * @var Database
      */
     private $database;
 
     /**
-     * @param Client $client
-     * @param string $dbName
+     * The config could be an array, string DSN or null. In case of null it will attempt to connect to localhost.
+     *
+     * $config = [
+     *   'dsn'                 => 'wamp://127.0.0.1:9090',
+     *   '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,
+     * ]
+     *
+     * or
+     *
+     * wamp://127.0.0.1:9090?max_retries=10
+     *
+     * @param array|string|null $config
      */
-    public function __construct(Client $client, string $dbName)
+    public function __construct($config = 'influxdb:')
     {
         $this->client = $client;
         $this->dbName = $dbName;
-        $this->measurMessages = 'messages';
-        $this->measurConsumers = 'consumers';
+        $this->measurementMessages = 'messages';
+        $this->measurementConsumers = 'consumers';
+
+        if (false == class_exists(Client::class)) {
+            throw new \LogicException('Seems client library is not installed. Please install "influxdb/influxdb-php"');
+        }
+
+        if (empty($config)) {
+            $config = $this->parseDsn('influxdb:');
+        } elseif (is_string($config)) {
+            $config = $this->parseDsn($config);
+        } elseif (is_array($config)) {
+            $config = empty($config['dsn']) ? $config : $this->parseDsn($config['dsn']);
+        } 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' => '9090',
+            'topic' => 'stats',
+            'max_retries' => 15,
+            'initial_retry_delay' => 1.5,
+            'max_retry_delay' => 300,
+            'retry_delay_growth' => 1.5,
+        ], $config);
+
+        $this->config = $config;
     }
 
     public function pushConsumerStats(ConsumerStats $event): void
@@ -72,7 +114,7 @@ public function pushConsumerStats(ConsumerStats $event): void
                 $values['finishedAtMs'] = $event->getFinishedAtMs();
             }
 
-            $points[] = new Point($this->measurConsumers, null, $tags, $values, $event->getTimestampMs());
+            $points[] = new Point($this->measurementConsumers, null, $tags, $values, $event->getTimestampMs());
         }
 
         $this->getDb()->writePoints($points, Database::PRECISION_MILLISECONDS);
@@ -97,7 +139,7 @@ public function pushMessageStats(MessageStats $event): void
         $runtime = $event->getTimestampMs() - $event->getReceivedAtMs();
 
         $points = [
-            new Point($this->measurMessages, $runtime, $tags, $values, $event->getTimestampMs()),
+            new Point($this->measurementMessages, $runtime, $tags, $values, $event->getTimestampMs()),
         ];
 
         $this->getDb()->writePoints($points, Database::PRECISION_MILLISECONDS);
@@ -112,4 +154,26 @@ private function getDb(): Database
 
         return $this->database;
     }
+
+    private function parseDsn(string $dsn): array
+    {
+        $dsn = new Dsn($dsn);
+
+        if (false === in_array($dsn->getSchemeProtocol(), ['wamp', 'ws'], true)) {
+            throw new \LogicException(sprintf(
+                'The given scheme protocol "%s" is not supported. It must be "wamp"',
+                $dsn->getSchemeProtocol()
+            ));
+        }
+
+        return array_filter(array_replace($dsn->getQuery(), [
+            'host' => $dsn->getHost(),
+            'port' => $dsn->getPort(),
+            'topic' => $dsn->getQueryParameter('topic'),
+            'max_retries' => $dsn->getInt('max_retries'),
+            'initial_retry_delay' => $dsn->getFloat('initial_retry_delay'),
+            'max_retry_delay' => $dsn->getInt('max_retry_delay'),
+            'retry_delay_growth' => $dsn->getFloat('retry_delay_growth'),
+        ]), function ($value) { return null !== $value; });
+    }
 }

WampStorage.php

@@ -58,6 +58,10 @@ class WampStorage implements StatsStorage
      */
     public function __construct($config = 'wamp:')
     {
+        if (false == class_exists(Client::class) || false == class_exists(PawlTransportProvider::class)) {
+            throw new \LogicException('Seems client libraries are not installed. Please install "thruway/client" and "thruway/pawl-transport"');
+        }
+
         if (empty($config)) {
             $config = $this->parseDsn('wamp:');
         } elseif (is_string($config)) {

composer.json

@@ -15,6 +15,11 @@
         "phpunit/phpunit": "~5.4.0",
         "enqueue/test": "0.9.x-dev"
     },
+    "suggest": {
+        "thruway/client": "Client for Thruway and the WAMP (Web Application Messaging Protocol).",
+        "thruway/pawl-transport": "Pawl WebSocket Transport for Thruway Client",
+        "influxdb/influxdb-php": "A PHP Client for InfluxDB, a time series database"
+    },
     "support": {
         "email": "[email protected]",
         "issues": "https://github.com/php-enqueue/enqueue-dev/issues",