# Pipeline to Pipeline forked configuration to enable use of the aggregation filter

**URL:** <https://discuss.elastic.co/t/pipeline-to-pipeline-forked-configuration-to-enable-use-of-the-aggregation-filter/358202>\
**Category:** Logstash\
**Tags:** aggregations\
**Created:** [April 25, 2024, 10:40am UTC](https://discuss.elastic.co/t/pipeline-to-pipeline-forked-configuration-to-enable-use-of-the-aggregation-filter/358202 "2024-04-25T10:40:44Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![finejason](https://avatars.discourse-cdn.com/v4/letter/f/df788c/32.png) [@finejason](https://discuss.elastic.co/u/finejason)\
**Post date:** [April 25, 2024, 10:40am UTC](https://discuss.elastic.co/t/pipeline-to-pipeline-forked-configuration-to-enable-use-of-the-aggregation-filter/358202/1 "2024-04-25T10:40:44Z")

</div>

I've set up a pipeline to pipeline configuration following advice given in this topic [Pipeline.workers configuration and aggregation filter](https://discuss.elastic.co/t/pipeline-workers-configuration-and-aggregation-filter/284461)

What I'm trying to solve is to be able to use the aggregate filter but not degrade performance by running everything through a single worker.

I have the input pipeline set up with 4 workers and it sends to 2 other pipelines, multi\_worker\_pipeline and single\_worker\_pipeline.  
[Pipeline to Pipeline guide](https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html#forked-path-pattern)

The multi worker pipeline is configured with 4 workers and the single worker is configured with 1 worker.

I'm trying to aggregate a message that is seen every minute and then produce an event when the line is not seen after 70 seconds using the inactivity\_timeout feature of the aggregate filter.

The issue I see is that the lines are not being aggregated as expected which makes the inactivity timeout fire for between 1 and 3 lines of the message I want to aggregate.

```auto
if [channel_name] =~ /.*/ {

	aggregate {
		task_id => 'aggregated_%{channel_name}_%{application_name}'
		code => '
			map["lines"] ||= 0
			map["lines"] += 1
			map["times_seen"] ||= []
			map["times_seen"] << Time.now()
		'
		timeout_task_id_field => 'lines_id'
		timeout => 3600
		inactivity_timeout => 70
		timeout_code => '
			event.set("timed_out_at", Time.now())
		'
		timeout_tags => ['_aggregatetimeout']
		timeout_timestamp_field => '@timestamp'
		push_map_as_event_on_timeout => true
	}

}

```

Pipelines

```auto
- pipeline.id: intake
  path.config: "/usr/share/logstash/pipeline/intake.conf"
  pipeline.workers: 4
  pipeline.ordered: false
  queue.type: persisted
- pipeline.id: multi-worker
  path.config: "/usr/share/logstash/pipeline/multi-worker.conf"
  pipeline.workers: 4
  pipeline.ordered: false
  queue.type: persisted
- pipeline.id: single-worker
  path.config: "/usr/share/logstash/pipeline/single-worker.conf"
  pipeline.workers: 1
  pipeline.ordered: true
  queue.type: persisted

```

Intake output

```auto
output { 
    pipeline { 
        send_to => ["multi-worker", "single-worker"] 
    }
}

```

All lines are seen every minute.  
All lines have the same 'application\_name' and 'channel\_name' used in task\_id.  
All lines are getting added to an aggregated map BUT not a single map as I would expect.  
This makes the inactivity timeout fire multiple times when it should not fire at all unless the log line is not seen for more than 70 seconds.

Any ideas of what could be going wrong?

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [April 25, 2024, 1:01pm UTC](https://discuss.elastic.co/t/pipeline-to-pipeline-forked-configuration-to-enable-use-of-the-aggregation-filter/358202/2 "2024-04-25T13:01:32Z")

</div>

This is also happens if you set all pipelines to use a single worker?

The `aggregate` filter requires a single worker, but if the data is coming from other pipelines that are not using a single worker, this may also impact in how it will work.

---

<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:** [April 25, 2024, 4:17pm UTC](https://discuss.elastic.co/t/pipeline-to-pipeline-forked-configuration-to-enable-use-of-the-aggregation-filter/358202/3 "2024-04-25T16:17:01Z")

</div>

> [@finejason](#):
>
> ```auto
> - pipeline.id: intake
> path.config: "/usr/share/logstash/pipeline/intake.conf"
> pipeline.workers: 4
> pipeline.ordered: false
> 
> ```

This will result in unordered events being sent to the "single\_worker" pipeline. The "intake" pipeline has to maintain order, which means it has to have a single worker. And yes, that will impact the overall throughput.

---

<div class="post-metadata">

**Author:** ![finejason](https://avatars.discourse-cdn.com/v4/letter/f/df788c/32.png) [@finejason](https://discuss.elastic.co/u/finejason)\
**Post date:** [April 25, 2024, 6:03pm UTC](https://discuss.elastic.co/t/pipeline-to-pipeline-forked-configuration-to-enable-use-of-the-aggregation-filter/358202/4 "2024-04-25T18:03:22Z")

</div>

Thank you for the clarification.

---

<div class="post-metadata">

**Author:** ![finejason](https://avatars.discourse-cdn.com/v4/letter/f/df788c/32.png) [@finejason](https://discuss.elastic.co/u/finejason)\
**Post date:** [April 25, 2024, 6:05pm UTC](https://discuss.elastic.co/t/pipeline-to-pipeline-forked-configuration-to-enable-use-of-the-aggregation-filter/358202/5 "2024-04-25T18:05:21Z")

</div>

Thank you.