为过去的记录分配标签

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中使用经过的插件，如下所示：

....

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”的用法，我不确定。我是Logtash的新手。但是，使用这种配置，我在kibana中找不到任何标签。我在哪里犯错？

谢谢

Answer 1

几条评论：

• 您的嵌套条件很好。
• 您使用正则表达式进行比较是错误的。 Use "=~"代替“ ==“。
• add_tag仅在基础过滤器起作用时起作用。虽然您可以使用kv {}很好，但是您实际上并没有对键/值做任何事情，因此可能会造成混淆。大多数人将mutate {}用于此类操作。

这些更改可能会使您步入正轨。

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

• 在原始grok {}中，将事件字段分为两部分。这样的模式应该可以做到：％{DATA：event_prefix} _％{DATA：event_verb}
• 将条件替换为与这两个字段完全匹配的条件。

不仅不但更“古怪”，而且在应用于大型数据集时也更快。