集中式日志分析平台 - ELK Stack - Logstash

作者: 大数据之心 | 来源:发表于2017-02-21 20:49 被阅读1367次

本文主要介绍 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 后的故障

然后有一个小插曲，愚昧的笔者点击了这个按钮：

Remove_index_pattern.png

删除 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，无法构建索引模板。

参考文献

Do you grok Grok?

网友评论

0溺水的鱼O:楼主研究的非常深入啊，最近ELK搭建正好用上，NB。
大数据之心:@0溺水的鱼O 咱们可以愉快的做同事吗~！

本文标题：集中式日志分析平台 - ELK Stack - Logstash

本文链接：https://www.haomeiwen.com/subject/rzouwttx.html

延伸阅读

深度阅读

您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！