ElasticSearch v8 support

.github/workflows/continuous-integration.yml

@@ -33,6 +33,33 @@ jobs:
       - name: "Checkout"
         uses: "actions/checkout@v2"
 
+      # required for elasticsearch
+      - name: Configure sysctl limits
+        run: |
+          sudo swapoff -a
+          sudo sysctl -w vm.swappiness=1
+          sudo sysctl -w fs.file-max=262144
+          sudo sysctl -w vm.max_map_count=262144
+
+      - name: Runs Elasticsearch
+        timeout-minutes: 1
+        continue-on-error: true
+        uses: elastic/elastic-github-actions/elasticsearch@master
+        with:
+          stack-version: 7.17.3
+
+      - name: Run CouchDB
+        timeout-minutes: 1
+        continue-on-error: true
+        uses: "cobot/couchdb-action@master"
+        with:
+          couchdb version: '3.2.2'
+
+      - name: Run MongoDB
+        uses: supercharge/mongodb-github-action@1.7.0
+        with:
+          mongodb-version: 5.0
+
       - name: "Install PHP"
         uses: "shivammathur/setup-php@v2"
         with:
@@ -60,6 +87,7 @@ jobs:
 
       - name: "Install latest dependencies"
         run: |
+          composer require --no-update --no-interaction --dev elasticsearch/elasticsearch:^7
           composer update ${{ env.COMPOSER_FLAGS }}
 
       - name: "Run tests"
@@ -72,3 +100,16 @@ jobs:
           composer require --no-update psr/log:^3
           composer update -W ${{ env.COMPOSER_FLAGS }}
           composer exec phpunit -- --verbose
+
+      - name: Runs Elasticsearch
+        timeout-minutes: 1
+        uses: elastic/elastic-github-actions/elasticsearch@master
+        with:
+          stack-version: 8.2.0
+
+      - name: "Run ElasticSearch tests with elasticsearch 8"
+        run: |
+          composer remove --no-update --dev ruflin/elastica
+          composer require --no-update --no-interaction --dev elasticsearch/elasticsearch:^8
+          composer update ${{ env.COMPOSER_FLAGS }}
+          composer exec phpunit -- --filter Elasticsearch --verbose

composer.json

@@ -19,7 +19,7 @@
     "require-dev": {
         "aws/aws-sdk-php": "^2.4.9 || ^3.0",
         "doctrine/couchdb": "~1.0@dev",
-        "elasticsearch/elasticsearch": "^7",
+        "elasticsearch/elasticsearch": "^7 || ^8",
         "graylog2/gelf-php": "^1.4.2",
         "mongodb/mongodb": "^1.8",
         "php-amqplib/php-amqplib": "~2.4 || ^3",

phpstan.neon.dist

@@ -35,3 +35,6 @@ parameters:
         - '#::popProcessor\(\) should return callable#'
         - '#Parameter \#1 \$ of callable \(callable\(Monolog\\Handler\\Record\): Monolog\\Handler\\Record\)#'
         - '#is incompatible with native type array.#'
+
+        # legacy elasticsearch namespace failures
+        - '# Elasticsearch\\#'

src/Monolog/Handler/ElasticsearchHandler.php

@@ -11,6 +11,7 @@
 
 namespace Monolog\Handler;
 
+use Elastic\Elasticsearch\Response\Elasticsearch;
 use Throwable;
 use RuntimeException;
 use Monolog\Logger;
@@ -19,6 +20,8 @@
 use InvalidArgumentException;
 use Elasticsearch\Common\Exceptions\RuntimeException as ElasticsearchRuntimeException;
 use Elasticsearch\Client;
+use Elastic\Elasticsearch\Exception\InvalidArgumentException as ElasticInvalidArgumentException;
+use Elastic\Elasticsearch\Client as Client8;
 
 /**
  * Elasticsearch handler
@@ -44,7 +47,7 @@
 class ElasticsearchHandler extends AbstractProcessingHandler
 {
     /**
-     * @var Client
+     * @var Client|Client8
      */
     protected $client;
 
@@ -54,11 +57,20 @@ class ElasticsearchHandler extends AbstractProcessingHandler
     protected $options = [];
 
     /**
-     * @param Client  $client  Elasticsearch Client object
-     * @param mixed[] $options Handler configuration
+     * @var bool
      */
-    public function __construct(Client $client, array $options = [], $level = Logger::DEBUG, bool $bubble = true)
+    private $needsType;
+
+    /**
+     * @param Client|Client8 $client  Elasticsearch Client object
+     * @param mixed[]        $options Handler configuration
+     */
+    public function __construct($client, array $options = [], $level = Logger::DEBUG, bool $bubble = true)
     {
+        if (!$client instanceof Client && !$client instanceof Client8) {
+            throw new \TypeError('Elasticsearch\Client or Elastic\Elasticsearch\Client instance required');
+        }
+
         parent::__construct($level, $bubble);
         $this->client = $client;
         $this->options = array_merge(
@@ -69,6 +81,14 @@ public function __construct(Client $client, array $options = [], $level = Logger
             ],
             $options
         );
+
+        if ($client instanceof Client8 || $client::VERSION[0] === '7') {
+            $this->needsType = false;
+            // force the type to _doc for ES8/ES7
+            $this->options['type'] = '_doc';
+        } else {
+            $this->needsType = true;
+        }
     }
 
     /**
@@ -133,16 +153,19 @@ protected function bulkSend(array $records): void
 
             foreach ($records as $record) {
                 $params['body'][] = [
-                    'index' => [
+                    'index' => $this->needsType ? [
                         '_index' => $record['_index'],
                         '_type'  => $record['_type'],
+                    ] : [
+                        '_index' => $record['_index'],
                     ],
                 ];
                 unset($record['_index'], $record['_type']);
 
                 $params['body'][] = $record;
             }
 
+            /** @var Elasticsearch */
             $responses = $this->client->bulk($params);
 
             if ($responses['errors'] === true) {
@@ -160,16 +183,20 @@ protected function bulkSend(array $records): void
      *
      * Only the first error is converted into an exception.
      *
-     * @param mixed[] $responses returned by $this->client->bulk()
+     * @param mixed[]|Elasticsearch $responses returned by $this->client->bulk()
      */
-    protected function createExceptionFromResponses(array $responses): ElasticsearchRuntimeException
+    protected function createExceptionFromResponses($responses): Throwable
     {
         foreach ($responses['items'] ?? [] as $item) {
             if (isset($item['index']['error'])) {
                 return $this->createExceptionFromError($item['index']['error']);
             }
         }
 
+        if (class_exists(ElasticInvalidArgumentException::class)) {
+            return new ElasticInvalidArgumentException('Elasticsearch failed to index one or more records.');
+        }
+
         return new ElasticsearchRuntimeException('Elasticsearch failed to index one or more records.');
     }
 
@@ -178,10 +205,14 @@ protected function createExceptionFromResponses(array $responses): Elasticsearch
      *
      * @param mixed[] $error
      */
-    protected function createExceptionFromError(array $error): ElasticsearchRuntimeException
+    protected function createExceptionFromError(array $error): Throwable
     {
         $previous = isset($error['caused_by']) ? $this->createExceptionFromError($error['caused_by']) : null;
 
+        if (class_exists(ElasticInvalidArgumentException::class)) {
+            return new ElasticInvalidArgumentException($error['type'] . ': ' . $error['reason'], 0, $previous);
+        }
+
         return new ElasticsearchRuntimeException($error['type'] . ': ' . $error['reason'], 0, $previous);
     }
 }

tests/Monolog/Handler/AmqpHandlerTest.php

@@ -78,7 +78,7 @@ public function testHandleAmqpExt()
 
     public function testHandlePhpAmqpLib()
     {
-        if (!class_exists('PhpAmqpLib\Connection\AMQPConnection')) {
+        if (!class_exists('PhpAmqpLib\Channel\AMQPChannel')) {
             $this->markTestSkipped("php-amqplib not installed");
         }
 

tests/Monolog/Handler/ElasticaHandlerTest.php

@@ -155,60 +155,6 @@ public function providerTestConnectionErrors()
         ];
     }
 
-    /**
-     * Integration test using localhost Elastic Search server version <7
-     *
-     * @covers Monolog\Handler\ElasticaHandler::__construct
-     * @covers Monolog\Handler\ElasticaHandler::handleBatch
-     * @covers Monolog\Handler\ElasticaHandler::bulkSend
-     * @covers Monolog\Handler\ElasticaHandler::getDefaultFormatter
-     */
-    public function testHandleIntegration()
-    {
-        $msg = [
-            'level' => Logger::ERROR,
-            'level_name' => 'ERROR',
-            'channel' => 'meh',
-            'context' => ['foo' => 7, 'bar', 'class' => new \stdClass],
-            'datetime' => new \DateTimeImmutable("@0"),
-            'extra' => [],
-            'message' => 'log',
-        ];
-
-        $expected = $msg;
-        $expected['datetime'] = $msg['datetime']->format(\DateTime::ISO8601);
-        $expected['context'] = [
-            'class' => '[object] (stdClass: {})',
-            'foo' => 7,
-            0 => 'bar',
-        ];
-
-        $client = new Client();
-        $handler = new ElasticaHandler($client, $this->options);
-
-        try {
-            $handler->handleBatch([$msg]);
-        } catch (\RuntimeException $e) {
-            $this->markTestSkipped("Cannot connect to Elastic Search server on localhost");
-        }
-
-        // check document id from ES server response
-        $documentId = $this->getCreatedDocId($client->getLastResponse());
-        $this->assertNotEmpty($documentId, 'No elastic document id received');
-
-        // retrieve document source from ES and validate
-        $document = $this->getDocSourceFromElastic(
-            $client,
-            $this->options['index'],
-            $this->options['type'],
-            $documentId
-        );
-        $this->assertEquals($expected, $document);
-
-        // remove test index from ES
-        $client->request("/{$this->options['index']}", Request::DELETE);
-    }
-
     /**
      * Integration test using localhost Elastic Search server version 7+
      *
@@ -274,6 +220,7 @@ protected function getCreatedDocId(Response $response)
         if (!empty($data['items'][0]['create']['_id'])) {
             return $data['items'][0]['create']['_id'];
         }
+        var_dump('Unexpected response: ', $data);
     }
 
     /**