# Logstash aggregate syslog entries **URL:** \ **Category:** Logstash\ **Created:** [February 9, 2021, 5:09pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783 "2021-02-09T17:09:33Z")\ **Posts on this page:** 14\ **Page:** 1
**Author:** ![Pablo\_Marques](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/pablo_marques/32/83656_2.png) [@Pablo\_Marques](https://discuss.elastic.co/u/Pablo_Marques)\ **Post date:** [February 9, 2021, 5:09pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/1 "2021-02-09T17:09:33Z")
Hello I am getting syslogs from Clearpass servers and using logstash to ingest them into elasticsearch. Some of these messages are received on multiple syslog packets, but they are really the same (big) message. I am trying to combine them (without much success) into one message before sending them to elastic. My grok entry is as follows: ```auto grok { match => [ "message", "<%{POSINT:syslog_pri}>%{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program} %{POSINT:syslog_pid} %{NONNEGINT:num_of_msgs} %{NONNEGINT:msg_num} %{GREEDYDATA:message}", "message", "<%{POSINT:syslog_id}>\.\.\.%{GREEDYDATA:message}" ] overwrite => ["message"] } ``` When a message has multiple entries "num\_of\_msgs" reflects the number of them (1,2,3,4 etc) and "msg\_num" is the sequence number for each message (0,1,2, etc). The common element between all entries is "syslog\_id". The idea is to concatenate all "message" entries with the same "syslog\_id" in the proper order, create an entry with that information and submit it to elasticsearch. Also do not want the partial messages in elasticsearch, only the combined one. I tried this, but it is not doing what I want at all. ```auto if [num_of_msgs] > "1" { aggregate { task_id => "%{syslog_pid}" code => "map['message'] ||=' '; map['message'] +=%{\n}+ event.get('message')" map_action => "create_or_update" push_map_as_event_on_timeout => true timeout => 10 timeout_tags => ['aggregated'] } if "aggregated" not in [tags] { drop{} } } ``` I did a lot of google search but I do not really found anything I can follow. Can you please help me with pointers or suggestions. Thank you ---
**Author:** ![wwalker](https://avatars.discourse-cdn.com/v4/letter/w/43a26b/32.png) [@wwalker](https://discuss.elastic.co/u/wwalker)\ **Post date:** [February 9, 2021, 5:40pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/2 "2021-02-09T17:40:14Z")
Do these messages have a unique ID? So, for example, if an event that comes in over 4 packets, does each contain a field with the same value that is globally unique? You could potentially use that as the index ID and the event will be updated in Elasticsearch. Since you don't list it, I would assume you are using the TCP input to receive the events. As long as the separate packets are contiguous, you might be able to use the multiline codec to stitch the events together. ---
**Author:** ![Pablo\_Marques](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/pablo_marques/32/83656_2.png) [@Pablo\_Marques](https://discuss.elastic.co/u/Pablo_Marques)\ **Post date:** [February 9, 2021, 5:54pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/3 "2021-02-09T17:54:14Z")
Thank you for your answer The common element between all entries is "syslog\_pid". I am using UDP So if I receive these two syslog entries like this: .... CPPM\_blah\_blah 1656456 2 0 "message 123 123 123" .... CPPM\_blah\_blah 1656456 2 1 "message 456 456 456" I want to submit to elasticsearch the following: .... CPPM\_blah\_blah 1656456 1 0 "message 123 123 123 message 456 456 456" ---
**Author:** ![wwalker](https://avatars.discourse-cdn.com/v4/letter/w/43a26b/32.png) [@wwalker](https://discuss.elastic.co/u/wwalker)\ **Post date:** [February 9, 2021, 7:12pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/4 "2021-02-09T19:12:25Z")
Ahh....that makes it more complex. I would guess something like 1. Setup elasticsearch output to use the syslog\_pid as the document id 2. Perform an elasticsearch lookup in your pipeline filter to pull the previously ingested event data. 3. Use another filter, maybe mutate, to create the array of values. ---
**Author:** ![Pablo\_Marques](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/pablo_marques/32/83656_2.png) [@Pablo\_Marques](https://discuss.elastic.co/u/Pablo_Marques)\ **Post date:** [February 9, 2021, 7:26pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/5 "2021-02-09T19:26:31Z")
There is a way using the logstash aggregate filter plugin, but I cannot figure out the right syntax for my case. > **[Aggregate filter plugin | Logstash Reference \[7.10\] | Elastic](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html)** ---
**Author:** ![wwalker](https://avatars.discourse-cdn.com/v4/letter/w/43a26b/32.png) [@wwalker](https://discuss.elastic.co/u/wwalker)\ **Post date:** [February 9, 2021, 7:59pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/6 "2021-02-09T19:59:02Z")
Interesting... So...and my interpretation may be wrong, but going off your existing 'code' configuration, `map['message'] ||=' '; map['message'] +=%{\n}+ event.get('message')`, I would read that to say "(1)Map previous event field message as blank if it doesn't exist. (2)Map previous event field message and append with a new line and the value of the current event's message field." What happens if you simplify it to `"event.set('message', map['message'])"`? I would expect it to concatenate the values into a single string. ---
**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:** [February 9, 2021, 8:50pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/7 "2021-02-09T20:50:26Z")
When using push\_map\_as\_event\_on\_timeout the only fields that will be present on the event are whatever you added to the map. You could try something like this: ``` aggregate { task_id => "%{syslog_pid}" code => ' map["syslog_pri"] ||= event.get("syslog_pri") map["syslog_timestamp"] ||= event.get("syslog_timestamp") map["syslog_hostname"] ||= event.get("syslog_hostname") map["syslog_program"] ||= event.get("syslog_program") map["syslog_pid"] ||= event.get("syslog_pid") map["message"] ||= [] map["message"] << event.get("message") event.cancel ' push_map_as_event_on_timeout => true timeout => 10 } ``` If that syslog\_pid really is a process id and not a message id then if you get two messages within 10 seconds this will combine them. To avoid that you would have to stop using push\_map\_as\_event\_on\_timeout and instead use a second aggregate filter with `map_action => "update"` and `end_of_task => true` when [msg\_num] is one less than [num\_of\_msgs] (which would require a ruby filter to do the maths). More like [example 2](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html#plugins-filters-aggregate-example2) than example 3. ---
**Author:** ![Pablo\_Marques](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/pablo_marques/32/83656_2.png) [@Pablo\_Marques](https://discuss.elastic.co/u/Pablo_Marques)\ **Post date:** [February 9, 2021, 9:02pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/8 "2021-02-09T21:02:55Z")
I believe ```auto syslog_pri ``` is a message id I should rename it to: syslog\_id or message\_id ---
**Author:** ![Pablo\_Marques](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/pablo_marques/32/83656_2.png) [@Pablo\_Marques](https://discuss.elastic.co/u/Pablo_Marques)\ **Post date:** [February 12, 2021, 4:59am UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/9 "2021-02-12T04:59:15Z")
aggregate is now working as expected. But instead of concatenate the messages, it adds values to an array: ```auto map["message"] << get.event("message") ``` message { [0] = " this is line 1" [1] = "this is line 2" etc } Is there a way to concatenate to the string instead of adding a new array entry? ---
**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:** [February 12, 2021, 7:55pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/10 "2021-02-12T19:55:03Z")
> [@Pablo\_Marques](#): > > Is there a way to concatenate to the string instead of adding a new array entry? Sure, you could ``` map["message"] ||= "" map["message"] += event.get("message") ``` possibly adding a delimiter as well. ---
**Author:** ![Pablo\_Marques](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/pablo_marques/32/83656_2.png) [@Pablo\_Marques](https://discuss.elastic.co/u/Pablo_Marques)\ **Post date:** [February 12, 2021, 8:02pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/11 "2021-02-12T20:02:36Z")
I get this error when I use += occurred {:error=\>#\, ---
**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:** [February 12, 2021, 8:06pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/12 "2021-02-12T20:06:07Z")
Did you change `map["message"] ||= []` to `map["message"] ||= ""`? ---
**Author:** ![Pablo\_Marques](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/pablo_marques/32/83656_2.png) [@Pablo\_Marques](https://discuss.elastic.co/u/Pablo_Marques)\ **Post date:** [February 12, 2021, 8:08pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/13 "2021-02-12T20:08:25Z")
I just did Seems to be working now. so by putting the square brackets the variable is defined as an array!! 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:** [March 12, 2021, 8:09pm UTC](https://discuss.elastic.co/t/logstash-aggregate-syslog-entries/263783/14 "2021-03-12T20:09:17Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.