# Aggregate filter not catching up: input start\_position = "beginning"
**URL:** https://discuss.elastic.co/t/aggregate-filter-not-catching-up-input-start-position-beginning/201428
**Category:** Logstash
**Created:** [September 27, 2019, 3:24pm UTC](https://discuss.elastic.co/t/aggregate-filter-not-catching-up-input-start-position-beginning/201428 "2019-09-27T15:24:41Z")
**Posts on this page:** 7
**Page:** 1
### Author:  [@Paul\_Ainslie](https://discuss.elastic.co/u/Paul_Ainslie)
#### Post date: [September 27, 2019, 3:24pm UTC](https://discuss.elastic.co/t/aggregate-filter-not-catching-up-input-start-position-beginning/201428/1 "2019-09-27T15:24:41Z")
One of the things I love about logstash is the "sincedb" database which tracks current position and — if interrupted — remembers the location of where it left off. So, if Logstash is down for, say 4 hours, once restarted it catches up and nothing is lost.
I discovered the aggregate filter recently, but it's not behaving this way. If I stop logstash for a while, it simply restarts from the beginning. 😬 This should work, shouldn't it?
I'm processing Nginx access.log records (written in JSON) via one pipeline with two outputs. One output is sending the raw records to ES whereas the 5 minute aggregate is sending writing to file. I also tried writing the aggregation to ES.
Should aggregations be in their own pipeline seperate from the raw processing & output?
{"logstash.version"=\>"7.3.2"}
Here's my input configuration.
```auto
input {
file {
path => "/var/log/nginx/access*"
start_position => "beginning"
sincedb_path => "/var/lib/logstash/plugins/inputs/file/.sincedb_nginx_update_alias"
exclude => "*.gz"
codec => json
}
}
```
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [September 28, 2019, 12:42pm UTC](https://discuss.elastic.co/t/aggregate-filter-not-catching-up-input-start-position-beginning/201428/2 "2019-09-28T12:42:30Z")
Using an aggregate filter does not affect what an input reads. By default an aggregate starts from scratch when logstash restarts. You can use the [aggregate\_maps\_path](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html#plugins-filters-aggregate-aggregate_maps_path) to persist the contents of the maps across restarts.
---
### Author:  [@Paul\_Ainslie](https://discuss.elastic.co/u/Paul_Ainslie)
#### Post date: [October 1, 2019, 12:48pm UTC](https://discuss.elastic.co/t/aggregate-filter-not-catching-up-input-start-position-beginning/201428/3 "2019-10-01T12:48:53Z")
Agreed, yes, I'm using `aggregate_maps_path`. Here's my configuration. And when I start logstash up it always confirms that `[INFO][logstash.filters.aggregate] Aggregate maps loaded from : /usr/share/logstash/aggregate_maps.db`.
```auto
input {
file {
path => "/var/log/nginx/access*"
start_position => "beginning"
sincedb_path => "/var/lib/logstash/plugins/inputs/file/.sincedb_v2"
exclude => "*.gz"
codec => json
}
}
filter {
dissect {
mapping => { "request" => "%{} /%{streamtype}/%{account}/%{stream}/%{unparsed_request} %{protocol}/%{httpversion}" }
}
geoip { source => "clientip" }
if [agent] != "-" and [agent] != "" {
useragent {
source => "agent"
target => "ua"
}
}
fingerprint {
concatenate_sources => true
source => ['clientip', 'agent', 'account', 'stream', 'cache_status', 'response']
method => "MURMUR3"
target => "eventid_5m"
}
# 5 minute aggregation
aggregate {
aggregate_maps_path => "/usr/share/logstash/aggregate_maps.db" # must be in a folder where user logstash has access
task_id => "%{eventid_5m}"
code => "
map['clientip_n_agent'] = event.get('clientip_n_agent')
map['account'] = event.get('account')
map['stream'] = event.get('stream')
map['cache_status'] = event.get('cache_status')
map['response'] = event.get('response')
map['streamtype'] = event.get('streamtype')
map['httpversion'] = event.get('httpversion')
map['host'] = event.get('host')
geoip_hash = {}
geoip_hash.store('continent_code', event.get('[geoip][continent_code]'))
geoip_hash.store('country_name', event.get('[geoip][country_name]'))
geoip_hash.store('country_code', event.get('[geoip][country_code2]'))
geoip_hash.store('region_name', event.get('[geoip][region_name]')) if event.get('[geoip][region_name]')
geoip_hash.store('city_name', event.get('[geoip][city_name]')) if event.get('[geoip][city_name]')
map['geoip'] = geoip_hash
ua_hash = {}
ua_hash.store('os', event.get('[ua][os]' ) ) if event.get('[ua][os]')
ua_hash.store('name', event.get('[ua][name]') ) if event.get('[ua][name]')
if event.get('[ua][os_major]')
os_ver = [event.get('[ua][os_major]').to_s, event.get('[ua][os_minor]').to_s].join('.')
ua_hash.store('ver', os_ver)
end
map['ua'] = ua_hash
map['bytes_sum'] ||= 0; map['bytes_sum'] += event.get('bytes').to_i;
"
push_map_as_event_on_timeout => true
timeout_task_id_field => "eventid_5m"
timeout => 60
timeout_tags => ['_5m_agg']
timeout_code => "
event.set('bytes', event.get('bytes_sum'))
"
}
}
output {
if "_5m_agg" in [tags] {
elasticsearch {
hosts => ["IP1", IP2", "IP3"]
index => "ms_agg_5m"
template_overwrite => true
}
# file {
# path => "/var/log/logstash/access_summary_5m.log"
# codec => "json_lines"
# }
}
}
```
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [October 1, 2019, 12:52pm UTC](https://discuss.elastic.co/t/aggregate-filter-not-catching-up-input-start-position-beginning/201428/4 "2019-10-01T12:52:41Z")
Does timeout\_timestamp\_field help?
---
### Author:  [@Paul\_Ainslie](https://discuss.elastic.co/u/Paul_Ainslie)
#### Post date: [October 2, 2019, 1:48am UTC](https://discuss.elastic.co/t/aggregate-filter-not-catching-up-input-start-position-beginning/201428/5 "2019-10-02T01:48:04Z")
Ah, adding `timeout_timestamp_field` took me one step further. Now when I start logstash, all the missing events (while logstash was down) are sent to ES. However, they all have a current `@timestamp` in ES. 😒 I'm trying to figure that out now.
---
### Author:  [@Paul\_Ainslie](https://discuss.elastic.co/u/Paul_Ainslie)
#### Post date: [October 2, 2019, 3:14am UTC](https://discuss.elastic.co/t/aggregate-filter-not-catching-up-input-start-position-beginning/201428/6 "2019-10-02T03:14:38Z")
It's working now! I needed to map the @timestamp field as follows, that way Elasticsearch doesn't use the @timestamp generated by Logstash at time of event, but instead uses the one from the logfile.
```auto
map['@timestamp'] = event.get('@timestamp')
```
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [October 30, 2019, 3:14am UTC](https://discuss.elastic.co/t/aggregate-filter-not-catching-up-input-start-position-beginning/201428/7 "2019-10-30T03:14:42Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.