Add with-source-only feature

[daipom]

Which issue(s) this PR fixes:
None.
Related to

• Update/Reload without downtime #4622

This feature may be useful on its own, but mainly, this feature will help #4622 feature.

What this PR does / why we need it:
Add --with-source-only option to fluentd command.
(Ref: Similar to existing --without-source option)
(We can use system-config option with_source_only as well)

It launches Fluentd with Input plugins only.
Here is the specification.

• Those Input plugins emits the events to SourceOnlyBufferAgent.
• The events is kept in the buf_file during the source-only-mode.
• Send SIGWINCH to the supervisor to cancels the mode.
• After canceled, the new agent starts to load the buffer.
• Not supported on Windows. (It would be possible, but I want to limit the cost of implementation and testing for now)

Docs Changes:
TODO

Release Note:

• Add with-source-only feature
  • fluentd command: add --with-source-only option
  • system config: add with_source_only option
• embedded plugin: add out_buffer plugin, which can be used for buffering and relabeling events.

new options for system config

We can use with_source_only system config option instead of --with-source-only command line option.

We can customize some buffer configs by setting source_only_buffer section.
If there is no need to customize, we don't need to set these configs.

• with_source_only: default nil
• source_only_buffer section
  • flush_thread_count: default 1
  • overflow_action: default drop_oldest_chunk
  • path: default nil (use unique path for the instance)
  • flush_interval: default nil (use the default of BufferOutput plugin: 10)
  • chunk_limit_size: default nil (use the default of BufferOutput plugin: 256 * 1024 * 1024)
  • total_limit_size: default nil (use the default of BufferOutput plugin: 64 * 1024 * 1024 * 1024)
  • compress: default nil (use the default of BufferOutput plugin: text)

example:

<system>
  <source_only_buffer>
    chunk_limit_size 100m
    compress gzip
    flush_thread_count 4
  </source_only_buffer>
</system>

new plugin: out_buffer

This feature internally uses the new plugin out_buffer.
This plugin can be used for buffering and relabeling events.

Example: in_udp -> out_buffer -> out_stdout

<source>
  @type udp
  tag foo.udp
  @label @buffer
  <parse>
    @type none
  </parse>
</source>

<match foo.**>
  @type stdout
</match>

<label @buffer>
  <match **>
    @type buffer
    @label @ROOT
    <buffer>
      path /path/to/buffer
    </buffer>
  </match>
</label>

Limitaion

The following warning logs occur when the agent starts to load the buffer after sending SIGWINCH.

2024-11-07 18:45:07 +0900 [warn]: #0 restoring buffer file: path = /tmp/fluent/source-only-buffer/...

This is because the setting of flush_at_shutdown is changed from false to true internally.

Behavior example

Config

<source>
  @type sample
  tag test
  auto_increment_key number
</source>

<match test.**>
  @type stdout
</match>

Behavior

$ bundle exec fluentd -c test.conf --with-source-only
...
{datetime} [info]: starting fluentd-1.17.1 pid=195838 ruby="3.2.2"
{datetime} [info]: spawn command to main:  cmdline=[...]
{datetime} [info]: #0 init worker0 logger path=nil rotate_age=nil rotate_size=nil
{datetime} [info]: adding match pattern="test.**" type="stdout"
{datetime} [info]: adding match pattern="**" type="buffer"
{datetime} [info]: #0 with-source-only: the emitted data will be stored in the buffer files under
/tmp/fluent/source-only-buffer/470d7c43-f10c-4f7f-9d00-75731d4fdc3b. You can send SIGWINCH to
the supervisor process to cancel with-source-only mode and process data.
{datetime} [info]: adding source type="sample"
{datetime} [warn]: #0 both of Plugin @id and path for <storage> are not specified. Using on-memory store.
{datetime} [warn]: #0 both of Plugin @id and path for <storage> are not specified. Using on-memory store.
{datetime} [info]: #0 starting fluentd worker pid=195858 ppid=195838 worker=0
{datetime} [info]: #0 fluentd worker is now running worker=0

### (Wait a moment and send SIGWINCH to the supervisor (kill -WINCH 195838)) ###

{datetime} [info]: #0 try to cancel with-source-only mode
{datetime} [info]: #0 cancel with-source-only mode and start the other plugins
{datetime} test: {"message":"sample","number":11}
{datetime} test: {"message":"sample","number":12}
{datetime} test: {"message":"sample","number":13}
{datetime} test: {"message":"sample","number":14}
{datetime} test: {"message":"sample","number":15}
{datetime} [info]: #0 shutting down output plugin type=:buffer plugin_id="object:d48"
{datetime} [info]: #0 starts the loading agent for with-source-only
{datetime} [info]: adding match pattern="**" type="buffer"
{datetime} [warn]: #0 restoring buffer file: path =
/tmp/fluent/source-only-buffer/470d7c43-f10c-4f7f-9d00-75731d4fdc3b/buffer.b626515249c8f8e1b5b487087c033fad5.log
{datetime} test: {"message":"sample","number":16}
{datetime} test: {"message":"sample","number":0} # <--- The data while source-only is loaded here!!
{datetime} test: {"message":"sample","number":1}
{datetime} test: {"message":"sample","number":2}
{datetime} test: {"message":"sample","number":3}
{datetime} test: {"message":"sample","number":4}
{datetime} test: {"message":"sample","number":5}
{datetime} test: {"message":"sample","number":6}
{datetime} test: {"message":"sample","number":7}
{datetime} test: {"message":"sample","number":8}
{datetime} test: {"message":"sample","number":9}
{datetime} test: {"message":"sample","number":10}
{datetime} test: {"message":"sample","number":17}
{datetime} test: {"message":"sample","number":18}
{datetime} test: {"message":"sample","number":19}

### Ctrl + C ###

{datetime} [info]: Received graceful stop
{datetime} test: {"message":"sample","number":20}
{datetime} [info]: #0 fluentd worker is now stopping worker=0
{datetime} [info]: #0 shutting down fluentd worker worker=0
{datetime} [info]: #0 shutting down input plugin type=:sample plugin_id="object:d70"
{datetime} [info]: #0 shutting down output plugin type=:buffer plugin_id="object:e10"
{datetime} [info]: #0 shutting down output plugin type=:stdout plugin_id="object:d20"
{datetime} [info]: Worker 0 finished with status 0

Behavior (Stop Fluentd without sending SIGWINCH)

$ bundle exec fluentd -c test.conf --with-source-only
...
{datetime} [info]: starting fluentd-1.17.1 pid=196617 ruby="3.2.2"
{datetime} [info]: spawn command to main:  cmdline=[...]
{datetime} [info]: #0 init worker0 logger path=nil rotate_age=nil rotate_size=nil
{datetime} [info]: adding match pattern="test.**" type="stdout"
{datetime} [info]: adding match pattern="**" type="buffer"
{datetime} [info]: #0 with-source-only: the emitted data will be stored in the buffer files under
/tmp/fluent/source-only-buffer/bbd9006d-bc41-418b-b346-f80888641dda. You can send SIGWINCH to
the supervisor process to cancel with-source-only mode and process data.
{datetime} [info]: adding source type="sample"
{datetime} [warn]: #0 both of Plugin @id and path for <storage> are not specified. Using on-memory store.
{datetime} [warn]: #0 both of Plugin @id and path for <storage> are not specified. Using on-memory store.
{datetime} [info]: #0 starting fluentd worker pid=196638 ppid=196617 worker=0
{datetime} [info]: #0 fluentd worker is now running worker=0

### Ctrl + C ###

{datetime} [info]: Received graceful stop
{datetime} [info]: #0 fluentd worker is now stopping worker=0
{datetime} [info]: #0 shutting down fluentd worker worker=0
{datetime} [info]: #0 shutting down input plugin type=:sample plugin_id="object:d70"
{datetime} [info]: #0 shutting down output plugin type=:buffer plugin_id="object:d48"
{datetime} [warn]: #0 some buffer files remain in /tmp/fluent/source-only-buffer/bbd9006d-bc41-418b-b346-f80888641dda.
Please consider recovering or saving the buffer files in the directory. To recover them, you can set
the buffer path manually to system config and retry, i.e., restart Fluentd with with-source-only mode
and send SIGWINCH again. Config Example:
<system>
  <source_only_buffer>
    path /tmp/fluent/source-only-buffer/bbd9006d-bc41-418b-b346-f80888641dda
  </source_only_buffer>
</system>

{datetime} [info]: Worker 0 finished with status 0

TODO

• Consider race condition
• Fix and add tests
• Consider how to stop the loading agent
  • Not supported because it is not mandatory.
• Consider configurablity of SourceOnlyBufferAgent.
• Consider Windows support
  • Don't support Windows for now. (It would be possible, but I want to limit the cost of implementation and testing for now)

[daipom]

I have added tests.

When making tests, I noticed cancel_source_only! didn't start plugins correctly, so I fixed it (production code diff).
(It was because the @with_source_only flag became false in the last of cancel_source_only!).

https://github.com/fluent/fluentd/pull/4661/files#diff-55c812cbe4793ae507d193d9dd642ce06c21d8e7bb41cd385740667e4f9e5033R266-R267

[kenhys]

LGTM.

[daipom]

Thanks for your review!
@Watson1978 @kenhys