# Aggregation in Ingest Pipeline of Elastic Agent - Custom Logs integration

**URL:** <https://discuss.elastic.co/t/aggregation-in-ingest-pipeline-of-elastic-agent-custom-logs-integration/313910>\
**Category:** Logs\
**Created:** [September 7, 2022, 4:56pm UTC](https://discuss.elastic.co/t/aggregation-in-ingest-pipeline-of-elastic-agent-custom-logs-integration/313910 "2022-09-07T16:56:29Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![rowra](https://avatars.discourse-cdn.com/v4/letter/r/bc79bd/32.png) [@rowra](https://discuss.elastic.co/u/rowra)\
**Post date:** [September 7, 2022, 4:56pm UTC](https://discuss.elastic.co/t/aggregation-in-ingest-pipeline-of-elastic-agent-custom-logs-integration/313910/1 "2022-09-07T16:56:29Z")

</div>

Hey,  
Currently I have a Logstash pipeline parsing, aggregating and finally delivering Postfix logs to the Elasticsearch. I want to retire Logstash completely and use Fleet to deploy a policy with Custom Logs integration. I managed to parse the logs but I'm not at the last step at which I need to aggregate them based on a field.  
(Postfix logs are structured in a way such as one logical process of a mail is split into numerous "physical" processes and thus loglines, only ever to be connected by the "queueid", which is a successfully parsed field by this time)

Basically I need to equivalent in Elastic Ingest Pipeline Processor of the following Logstash snippet:

```auto
filter {
  if ![queueid] {
    drop {}
  } else {
    aggregate {
      task_id => "%{queueid}"
      aggregate_maps_path => "/inputs/.aggregate_maps"
      code => "
        map.merge!(event)
      "
      map_action => "create_or_update"
      push_previous_map_as_event => false
      push_map_as_event_on_timeout => true
      timeout => 30
      timeout_tags => ['aggregated']
    }
  }
}

```

How would one be able to reproduce such functionality in terms of Ingest Pipeline's processors?  
Thanks

---

<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 5, 2022, 4:57pm UTC](https://discuss.elastic.co/t/aggregation-in-ingest-pipeline-of-elastic-agent-custom-logs-integration/313910/2 "2022-10-05T16:57:09Z")

</div>

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