[Monolog] Added ElasticsearchLogstashHandler

Author: lyrixx

src/Symfony/Bridge/Monolog/CHANGELOG.md

@@ -5,6 +5,7 @@ CHANGELOG
 -----
 
 * The `RouteProcessor` class has been made final
+* Added `ElasticsearchLogstashHandler`
 
 4.3.0
 -----

src/Symfony/Bridge/Monolog/Handler/ElasticsearchLogstashHandler.php

@@ -0,0 +1,125 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Bridge\Monolog\Handler;
+
+use Monolog\Formatter\FormatterInterface;
+use Monolog\Formatter\LogstashFormatter;
+use Monolog\Handler\AbstractHandler;
+use Monolog\Logger;
+use Symfony\Component\HttpClient\HttpClient;
+use Symfony\Contracts\HttpClient\HttpClientInterface;
+
+/**
+ * Push logs directly to Elasticsearch and format them according to Logstash specification.
+ *
+ * This handler dials directly with the HTTP interface of elasticsearch. This
+ * means it will slow down your application if elasticsearch takes times to
+ * answer. Even if all HTTP calls are done asynchronously.
+ *
+ * In a development environment, it's fine to keep the default configuration:
+ * For each log, an HTTP request will be made to push the log to Elasticsearch.
+ *
+ * But in a production environment, it's highly recommended to wrap this handler
+ * in an handler with buffering capability (like the FingersCrossedHandler, or
+ * BufferHandler) in order to call Elasticsearch only once. For even better
+ * performance and fault tolerance, a proper ELK stack is recommended.
+ *
+ * @author Grégoire Pineau <lyrixx@lyrixx.info>
+ */
+class ElasticsearchLogstashHandler extends AbstractHandler
+{
+    private $endpoint;
+    private $index;
+    private $client;
+    private $responses;
+
+    public function __construct(string $endpoint = 'http://127.0.0.1:9200', string $index = 'monolog', HttpClientInterface $client = null, int $level = Logger::DEBUG, bool $bubble = true)
+    {
+        if (!interface_exists(HttpClientInterface::class)) {
+            throw new \LogicException(sprintf('%s handler needs symfony/http-client, please run `composer require symfony/http-client`.', __CLASS__));
+        }
+
+        parent::__construct($level, $bubble);
+        $this->endpoint = $endpoint;
+        $this->index = $index;
+        $this->client = $client ?: HttpClient::create(['timeout' => 1]);
+        $this->responses = new \SplObjectStorage();
+    }
+
+    public function handle(array $record): bool
+    {
+        if (!$this->isHandling($record)) {
+            return false;
+        }
+
+        $this->sendToElasticsearch([$record]);
+
+        return false === $this->bubble;
+    }
+
+    public function handleBatch(array $records): void
+    {
+        $records = array_filter($records, function (array $record) {
+            return $this->isHandling($record);
+        });
+
+        if (!$records) {
+            return;
+        }
+
+        $this->sendToElasticsearch($records);
+    }
+
+    protected function getDefaultFormatter(): FormatterInterface
+    {
+        return new LogstashFormatter('application', null, null, 'ctxt_', LogstashFormatter::V1);
+    }
+
+    private function sendToElasticsearch(array $records)
+    {
+        $formatter = $this->getFormatter();
+
+        $body = '';
+        foreach ($records as $record) {
+            if ($this->processors) {
+                foreach ($this->processors as $processor) {
+                    $record = \call_user_func($processor, $record);
+                }
+            }
+
+            $body .= json_encode([
+                'index' => [
+                    '_index' => $this->index,
+                    '_type' => '_doc',
+                ],
+            ]);
+            $body .= "\n";
+            $body .= $formatter->format($record);
+            $body .= "\n";
+        }
+
+        $response = $this->client->request('POST', $this->endpoint.'/_bulk', [
+            'body' => $body,
+            'headers' => [
+                'Content-Type' => 'application/json',
+            ],
+        ]);
+
+        $this->responses->attach($response);
+
+        foreach ($this->client->stream($this->responses, 0) as $response => $chunk) {
+            if (!$chunk->isTimeout() && $chunk->isFirst()) {
+                $this->responses->detach($response);
+            }
+        }
+    }
+}