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 поле корректно добавляется ко всем событиям, - что то, что я хочу - но значение этого поля не сама дата (значение %{mydate}), но фактическая строка "%{mydate}", за исключением случаев, когда фактически выполняется сопоставление в первый раз (при анализе фактической даты в моем файле журнала поле grepdate содержит правильное значение)

Что я могу сделать, чтобы исправить это?

Любая помощь очень ценится.

Edit:

Я сейчас пытаюсь решение, которое включает в себя использование 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, так как memorize не является официальным и 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 поле с датой, которая была на первой строке журнала.