Laravel使用Elasticsearch存储日志折腾笔记

2019-03-25 本文已影响0人孤城浪子55555

为什么要用Elasticsearch存储Laravel日志而不是直接使用默认的文件存储？
1. 当PHP部署在多台服务器时，如果需要查找日志则要在每台服务器上面进行查找。
2. 通常日志是按天分割的，如果不确定是哪一天还需要在好几个文件里面进行查找，然后需要查找的文件数就变成了不确定的天数*负载均衡的服务器数量。
3. 在服务器上面直接通过命令行查询查找日志内容真的不方便。
开始折腾
1. 首先得有Elasticsearch服务器，自己在服务器上面安装或者使用第三方提供的服务，我这里直接使用AWS的服务。
2. 因为Elasticsearch就是通过标准的RESTful接口进行操作，所以PHP也就不用安装什么扩展了，但是为了方便使用还是要安装一个Packagist：
```
https://packagist.org/packages/elasticsearch/elasticsearch
composer require elasticsearch/elasticsearch
```
如果没使用过可以看中文文档：
```
https://www.elastic.co/guide/cn/elasticsearch/php/current/index.html
```
这里我就不安装Laravel的package，毕竟只是把日志写到Elasticsearch就行了，所以自己动手写个简单的ElasticsearchClient类，里面就只有一个getClient方法：

<?php

/**
 *===================================================
 * Filename:ElasticsearchClient.php
 * Author:f4ck_langzi@foxmail.com
 * Date:2018-06-15 18:31
 *===================================================
 **/

namespace App\Libs;

use Elasticsearch\ClientBuilder;

class ElasticsearchClient
{
    private $client;

    public function __construct()
    {
        $hosts = config('elasticsearch.hosts');
        $this->client = ClientBuilder::create()->setHosts($hosts)->build();
    }

    public function getClient()
    {
        return $this->client;
    }
}

为了能够配置Elasticsearch相关信息，我创建了配置文件elasticsearch.php，只有hosts和log_name(Index):

<?php
/**
 *===================================================
 * Filename:elasticsearch.php
 * Author:f4ck_langzi@foxmail.com
 * Date:2018-06-15 18:32
 *===================================================
 **/
return [
    'hosts'=>[
        env('ELASTIC_HOST')
    ],
    'log_name'=>env('ELASTIC_LOG_NAME')
];

现在就可以在Laravel中通过(new ElasticsearchClient())->getClient()来获取到Elasticsearch的Client对象了。

在页面的每一次请求中肯定会打印多次日志，如果每次打印日志都要创建Elasticsearch的Client对象就会会消耗一定的时间和性能，为了能够更加优雅的使用Client对象，我创建了一个ElasticsearchClientProvider:

<?php

namespace App\Providers;

use App\Libs\ElasticsearchClient;
use Illuminate\Support\ServiceProvider;

class ElasticsearchClientProvider extends ServiceProvider
{
    protected $defer = true;

    /**
     * Bootstrap the application services.
     *
     * @return void
     */
    public function boot()
    {
        //
    }

    /**
     * Register the application services.
     *
     * @return void
     */
    public function register()
    {
        $this->app->singleton('elasticsearch', function () {
            return new ElasticsearchClient();
        });
    }

    public function provides()
    {
        return ['elasticsearch'];
    }

}

有了这个就不用每次都来new一次，使用的时候通过app('elasticsearch')->getClient();直接从容器中拿出来就即可；其实还可以写个Facade门脸类来进一步简化代码，这里就不去麻烦了。

以上步骤只是把Elasticsearch集成到了Laravel中，要想把日志直接放到Elasticsearch还需要一些工作。
接下来修改Laravel默认的Log存储方式为Elasticsearch，通过网上查询资料发现有两种方式可以修改：

http://www.muyesanren.com/2017/09/15/laravel-how-to-store-logging-with-mongodb/

第一种是在bootstrap/app.php的return $app之前修改，第二种是在app/providers/AppServiceProvider.php中修改，我采用更加友好的第二种方式。由于参考的文章使用的是MongoDB来存储日志，因此他的代码是这样的：

$monolog = Log::getMonolog();
$mongoHost = env('MONGO_DB_HOST');
$mongoPort = env('MONGO_DB_PORT');
$mongoDsn = 'mongodb://' . $mongoHost . ':' . $mongoPort;
$mongoHandler = new \Monolog\Handler\MongoDBHandler(new \MongoClient($mongoDsn), 'laravel_project_db', 'logs');
$monolog->pushHandler($mongoHandler);

但是我这里不能使用MongoDBHandler，于是自己动手创建一个ElasticsearchLogHandler继承Monolog\Handler\AbstractProcessingHandler并实现write(别问我怎么知道需要继承这个类的，我也是看到MongoDBHandler继承了这个类才知道的)：

<?php

/**
 *===================================================
 * Filename:ElasticsearchLogHandler.php
 * Author:f4ck_langzi@foxmail.com
 * Date:2018-06-15 11:57
 *===================================================
 **/

namespace App\Libs;

use App\Jobs\ElasticsearchLogWrite;
use Monolog\Handler\AbstractProcessingHandler;

class ElasticsearchLogHandler extends AbstractProcessingHandler
{
    protected function write(array $record)
    {
        //只
        if ($record['level'] >= 200)
            dispatch((new ElasticsearchLogWrite($record)));
    }
}

调试过程中我发现每次打印日志都会执行write方法，于是准备在write函数里面动手写【吧日志存储到elasticsearch的逻辑】，我尝试了，然后发现每次打印日志就要写一次，还是同步的.....，所以我搞了一个Job把这个操作放到队列中执行，就是长这个样子：

<?php

namespace App\Jobs;

use Elasticsearch\Client;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Contracts\Queue\ShouldQueue;

class ElasticsearchLogWrite extends Job implements ShouldQueue
{
    use InteractsWithQueue;

    private $params;

    /**
     * Create a new job instance.
     *
     * @return void
     */
    public function __construct(array $record)
    {
        unset($record['context']);
        unset($record['extra']);
        $record['datetime']=$record['datetime']->format('Y-m-d H:i:s');
        $this->params = [
            'index' => config('elasticsearch.log_name'),
            'type'  => 'log',
            'body'  => $record,
        ];
    }

    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    {
        $client = app('elasticsearch')->getClient();
        if ($client instanceof Client) {
            $client->index($this->params);
        }
    }
}

到目前为止基本是实现功能了，这个时候每条日志会同时写进文件和Elasticsearch，如果你不希望日志还写进文件中可以在

$monolog->pushHandler(
    new ElasticsearchLogHandler()
);

之前使用$monolog->popHandler();把默认的文件存储去掉。

上面最然是实现了把日志写进Elasticsearch中，但是每条日志都要写一次，就算是放到队列里面当日志量比较大的时候也是可能把redis撑爆的。那么有没有什么办法可以在每次请求结束的时候一次性写入到Elasticsearch呢？答案肯定是有的，因为我发现了\vendor\monolog\monolog\src\Monolog\Handler\ElasticSearchHandler.php这个文件，原来Monolog已经自带了把日志写入到Elasticsearch的功能，我之前居然都没有去找找....
代码如下：

<?php

/*
 * This file is part of the Monolog package.
 *
 * (c) Jordi Boggiano <j.boggiano@seld.be>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Monolog\Handler;

use Monolog\Formatter\FormatterInterface;
use Monolog\Formatter\ElasticaFormatter;
use Monolog\Logger;
use Elastica\Client;
use Elastica\Exception\ExceptionInterface;

/**
 * Elastic Search handler
 *
 * Usage example:
 *
 *    $client = new \Elastica\Client();
 *    $options = array(
 *        'index' => 'elastic_index_name',
 *        'type' => 'elastic_doc_type',
 *    );
 *    $handler = new ElasticSearchHandler($client, $options);
 *    $log = new Logger('application');
 *    $log->pushHandler($handler);
 *
 * @author Jelle Vink <jelle.vink@gmail.com>
 */
class ElasticSearchHandler extends AbstractProcessingHandler
{
    /**
     * @var Client
     */
    protected $client;

    /**
     * @var array Handler config options
     */
    protected $options = array();

    /**
     * @param Client  $client  Elastica Client object
     * @param array   $options Handler configuration
     * @param int     $level   The minimum logging level at which this handler will be triggered
     * @param Boolean $bubble  Whether the messages that are handled can bubble up the stack or not
     */
    public function __construct(Client $client, array $options = array(), $level = Logger::DEBUG, $bubble = true)
    {
        parent::__construct($level, $bubble);
        $this->client = $client;
        $this->options = array_merge(
            array(
                'index'          => 'monolog',      // Elastic index name
                'type'           => 'record',       // Elastic document type
                'ignore_error'   => false,          // Suppress Elastica exceptions
            ),
            $options
        );
    }

    /**
     * {@inheritDoc}
     */
    protected function write(array $record)
    {
        $this->bulkSend(array($record['formatted']));
    }

    /**
     * {@inheritdoc}
     */
    public function setFormatter(FormatterInterface $formatter)
    {
        if ($formatter instanceof ElasticaFormatter) {
            return parent::setFormatter($formatter);
        }
        throw new \InvalidArgumentException('ElasticSearchHandler is only compatible with ElasticaFormatter');
    }

    /**
     * Getter options
     * @return array
     */
    public function getOptions()
    {
        return $this->options;
    }

    /**
     * {@inheritDoc}
     */
    protected function getDefaultFormatter()
    {
        return new ElasticaFormatter($this->options['index'], $this->options['type']);
    }

    /**
     * {@inheritdoc}
     */
    public function handleBatch(array $records)
    {
        $documents = $this->getFormatter()->formatBatch($records);
        $this->bulkSend($documents);
    }

    /**
     * Use Elasticsearch bulk API to send list of documents
     * @param  array             $documents
     * @throws \RuntimeException
     */
    protected function bulkSend(array $documents)
    {
        try {
            $this->client->addDocuments($documents);
        } catch (ExceptionInterface $e) {
            if (!$this->options['ignore_error']) {
                throw new \RuntimeException("Error sending messages to Elasticsearch", 0, $e);
            }
        }
    }
}

但是很明显我们是不能直接拿来就使用的，因为它使用的Client并不是Elasticsearch\Client而是Elastica\Client，可是我只安装了前者...，那么现在怎么办呢？

目前有两个解决办法：
1. 自己动手改造ElasticSearchHandler，太麻烦。
2. 再安装一个包咯。
  采用第二种方法：

https://packagist.org/packages/ruflin/elastica
composer require ruflin/elastica
http://elastica.io/getting-started/

然后在AppServiceProvider中直接：

$client = new \Elastica\Client(['host'=>'127.0.0.1','port'=>9200]);
$options = [
    'index' => 'dating-logs-new',
    'type'  => 'log',
];
$handler = new ElasticSearchHandler($client, $options,200);
$log = Log::getMonolog();
$log->pushHandler($handler);

本来超级简单的东西被我搞得这么复杂。

呸呸呸，我还以为自带的ElasticSearchHandler是在每次请求结束的时候一次性把日志写进去的，我看了下源码发现还是每条日志都请求了网络。
现在还得想办法实现每次请求结束后统一写入日志的功能。自己动手，丰衣足食，既然要改造肯定就不能用自带的ElasticSearchHandler了，就在之前的ElasticsearchLogHandler动手吧。

我们要实现的功能是在每次请求结束时批量写入日志到Elasticsearch中去，所以需要准备两个条件：
1. 代码无论页面是否报错都会执行。这个我所知道的就是在中间件中完成：https://laravel-china.org/docs/laravel/5.5/middleware/1294#terminable-middleware
2. 要有批量写入的接口。这个条件比较简单实现，上面的两个扩展包都有这个功能，由于之前一直用的elasticsearch/elasticsearch扩展，我们就还是使用这个，文档：https://www.elastic.co/guide/cn/elasticsearch/php/current/_indexing_documents.html

首先来创建一个中间件

php artisan make:middleware ElasticsearchBulkWrite

添加一个terminate方法

<?php

namespace App\Http\Middleware;

use Closure;

class ElasticsearchBulkWrite
{
    /**
     * Handle an incoming request.
     *
     * @param  \Illuminate\Http\Request  $request
     * @param  \Closure  $next
     * @return mixed
     */
    public function handle($request, Closure $next)
    {
        return $next($request);
    }

    public function terminate($request, $response)
    {

    }
}

还要添加到app/Http/Kernel.php文件的全局中间件中

/**
 * The application's global HTTP middleware stack.
 *
 * These middleware are run during every request to your application.
 *
 * @var array
 */
protected $middleware = [
    \Illuminate\Foundation\Http\Middleware\CheckForMaintenanceMode::class,
    ElasticsearchBulkWrite::class,
];

等会儿我们直接在terminate函数中写批量写入到Elasticsearch的方法即可。

改造日志写入流程。
之前的流程是：

原先流程图

    st=>start: 产生日志
    op1=>operation: My Operation
    A(产生日志) -->B(ElasticsearchLogHandler的write方法分发队列)
    B -->C(Job任务队列执行写入)
    e=>end:

改造之后的流程：

改造后的流程图

graph TD
    A(等待产生日志) -->B(ElasticsearchLogHandler的write方法暂存文日志到ElasticsearchClient的$documents属性中)
    B -->C{请求是否结束}
    C -->|否| A
    C -->|是| D(请求结束后中间件terminate拿出之前暂存的日志一次性分发到队列中)
    D -->F(Job任务队列批量写入)

要实现后面的流程则要对ElasticsearchLogHandler，ElasticsearchClient和App\Jobs\ElasticsearchLogWrite三个文件进行改造：

ElasticsearchLogHandler只实现添加日志的功能：

<?php
/**
 *===================================================
 * Filename:ElasticsearchLogHandler.php
 * Author:f4ck_langzi@foxmail.com
 * Date:2018-06-15 11:57
 *===================================================
 **/

namespace App\Libs;

use Monolog\Handler\AbstractProcessingHandler;

class ElasticsearchLogHandler extends AbstractProcessingHandler
{
    protected function write(array $record)
    {
        if ($record['level'] >= 200)
            app('elasticsearch')->addDocument($record);
    }
}

ElasticsearchClient实现添加文档和获取所有已添加日志的功能：

<?php

/**
 *===================================================
 * Filename:ElasticsearchClient.php
 * Author:f4ck_langzi@foxmail.com
 * Date:2018-06-15 18:31
 *===================================================
 **/

namespace App\Libs;

use Elasticsearch\ClientBuilder;

class ElasticsearchClient
{
    protected $client;

    protected $documents = [];

    public function __construct()
    {
        $hosts = config('elasticsearch.hosts');
        $this->client = ClientBuilder::create()->setHosts($hosts)->build();
    }

    public function getClient()
    {
        return $this->client;
    }

    /**
     * @function Name addDocument
     * @description 添加日志
     * @param array $document
     */
    public function addDocument(array $document)
    {
        $this->documents[] = $document;
    }

    /**
     * @function Name getDocuments
     * @description 获取所有已添加日志
     * @return mixed
     */
    public function getDocuments()
    {
        return $this->documents;
    }
}

实现批量写入到Elasticsearch功能：

<?php

namespace App\Jobs;

use Elasticsearch\Client;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Contracts\Queue\ShouldQueue;

class ElasticsearchLogWrite extends Job implements ShouldQueue
{
    use InteractsWithQueue;

    private $params;

    /**
     * Create a new job instance.
     *
     * @return void
     */
    public function __construct(array $records)
    {
        $this->params['body'] = [];
        //A good start are 500 documents per bulk operation. Depending on the size of your documents you’ve to play around a little how many documents are a good number for your application.
        foreach ($records as $record) {
            unset($record['context']);
            unset($record['extra']);
            $record['datetime'] = $record['datetime']->format('Y-m-d H:i:s');
            $this->params['body'][] = [
                'index' => [
                    '_index' => config('elasticsearch.log_name'),
                    '_type'  => 'log',
                ],
            ];
            $this->params['body'][] = $record;
        }
    }

    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    {
        $client = app('elasticsearch')->getClient();
        if ($client instanceof Client) {
            $client->bulk($this->params);
        }
    }
}

最后在中间件中加入分发代码即可

public function terminate($request, $response)
{
    $documents = app('elasticsearch')->getDocuments();
    //需要判断是否有日志
    if (count($documents) > 0)
        dispatch(new ElasticsearchLogWrite($documents));
}

这样基本上就是实现功能啦。当然其中还有很多细节是需要去完善的，这里只是记录了整个折腾过程，看起来可能会比较乱。

Laravel使用Elasticsearch存储日志折腾笔记

猜你喜欢

热点阅读