Logstash: Przechowywanie wartości w różnych zdarzeniach

Question

Mam datę, która jest obecna tylko raz w każdym pliku dziennika i próbuję dodać tę datę do wszystkich następujących zdarzeń po tym, jak została dopasowana raz, co powoduje, że działa ona jak zmienna globalna w kilka sposobów. (Data znajduje się na górze dokumentu i nie jestem w stanie używać multiline lub dokonać zmiany nazwy pliku lub zawartości)

za to, moje podejście jest użycie grep filtra 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" 
} 

Wyrażenie regularne:

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

Co Zauważyłem to, że pole grepdate jest poprawnie dodawane do wszystkich wydarzeń - czyli to, co chcę - ale wartość tego pola nie jest sama data (wartość %{mydate}), ale rzeczywisty ciąg "%{mydate}", z wyjątkiem sytuacji, gdy faktycznie jest dopasowywany po raz pierwszy (podczas analizowania faktycznej daty w moim pliku dziennika, pole grepdate zawiera poprawną wartość)

Co mogę zrobić, aby to naprawić?

Każda pomoc jest bardzo doceniana.

Edit:

Jestem teraz próbuje rozwiązanie, które obejmuje korzystanie z wtyczki memorize. Jednak otrzymuję następujący błąd:

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

Czy istnieje sposób, aby ten wątek filtra był bezpieczny?

Answer 1

Może powinieneś użyć oficjalnego aggregate filter, ponieważ memorize nie jest oficjalny i will not work with Logstash >2.0.

byłoby to tak:

# 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 
    } 
} 

wszystkie wydarzenia będzie wtedy mieć mydate pole z datą, która była na pierwszej linii dziennika.