Kibana 在选择时间戳和收到_at 时未显示正确的数据答案

【问题标题】：Kibana not showing the correct data while choosing with timestamp & received_atKibana 在选择时间戳和收到_at 时未显示正确的数据
【发布时间】：2019-08-04 08:38:42
【问题描述】：

我在logstash.conf 文件下方，我看到数据处理正确，但今天我看到非常奇怪的问题，noi-syslog 的索引没有显示正确的syslog_timestamp。

input {
  file {
    path => [ "/scratch/rsyslog/*/messages.log" ]
    start_position => beginning
    sincedb_path => "/dev/null"
    max_open_files => 64000
    type => "noi-syslog"
  }
  file {
    path => [ "/scratch/rsyslog_CISCO/*/network.log" ]
    start_position => beginning
    sincedb_path => "/dev/null"
    max_open_files => 64000
    type => "apic_logs"
  }
}

filter {
  if [type] == "noi-syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp } %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      remove_field => [ "host", "path" ]
    }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
 }
}
  if [type] == "apic_logs" {
    grok {
      match => { "message" => "%{CISCOTIMESTAMP:syslog_timestamp} %{CISCOTIMESTAMP} %{SYSLOGHOST:syslog_hostname} (?<prog>[\w._/%-]+) %{SYSLOG5424SD:fault_code}%{SYSLOG5424SD:fault_state}%{SYSLOG5424SD:crit_info}%{SYSLOG5424SD:log_severity}%{SYSLOG5424SD:log_info} %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      remove_field => [ "host", "path" ]
    }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
  }
 }
}
output {
        if [type] == "noi-syslog" {
        elasticsearch {
                hosts => "noida-elk:9200"
                manage_template => false
                index => "noi-syslog-%{+YYYY.MM.dd}"
                document_type => "messages"
  }
 }
}

output {
        if [type] == "apic_logs" {
        elasticsearch {
                hosts => "noida-elk:9200"
                manage_template => false
                index => "apic_logs-%{+YYYY.MM.dd}"
                document_type => "messages"
  }
 }
}

`apic_logs` & `noi-syslog` 的索引：

$ curl -s -XGET http://127.0.0.1:9200/_cat/indices?v |  grep apic_logs
green  open   noi-syslog-2019.03.13           Fz1Rht65QDCYCshmSjWO4Q   5   1    6845696            0      2.2gb            1gb
green  open   noi-rmlog-2019.03.13            W_VW8Y1eTWq-TKHAma3DLg   5   1     148613            0     92.6mb           45mb
green  open   apic_logs-2019.03.13            pKz61TS5Q-W2yCsCtrVvcQ   5   1    1606765            0    788.6mb        389.7mb

使用@timesatmp 选择apic_logs 索引时Kibana 页面正确显示所有字段，但对于Linux 系统日志索引noi-syslog 无法正常工作。

noi-syslog not显示所有字段，同时选择@timestamp但显示_grokparsefailure标签，另一个事实是在选择received_at 987654340 @它显示所有字段但没有显示及时显示数据。

下面是用received_at选择的图片

下面是用@timestamp选择的图片

在 logstash 日志中：

# tail -5 log-cohort_deprecation.log
[2019-03-13T20:16:29,112][WARN ][o.e.d.a.a.i.t.p.PutIndexTemplateRequest] [noida-elk.cadence.com] Deprecated field [template] used, replaced by [index_patterns]
[2019-03-13T20:16:30,548][WARN ][o.e.d.a.a.i.t.p.PutIndexTemplateRequest] [noida-elk.cadence.com] Deprecated field [template] used, replaced by [index_patterns]
[2019-03-13T20:19:45,935][WARN ][o.e.d.a.a.i.t.p.PutIndexTemplateRequest] [noida-elk.cadence.com] Deprecated field [template] used, replaced by [index_patterns]
[2019-03-13T20:19:48,644][WARN ][o.e.d.a.a.i.t.p.PutIndexTemplateRequest] [noida-elk.cadence.com] Deprecated field [template] used, replaced by [index_patterns]
[2019-03-13T20:20:13,069][WARN ][o.e.d.a.a.i.t.p.PutIndexTemplateRequest] [noida-elk.cadence.com] Deprecated field [template] used, replaced by [index_patterns]

系统内存使用情况：

             total       used       free     shared    buffers     cached
Mem:         32057      31794        263          0        210      18206
-/+ buffers/cache:      13378      18679
Swap:       102399        115     102284

总内存 32GB 我已为每个 Elastic 和 Logstash 分配了 8GB，我怀疑这是否会导致问题。

删除`grokparsefailure` 标签的解决方法：

filter {
  if [type] == "noi-syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      remove_field => [ "host", "path" ]
    }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
 }
 if "_grokparsefailure" in [tags] {
         drop { }
 }
}

1- 或者一个替代方案只是一个想法..

filter {
  if [type] == "noi-syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      remove_field => [ "host", "path" ]
    }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
 }
  if "_grokparsefailure" in [tags] {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp } %{SYSLOGHOST:syslog_hostname} %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      remove_field => [ "host", "path" ]
   }
  }     
 }
}

2- 或者另一种选择只是一个想法..

filter {
  if [type] == "noi-syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      remove_field => [ "host", "path" ]
    }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
 }
  elif "_grokparsefailure" in [tags] {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp } %{SYSLOGHOST:syslog_hostname} %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      remove_field => [ "host", "path" ]
   }
   else "_grokparsefailure" in [tags] {
         drop { }
  }
 }

【问题讨论】：

问题出在您的grok 过滤器中，它没有正确解析您的消息，您要用作时间戳的字段syslog_timestamp 取决于消息的正确解析，如果grok 不起作用，该字段没有被创建。您需要使用给您错误的消息的文本示例和您的 grok 过滤器来更新您的问题，以便人们可以尝试重现您的管道。

标签： elasticsearch logstash elastic-stack logstash-configuration kibana-6

【解决方案1】：

这里的问题是noi-syslog 类型的示例中的消息彼此不同，并且您的grok 过滤器仅适用于第一个，当grok 无法解析消息时它会添加一个标签命名为_grokparsefailure。

您在grok 上工作的第一个示例是：

Mar 13 15:55:02 hostname /usr/bin/crontab[32708]: (root) LIST (root)

grok 失败的第二个例子是：

Mar 12 11:01:02 hostname run-parts(/etc/cron.hourly)[3970 starting mcelog.cron

第二条消息是错误的，它在 PID 3970 之后缺少右括号 (]) 和冒号 (:)，因此您的 grok 模式不起作用。

由于你的grok失败了，所以syslog_timestamp这个字段不存在，所以你的date过滤器也没什么用，@timestamp会被设置为事件进入logstash管道的时间。

您需要为您拥有的每种消息模式设置一个grok 模式，纠正syslog_timestamp 的一种快速方法是捕获失败的消息并应用另一个grok 过滤器以获取syslog_timestamp字段和消息的其余部分在另一个字段中。

尝试将以下条件添加到您的管道中。

if "_grokparsefailure" in [tags] {
  grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp } %{SYSLOGHOST:syslog_hostname} %{GREEDYDATA:rest_of_syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      remove_field => [ "host", "path" ]
   } 
}

grok 的结果会是这样的：

{
  "syslog_hostname": "hostname",
  "syslog_timestamp": "Mar 12 11:01:02",
  "rest_of_syslog_message": "run-parts(/etc/cron.hourly)[3970 starting mcelog.cron"
}

【讨论】：

leandrojmp，谢谢您的宝贵答案+1，但是我通过删除grokparsefailure标签添加了解决方法，以便可以处理其他数据，但我需要看看我如何适合您的方法ino 我的 logstash 过滤器。
你需要修复你的grok，这是你的主要问题，你现在有两种不同的消息模式，你的 grok 只适用于其中一个，你需要一个适用于两者的 grok 或添加另一个适用于第二条消息的模式。我的建议是在你的 grok 之后使用上面的条件来获得正确的 syslog_timestamp 并创建一个 grok 模式来解析 rest_of_syslog_message 字段。
你好@leandrojmp，你是绝对正确的，我刚刚更新了我的帖子，我申请了临时解决方法，但是我只是保留了一些想法作为编辑，同时牢记你的建议，你能建议这些是否可行。
我认为捕获具有_grokparsefailure 的事件比丢弃事件更好，这样您仍然可以获得消息。您的问题只是grok，您需要查看您有多少种不同的消息模式并制作一个grok 模式来解析它们中的每一个。