1、Grok正则插件

Grok 是 Logstash 最重要的插件。你可以在 grok 里预定义好命名正则表达式，在稍后(grok参数或者其他正则表达式里)引用它。

Grok 支持把预定义的 grok 表达式 写入到文件中，官方提供的预定义 grok 表达式见：https://github.com/logstash/logstash/tree/v1.4.2/patterns。

正则表达式语法

运维工程师多多少少都会一点正则。你可以在 grok 里写标准的正则，像下面这样：

\s+(?<request_time>\d+(?:\.\d+)?)\s+

小贴士：这个正则表达式写法对于 Perl 或者 Ruby 程序员应该很熟悉了，Python 程序员可能更习惯写 (?P<name>pattern)，没办法，适应一下吧。

现在给我们的配置文件添加第一个过滤器区段配置。配置要添加在输入和输出区段之间(logstash 执行区段的时候并不依赖于次序，不过为了自己看得方便，还是按次序书写吧)：

input{ stdin{} }
filter{
    grok {
        match => {
                "message" => "%{INT:int}"
        }
    }
}

output{ stdout{ codec => rubydebug }  }

运行 logstash 进程然后输入 "begin 123"，你会看到类似下面这样的输出：

{
         "message" => "begin 123",
        "@version" => "1",
      "@timestamp" => "2014-08-09T11:55:38.186Z",
            "host" => "raochenlindeMacBook-Air.local",
    "int" => "123"
}

这个时候int成功获取到了数值，但是会发现这是个字符串的，可以做如下修改改变类型

filter{
    grok {
        match => {
                "message" => "%{INT:int:int}"  # 改变之后的
              #"message" => “%{INT:int}”  # 改变之前的
        }
    }
}

我们可以清除的看到两者的区别
grok 表达式的打印复制格式的完整语法是下面这样的：

%{PATTERN_NAME:capture_name:data_type}

小贴士：data_type 目前只支持两个值：int 和 float。

重新运行进程然后可以得到如下结果：

{
         "message" => "begin 123",
        "@version" => "1",
      "@timestamp" => "2014-08-09T11:55:38.186Z",
            "host" => "raochenlindeMacBook-Air.local",
    "int" => 123
}

这次获取到的结果是一个int类型的值

自定义正则

实际应用中，我们需要处理各种各样的数据，大部分都可以通过官方提供的grok表达式直接调用

# 调用的语法为    
%{PATTERN_NAME:capture_name:data_type}

但是有些特殊情况这些无法满足我们的需求，这个时候我们就需要自己编写正则了，正则的格式如下

(?<name>正则表达式)  # 这只是定义正则的一种方式  还有其他方式

最佳实战

实际运用中，我们需要处理各种各样的日志文件，如果你都是在配置文件里各自写一行自己的表达式，就完全不可管理了。所以，我们建议是把所有的 grok 表达式统一写入到一个地方。然后用 filter/grok 的 patterns_dir选项来指明。

下面是一个实例
配置文件

input { stdin{}  }
filter{
    grok{
        patterns_dir => "/root/patherns"  # 只想自定义文件所在的目录
        match => {
            "message" => "%{JIAO:user} %{JIAOX:int}"  # 进行过滤
        }
        remove_field => "message"  # 过滤成功之后就删除message数据
    }
}

output{ stdout{ codec=>rubydebug } }

/root/patherns/jiao 自定义文件

# 前面为字段名  后面为字段名对应的正则或者引用grok表达式
JIAO %{USERNAME}    # 引用grok表达式
JIAOX \d+  # 自定义正则

在命令行输入 alice 123456 , 查看控制台输出结果

{
           "int" => "123456",
          "user" => "alice",
      "@version" => "1",
          "host" => "a656fc35ea31",
       "message" => "alice 123456",
    "@timestamp" => 2019-11-23T09:04:58.276Z
}

可以看到文件内自定义的正则也可以正常使用

多项选择

有时候我们会碰上一个日志有多种可能格式的情况。这时候要写成单一正则就比较困难，或者全用 | 隔开又比较丑陋。这时候，logstash 的语法提供给我们一个有趣的解决方式。

文档中，都说明 logstash/filters/grok 插件的 match 参数应该接受的是一个 Hash 值。但是因为早期的 logstash 语法中 Hash 值也是用 [] 这种方式书写的，所以其实现在传递 Array 值给 match 参数也完全没问题。所以，我们这里其实可以传递多个正则来匹配同一个字段：

match => [
    "message", "(?<request_time>\d+(?:\.\d+)?)",
    "message", "%{SYSLOGBASE} %{DATA:message}",
    "message", "(?m)%{WORD}"
]

logstash 会按照这个定义次序依次尝试匹配，到匹配成功为止。虽说效果跟用 | 分割写个大大的正则是一样的，但是可阅读性好了很多。

我强烈建议每个人都要使用 Grok Debugger 来调试自己的 grok 表达式。

时间处理

date过滤器用于解析字段中的日期，然后使用该日期或时间戳作为事件的logstash时间戳。

例如，syslog事件通常有这样的时间戳：

"Apr 17 09:32:01"

你应该使用MMM dd HH:mm:ss的日期格式来解析这个。

日期过滤器对于事件的排序和对旧数据的回填特别重要，如果在你的事件中没有得到正确的日期，那么以后搜索它们可能会出现顺序不对。

如果没有这个过滤器，logstash将根据第一次看到事件（在输入时）的时间（如果时间戳还没有在事件中设置）选择一个时间戳，例如，对于文件输入，时间戳被设置为每次读取的时间。

通过日期过滤器过滤之后，时间戳会按照日志文件内日志发生的时间，例如nginx访问日志，这样就可以知道什么时候，哪个ip访问了我们的服务器，具体访问了哪台服务器，以什么方式访问等等信息

日期实例

配置文件

input{ stdin{} }
filter{
    grok {
        match => {
                "message" => "(?<time>.*)"  # 定义字段名  time 
        } 
    }
    date {
        match => ["time","yyyy-MM-dd HH:mm:ss"]   # 上面提供的字段名time
        # 以及要在命令行的时间对应的时间格式  

     }

}

注意日期格式一定要和日志内的时间格式或者自己输入的时间格式一致

执行命令，输入时间，查看对应的结果

2000-11-11 10:10:10  # 这是输入的时间
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/awesome_print-1.7.0/lib/awesome_print/formatters/base_formatter.rb:31: warning: constant ::Fixnum is deprecated
{
          "time" => "2000-11-11 10:10:10",  # time字段
          "host" => "a656fc35ea31",
      "@version" => "1",
    "@timestamp" => 2000-11-11T10:10:10.000Z,  # 可以看到时间戳变成了命令行输入的时间
       "message" => "2000-11-11 10:10:10"
}

下面是用来解析日期和时间的方法：
y 年
- yyyy => 全年数字，例如：2015。
- yy => 两位数的年，例如：15代表2015年。
M 一年中的月份
- M => 最小位数的月份，例如：1为1月，12为12月。
- MM => 两位数的月份，如果需要将用零填充，例如：01为1月，12为12月。
- MMM => 缩写月文本，例如：Jan为1月，注意：使用的语言取决于你的语言环境，查看locale设置如何更改语言。
- MMMM => 全月的文本，例如：January，注意：使用的语言取决于你的语言环境。
d 一月中的天
- d => 最小位数的天，例如：1为一个月的第一天。
- dd => 两位数的天，如果需要将用零填充，例如：01为一个月的第一天。
H 一天中的小时（24小时时钟）
- H => 最小位数的小时，例如：0为午夜。
- HH => 两位数的小时，如果需要，零填充，例如：00为午夜。
m 一小时中的分钟（每小时60分钟）
- m => 最小位数的分钟，例如：0。
- mm => 两位数的分钟，如果需要，零填充，例如：00。
s 一分钟中的秒（每分钟60秒）
- s => 最小位数的秒，例如：0。
- ss => 两位数的秒，如果需要，零填充，例如：00。
S 每秒最大精度的分数为毫秒(SSS)，超过这个，追加零
- S => 十分之一秒，例如：0表示亚秒值012。
- SS => 百分之一秒，例如：01为亚秒值01。
- SSS => 千分之一秒，例如：012为亚秒值012。
Z 时区偏移或标识
- Z => 时区偏移结构为HHmm（Zulu/UTC的小时和分钟偏移），例如：-0700。
- ZZ => 时区偏移结构为HH:mm（在小时和分钟偏移量之间的冒号），例如：-07:00。
- ZZZ => 时区的标识，例如：America/Los_Angeles，注意：有效的id列在Joda.org可用时区页面上。
z 时区名称，无法解析时区名称（z）。
w 一年中的周
- w => 最小位数的周，例如：1。
- ww => 两位数的周，如果需要，零填充，例如：01。
D 一年中的天
e 一周中的星期（数字）
E 一周中的星期（文本）
- E, EE, EEE => 一周中缩短的星期，例如：Mon，Tue，Wed，Thu，Fri，Sat，Sun，注意：它的实际语言取决于你的语言环境。
- EEEE => 一周的星期全称，例如：Monday，Tuesday，...，注意：它的实际语言取决于你的语言环境。
  查看更多点击此处

GeoIP 地址查询归类

GeoIP 是最常见的免费 IP 地址归类查询库，同时也有收费版可以采购。GeoIP 库可以根据 IP 地址提供对应的地域信息，包括国别，省市，经纬度等，对于可视化地图和区域统计非常有用。

配置示例

input { stdin{ }  }

filter {
    geoip {

        source => "message"  # 对命令行输入的ip进行地址归类查询
        # source是固定的，对应的字段可以是任一处理后的字段，  
        # 注意  只有公网ip才可以
    }

}


output { stdout { codec => rubydebug  } }

运行结果

183.60.92.253
{
       "message" => "183.60.92.253",
         "geoip" => {
             "longitude" => 113.25,
         "country_code3" => "CN",
           "region_name" => "Guangdong",
         "country_code2" => "CN",
              "location" => {
            "lon" => 113.25,
            "lat" => 23.1167
        },
          "country_name" => "China",
           "region_code" => "GD",
              "timezone" => "Asia/Shanghai",
              "latitude" => 23.1167,
        "continent_code" => "AS",
                    "ip" => "183.60.92.253"
    },
          "host" => "a656fc35ea31",
    "@timestamp" => 2019-11-23T10:39:59.990Z,
      "@version" => "1"
}

配置说明

GeoIP 库数据较多，如果你不需要这么多内容，可以通过 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.location 是 logstash 通过 latitude 和 longitude额外生成的数据。所以，如果你是想要经纬度又不想重复数据的话，应该像下面这样做：

filter { 
    geoip { 
        fields => ["city_name", "country_code2", "country_name", "latitude", "longitude", "region_name"]   # 指定显示的字段
        remove_field => ["[geoip][latitude]", "[geoip][longitude]"]  
# 在显示字段的基础 上移除某些字段
    } 
}

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

所以读者在测试时，如果使用了诸如 127.0.0.1, 172.16.0.1, 182.168.0.1, 10.0.0.1 等内网地址，会发现没有对应输出！

Logstash_过滤插件