Logstash 配置总结

一时失言乱红尘 2022-06-02 23:18 167阅读 0赞

\#整个配置文件分为三部分：input,filter,output  
\#参考这里的介绍 https://www.elastic.co/guide/en/logstash/current/configuration-file-structure.html  
input \{  
  \#file可以多次使用，也可以只写一个file而设置它的path属性配置多个文件实现多文件监控  
  file \{  
    \#type是给结果增加了一个属性叫type值为"<xxx>"的条目。这里的type，对应了ES中index中的type，即如果输入ES时，没有指定type，那么这里的type将作为ES中index的type。  
    type => "apache-access"   
    path => "/apphome/ptc/Windchill\_10.0/Apache/logs/access\_log\*"  
    \#start\_position可以设置为beginning或者end，beginning表示从头开始读取文件，end表示读取最新的，这个也要和ignore\_older一起使用。  
    start\_position => beginning  
    \#sincedb\_path表示文件读取进度的记录，每行表示一个文件，每行有两个数字，第一个表示文件的inode，第二个表示文件读取到的位置（byteoffset）。默认为$HOME/.sincedb\*  
    sincedb\_path => "/opt/logstash-2.3.1/sincedb\_path/access\_progress"  
    \#ignore\_older表示了针对多久的文件进行监控，默认一天，单位为秒，可以自己定制，比如默认只读取一天内被修改的文件。  
    ignore\_older => 604800  
    \#add\_field增加属性。这里使用了$\{HOSTNAME\}，即本机的环境变量，如果要使用本机的环境变量，那么需要在启动命令上加--alow-env。  
    add\_field => \{"log\_hostname"=>"$\{HOSTNAME\}"\}  
    \#这个值默认是\\n 换行符，如果设置为空""，那么后果是每个字符代表一个event  
    delimiter => ""  
    \#这个表示关闭超过（默认）3600秒后追踪文件。这个对于multiline来说特别有用。... 这个参数和logstash对文件的读取方式有关，两种方式read tail，如果是read  
    close\_older => 3600  
    coodec => multiline \{  
      pattern => "^\\s"  
      \#这个negate是否定的意思，意思跟pattern相反，也就是不满足patter的意思。  
\#      negate => ""  
      \#what有两个值可选 previous和next，举例说明，java的异常从第二行以空格开始，这里就可以pattern匹配空格开始，what设置为previous意思是空格开头这行跟上一行属于同一event。另一个例子，有时候一条命令太长，当以\\结尾时表示这行属于跟下一行属于同一event，这时需要使用negate=>true，what=>'next'。  
      what => "previous"  
      auto\_flush\_interval => 60  
    \}  
  \}  
  file \{   
    type => "methodserver-log"   
    path => "/apphome/ptc/Windchill\_10.0/Windchill/logs/MethodServer-1604221021-32380.log"   
    start\_position => beginning   
    sincedb\_path => "/opt/logstash-2.3.1/sincedb\_path/methodserver\_process"  
\#    ignore\_older => 604800  
  \}  
\}  
filter\{  
  \#执行ruby程序，下面例子是将日期转化为字符串赋予daytag  
  ruby \{  
    code => "event\['daytag'\] = event.timestamp.time.localtime.strftime('%Y-%m-%d')"  
  \}  
  \# if \[path\] =~ "access" \{\} else if \[path\] =~ "methodserver" \{\} else if \[path\] =~ "servermanager" \{\} else \{\} 注意语句结构  
  if \[path\] =~ "MethodServer" \{ \#z这里的=~是匹配正则表达式  
    grok \{  
      patterns\_dir => \["/opt/logstash-2.3.1/patterns"\] \#自定义正则匹配  
\#      Tue 4/12/16 14:24:17: TP-Processor2: hirecode---->77LS  
      match => \{ "message" => "%\{DAY:log\_weekday\} %\{DATE\_US:log\_date\} %\{TIME:log\_time\}: %\{GREEDYDATA:log\_data\}"\}  
    \}  
    \#mutage是做转换用的  
    mutate \{   
      replace => \{ "type" => "apache" \} \#替换属性值  
      convert => \{ \#类型转换  
        "bytes" => "integer" \#例如还有float  
        "duration" => "integer"  
        "state" => "integer"  
      \}  
    \#date主要是用来处理文件内容中的日期的。内容中读取的是字符串，通过date将它转换为@timestamp。参考https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html\#plugins-filters-date-match  
\#    date \{  
\#      match => \[ "logTime" , "dd/MMM/yyyy:HH:mm:ss Z" \]  
\#    \}  
  \}else if \[type\] in \['tbg\_qas','mbg\_pre'\] \{ \# if ... else if ... else if ... else结构  
  \}else \{  
    drop\{\} \# 将event丢弃  
  \}  
\}  
output \{  
  stdout\{ codec=>rubydebug\} \# 直接输出，调试用起来方便  
  \# 输出到redis  
  redis \{  
    host => '10.120.20.208'  
    data\_type => 'list'  
    key => '10.99.201.34:access\_log\_2016-04'  
  \}  
  \# 输出到ES  
  elasticsearch \{  
    hosts =>"192.168.0.15:9200"  
    index => "%\{sysid\}\_%\{type\}"  
    document\_type => "%\{daytag\}"  
  \}