搭建ELK(ElasticSearch+Logstash+Kibana)日志分析系统(四) logstash codec和filter 配置

2017-09-13 by qq_24879495

这一节来介绍codec插件和filter插件,然后关于logstash的常见配置就介绍完毕了。下一节介绍通过redis扩展logsatsh。


1、logsatsh codec (编码插件)插件

Codec 来自 Coder/decoder 两个单词的首字母缩写,Logstash 不只是一个input | filter | output 的数据流,而是一个input | decode | filter | encode | output 的数据流,codec 就是用来 decode、encode 事件的。简单说,就是在logstash读入的时候,通过codec编码解析日志为相应格式,从logstash输出的时候,通过codec解码成相应格式。rubydebug 就是一种 codec!虽然它一般只会用在 stdout 插件中,作为配置测试或者调试的工具。下面介绍几种常见的格式。

1)JSON 格式

如果事件数据是JSON格式,就可以通过 codec=>json 来解析输入的日志,配置示例:

    input {      file {          path => "/home/husen/test.log"          codec => "json" ## 以JSON格式读取日志      }  }  


2)Multiline 格式

有些日志事件在打印的时候,可能会有多行,比如java 错误日志。而logstash默认是一行一行读取的,这个时候就要用到multiline格式了。Multiline通过正则来匹配多行日志,常见是只匹配开头的符号。比如

<2017-08-31 10:26:56,699> hello world

<2017-08-31 10:26:56,699>hello world

这两行日志都是以 < 开头的,那么久用正则匹配 /<就可以了。配置示例如下:

    intput{      file {      path =>[          "/home/husen/log.log"      ]            type => "portal"      start_position => "beginning"      sincedb_path => "/dev/null" #从头读      codec => multiline {      pattern => "^/<"  ##匹配每行的开头      negate => true    ## 是否开启正则匹配      what => "previous" ##未匹配的内容向前合并 如果为next,则向后合并      }  }  


2)plain 格式

主要用于事件之间没有分隔的纯文本,可自定义格式,匹配实例:

    output {      stdout {         codec => plain {              format => "%{@timestamp} %{message} %{type}" ## format可不写,不写为空,写了就定义格式          }      }  }  


2、logstash filter (过滤器)插件

过滤器扩展了进入过滤器的原始数据,可以进行复杂的逻辑处理,如匹配正则、增删改字段等,这里介绍常见的几个

1)date - 时间过滤器

可以用来转换时间的格式,并应用到@timestamp 字段上,配置示例
    filter {      grok {          match => ["message", "%{HTTPDATE:logdate}"] ##匹配日志中的时间      }        date {      match => [ "logdate", "yyyy-MM-dd HH:mm:ss,SSS" ] ##转换格式      locale => "cn" ##确定本地时区      target => "@timestamp" ##重写到@timestamp字段      }  }  

2)grok - 正则过滤器

Grok 是 Logstash 最重要的插件。你可以在 grok 里预定义好命名正则表达式,然后匹配日志中的内容,并以JSON格式展现。

配置示例:

    filter {      grok {          match => [ ## 可以定义多个正则              "message", "%{IP:logIp} .* %{USERNAME:username}",   ## 匹配具有ip地址和用户名的日志              "message", "%{MAC:logMac}"  ##匹配具有mac地址的日志          ]      }  }  

3)geoip - IP 过滤器

GeoIP 是最常见的免费 IP 地址归类查询库。GeoIP 库可以根据 IP 地址提供对应的地域信息,包括国别,省市,经纬度等,

对于可视化地图和区域统计非常有用。配置示例:

    filter {      geoip {          source => "message" #匹配message      }  }    ## 结果   "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          ]      }  }  

4)json - JSON过滤器

直接解析日志中的具有JSON格式的数据,配置示例

    filter {      json {          source => "message" ##数据来源          target => "json_result" ##匹配后的命名 如果不打算使用多层结构的话,删掉 target 配置即可      }  }    ##输出结果  {      "@version": "1",      "@timestamp": "2017-09-13T08:11:33.000Z",      "host": "192.168.1.1",      "message": "{/"username/":husen,/"password/":/"123456/"}",      "json_result": {          "username": husen,          "password": "123456"      }  }  


5)kv - KV过滤器

    kv {         prefix => "url_"         source => "url_args"         field_split => "&"         include_keys => [ "uid", "cip" ]         remove_field => [ "url_args", "uri", "request" ]     }  # 上例即表示,除了 url_uid 和 url_cip 两个字段以外,其他的 url_* 都不保留。  


6)mutate -  数据修改过滤器

mutate提供了丰富的基础类型数据处理能力。包括类型转换,字符串处理和字段处理等。配置示例

a. 类型转换
    ## 类型转换 可以设置的转换类型包括:"integer","float" 和 "string"  filter {      mutate {          convert => ["request_time", "float"]      }  }


b. 字符串处理 - split 切割
    filter {      mutate {          split => ["message", "|"]      }  }    ##随意输入一串以|分割的字符,比如 "123|321|adfd|dfjld*=123",可以看到如下输出:    "message" => [      [0] "123",      [1] "321",      [2] "adfd",      [3] "dfjld*=123"  ]  

c. 字符串处理 - join 加入
    filter {      mutate {          split => ["message", "|"]      }        mutate {          join => ["message", ","]      }  }    ##我们在之前已经用 split 割切的基础再 join 回去    "message" => "123,321,adfd,dfjld*=123",  
d.  字符串处理 - merge 合并
    filter {      mutate {          split => ["message", "|"]      }        mutate {          merge => ["message", "message"]      }  }    ##我们在之前已经用 split 割切的基础再合并,会复制一份     "message" => [          [0] "123",          [1] "321",          [2] "adfd",          [3] "dfjld*=123",          [4] "123",          [5] "321",          [6] "adfd",          [7] "dfjld*=123"      ],  

e. 字符串处理 - strip 去掉空格
    mutate {          strip => ["type"] ## 去掉字符串前后空格,如果有的话      }  

f.  字段处理 

rename 重命名 
  update  更新某个字段的内容。如果字段不存在,不会新建
 replace 作用和 update 类似,但是当字段不存在的时候,它会起到 add_field 参数一样的效果,自动添加新的字段

    filter {      mutate {          rename => ["syslog_host", "host"]      }  }  

最新更新:

第七城市

栏目导航(关闭)