formatter_csv: fix memory leak

[Watson1978]

Which issue(s) this PR fixes:
Fixes #3627

What this PR does / why we need it:
Seems in_exec plugin creats Thread object each interval and it invoke formatter_csv on the thread.

If it use a Thread object as hash key when caching with formatter_csv, that thread will be permanently referenceable and will not be collected by the GC.

Here is memory usage of Ruby's process.

Reproduce

See #3627

<system>
  process_name fluentd_monitor
</system>

<source>
  @type exec
  tag hardware.memory
  run_interval 0.1
  command /home/watson/prj/sandbox/fluentd/get_memory_metrics.sh
  <parse>
    @type json
    types mem_available:float,swap_total:float,swap_free:float
  </parse>
</source>

<match **>
  @type exec
  command /home/watson/prj/sandbox/fluentd/psql_memory.sh
  <format>
    @type csv
    force_quotes false
    delimiter |
    fields time_local,mem_available,swap_total,swap_free
  </format>
  <buffer time>
    @type file
    path /home/watson/prj/sandbox/fluentd/log
    timekey 50s
    timekey_wait 0s
    flush_mode lazy
    total_limit_size 200MB
    retry_type periodic  # exponential backoff is broken https://github.com/fluent/fluentd/issues/3609
    retry_max_times 0  # need to go to secondary as soon as possible https://github.com/fluent/fluentd/issues/3610
  </buffer>
</match>

Docs Changes:

Release Note:
formatter_csv: fix memory leak

[daipom]

Thanks! Good catch!

It seems that in #2529, input plugins were assumed to spawn only a fixed number of threads for emitting.
However, in_exec for example, makes threads for emitting without limit.
The cache would then become infinitely large.
We need to fix this cache.

[daipom]

The thread key would need to be unique for each plugin instance.

The following out_http code has a similar feature.
I want to align the code with this one unless there is a special reason not to.

fluentd/lib/fluent/plugin/out_http.rb

Lines 109 to 111 in 2e896ec

	def connection_cache_id_thread_key
	"#{plugin_id}_connection_cache_id"
	end

For example, I want the following getter.

def csv_thread_key
  "#{plugin_id}_csv"
end

[Watson1978]

@daipom Thanks. I fixed the key name at last commit.

[daipom]

Sorry, I forgot that this is a owned plugin.
It would not be expected to have an owned plugin with PluginId.

So, could you please fix the following points?

(Owned plugins are unique to the owner's id and usage.)

[Watson1978]

When it invoke CsvFormatter via Fluent::Compat::TextFormatter, looks like it does not set owner...

[daipom]

When it invoke CsvFormatter via Fluent::Compat::TextFormatter, looks like it does not set owner...

Sorry, I missed this point.
I have pushed another idea to fix this issue.

• 80c02fb

@Watson1978 Could you please review this commit?
If there seem to be some problems, we can revert it.

The aim of this commit is to ensure that the compat layer does not affect this cache feature.
(Because I'm not sure how to make the key unique on the compat layer context.)

[daipom]

Thanks for fixing the CI failures my commit caused!

It seems possible that there is no owner. (FluentBinlogReader)
It would be a good solution to disable the cache feature in such a case.

I commented on some minor points.
Please check them.

[daipom]

LGTM! Thanks for your contribution!