如何在logstash中进行两个条件检查并编写更好的配置文件

Question

我正在使用logstash 1.4.2,

我在客户端日志服务器中有logstash-forwarder.conf,就像这样

{
    "network": {
      "servers": [ "xxx.xxx.xxx.xxx:5000" ],
      "timeout": 15,
      "ssl ca": "certs/logstash-forwarder.crt"
    },
  "files": [
       {
          "paths": [ "/var/log/messages" ],
          "fields": { "type": "syslog" }
        },
        {

          "paths": [ "/var/log/secure" ],
          "fields": { "type": "linux-syslog" }
        }
         ]
}

================================================== =======

在logstash服务器中

1. filter.conf

filter {
  if [type] == "syslog" {
date {
        locale => "en"
        match => ["syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
        timezone => "Asia/Kathmandu"
        target => "@timestamp"
        add_field => { "debug" => "timestampMatched"}
   }
    grok {
      match => { "message" => "\[%{WORD:messagetype}\]%{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }
    syslog_pri { }
  }
if [type] == "linux-syslog" {
date {
        locale => "en"
        match => ["syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
        timezone => "Asia/Kathmandu"
        target => "@timestamp"
        add_field => { "debug" => "timestampMatched"}
   }
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }
    syslog_pri { }
mutate { replace => [ "syslog_timestamp", "%{syslog_timestamp} +0545" ] }

  }
}

================================================== =====

2. output.conf

output {
    if [messagetype] == "WARNING" {
 elasticsearch { host => "xxx.xxx.xxx.xxx" }
  stdout { codec => rubydebug }
}

 if [messagetype] == "ERROR" {
 elasticsearch { host => "xxx.xxx.xxx.xxx" }
  stdout { codec => rubydebug }
}

if [type] == "linux-syslog" {
 elasticsearch { host => "xxx.xxx.xxx.xxx" }
  stdout { codec => rubydebug }
}

}

================================================== =====

我希望所有日志都从/ var/log/secure转发而只有/ var/log/messages中的ERROR和WARNING日志,我知道这不是一个好的配置.我希望有人能告诉我更好的方法.

Answer 1

我更喜欢在过滤器块中做出有关事件的决定.我的输入和输出块通常很简单.从那里,我看到两个选项.

使用下拉过滤器

该drop过滤器会导致事件被丢弃.它永远不会成为你的输出:

filter {
    #other processing goes here

    if [type] == "syslog" and [messagetype] not in ["ERROR", "WARNING"] {
        drop {}
    }
}

这样做的好处是它非常简单.

缺点是该事件刚刚被取消.它根本不会输出.哪个好,如果这就是你想要的.

使用标签

许多过滤器允许您添加标签,这对于在插件之间进行通信决策非常有用.您可以附加一个标记,告诉输出块将事件发送到ES:

filter {
    #other processing goes here

    if [type] == "linux-syslog" or [messagetype] in ["ERROR", "WARNING"] {
        mutate {
            add_tag => "send_to_es"
        }
    }
}

output {
    if "send_to_es" in [tags] {
        elasticsearch {
            #config goes here
        }
    }
}

这样做的好处是它可以实现精细控制.

这样做的缺点是它需要更多的工作,并且您的ES数据最终会受到一点污染(标签将在ES中可见并可搜索).