ignore incoming logstash entries that are older than a given date

Question

I want Logstash, when it's processing input entries, to simply drop entries that are older than N days.

I assume I'll use the date module and obviously drop, but I don't know how to connect them.

Answer 1

The only way that I know to do date level comparison is via Ruby code. You need the date filter to parse the timestamp (that's its own issue).

Once you parse the date into a field (e.g., event["@timestamp"]), then you can use it to determine if you want to ignore it or not:

5.0:

ruby {
  code => "event.cancel if (Time.now.to_f - event.get('@timestamp').to_f) > (60 * 60 * 24 * 5)"
}

Pre-5.x:

ruby {
  code => "event.cancel if (Time.now.to_f - event['@timestamp'].to_f) > (60 * 60 * 24 * 5)"
}

In this case, 5 is N.

Also, it's worth pointing out that this is relative to the machine time where Logstash happens to be running. If it's inaccurate, then it will impact date math. Similarly, if the source machine's system clock is wrong, then it too can be a problem.

Drawing on Alain's good point, you could use this store the lag time, in addition to just dropping based on it.

5.0:

ruby {
  code => "event.set('lag_seconds', Time.now.to_f - event.get('@timestamp').to_f))"
}

# 5 represents the number of days to allow
if [lag_seconds] > (60 * 60 * 24 * 5) {
  drop { }
}

Pre-5.x:

ruby {
  code => "event['lag_seconds'] = Time.now.to_f - event['@timestamp'].to_f)"
}

# 5 represents the number of days to allow
if [lag_seconds] > (60 * 60 * 24 * 5) {
  drop { }
}

Using this approach, you would then be indexing lag_seconds, which is a fractional amount, thereby allowing you to analyze lag in your index if this goes into ES or some other data store.