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 语义从字符串转换为整数。目前唯一支持的转换是 intfloat

使用语义和语法,我们可以从一个示例日志中提取有用的字段,如下面这个虚构的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的主要内容,如果未能解决你的问题,请参考以下文章

使用Logstash filter grok过滤日志文件

Logstash 常用 filter 插件介绍

Elastic Stack Logstash基础用法

Logstash语法常用案例解析

logstash使用filter删除不需要的日志

使用Logstash filter grok过滤日志文件