Logstash：跨事件保持价值

Question

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

为此，我的方法是使用 grep 过滤器和 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 字段已正确添加到所有事件中 - 这正是我想要的 - 但该字段的值不是日期本身（[=16= 的值]), 但实际的字符串 "%{mydate}", 除了第一次实际匹配时 (在解析我的日志文件中的实际日期时, grepdate 字段包含正确的值)

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

非常感谢任何帮助。

编辑：

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

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

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

Answer 1

也许你应该使用官方的aggregate filter for this, since memorize is not official and 。

它会是这样的：

# 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 字段，其中包含第一行日志中的日期。

Logstash：跨事件保持价值

Logstash: Keeping a value across events

grep

date

multiline

logstash

logstash-grok