Pr5449 verify

lib/fluent/command/cat.rb

@@ -326,7 +326,7 @@ def abort_message(time, record)
 when 'json'
   begin
     while line = $stdin.gets
-      record = JSON.parse(line, Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+      record = JSON.parse(line, **Fluent::DEFAULT_JSON_PARSE_OPTIONS)
       w.write(record)
     end
   rescue

lib/fluent/compat/exec_util.rb

@@ -80,7 +80,7 @@ class JSONParser < Parser
         BYTES_TO_READ = 8192
 
         def call(io)
-          parser = JSON::ResumableParser.new(Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+          parser = JSON::ResumableParser.new(**Fluent::DEFAULT_JSON_PARSE_OPTIONS)
           begin
             chunk = +"".b
             while io.readpartial(BYTES_TO_READ, chunk)

lib/fluent/config/literal_parser.rb

@@ -253,7 +253,7 @@ def scan_json(is_array)
             # '{"foo":"bar", #' -> '{"foo":"bar"}' (to check)
             parsed = nil
             begin
-              parsed = JSON.parse(buffer + line_buffer.rstrip.sub(/,$/, '') + (is_array ? "]" : "}"), Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+              parsed = JSON.parse(buffer + line_buffer.rstrip.sub(/,$/, '') + (is_array ? "]" : "}"), **Fluent::DEFAULT_JSON_PARSE_OPTIONS)
             rescue JSON::ParserError
               # This '#' is in json string literals
             end
@@ -288,7 +288,7 @@ def scan_json(is_array)
 
           line_buffer << char
           begin
-            result = JSON.parse(buffer + line_buffer, Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+            result = JSON.parse(buffer + line_buffer, **Fluent::DEFAULT_JSON_PARSE_OPTIONS)
           rescue JSON::ParserError
             # Incomplete json string yet
           end

lib/fluent/config/types.rb

@@ -201,7 +201,7 @@
       return nil if val.nil?
 
       param = if val.is_a?(String)
-                val.start_with?('{') ? JSON.parse(val, Fluent::DEFAULT_JSON_PARSE_OPTIONS) : Hash[val.strip.split(/\s*,\s*/).map{|v| v.split(':', 2)}]
+                val.start_with?('{') ? JSON.parse(val, **Fluent::DEFAULT_JSON_PARSE_OPTIONS) : Hash[val.strip.split(/\s*,\s*/).map{|v| v.split(':', 2)}]
               else
                 val
               end
@@ -228,7 +228,7 @@
       return nil if val.nil?
 
       param = if val.is_a?(String)
-                val.start_with?('[') ? JSON.parse(val, Fluent::DEFAULT_JSON_PARSE_OPTIONS) : val.strip.split(/\s*,\s*/)
+                val.start_with?('[') ? JSON.parse(val, **Fluent::DEFAULT_JSON_PARSE_OPTIONS) : val.strip.split(/\s*,\s*/)
               elsif val.is_a?(Array)
                 val
               elsif val.is_a?(Numeric) || val == true || val == false

lib/fluent/daemon.rb

@@ -10,5 +10,5 @@
 
 server_module = Fluent.const_get(ARGV[0])
 worker_module = Fluent.const_get(ARGV[1])
-params = JSON.parse(ARGV[2], Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+params = JSON.parse(ARGV[2], **Fluent::DEFAULT_JSON_PARSE_OPTIONS)
 ServerEngine::Daemon.run_server(server_module, worker_module) { Fluent::Supervisor.serverengine_config(params) }

lib/fluent/plugin/filter_record_transformer.rb

@@ -117,7 +117,7 @@ def filter_stream(tag, es)
 
     def parse_value(value_str)
       if value_str.start_with?('{', '[')
-        JSON.parse(value_str, Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+        JSON.parse(value_str, **Fluent::DEFAULT_JSON_PARSE_OPTIONS)
       else
         value_str
       end

lib/fluent/plugin/in_forward.rb

@@ -257,7 +257,7 @@ def read_messages(conn, &block)
         unless feeder
           first = data[0]
           if first == '{' || first == '[' # json
-            parser = JSON::ResumableParser.new(Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+            parser = JSON::ResumableParser.new(**Fluent::DEFAULT_JSON_PARSE_OPTIONS)
             serializer = :to_json.to_proc
             feeder = ->(d){
               parser << d

lib/fluent/plugin/in_monitor_agent.rb

@@ -102,7 +102,7 @@ def render_json(obj, code: 200, pretty_json: nil)
       end
 
       def render_ltsv(obj, code: 200)
-        normalized = JSON.parse(obj.to_json, Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+        normalized = JSON.parse(obj.to_json, **Fluent::DEFAULT_JSON_PARSE_OPTIONS)
         text = ''
         normalized.each do |hash|
           row = []

lib/fluent/plugin/in_sample.rb

@@ -45,7 +45,7 @@ class SampleInput < Input
     desc "The sample data to be generated. An array of JSON hashes or a single JSON hash."
     config_param :sample, alias: :dummy, default: [{"message" => "sample"}] do |val|
       begin
-        parsed = JSON.parse(val, Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+        parsed = JSON.parse(val, **Fluent::DEFAULT_JSON_PARSE_OPTIONS)
       rescue JSON::ParserError => ex
         # Fluent::ConfigParseError, "got incomplete JSON" will be raised
         # at literal_parser.rb with --use-v1-config, but I had to

lib/fluent/plugin/in_unix.rb

@@ -158,7 +158,7 @@ def on_read(data)
         first = data[0]
         if first == '{'.freeze || first == '['.freeze
           m = method(:on_read_json)
-          @parser = JSON::ResumableParser.new(Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+          @parser = JSON::ResumableParser.new(**Fluent::DEFAULT_JSON_PARSE_OPTIONS)
         else
           m = method(:on_read_msgpack)
           @parser = Fluent::MessagePackFactory.msgpack_unpacker

lib/fluent/plugin/parser_json.rb

@@ -38,7 +38,7 @@ class JSONParser < Parser
 
       # Use a shared proc rather than a per-call lambda so that
       # configure_json_parser returns the same object every time.
-      JSON_PARSE_PROC = ->(text) { JSON.parse(text, Fluent::DEFAULT_JSON_PARSE_OPTIONS) }
+      JSON_PARSE_PROC = ->(text) { JSON.parse(text, **Fluent::DEFAULT_JSON_PARSE_OPTIONS) }
 
       def configure(conf)
         if conf.has_key?('time_format')
@@ -96,7 +96,7 @@ def parser_type
       end
 
       def parse_io(io, &block)
-        parser = JSON::ResumableParser.new(Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+        parser = JSON::ResumableParser.new(**Fluent::DEFAULT_JSON_PARSE_OPTIONS)
         begin
           chunk = +"".b
           while io.readpartial(@stream_buffer_size, chunk)

lib/fluent/plugin/parser_syslog.rb

@@ -71,6 +71,7 @@ def initialize
         @time_parser_rfc5424 = nil
         @space_count_rfc3164 = nil
         @space_count_rfc5424 = nil
+        @time_format_starts_with_space_rfc3164 = false
         @skip_space_count_rfc3164 = false
         @skip_space_count_rfc5424 = false
         @time_parser_rfc5424_without_subseconds = nil
@@ -123,10 +124,12 @@ class << self
 
       def setup_time_parser_3164(time_fmt)
         @time_parser_rfc3164 = time_parser_create(format: time_fmt)
+        @time_format_starts_with_space_rfc3164 = time_fmt.start_with?(SPLIT_CHAR)
         if ['%b %d %H:%M:%S', '%b %d %H:%M:%S.%N'].include?(time_fmt)
           @skip_space_count_rfc3164 = true
         end
         @space_count_rfc3164 = time_fmt.squeeze(' ').count(' ') + 1
+        @space_count_rfc3164 -= 1 if @time_format_starts_with_space_rfc3164
       end
 
       def setup_time_parser_5424(time_fmt)
@@ -162,6 +165,8 @@ def parse_auto(text, &block)
       end
 
       SPLIT_CHAR = ' '.freeze
+      DOT_CHAR = '.'.freeze
+      RFC3164_PRI_DIGITS_REGEXP = /\A[0-9]{1,3}\z/
 
       def parse_rfc3164_regex(text, &block)
         idx = 0
@@ -180,13 +185,26 @@ def parse_rfc3164_regex(text, &block)
 
         i = idx - 1
         sq = false
+        first_time_field = true
         @space_count_rfc3164.times do
           while text[i + 1] == SPLIT_CHAR
+            if first_time_field
+              unless @time_format_starts_with_space_rfc3164
+                idx += 1
+                i += 1
+                break
+              end
+            end
             sq = true
             i += 1
           end
 
+          first_time_field = false
           i = text.index(SPLIT_CHAR, i + 1)
+          unless i
+            yield nil, nil
+            return
+          end
         end
 
         time_str = sq ? text.slice(idx, i - idx).squeeze(SPLIT_CHAR) : text.slice(idx, i - idx)
@@ -287,16 +305,30 @@ def parse_rfc3164(text, &block)
           end
         end
 
+        if text[cursor] == SPLIT_CHAR
+          cursor = rfc3164_space_cursor(text, cursor)
+          unless cursor
+            yield nil, nil
+            return
+          end
+        end
+
         if @skip_space_count_rfc3164
           # header part
           time_size = 15 # skip Mmm dd hh:mm:ss
-          time_end = text[cursor + time_size]
+          time_end_index = cursor + time_size
+          time_end = text[time_end_index]
+
           if time_end == SPLIT_CHAR
             time_str = text.slice(cursor, time_size)
             cursor += 16 # time + ' '
-          elsif time_end == '.'.freeze
+          elsif time_end == DOT_CHAR
             # support subsecond time
-            i = text.index(SPLIT_CHAR, time_size)
+            i = text.index(SPLIT_CHAR, time_end_index)
+            unless i
+              yield nil, nil
+              return
+            end
             time_str = text.slice(cursor, i - cursor)
             cursor = i + 1
           else
@@ -312,14 +344,18 @@ def parse_rfc3164(text, &block)
               i += 1
             end
             i = text.index(SPLIT_CHAR, i + 1)
+            unless i
+              yield nil, nil
+              return
+            end
           end
 
-          time_str = sq ? text.slice(idx, i - cursor).squeeze(SPLIT_CHAR) : text.slice(cursor, i - cursor)
+          time_str = sq ? text.slice(cursor, i - cursor).squeeze(SPLIT_CHAR) : text.slice(cursor, i - cursor)
           cursor = i + 1
         end
 
         i = text.index(SPLIT_CHAR, cursor)
-        if i.nil?
+        unless i
           yield nil, nil
           return
         end
@@ -363,12 +399,23 @@ def parse_rfc3164(text, &block)
         msg.chomp!
         record['message'] = msg
 
-        time = @time_parser_rfc3164.parse(time_str)
+        begin
+          time = @time_parser_rfc3164.parse(time_str)
+        rescue Fluent::TimeParser::TimeParseError
+          yield nil, nil
+          return
+        end
         record['time'] = time_str if @keep_time_key
 
         yield time, record
       end
 
+      def rfc3164_space_cursor(text, cursor)
+        return if @with_priority && !RFC3164_PRI_DIGITS_REGEXP.match?(text.slice(1, cursor - 2))
+
+        @time_format_starts_with_space_rfc3164 ? cursor : cursor + 1
+      end
+
       NILVALUE = '-'.freeze
 
       def parse_rfc5424(text, &block)

lib/fluent/plugin/sd_file.rb

@@ -76,7 +76,7 @@ def parser
             -> (v) { YAML.safe_load(v).map }
           when :json
             require 'json'
-            -> (v) { JSON.parse(v, Fluent::DEFAULT_JSON_PARSE_OPTIONS) }
+            -> (v) { JSON.parse(v, **Fluent::DEFAULT_JSON_PARSE_OPTIONS) }
           end
       end
 

lib/fluent/plugin/storage_local.rb

@@ -90,7 +90,7 @@ def configure(conf)
                 log.warn "detect empty plugin storage file during startup. Ignored: #{@path}"
                 return
               end
-              data = JSON.parse(data, Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+              data = JSON.parse(data, **Fluent::DEFAULT_JSON_PARSE_OPTIONS)
               raise Fluent::ConfigError, "Invalid contents (not object) in plugin storage file: '#{@path}'" unless data.is_a?(Hash)
             rescue => e
               log.error "failed to read data from plugin storage file", path: @path, error: e
@@ -114,7 +114,7 @@ def load
         return unless File.exist?(@path)
         begin
           json_string = File.open(@path, 'r:utf-8:utf-8'){ |io| io.read }
-          json = JSON.parse(json_string, Fluent::DEFAULT_JSON_PARSE_OPTIONS)
+          json = JSON.parse(json_string, **Fluent::DEFAULT_JSON_PARSE_OPTIONS)
           unless json.is_a?(Hash)
             log.error "broken content for plugin storage (Hash required: ignored)", type: json.class
             log.debug "broken content", content: json_string

test/plugin/test_in_syslog.rb

@@ -124,6 +124,79 @@ def test_time_format(data)
     }
   end
 
+  data(
+    'regexp/rfc3164/parser priority' => ['regexp', 'rfc3164', true],
+    'string/rfc3164/parser priority' => ['string', 'rfc3164', true],
+    'regexp/auto/parser priority' => ['regexp', 'auto', true],
+    'string/auto/parser priority' => ['string', 'auto', true],
+    'regexp/rfc3164/input priority' => ['regexp', 'rfc3164', false],
+    'string/rfc3164/input priority' => ['string', 'rfc3164', false],
+    'regexp/auto/input priority' => ['regexp', 'auto', false],
+    'string/auto/input priority' => ['string', 'auto', false],
+  )
+  def test_space_between_rfc3164_priority_and_header(data)
+    parser_engine, message_format, with_priority = data
+    d = create_driver([
+      ipv4_config,
+      'severity_key severity',
+      'facility_key facility',
+      '<parse>',
+      "  parser_engine #{parser_engine}",
+      "  message_format #{message_format}",
+      "  with_priority #{with_priority}",
+      '</parse>',
+    ].join("\n"))
+
+    message = 'Apr 25 16:43:29 PAA-SW1-1 General[procLOG]: main.c(257) 272264 %% Stopping System API  application'
+    d.run(expect_emits: 2) do
+      u = UDPSocket.new
+      u.connect('127.0.0.1', @port)
+      u.send("<14>#{message}", 0)
+      u.send("<14> #{message}", 0)
+    end
+
+    assert_equal(2, d.events.size)
+    assert_equal(d.events[0][1], d.events[1][1])
+    assert_equal(d.events[0][2], d.events[1][2])
+    d.events.each do |tag, time, record|
+      assert_equal('syslog.user.info', tag)
+      assert_equal(event_time('Apr 25 16:43:29', format: '%b %d %H:%M:%S'), time)
+      assert_equal('user', record['facility'])
+      assert_equal('info', record['severity'])
+      assert_equal('PAA-SW1-1', record['host'])
+      assert_equal('General', record['ident'])
+      assert_equal('main.c(257) 272264 %% Stopping System API  application', record['message'])
+    end
+  end
+
+  data('regexp' => 'regexp', 'string' => 'string')
+  def test_space_after_input_owned_priority_preserves_input_priority_syntax(parser_engine)
+    d = create_driver([
+      ipv4_config,
+      'emit_unmatched_lines true',
+      '<parse>',
+      "  parser_engine #{parser_engine}",
+      '  with_priority false',
+      '</parse>',
+    ].join("\n"))
+
+    messages = [
+      '<1234>Apr 25 16:43:29 host app: message',
+      '<1234> Apr 25 16:43:29 host app: message',
+      '<ab> Apr 25 16:43:29 host app: message',
+    ]
+    d.run(expect_emits: 3) do
+      u = UDPSocket.new
+      u.connect('127.0.0.1', @port)
+      messages.each { |message| u.send(message, 0) }
+    end
+
+    assert_equal(3, d.events.size)
+    assert_equal(d.events[0], d.events[1])
+    assert_equal('syslog.unmatched', d.events[2][0])
+    assert_equal(messages[2], d.events[2][2]['unmatched_line'])
+  end
+
   def test_msg_size
     d = create_driver
     tests = create_test_case