使用 Logstash 解析日志
编辑使用 Logstash 解析日志
编辑在存储您的第一个事件中，您创建了一个基本的 Logstash 管道来测试您的 Logstash 设置。在现实世界中，Logstash 管道会稍微复杂一些：它通常具有一个或多个输入、过滤器和输出插件。
在本节中，您将创建一个 Logstash 管道，该管道使用 Filebeat 将 Apache Web 日志作为输入，解析这些日志以从日志中创建特定的、命名的字段，并将解析的数据写入 Elasticsearch 集群。您将在配置文件中定义管道，而不是在命令行中定义管道配置。
要开始，请单击此处下载本示例中使用的示例数据集。解压缩该文件。
配置 Filebeat 将日志行发送到 Logstash
编辑在创建 Logstash 管道之前，您将配置 Filebeat 将日志行发送到 Logstash。Filebeat 客户端是一个轻量级、资源友好的工具，用于从服务器上的文件中收集日志，并将这些日志转发到您的 Logstash 实例进行处理。Filebeat 的设计目标是可靠性和低延迟。Filebeat 在主机上的资源占用量很小，并且Beats 输入插件最大限度地减少了 Logstash 实例上的资源需求。
在典型的用例中，Filebeat 在与运行 Logstash 实例的计算机不同的计算机上运行。在本教程中，Logstash 和 Filebeat 在同一台计算机上运行。
默认的 Logstash 安装包括Beats 输入插件。Beats 输入插件使 Logstash 能够接收来自 Elastic Beats 框架的事件，这意味着任何为与 Beats 框架一起工作而编写的 Beat，例如 Packetbeat 和 Metricbeat，也可以将事件数据发送到 Logstash。
要在您的数据源计算机上安装 Filebeat，请从 Filebeat 产品页面下载相应的软件包。您还可以参考Filebeat 快速入门，以获取其他安装说明。
安装 Filebeat 后，您需要对其进行配置。打开位于 Filebeat 安装目录中的 filebeat.yml 文件，并将内容替换为以下行。确保 paths 指向您之前下载的示例 Apache 日志文件 logstash-tutorial.log。
filebeat.inputs:
- type: log
paths:
- /path/to/file/logstash-tutorial.log
output.logstash:
hosts: ["localhost:5044"]
保存更改。
为了使配置简单，您不会像在现实场景中那样指定 TLS/SSL 设置。
在数据源计算机上，使用以下命令运行 Filebeat
sudo ./filebeat -e -c filebeat.yml -d "publish"
如果以 root 身份运行 Filebeat，则需要更改配置文件所有权（请参阅 _Beats Platform Reference_ 中的配置文件所有权和权限）。
Filebeat 将尝试连接端口 5044。在 Logstash 启动并使用活动 Beats 插件之前，该端口上不会有任何响应，因此您现在看到的任何有关无法连接到该端口的消息都是正常的。
配置 Logstash 以进行 Filebeat 输入
编辑接下来，您将创建一个 Logstash 配置管道，该管道使用 Beats 输入插件接收来自 Beats 的事件。
以下文本表示配置管道的框架
# 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 {
}
此框架是非功能的，因为输入和输出部分没有任何有效的选项定义。
要开始，请将框架配置管道复制并粘贴到主 Logstash 目录中名为 first-pipeline.conf 的文件中。
接下来，通过将以下行添加到 first-pipeline.conf 文件的 input 部分，配置 Logstash 实例以使用 Beats 输入插件
beats {
port => "5044"
}
您将在稍后配置 Logstash 以写入 Elasticsearch。现在，您可以将以下行添加到 output 部分，以便在运行 Logstash 时将输出打印到 stdout
stdout { codec => rubydebug }
完成后，first-pipeline.conf 的内容应如下所示
input {
beats {
port => "5044"
}
}
# The filter part of this file is commented out to indicate that it is
# optional.
# filter {
#
# }
output {
stdout { codec => rubydebug }
}
要验证您的配置，请运行以下命令
bin/logstash -f first-pipeline.conf --config.test_and_exit
--config.test_and_exit 选项会解析您的配置文件并报告任何错误。
如果配置文件通过配置测试，请使用以下命令启动 Logstash
bin/logstash -f first-pipeline.conf --config.reload.automatic
--config.reload.automatic 选项启用自动配置重新加载，这样您就不必每次修改配置文件时都停止并重新启动 Logstash。
当 Logstash 启动时，您可能会看到一个或多个有关 Logstash 忽略 pipelines.yml 文件的警告消息。您可以安全地忽略此警告。pipelines.yml 文件用于在单个 Logstash 实例中运行多个管道。对于此处显示的示例，您正在运行单个管道。
如果您的管道工作正常，您应该看到一系列类似以下内容的事件写入控制台
{
"@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"
},
"input" => {
"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 过滤器插件解析 Web 日志
编辑现在您有了一个可以从 Filebeat 读取日志行的工作管道。但是，您会注意到日志消息的格式并不理想。您希望解析日志消息以从日志中创建特定的、命名的字段。为此，您将使用 grok 过滤器插件。
grok 过滤器插件是 Logstash 中默认提供的多个插件之一。有关如何管理 Logstash 插件的详细信息，请参阅插件管理器的参考文档。
grok 过滤器插件使您能够将非结构化的日志数据解析为结构化且可查询的数据。
由于 grok 过滤器插件会在传入的日志数据中查找模式，因此配置插件需要您决定如何识别您用例感兴趣的模式。Web 服务器日志示例的代表性行如下所示
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 地址很容易识别，括号中的时间戳也很容易识别。要解析数据，您可以使用 %{COMBINEDAPACHELOG} grok 模式，该模式使用以下架构构造 Apache 日志中的行
|
信息 |
字段名称 |
|
IP 地址 |
|
|
用户 ID |
|
|
用户身份验证 |
|
|
时间戳 |
|
|
HTTP 动词 |
|
|
请求主体 |
|
|
HTTP 版本 |
|
|
HTTP 状态代码 |
|
|
已服务的字节数 |
|
|
引用 URL |
|
|
用户代理 |
|
如果您在构建 grok 模式时需要帮助，请尝试使用Grok 调试器。Grok 调试器是基本许可证下的 X-Pack 功能，因此可以免费使用。
编辑 first-pipeline.conf 文件，并将整个 filter 部分替换为以下文本
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}"}
}
}
完成后，first-pipeline.conf 的内容应如下所示
input {
beats {
port => "5044"
}
}
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}"}
}
}
output {
stdout { codec => rubydebug }
}
保存更改。由于您已启用自动配置重新加载，因此无需重新启动 Logstash 即可接收您的更改。但是，您需要强制 Filebeat 从头开始读取日志文件。为此，请转到 Filebeat 正在运行的终端窗口并按 Ctrl+C 以关闭 Filebeat。然后删除 Filebeat 注册表文件。例如，运行
sudo rm data/registry
由于 Filebeat 将其采集的每个文件的状态存储在注册表中，因此删除注册表文件会强制 Filebeat 从头开始读取其采集的所有文件。
接下来，使用以下命令重新启动 Filebeat
sudo ./filebeat -e -c filebeat.yml -d "publish"
如果 Filebeat 需要等待 Logstash 重新加载配置文件，则 Filebeat 可能需要稍作延迟才会开始处理事件。
在 Logstash 应用 grok 模式后，事件将具有以下 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"
},
"input" => {
"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 过滤器插件增强数据
编辑除了解析日志数据以进行更好的搜索之外，过滤器插件还可以从现有数据中派生补充信息。例如，geoip 插件会查找 IP 地址，从这些地址中派生地理位置信息，并将该位置信息添加到日志中。
通过将以下行添加到 first-pipeline.conf 文件的 filter 部分，配置 Logstash 实例以使用 geoip 过滤器插件
geoip {
source => "clientip"
}
geoip 插件配置要求您指定包含要查找的 IP 地址的源字段的名称。在本示例中，clientip 字段包含 IP 地址。
由于过滤器按顺序评估，请确保 geoip 部分位于配置文件的 grok 部分之后，并且 grok 和 geoip 部分都嵌套在 filter 部分中。
完成后，first-pipeline.conf 的内容应如下所示
input {
beats {
port => "5044"
}
}
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}"}
}
geoip {
source => "clientip"
}
}
output {
stdout { codec => rubydebug }
}
保存更改。要强制 Filebeat 从头开始读取日志文件（如之前所做的那样），请关闭 Filebeat（按 Ctrl+C），删除注册表文件，然后使用以下命令重新启动 Filebeat
sudo ./filebeat -e -c filebeat.yml -d "publish"
请注意，事件现在包含地理位置信息
{
"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 中了。
您可以在自己的硬件上运行 Elasticsearch，也可以使用我们托管在 AWS、GCP 和 Azure 上的 Elasticsearch 服务。免费试用 Elasticsearch 服务。
Logstash 管道可以将数据索引到 Elasticsearch 集群中。编辑 first-pipeline.conf 文件，并将整个 output 部分替换为以下文本
output {
elasticsearch {
hosts => [ "localhost:9200" ]
}
}
使用此配置，Logstash 使用 http 协议连接到 Elasticsearch。以上示例假定 Logstash 和 Elasticsearch 在同一实例上运行。您可以使用 hosts 配置来指定类似于 hosts => [ "es-machine:9092" ] 的内容，以此指定远程 Elasticsearch 实例。
此时，您的 first-pipeline.conf 文件已正确配置了输入、过滤器和输出部分，如下所示
input {
beats {
port => "5044"
}
}
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}"}
}
geoip {
source => "clientip"
}
}
output {
elasticsearch {
hosts => [ "localhost:9200" ]
}
}
保存更改。要强制 Filebeat 从头开始读取日志文件（如之前所做的那样），请关闭 Filebeat（按 Ctrl+C），删除注册表文件，然后使用以下命令重新启动 Filebeat
sudo ./filebeat -e -c filebeat.yml -d "publish"
测试您的管道
编辑现在，Logstash 管道已配置为将数据索引到 Elasticsearch 集群中，您可以查询 Elasticsearch。
尝试对 Elasticsearch 进行测试查询，查询基于 grok 过滤器插件创建的字段。将 $DATE 替换为当前日期，格式为 YYYY.MM.DD
curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=response=200'
索引名称中使用的日期基于 UTC，而不是 Logstash 运行的时区。如果查询返回 index_not_found_exception，请确保 logstash-$DATE 反映了索引的实际名称。要查看可用索引的列表，请使用此查询：curl 'localhost:9200/_cat/indices?v'。
您应该会收到多个命中结果。例如
{
"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"
},
"input": {
"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
curl -XGET 'localhost:9200/logstash-$DATE/_search?pretty&q=geoip.city_name=Buffalo'
一些日志条目来自布法罗，因此查询会产生以下响应
{
"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"
},
"input": {
"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 数据
请参阅 Filebeat 快速入门文档，了解有关为 Filebeat 加载 Kibana 索引模式的信息。
您已成功创建了一个管道，该管道使用 Filebeat 将 Apache Web 日志作为输入，解析这些日志以从日志中创建特定的命名字段，并将解析后的数据写入 Elasticsearch 集群。接下来，您将学习如何创建使用多个输入和输出插件的管道。