通过Logstash解析日志 · Logstash中文文档（个人翻译）

[TOC] ### 通过Logstash解析日志上面的示例中你已经创建了一个最基本的Logstash管道来测试你的Logstash，在实际的情形中，Logstash通常面临着更加复杂的场景：一个或多个input、filter和output插件。本段，你将创建一个Logstash管道，使用Filebeat采集的Apache日志作为input，从日志中解析特殊字段，然后将解析后的数据写入一个Elasticsearch集群。这次你将定义一个配置文件，而不是在命令行中设定。在[这里](https://download.elastic.co/demos/logstash/gettingstarted/logstash-tutorial.log.gz)下载示例数据，以便我们可以开始。 #### 配置Filebeat来发送日志到Logstash 在创建Logstash管道之前，你需要配置Filebeat来将日志发送到Logstash。Filebeat是一个轻量的、友好的日志收集客户端，它从你服务器上的文件中收集日志，并转发到Logstash以进行实时处理。Filebeat是为可靠性和低延迟设计的。Filebeat可以在你的主机上占用更少的资源，使用`Beats input`插件将让你的Logstash实例的资源使用降到最低。 > NOTE：在一个典型的使用场景中，Filebeat运行在和Logstash不同的单独的主机上，在这个演示的实例中我们将Logstash和Filebeat运行在同一台主机上。默认情况下Logstash会安装`Beats input`插件。Beats input插件允许Logstash接收来自Elastic Beats框架的事件，这意味着任何基于Beats框架的Beat written如Packetbeat和Metricbeta都可以给Logstash发送事件。要在你的数据源主机上安装适合的Filebeat，可以在[这里](https://www.elastic.co/downloads/beats/filebeat)下载。你也可以在[这里](https://www.elastic.co/guide/en/beats/filebeat/6.4/filebeat-getting-started.html)查看Beats的安装文档。安装完成之后你需要进行一些配置。在你的安装目录中找到并打开`filebeat.yml`文件，然后使用下面的内容替换其中的内容。确保`paths`指向下载的测试用的Apache日志文件`logstash-tutorial.log`。 ```yaml filebeat.prospectors: - type: log paths: - /path/to/file/logstash-tutorial.log ① output.logstash: hosts: ["localhost:5044"] ``` ① 文件或目录的绝对路径。（原文：Absolute path to the file or files that Filebeat processes.）保存更改。为了更加简单，这里不用配置TLS/SSL。在生产中，通常需要配置这些。在数据源的主机上执行下面的命令来运行Filebeat： ```shell sudo ./filebeat -e -c filebeat.yml -d "publish" ``` > NOTE：如果你以root的身份运行Filebeat,你需要更改配置文件的所有权.（参考：[Config File OwnerShip and Permissions](https://www.elastic.co/guide/en/beats/libbeat/6.4/config-file-permissions.html)）。 Filebeat使用5044进行连接，在Logstash启动Beats插件之前，Filebeat不会收到任何响应，所以此时你看到的任何连接此端口失败的消息都是正常的。 #### 配置Logstash使用Filebeat作为输入接下来，你将创建一个Logstash管道使用Beats input插件接收来自Beats的事件。下面是一个配置文件的框架： ```json # The # character at the beginning of a line indicates a comment. Use # comments to describe your configuration. input { } # The filter part of this file is commented out to indicate that it is # optional. # filter { # # } output { } ``` 这个框架没有任何功能，因为在input 和 output 没有定义任何选项。在Logstash的home目录下创建一个文件，名为`first-pipeline.conf`，将上述内容复制进去。（Logstash的home目录应该是指Logstash安装目录。）接下来，在`first-pipeline.conf`文件中的`input`配置段添加以下内容来使用Beats input 插件： ```json beats { port => "5044" } ``` 你可以稍后配置Elasticsearch。现在，将下面的内容填入到配置文件中的`output`配置段。其作用是在你运行Logstash的时候将输出结果输出到stdout。 ```json stdout { codec => rubydebug } ``` 完成之后`first-pipeline.conf`的内容应该如下： ```json input { beats { port => "5044" } } # The filter part of this file is commented out to indicate that it is # optional. # filter { # # } output { stdout { codec => rubydebug } } ``` 你可以使用下面的命令来检查你的配置文件： ```shell bin/logstash -f first-pipeline.conf --config.test_and_exit ``` `--config.test_and_exit`选项会分析你的配置文件并将其中的错误输出。如果配置文件通过了检查，你可以使用下面的命令来启动Logstash： ```shell bin/logstash -f first-pipeline.conf --config.reload.automatic ``` `--config.reload.automatic`选项可以让Logstash在你修改配置文件之后重载而不必重新启动。当Logstash启动之后，你可能会看到一到多条关于忽略`pipelines.yml`文件的警告信息。你可以忽略这些警告。`pipelines.yml`文件是用来在一个Logstash示例中运行[多个管道](https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html)使用的。在这个演示的示例中，你只运行一个管道。如果运行正确你应该能在屏幕上看到如下的输出信息： ```json { "@timestamp" => 2017-11-09T01:44:20.071Z, "offset" => 325, "@version" => "1", "beat" => { "name" => "My-MacBook-Pro.local", "hostname" => "My-MacBook-Pro.local", "version" => "6.0.0" }, "host" => "My-MacBook-Pro.local", "prospector" => { "type" => "log" }, "source" => "/path/to/file/logstash-tutorial.log", "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"", "tags" => [ [0] "beats_input_codec_plain_applied" ] } ... ``` #### 使用Grok Filter插件解析Web Logs 现在你已经有了一个正常工作的从Filebeat读取日志的管道。但是，你知道日志的格式并不理想。你希望解析这些日志文件中的特殊字段。为了实现这个，你需要用到`grok` filter插件。 [Grok](http://www.elastic.co/guide/en/logstash/6.4/plugins-filters-grok.html) filter插件是少数Logstash默认可以使用的插件之一。更多关于管理Logstash插件的信息，参考[reference documentation](https://www.elastic.co/guide/en/logstash/current/working-with-plugins.html)。 `Grok`插件可以将非结构化的数据转换为高质量的结构化数据。由于`grok`插件从传入的日志中进行匹配查找，所以需要你根据自己感兴趣的数据来配置插件。如典型的Web服务日志如下： ```text 83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36" ``` 很容易辨认出最开始的是IP地址，方括号中的是时间标签。对于Apache日志你可以使用`%{COMBINEDAPACHELOG}`来将数据解析成如下结构。 | Information | Field Name | | :------------------ | ------------- | | IP Address | `clientip` | | User ID | `ident` | | User Authentication | `auth` | | timestamp | `timestamp` | | HTTP Verb | `verb` | | Request body | `request` | | HTTP Version | `httpversion` | | HTTP Status Code | `response` | | Bytes served | `bytes` | | Referrer URL | `referrer` | | User agent | `agent` | > TIP：如果在构建Grok匹配模式上需要帮助，你可以使用[Grok Debugger](https://www.elastic.co/guide/en/kibana/6.4/xpack-grokdebugger.html)。Grok Debugger是基本许可下的X-Pack的一个附加特性，可以免费使用。编辑`first-pipeline.conf`文件并使用下面的内容替换其中的`filter`字段： ```json filter { grok { match => { "message" => "%{COMBINEDAPACHELOG}"} } } ``` 完成之后`first-pipeline.conf`中的内容应该如下： ```json input { beats { port => "5044" } } filter { grok { match => { "message" => "%{COMBINEDAPACHELOG}"} } } output { stdout { codec => rubydebug } } ``` 保存更改，由于已经开启了自动重载配置文件，因此你不必重新启动Logstash来使配置生效。但是你需要让Filebeat强制从头开始读取日志文件。在对应的终端上使用Ctrl+C组合键来关闭Filebeat。然后删除Filebeat的registry文件，命令如下： ```shell sudo rm data/registry ``` Filebeat的registry文件保存了它读取的文件的状态，删除之后可以强制让Filebeat重新从头读取这些文件。然后使用下面的命令重启Filebeat： ```shell sudo ./filebeat -e -c filebeat.yml -d "publish" ``` 你可能需要等待一小会儿如果Filebeat等待Logstash重读配置文件的话。在Logstash完成重读并匹配完成之后，事件将用JSON的格式表示如下： ```json { "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png", "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"", "offset" => 325, "auth" => "-", "ident" => "-", "verb" => "GET", "prospector" => { "type" => "log" }, "source" => "/path/to/file/logstash-tutorial.log", "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"", "tags" => [ [0] "beats_input_codec_plain_applied" ], "referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"", "@timestamp" => 2017-11-09T02:51:12.416Z, "response" => "200", "bytes" => "203023", "clientip" => "83.149.9.216", "@version" => "1", "beat" => { "name" => "My-MacBook-Pro.local", "hostname" => "My-MacBook-Pro.local", "version" => "6.0.0" }, "host" => "My-MacBook-Pro.local", "httpversion" => "1.1", "timestamp" => "04/Jan/2015:05:13:42 +0000" } ``` 需要注意的是原本的消息内容会被保留，但消息同时也会被分割成不同的字段。 #### 使用GeoIP插件丰富你的数据除了解析日志以便于搜索，filter插件还可以从现有的数据中进行扩展。举个例子，`geoip`插件可以从IP地址获取物理位置信息并将其添加到日志中。在`first-pipeline.conf`文件的`filter`字段中添加以下内容来使用`geoip`filter插件： ```json geoip { source => "clientip" } ``` `geoip`插件的配置需要你指定包含要查找的IP地址的源字段的名称。在示例中，`clientip`字段包含IP地址。因为filter是按照顺序进行解析，所以配置文件中的`geoip`字段要在`grok`字段之后且这两个字段都要在`filter`字段中当你做完之后，`first-pipeline.conf`文件中的内容应该如下： ```json input { beats { port => "5044" } } filter { grok { match => { "message" => "%{COMBINEDAPACHELOG}"} } geoip { source => "clientip" } } output { stdout { codec => rubydebug } } ``` 保存设置，和之前一样强制停止Filebeat然后删除registry文件，然后使用下面的命令重启Filebeat： ```shell sudo ./filebeat -e -c filebeat.yml -d "publish" ``` 注意事件中现在已经包含了物理位置信息： ```json { "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png", "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"", "geoip" => { "timezone" => "Europe/Moscow", "ip" => "83.149.9.216", "latitude" => 55.7485, "continent_code" => "EU", "city_name" => "Moscow", "country_name" => "Russia", "country_code2" => "RU", "country_code3" => "RU", "region_name" => "Moscow", "location" => { "lon" => 37.6184, "lat" => 55.7485 }, "postal_code" => "101194", "region_code" => "MOW", "longitude" => 37.6184 }, ... ``` #### 将你的数据索引到Elasticsearch 现在web日志已经按照字段进行分割，你已经准备好将数据写入Elasticsearch。 > TIP：你可以在你自己的硬件上运行Elasticsearch，也可以使用我们的Elastic云上的[Elasticsearch主机服务](https://www.elastic.co/cloud/elasticsearch-service)。AWS和GCP都提供了Elasticsearch服务。[免费试用一下](https://www.elastic.co/cloud/elasticsearch-service/signup)。 Logstash可以将数据索引到Elasticsearch集群。编辑`first-pipeline.conf`文件中的`output`字段，内容如下： ```json output { elasticsearch { hosts => [ "localhost:9200" ] } } ``` 这个配置中，Logstash使用HTTP协议连接Elasticsearch。上面的示例中Logstash和Elasticsearch运行在相同的实例中。你也可以指定一个远程Elasticsearch实例通过配置`hosts`如：`hosts => ["es-machine:9200"]`。到这里，你的`first-pipeline.conf`包含input,filter,和output配置，没问题的话，应该是下面这样： ```json input { beats { port => "5044" } } filter { grok { match => { "message" => "%{COMBINEDAPACHELOG}"} } geoip { source => "clientip" } } output { elasticsearch { hosts => [ "localhost:9200" ] } } ``` 保存配置，像之前一样强制中止Filebeat，删除registry文件，然后使用下面的命令重启Filebeat： ```shell sudo ./filebeat -e -c filebeat.yml -d "publish" ``` #### 测试你的管道现在你的Logstash已经将数据存储到Elasticsearch集群中，你可以在Elasticsearch集群中进行查询。尝试在Elasticsearch中查询`grok`filter插件创建的字段。用YYYY.MM.DD格式的当前时间替换下面的$DATE: ```shell curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=response=200' ``` > NOTE：数据使用的索引名称基于UTC时间，并非Logstash运行的当地时间。如果查询返回`index_not_found_exception`，确保`logstash-$DATE`对应的名字是正确的索引名。可以使用这个查询指令来查看所有可用的索引：`curl 'localhost:9200/_cat/indices?v'` 应该会返回多个命中结果。如下： ```json { "took": 50, "timed_out": false, "_shards": { "total": 5, "successful": 5, "skipped": 0, "failed": 0 }, "hits": { "total": 98, "max_score": 2.793642, "hits": [ { "_index": "logstash-2017.11.09", "_type": "doc", "_id": "3IzDnl8BW52sR0fx5wdV", "_score": 2.793642, "_source": { "request": "/presentations/logstash-monitorama-2013/images/frontend-response-codes.png", "agent": """"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"""", "geoip": { "timezone": "Europe/Moscow", "ip": "83.149.9.216", "latitude": 55.7485, "continent_code": "EU", "city_name": "Moscow", "country_name": "Russia", "country_code2": "RU", "country_code3": "RU", "region_name": "Moscow", "location": { "lon": 37.6184, "lat": 55.7485 }, "postal_code": "101194", "region_code": "MOW", "longitude": 37.6184 }, "offset": 2932, "auth": "-", "ident": "-", "verb": "GET", "prospector": { "type": "log" }, "source": "/path/to/file/logstash-tutorial.log", "message": """83.149.9.216 - - [04/Jan/2015:05:13:45 +0000] "GET /presentations/logstash-monitorama-2013/images/frontend-response-codes.png HTTP/1.1" 200 52878 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"""", "tags": [ "beats_input_codec_plain_applied" ], "referrer": """"http://semicomplete.com/presentations/logstash-monitorama-2013/"""", "@timestamp": "2017-11-09T03:11:35.304Z", "response": "200", "bytes": "52878", "clientip": "83.149.9.216", "@version": "1", "beat": { "name": "My-MacBook-Pro.local", "hostname": "My-MacBook-Pro.local", "version": "6.0.0" }, "host": "My-MacBook-Pro.local", "httpversion": "1.1", "timestamp": "04/Jan/2015:05:13:45 +0000" } }, ... ``` 尝试使用从IP地址派生出的地理信息来查询。使用当前时间替换$DATE，用YYYY.MM.DD格式的时间： ```shell curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=geoip.city_name=Buffalo' ``` 有些日志来自Buffalo，所以应该会返回下面的结果： ```json { "took": 9, "timed_out": false, "_shards": { "total": 5, "successful": 5, "skipped": 0, "failed": 0 }, "hits": { "total": 2, "max_score": 2.6390574, "hits": [ { "_index": "logstash-2017.11.09", "_type": "doc", "_id": "L4zDnl8BW52sR0fx5whY", "_score": 2.6390574, "_source": { "request": "/blog/geekery/disabling-battery-in-ubuntu-vms.html?utm_source=feedburner&utm_medium=feed&utm_campaign=Feed%3A+semicomplete%2Fmain+%28semicomplete.com+-+Jordan+Sissel%29", "agent": """"Tiny Tiny RSS/1.11 (http://tt-rss.org/)"""", "geoip": { "timezone": "America/New_York", "ip": "198.46.149.143", "latitude": 42.8864, "continent_code": "NA", "city_name": "Buffalo", "country_name": "United States", "country_code2": "US", "dma_code": 514, "country_code3": "US", "region_name": "New York", "location": { "lon": -78.8781, "lat": 42.8864 }, "postal_code": "14202", "region_code": "NY", "longitude": -78.8781 }, "offset": 22795, "auth": "-", "ident": "-", "verb": "GET", "prospector": { "type": "log" }, "source": "/path/to/file/logstash-tutorial.log", "message": """198.46.149.143 - - [04/Jan/2015:05:29:13 +0000] "GET /blog/geekery/disabling-battery-in-ubuntu-vms.html?utm_source=feedburner&utm_medium=feed&utm_campaign=Feed%3A+semicomplete%2Fmain+%28semicomplete.com+-+Jordan+Sissel%29 HTTP/1.1" 200 9316 "-" "Tiny Tiny RSS/1.11 (http://tt-rss.org/)"""", "tags": [ "beats_input_codec_plain_applied" ], "referrer": """"-"""", "@timestamp": "2017-11-09T03:11:35.321Z", "response": "200", "bytes": "9316", "clientip": "198.46.149.143", "@version": "1", "beat": { "name": "My-MacBook-Pro.local", "hostname": "My-MacBook-Pro.local", "version": "6.0.0" }, "host": "My-MacBook-Pro.local", "httpversion": "1.1", "timestamp": "04/Jan/2015:05:29:13 +0000" } }, ... ``` 如果你使用Kibana来可视化你的数据，你可以在Kibana中查看你通过Filebeat获取的数据： ![Discovering Filebeat data in Kibana](https://www.elastic.co/guide/en/logstash/current/static/images/kibana-filebeat-data.png) 查看[Filebeat getting started docs](https://www.elastic.co/guide/en/beats/filebeat/6.4/filebeat-getting-started.html)获取关于在Kibana中加载Filebeat索引的信息。你已经成功的创建了一个Logstash管道，通过Filebeat采集Apache日志作为input，分析其中的特殊字段，然后将分析后的字段写入到Elasticsearch集群中。接下来，即将学习如何创建一个管道同时使用多个input和output插件。