How to drop a given event in logstash based on dat

问题:

I have a Kafka JSON stream that I set as my input in logstash.

I would like to drop events for which dates are before a given date (say, today's midnight).

I can parse the input correctly (it's json, so it's default), and I can print it to stdout with the json codec.

How do I filter the date? Is there something like:

filter {
  if [date] <= "some date" {
    drop { }
  }
}

回答1:

Before your date{} filter, drop into ruby{} and tuck away the server's current time:

event['server_timestamp'] = event['@timestamp']

Then use your date{} filter as normal to reset @timestamp to the event's time.

After that, drop into ruby{} again to compute the difference:

event['lag'] = ( ( event['server_timestamp'] - event['@timestamp'] ) ).to_f

And back in logstash, check the lag against your constraints:

# seconds!
if [lag] > 60 {
     drop{}
}

If you don't want to compare to the server's time, you can use any time you want. When I tried to use ruby's Datetime, it seemed to drop milliseconds, so beware of that.