Logstash过滤器插件filter

╰半橙微兮° 2023-03-02 04:51 15阅读 0赞

#### 一、grok正则捕获 ####

grok是一个十分强大的Logstash filter插件，它可以通过正则解析任意文本，将非结构化日志数据弄成结构化和方便查询的结构。它是目前Logstash中解析非结构化日志数据最好的方式

**grok的语法规则：**  ```%{语法：语义}```
    “语法”指的是匹配的模式。例如使用NUMBER模式可以匹配出数字，IP模式则会匹配出127.0.0.1这样的IP地址
    **举例：**
    原始数据：172.16.213.132 [07/Feb/2018:16:24:19 +0800] "GET /HTTP/1.1" 403 5039
    1）过滤IP

input {
            stdin {
       }
    }
    filter{
         grok{
                match => {"message" => "%{IPV4:ip}"}
       }
    }
    output {
             stdout {
         }
    }

**运行结果：**

[root@:172.31.22.29 /etc/logstash/conf.d]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/l2.conf
    Sending Logstash logs to /var/log/logstash which is now configured via log4j2.properties172.16.213.132 [07/Feb/2018:16:24:19 +0800]"GET /HTTP/1.1" 403 5039　　　　　　#手动输入此行信息
    {
           "message" => "172.16.213.132 [07/Feb/2018:16:24:19 +0800]\"GET /HTTP/1.1\" 403 5039",
                "ip" => "172.16.213.132",
          "@version" => "1",
              "host" => "ip-172-31-22-29.ec2.internal",
        "@timestamp" => 2019-01-22T09:48:15.354Z
    }

2）过滤时间戳

filter{
            grok{
               match => {"message" => "%{IPV4:ip}\ \[%{HTTPDATE:timestamp}\]"}
        }
    }

运行结果：

[root@:172.31.22.29 /etc/logstash/conf.d]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/l2.conf
    Sending Logstash logs to /var/log/logstash which is now configured via log4j2.properties172.16.213.132 [07/Feb/2018:16:24:19 +0800]"GET /HTTP/1.1" 403 5039　　　　手动输入此行信息
    {
          "@version" => "1",
         "timestamp" => "07/Feb/2018:16:24:19 +0800",
        "@timestamp" => 2019-01-22T10:16:14.205Z,
           "message" => "172.16.213.132 [07/Feb/2018:16:24:19 +0800]\"GET /HTTP/1.1\" 403 5039",
                "ip" => "172.16.213.132",
              "host" => "ip-172-31-22-29.ec2.internal"
    }

可以看到我们已经过滤成功了，在配置文件中grok其实是使用正则表达式来进行过滤的。我们做个小实验，比如我现在在例子中的数据ip后面添加两个“-”。如图所示：

172.16.213.132 - - \[07/Feb/2018:16:24:19 +0800\] “GET /HTTP/1.1” 403 5039

那么此时在配置文件中我就需要这样子来写：

filter{
         grok{
             match => {"message" => "%{IPV4:ip}\ -\ -\ \[%{HTTPDATE:timestamp}\]"}
         }
    }

那么此时在match行我就要匹配两个“-”，否则grok就不能正确匹配数据，从而不能解析数据

运行结果：

[root@:172.31.22.29 /etc/logstash/conf.d]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/l2.conf
    Sending Logstash logs to /var/log/logstash which is now configured via log4j2.properties172.16.213.132 - - [07/Feb/2018:16:24:19 +0800] "GET /HTTP/1.1" 403 5039　　　　　　#手动输入此行内容，然后按下enter键。
    {
        "@timestamp" => 2019-01-22T10:25:46.687Z,
                "ip" => "172.16.213.132",
           "message" => "172.16.213.132 - - [07/Feb/2018:16:24:19 +0800] \"GET /HTTP/1.1\" 403 5039",
         "timestamp" => "07/Feb/2018:16:24:19 +0800",
          "@version" => "1",
              "host" => "ip-172-31-22-29.ec2.internal"
    }

这时候我们就得到了信息，我这里是匹配IP和时间，当然你也可以直接匹配时间即可：

filter{
            grok{
                match => {"message" => "\ -\ -\ \[%{HTTPDATE:timestamp}\]"}
         }
    }

这个时候我们更加能理解grok使用正则匹配数据了  
**注意：正则中，匹配空格和中括号要加上转义符**

3）过滤出报文头信息  
首先来写匹配的正则模式

filter{
            grok{
                match => {"message" => "\ %{QS:referrer}\ "}
         }
    }

运行结果：

[root@:172.31.22.29 /etc/logstash/conf.d]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/l2.conf
    Sending Logstash logs to /var/log/logstash which is now configured via log4j2.properties172.16.213.132 - - [07/Feb/2018:16:24:19 +0800] "GET /HTTP/1.1" 403 5039
    {
        "@timestamp" => 2019-01-22T10:47:37.127Z,
           "message" => "172.16.213.132 - - [07/Feb/2018:16:24:19 +0800] \"GET /HTTP/1.1\" 403 5039",
          "@version" => "1",
              "host" => "ip-172-31-22-29.ec2.internal",
          "referrer" => "\"GET /HTTP/1.1\""
    }

#### 二、date插件 ####

在上面我们有个例子是讲解timestamp字段，表示取出日志中的时间。但是在显示的时候除了显示你指定的timestamp外，还有一行是@timestamp信息，这两个时间是不一样的，@timestamp表示系统当前时间。两个时间并不是一回事，在ELK的日志处理系统中，@timestamp字段会被elasticsearch用到，用来标注日志的生产时间，如此一来，日志生成时间就会发生混乱，要解决这个问题，需要用到另一个插件，即date插件，这个时间插件用来转换日志记录中的时间字符串，变成Logstash::Timestamp对象，然后转存到@timestamp字段里面

接下来我们在配置文件中配置一下：

filter{
          grok{
               match => {"message" => "\ -\ -\ \[%{HTTPDATE:timestamp}\]"}
          }
          date{
               match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
        }
    }

注意：时区偏移量需要用一个字母Z来转换。还有这里的“dd/MMM/yyyy”，你发现中间是三个大写的M，没错，这里确实是三个大写的M，我尝试只写两个M的话，转换失败

运行结果：

[root@:172.31.22.29 /etc/logstash/conf.d]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/l2.conf
    Sending Logstash logs to /var/log/logstash which is now configured via log4j2.properties172.16.213.132 - - [07/Feb/2018:16:24:19 +0800] "GET /HTTP/1.1" 403 5039        #手动输入此行信息
    {
              "host" => "ip-172-31-22-29.ec2.internal",
         "timestamp" => "07/Feb/2018:16:24:19 +0800",
        "@timestamp" => 2018-02-07T08:24:19.000Z,
           "message" => "172.16.213.132 - - [07/Feb/2018:16:24:19 +0800] \"GET /HTTP/1.1\" 403 5039",
          "@version" => "1"
    }

会发现@timestamp时间转换成功，还有一点就是在时间少8个小时

上面有几个例子已经讲到了date的用法。date插件对于排序事件和回填旧数据尤其重要，它可以用来转换日志记录中的时间字段，变成Logstash：：timestamp对象，然后转存到@timestamp字段里面。

为什么要使用这个插件呢？

*  一方面由于Logstash会给收集到的每条日志自动打上时间戳（即@timestamp），但是这个时间戳记录的是input接收数据的时间，而不是日志生成的时间（因为日志生成时间与input接收的时间肯定不同），这样就可能导致搜索数据时产生混乱
 *  另一方面，在上面那段rubydebug编码格式的输出中，@timestamp字段虽然已经获取了timestamp字段的时间，但是仍然比北京时间晚了8个小时，这是因为在Elasticsearch内部，对时间类型字段都是统一采用UTC时间，而日志统一采用UTC时间存储，是国际安全、运维界的一个共识。其实这并不影响什么，因为ELK已经给出了解决方案，那就是在Kibana平台上，程序会自动读取浏览器的当前时区，然后在web页面自动将UTC时间转换为当前时区的时间

如果你要解析你的时间，你要使用字符来代替，用于解析日期和时间文本的语法使用字母来指示时间（年、月、日、时、分等）的类型。以及重复的字母来表示该值的形式。在上面看到的"dd/MMM/yyy:HH:mm:ss Z"，他就是使用这种形式，我们列出字符的含义：  
![在这里插入图片描述][format_png_pic_center]  
那我们是依据什么写出“dd/MMM/yyy:HH:mm:ss Z”这样子的形式的呢？比如上面的试验数据是：

172.16.213.132 - - \[07/Feb/2018:16:24:19 +0800\] “GET /HTTP/1.1” 403 5039  
现在我们想转换时间，那就要写出"dd/MMM/yyy:HH:mm:ss Z"，你发现中间有三个M，你要是写出两个就不行了，因为我们查表发现两个大写的M表示两位数字的月份，可是我们要解析的文本中，月份则是使用简写的英文，所以只能去找三个M。还有最后为什么要加上个大写字母Z，因为要解析的文本中含有“+0800”时区偏移，因此我们要加上去，否则filter就不能正确解析文本数据，从而转换时间戳失败

#### 三、remove\_field的用法 ####

remove\_field的用法也是很常见的，其作用是去重，在前面的例子中你也看到了，不管是我们要输出什么样子的信息，都是有两份数据，即message里面是一份，HTTPDATE或者IP里面也有一份，这样子就造成了重复，过滤的目的就是筛选出有用的信息，重复的不要，因此我们看看如何去重呢？

1）输出IP为例

filter{
            grok{
                  match => {"message" => "%{IP:ip_address}"}
                  remove_field => ["message"]
            } 
    }

运行结果：

[root@:172.31.22.29 /etc/logstash/conf.d]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/l5.conf
    Sending Logstash logs to /var/log/logstash which is now configured via log4j2.properties172.16.213.132 - - [07/Feb/2018:16:24:19 +0800] "GET /HTTP/1.1" 403 5039　　　　　　#手动输入此行内容并按enter键
    {
        "ip_address" => "172.16.213.132",
              "host" => "ip-172-31-22-29.ec2.internal",
          "@version" => "1",
        "@timestamp" => 2019-01-22T12:16:58.918Z
    }

这时候你会发现没有之前显示的那个message的那一行信息了。因为我们使用remove\_field把他移除了

#### 四、数据修改mutate插件 ####

mutate插件是logstash另一个非常重要的插件，它提供了丰富的基础类型数据处理能力，包括重命名、删除、替换、修改日志事件中的字段。我们这里举几个常用的mutate插件：字段类型转换功能covert、正则表达式替换字段功能gsub、分隔符分隔字符串为数值功能split、重命名字段功能rename、删除字段功能remove\_field

1）字段类型转换convert  
先修改配置文件

filter{
            grok{
                  match => {"message" => "%{IPV4:ip}"}
                  remove_field => ["message"]
            }
            mutate{
                  convert => ["ip","string"]
            }
    }

或者这样子写也行，写法区别较小：

filter{
            grok{
                 match => {"message" => "%{IPV4:ip}"}
                 remove_field => ["message"]
            }
            mutate{
                  convert => {
                       "ip" => "string"
                 }
          }
    }

运行结果：

[root@:172.31.22.29 /etc/logstash/conf.d]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/l6.conf
    Sending Logstash logs to /var/log/logstash which is now configured via log4j2.properties172.16.213.132 - - [07/Feb/2018:16:24:9 +0800] "GET /HTTP/1.1" 403 5039
    {
        "@timestamp" => 2019-01-23T04:13:55.261Z,
                "ip" => "172.16.213.132",
              "host" => "ip-172-31-22-29.ec2.internal",
          "@version" => "1"
    }

在这里的ip行中，效果可能不太明显，但是确实是已经转化成string模式了

2）正则表达式替换匹配字段  
gsub可以通过正则表达式替换字段中匹配到的值，但是这本身只对字符串字段有效

首先修改配置文件

filter{
            grok{
                 match => {"message" => "%{QS:referrer}"}
                 remove_field => ["message"]
            }
            mutate{
                 gsub => ["referrer","/","-"]
          }
    }

运行结果：

172.16.213.132 - - [07/Feb/2018:16:24:9 +0800] "GET /HTTP/1.1" 403 5039
    {
              "host" => "ip-172-31-22-29.ec2.internal",
        "@timestamp" => 2019-01-23T05:51:30.786Z,
          "@version" => "1",
          "referrer" => "\"GET -HTTP-1.1\""
    }

很不错，确实对QS的部分的分隔符换做横杠了

3）分隔符分隔字符串为数组  
split可以通过指定的分隔符分隔字段中的字符串为数组

首先修改配置文件

filter{
            mutate{
                    split => ["message","-"]
                    add_field => ["A is lower case :","%{[message][0]}"]
            }
    }

这里的意思是对一个字段按照“-”进行分隔为数组

运行结果：

a-b-c-d-e-f-g　　　　　　　　　　　　#手动输入此行内容，并按下enter键。
    {
        "A is lower case :" => "a",
                  "message" => [
            [0] "a",
            [1] "b",
            [2] "c",
            [3] "d",
            [4] "e",
            [5] "f",
            [6] "g"
        ],
               "host" => "ip-172-31-22-29.ec2.internal",
               "@version" => "1",
               "@timestamp" => 2019-01-23T06:07:18.062Z
    }

4）重命名字段  
rename可以实现重命名某个字段的功能

filter{
            grok{
                  match => {"message" => "%{IPV4:ip}"}
                  remove_field => ["message"]
            }
            mutate{
                    convert => {
                        "ip" => "string"
                    }
                    rename => {
                        "ip"=>"IP"
                    }
            }
    }

rename字段使用大括号\{\}括起来，其实我们也可以使用中括号达到同样的目的

mutate{
            convert => {
                  "ip" => "string"
            }
            rename => ["ip","IP"]
     }

运行结果：

172.16.213.132 - - [07/Feb/2018:16:24:9 +0800] "GET /HTTP/1.1" 403 5039　　　　　　#手动输入此内容
    {
          "@version" => "1",
          "@timestamp" => 2019-01-23T06:20:21.423Z,
          "host" => "ip-172-31-22-29.ec2.internal",
          "IP" => "172.16.213.132"
    }

5）删除字段，这个不多说，我们上面已经有例子了  
6）添加字段add\_field  
添加字段多用于split分隔中

filter {
      mutate {
        split => ["message", "|"]
          add_field => {
            "timestamp" => "%{[message][0]}"　　　　
            }　
        }
    }

**添加字段后，该字段会与@timestamp一样同等格式显示出来**

[format_png_pic_center]: /images/20230208/e8c9dedf1d894b938b8d9761cba7e285.png