使用 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 平台参考》中的 配置文件所有权和权限）。
Filebeat 将尝试连接到 5044 端口。在 Logstash 启动并启动活动 Beats 插件之前，该端口上不会有任何响应，因此，现在您看到的关于无法连接到该端口的任何消息都是正常的。
配置 Logstash 以接收 Filebeat 输入
编辑接下来，您将创建一个使用 Beats 输入插件接收来自 Beats 事件的 Logstash 配置管道。
以下文本表示配置管道的框架
# 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 { 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 了。
尝试基于 grok 过滤器插件创建的字段对 Elasticsearch 进行测试查询。将 $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集群。接下来，您将学习如何创建一个使用多个输入和输出插件的管道。