Как в фильтрах Logstash осуществить выборку определенных полей?

Question

[Антон Кикоть]

Вопрос к Гуру Logstash, познавшим Дзен.
У меня есть простыня, сшитая ранее (еще на этапе загрузки логов) после этого она попадает в фильтры, и вот здесь-то начинается колдовство. Пример сшитого лога:

T1_INPUT_XML]|[field_n2.5d6754ffg2]|[createTemplates]|[2014.03.25 10:00:04.593]|[templates]|[<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<create xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    <entry id="11111111">
        <request>
            <params pkey_template="111111111" pkey_ps="11111111111" pkey_pb="1111111111" pb_create_timestamp="2014-03-25 10:00:04.589" account="123456774" mfo="123456" dest="За водопостачання від , адреса , о/р 1111111111" total_sum="12.65" pkey_company_kind="789" 
...
    <meters/>

Здесь мне уже надо работать только с полем

T1_INPUT_XML]|[field_n2.5d6754ffg2]|[createTemplates]|[2014.03.25 10:00:04.593]|[templates]|[<?xml version="1.0" encoding="UTF-8" standalone="yes"?>

и разложить его на составляющие. Как то:
@timing_level => T1
@app_timestamp => 2014.03.25 10:00:04.593
@session_id => field_n2.5d6754ffg2
@service_name => createTemplates
и уже измененную простыню затем передать в БД
Вопрос - каким фильтром это можно сделать? Проблема в том, что значения необходимого поля динамически изменяются. Пробовал grep и создавать новые field, но как им передавать динамические значения, пока понять не могу. Может кто помочь ?

Answer 1

[Антон Кикоть]

И по традиции отвечу сам )
для себя выбрал такой механизм, может кому понадобится

input {  
    file {
        codec => plain  #сначала получаем логи как они есть 
#       codec => multiline {
#       pattern => "T" ##создаем патерны (теги) по которым будет разбираться простыня
#       negate => true
#       what => "previous" ##определяет положение сшивающего тега(в данном случае в начале листа тег и инфа до следующего указаного в патерне)если ставить "next" то патерн попадает в предыдущий лист
#       }
        path => "/home/kab/logs/*.log"
        start_position => "beginning" 
        type => "syslog"
        }
}
filter {

##############################WORK BLOCK#######################################
    mutate { # избавляемся от ненужных символов
        type => "syslog"
        gsub => ["message","\]|\["," "] # delete from log symbol"[]"
        gsub => ["message","\|"," "] # delete from log symbol "|"
    }
    grok { # обрабатываем поступающее сообщение (шапку) так как она +- стандартна, остальное пройдет через фильтр неизменным
        type => "syslog"
        pattern => "%{SYSLOGPROG:timing_level}   %{HOST:app_name}%{NOTSPACE:IDsession}   %{WORD:jpkg_name}   %{NOTSPACE:date} %{TIME:app_timestamp}   %{PROG:method}"
        }
   multiline { #ну а теперь сшиваем все 
        type => "syslog"
        pattern => "^(T0_)|(T1_)"
        negate => true
        what => "previous"
        }
###############################################################################
   mutate { # избавляемся от ненужных данных
        type => "syslog"
        remove => [tags]
        remove => [program]
        }
}
output { # и передаем на выход
   stdout {
        codec => "rubydebug"
        }
   elasticsearch {
        host => "localhost"
        }
}

Как то так