How to select certain fields in Logstash filters?

Question

logstash

A question for the Logstash Guru who knows Zen.
I have a sheet sewn earlier (still at the stage of loading the logs) after that it gets into the filters, and this is where the witchcraft begins. An example of a stitched log:

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/>

Here I already need to work only with the field

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

and break it down into its components. Like this:
@timing_level => T1
@app_timestamp => 2014.03.25 10:00:04.593
@session_id => field_n2.5d6754ffg2
@service_name => createTemplates
and then transfer the already changed sheet to the database
Question - what filter can this be done? The problem is that the values ​​of the required field change dynamically. I tried grep and create new fields, but I still can't figure out how to pass dynamic values ​​to them. Can anyone help?

Answer 1

And by tradition I will answer myself)
I chose such a mechanism for myself, maybe someone will need

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

Something like this