在 logstash 中为 elapsed 分配标签

Question

我有这样的日志文件：

2015-12-05 05:00:51 prefix1_sent 1
2015-12-05 05:00:52 prefix2_sent 2
2015-12-05 05:00:53 prefix1_received 1
2015-12-05 05:00:54 prefix2_received 2

我想了解特定事件花费了多少时间。我在 logstash 中使用 elapsed 插件是这样的：

....

grok {
        patterns_dir => "path"
        match => { 
            "message" => "%{TIMESTAMP:logtime}, %{EVENT:event}, %{SPLIT:ID}"
        }
    }

filter{
    if [event] == "(.)*sent" {
        if [event] == "prefix1(.)*" {
            kv {
                add_tag => [ "prefix1_sent" ]
            }
        } else if [event] == "prefix2(.)*" {
            kv {
                add_tag => [ "prefix2_sent" ]
            }
    } else if [event] == "(.)*received" {
        if [event] == "prefix1(.)*" {
            kv {
                add_tag => [ "prefix1_received" ]
            }
        } else if [event] == "prefix2(.)*" {
            kv {
                add_tag => [ "prefix2_received" ]
            }
        }
    }
}
filter{
    elapsed {
        start_tag => "prefix1_sent"
        end_tag => "prefix1_received"
        unique_id_field => "ID"
        new_event_on_match => false
    }
}
filter{
    elapsed {
        start_tag => "prefix2_sent"
        end_tag => "prefix2_received"
        unique_id_field => "ID"
        new_event_on_match => false
    }
}

....

我不确定我的嵌套 if 语句语法是否正确，我是否理解 "kv" 的用法。我是 logstash 的新手。但是我在 kibana 中找不到任何带有此配置的标签。我哪里出错了？

谢谢

Answer 1

几条评论：

你的嵌套条件没问题。
您使用正则表达式进行的比较是错误的。 Use "=~" 而不是“==”。
add_tag 仅在底层过滤器有效时有效。虽然您使用 kv{} 没问题，但您实际上并没有对 key/values 做任何事情，所以它可能会造成混淆。大多数人使用 mutate{} 来做这样的事情。

这些更改可能会让您走上正轨。

对于一些未经请求的建议，请查看您的工作流程。首先，您应用带有 grok 的正则表达式来创建三个字段。然后在执行 add_field 之前再次对其中之一进行正则表达式（2-3 次）。试试这个：

在原来的grok{}中，将事件字段拆分为两部分。像这样的模式应该这样做： %{DATA:event_prefix}_%{DATA:event_verb}
重写您的条件，使其与这两个字段完全匹配。

不仅"grok-ish"，而且应用于大数据集时速度会更快。

在 logstash 中为 elapsed 分配标签

Assigning tags for elapsed in logstash

syntax

elasticsearch

logstash

kibana-4