Backport(v1.16): formatter_csv: fix memory leak (#4864)

lib/fluent/compat/formatter.rb

@@ -101,6 +101,12 @@ class LabeledTSVFormatter < Fluent::Plugin::LabeledTSVFormatter
 
       class CsvFormatter < Fluent::Plugin::CsvFormatter
         # TODO: warn when deprecated
+
+        # Do not cache because it is hard to consider the thread key correctly.
+        # (We can try, but it would be low priority.)
+        def csv_cacheable?
+          false
+        end
       end
 
       class SingleValueFormatter < Fluent::Plugin::SingleValueFormatter

lib/fluent/plugin/formatter_csv.rb

@@ -35,6 +35,22 @@ class CsvFormatter < Formatter
       config_param :fields, :array, value_type: :string
       config_param :add_newline, :bool, default: true
 
+      def csv_cacheable?
+        !!owner
+      end
+
+      def csv_thread_key
+        csv_cacheable? ? "#{owner.plugin_id}_csv_formatter_#{@usage}_csv" : nil
+      end
+
+      def csv_for_thread
+        if csv_cacheable?
+          Thread.current[csv_thread_key] ||= CSV.new("".force_encoding(Encoding::ASCII_8BIT), **@generate_opts)
+        else
+          CSV.new("".force_encoding(Encoding::ASCII_8BIT), **@generate_opts)
+        end
+      end
+
       def configure(conf)
         super
 
@@ -51,12 +67,10 @@ def configure(conf)
 
         @generate_opts = {col_sep: @delimiter, force_quotes: @force_quotes, headers: @fields,
                           row_sep: @add_newline ? :auto : "".force_encoding(Encoding::ASCII_8BIT)}
-        # Cache CSV object per thread to avoid internal state sharing
-        @cache = {}
       end
 
       def format(tag, time, record)
-        csv = (@cache[Thread.current] ||= CSV.new("".force_encoding(Encoding::ASCII_8BIT), **@generate_opts))
+        csv = csv_for_thread
         line = (csv << record).string.dup
         # Need manual cleanup because CSV writer doesn't provide such method.
         csv.rewind
@@ -65,7 +79,7 @@ def format(tag, time, record)
       end
 
       def format_with_nested_fields(tag, time, record)
-        csv = (@cache[Thread.current] ||= CSV.new("".force_encoding(Encoding::ASCII_8BIT), **@generate_opts))
+        csv = csv_for_thread
         values = @accessors.map { |a| a.call(record) }
         line = (csv << values).string.dup
         # Need manual cleanup because CSV writer doesn't provide such method.