使用 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 插件之前，该端口上不会有任何响应，因此您看到的有关在该端口上无法连接的任何消息在目前都是正常的。
为 Filebeat 输入配置 Logstash编辑
接下来，您将创建一个 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 { 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 重新加载配置文件，则可能需要稍等片刻才能开始处理事件。
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编辑
现在，网络日志已分解为特定字段，您已准备好将数据导入 Elasticsearch。
您可以在自己的硬件上运行 Elasticsearch，也可以使用我们托管的 Elasticsearch 服务，该服务可在 AWS、GCP 和 Azure 上使用。 免费试用 Elasticsearch 服务.
Logstash 管道可以将数据索引到 Elasticsearch 集群中。编辑 first-pipeline.conf 文件，并将整个 output 部分替换为以下文本
output {
elasticsearch {
hosts => [ "localhost:9200" ]
}
}
使用此配置，Logstash 使用 http 协议连接到 Elasticsearch。以上示例假设 Logstash 和 Elasticsearch 在同一实例上运行。您可以通过使用 hosts 配置指定远程 Elasticsearch 实例，例如 hosts => [ "es-machine:9092" ]。
此时，您的 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 的 Kibana 索引模式的信息，请参阅 Filebeat 快速入门文档。
您已成功创建了一个管道，该管道使用 Filebeat 将 Apache 网络日志作为输入，解析这些日志以从日志中创建特定的命名字段，并将解析后的数据写入 Elasticsearch 集群。接下来，您将学习如何创建使用多个输入和输出插件的管道。