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 <hayashi@clear-code.com>

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

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