How to Split the one field into multiple fields using kv plugins

Elastic Stack Logstash
1

Hi ,

This is Sandeep,

I am new at logstash, here i am trying to fatch data from .csv file to elasticsearch using logstash and i have 6 fields total , in that 6 fields one of the field(log_message) have actually data i need to split that field as multiple fields ,
here is my config file

input{
    file{
        path => "/home/Data/logs.csv"   
       start_position => "beginning"
        sincedb_path => "/dev/null"
    }
}
filter{
    csv{
       separator => ","
        columns => ["log_id","log_datetime","log_type","log_message","log_user" ]
    	}
    date { match => ["log_datetime", "yyyy-MM-dd HH:mm:ss"] }	

    mutate { convert => [ "log_id","integer"] }
  
     kv { field_split_pattern => "[\"]+" value_split_pattern => "->" }
     kv { field_split_pattern => "[\( \)]+" value_split_pattern => "->" }
     kv { field_split_pattern => "[ ]+" value_split_pattern => ":" }

}

output{
    elasticsearch{
        hosts =>"http://localhost:9200"
        index =>"log_infomation"
        document_type => "log_info"
    }
    stdout { codec => "rubydebug" }
}

I am getting the data when i use
kv { field_split_pattern => "["]+" value_split_pattern => "->" }
kv { field_split_pattern => "[( )]+" value_split_pattern => "->" }
kv { field_split_pattern => "+" value_split_pattern => ":" }

but in my log_message field some rows are like this (assume 1st row)
"(Institution Id->77) (data Id->127) QUERY-> insert into tablename1(col1,col2) values(val1,val2) ERROR->Table 'tablename1' doesn't exist"

some are (assume 2st row)
"QUERY-> update set tablename2 col1=val1 some conditions
ERROR: Table 'Tablename2' doesn't exist"

and some rows are like this (assume 3rd row)
"QUERY-> update set tablename3 col1=val1 some conditions
ERROR: You have SQL syntax error "

if i apply "kv" filter plugins i am getting the results like this ,
1st document like this from the first row data,
{
"ERROR" : "Table", //here i am getting only first word from log_message field
"QUERY" : "insert", //here also
"Id" :[
"77",
"127"
]
},
2nd document like this,
{
"ERROR" : "Table", //here i am getting only first word
"QUERY" : "update", //here also
},
3rd document like this
{
"Error" : "update",
"Query" : "You"
}

but here my expected result is
{
"ERROR" : "Table 'tablename1' doesn't exist",
"QUERY" : "insert into tablename1(col1,col2) values(val1,val2) ", //i need total query here not only insert command
"institution Id" : 77, //here i need field name like "institution Id"
"data Id" : 127 //here i need field name like "data Id"
},
{
"ERROR" : "Table 'tablename2' doesn't exist",
"QUERY" : "update set tablename2 col1=val1 some conditions", //i need total query here not only insert command
"institution Id" : 77, //here i need field name like "institution Id"
"data Id" : 127 //here i need field name like "data Id"
},
{
"Query": "update set tablename3 col1=val1 some conditions",
"Error": " You have SQL syntax error"
}

OR

{

"QUERY" : "insert into tablename1(col1,col2) values(val1,val2) "ERROR" => "Table 'tablename1' doesn't exist"
"ERROR" : "Table 'tablename1' doesn't exist",
"institution Id" : 77, //here i need field name like "institution Id"
"data Id" : 127 //here i need field name like "data Id"
},
{
"QUERY" : "update set tablename2 col1=val1 some conditions "ERROR" => "Table 'tablename2' doesn't exist"
"ERROR" : "Table 'tablename2' doesn't exist",
"institution Id" : 77, //here i need field name like "institution Id"
"data Id" : 127 //here i need field name like "data Id"
},
{
"Query": "update set tablename3 col1=val1 some conditions Error: You have SQL syntax error",
"Error": " You have SQL syntax error"
}

what should i do in this case ,can any one help me please ?

Thanks
Sandeep

2

I don't think your data is regular enough to use a kv plugin. grok might work.

2 Likes
3

Hi @Badger

Thank you for reply,

i have less knowledge on grok filter because i am new at logstash so,

can i do this grok filter only particular field like log_message because i need only one field data .

columns are like this ,
columns => ["log_id","log_datetime","log_type","log_message","log_user" ]

this log_message data shown in above context .

please help me ,

Thanks
Sandeep

4

If an example of log_message is

(Institution Id->77) (data Id->127) QUERY-> insert into tablename1(col1,col2) values(val1,val2) ERROR->Table 'tablename1' doesn't exist

Then you could parse that (and the other 2) using this

    grok { match => { "log_message" => "^%{DATA:prefix}QUERY\-> %{DATA:query} ERROR(: |\->)%{DATA:error}$" } }
    if [prefix] {
        ruby {
            code => '
                m = event.get("prefix").scan(/\(([^-]+)\->([^\)]+)\)/)
                m.each.each { |k, v|
                    event.set(k, v)
                }
            '
        }
    }

In the ruby filter the scan function matches a regexp against the "prefix" field. The regexp looks for (key->value) patterns and returns an array of them. Each match is on entry in the array, and that entry itself is an array, the first entry of which is the key, and the second is the value.

2 Likes
5

Hi @Badger ,

Thank you so much for replaying,

It is working perfectly ,i really appreciate you thank you so much .

Thanks
Sandeep

6

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.