in_tail: add option to skip long lines

There is a case that processing unexpected long line causes a BufferChunkOverflow exception. With #3562 (Fluentd v1.14.3), it can just ignore only problematic long record, but it always raise an exception. Thus it has performance regression. In this fix, skip_long_lines option enables to just skip processing event further more. Signed-off-by: Kentaro Hayashi <[email protected]>
fluent · Nov 29, 2021 · be9943a · be9943a
1 parent 438a82a
commit be9943a
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 0 deletions.
diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb
@@ -113,6 +113,8 @@ def initialize
     config_param :path_timezone, :string, default: nil
     desc 'Follow inodes instead of following file names. Guarantees more stable delivery and allows to use * in path pattern with rotating files'
     config_param :follow_inodes, :bool, default: false
+    desc 'Skip long lines which size is longer than specified length'
+    config_param :skip_long_lines, :integer, default: 0
 
     config_section :parse, required: false, multi: true, init: true, param_name: :parser_configs do
       config_argument :usage, :string, default: 'in_tail_parser'
@@ -594,6 +596,9 @@ def flush_buffer(tw, buf)
 
     # @return true if no error or unrecoverable error happens in emit action. false if got BufferOverflowError
     def receive_lines(lines, tail_watcher)
+      lines = lines.reject do |line|
+        @skip_long_lines > 0 ? line.bytesize > @skip_long_lines : false
+      end
       es = @receive_handler.call(lines, tail_watcher)
       unless es.empty?
         tag = if @tag_prefix || @tag_suffix

diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb
@@ -1707,6 +1707,25 @@ def test_tag_prefix_and_suffix_ignore
       mock(plugin.router).emit_stream('pre.foo.bar.log.post', anything).once
       plugin.receive_lines(['foo', 'bar'], DummyWatcher.new('foo.bar.log'))
     end
+
+    def test_skip_long_lines
+      config = config_element("", "", {
+                                "tag" => "skip_long_lines",
+                                "path" => "#{TMP_DIR}/with_long_lines.txt",
+                                "format" => "none",
+                                "read_from_head" => true,
+                                "skip_long_lines" => 128
+                              })
+      File.open("#{TMP_DIR}/with_long_lines.txt", "w+") do |f|
+        f.puts "foo"
+        f.puts "x" * 129
+        f.puts "bar"
+      end
+      d = create_driver(config, false)
+      d.run(expect_records: 2)
+      assert_equal([{"message" => "foo"},{"message" => "bar"}],
+                   d.events.collect { |event| event.last })
+    end
   end
 
   # Ensure that no fatal exception is raised when a file is missing and that