# Symantec Message Gateway Multiline Logs Parsing **URL:** \ **Category:** Logstash\ **Created:** [March 17, 2021, 7:28am UTC](https://discuss.elastic.co/t/symantec-message-gateway-multiline-logs-parsing/267455 "2021-03-17T07:28:47Z")\ **Posts on this page:** 5\ **Page:** 1
**Author:** ![msszafar](https://avatars.discourse-cdn.com/v4/letter/m/ed655f/32.png) [@msszafar](https://discuss.elastic.co/u/msszafar)\ **Post date:** [March 17, 2021, 7:28am UTC](https://discuss.elastic.co/t/symantec-message-gateway-multiline-logs-parsing/267455/1 "2021-03-17T07:28:47Z")
Hi Community, I've multiline logs of Symantec message gateway (email server). > {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:21:59 smtp01 bmserver: 1614684119|0a0a5199-534c370000006c49-29-603e1fd7fcf8|MSG\_SIZE|3682","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} Problem is that I couldn't combine logs that contains same value of "0a0a5199-534c370000006c49-29-603e1fd7fcf8". I'm using regex value like this one input { tcp { port =\> 5097 type =\> "smg" codec =\> multiline { pattern =\> "\d+|(.\*?)|" what =\> "next" } } } What it does is like it checks if there is one or more digit and after digit, there must be pipe "|", in between two pipes "|" and "|" there is value. When you match it, then please combine all the lines that contain the same value of unique id like this one "0a0a5199-534c370000006c49-29-603e1fd7fcf8". This identifier "0a0a5199-534c370000006c49-29-603e1fd7fcf8" is getting changed but I am unable to combine the logs of unique based on the unique identifier. **These are the logs entries where unique identifier is "0a0a5199-534c370000006c49-29-603e1fd7fcf8":** {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:21:59 smtp01 bmserver: 1614684119|0a0a5199-534c370000006c49-29-603e1fd7fcf8|MSG\_SIZE|3682","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:21:59 smtp01 bmserver: 1614684119|0a0a5199-534c370000006c49-29-603e1fd7fcf8|EHLO|hostname\_here","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:21:59 smtp01 bmserver: 1614684119|0a0a5199-534c370000006c49-29-603e1fd7fcf8|LOGICAL\_IP|10.10.81.133","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:21:59 smtp01 bmserver: 1614684119|0a0a5199-534c370000006c49-29-603e1fd7fcf8|UNTESTED|xyz@gmail.com|submission|spam|bulk|newsletter|suspicious\_url|gray|safe|opl|has\_urls|unscannable\_pmc|content\_740|content\_1423808626610|content\_1532019171118|content\_500|content\_1542184136209|content\_1614087423761|content\_720|content\_750|content\_600|content\_1454394469379|content\_1530882675851|content\_1543489360725|content\_1548324806784|content\_1415274004379|content\_700|content\_1569318369896|content\_730|content\_760|content\_1569928837176|content\_1548326156991|content\_1507892598349|content\_520|content\_521|content\_710|sys\_deny\_ip|sys\_allow\_ip|sys\_allow\_email|sys\_deny\_email|dns\_allow|dns\_deny|user\_allow|user\_deny|freq\_va|freq\_dha|freq\_sa|connection\_class\_0|connection\_class\_1|connection\_class\_2|connection\_class\_3|connection\_class\_4|connection\_class\_5|connection\_class\_6|connection\_class\_7|connection\_class\_8|connection\_class\_9|senderauth\_batv\_sign|senderauth\_batv\_fail|blockedlang|knownlang","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} And here are the logs where unique identifier is **0a0a5199-534c370000006c49-2b-603e1fdc40df** {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:22:04 smtp01 bmserver: 1614684124|0a0a5199-534c370000006c49-2b-603e1fdc40df|MSG\_SIZE|4447","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:22:04 smtp01 bmserver: 1614684124|0a0a5199-534c370000006c49-2b-603e1fdc40df|EHLO|hostname","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:22:04 smtp01 bmserver: 1614684124|0a0a5199-534c370000006c49-2b-603e1fdc40df|LOGICAL\_IP|10.10.x.x","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} **Output Desired** There should be only two logs entries becuase unique identifiers are two. How can I achieve it? Please help ... ---
**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 17, 2021, 4:28pm UTC](https://discuss.elastic.co/t/symantec-message-gateway-multiline-logs-parsing/267455/2 "2021-03-17T16:28:32Z")
> [@msszafar](#): > > pattern =\> "\d+|(.\*?)|" That will match any line that contains numbers, or any line that contains anything. That means it will match every line. | is used for alternation ("or") in a regexp. It is possible you mean `"\d+\|(.*?)\|"`, but even then, it is not going to match the capture group. It will still match any number followed by a pipe, followed by anything, followed by a pipe. I suggest you try an aggregate filter. Take a look at [example 3](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html#plugins-filters-aggregate-example3). ---
**Author:** ![msszafar](https://avatars.discourse-cdn.com/v4/letter/m/ed655f/32.png) [@msszafar](https://discuss.elastic.co/u/msszafar)\ **Post date:** [March 18, 2021, 10:36am UTC](https://discuss.elastic.co/t/symantec-message-gateway-multiline-logs-parsing/267455/3 "2021-03-18T10:36:41Z")
Thank you for your prompt reply. I've explored aggregate filter but it doesn't resolve the issue. The aggregate filter requires two fields `code` and `task_id`. The value `code` allows us to use `map` or `event` that performs aggregation on the fields extracted using GROK or some other filter. Output is only one field that contains the aggregated results in numbers like 2+2-2=2 But my issue is to combine the multiline logs into single line because SMG logsource send multiline logs that are meaning less unless you don't combine multiple lines into single line. This can be achieved using unique identifier. **Here is the logstash code** > input { > file { > path =\> "/home/zafar.iqbal/logstash/logstash-7.11.1/bin/aggregate.log" > sincedb\_path =\> "/dev/null" > start\_position =\> "beginning" > type =\> "smg" > } > } > filter { > if [type] == "smg" { > mutate { rename =\> {"host" =\> "[host][ip]"} remove\_field =\> ["host"]} > grok { > match =\> {"message" =\> "\<%{POSINT}\>%{MONTH:month} %{MONTHDAY:day} %{NOTSPACE:time} %{DATA:[host][name]} %{GREEDYDATA:[event][original]}"} > }#extracted the raw payload in event.original field > grok { > match =\> {"[event][original]" =\> "%{DATA:[event][module]}: %{NUMBER}|%{DATA:unique\_id}|"} > }#got unique id value in field "unique\_id" > mutate {remove\_field =\> ["[host][ip]", "month", "day", "time", "[host][name]", "[event][module]", "message"]} > aggregate { > task\_id =\> "%{unique\_id}" > code =\> "map['variable\_name'] = 0" > #push\_map\_as\_event\_on\_timeout =\> true > #timeout\_task\_id\_field =\> "user\_id" > #timeout =\> 30 # 1 hour timeout, user activity will be considered finished one hour after the first event, even if events keep coming > #inactivity\_timeout =\> 30 # 5 minutes timeout, user activity will be considered finished if no new events arrive 5 minutes after the last event > #timeout\_tags =\> ['\_aggregatetimeout'] > #timeout\_code =\> "event.set('several\_clicks', event.get('clicks') \> 1)" > }#end\_aggregate\_filter > }#end\_main\_if\_condition > } > output { > if [type] == "smg" { > stdout {codec =\> rubydebug} > } > } ---
**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 18, 2021, 7:02pm UTC](https://discuss.elastic.co/t/symantec-message-gateway-multiline-logs-parsing/267455/4 "2021-03-18T19:02:43Z")
> [@msszafar](#): > > {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:22:04 smtp01 bmserver: 1614684124|0a0a5199-534c370000006c49-2b-603e1fdc40df|MSG\_SIZE|4447","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} > {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:22:04 smtp01 bmserver: 1614684124|0a0a5199-534c370000006c49-2b-603e1fdc40df|EHLO|hostname","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} > {"port":49542,"host":"10.10.x.x","type":"smg","message":"\<158\>Mar 2 16:22:04 smtp01 bmserver: 1614684124|0a0a5199-534c370000006c49-2b-603e1fdc40df|LOGICAL\_IP|10.10.x.x","@timestamp":"2021-03-04T07:57:47.190Z","@version":"1"} You can add anything you want to the map, then push the contents of the map as a new event on timeout. With those messages you could parse out items using grok ``` grok { pattern_definition => { "GUID" => "%{BASE16NUM}-%{BASE16NUM}-%{BASE16NUM}-%{BASE16NUM}" } break_on_match => false match => { "message" => [ "\|%{GUID:guid}\|", "MSG_SIZE\|%{INT:msgsize}", "LOGICAL_IP\|%(IPV4:clientIp}", "\|%{WORD:command}|%{HOSTNAME:clientHost}$" ] } } ``` then aggregate them based on the id ``` aggregate { task_id => "%{guid}" code => ' map["@timestamp"] ||= event.get("@timestamp") map["port"] ||= event.get("port") map["host"] ||= event.get("host") map["type"] ||= event.get("type") map["msgsize"] ||= event.get("msgsize") map["clientIp"] ||= event.get("clientIp") map["command"] ||= event.get("command") map["clientHost"] ||= event.get("clientHost") ' push_map_as_event_on_timeout => true timeout_task_id_field => "guid" timeout => 10 } ``` Alternatively, if you want all of the messages ``` map["messages"] ||= [] map["messages"] << event.get("message") ``` ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [April 15, 2021, 7:02pm UTC](https://discuss.elastic.co/t/symantec-message-gateway-multiline-logs-parsing/267455/5 "2021-04-15T19:02:47Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.