Refactor OTLPMetricsClient to use ESQL with TS and use CSV Streaming

README.md

@@ -8,9 +8,15 @@ This module requires the frontend module:
 
 - https://github.com/NETWAYS/icingaweb2-module-perfdatagraphs
 
+## Known Issues
+
+> Allowed memory size exhausted
+
+Lower the `max_data_points` so that less data is returned or reduce the number of metrics being fetched via the `metrics_include/exclude` settings.
+
 ## Installation Requirements
 
 * PHP version ≥ 8.0
 * Icinga2 ElasticsearchWriter or OTLPMetricsWriter
 * IcingaDB or IDO Database
-* Elasticsearch
+* Elasticsearch (OTLP requires at least Elasticsearch 9.2)

application/forms/PerfdataGraphsElasticsearchConfigForm.php

@@ -73,6 +73,17 @@ public function createElements(array $formData)
             'description' => t('Skip the TLS verification'),
             'label' => t('Skip the TLS verification')
         ]);
+
+        $this->addElement('number', 'elasticsearch_api_max_data_points', [
+            'label' => t('The maximum numbers of datapoints each series returns'),
+            'description' => t(' '),
+            'description'   => t(
+                'The maximum numbers of datapoints each series returns.'
+                    . ' The module will use an aggregations query to downsample to this number.'
+            ),
+            'required' => false,
+            'placeholder' => 10000,
+        ]);
     }
 
     public function addSubmitButton()
@@ -160,10 +171,7 @@ public static function validateFormData($form): array
         $password = $form->getValue('elasticsearch_api_password', '');
         $index = $form->getValue('elasticsearch_api_index', 'icinga2');
         $tlsVerify = (bool) $form->getValue('elasticsearch_api_tls_insecure', false);
-
-        // TODO: Not yet implemented
-        $maxDataPoints = 10000;
-        // $maxDataPoints = $form->getValue('elasticsearch_max_data_points', 10000);
+        $maxDataPoints = (int) $form->getValue('elasticsearch_api_max_data_points', 5000);
 
         $writer = $form->getValue('elasticsearch_icinga_writer', '');
 

library/Perfdatagraphselasticsearch/Client/BaseClient.php

@@ -22,16 +22,14 @@
 abstract class BaseClient
 {
     protected Transport $transport;
-
-    // TODO: Currently unused
     protected int $maxDataPoints;
 
     /**
      * parseDuration parses the duration string from the frontend
      * into something we can use with the API (from parameter).
      *
      * @param string $duration ISO8601 Duration
-     * @param string $now current time (used in testing)
+     * @param DateTime $now current time (used in testing)
      * @return string
      */
     public function parseDuration(\DateTime $now, string $duration): string
@@ -137,6 +135,21 @@ public function search(array $params = [])
         return $d;
     }
 
+    public function query(string $query = '')
+    {
+        $uri = '_query?format=csv';
+        $method = 'POST';
+
+        $body = Json::encode(['query' => $query]);
+
+        $req = new Request($method, $uri, [], $body);
+
+        $response = $this->transport->sendRequest($req, true);
+        // TODO: Error handling
+
+        return $response;
+    }
+
     /**
      * status tests connectivity to the Elasticsearch cluster
      * @return array

library/Perfdatagraphselasticsearch/Client/ESInterface.php

@@ -19,5 +19,6 @@ public function fetchMetrics(
         bool $isHostCheck,
         array $includeMetrics,
         array $excludeMetrics,
+        int $checkInterval
     ): PerfdataResponse;
 }

library/Perfdatagraphselasticsearch/Client/ElasticsearchClient.php

@@ -13,8 +13,9 @@
 use Icinga\Application\Logger;
 use Icinga\Util\Json;
 
-use GuzzleHttp\Client;
+use DateTime;
 use Exception;
+use GuzzleHttp\Client;
 
 /**
  * ElasticsearchClient is used with with Icinga2 ElasticsearchWriter
@@ -63,7 +64,7 @@ public static function fromConfig(Config $moduleConfig = null): ESInterface
             'api_url' => 'http://localhost:9200',
             'api_index' => 'icinga2',
             'api_timeout' => 10,
-            'api_max_data_points' => 10000,
+            'api_max_data_points' => 5000,
             'api_username' => '',
             'api_password' => '',
             'api_tls_insecure' => false,
@@ -83,11 +84,11 @@ public static function fromConfig(Config $moduleConfig = null): ESInterface
         $baseURI = rtrim($moduleConfig->get('elasticsearch', 'api_url', $default['api_url']), '/');
         $index = rtrim($moduleConfig->get('elasticsearch', 'api_index', $default['api_index']), 'icinga2');
         $timeout = (int) $moduleConfig->get('elasticsearch', 'api_timeout', $default['api_timeout']);
-        $maxDataPoints = (int) $moduleConfig->get('elasticsearch', 'api_max_data_points', $default['api_max_data_points']);
         $username = $moduleConfig->get('elasticsearch', 'api_username', $default['api_username']);
         $password = $moduleConfig->get('elasticsearch', 'api_password', $default['api_password']);
         // Hint: We use a "skip TLS" logic in the UI, but Guzzle uses "verify TLS"
         $tlsVerify = !(bool) $moduleConfig->get('elasticsearch', 'api_tls_insecure', $default['api_tls_insecure']);
+        $maxDataPoints = (int) $moduleConfig->get('elasticsearch', 'api_max_data_points', $default['api_max_data_points']);
 
         return new static($baseURI, $username, $password, $maxDataPoints, $timeout, $tlsVerify, $index);
     }
@@ -126,7 +127,11 @@ public function fetchMetrics(
         bool $isHostCheck,
         array $includeMetrics,
         array $excludeMetrics,
+        int $checkInterval = 0
     ): PerfdataResponse {
+        $now = new DateTime();
+        $from = $this->parseDuration($now, $from);
+
         $params = [
             'size' => 2000,
             'sort' => '@timestamp:asc',

library/Perfdatagraphselasticsearch/Client/EsqlCsvParser.php

@@ -0,0 +1,96 @@
+<?php
+
+namespace Icinga\Module\Perfdatagraphselasticsearch\Client;
+
+use GuzzleHttp\Psr7\Stream;
+
+/**
+ * EsqlCsvParser takes a CSV Stream and returns nice little Records
+ */
+class EsqlCsvParser
+{
+    private $response;
+    private $resource;
+    private $stream;
+
+    public $closed;
+
+    public function __construct(Stream $response)
+    {
+        $this->response = $response;
+        $this->resource = $response->detach();
+        $this->closed = false;
+    }
+
+    public function each()
+    {
+        // avg_threshold,avg_perfdata,attributes.perfdata_label,attributes.threshold_type,bucket_epoch_s
+        // 0.0,,load1,min,1783430880
+        // ,0.09,load1,,1783430880
+        // 3.0,,load15,warning,1783430880
+        try {
+            while (($csv = fgetcsv($this->resource)) !== false) {
+                if (!isset($csv) || (count($csv) == 1 && $csv[0] == null)) {
+                    continue;
+                }
+
+                // Skip the header
+                if ($csv[0] == 'avg_threshold') {
+                    continue;
+                }
+
+                $result = $this->parseLine($csv);
+
+                if ($result instanceof EsqlRecord) {
+                    yield $result;
+                }
+            }
+        } finally {
+            $this->closeConnection();
+        }
+    }
+
+    private function parseLine(array $csv): EsqlRecord
+    {
+        // 0              1             2                          3                          4
+        // avg_threshold, avg_perfdata, attributes.perfdata_label, attributes.threshold_type, bucket_epoch_s
+        // 0.0,,load1,min,1783430880
+        // ,0.09,load1,,1783430880
+        // 3.0,,load15,warning,1783430880
+        $label = $csv[2];
+        $timestamp = $csv[4];
+        $value = $csv[1] === '' ? null: floatval($csv[1]);
+        $recordType = $csv[3] === '' ? 'value': $csv[3];
+
+        $warn = null;
+        $crit = null;
+        $unit = '';
+
+        if ($recordType == 'warning') {
+            $warn = $csv[0] === '' ? null: floatval($csv[0]);
+        }
+
+        if ($recordType == 'critical') {
+            $crit = $csv[0] === '' ? null: floatval($csv[0]);
+        }
+
+        $record = new EsqlRecord($recordType, $label, $timestamp, $value, $warn, $crit, $unit);
+
+        return $record;
+    }
+
+    private function closeConnection(): void
+    {
+        # Close CSV Parser
+        $this->closed = true;
+        if (isset($this->response)) {
+            $this->response->close();
+        }
+        if (is_resource($this->resource)) {
+            fclose($this->resource);
+        }
+
+        unset($this->response);
+        unset($this->resource);
+    }
+}

library/Perfdatagraphselasticsearch/Client/EsqlRecord.php

@@ -0,0 +1,70 @@
+<?php
+
+namespace Icinga\Module\Perfdatagraphselasticsearch\Client;
+
+/**
+ * EsqlRecord represents a single CSV line
+ */
+class EsqlRecord
+{
+    protected string $recordType;
+    protected string $label;
+    protected int $timestamp;
+    protected ?float $value;
+    protected ?float $warning;
+    protected ?float $critical;
+    protected ?string $unit;
+
+    public function __construct(
+        string $recordType,
+        string $label,
+        int $timestamp,
+        ?float $value,
+        ?float $warn,
+        ?float $crit,
+        ?string $unit,
+    ) {
+        $this->recordType = $recordType;
+        $this->label = $label;
+        $this->timestamp = $timestamp;
+        $this->value = $value;
+        $this->warning = $warn;
+        $this->critical = $crit;
+        $this->unit = $unit;
+    }
+
+    public function getRecordType(): string
+    {
+        return $this->recordType;
+    }
+
+    public function getLabel(): string
+    {
+        return $this->label;
+    }
+
+    public function getTimestamp(): int
+    {
+        return $this->timestamp;
+    }
+
+    public function getValue(): ?float
+    {
+        return $this->value;
+    }
+
+    public function getWarning(): ?float
+    {
+        return $this->warning;
+    }
+
+    public function getCritical(): ?float
+    {
+        return $this->critical;
+    }
+
+    public function getUnit(): ?string
+    {
+        return $this->unit;
+    }
+}