系统日志到 logstash _grokparsefailure

Question

我有 ELK 5.4 版。我从许多网络机器（如 cisco、Juniper、Fortigate、F5）获取系统日志。

logstash.conf 看起来像：

input {
  tcp {
    port => 514
    type => syslog
  }
  udp {
    port => 514
    type => syslog
  }
}

filter {
  if [type] == "syslog" {
    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 { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
    if !("_grokparsefailure" in [tags]) {
      mutate {
        replace => [ "@source_host", "%{syslog_hostname}" ]
        replace => [ "@message", "%{syslog_message}" ]
      }
    }
    mutate {
      remove_field => [ "syslog_hostname", "syslog_message", "syslog_timestamp" ]
    }
  }
}

我收到很多 _grokparsefailure。也许有人有一个 logstash.conf 的例子，它会在没有 _grokparsefailure 的情况下理解所有机器的系统日志？

以下是一些日志示例：

2018-01-30T09:30:14.675+0200: %ADMIN-6-INFO: [local] recorded login for administrator: (user)
Jan 25 17:26:47.767: %LINEPROTO-5-UPDOWN: Line protocol on Interface FastEthernet8/11, changed state to down
Jan 30 01:13:43 SERVER_NAME lldpd[70354]: task_process_events: no read/accept method for (null) socket 0 <39>
Feb 1 09:14:43 SERVER_NAME rshd[64810]: user@user2 as root: cmd='/xxxx/xxxx/xxxx'

谢谢 TA

Answer 1

我用这个网站来测试模式：

http://grokconstructor.appspot.com/do/match

此网站用于查找预制图案：

https://github.com/elastic/logstash/blob/v1.4.2/patterns/grok-patterns

第一条消息是这样的模式：

%{TIMESTAMP_ISO8601:time}: %{SYSLOGPROG} \[%{WORD:level}\] %{GREEDYDATA:message}

对于第三条消息，你可以使用这个（如果你喜欢的话）：

%{MONTH:month} %{MONTHDAY:day} %{TIME:time} %{NOTSPACE:serverName} %{NOTSPACE:task}: %{GREEDYDATA:message}

在 logstash 中，您可以像这样使用它们：

filter{
  grok{
    match=>{"message" => <pattern1>}
  }
  if "_grokparsefailure" in [tags]{
    mutate{ remove_tag => [ "_grokparsefailure" ] }
    grok{
      match=>{"message" => <pattern2>}
    }
  }
  if "_grokparsefailure" in [tags]{
    mutate{ remove_tag => [ "_grokparsefailure" ] }
    grok{
      match=>{"message" => <pattern3>}
    }
  }
}

我们的想法是编写您的模式以匹配日志，如果一个不适合，您删除 _grokparsefailure 标记并尝试另一个。如果失败，则不会尝试下一个。这有点乏味，但我没有更好的解决方案。如果您可以按来源标记日志，则可以通过编写 if 条件来加快此过程。
我希望这对您有所帮助，或者至少给您一些指导。

Answer 2

我建议您将解析失败写入文件，以便您可以根据失败创建新模式。

output {
  if "_grokparsefailure" in [tags] {
    file {
      path => "/tmp/var/log/parse_failures.log
    }
  }
}

并非所有设备都遵循系统日志 RFC，因此几乎不可能列出所有可用模式。

系统日志到 logstash _grokparsefailure

syslog to logstash _grokparsefailure

tags

syslog

filter

logstash

elastic-stack