logstash - 如何在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.过滤器.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. 输出.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导致事件被丢弃。它永远不会出现在您的输出中:

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 中可见和可搜索)。