buffer: add evacuate_limit_size to keep evacuated chunks from filling the disk

lib/fluent/plugin/buf_file.rb

@@ -234,7 +234,7 @@ def evacuate_chunk(chunk)
           raise ArgumentError, "The chunk must be FileChunk, but it was #{chunk.class}."
         end
 
-        backup_dir = File.join(backup_base_dir, 'buffer', safe_owner_id)
+        backup_dir = evacuate_dir
         FileUtils.mkdir_p(backup_dir, mode: system_config.dir_permission || Fluent::DEFAULT_DIR_PERMISSION) unless Dir.exist?(backup_dir)
 
         FileUtils.copy([chunk.path, chunk.meta_path], backup_dir)

lib/fluent/plugin/buf_file_single.rb

@@ -250,7 +250,7 @@ def evacuate_chunk(chunk)
           raise ArgumentError, "The chunk must be FileSingleChunk, but it was #{chunk.class}."
         end
 
-        backup_dir = File.join(backup_base_dir, 'buffer', safe_owner_id)
+        backup_dir = evacuate_dir
         FileUtils.mkdir_p(backup_dir, mode: system_config.dir_permission || Fluent::DEFAULT_DIR_PERMISSION) unless Dir.exist?(backup_dir)
 
         FileUtils.copy(chunk.path, backup_dir)

lib/fluent/plugin/buffer.rb

@@ -72,6 +72,9 @@ class BufferChunkOverflowError < BufferError; end # A record size is larger than
       desc 'If true, chunks are thrown away when unrecoverable error happens'
       config_param :disable_chunk_backup, :bool, default: false
 
+      desc 'The approximate size limit of the evacuated chunk files in the evacuation directory. When the retry limit is reached, queued chunks that fit within the remaining capacity are evacuated, and the others are purged without being evacuated. The total size may exceed this limit due to chunk metadata or concurrent evacuation by multiple workers. Set to 0 to disable evacuation. nil (the default) means no limit.'
+      config_param :evacuate_limit_size, :size, default: nil
+
       Metadata = Struct.new(:timekey, :tag, :variables, :seq) do
         def initialize(timekey, tag, variables)
           super(timekey, tag, variables, 0)
@@ -227,6 +230,20 @@ def configure(conf)
         unless @queue_limit_length.nil?
           @total_limit_size = @chunk_limit_size * @queue_limit_length
         end
+        if @evacuate_limit_size
+          raise Fluent::ConfigError, "evacuate_limit_size must be 0 or larger" if @evacuate_limit_size < 0
+          if !persistent?
+            log.warn "evacuate_limit_size is ignored because this buffer does not evacuate chunks"
+            @evacuate_limit_size = nil
+          elsif @evacuate_limit_size > 0
+            unless owner.plugin_id_configured?
+              log.warn "evacuate_limit_size should be used with @id, because the evacuation directory without @id can change when the configuration is changed or reloaded"
+            end
+            if @evacuate_limit_size < @chunk_limit_size
+              log.warn "evacuate_limit_size is smaller than chunk_limit_size, so full chunks may not be evacuated", evacuate_limit_size: @evacuate_limit_size, chunk_limit_size: @chunk_limit_size
+            end
+          end
+        end
         @stage_length_metrics = metrics_create(namespace: "fluentd", subsystem: "buffer", name: "stage_length",
                                                help_text: 'Length of stage buffers', prefer_gauge: true)
         @stage_length_metrics.set(0)
@@ -642,12 +659,27 @@ def purge_chunk(chunk_id)
       def clear_queue!
         log.on_trace { log.trace "clearing queue", instance: self.object_id }
 
+        evacuate_budget = nil
+        skipped_num = 0
+        skipped_bytesize = 0
+
         synchronize do
           until @queue.empty?
             begin
               q = @queue.shift
-              evacuate_chunk(q)
-              log.trace("purging a chunk in queue"){ {id: dump_unique_id_hex(chunk.unique_id), bytesize: chunk.bytesize, size: chunk.size} }
+              if @evacuate_limit_size.nil?
+                evacuate_chunk(q)
+              elsif @evacuate_limit_size > 0
+                evacuate_budget ||= remaining_evacuate_size
+                if q.bytesize <= evacuate_budget
+                  evacuate_chunk(q)
+                  evacuate_budget -= q.bytesize
+                else
+                  skipped_num += 1
+                  skipped_bytesize += q.bytesize
+                end
+              end
+              log.trace("purging a chunk in queue"){ {id: dump_unique_id_hex(q.unique_id), bytesize: q.bytesize, size: q.size} }
               q.purge
             rescue => e
               log.error "unexpected error while clearing buffer queue", error_class: e.class, error: e
@@ -656,6 +688,10 @@ def clear_queue!
           end
           @queue_size_metrics.set(0)
         end
+
+        if skipped_num > 0
+          log.warn "some chunks are purged without being evacuated to stay within evacuate_limit_size.", evacuate_limit_size: @evacuate_limit_size, chunks: skipped_num, bytesize: skipped_bytesize
+        end
       end
 
       def evacuate_chunk(chunk)
@@ -1004,6 +1040,24 @@ def safe_owner_id
       def backup_base_dir
         system_config.root_dir || DEFAULT_BACKUP_DIR
       end
+
+      def evacuate_dir
+        File.join(backup_base_dir, 'buffer', safe_owner_id)
+      end
+
+      def evacuated_bytesize
+        dir = evacuate_dir
+        Dir.children(dir, encoding: dir.encoding).sum { |name| File.size?(File.join(dir, name)).to_i }
+      rescue Errno::ENOENT
+        0
+      end
+
+      def remaining_evacuate_size
+        @evacuate_limit_size - evacuated_bytesize
+      rescue => e
+        log.warn "failed to get the size of the evacuated chunk files. queued chunks are purged without being evacuated.", error: e
+        0
+      end
     end
   end
 end

test/plugin/test_buf_file.rb

@@ -1696,5 +1696,40 @@ def evacuate_dir(plugin_id)
 
       assert_equal(expected_records, actual_records)
     end
+
+    test 'evacuate_limit_size bounds the evacuated files over repeated retry limits' do
+      plugin_id = "test_output"
+      tag = "test.1"
+      buffer_conf = {
+        "path" => @buf_dir,
+        "flush_mode" => "interval",
+        "flush_interval" => "1s",
+        "retry_type" => "periodic",
+        "retry_max_times" => 1,
+        "retry_randomize" => false,
+        "evacuate_limit_size" => 400,
+      }
+
+      configure_output(plugin_id, "tag", buffer_conf) do |output|
+        start_plugin(output)
+
+        reach_retry_limit = lambda do
+          errors = output.num_errors
+          emit_events(output, tag, dummy_event_stream)
+          wait { output.num_errors == errors + 1 }
+          wait { output.retry }
+          proceed_to_next_retry(output)
+          wait { output.num_errors == errors + 2 }
+          wait { output.retry.nil? }
+          Dir.glob(File.join(evacuate_dir(plugin_id), "*.log")).size
+        end
+
+        assert_equal [1, 2, 2, 2], Array.new(4) { reach_retry_limit.call }
+        assert_equal 2, output.log.out.logs.count { |l| l.include?("to stay within evacuate_limit_size") }
+
+        FileUtils.rm(Dir.glob(File.join(evacuate_dir(plugin_id), "*")))
+        assert_equal 1, reach_retry_limit.call
+      end
+    end
   end
 end

test/plugin/test_buf_file_single.rb

@@ -1219,5 +1219,40 @@ def evacuate_dir(plugin_id)
 
       assert_equal(expected_records, actual_records)
     end
+
+    test 'evacuate_limit_size bounds the evacuated files over repeated retry limits' do
+      plugin_id = "test_output"
+      tag = "test.1"
+      buffer_conf = {
+        "path" => @buf_dir,
+        "flush_mode" => "interval",
+        "flush_interval" => "1s",
+        "retry_type" => "periodic",
+        "retry_max_times" => 1,
+        "retry_randomize" => false,
+        "evacuate_limit_size" => 300,
+      }
+
+      configure_output(plugin_id, "tag", buffer_conf) do |output|
+        start_plugin(output)
+
+        reach_retry_limit = lambda do
+          errors = output.num_errors
+          emit_events(output, tag, dummy_event_stream)
+          wait { output.num_errors == errors + 1 }
+          wait { output.retry }
+          proceed_to_next_retry(output)
+          wait { output.num_errors == errors + 2 }
+          wait { output.retry.nil? }
+          Dir.children(evacuate_dir(plugin_id)).size
+        end
+
+        assert_equal [1, 2, 2, 2], Array.new(4) { reach_retry_limit.call }
+        assert_equal 2, output.log.out.logs.count { |l| l.include?("to stay within evacuate_limit_size") }
+
+        FileUtils.rm(Dir.glob(File.join(evacuate_dir(plugin_id), "*")))
+        assert_equal 1, reach_retry_limit.call
+      end
+    end
   end
 end

test/plugin/test_buffer.rb

@@ -85,6 +85,11 @@ def generate_chunk(metadata)
       DummyMemoryChunk.new(metadata, compress: @compress, decompression_size_limit: @decompression_size_limit)
     end
   end
+  class DummyPersistentPlugin < DummyPlugin
+    def persistent?
+      true
+    end
+  end
 end
 
 # DummyPlugin#resume pre-seeds staged/queued chunks; tests that assert exact
@@ -96,11 +101,11 @@ def resume
 end
 
 class BufferTest < Test::Unit::TestCase
-  def create_buffer(hash)
+  def create_buffer(hash, klass: FluentPluginBufferTest::DummyPlugin)
     buffer_conf = config_element('buffer', '', hash, [])
     owner = FluentPluginBufferTest::DummyOutputPlugin.new
     owner.configure(config_element('ROOT', '', {}, [ buffer_conf ]))
-    p = FluentPluginBufferTest::DummyPlugin.new
+    p = klass.new
     p.owner = owner
     p.configure(buffer_conf)
     p
@@ -513,6 +518,141 @@ def create_chunk_es(metadata, es)
       assert{ qchunks.all?{ |c| c.purged } }
     end
 
+    test '#clear_queue! evacuates every queued chunk by default (no evacuate_limit_size)' do
+      evacuated = []
+      @p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      qchunks = @p.queue.dup
+
+      assert_nil @p.evacuate_limit_size
+
+      @p.clear_queue!
+
+      assert_equal [100, 100, 3], evacuated
+      assert_equal [], @p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+    end
+
+    test '#clear_queue! skips evacuation entirely when evacuate_limit_size is 0, but still purges' do
+      p = create_buffer({'evacuate_limit_size' => 0}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      assert_equal [], evacuated
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+      assert{ $log.out.logs.none? { |l| l.include?("to stay within evacuate_limit_size") } }
+    end
+
+    test '#clear_queue! bounds the total evacuated size to evacuate_limit_size' do
+      p = create_buffer({'evacuate_limit_size' => 200}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.define_singleton_method(:evacuated_bytesize) { 0 }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      # queued chunks are 100, 100 and 3 bytes; the first two fill the 200 byte
+      # budget, so the last one is purged without being evacuated
+      assert_equal [100, 100], evacuated
+      assert_equal 200, evacuated.sum
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+    end
+
+    test '#clear_queue! skips the chunks which do not fit in evacuate_limit_size' do
+      p = create_buffer({'evacuate_limit_size' => 150}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.define_singleton_method(:evacuated_bytesize) { 0 }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      assert_equal [100, 3], evacuated
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+      skipped_logs = $log.out.logs.select { |l| l.include?("to stay within evacuate_limit_size") }
+      assert_equal 1, skipped_logs.size
+      assert{ skipped_logs.first.include?("chunks=1 bytesize=100") }
+    end
+
+    test '#clear_queue! counts already evacuated files toward evacuate_limit_size' do
+      p = create_buffer({'evacuate_limit_size' => 200}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.define_singleton_method(:evacuated_bytesize) { 100 }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      assert_equal [100], evacuated
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+    end
+
+    test '#clear_queue! purges all chunks without evacuation if it fails to get the size of the evacuated files' do
+      p = create_buffer({'evacuate_limit_size' => 200}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.define_singleton_method(:evacuate_dir) { raise Errno::EACCES, "/tmp/fluent/buffer/test" }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      assert_equal [], evacuated
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+      assert{ $log.out.logs.any? { |l| l.include?("failed to get the size of the evacuated chunk files") } }
+      assert{ $log.out.logs.any? { |l| l.include?("to stay within evacuate_limit_size") && l.include?("chunks=3 bytesize=203") } }
+    end
+
+    test 'ignores evacuate_limit_size if the buffer is not persistent' do
+      p = create_buffer({'evacuate_limit_size' => 200})
+
+      assert_nil p.evacuate_limit_size
+      assert{ $log.out.logs.any? { |l| l.include?("evacuate_limit_size is ignored") } }
+    end
+
+    test 'raises ConfigError if evacuate_limit_size is negative' do
+      assert_raise(Fluent::ConfigError) do
+        create_buffer({'evacuate_limit_size' => -1})
+      end
+    end
+
+    data(
+      'without @id' => [{}, true],
+      'with @id' => [{'@id' => 'test_output'}, false],
+    )
+    test 'warns that evacuate_limit_size should be used with @id' do |(owner_conf, warned)|
+      buffer_conf = config_element('buffer', '', {'evacuate_limit_size' => '1g'})
+      owner = FluentPluginBufferTest::DummyOutputPlugin.new
+      owner.configure(config_element('ROOT', '', owner_conf, [buffer_conf]))
+      p = FluentPluginBufferTest::DummyPersistentPlugin.new
+      p.owner = owner
+      p.configure(buffer_conf)
+
+      assert_equal warned, $log.out.logs.any? { |l| l.include?("should be used with @id") }
+    end
+
+    data(
+      'smaller than chunk_limit_size' => ['1m', true],
+      'larger than chunk_limit_size' => ['1g', false],
+    )
+    test 'warns if evacuate_limit_size is smaller than chunk_limit_size' do |(limit, warned)|
+      create_buffer({'evacuate_limit_size' => limit, 'chunk_limit_size' => '8m'}, klass: FluentPluginBufferTest::DummyPersistentPlugin)
+
+      assert_equal warned, $log.out.logs.any? { |l| l.include?("smaller than chunk_limit_size") }
+    end
+
     test '#write returns immediately if argument data is empty array' do
       assert_equal [@dm0,@dm1,@dm1], @p.queue.map(&:metadata)
       assert_equal [@dm2,@dm3], @p.stage.keys