Logstash的filter插件-Grok
Posted shark_西瓜甜
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Logstash的filter插件-Grok相关的知识,希望对你有一定的参考价值。
一、介绍
Grok是一种将非结构化日志数据解析为结构化和可查询数据的好方法。
这个工具非常适合于syslog日志、apache和其他web服务器日志、mysql日志,以及通常为人类而不是计算机使用编写的任何日志格式。
二、选择
Grok or Dissect?还是两者兼有?
dissect filter插件是使用分隔符将非结构化事件数据提取到字段中的另一种方法。
Dissect与Grok的不同之处在于它不使用正则表达式,而且速度更快。当数据可靠地重复时,Dissect工作得很好。当文本的结构因行而异时,Grok是一个更好的选择。
当行的一部分可靠地重复,但整行没有重复时,您可以对混合用例使用Dissect和Grok。分离过滤器可以解构重复的线段。Grok过滤器可以处理剩余的字段值,具有更大的regex可预测性。
三、Grok 基础
Grok的工作原理是将文本模式组合成与日志匹配的内容。
grok模式的语法: %{SYNTAX:SEMANTIC}
SYNTAX 是与文本匹配的模式的名称。例如,3.44将由 NUMBER 模式匹配,55.3.244.1将由 IP 模式匹配。
SEMANTIC (语义)是您为匹配的文本段指定的标识符(可理解为字段名称)。例如,3.44可以是一个事件的持续时间,因此可以简单地称之为 duration
。此外,字符串55.3.244.1可能标识发出请求的 client
。
对于上面的示例,grok过滤器如下所示:
%{NUMBER:duration} %{IP:client}
您可以选择向 grok 模式添加数据类型转换。默认情况下,所有语义都保存为字符串。如果您希望转换语义的数据类型,例如,将字符串更改为整数,然后用目标数据类型作为后缀。例如 %{NUMBER:num:int}
它将 num
语义从字符串转换为整数。目前唯一支持的转换是 int
和 float
。
使用语义和语法,我们可以从一个示例日志中提取有用的字段,如下面这个虚构的http请求日志:
55.3.244.1 GET /index.html 15824 0.043
其模式可能是:
%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}
更现实的例子是,让我们从文件中读取以下日志:
input {
file {
path => "/var/log/http.log"
}
}
filter {
grok {
match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
}
}
在grok过滤器之后,事件中会有几个额外的字段:
client: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
duration: 0.043
四、正则表达式
Grok位于正则表达式之上,因此任何正则表达式在Grok中都是有效的。正则表达式库是Oniguruma,您可以在 Oniguruma站点上看到完全支持的regexp语法。
五、自定义模式
有时候logstash没有你需要的模式。为此,你有几个选择。
1 首先,可以使用Oniguruma语法对捕获到的内容命名,这样可以匹配一段文本并将其保存为字段:
(?<field_name>the pattern here)
例如,日志的后缀 队列 id 的值为10或11个字符的十六进制。我可以这样轻松地捕捉到:
(?<queue_id>[0-9A-F]{10,11})
一个实际的示例如下
input { stdin {}}
filter {
grok {
match => {"message" => "(?<queue_id>[0-9A-F]{10,11})" }
}
}
output { stdout {}}
在终端输入 BEF25A72965
会看到如下输出
{
"@version" => "1",
"host" => "es03",
"queue_id" => "BEF25A72965",
"message" => "BEF25A72965",
"@timestamp" => 2021-06-07T13:54:19.661Z
}`
2 或者,您可以创建一个自定义模式文件。
创建一个名为patterns的目录,其中包含一个名为 postfix 的文件(文件名无关紧要,但要为自己命名)
在该文件中,编写所需的模式,作为模式名称、空格,然后是该模式的regexp。
例如,如上所述执行postfix queue id示例:
# ./patterns/postfix 的内容:
POSTFIX_QUEUEID [0-9A-F]{10,11}
然后使用这个插件中的 patterns_dir
设置告诉logstash你的自定义patterns目录在哪里。下面是一个完整的示例和一个示例日志:
Jan 1 06:25:43 mailserver14 postfix/cleanup[21403]: BEF25A72965: message-id=<20130101142543.5828399CCAF@mailserver14.example.com>
filter {
grok {
patterns_dir => ["./patterns"]
match => { "message" => "%{SYSLOGBASE} %{POSTFIX_QUEUEID:queue_id}: %{GREEDYDATA:syslog_message}" }
}
}
以上内容将匹配并产生以下字段:
timestamp: Jan 1 06:25:43
logsource: mailserver14
program: postfix/cleanup
pid: 21403
queue_id: BEF25A72965
syslog_message: message-id=<20130101142543.5828399CCAF@mailserver14.example.com>
六、关于匹配的细节
[root@es03 logstash]# cat testgrok-dir.conf
input { stdin {}}
filter {
grok {
patterns_dir => ["./patterns"]
match => {"message" => "%{POSTFIX_QUEUEID:queue_id}:" }
}
}
output { stdout {}}
以上是关于Logstash的filter插件-Grok的主要内容,如果未能解决你的问题,请参考以下文章