# How to validate the data after parsing with KV Filter **URL:** \ **Category:** Logstash\ **Created:** [January 6, 2017, 6:55pm UTC](https://discuss.elastic.co/t/how-to-validate-the-data-after-parsing-with-kv-filter/70811 "2017-01-06T18:55:41Z")\ **Posts on this page:** 5\ **Page:** 1
**Author:** ![sandeepkanabar](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/sandeepkanabar/32/79399_2.png) [@sandeepkanabar](https://discuss.elastic.co/u/sandeepkanabar)\ **Post date:** [January 6, 2017, 6:55pm UTC](https://discuss.elastic.co/t/how-to-validate-the-data-after-parsing-with-kv-filter/70811/1 "2017-01-06T18:55:41Z")
I have log messages like: ``` 2017-01-06 19:27:53,893 INFO [[ACTIVE] ExecuteThread: '10' for queue: 'weblogic.kernel.Default (self-tuning)'] com.symantec.olp.sas.log.SummaryLogAspect - {country=JP, svid=182, typ=sassum, vid=1120, xffmode=0, initallocdate=2014-12-15T01:08:24Z, xffip=222.230.107.165, seatcnt=1, plgid=2, api=941753644, oslocale=JPN, fng=CJ6FRE1208VMNRQG, req=/228228131/28RXAAPB1DqJj/RSLHL940/EMBXtu+/f+/Zeb/KV1Q/DTXZBFC94ZE5AOmz/mDCqB7zJOARDQO/166180202502162557303662649078783407201612&D09DEEFB7E78065D?NAM=SFlBUy1WQTE2&MFN=VkFJTyBDb3Jwb3JhdGlvbg==&MFM=VkpQMTEx&OLA=JPN&OLO=JPN, lpmv=470, oslang=JPN, ctok=166180202502162557303662649078783407201612, resptime=119, epid=70D3B811A994477F957A90985109BE9D, campnid=0, remip=222.230.107.165, lictype=SOS, dbepid=70D3B811A994477F957A90985109BE9D, cid=nav1sasapppex02.msp.symantec.com1481215212435, status=10002, siid=240, skum=21356539, skup=01001230, psn=O749UPCN8KSY, cip=84.100.138.144, mname=VAIO Corporation, puid=1199, skuf=01100470, st=1481765738387, prbid=5967, mmodel=VJP111, clang=EN, pnfi=1120, cprbid=745, cpmv=7428, euip=222.230.107.165, prcdline=2, dvnm=HYAS-VA16, remdays=0, seatid=ah00s8CIdqUQyW2V, sasvid=106, xlsid=3730, baseactkey=186635290403122706518307794, coupon=651218, translogid=75033f05-9cf2-48e2-b924-fc2441d11d33} 2017-01-06 19:28:03,894 INFO [[ACTIVE] ExecuteThread: '10' for queue: 'weblogic.kernel.Default (self-tuning)'] com.symantec.olp.sas.log.SummaryLogAspect - {country=JP, svid=182, typ=sassum, vid=1120, xffmode=0, initallocdate=2014-12-15T01:08:24Z, xffip=222.230.107.165, seatcnt=1, plgid=2, api=228228131, oslocale=JPN, fng=1TA6U8RVL0JQXA0N, req=/228228131/28RXAAPB1DqJj/RSLHL940/EMBXtu+/f+/Zeb/KV1Q/DTXZBFC94ZE5AOmz/mDCqB7zJOARDQO/166180202502162557303662649078783407201612&D09DEEFB7E78065D?NAM=SFlBUy1WQTE2&MFN=VkFJTyBDb3Jwb3JhdGlvbg==&MFM=VkpQMTEx&OLA=JPN&OLO=JPN, lpmv=470, oslang=JPN, ctok=166180202502162557303662649078783407201612, resptime=119, epid=70D3B811A994477F957A90985109BE9D, campnid=0, remip=222.230.107.165, lictype=SOS, dbepid=70D3B811A994477F957A90985109BE9D, cid=nav1sasapppex02.msp.symantec.com1481215212435, status=0000, siid=240, skum=21356539, skup=01001230, psn=28MHHH2VPR4T, cip=222.230.107.165, mname=VAIO Corporation, puid=1199, skuf=01100470, st=1481765738387, prbid=5967, mmodel=VJP111, clang=EN, pnfi=1120, cprbid=745, cpmv=1027, euip=222.230.107.165, prcdline=2, dvnm=HYAS-VA16, remdays=0, seatid=StrlisGXA4yAt1ad, sasvid=130, xlsid=2820, baseactkey=028200017462383754273799438, coupon=123456, translogid=72df4536-6038-4d1c-b213-d0ff5c3c20fb} ``` I use the below grok pattern to match against these: `(?m)%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:severity} \[%{GREEDYDATA:thread}\] %{JAVACLASS:className} - \{%{GREEDYDATA:logmsg}\}` Post that, I use the KV filter to split the fields in logmsg field and include only those fields that are of interest to me. My question is: _How can I validate the format of those fields that are of interest to me?_ One thing I need to mention is - the log contains different numbers of fields in logmsg that's why I've used `GREEDYDATA` My Logstash.conf is as follows: ``` input { kafka { bootstrap_servers => "brokers_list" topics => ["transaction-log"] codec => "json" } } filter { grok { match => ["message", "(?m)%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:severity} \[%{GREEDYDATA:thread}\] %{JAVACLASS:className} - \{%{GREEDYDATA:logmsg}\}" ] #overwrite => ["message"] } if "_grokparsefailure" not in [tags] { kv { field_split => ", " source => "logmsg" include_keys => ["api", "fng", "status", "cip", "cpmv", "translogid", "coupon", "baseactkey", "xlsid", "sasvid", "seatid", "srcHostname", "serverId"] allow_duplicate_values => false remove_field => ["message", "kafka.*", "logmsg"] } } if [api] != "228228131" { mutate { add_tag => "_grokparsefailure" } } date { # use timestamp from the log "match" => ["timestamp", "YYYY-MM-dd HH:mm:ss,SSS"] target => "@timestamp" } mutate { remove_field => ["timestamp"] # remove unused stuff } } output { if "_grokparsefailure" not in [tags] { kafka { topic_id => "invalid topic" bootstrap_servers => "brokers_list" codec => json {} } } else { kafka { topic_id => "valid topic" bootstrap_servers => "brokers_list" codec => json { } } } } ``` After parsing with `KV` filter, I'm checking the value of `api` field and if it is NOT equal to `228228131`, then I add \_grokparsefailure tag to it and don't process further. I want to be able to validate the format of the fields listed in Include\_keys like `cip` which is client IP? How can I validate the data format for those fields? Since my log contains different number of fields, at the grok level I can't validate. Only after parsing with KV I can get those fields and validate them. By validation i mean, validate that they conform to the type defined in ES Index. This is because in case, they are not conforming, I want to send them to invalid-topic in Kafka. Should I use ruby filter to validate? If so, can you give me a sample? Or should I rebuild the event after KV parsing and again use grok on that newly created event. Will v much appreciate some sample showing these. ---
**Author:** ![sandeepkanabar](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/sandeepkanabar/32/79399_2.png) [@sandeepkanabar](https://discuss.elastic.co/u/sandeepkanabar)\ **Post date:** [January 7, 2017, 1:55am UTC](https://discuss.elastic.co/t/how-to-validate-the-data-after-parsing-with-kv-filter/70811/2 "2017-01-07T01:55:56Z")
@magnusbaeck Any suggestions please? ---
**Author:** ![warkolm](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/warkolm/32/39224_2.png) [@warkolm](https://discuss.elastic.co/u/warkolm)\ **Post date:** [January 7, 2017, 5:12am UTC](https://discuss.elastic.co/t/how-to-validate-the-data-after-parsing-with-kv-filter/70811/3 "2017-01-07T05:12:09Z")
Logstash is not that great at validation. The best thing I can think of would be to use date/ip filters to see if they can process the field, then conditional match them if they don't. > [@sandeepkanabar](#): > > @magnusbaeck Any suggestions please? Please don't ping people like that, unless they are already in the topic answering. If you'd like guaranteed answers then there are support subscriptions, otherwise the time most of us spend here is voluntary so we ask that you respect that 🙂 ---
**Author:** ![sandeepkanabar](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/sandeepkanabar/32/79399_2.png) [@sandeepkanabar](https://discuss.elastic.co/u/sandeepkanabar)\ **Post date:** [January 9, 2017, 10:26am UTC](https://discuss.elastic.co/t/how-to-validate-the-data-after-parsing-with-kv-filter/70811/4 "2017-01-09T10:26:19Z")
> [@warkolm](#): > > Please don't ping people like that, unless they are already in the topic answering. If you'd like guaranteed answers then there are support subscriptions, otherwise the time most of us spend here is voluntary so we ask that you respect that 🙂 My sincerest apologies, @warkolm. Thanks for letting me know. We do have support subscription but it was Friday :(. I have been at it since so long and I thought if I could get the answer, I could try it out over the weekend. Thank you. ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [February 6, 2017, 10:26am UTC](https://discuss.elastic.co/t/how-to-validate-the-data-after-parsing-with-kv-filter/70811/5 "2017-02-06T10:26:46Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.