Logstash的过滤插件及其应用

Posted 2023-03-07

一、Logstash的过滤插件

数据从源传输到存储库的过程中，Logstash 过滤器能够解析各个事件，识别已命名的字段以构建结构，并将它们转换成通用格式，以便进行更强大的分析和实现商业价值。 Logstash 能够动态地转换和解析数据，不受格式或复杂度的影响。
      常见的过滤插件的用途:

• 利用 Grok 从非结构化数据中派生出结构
• 从 IP 地址破译出地理坐标
• 利用 useragent 从请求中分析操作系统、设备类型
• 简化整体处理，不受数据源、格式或架构的影响

1.1 Grok插件

Grok 是一个过滤器插件，可帮助描述日志格式的结构。有超过200种 grok模式抽象概念，如IPv6地址，UNIX路径和月份名称。

参考网站：​​http://grokdebug.herokuapp.com/discover?#​​

使用正则表达式%COMBINEDAPACHELOG可将nginx，tomcat等的日志转换为json格式。当然也可以在kibana界面实现

      如果用logstash执行，则logstash对应的配置文件nginx_grok_stdout.conf的内容如下：

input 
  file 
    path => "/var/log/nginx/access.log"
    type => "nginx-accesslog"
    start_position => "beginning"
    stat_interval => "3" #每3s检测文件是否更新
  

filter 
  #将nginx日志格式化为json格式
  grok 
    match => 
      "message" => "%COMBINEDAPACHELOG" #将message字段转化为指定的Json格式
    
  

output 
  stdout 
    codec => rubydebug
  

执行：/usr/share/logstash/bin/logstash –f nginx_grok_stdout.conf

1.2 Geoip插件

kibana支持绘制热点地图。所谓热点地图，即根据访问日志的客户端ip地址信息，在一张地图上画出访问客户端的来源，并根据不同的来源密度给予不同的颜色。这样，就可以直观的体现我们服务的客户端地域来源。而Geoip可以根据ip地址可以得到该ip地址所对应的地理位置，便于后续的地理数据分析
      该插件有如下信息：

 "message" => "183.60.92.253",
 "@version" => "1",
 "@timestamp" => "2014-08-07T10:32:55.610Z",
 "host" => "raochenlindeMacBook-Air.local",
 "geoip" => 
 "ip" => "183.60.92.253",
 "country_code2" => "CN",
 "country_code3" => "CHN",
 "country_name" => "China",
 "continent_code" => "AS",
 "region_name" => "30",
 "city_name" => "Guangzhou",
 "latitude" => 23.11670000000001,
 "longitude" => 113.25,
 "timezone" => "Asia/Chongqing",
 "real_region_name" => "Guangdong",
 "location" => [
 [0] 113.25,
 [1] 23.11670000000001
 ]
 

      可以用fields选项指定所需要的字段。

filter 
  geoip 
    fields => ["city_name", "continent_code", "country_code2", "country_code3", "country_name", "dma_code", "ip", "latitude", "longitude", "postal_code", "region_name", "timezone"]
  

注意：geoip 插件的 “source” 字段可以是任一处理后的字段，比如 “ip”，但是字段内容却需要小心！geoip 库内只存有公共网络上的ip信息，查询不到结果的，会直接返回 null，而 logstash 的 geoip 插件对 null 结果的处理是：不生成对应的 geoip.字段。

参考​​http://www.openi.cn/read/logstash-best-practice-cn/filter-geoip.md​​​

vim geoip.conf
input 
  file 
    path => "/var/log/nginx/access.log"
    type => "nginx-accesslog"
    start_position => "beginning"
    stat_interval => "3"
  

filter 
  #将nginx日志格式化为json格式 
  grok 
    match => 
      "message" => "%COMBINEDAPACHELOG"
    
  
  #以上面提取clientip字段为源,获取地域信息
  geoip 
    source => "clientip" 
    #fields => ["continent_code", "country_name", "region_name", "latitude", "longitude", "timezone"]
  

output 
  elasticsearch
    hosts=>["192.168.131.12:9200"]
    index=>"logstash-geoip-%+YYYY.MM.dd"
  

执行：/usr/share/logstash/bin/logstash -f geoip.conf
      同时这里仿照nginx的日志格式，添加一段带外网ip地址的日志到nginx日志：

echo 106.11.248.144 - - [02/Mar/2023:21:26:12 +0800] "GET / HTTP/1.1" 304 0 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (Khtml, like Gecko) Chrome/110.0.0.0 Safari/537.36 Edg/110.0.1587.57" >> /var/log/nginx/access.log

Kibana得到的索引如下：

之后创建一个坐标地图版本的可视化

填写对应的索引，之后就能得到一张地域热图

1.3 Date插件

Date插件可以将日志中的日期字符串解析为日志类型。然后替换@timestamp 字段(此字段默认为当前放入logstash的时间而非日志本身的时间)或指定的其他字段。选项说明如下：

• match 类型为数组，用于指定日期匹配的格式，可以以此指定多种日期格式
• target 类型为字符串，用于指定赋值的字段名，默认是@timestamp
• timezone 类型为字符串，用于指定时区域

vim date.conf
input 
  file 
    path => "/var/log/nginx/access.log"
    type => "nginx-accesslog"
    start_position => "beginning"
    stat_interval => "3"
  

filter 
  #将nginx日志格式化为json格式
  grok 
    match => 
      "message" => "%COMBINEDAPACHELOG"
    
  
  geoip 
    source => "clientip"
    #fields => ["continent_code", "country_name", "region_name", "latitude", "longitude", "timezone"]
  
  #解析date日期格式为: 3/Jul/2022:20:07:27 +0800,
  date 
    match => ["timestamp", "yyyy-MM-dd HH:mm:ss Z"]
    #将时间覆盖原有的@timestamp字段
    target => "@timestamp"
    timezone => "Asia/Shanghai"
  

output 
  stdout 
    codec => rubydebug
  

执行：/usr/share/logstash/bin/logstash –f date.conf

跟之前的测试方法类似，还是追加一条格式相仿的nginx日志，看测试结果：

      由于@timestamp自身使用的UTC时区(北京时间减去8小时)，因此显示的结果正常

1.4 Useragent插件

Useragent 插件可以根据请求中的 user-agent 字段，解析出浏览器设备、操作系统等信息​

Vim useragent.conf
input 
  file 
    path => "/var/log/nginx/access.log"
    type => "nginx-accesslog"
    start_position => "beginning"
    stat_interval => "3"
  

filter 
  #将nginx日志格式化为json格式 
  grok 
    match => 
      "message" => "%COMBINEDAPACHELOG"
    
  
  #解析date日期格式为: 3/Jul/2022:20:07:27 +0800
  date 
    match => ["timestamp", "yyyy-MM-dd HH:mm:ss Z"]
    #将时间覆盖原有的@timestamp字段
    target => "@timestamp"
    timezone => "Asia/Shanghai"
  
  #提取agent字段，进行解析
  useragent 
    #指定从哪个字段获取数据
    source => "agent"
    #转换后的新字段
    target => "useragent"
  

output 
  stdout 
    codec => rubydebug
  

执行：/usr/share/logstash/bin/logstash –f useragent.conf

在本机访问nginx服务后，日志输出如下：

1.5 Mutate插件

Mutate 插件主要是对字段进行、类型转换、删除、替换、更新等操作，可以使用以下函数：

• remove_field：删除字段
• split：字符串切割，相当于awk取列
• add_field：添加字段
• convert：类型转换
• gsub：字符串替换

1.5.1 remove_field 删除字段

vim remove_field.conf
input 
  file 
    path => "/var/log/nginx/access.log"
    type => "nginx-accesslog"
    start_position => "beginning"
    stat_interval => "3"
  

filter 
  #将nginx日志格式化为json格式 
  grok 
    match => 
      "message" => "%COMBINEDAPACHELOG"
    
  
  #解析date日期格式为: 3/Jul/2022:20:07:27 +0800
  date 
    match => ["timestamp", "yyyy-MM-dd HH:mm:ss Z"]
    #将时间覆盖原有的@timestamp字段
    target => "@timestamp"
    timezone => "Asia/Shanghai"
  
  #mutate--删除操作
  mutate 
    remove_field => ["headers","message", "agent"]
  

output 
  stdout 
    codec => rubydebug
  

再次执行，就会发现显示的日志中没了headers、message、agent字段：

1.5.2 Split切割&add_field添加字段

mutate 中的 split 字符切割， 指定字段做为分隔符,生成新的字段名
      示例: 1000|提交订单|2023-03-0522:37:21​

Vim add_field.conf
input 
  http 
    port =>6666
  

filter 
  #mutate 切割操作
  mutate 
    #字段分隔符
    split =>  "message" => "|" 
    #添加字段,将message的列表的第0个元素添加字段名user_id.以此类推
    add_field => 
      "user_id" => "%[message][0]" 
      "action" => "%[message][1]"
      "time" => "%[message][2]"
    
    #删除无用字段
    remove_field => ["headers","message"]
  

output 
  stdout 
    codec => rubydebug
  

执行：/usr/share/logstash/bin/logstash -f add_field.conf
      用curl提交日志，可看到如下的输出信息：

curl -XPOST -d 1000|提交订单|2023-03-05 22:41:30 192.168.131.14:6666