# Logstash Filter **URL:** \ **Category:** Logstash\ **Created:** [March 4, 2021, 5:20am UTC](https://discuss.elastic.co/t/logstash-filter/266182 "2021-03-04T05:20:14Z")\ **Posts on this page:** 4\ **Page:** 1
**Author:** ![Anu6006](https://avatars.discourse-cdn.com/v4/letter/a/47e85d/32.png) [@Anu6006](https://discuss.elastic.co/u/Anu6006)\ **Post date:** [March 4, 2021, 5:20am UTC](https://discuss.elastic.co/t/logstash-filter/266182/1 "2021-03-04T05:20:14Z")
Hi Team, I'm trying to get **NSG Flow logs** through _Azure Blob Storage_ (logstash-input-azureblob (0.9.12)). But the filtering part is not working properly and showing the errors **\_jsonparsefailure, \_split\_type\_failure, \_dateparsefailure**. ![Nsg_0](https://us1.discourse-cdn.com/elastic/original/3X/3/d/3d99c04d30b6ab3c5079d96b31998e7600947df4.png) **Types of logs:** { "records": [{ "time": "2021-03-02T11:00:06.8483928Z", "systemId": "", "macAddress": "", "category": "NetworkSecurityGroupFlowEvent", "resourceId": "", "operationName": "NetworkSecurityGroupFlowEvents", "properties": { "Version": 2, "flows": [{ "rule": "DefaultRule\_AllowInternetOutBound", "flows": [{ "mac": "000D3A0A6B1F", "flowTuples": ["1614682747,192.168.200.4,104.211.232.185,36368,443,T,O,A,E,8,1553,1,74", "1614682748,192.168.200.4,104.211.244.138,39948,9200,T,O,A,E,4,399,1,74", "1614682749,192.168.200.4,104.211.232.185,36396,443,T,O,A,B,,,,", "1614682753,192.168.200.4,104.211.232.185,36378,443,T,O,A,E,2,657,1,74", "1614682755,192.168.200.4,104.211.232.185,36404,443,T,O,A,B,,,,", "1614682759,192.168.200.4,104.211.232.185,36388,443,T,O,A,E,8,1553,1,74", "1614682761,192.168.200.4,104.211.232.185,36412,443,T,O,A,B,,,,", "1614682765,192.168.200.4,104.211.244.138,39992,9200,T,O,A,B,,,,", "1614682765,192.168.200.4,104.211.232.185,36396,443,T,O,A,E,8,1553,1,74", "1614682767,192.168.200.4,104.211.232.185,36422,443,T,O,A,B,,,,", "1614682768,192.168.200.4,20.43.120.7,46234,443,T,O,A,B,,,,", "1614682769,192.168.200.4,20.43.120.7,46236,443,T,O,A,B,,,,", "1614682771,192.168.200.4,20.43.120.7,46238,443,T,O,A,B,,,,", "1614682771,192.168.200.4,104.211.232.185,36404,443,T,O,A,E,8,1553,1,74", "1614682773,192.168.200.4,104.211.232.185,36446,443,T,O,A,B,,,,", "1614682777,192.168.200.4,104.211.232.185,36412,443,T,O,A,E,8,1553,1,74", "1614682779,192.168.200.4,104.211.232.185,36456,443,T,O,A,B,,,,", "1614682782,192.168.200.4,104.211.244.138,39992,9200,T,O,A,E,4,399,1,74", "1614682783,192.168.200.4,104.211.232.185,36422,443,T,O,A,E,8,1553,1,74", "1614682785,192.168.200.4,104.211.232.185,36464,443,T,O,A,B,,,,", "1614682785,192.168.200.4,20.43.120.7,46234,443,T,O,A,E,1,74,1,66", "1614682786,192.168.200.4,20.43.120.7,46236,443,T,O,A,E,1,74,1,66", "1614682788,192.168.200.4,20.43.120.7,46238,443,T,O,A,E,1,74,1,66", "1614682790,192.168.200.4,104.211.232.185,36446,443,T,O,A,E,8,1553,1,74", "1614682791,192.168.200.4,104.211.232.185,36472,443,T,O,A,B,,,,", "1614682796,192.168.200.4,104.211.232.185,36456,443,T,O,A,E,8,1553,1,74", "1614682797,192.168.200.4,104.211.232.185,36480,443,T,O,A,B,,,,", "1614682802,192.168.200.4,104.211.232.185,36464,443,T,O,A,E,8,1553,1,74"] }] }, { "rule": "DefaultRule\_DenyAllInBound", "flows": [{ "mac": "", "flowTuples": ["1614682743,149.81.167.130,192.168.200.4,49192,56596,T,I,D,B,,,,", "1614682770,162.142.125.159,192.168.200.4,51377,6021,T,I,D,B,,,,", "1614682773,167.248.133.30,192.168.200.4,62339,8010,T,I,D,B,,,,", "1614682775,162.142.125.146,192.168.200.4,2654,9015,T,I,D,B,,,,", "1614682779,162.142.125.156,192.168.200.4,22192,9950,T,I,D,B,,,,", "1614682779,162.142.125.147,192.168.200.4,21145,52869,T,I,D,B,,,,", "1614682782,103.127.186.26,192.168.200.4,53363,445,T,I,D,B,,,,"] }] }, { "rule": "UserRule\_Port\_9200", "flows": [{ "mac": "", "flowTuples": ["1614682748,104.211.244.138,192.168.200.4,39948,9200,T,I,A,E,4,399,1,74", "1614682765,104.211.244.138,192.168.200.4,39992,9200,T,I,A,B,,,,", "1614682782,104.211.244.138,192.168.200.4,39992,9200,T,I,A,E,4,399,1,74", "1614682798,192.168.250.4,192.168.200.4,13963,9200,T,I,A,E,11,2337,11,3987"] }] }, { "rule": "UserRule\_SSH", "flows": [{ "mac": "", "flowTuples": ["1614682757,218.92.0.192,192.168.200.4,54565,22,T,I,A,B,,,,", "1614682782,218.92.0.192,192.168.200.4,54565,22,T,I,A,E,11,2037,23,5180", "1614682788,218.92.0.192,192.168.200.4,41062,22,T,I,A,B,,,,"] }] }] } }, **The filters which are used:** filter { split { field =\> "[records]" } split { field =\> "[records][properties][flows]" } split { field =\> "[records][properties][flows][flows]" } split { field =\> "[records][properties][flows][flows][flowTuples]" } mutate { split =\> { "[records][resourceId]" =\> "/" } add\_field =\> { "Subscription" =\> "%{[records][resourceId][2]}" "ResourceGroup" =\> "%{[records][resourceId][4]}" "NetworkSecurityGroup" =\> "%{[records][resourceId][8]}" } convert =\> { "Subscription" =\> "string" } convert =\> { "ResourceGroup" =\> "string" } convert =\> { "NetworkSecurityGroup" =\> "string" } split =\> { "[records][properties][flows][flows][flowTuples]" =\> "," } add\_field =\> { "unixtimestamp" =\> "%{[records][properties][flows][flows][flowTuples][0]}" "srcIp" =\> "%{[records][properties][flows][flows][flowTuples][1]}" "destIp" =\> "%{[records][properties][flows][flows][flowTuples][2]}" "srcPort" =\> "%{[records][properties][flows][flows][flowTuples][3]}" "destPort" =\> "%{[records][properties][flows][flows][flowTuples][4]}" "protocol" =\> "%{[records][properties][flows][flows][flowTuples][5]}" "trafficflow" =\> "%{[records][properties][flows][flows][flowTuples][6]}" "traffic" =\> "%{[records][properties][flows][flows][flowTuples][7]}" } convert =\> { "unixtimestamp" =\> "integer" } convert =\> { "srcPort" =\> "integer" } convert =\> { "destPort" =\> "integer" } } date { match =\> ["unixtimestamp", "UNIX"] }} I want to split the 'value' field that is a JSON array into multiple JSONs. **Please help me to filter this properly.** Please guide. Thank you in advance. ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [March 4, 2021, 5:35pm UTC](https://discuss.elastic.co/t/logstash-filter/266182/2 "2021-03-04T17:35:36Z")
This part is fine ``` split { field => "[records]" } split { field => "[records][properties][flows]" } split { field => "[records][properties][flows][flows]" } split { field => "[records][properties][flows][flows][flowTuples]" } ``` That gets you records like ``` { "records" => { "resourceId" => "", "properties" => { "Version" => 2, "flows" => { "rule" => "DefaultRule_AllowInternetOutBound", "flows" => { "mac" => "000D3A0A6B1F", "flowTuples" => "1614682747,192.168.200.4,104.211.232.185,36368,443,T,O,A,E,8,1553,1,74" } } }, "category" => "NetworkSecurityGroupFlowEvent", "operationName" => "NetworkSecurityGroupFlowEvents", "macAddress" => "", "time" => "2021-03-02T11:00:06.8483928Z", "systemId" => "" } ... } ``` The problem is in your mutate filter. Filter operations are [processed in a fixed order](https://www.elastic.co/guide/en/logstash/master/plugins-filters-mutate.html#plugins-filters-mutate-proc_order). coerce first, then rename etc. You can check [the code](https://github.com/elastic/logstash/blob/dbc7bbe179248eb93bf83e413c1dc4e7c93b1b7d/logstash-core/lib/logstash/filters/base.rb#L197) to see the order of the common options. You will need to use multiple mutate options to force the order you want. ---
**Author:** ![Anu6006](https://avatars.discourse-cdn.com/v4/letter/a/47e85d/32.png) [@Anu6006](https://discuss.elastic.co/u/Anu6006)\ **Post date:** [March 12, 2021, 8:21am UTC](https://discuss.elastic.co/t/logstash-filter/266182/3 "2021-03-12T08:21:31Z")
> [@Anu6006](#): > > Hi Team, > I'm trying to get **NSG Flow logs** through _Azure Blob Storage_ (logstash-input-azureblob (0.9.12)). Thanks a lot 😇 ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [April 9, 2021, 8:22am UTC](https://discuss.elastic.co/t/logstash-filter/266182/4 "2021-04-09T08:22:23Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.