Logstash日志解析的Grok模式详解

1.Grok filter入门
- 1.1 什么是 Grok?
- 1.2 Grok 是如何工作的？
- 1.3 一个 Logstash Grok例子
- 1.4 常见的例子
  - 1.4.1 Syslog
  - 1.4.2 Apache access logs
  - 1.4.3 Elasticsearch
  - 1.4.4 Redis
  - 1.4.5 MogoDB
  - 1.4.6 AWS
  - 1.4.7 总结
- 1.5 总结
2. 日志解析的Grok模式示例
- 2.1 什么是 grok?
- 2.2 Grok 是如何工作的？
- 2.3 Logstash Grok 模式例子
- 2.4 Grok 数据类型转换
- 2.5 _grokparsefailure
- 2.6 操作数据
- 2.7 完整示例
  - 2.7.1 提取 IP
  - 2.7.2 提取 timestamps
  - 2.7.3 提取 verbs
  - 2.7.4 提取 request
  - 2.7.5 提取 status
  - 2.7.6 提取 Bytes
  - 2.7.7 提取 referrer
  - 2.7.8 忽略数据及提取 os

1.Grok filter入门

有效分析和查询送入Elastic Stack的数据的能力取决于信息的可读性。这意味着，当将非结构化数据摄取到系统中时，必须将其转换为结构化数据。

通常，这个至关重要的任务留给Logstash（尽管还有其他日志传送器可用，比如 Fluentd）。无论你定义什么数据源，都必须提取日志并执行一些魔力来美化它们，以确保在将它们输出到Elasticsearch之前正确地对其进行了解析。

Logstash中的数据操作是使用过滤器插件执行的。本文重点介绍最流行和有用的过滤器插件之一 ——Logstash grok过滤器，该过滤器用于将非结构化数据解析为结构化数据。

1.1 什么是 Grok?

最初的术语实际上是很新的由罗伯特·A·海因莱因（Robert A. Heinlein）在他的1961年的《陌生的土地上的陌生人》一书中创造的，指的是理解某种东西，使人们真正沉浸其中。这是 grok 语言和 Logstash grok 插件的合适名称，它们可以以一种格式修改信息并将其浸入另一种格式（特别是 JSON）。已经有数百种用于记录的 Grok 模式。

1.2 Grok 是如何工作的？

简而言之，grok是一种将行与正则表达式匹配，将行的特定部分映射到专用字段中以及根据此映射执行操作的方法。

内置了超过 200 种 Logstash 模式，用于过滤 AWS，Bacula，Bro，Linux-Syslog 等中的单词，数字和日期等项目。如果找不到所需的模式，则可以编写自己的自定义模式。还有多个匹配模式的选项，可简化表达式的编写以捕获日志数据。

这是 Logstash grok 过滤器的基本语法格式：

%{PATTERN:FieldName}

这将匹配预定义的模式，并将其映射到特定的标识字段。由于grok本质上是基于正则表达式的组合，因此你也可以创建自己的基于正则表达式的 grok 过滤器。例如：

(?\d\d-\d\d-\d\d)

这将使 22-22-22（或任何其他数字）的正则表达式与字段名称匹配。

1.3 一个 Logstash Grok例子

这个 grok 调试工具是开始构建 grok 过滤器的好方法：Grok Debugger。你也可以在 Kibana 中找到 grok 的调试工具：

使用此工具，你可以粘贴日志消息并逐步构建 grok 模式，同时连续测试编译。通常，我建议从 ％{GREEDYDATA：message} 模式开始，然后逐步添加越来越多的模式。

在上面的示例中，我将从以下内容开始：

%{GREEDYDATA:message}

比如针对这条信息：

83.149.9.216 - - [17/May/2015:10:05:03 +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"

我们可以看到整个信息都赋予给 message 了。

我们再接着尝试如下的例子：

%{IPORHOST:clientip} - - \[%{HTTPDATE:timestamp}\] %{GREEDYDATA:message}

在上面，我们可以看到 clientip, message及timestamp。

1.4 常见的例子

以下是一些示例，可帮助你熟悉如何构造grok过滤器：

1.4.1 Syslog

用 Grok 解析 syslog 消息是新用户更普遍的需求之一。 syslog 还有几种不同的日志格式，因此请牢记编写自己的自定义 grok 模式。这是常见系统日志解析的一个示例：

May  4 00:10:36 liuxg xpcproxy[69746]: libcoreservices: _dirhelper_userdir: 557: bootstrap_look_up returned (ipc/send) invalid destination port

grok解析为：

%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?:%{GREEDYDATA:syslog_message}

在logstash中，我们应该使用如下的grok filter：

grok {
   match => { 
    "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp}
    %{SYSLOGHOST:syslog_hostname}
    %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?:
    %{GREEDYDATA:syslog_message}" 
  }
}

1.4.2 Apache access logs

80.135.37.131 - - [11/Sep/2019:23:56:45 +0000] "GET /item/giftcards/4852 HTTP/1.1" 200 91 "-" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.6; rv:9.0.1) Gecko/20100101 Firefox/9.0.1"

grok解析为：

%{COMBINEDAPACHELOG}

在logstash中，我们应该使用如下的grok filter：

grok  {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
}

1.4.3 Elasticsearch

[2017-09-10T12:07:26,683][WARN ][index.indexing.slowlog.index] [GOgO9TD] [testindex-slowlogs/yNbyYk1ARSW_hd0YRh6J0A] took[142.3micros], took_millis[0], type[product], id[105], routing[] , source[{"price":9925,"name":"Nariko"}]

grok解析为：

\[%{TIMESTAMP_ISO8601:timestamp}\]\[%{DATA:loglevel}%{SPACE}\]\[%{DATA:source}\]%{SPACE}\[%{DATA:node}\]%{SPACE}\[%{DATA:index}\] %{NOTSPACE} \[%{DATA:updated-type}\]

\[%{TIMESTAMP_ISO8601:timestamp}\]\[%{DATA:loglevel}%{SPACE}\]\[%{DATA:source}\]%{SPACE}\[%{DATA:node}\] (\[%{NOTSPACE:index}\]\[%{NUMBER:shards}\])?%{GREEDYDATA}

在logstash中，我们应该使用如下的grok filter：

grok {
      match => ["message", "\[%{TIMESTAMP_ISO8601:timestamp}\]\[%{DATA:loglevel}%{SPACE}\]\[%{DATA:source}%{SPACE}\]%{SPACE}\[%{DATA:node}\]%{SPACE}\[%{DATA:index}\] %{NOTSPACE} \[%{DATA:updated-type}\]",
                "message", "\[%{TIMESTAMP_ISO8601:timestamp}\]\[%{DATA:loglevel}%{SPACE}\]\[%{DATA:source}%{SPACE}\]%{SPACE}\[%{DATA:node}\] (\[%{NOTSPACE:Index}\]\[%{NUMBER:shards}\])?%{GREEDYDATA}"
      ]
   }

上面有两种匹配的方法，只要其中的一种可以匹配就可以了。

1.4.4 Redis

grok {
       match => ["redistimestamp", "\[%{MONTHDAY} %{MONTH} %{TIME}]",
                ["redislog", "\[%{POSINT:pid}\] %{REDISTIMESTAMP:timestamp}"],
                ["redismonlog", "\[%{NUMBER:timestamp} \[%{INT:database} %{IP:client}:%{NUMBER:port}\] "%{WORD:command}"\s?%{GREEDYDATA:params}"]
      ]
    }

1.4.5 MogoDB

MONGO_LOG %{SYSLOGTIMESTAMP:timestamp} \[%{WORD:component}\] %{GREEDYDATA:message}MONGO_QUERY \{ (?<={ ).*(?= } ntoreturn:) \}MONGO_SLOWQUERY %{WORD} %{MONGO_WORDDASH:database}\.%{MONGO_WORDDASH:collection} %{WORD}: %{MONGO_QUERY:query} %{WORD}:%{NONNEGINT:ntoreturn} %{WORD}:%{NONNEGINT:ntoskip} %{WORD}:%{NONNEGINT:nscanned}.*nreturned:%{NONNEGINT:nreturned}..+ (?``[0-9]+)msMONGO_WORDDASH \b[\w-]+\bMONGO3_SEVERITY \wMONGO3_COMPONENT %{WORD}|-MONGO3_LOG %{TIMESTAMP_ISO8601:timestamp} %{MONGO3_SEVERITY:severity} %{MONGO3_COMPONENT:component}%{SPACE}(?:\[%{DATA:context}\])? %{GREEDYDATA:message}

1.4.6 AWS

ELB_ACCESS_LOG %{TIMESTAMP_ISO8601:timestamp} %{NOTSPACE:elb} %{IP:clientip}:%{INT:clientport:int} (?:(%{IP:backendip}:?:%{INT:backendport:int})|-) %{NUMBER:request_processing_time:float} %{NUMBER:backend_processing_time:float} %{NUMBER:response_processing_time:float} %{INT:response:int} %{INT:backend_response:int} %{INT:received_bytes:int} %{INT:bytes:int} "%{ELB_REQUEST_LINE}"CLOUDFRONT_ACCESS_LOG (?``%{YEAR}-%{MONTHNUM}-%{MONTHDAY}\t%{TIME})\t%{WORD:x_edge_location}\t(?:%{NUMBER:sc_bytes:int}|-)\t%{IPORHOST:clientip}\t%{WORD:cs_method}\t%{HOSTNAME:cs_host}\t%{NOTSPACE:cs_uri_stem}\t%{NUMBER:sc_status:int}\t%{GREEDYDATA:referrer}\t%{GREEDYDATA:agent}\t%{GREEDYDATA:cs_uri_query}\t%{GREEDYDATA:cookies}\t%{WORD:x_edge_result_type}\t%{NOTSPACE:x_edge_request_id}\t%{HOSTNAME:x_host_header}\t%{URIPROTO:cs_protocol}\t%{INT:cs_bytes:int}\t%{GREEDYDATA:time_taken:float}\t%{GREEDYDATA:x_forwarded_for}\t%{GREEDYDATA:ssl_protocol}\t%{GREEDYDATA:ssl_cipher}\t%{GREEDYDATA:x_edge_response_result_type}

1.4.7 总结

Logstash grok 只是一种过滤器，可以在将日志转发到 Elasticsearch 之前将其应用于你的日志。因为它在日志管道中扮演着如此重要的角色，所以 grok 也是最常用的过滤器之一。

以下是一些有用的资源列表，可以帮助你深入了解：

此处提供了可用于操作日志的可用操作的完整列表，以及它们的输入类型和默认值。

2.7 完整示例

让我们从一个示例非结构化日志消息开始，然后我们将使用 Grok 模式对其进行结构化：

128.39.24.23 - - [25/Dec/2021:12:16:50 +0000] "GET /category/electronics HTTP/1.1" 200 61 "/category/finance" "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)"

想象一下搜索数百万条看起来像这样的日志行！看起来很可怕。这就是为什么我们有像 Grok 这样的解析语言——让数据更易于阅读和搜索。

查看该数据并使用 Kibana 等分析工具进行搜索的一种更简单的方法是将其分解为具有值的字段，如下表所示：

1.ip:128.39.24.23
2.timestamp:25/Dec/2021:12:16:50 +0000
3.verb:GET
4.request:/category/electronics HTTP/1.1
5.status: 200
6.bytes: 61
7.referrer:/category/finance
8.os: Windows

让我们使用一个示例 Grok 模式来生成这些字段。以下部分将展示 Grok 模式语法以生成上述每个字段。

对于每个部分，我们的 Grok 模式都会扩展，因为它包含更多要解析的字段。这些模式是 Grok 可以阅读的正则表达式——我们可以使用它们来表达我们的信息。

下面是一些有用的链接，可以帮助你开始使用一些 Grok 模式。但我们将在博客的其余部分提供更多示例。

Grok pattern
Grok Debugger
在 Grok pattern 页面，我们可以找到我们需要的预定义好的 pattern：

从某种意义上讲，这些 pattern 在本质上就是基于正则表达式的组合。也可以创建自己定义的 pattern。

让我们开始构建 Grok 模式来结构化数据。

我们使用以下方式描述模式的开始： ^
语法如下： %{pattern:Name_of_the_field}

请注意：不建议使用空格来描述字段的名称。

2.7.1 提取 IP

假设我们要提取 IP，我们可以使用 IP 方法：

^%{IP:ip}

然后我们有这些： – –

为了告诉 Grok 忽略它们，我们只需将它们添加到我们的模式中。

^%{IP:ip} - -

这个模式给了我们这个字段：

ip:128.39.24.23

2.7.2 提取 timestamps

在我们非结构化日志消息的下一部分中，我们将时间戳 “含” 在一个数组中：

[25/Dec/2021:12:16:50 +0000]
要提取它，我们需要使用正则表达式和 HTTPDATE 方法，同时在外面添加括号，以便 Grok 知道忽略它们：

\[%{HTTPDATE:timestamp}\]

基于我们之前的模式，我们现在有：

^%{IP:ip} - - \[%{HTTPDATE:timestamp}\]

这给了我们：

"ip": "128.39.24.23",
"timestamp": "25/Dec/2021:12:16:50 +0000"

回到我们最初的非结构化消息，看起来我们有一个空间，时间戳结束，“GET 开始”。我们还需要告诉 Grok 忽略空格。为此，我们只需按键盘上的空格键，或者我们可以使用 %{SPACE} -> 直到 4 个空格。

2.7.3 提取 verbs

是时候提取 GET 字段了。首先，我们需要告诉 Grok 忽略引号 - 然后使用 WORD 方法，我们将通过编写来做到这一点：

"%{WORD:verb}

所以，现在我们的模式是：

^%{IP:ip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:verb}

这给了我们这些字段：

"ip": "128.39.24.23",
"verb": "GET",
"timestamp": "25/Dec/2021:12:16:50 +0000"

2.7.4 提取 request

为了提取请求 -> /category/electronics HTTP/1.1"，我们需要使用 DATA 方法，它本质上是正则表达式中的通配符。

这意味着我们需要添加一个句号来提取此信息，以告诉 DATA 方法在哪里停止——否则，它不会捕获任何数据。我们可以使用引号作为停止标记：

%{DATA:request}"

现在，我们有以下 Grok 模式：

^%{IP:ip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request}"

这给了我们这些字段：

"request": "/category/electronics HTTP/1.1",
"ip": "128.39.24.23",
"verb": "GET",
"timestamp": "25/Dec/2021:12:16:50 +0000"

2.7.5 提取 status

接下来是状态，但我们在请求结束和状态之间再次有一个空格，我们可以添加一个空格或 %{SPACE}。为了提取数字，我们使用 NUMBER 方法。

%{NUMBER:status}

现在我们的模式扩展到：

^%{IP:ip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request}" %{NUMBER:status}

这给了我们如下的字段：

"request": "/category/electronics HTTP/1.1",
"ip": "128.39.24.23",
"verb": "GET",
"timestamp": "25/Dec/2021:12:16:50 +0000",
"status": "200"

2.7.6 提取 Bytes

为了提取字节，我们需要再次使用NUMBER方法，但在此之前，我们需要使用常规空格或 %{SPACE}

^%{IP:ip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request}" %{NUMBER:status} %{NUMBER:bytes}

这给了我们如下的字段：

"request": "/category/electronics HTTP/1.1",
"bytes": "61",
"ip": "128.39.24.23",
"verb": "GET",
"timestamp": "25/Dec/2021:12:16:50 +0000",
"status": "200"

2.7.7 提取 referrer

要提取"referrer"，我们需要使用常规空格或 %{SPACE} -> " 所以 Grok 会忽略它：

%{DATA:referrer}"

如你所见，我们添加 DATA 直到它遇到 " 字符：

这给了我们如下的字段：

"request": "/category/electronics HTTP/1.1",
"referrer": "",
"bytes": "61",
"ip": "128.39.24.23",
"verb": "GET",
"timestamp": "25/Dec/2021:12:16:50 +0000",
"status": "200"

2.7.8 忽略数据及提取 os

现在我们想忽略这个数据 ->" "Mozilla/5.0 (compatible; MSIE 9.0; 为此，我们将使用 "Mozilla/5.0"的 DATA 方法，但不写入该字段将忽略它。

然后我们将使用不带冒号或字段名的 WORD 方法忽略 (compatible；

最后，我们将使用 DATA 忽略 MSIE 9.0；

这给我们留下了以下模式来忽略该数据

%{DATA}\(%{WORD};%{DATA};

为了进一步解释这种模式......

( -> 停止直到它到达 (
; -> 停止直到到达 ;

请注意上面的模式中的 ( 字符。第一个 DATA 遇到 ( 即停止，也就是第一个数据。第二个 DATA 停止直到遇到 ; 符号。同样地，第三个 DATA 停止直到遇到 ;

现在我们可以使用 WORD 方法提取 os -> %{WORD:os}。就是这样！现在我们剩下以下 Grok 模式来构建我们的数据。

^%{IP:ip} - - \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request}" %{NUMBER:status} %{NUMBER:bytes} %{DATA:referrer}" %{DATA}\(%{WORD};%{DATA}; %{WORD:os}

这为我们提供了这些整洁的字段，我们可以使用这些字段更轻松地搜索和可视化我们的日志数据：

"request": "/category/electronics HTTP/1.1",
"referrer": "\"/category/finance",
"os": "Windows",
"bytes": "61",
"ip": "128.39.24.23",
"verb": "GET",
"timestamp": "25/Dec/2021:12:16:50 +0000",
"status": "200"

至此，我们已经完成了对整个日志信息的结构化分析。

elk

Logstash日志解析的Grok模式详解

1.Grok filter入门

1.1 什么是 Grok?

1.2 Grok 是如何工作的？

1.3 一个 Logstash Grok例子

1.4 常见的例子

1.4.1 Syslog

1.4.2 Apache access logs

1.4.3 Elasticsearch

1.4.4 Redis

1.4.5 MogoDB

1.4.6 AWS

1.4.7 总结

2.7 完整示例

2.7.1 提取 IP

2.7.2 提取 timestamps

2.7.3 提取 verbs

2.7.4 提取 request

2.7.5 提取 status

2.7.6 提取 Bytes

2.7.7 提取 referrer

2.7.8 忽略数据及提取 os

相关

ELK 简单实验

ELK查询和汇总

ELK 使用案例

ELK + Kafka

windows下部署ELK详细教程

4 - 基于ELK的ElasticSearch 7.8.x技术整理 - 高级篇（续） - 更新完毕

K8S-ELK日志系统之五：logstash+kabana

ELK-Kibana7.7应用

日志收集管理-ELK_01

日志收集管理-ELK_01

日志收集管理-ELK_01

日志收集管理-ELK_01

标签