Logstash: Keeping einen Wert über Ereignisse

Question

Ich habe ein Datum, das nur einmal in jeder Protokolldatei vorhanden ist, und ich versuche, dieses Datum zu allen folgenden Ereignissen hinzuzufügen, nachdem es einmal übereinstimmte, so dass es wie eine globale Variable in einige Wege. (Das Datum am oberen Rand des Dokuments ist, und ich bin nicht in der Lage multiline zu verwenden oder Änderungen an den Dateinamen oder Inhalt)

Dazu mein Ansatz ist es, ein grep Filter mit drop => false zu verwenden.

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

Regulärer Ausdruck:

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

Was ich feststellen, dass das grepdate Feld korrekt zu allen Veranstaltungen hinzugefügt werden wird - das ist, was ich will - aber der Wert dieses Feldes ist nicht die Datum selbst (der Wert von %{mydate}), aber die tatsächliche Zeichenfolge "%{mydate}", außer wenn tatsächlich zum ersten Mal verglichen wird (beim Analysieren des tatsächlichen Datums in meiner Protokolldatei enthält das grepdate Feld den richtigen Wert)

Was kann ich tun, um das Problem zu beheben?

Jede Hilfe wird sehr geschätzt.

Edit:

ich jetzt eine Lösung versucht, die die Verwendung der memorize Plugin enthält. Ich erhalte jedoch den folgenden Fehler:

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

Gibt es eine Möglichkeit, diesen Filter threadsicher zu machen?

Answer 1

Vielleicht sollten Sie die offizielle aggregate filter dafür verwenden, da memorize nicht offiziell ist und will not work with Logstash >2.0.

Es würde so gehen:

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

Alle Ihre Veranstaltungen werden dann ein mydate Feld mit dem Datum, das auf der ersten Log-Zeile war.