一、背景

我们的应用程序通常每天都会产生非常多的日志，这些日志大多都是一个字符串的格式，那么我们如果想从中提取一些有用的信息（比如：请求的时间、日志的级别等），那么应该如果实现呢？

二、解决思路

针对以上的问题，我们可以通过正则表达式来匹配我们的日志内容，从而达到提取到有用的数据。而 logstash的grok正好可以帮助我们做到这种事。

三、前置知识

grok插件为我们提供了大概120种可用的模式。可以简单理解，grok提供好了120种定义好的可用这则。
1. 比如： INT 对应的正则(?:[+-]?(?:[0-9]+)) 表示正负数字。
2. grok 提供好的可用模式。 https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns
grok语法的格式
1. %{SYNTAX:SEMANTIC}
  1. SYNTAX 表示需要匹配文本的格式。比如：文本内容: 123 可以被 INT匹配到。
  2. SEMANTIC表示 SYNTAX 匹配到文本后，将内容保存到那个字段中。比如：%{INT:age} 则匹配到后的age字段有值。
自定义grok模式(custom patterns)
1. 语法格式：(?<field_name>the pattern here)
2. 比如: (?<userName>[a-zA-Z]{3,5}) 用户名只能是 3到5位的字母
grok调试网站
1. 可以在此网站调试我们的grok表达式是否编写正确 http://grokdebug.herokuapp.com/
grok解析失败和超时会增加如下标签
1. 解析失败，会在生成数据中的tags中介增加 _grokparsefailure标签
2. 解析超时，会在生成数据中的tags中增加_groktimeout标签

四、实现步骤

1、准备测试数据

[9708] 2021-05-13 11:14:51.873 [http-nio-8080-exec-1] INFO  org.springframework.web.servlet.DispatcherServlet#initServletBean:547 -Completed initialization in 1 ms
[9708] 2021-05-13 11:14:51.910 [http-nio-8080-exec-1] ERROR com.huan.study.LogController#showLog:32 -请求:[/showLog]发生了异常
java.lang.ArithmeticException: / by zero
    at com.huan.study.LogController.showLog(LogController.java:30)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

2、编写`grok`表达式

(?m)^\[%{INT:pid}\]%{SPACE}%{TIMESTAMP_ISO8601:createTime}%{SPACE}\[%{DATA:threadName}\]%{SPACE}%{LOGLEVEL:LEVEL}%{SPACE}%{JAVACLASS:javaClass}#(?<methodName>[a-zA-Z_]+):%{INT:linenumber}%{SPACE}-%{GREEDYDATA:msg}

注意⚠️：

1、如果要匹配多行文本，比如java中的异常堆栈,则在表达式前需要加上(?m)。

2、使用\可以进行转义。

3、其中的(?<methodName>[a-zA-Z_]+) methodName 为我们自定义的正则。

3、编写 logstash pipeline文件

input {
    file {
        id => "mutate-id"
        path => ["/Users/huan/soft/elastic-stack/logstash/logstash/pipeline.conf/filter-grok/*.log"]
        start_position => "beginning"
        sincedb_path => "/Users/huan/soft/elastic-stack/logstash/logstash/pipeline.conf/filter-grok/sincedb.db"
        codec => multiline {
             pattern => "^\[+"
             negate => "true"
             what => "previous"
             charset => "UTF-8"
             auto_flush_interval => 2
        }
    }
}

filter {

    grok {
        match => {
            "message" => "(?m)^\[%{INT:pid}\]%{SPACE}%{TIMESTAMP_ISO8601:createTime}%{SPACE}\[%{DATA:threadName}\]%{SPACE}%{LOGLEVEL:LEVEL}%{SPACE}%{JAVACLASS:javaClass}#(?<methodName>[a-zA-Z_]+):%{INT:linenumber}%{SPACE}-%{GREEDYDATA:msg}"
        }
    }
}

output {
    stdout {
        codec => rubydebug {

        }
    }
}