Experimental: Replace Yajl to JSON for handling JSON stream

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

Author: Watson1978

Gemfile

@@ -3,6 +3,7 @@ source 'https://rubygems.org/'
 gemspec
 
 gem 'benchmark'
+gem 'json', git: 'https://github.com/byroot/json.git', branch: 'resumable-parser'
 
 local_gemfile = File.join(File.dirname(__FILE__), "Gemfile.local")
 if File.exist?(local_gemfile)

fluentd.gemspec

@@ -29,7 +29,6 @@ Gem::Specification.new do |gem|
 
   gem.add_runtime_dependency("bundler")
   gem.add_runtime_dependency("msgpack", [">= 1.3.1", "< 2.0.0"])
-  gem.add_runtime_dependency("yajl-ruby", ["~> 1.0"])
   gem.add_runtime_dependency("cool.io", [">= 1.4.5", "< 2.0.0"])
   gem.add_runtime_dependency("serverengine", [">= 2.3.2", "< 3.0.0"])
   gem.add_runtime_dependency("http_parser.rb", [">= 0.5.1", "< 0.9.0"])

lib/fluent/compat/exec_util.rb

@@ -16,7 +16,6 @@
 
 require 'msgpack'
 require 'json'
-require 'yajl'
 
 require 'fluent/engine'
 require 'fluent/plugin'
@@ -78,9 +77,15 @@ def each_line(line)
 
       class JSONParser < Parser
         def call(io)
-          y = Yajl::Parser.new
-          y.on_parse_complete = @on_message
-          y.parse(io)
+          parser = JSON::Ext::ResumableParser.new({})
+          buffer_size = 8192
+
+          while (chunk = io.read(buffer_size))
+            parser << chunk
+            while parser.parse
+              @on_message.call(parser.value)
+            end
+          end
         end
       end
 

lib/fluent/config/literal_parser.rb

@@ -17,7 +17,6 @@
 require 'stringio'
 
 require 'json'
-require 'yajl'
 require 'socket'
 require 'ripper'
 

lib/fluent/load.rb

@@ -4,7 +4,6 @@
 require 'stringio'
 require 'fileutils'
 require 'json'
-require 'yajl'
 require 'uri'
 require 'msgpack'
 require 'strptime'

lib/fluent/plugin/in_forward.rb

@@ -17,7 +17,7 @@
 
 require 'fluent/plugin/input'
 require 'fluent/msgpack_factory'
-require 'yajl'
+require 'json'
 require 'digest'
 require 'securerandom'
 
@@ -248,13 +248,15 @@ def read_messages(conn, &block)
         unless feeder
           first = data[0]
           if first == '{' || first == '[' # json
-            parser = Yajl::Parser.new
-            parser.on_parse_complete = ->(obj){
-              block.call(obj, bytes, serializer)
-              bytes = 0
-            }
+            parser = JSON::Ext::ResumableParser.new({})
             serializer = :to_json.to_proc
-            feeder = ->(d){ parser << d }
+            feeder = ->(d){
+              parser << d
+              while parser.parse
+                block.call(parser.value, bytes, serializer)
+                bytes = 0
+              end
+            }
           else # msgpack
             parser = Fluent::MessagePackFactory.msgpack_unpacker
             serializer = :to_msgpack.to_proc

lib/fluent/plugin/in_unix.rb

@@ -19,7 +19,7 @@
 require 'fluent/msgpack_factory'
 
 require 'cool.io'
-require 'yajl'
+require 'json'
 require 'fileutils'
 require 'socket'
 
@@ -158,8 +158,7 @@ def on_read(data)
         first = data[0]
         if first == '{'.freeze || first == '['.freeze
           m = method(:on_read_json)
-          @parser = Yajl::Parser.new
-          @parser.on_parse_complete = @on_message
+          @parser = JSON::Ext::ResumableParser.new({})
         else
           m = method(:on_read_msgpack)
           @parser = Fluent::MessagePackFactory.msgpack_unpacker
@@ -173,6 +172,9 @@ def on_read(data)
 
       def on_read_json(data)
         @parser << data
+        while @parser.parse
+          @on_message.call(@parser.value)
+        end
       rescue => e
         @log.error "unexpected error in json payload", error: e.to_s
         @log.error_backtrace

lib/fluent/plugin/parser_json.rb

@@ -18,7 +18,6 @@
 require 'fluent/time'
 require 'fluent/oj_options'
 
-require 'yajl'
 require 'json'
 
 module Fluent
@@ -28,12 +27,10 @@ class JSONParser < Parser
 
       config_set_default :time_key, 'time'
       desc 'Set JSON parser'
+      # NOTE: Contains yajl for backward compatibility
       config_param :json_parser, :enum, list: [:oj, :yajl, :json], default: :oj
 
-      # The Yajl library defines a default buffer size of 8KiB when parsing
-      # from IO streams, so maintain this for backwards-compatibility.
-      # https://www.rubydoc.info/github/brianmario/yajl-ruby/Yajl%2FParser:parse
-      desc 'Set the buffer size that Yajl will use when parsing streaming input'
+      desc 'Set the buffer size that JSON parser will use when parsing streaming input'
       config_param :stream_buffer_size, :integer, default: 8192
 
       config_set_default :time_type, :float
@@ -54,8 +51,8 @@ def configure_json_parser(name)
 
           log&.info "Oj is not installed, and failing back to JSON for json parser"
           configure_json_parser(:json)
-        when :json then [JSON.method(:parse), JSON::ParserError]
-        when :yajl then [Yajl.method(:load), Yajl::ParseError]
+        when :yajl, :json # NOTE: Fallback yajl to json for backward compatibility
+          [JSON.method(:parse), JSON::ParserError]
         else
           raise "BUG: unknown json parser specified: #{name}"
         end
@@ -94,11 +91,15 @@ def parser_type
       end
 
       def parse_io(io, &block)
-        y = Yajl::Parser.new
-        y.on_parse_complete = ->(record){
-          block.call(parse_time(record), record)
-        }
-        y.parse(io, @stream_buffer_size)
+        parser = JSON::Ext::ResumableParser.new({})
+        while (chunk = io.read(@stream_buffer_size))
+          parser << chunk
+
+          while parser.parse
+            record = parser.value
+            block.call(parse_time(record), record)
+          end
+        end
       end
     end
   end

test/plugin/test_out_stdout.rb

@@ -78,13 +78,8 @@ def create_driver(conf = CONFIG)
       end
       assert_equal "#{Time.at(time).localtime.strftime(TIME_FORMAT)} test: {\"test\":\"test1\"}\n", out
 
-      if data == 'yajl'
-        # NOTE: Float::NAN is not jsonable
-        assert_raise(JSON::GeneratorError) { d.feed('test', time, {'test' => Float::NAN}) }
-      else
-        out = capture_log { d.feed('test', time, {'test' => Float::NAN}) }
-        assert_equal "#{Time.at(time).localtime.strftime(TIME_FORMAT)} test: {\"test\":NaN}\n", out
-      end
+      out = capture_log { d.feed('test', time, {'test' => Float::NAN}) }
+      assert_equal "#{Time.at(time).localtime.strftime(TIME_FORMAT)} test: {\"test\":NaN}\n", out
     end
 
     test 'emit in hash format' do

test/plugin/test_parser_json.rb

@@ -11,7 +11,7 @@ def setup
   sub_test_case "configure_json_parser" do
     data("oj", [:oj, [Oj.method(:load), Oj::ParseError]])
     data("json", [:json, [JSON.method(:parse), JSON::ParserError]])
-    data("yajl", [:yajl, [Yajl.method(:load), Yajl::ParseError]])
+    data("yajl", [:yajl, [JSON.method(:parse), JSON::ParserError]])
     def test_return_each_loader((input, expected_return))
       result = @parser.instance.configure_json_parser(input)
       assert_equal expected_return, result