fluent-plugin-time-cutoff

Fluentd filter plugin to cut off log records with a timestamp outside of given window.

Examples

Example 1: Delete all logs older than 12 hours, pass through records from the future

Configuration

<filter **>
  @type time_cutoff
  @id time_cutoff
 
  old_cutoff 12h
  old_action drop

  new_action pass
  new_log false
</filter>

Input

# Local time is 2025-04-08T14:43:14+03:00
fluent-cat --event-time '2025-04-08T02:42:00+03:00' test.cutoff <<< '{"message": "hello"}'
fluent-cat --event-time '2025-04-08T12:42:00+03:00' test.cutoff <<< '{"message": "hello"}'
fluent-cat --event-time '2025-04-09T14:42:00+03:00' test.cutoff <<< '{"message": "hello"}'

Output

2025-04-08 14:43:24 +0300 [warn]: [time_cutoff] Record caught [old, drop]: "test.cutoff: 2025-04-08T02:42:00+03:00 {"message":"hello"}".
2025-04-08 12:42:00.000000000 +0300 test.cutoff: {"message":"hello"}
2025-04-09 14:42:00.000000000 +0300 test.cutoff: {"message":"hello"}

Example 2: Delete all logs older than 24 hours (omit log), rewrite timestamps for all records newer than 3 hours:

Configuration

<filter **>
  @type time_cutoff
  @id time_cutoff
 
  old_cutoff 24h
  old_action drop
  old_log false

  new_cutoff 3h
  new_action replace_timestamp
</filter>

Input

# Local time is 2025-04-08T14:48:42+03:00
fluent-cat --event-time '2025-04-07T14:48:00+03:00' test.cutoff <<< '{"message": "hello"}'
fluent-cat --event-time '2025-04-08T14:48:00+03:00' test.cutoff <<< '{"message": "hello"}'
fluent-cat --event-time '2025-04-08T18:48:00+03:00' test.cutoff <<< '{"message": "hello"}'

Output

2025-04-08 14:48:00.000000000 +0300 test.cutoff: {"message":"hello"}
2025-04-08 14:49:13 +0300 [warn]: [time_cutoff] Record caught [new, replace_timestamp]: "test.cutoff: 2025-04-08T18:48:00+03:00 {"message":"hello"}".
2025-04-08 14:49:13.090783000 +0300 test.cutoff: {"message":"hello","source_time":"2025-04-08T18:48:00+03:00"}

Installation

RubyGems

$ gem install fluent-plugin-time-cutoff

Bundler

Add following line to your Gemfile:

gem "fluent-plugin-time-cutoff"

And then execute:

$ bundle

Configuration

<filter **>
  @type time_cutoff
  @id time_cutoff
 
  old_cutoff 24h
  old_action pass
  old_log full

  new_cutoff 24h
  new_action pass
  new_log full

  source_time_key source_time
  source_time_format iso8601

  bad_time_action replace_timestamp
  bad_time_log full
  bad_time_key source_time_raw

  time_key_conflict skip

  metrics false
</filter>

old_cutoff (time) (optional)

Records older than this amount of time are deemed "old". Defaults to 24 hours.

Must not be negative.

Default value: 86400.

old_action (enum) (optional)

The action that will be performed to all "old" messages. Defaults to passthrough.

Available values: pass, replace_timestamp, drop

Default value: pass.

old_log (enum) (optional)

Log the "old" messages to Fluentd's own log. Defaults to "full".

Available values: none (or false/no), metadata, full (or true/yes)

See Logging for what each mode logs.

Default value: full.

new_cutoff (time) (optional)

Records newer than this amount of time are deemed "new". Defaults to 24 hours.

Must not be negative.

Default value: 86400.

new_action (enum) (optional)

The action that will be performed to all "new" messages. Defaults to passthrough.

Available values: pass, replace_timestamp, drop

Default value: pass.

new_log (enum) (optional)

Log the "new" messages to Fluentd's own log. Defaults to "full".

Available values: none (or false/no), metadata, full (or true/yes)

See Logging for what each mode logs.

Default value: full.

source_time_key (string) (optional)

The key that will hold the original time (for :replace_timestamp action). Defaults to "source_time".

Must not be empty or equal to bad_time_key.

Default value: source_time.

source_time_format (enum) (optional)

Time format for the original timestamp field. Defaults to ISO8601-formatted string.

Available values: epoch, epoch_float, iso8601

Default value: iso8601.

bad_time_action (enum) (optional)

The action that will be performed to all messages with an unparseable time. Defaults to replacing it with current time.

Event times that aren't a Fluent::EventTime are parsed before the age check. Integers, floats, numeric strings and ISO 8601 datetime strings are accepted. ISO 8601 strings may use a space instead of T and may have a space before the timezone (e.g. 2025-04-08 14:43:14 +0300 or 2025-04-08 11:43:14 UTC, as produced by Ruby's Time#to_s). Strings without a timezone are treated as local time. Any other time is deemed unparseable and is handled according to this action (and logged, unless bad_time_log is none).

Available values: replace_timestamp, drop

Default value: replace_timestamp.

bad_time_log (enum) (optional)

Log the messages with an unparseable time to Fluentd's own log. Defaults to "full".

Available values: none (or false/no), metadata, full (or true/yes)

See Logging for what each mode logs.

Default value: full.

bad_time_key (string) (optional)

The key that will hold the original unparseable time (for :replace_timestamp action). Defaults to "source_time_raw".

Must not be empty or equal to source_time_key.

The original time is stored as is if it's a string, or as its Ruby #inspect representation otherwise (e.g. nil), so this field always holds a string. It is kept separate from source_time_key, so that source_time_key always holds a value in source_time_format and doesn't cause type conflicts in downstream storage.

Default value: source_time_raw.

time_key_conflict (enum) (optional)

The action that will be performed when the record already has source_time_key or bad_time_key. Defaults to keeping the existing value.

This happens e.g. when a record passes through more than one time_cutoff filter, in which case the existing value is the true original time. The event time itself is replaced with current time in any case.

• skip: keep the existing value.
• warn: keep the existing value and log a warning to Fluentd's own log. The warning is logged even if the record's log mode is none, but includes the record only in full mode.
• overwrite: replace the existing value.

Available values: skip, warn, overwrite

Default value: skip.

metrics (bool) (optional)

Count caught records in a Prometheus counter (needs the prometheus-client gem). Defaults to "false".

See Metrics for details.

Default value: false.

Logging

Each caught record is logged as a warning to Fluentd's own log, according to the old_log, new_log or bad_time_log mode:

• none (or false/no): nothing is logged.
• metadata: only the tag, the event time and its age (or the raw unparseable time) are logged, but not the record itself. Use it for records that may hold sensitive data, e.g. audit logs.
• full (or true/yes): the whole record is logged as well.

# metadata
[warn]: [time_cutoff] Record caught [old, drop]: "vault.audit: 2025-04-07T14:43:24+03:00 (age 1d 0h 0m 0s)".
[warn]: [time_cutoff] Record caught [new, pass]: "vault.audit: 2025-04-08T16:43:29+03:00 (age -2h 0m 5s)".
[warn]: [time_cutoff] Record caught [bad_time, drop]: "vault.audit: "not-a-time"".

# full
[warn]: [time_cutoff] Record caught [old, drop]: "vault.audit: 2025-04-07T14:43:24+03:00 {"message":"hello"}".
[warn]: [time_cutoff] Record caught [bad_time, drop]: "vault.audit: "not-a-time" {"message":"hello"}".

Metrics

With metrics true, caught records are counted in a Prometheus counter:

fluentd_time_cutoff_records_total{plugin_id, worker_id, hostname, tag, reason, action}

• plugin_id: the filter's @id. Give each time_cutoff filter a stable @id to tell them apart.
• worker_id, hostname: the Fluentd worker and host, so that series from multiple workers (e.g. on fluent-plugin-prometheus's /aggregated_metrics endpoint) and hosts don't collide.
• tag: the record's tag.
• reason: why the record was caught: old, new or bad_time.
• action: what was done to it: pass, replace_timestamp or drop.

Records inside the time window aren't counted. Records are counted regardless of their log mode, so the counter can replace per-record logging.

The counter is registered in the default registry of the prometheus-client gem (version 2.1 or newer), which is what fluent-plugin-prometheus serves, so adding its prometheus input is enough to expose it:

<source>
  @type prometheus
</source>

<filter **>
  @type time_cutoff
  @id cutoff_main
  metrics true
</filter>

The prometheus-client gem is an optional dependency of this plugin. If it's not installed, enabling metrics only logs a warning on startup. Errors while updating the counter are logged once and never affect how records are processed.

Example query, records dropped per second for being too old, by tag:

sum by (tag) (rate(fluentd_time_cutoff_records_total{reason="old", action="drop"}[5m]))

Copyright

• Copyright(c) 2025-2026 Qrator Labs and Serge Tkatchouk
• License: MIT