Logstash: сохранение значения между событиями

У меня есть дата, которая присутствует только один раз в каждом файле журнала, и я пытаюсь добавить эту дату ко всем последующим событиям после того, как она была сопоставлена ​​один раз, что в некотором роде заставляет ее действовать как глобальная переменная. (Дата указана вверху документа, и я не могу использовать 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 содержит правильное значение)

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

Любая помощь приветствуется.

Изменить:

Сейчас я пробую решение, включающее использование плагина memorizeplugin. Однако я получаю следующую ошибку:

Невозможно использовать более 1 обработчика фильтров, поскольку следующие плагины не работают более чем с одним обработчиком: memorize

Есть ли способ сделать этот фильтр потокобезопасным?


date logstash multiline logstash-grok grep
person halpsb    schedule 25.03.2015    source источник

Ответы (1)


arrow_upward
5
arrow_downward

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

person Val    schedule 25.05.2016
comment
Было ли это вообще полезно? - person Val; 16.05.2017