in_tail: add option to skip long lines

There is a case that processing unexpected long line causes a
BufferChunkOverflow exception.

With fluent#3562 (Fluentd v1.14.3), it can just ignore only problematic long
record during processing buffer chunk, 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 in in_tail plugin side.

Signed-off-by: Kentaro Hayashi <[email protected]>

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, :size, default: nil
 
     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,11 @@ 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_line = @skip_long_lines ? line.bytesize > @skip_long_lines : false
+        log.debug "received line length is longer than #{@skip_long_lines}: #{line[0..9]}..." if skip_line
+        skip_line
+      end
       es = @receive_handler.call(lines, tail_watcher)
       unless es.empty?
         tag = if @tag_prefix || @tag_suffix

test/plugin/test_in_tail.rb

@@ -1707,6 +1707,36 @@ 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
+
+    data(
+      small: ["128", 128],
+      KiB: ["1k", 1024]
+    )
+    test 'skip_long_lines' do |(label, size)|
+      config = config_element("", "", {
+                                "tag" => "skip_long_lines",
+                                "path" => "#{TMP_DIR}/with_long_lines.txt",
+                                "format" => "none",
+                                "read_from_head" => true,
+                                "skip_long_lines" => label,
+                                "log_level" => "debug"
+                              })
+      File.open("#{TMP_DIR}/with_long_lines.txt", "w+") do |f|
+        f.puts "foo"
+        f.puts "x" * size # 'x' * size + \n > @skip_long_lines
+        f.puts "bar"
+      end
+      d = create_driver(config, false)
+      d.run(expect_records: 2)
+      assert_equal([
+                     [{"message" => "foo"},{"message" => "bar"}],
+                     true,
+                   ],
+                   [
+                     d.events.collect { |event| event.last },
+                     d.logs.any? { |log| log.end_with?("received line length is longer than #{size}: xxxxxxxxxx...\n") },
+                   ])
+    end
   end
 
   # Ensure that no fatal exception is raised when a file is missing and that