本文主要介绍 Grok 的基本语法，以及一个简单的基于 Filebeat 日志进行结构化处理的例子。
Grok的初衷是为了解析异构文本并进行结构化处理，使其变得可查询。尤其适合 syslog logs，apache and other webserver logs，mysql logs，以及任何方便人类阅读而生的日志。Logstash 默认就预定义了 120 种日志模板，可以在这里下载 https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns ，可以通过 patterns_dir 配置进行添加。有几个经常使用的网站：
- Grok debugger，帮助校验正则是否正确，并提出修改建议；
- GrokConstructor，则更进一步，根据给定的日志，给出建议的正则；
- Grok-Patterns，全部的 Grok 内置 Pattern；
- Logstash-Filter Plugin-Grok，官方 Grok 文档；
我们用的最常用的还是 Grok debugger 和 Grok-Patterns 。在 Grok-Patterns 中查看如何去进行利用预定义模板进行匹配，然后在 Grok debugger 中进行校验我们写的基本正则是否正确。本文以 Filebeat 的日志解析为例，阐述如何使用 Logstash + Grok 进行日志的解析、过滤。
Grok 基础
Grok Pattern 的基本语法是 %{SYNTAX:SEMANTIC} 。 SYNTAX 是 Pattern 的名称。比如说 3.44 会被匹配匹配到 NUMBER Pattern，55.3.244.1 会被匹配到 IP Pattern。SEMANTIC 是对文本内容的定义，相当于 json 中的 key。比如，3.44 可以被定义为 event 的持续时间 duration，55.3.244.1 可以被定义为发起请求的 client 。对于这个例子来说，Grok 过滤器的定义如下：
%{NUMBER:duration} %{IP:client}
我们还可以使用强制转换，比如我们可以把 num 这个 SEMANTIC 转换成整型：
%{NUMBER:num:int}
举一个 http 请求日志的简单例子，日志文本如下：
55.3.244.1 GET /index.html 15824 0.043
我们对应的 Grok Pattern 应该如下：
%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}
Logstash 的配置如下：
input {
file {
path => "/var/log/http.log"
}
}
filter {
grok {
match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
}
}
最终解析出来的数据如下：
client: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
duration: 0.043
正则表达式
Grok 支持所有正则表达式，可以参考Oniguruma Regular Expressions。
自定义模板
为了满足预定义的 Pattern 无法满足需求的情况， Logstash 有自定义模板机制。首先，可以使用 Oniguruma 语法去捕捉一部分文本，并定义为 field：
(?<field_name>the pattern here)
比如，postfix 日志有 queue_id 字段包含了10-11位16进制的字符。我们可以通过以下正则进行捕捉：
(?<queue_id>[0-9A-F]{10,11})
或者，对于常用的自定义模板，我们可以创建自定义模板文件，并在配置里引用，达到复用的目的。创建一个名为 patterns 的模板，其中包含一个名为 extra 的文件，在该文件中，对 pattern 进行命名，然后编写该模板的正则表达式。例如，postfix queue id例子如下：
# contents of ./patterns/postfix:
POSTFIX_QUEUEID [0-9A-F]{10,11}
然后在 Logstash 配置中申明 patterns_dir 告诉 grok 插件去哪里寻找对应的自定义模板：
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>
其余一些概要性质的学习请自定看官方文档，这里就不进行翻译了：Synopsis。
日志分析实例
接下来我们对 Filebeat 日志进行分析，我们先来看看 Filebeat 日志长啥样儿：
2017-02-20T20:22:42+08:00 INFO Non-zero metrics in the last 30s: publish.events=1 libbeat.logstash.publish.write_bytes=391 libbeat.publisher.published_events=1 libbeat.logstash.call_count.PublishEvents=1 registrar.states.update=1 libbeat.logstash.published_and_acked_events=1 registrar.writes=1 libbeat.logstash.publish.read_bytes=6
2017-02-21T14:05:08+08:00 INFO Non-zero metrics in the last 30s: registrar.writes=1 registrar.states.update=1 libbeat.logstash.published_and_acked_events=1 libbeat.logstash.publish.read_byte
s=6 libbeat.logstash.call_count.PublishEvents=1 libbeat.publisher.published_events=1 libbeat.logstash.publish.write_bytes=383 publish.events=1
Grok 结构化处理
会发现大致分为几个部分： 日志时间、日志类型、日志消息、日志详情。修改 Logstash 的配置，以进行 grok 结构化：
input {
beats {
port => 5044
}
}
filter {
grok {
match => {
"message" => "^%{YEAR:year}-%{MONTHNUM:month}-%{MONTHDAY:day}T%{HOUR:hour}:?%{MINUTE:minute}(?::?%{SECOND:second})?%{ISO8601_TIMEZONE:timzone}? %{LOGLEVEL:log_level} %{DATA:log_msg}: %
{GREEDYDATA:log_msg_detail}$"
}
overwrite => ["message"]
}
}
output {
elasticsearch {
hosts => "172.16.134.2:9200"
manage_template => false
index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
document_type => "%{[@metadata][type]}"
}
}
重启 Logstash 后我们在 Kibana 中可以看到 Availiable Fileds 中出现了几个新增的 fileds：year month day hour minute second timezone log_level log_msg log_msg_detail 。接着我们需要在 Management -> Index Pattern 中执行刷新，把 fields 映射关系刷入缓存。 这个时候我们在 Index Pattern 中发现，类似 hour 这样的 fields 类型为 string，不符合我们的需求。怎么解决呢？我们利用 mutate 插件来实现：
filter {
grok {
match => {
"message" => "^%{YEAR:year}-%{MONTHNUM:month}-%{MONTHDAY:day}T%{HOUR:hour}:?%{MINUTE:minute}(?::?%{SECOND:second})?%{ISO8601_TIMEZONE:timzone}? %{LOGLEVEL:log_level} %{DATA:log_msg}: %{GREEDYDATA:log_msg_detail}$"
}
overwrite => ["message"]
}
mutate {
convert => {"year" => "integer"}
convert => {"month" => "integer"}
convert => {"day" => "integer"}
convert => {"hour" => "integer"}
convert => {"minute" => "integer"}
convert => {"second" => "integer"}
}
}
重启 Logstash 服务，但是刷新 Index Pattern ，还是看到对于 fields 的类型是 string。原因是 ES 只允许使用 API 去修改，我们需要调用 Mapping API 去修改 ES 中的数据类型。
小插曲 - 删除 Index Pattern 后的故障
然后有一个小插曲，愚昧的笔者点击了这个按钮：
删除 filebeat-* 这个 Index Pattern 之后 Kibana 的 Discover 页面无法进入，查看请求失败的日志：
{"error":{"root_cause":[{"type":"index_not_found_exception","reason":"no such index","resource.type":"index_or_alias","resource.id":"packetbeat-*","index_uuid":"_na_","index":"packetbeat-*"}],"type":"index_not_found_exception","reason":"no such index","resource.type":"index_or_alias","resource.id":"packetbeat-*","index_uuid":"_na_","index":"packetbeat-*"},"status":404}
找到社区一篇文章Kibana blank after deleting index找到了解决方案，在终端执行以下调用以强刷数据去除不一致：
curl -XPOST '172.16.134.2:9200/.kibana/_update_by_query?pretty&wait_for_completion&refresh' -H 'Content-Type: application/json' -d'
{
"script": {
"inline": "ctx._source.defaultIndex = null",
"lang": "painless"
},
"query": {
"term": {
"_type": "config"
}
}
}
'
这时候再在 Index Pattern 中把 filebeat-* 添加回，即可看到year month day hour minute second timezone log_level log_msg log_msg_detail 的数据类型变成了 number。
Kibana Discover 查询
Kibana 的 Discover 界面提供了超便利的查询，以下举几个常用的查询例子。主要的查询语法可以看这里。
根据时间 range 查询
我们已经结构化了时间维度的信息，比如我们需要查询 2017-02-21 19:30:00 ~ 2017-02-21 19:40:00 的日志数据，可以这么写，请注意 @timestamp filed 不是日志数据，是日志数据写入 ES 的时间戳：
year:2017 AND month:2 AND day:21 AND hour:19 AND minute: [30 TO 40]
按照日志类型 log_level 过滤
我们之前定义了 log_level filed，比如我们要查询 ERROR 类型的日志，很简单：
log_level:ERROR
太简单了，所以笔者不进行复述了，读者可以结合 Filter 进行各种查询组合得到自己想要的结果。然后如上一篇所述，可以定义为 Visulazation 加入 Dashboard。
Grok 注意事项
- Grok 在匹配失败的时候性能可能并不那么好
- 多留意
_grokparsefailures出现的频率和出现时候的性能 - 写正则的时候记得打锚点
- 不使用锚点的时候分层Grok处理的性能会比不分层的性能好，不过打了锚点的话两个都一样
- 多使用LogStash的性能监控功能，后续还可以拿来分析用
具体可以查看这篇中文译文：你真的了解Grok吗。
Q & A
Q: 在 Discover 中添加 filebeat-* Index Pattern 时报错：unable to fetch mapping do you have indices mapping the pattern。
A: 原因是数据没有流入 ES，无法构建索引模板。
