Logstash:跨事件保持价值 [英] Logstash: Keeping a value across events

问题描述

我有一个在每个日志文件中只出现一次的日期，我试图在匹配一次后将此日期添加到所有后续事件中，使其在某些方面表现得像一个全局变量.(日期在文档顶部，我无法使用 multiline 或更改文件名或内容)

I have a date that is only present once in every log file and I am trying to add this date to all following events after it has been matched once, making it act like a global variable in some ways. (The date is at the top of the document and I am unable to use multiline or make changes to the file name or content)

为此，我的方法是使用带有 drop => 的 grep 过滤器.错误.

For this, my approach is to use a grep filter with drop => false.

grok {
    patterns_dir => "[...]"
    match => [ "message", "%{DATELINE}" ]
    tag_on_failure => [ ]
}
grep {
    add_field => { "grepdate" => "%{mydate}" }
    drop => false
}
date {
    locale => "en"
    timezone => "Europe/Paris"
    match => [ "grepdate", "yyyyMMdd" ]
    target => "grepdate"
}

正则表达式:

DATELINE (= Date: (?<mydate>[0-9]{8}))

我注意到 grepdate 字段被正确添加到所有事件中 - 这是我想要的 - 但该字段的值不是日期本身(%{mydate})，但实际的字符串 "%{mydate}"，除非第一次实际匹配时(在我的日志文件中解析实际日期时，grepdate 字段包含正确的值)

What I notice is that the grepdate field is correctly being added to all events - which is what I want - but the value of that field is not the date itself (the value of %{mydate}), but the actual string "%{mydate}", except when actually being matched for the first time (when parsing the actual date in my log file, the grepdate field contains the correct value)

我该怎么做才能解决这个问题?

What can I do to fix this?

非常感谢任何帮助.

我现在正在尝试一种解决方案，其中包括使用 memorize 插件.但是，我收到以下错误:

I am now trying a solution that includes the use of the memorizeplugin. However, I am getting the following error:

不能使用超过 1 个过滤器工作器，因为以下插件不要与一个以上的工人一起工作:记住

Cannot use more than 1 filter worker because the following plugins don't work with more than one worker: memorize

有没有办法让这个过滤器线程安全?

Is there a way to make this filter thread-safe?

推荐答案

也许你应该使用官方的 aggregate 过滤器，因为 memorize 不是官方的，而且 不适用于 Logstash >2.0.

Maybe you should use the official aggregate filter for this, since memorize is not official and will not work with Logstash >2.0.

它会是这样的:

# same as what you have now
grok {
    patterns_dir => "[...]"
    match => [ "message", "%{DATELINE}" ]
    tag_on_failure => [ "not_date_line" ]

}
# add a fictional taskId field to correlate all lines
mutate {
   add_field => { "taskId" => "all" }
}

# if we're processing the first line, remember the date
if "not_date_line" not in [tags] {
    aggregate {
        task_id => "%{taskId}"
        code => "map['mydate'] = event['mydate']"
    }
} 
# if we're processing the next lines, add the date
else {
    aggregate {
        task_id => "%{taskId}"
        code => "event['mydate'] = map['mydate']"
        map_action => "update"
        timeout => 0
    }
}

然后，您的所有事件都会有一个 mydate 字段，其中包含第一条日志行上的日期.

All your events will then have a mydate field with the date that was on the first log line.

这篇关于Logstash:跨事件保持价值的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！