Logstash：Grok filter 入门

> 本文由 [简悦 SimpRead](http://ksria.com/simpread/) 转码， 原文地址 [blog.csdn.net](https://blog.csdn.net/UbuntuTouch/article/details/105922198)

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

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

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

在之前我的另外一篇文章 “[Elasticsearch：Elastic 可观测性 - 数据结构化及处理](https://elasticstack.blog.csdn.net/article/details/105790798 "Elasticsearch：Elastic可观测性 - 数据结构化及处理")”，我也介绍了如何使用 ingest node 来对数据进行加工和格式化。

Logstash 日志解析的 Grok 模式示例

[Logstash 日志解析的 Grok 模式示例_哔哩哔哩_bilibili](https://www.bilibili.com/video/BV1g3411W7fe "Logstash 日志解析的 Grok 模式示例_哔哩哔哩_bilibili")

什么是 Grok?
=========

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

Grok 是如何工作的？
============

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

内置了超过 200 种 [Logstash 模式](https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html "Logstash 模式")，用于过滤 AWS，Bacula，Bro，Linux-Syslog 等中的单词，数字和日期等项目。 如果找不到所需的模式，则可以编写自己的自定义模式。 还有多个匹配模式的选项，可简化表达式的编写以捕获日志数据。

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

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

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

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

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

一个 Logstash Grok 例子
===================

这个 grok 调试工具是开始构建 grok 过滤器的好方法：[Grok Debugger](https://grokdebug.herokuapp.com/ "Grok Debugger")。你也可以在 Kibana 中找到 grok 的调试工具：

![](https://img-blog.csdnimg.cn/20200504195832508.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70)

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

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

```
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}"
      ]
   }
```

比如针对这条信息：

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"

![](https://img-blog.csdnimg.cn/20200504200321781.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70)

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

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

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

![](https://img-blog.csdnimg.cn/20200504200734239.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70)

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

常见的例子
=====

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

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

![](https://img-blog.csdnimg.cn/20200504201353462.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70)

在 Logstash 中，我们应该使用如下的 grok filter:

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

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  {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
}
```

![](https://img-blog.csdnimg.cn/20200504201800945.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70)

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 {
      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}"
      ]
   }
```

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

![](https://img-blog.csdnimg.cn/20200504202133253.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1VidW50dVRvdWNo,size_16,color_FFFFFF,t_70)

总结
==

Logstash grok 只是在将日志转发到 Elasticsearch 之前可以应用于你的日志的一种过滤器。 由于 grok 在测井管道中起着至关重要的作用，因此它也是最常用的过滤器之一。