如何在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 {
  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 中可见和可搜索）。