source cloudwatch_logs after end_time has passed

[markuman]

Problem

I read logs from 1 day from cwl and want to write it to s3. This is the only task my fluentd has.
That means there comes a time when all logs are read from cwl and are written to s3. but how do you recognize that?

<source>
  @type cloudwatch_logs
  tag cloudwatch.in
  log_group_name /ecs/my_log
  use_log_stream_name_prefix true
  state_file fluent/group_stream.in.state
  start_time "2021-03-17 00:00:00Z"
  end_time "2021-03-17 23:59:59Z"
  time_range_format "%Y-%m-%d %H:%M:%S%z"
  throttling_retry_seconds 100
</source>

<match cloudwatch.in>
  @type s3
  s3_bucket my_bucket
  s3_region "#{ENV['AWS_DEFAULT_REGION']}"
  path "mbtest/${tag}/%Y/%m/%d/%H/"
  s3_object_key_format %{path}${tag}_%{index}.%{file_extension}
  <format>
    @type single_value
    message_key log
  </format>
  <buffer tag,time>
    @type file
    path "/fluent/s3/"
    timekey 3600 # 1 hour partition
    timekey_wait 10m
    timekey_use_utc false
    chunk_limit_size 256m
  </buffer>
</match>

[Watson1978]

Thanks for the report — this is expected behavior rather than a bug, so let me clarify how start_time / end_time work here.

This input plugin runs as a continuous poller, not as a one-shot batch job. start_time and end_time are just passed through to the CloudWatch Logs GetLogEvents API as filters on which events are returned; they do not make Fluentd stop once end_time is reached.
The internal loop keeps polling every fetch_interval seconds (default 60) until the plugin is shut down:

def run
  @next_fetch_time = Time.now
  until @finished          # @finished only becomes true on shutdown
    ...
    events = get_events(log_group_name, log_stream_name)
    events.each { |event| emit(...) }
    ...
    sleep 1
  end
end

Because of this, there is currently no built-in "the window has been fully read" event or signal. Once every event inside the [start_time, end_time) range has been read, next_forward_token stops advancing and no new records are emitted, but the plugin simply keeps polling (harmlessly) forever until you stop it.

For a bounded backfill like yours, the practical way to recognize completion is:

• Watch for the point where new records stop flowing — e.g. the position kept in your state_file / next-token storage stops changing and no new S3 objects are being created. At that point the window has been fully read.
• Then stop Fluentd (SIGTERM). To make sure the last buffered chunk is actually written to S3 on stop, add flush_at_shutdown true to the <buffer> of your s3 output.

One note on the S3 side: your chunks are keyed by event time (<buffer tag,time> with timekey 3600 / timekey_wait 10m). Since your target day (2021-03-17) is in the past, timekey + timekey_wait has already elapsed, so chunks become flushable promptly instead of waiting in real time — but flush_at_shutdown true is still the safest way to guarantee the final partial chunk is flushed when you shut Fluentd down.

In short: the plugin will not auto-terminate at end_time; please treat this as a long-running job that you stop yourself once the source stops producing new records. If an "automatically stop (or emit a completion marker) after end_time" behavior would be useful, that would be a feature request rather than existing behavior — feel free to open one if you'd like it considered.