Как обрабатывать многострочную запись в журнале с помощью фильтра logstash?

История:

У меня есть пользовательский сгенерированный файл журнала, который имеет следующий шаблон:

[2014-03-02 17:34:20] - 127.0.0.1|ERROR| E:\xampp\htdocs\test.php|123|subject|The error message goes here ; array (
  'create' => 
  array (
    'key1' => 'value1',
    'key2' => 'value2',
    'key3' => 'value3'
  ),
)
[2014-03-02 17:34:20] - 127.0.0.1|DEBUG| flush_multi_line

Вторая запись [2014-03-02 17:34:20] - 127.0.0.1|DEBUG| flush_multi_line Является фиктивной линией, чтобы позволить logstash знать, что событие с несколькими линиями завершено, эта строка будет удалена позже.

Мой конфигурационный файл следующий:

input {
  stdin{}
}

filter{
  multiline{
      pattern => "^\["
      what => "previous"
      negate=> true
  }
  grok{
    match => ['message',"\[.+\] - %{IP:ip}\|%{LOGLEVEL:loglevel}"]
  }

  if [loglevel] == "DEBUG"{ # the event flush  line
    drop{}
  }else if [loglevel] == "ERROR"  { # the first line of multievent
    grok{
      match => ['message',".+\|.+\| %{PATH:file}\|%{NUMBER:line}\|%{WORD:tag}\|%{GREEDYDATA:content}"] 
    }
  }else{ # its a new line (from the multi line event)
    mutate{
      replace => ["content", "%{content} %{message}"] # Supposing each new line will override the message field
    }
  }  
}

output {
  stdout{ debug=>true }
}

Вывод для поля content: The error message goes here ; array (

Проблема:

Моя проблема заключается в том, что я хочу сохранить остальную часть многострочного поля содержимого:

The error message goes here ; array (
  'create' => 
  array (
    'key1' => 'value1',
    'key2' => 'value2',
    'key3' => 'value3'
  ),
)

Итак, я могу удалить поле сообщения позже.

В поле @message содержится все многострочное событие, поэтому я попробовал фильтр mutate с функцией replace, но я ' m просто не может заставить его работать:(.

Я не понимаю, как работает Multiline filter, если кто-то может пролить свет на это, было бы очень благодарно.

Спасибо,

Абду.

Ответ 1

Я просмотрел исходный код и узнал, что:

Многострочный фильтр отменит все события , которые считаются последующим событием, затем добавьте эту строку в исходное поле сообщения, что означает любое фильтры, которые после многострочного фильтра не будут применяться в этом случае
Единственное событие, которое когда-либо будет проходить фильтр, - это тот, который считается новым (что-то, начинающееся с [ в моем случае)

Вот рабочий код:

input {
   stdin{}
}  

filter{
      if "|ERROR|" in [message]{ #if this is the 1st message in many lines message
      grok{
        match => ['message',"\[.+\] - %{IP:ip}\|%{LOGLEVEL:loglevel}\| %{PATH:file}\|%{NUMBER:line}\|%{WORD:tag}\|%{GREEDYDATA:content}"]
      }

      mutate {
        replace => [ "message", "%{content}" ] #replace the message field with the content field ( so it auto append later in it )
        remove_field => ["content"] # we no longer need this field
      }
    }

    multiline{ #Nothing will pass this filter unless it is a new event ( new [2014-03-02 1.... )
        pattern => "^\["
        what => "previous"
        negate=> true
    }

    if "|DEBUG| flush_multi_line" in [message]{
      drop{} # We don't need the dummy line so drop it
    }
}

output {
  stdout{ debug=>true }
}

Приветствия,

Абду

Ответ 2

В этом выпуске упоминается grok и многострочная обработка https://logstash.jira.com/browse/LOGSTASH-509

Просто добавьте "(? m)" перед вашим регулярным выражением grok, и вам не понадобится мутация. Пример из выпуска:

pattern => "(?m)<%{POSINT:syslog_pri}>(?:%{SPACE})%{GREEDYDATA:message_remainder}"

Ответ 3

Многострочный фильтр добавит в сообщение "\n". Например:

"[2014-03-02 17:34:20] - 127.0.0.1|ERROR| E:\\xampp\\htdocs\\test.php|123|subject|The error message goes here ; array (\n  'create' => \n  array (\n    'key1' => 'value1',\n    'key2' => 'value2',\n    'key3' => 'value3'\n  ),\n)"

Однако фильтр grok не может разобрать "\n". Поэтому вам нужно подставить \n другому персонажу, говорит пустое место.

mutate {
    gsub => ['message', "\n", " "]
}

Затем шаблон grok может анализировать сообщение. Например:

 "content" => "The error message goes here ; array (   'create' =>    array (     'key1' => 'value1',     'key2' => 'value2',     'key3' => 'value3'   ), )"

Ответ 4

Не проблема просто упорядочение фильтров. Заказ очень важен для регистрации тайника. Вам не нужна другая строка, указывающая, что вы закончили вывод многострочной строки журнала. Просто убедитесь, что многострочный фильтр появляется перед grok (см. Ниже)

P.s. Мне удалось проанализировать многострочную строку строки журнала, где xml был добавлен в конец строки журнала, и он охватывал несколько строк, и все же я получил хороший чистый объект xml в моей эквивалентной переменной контента (с именем xmlrequest ниже). Прежде чем вы что-нибудь скажете о регистрации xml в журналах... Я знаю... это не идеальный... но это для еще одной дискуссии:)):

filter { 
multiline{
        pattern => "^\["
        what => "previous"
        negate=> true
    }

mutate {
    gsub => ['message', "\n", " "]
}

mutate {
    gsub => ['message', "\r", " "]
}

grok{
        match => ['message',"\[%{WORD:ONE}\] \[%{WORD:TWO}\] \[%{WORD:THREE}\] %{GREEDYDATA:xmlrequest}"]
    }

xml {
source => xmlrequest
remove_field => xmlrequest
target => "request"
  }
}