# Logstash aggregate array of objects based on key field

**URL:** <https://discuss.elastic.co/t/logstash-aggregate-array-of-objects-based-on-key-field/250249>\
**Category:** Logstash\
**Created:** [September 28, 2020, 5:51pm UTC](https://discuss.elastic.co/t/logstash-aggregate-array-of-objects-based-on-key-field/250249 "2020-09-28T17:51:51Z")\
**Posts on this page:** 7\
**Page:** 1

<div class="post-metadata">

**Author:** ![rfferrao](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rfferrao/32/76344_2.png) [@rfferrao](https://discuss.elastic.co/u/rfferrao)\
**Post date:** [September 28, 2020, 5:51pm UTC](https://discuss.elastic.co/t/logstash-aggregate-array-of-objects-based-on-key-field/250249/1 "2020-09-28T17:51:51Z")

</div>

I'm receiving some logs like these:

```auto
datetime="2020-09-2020T14:40:00-0300" action="tunnel-up" tunneltype="ssl-tunnel" tunnelid=1171992131 remip=192.168.0.1 user="someuser" group="vpn_ssl_group"

```

and

```auto
datetime="2020-09-2020T15:40:00-0300" action="tunnel-down" tunneltype="ssl-tunnel" tunnelid=1171992131 remip=192.168.0.1 user="someuser" group="vpn_ssl_group"

```

I'd like to be able to aggregate fields `datetime` and `action` grouped by the `tunnelid` field into an array of objects.

Final document should look like this:

```auto
{
    "datetime": "2020-09-2020T14:40:00-0300",
    "tunnelid": 1171992131,
    "user": "someuser",
    "group": "vpn_ssl_group",
    "remip": "192.168.0.1",
    "logs": [{
        "datetime": "2020-09-2020T14:40:00-0300",
        "action": "tunnel-up"
    },
    {
        "datetime": "2020-09-2020T15:40:00-0300",
        "action": "tunnel-down"
    }],
    "duration": 3600
}

```

Is there any way I can do that? If so, can you please show me how?

---

<div class="post-metadata">

**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:** [September 28, 2020, 8:23pm UTC](https://discuss.elastic.co/t/logstash-aggregate-array-of-objects-based-on-key-field/250249/2 "2020-09-28T20:23:09Z")

</div>

> [@rfferrao](#):
>
> Is there any way I can do that?

Use an aggregate filter. Look at example 3 in the [documentation](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html).

Make sure you have a single pipeline worker thread. Note that when the filter pushes an event based on the map, the only fields on the event will be what you added to the map.

---

<div class="post-metadata">

**Author:** ![rfferrao](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rfferrao/32/76344_2.png) [@rfferrao](https://discuss.elastic.co/u/rfferrao)\
**Post date:** [September 28, 2020, 9:57pm UTC](https://discuss.elastic.co/t/logstash-aggregate-array-of-objects-based-on-key-field/250249/3 "2020-09-28T21:57:32Z")

</div>

> [@Badger](#):
>
> single pipeline worker

Must I have a single pipeline worker for every configuration file or can I specify the one which has the aggregation filter? If it's possible to split them, can I have a configuration file which keeps the input from the main conf and just adds the aggregation filter and put it in another pipeline?

---

<div class="post-metadata">

**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:** [September 28, 2020, 11:10pm UTC](https://discuss.elastic.co/t/logstash-aggregate-array-of-objects-based-on-key-field/250249/4 "2020-09-28T23:10:16Z")

</div>

The pipeline which contains the aggregate filter must only use a single worker thread. If you need to maintain event order then that applies to other pipelines too.

However, if you are OK with events getting re-ordered, which I expect you are for your use case then other pipelines could have multiple workers.

You can specify the number of workers for each pipeline in pipelines.yml

If the aggregate pipeline feeds the multi-worker pipeline then it is not going to scale well. If the multi-worker pipeline does a lot of work on each event (especially expensive calls like http, geoip, or dns filters) and then feeds the aggregate pipeline it may scale quite well.

---

<div class="post-metadata">

**Author:** ![rfferrao](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rfferrao/32/76344_2.png) [@rfferrao](https://discuss.elastic.co/u/rfferrao)\
**Post date:** [September 28, 2020, 11:33pm UTC](https://discuss.elastic.co/t/logstash-aggregate-array-of-objects-based-on-key-field/250249/5 "2020-09-28T23:33:15Z")

</div>

Thank you for the feedback, Badger, it's been very helpful! Last question, would you recommend pipeline to pipeline communication for this use case? By the way, I'm currently using tcp as the input and elasticsearch as the output plugins in the single configuration file I have set up.

---

<div class="post-metadata">

**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:** [September 29, 2020, 12:11am UTC](https://discuss.elastic.co/t/logstash-aggregate-array-of-objects-based-on-key-field/250249/6 "2020-09-29T00:11:22Z")

</div>

> [@rfferrao](#):
>
> would you recommend pipeline to pipeline communication for this use case?

Test it, measure it. If you cannot see a difference between the two options then choose the simpler one.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [October 27, 2020, 12:11am UTC](https://discuss.elastic.co/t/logstash-aggregate-array-of-objects-based-on-key-field/250249/7 "2020-10-27T00:11:50Z")

</div>

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