在我之前的文章 “Elastic可观测性 - 数据结构化及处理”,讲述了如果把一个非结构化的数据变为一个结构化的数据结构。其中Grok processor 是非常重要的一个。在今天的文章中,我们来更加深入地对它进行描述。今天的这个 Grok 的实践也适用于 Logstash 的 Grok filer。关于 Logstash Grok 的教程,你可以参阅文章 “Logstash:Grok filter 入门”。
我们先来看一下如下的一个日志:
157.97.192.70 2019 09 29 00:39:02.912 myserver Process 107673 Initializing
在上面的日志中,我们可以看到一个日期信息:2019 09 29 00:39:02.912。它是被空格字符串所分开,如果没有正确的 Grok pattern 来帮我们提取的话,我们将会很难提取到一个完整的日期。我们的日志信息符合如下的一个数据结构:
ip timestamp server Process process_id action
首先,我们打开 Kibana:
我们可以先提取 IP:
之后的,就是年,月,日,及时间。我们可以通过如下的方式来进行提取:
- 运用 YEAY 来提取年份
- 运用 MONTHNUM 来提取月份
- 运用 MONTHDAY 来提取日期
- 运用 TIME 来提取时间
- 运用 WORD 来提取一个单词
- 运用 NUMBER 来提取一个数值
- 对于 Process 来说,我们就不提取了,忽略它
这样,我们可以使用如下的 Grok pattern:
%{IP:ip} %{YEAR:year} %{MONTHNUM:month} %{MONTHDAY:day} %{TIME:time} %{WORD:server} Process %{NUMBER:process_id} %{WORD:action}
显然,它正确地解析了我们的日志,但是美中不足的是我们最终需要的是一个真正的日期,而不是用 year, month, day, time 来表示的一个时间。我们可以点击上面的 custerm pattern,并输入一下的句子:
EVENTDATE %{YEAR} %{MONTHNUM} %{MONTHDAY} %{TIME}
在上面,我们定义了 EVENDATE 为 YEAR, MONTHNUM, MONTHDAY 及 TIME 的组合。那么我们该如和应用上面的 custom patttern呢?
我们必须修改上面的 Grok pattern 为:
%{IP:ip} %{EVENTDATE:@timestamp} %{WORD:server} Process %{NUMBER:process_id} %{WORD:action}
从上面,我们可以看出来,我们的 EVENTDATE 起作用了。它正确地解析了我们的时间。
那么在我们实际的使用中,我们该如何地应用呢?
我们可以创建如下的一个命令:
POST /_ingest/pipeline/_simulate
{
"pipeline": {
"processors": [
{
"grok": {
"field": "message",
"patterns": [
"%{IP:ip} %{EVENTDATE:@timestamp} %{WORD:server} Process %{NUMBER:process_id} %{WORD:action}"
],
"pattern_definitions": {
"EVENTDATE": "%{YEAR} %{MONTHNUM} %{MONTHDAY} %{TIME}"
}
}
}
]
},
"docs": [
{
"_source": {
"message": "157.97.192.70 2019 09 29 00:39:02.912 myserver Process 107673 Initializing"
}
}
]
}
运行上面的命令:
{
"docs" : [
{
"doc" : {
"_index" : "_index",
"_type" : "_doc",
"_id" : "_id",
"_source" : {
"server" : "myserver",
"process_id" : "107673",
"@timestamp" : "2019 09 29 00:39:02.912",
"ip" : "157.97.192.70",
"action" : "Initializing",
"message" : "157.97.192.70 2019 09 29 00:39:02.912 myserver Process 107673 Initializing"
},
"_ingest" : {
"timestamp" : "2020-06-15T08:33:01.28191Z"
}
}
}
]
}
上面显示我们的日志被正确地解析并结构化。
另外一种方法是通过 set processor 来把上面的日期相关的字段来组成我们需要的 @timestamp 字段。
POST /_ingest/pipeline/_simulate
{
"pipeline": {
"processors": [
{
"grok": {
"field": "message",
"patterns": [
"%{IP:ip} %{YEAR} %{MONTHNUM} %{MONTHDAY} %{TIME} %{WORD:server} Process %{NUMBER:process_id} %{WORD:action}"
]
}
},
{
"set": {
"field": "@timestamp",
"value": "{{year}} {{month}} {{day}} {{time}}"
}
}
]
},
"docs": [
{
"_source": {
"message": "157.97.192.70 2019 09 29 00:39:02.912 myserver Process 107673 Initializing"
}
}
]
}
在上面,我们通过:
{
"set": {
"field": "@timestamp",
"value": "{{year}} {{month}} {{day}} {{time}}"
}
}
来把 @timestamp 进行定义,它组合了 year, month, day 及 time 的值。