Как разделить событие Logstash, содержащее несколько раз один и тот же шаблон

Я читаю ввод в формате xml и пытаюсь извлечь каждую строку таблицы html как отдельное событие.

Например, если мой ввод:

<xml> <table> <tr> <td> 1 </td> <td> 2 </td> </tr> <tr> <td> 3 </td> <td> 4 </td> </tr> </table> </xml>

Я хочу, чтобы результат был:

{
       "message" => "<tr> <td> 1 </td> <td> 2 </td> </tr>",
      "@version" => "1",
    "@timestamp" => "2015-03-20T10:30:38.234Z",
          "host" => "VirtualBox"
}
{
       "message" => "<tr> <td> 3 </td> <td> 4 </td> </tr>",
      "@version" => "1",
    "@timestamp" => "2015-03-20T10:30:38.234Z",
          "host" => "VirtualBox"
}

Проблема в том, что мне нужно разделить событие на несколько событий. Использование разделительного фильтра не сработало, потому что он удаляет строку, используемую в качестве «терминатора».

Я разработал собственный шаблон grok для извлечения содержимого строки html: (?<data><tr>(.)*?</tr>)

К сожалению, этот шаблон обнаруживает только первое вхождение, и хотя в одном XML-файле имеется конечное число вхождений, количество строк заранее неизвестно.

Взглянув на JIRA-703 на веб-сайте logstash, я боюсь, что grok не может найти один шаблон несколько раз (на данный момент Марс 2015 г.)

Должен ли я кодировать свой собственный фильтр? Можно ли сохранить каждое совпадение фильтра grok как новое событие?

Вы можете посмотреть мой фильтр

    input {
        stdin { }
    }

    filter {
        mutate {
            gsub => ["message", "<tr>", "[split]<tr>"]
        }
        mutate {
            gsub => ["message", "</tr>", "</tr>[split]"]
        }
        split {
            terminator => "[split]"
        }
        grok {
            patterns_dir => "../patterns"
            #voir pourquoi le meme pattern plusieurs fois ne fonctionne pas
            #https://logstash.jira.com/browse/LOGSTASH-703
            match => ["message", "%{HTML_ROW_LINE:data}" ]
        }
    }

    output {
        stdout {
            codec => rubydebug
        }
    }

Я обнаружил, что когда я разделяю событие до и после строки, фильтр grok больше не работает. Я действительно получаю то, что хочу, в поле «сообщение», но больше не в поле «данные», как хотел.

Странно то, что я не получаю тег "_grokparsefailure", пока я не получаю поле данных. Кажется, это указывает на то, что совпадение действительно есть, но оно не хранится в поле.


person vdolez    schedule 20.03.2015    source источник
comment
Вам нужен фильтр, который создает новые события, и для этого подойдет split{}. Можете ли вы снова добавить строку терминатора с помощью mutate-›gsub{} ?   -  person Alain Collins    schedule 22.03.2015
comment
Я действительно изучал методы mutate-›gsub. Но поскольку я разделил свои события, я думаю, что фильтр grok больше не соответствует. Спасибо за ваш комментарий в любом случае :)   -  person vdolez    schedule 23.03.2015
comment
После того, как вы запустите split{}, все фильтры после этого должны быть применены.   -  person Alain Collins    schedule 23.03.2015
comment
Я обнаружил, что JIRA говорит об ошибке фильтра Split: logstash.jira.com/browse/LOGSTASH -1312 Видимо будет исправлено в следующей версии (1.5). Сейчас я думаю, что просто создам временный выходной файл, который будет использоваться в качестве входных данных.   -  person vdolez    schedule 23.03.2015