参考文档

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 等内网地址，会发现没有对应输出！