WEBKT

Logstash Grok Filter 高级用法与技巧：从入门到精通

6 0 0 0

Logstash Grok Filter 高级用法与技巧：从入门到精通

什么是 Grok？

为什么要用 Grok？

Grok 的基本语法

自定义正则表达式模式

Grok Debugger

多个 Match 规则

覆盖字段

实例演示

性能优化

总结

Logstash Grok Filter 高级用法与技巧：从入门到精通

你好！相信你已经对 Logstash 有了一定的了解，并且可能已经在使用它来处理你的日志数据了。Logstash 强大的地方之一就是它的 filter 插件，而 grok 又是其中最常用的 filter 之一。今天我们就来深入探讨一下 grok 的高级用法和技巧，让你能够更加灵活高效地处理各种复杂的日志格式。

什么是 Grok？

简单来说，grok 就是一个帮你把非结构化日志数据转换成结构化数据的工具。它通过预定义的或者自定义的正则表达式模式来匹配日志中的各个字段，然后把这些字段提取出来，变成一个个键值对，方便你后续的分析和处理。

为什么要用 Grok？

想象一下，你有一堆杂乱无章的日志数据，比如这样的：

127.0.0.1 - - [11/Dec/2023:00:01:45 -0800] "GET /apache_pb.gif HTTP/1.0" 200 2326

如果你想从中提取出 IP 地址、时间、请求方法、URL、状态码、响应大小这些信息，你会怎么做？手动写正则表达式？太麻烦了！而且不同的日志格式，正则表达式还不一样，想想都头疼。

这时候，grok 就派上用场了。你可以用一个简单的 grok 表达式来搞定：

%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response} %{NUMBER:bytes}

这样，grok 就会自动帮你把这些字段提取出来，变成这样的结构化数据：

{
  "clientip": "127.0.0.1",
  "ident": "-",
  "auth": "-",
  "timestamp": "11/Dec/2023:00:01:45 -0800",
  "verb": "GET",
  "request": "/apache_pb.gif",
  "httpversion": "1.0",
  "response": "200",
  "bytes": "2326"
}

是不是很方便？

Grok 的基本语法

Grok 的基本语法很简单，就是：

%{PATTERN:FieldName}

其中：

  • PATTERN 是预定义的或者自定义的正则表达式模式。
  • FieldName 是你给这个字段起的名字。

Logstash 内置了很多常用的正则表达式模式，你可以直接拿来用。比如：

  • %{NUMBER:bytes}：匹配数字，并将结果存储在 bytes 字段中。
  • %{IP:clientip}：匹配 IP 地址，并将结果存储在clientip字段。
  • %{WORD:method}：匹配单个单词, 并将结果存储在method字段。
  • %{HTTPDATE:timestamp}：匹配 HTTP 的日期格式。

你可以在这里找到 Logstash 内置的所有模式：https://github.com/logstash-plugins/logstash-patterns-core/blob/main/patterns/ecs-v1/grok-patterns

自定义正则表达式模式

如果内置的模式不能满足你的需求，你还可以自定义正则表达式模式。有两种方法：

  1. 在 grok 表达式中直接定义：

    (?<field_name>the pattern here)

    例如，你想匹配一个由字母和数字组成的 ID：

    (?<id>[a-zA-Z0-9]+)

  2. 创建自定义模式文件：

    • 创建一个名为 patterns 的文件夹 (名字不能变).
    • 在该文件夹下创建一个任意名称的文本文件, 例如 my_patterns.
    • 在该文件中，每一行定义一个模式，格式为：pattern_name regex_pattern

    例如，在 my_patterns 文件中添加：

    MY_CUSTOM_ID [a-zA-Z0-9]+

    然后在 Logstash 配置文件中，使用 patterns_dir 选项指定 patterns 文件夹的路径：

    		filter {
    		  grok {
    		    patterns_dir => ["./patterns"]
    		    match => { "message" => "%{MY_CUSTOM_ID:id}" }
    		  }
    		}

Grok Debugger

调试 grok 表达式是件很头疼的事情，特别是当表达式很复杂的时候。不过，Logstash 提供了一个非常方便的工具：Grok Debugger。 它是 Kibana 的一部分，你可以通过 Kibana 的 Dev Tools 来访问它。

使用 Grok Debugger 的步骤：

  1. 打开 Kibana。
  2. 进入 Dev Tools。
  3. 选择 Grok Debugger。
  4. 输入你的日志样本和 grok 表达式。
  5. 点击 "Simulate"。

Debugger 会显示匹配的结果，以及每个字段的值。如果匹配失败，它会告诉你哪里出错了。通过这个工具你可以快速调试你的 grok 表达式。

多个 Match 规则

有时候，你的日志可能有多种不同的格式。这时候，你可以使用多个 match 规则来处理：

filter {
  grok {
    match => {
      "message" => [
        "%{PATTERN1:field1}",
        "%{PATTERN2:field2}",
        "%{PATTERN3:field3}"
      ]
    }
  }
}

Logstash 会按照顺序尝试每一个 match 规则，直到找到一个匹配的为止。如果所有的规则都不匹配，那么这条日志就不会被解析，你可以在grok里配置tag_on_failure决定打什么标签。

覆盖字段

默认情况下，如果 grok 表达式中提取的字段已经存在，Logstash 会保留原来的值。如果你想用 grok 提取的值覆盖原来的值，可以使用 overwrite 选项：

filter {
  grok {
    match => { "message" => "%{IP:clientip}" }
    overwrite => [ "clientip" ]
  }
}

实例演示

下面我们通过几个实例来演示如何使用 grok 解析不同格式的日志数据。

1. 解析 Apache 访问日志

127.0.0.1 - - [11/Dec/2023:00:01:45 -0800] "GET /apache_pb.gif HTTP/1.0" 200 2326

filter {
  grok {
    match => { "message" => "%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] \"%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:response} %{NUMBER:bytes}" }
  }
}

2. 解析 Nginx 错误日志

2023/12/11 10:38:43 [error] 1234#1234: *1 client closed connection while waiting for request, client: 192.168.1.1, server: example.com, request: "GET / HTTP/1.1", upstream: "http://127.0.0.1:8080/", host: "example.com"

filter{
    grok{
        match => {"message" => "%{TIMESTAMP_ISO8601:timestamp} \[%{LOGLEVEL:loglevel}\] %{POSINT:pid}#%{NUMBER:threadid}: \*%{NUMBER:connectionid} %{GREEDYDATA:errormessage}, client: %{IP:clientip}, server: %{HOSTNAME:server}(, request: \"%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\")?(, upstream: \"%{URI:upstream}\")?(, host: \"%{HOSTNAME:host}\")?"}
    }
}

3. 解析自定义日志

假设你的应用程序输出这样的日志：

[INFO] 2023-12-11 12:34:56 - User 'john' logged in from 192.168.1.100

filter {
  grok {
    match => { "message" => "\[%{LOGLEVEL:level}\] %{TIMESTAMP_ISO8601:timestamp} - User '%{USERNAME:username}' logged in from %{IP:ip}" }
  }
}

4. 多行日志

有些日志是多行的，比如 Java 的异常堆栈信息。这种情况下，你需要先使用 multiline codec 或者 filter 将多行日志合并成一行，然后再使用 grok 解析。
例如，以下是一个java异常：

java.lang.NullPointerException
        at com.example.myproject.MyClass.myMethod(MyClass.java:123)
        at com.example.myproject.AnotherClass.anotherMethod(AnotherClass.java:456)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

我们可以通过以下配置，先将多行合并，再解析：

input {
  stdin {
    codec => multiline {
      pattern => "^\\s"
      what => "previous"
    }
  }
}
 
filter {
  grok {
    match => { "message" => "(?m)%{JAVACLASS:class}\\n%{GREEDYDATA:stacktrace}" }
  }
}

这里，(?m)开启了多行模式。

性能优化

当处理大量日志数据时，grok 的性能可能会成为一个瓶颈。下面是一些优化 grok 性能的技巧：

  1. 使用最具体的模式： 尽量使用最具体的模式，避免使用过于通用的模式，比如 DATA。因为通用的模式需要匹配更多的字符，会降低性能。
  2. 避免回溯： 正则表达式的回溯会导致性能急剧下降。尽量避免使用复杂的正则表达式，特别是那些包含多个 .* 或者 .+ 的表达式。
  3. 锚定表达式： 使用 ^ 和 $ 来锚定表达式的开头和结尾，可以避免不必要的匹配。
  4. 重用模式： 如果你需要多次使用同一个模式，可以将它定义成一个自定义模式，然后重复使用，避免重复编译。
  5. 减少match数量： match规则是顺序执行，如果前面的match匹配成功，后面的match则不会执行，那么把最常用的match放在前面会减少匹配次数。

总结

Grok 是 Logstash 中一个非常强大且灵活的工具，可以帮助你轻松地解析各种复杂的日志格式。通过掌握 grok 的高级用法和技巧，你可以更加高效地处理你的日志数据，从中提取出有价值的信息。希望这篇文章对你有所帮助，如果你还有其他问题，欢迎随时交流！

希望你在使用 Logstash 的过程中越来越得心应手，成为一个真正的日志处理专家！

技术老炮儿 LogstashGrok日志处理

评论点评

打赏赞助
sponsor

感谢您的支持让我们更好的前行

分享

QRcode

https://www.webkt.com/article/8353