Logstash: Mantenere un valore tra gli eventi

Question

Ho una data che è presente solo una volta in ogni file di registro e sto cercando di aggiungere questa data a tutti gli eventi successivi dopo che è stata abbinata una volta, facendolo agire come una variabile globale in in qualche modo. (La data è nella parte superiore del documento e sono in grado di utilizzare multiline o apportare modifiche al nome del file o contenuto)

Per questo, il mio approccio è quello di utilizzare un filtro grep con 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" 
} 

Espressione regolare:

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

Quello che noto è che il campo grepdate sia correttamente viene aggiunto a tutti gli eventi - che è quello che voglio - ma il valore di quel campo non è il data stessa (il valore di %{mydate}), ma la stringa effettiva "%{mydate}", tranne quando in realtà viene abbinata per la prima volta (quando si analizza la data effettiva nel mio file di registro, il campo grepdate contiene il valore corretto)

Cosa posso fare per risolvere questo problema?

Qualsiasi aiuto è molto apprezzato.

Edit:

Ora sto cercando una soluzione che prevede l'uso del plugin memorize. Tuttavia, viene visualizzato il seguente errore:

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

C'è un modo per rendere questo filtro thread-safe?

Answer 1

Forse dovresti usare lo aggregate filter ufficiale per questo, poiché memorize non è ufficiale e will not work with Logstash >2.0.

Sarebbe andata così:

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

Tutti i tuoi eventi verranno poi avere un campo mydate con la data che era sulla prima riga del registro.