From 96664acdddba6eb069cc006c9e69ed45717c107e Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Mon, 29 Aug 2016 19:44:49 +0900
Subject: [PATCH 01/19] Port parser filter from fluent-plugin-parser. fix #1189

---
 lib/fluent/plugin/filter_parser.rb | 114 +++++
 test/plugin/test_filter_parser.rb  | 713 +++++++++++++++++++++++++++++
 2 files changed, 827 insertions(+)
 create mode 100644 lib/fluent/plugin/filter_parser.rb
 create mode 100644 test/plugin/test_filter_parser.rb

diff --git a/lib/fluent/plugin/filter_parser.rb b/lib/fluent/plugin/filter_parser.rb
new file mode 100644
index 0000000000..7faa9022d8
--- /dev/null
+++ b/lib/fluent/plugin/filter_parser.rb
@@ -0,0 +1,114 @@
+require 'fluent/plugin/filter'
+require 'fluent/plugin/parser'
+
+module Fluent::Plugin
+  class ParserFilter < Filter
+    Fluent::Plugin.register_filter('parser', self)
+
+    config_param :key_name, :string
+    config_param :reserve_data, :bool, default: false
+    config_param :inject_key_prefix, :string, default: nil
+    config_param :replace_invalid_sequence, :bool, default: false
+    config_param :hash_value_field, :string, default: nil
+    config_param :suppress_parse_error_log, :bool, default: false
+    config_param :time_parse, :bool, default: true
+    config_param :ignore_key_not_exist, :bool, default: false
+
+    attr_reader :parser
+
+    def initialize
+      super
+      require 'time'
+    end
+
+    def configure(conf)
+      super
+
+      @parser = Fluent::TextParser.new
+      @parser.estimate_current_event = false
+      @parser.configure(conf)
+      if !@time_parse && @parser.parser.respond_to?("time_key=".to_sym)
+        # disable parse time
+        @parser.parser.time_key = nil
+      end
+
+      self
+    end
+
+    def filter_stream(tag, es)
+      new_es = Fluent::MultiEventStream.new
+      es.each do |time, record|
+        raw_value = record[@key_name]
+        if raw_value.nil?
+          log.warn "#{@key_name} does not exist" unless @ignore_key_not_exist
+          new_es.add(time, handle_parsed(tag, record, time, {})) if @reserve_data
+          next
+        end
+        begin
+          @parser.parse(raw_value) do |t, values|
+            if values
+              t ||= time
+              r = handle_parsed(tag, record, t, values)
+              new_es.add(t, r)
+            else
+              log.warn "pattern not match with data '#{raw_value}'" unless @suppress_parse_error_log
+              if @reserve_data
+                t = time
+                r = handle_parsed(tag, record, time, {})
+                new_es.add(t, r)
+              end
+            end
+          end
+        rescue Fluent::TextParser::ParserError => e
+          log.warn e.message unless @suppress_parse_error_log
+        rescue ArgumentError => e
+          if @replace_invalid_sequence
+            unless e.message.index("invalid byte sequence in") == 0
+              raise
+            end
+            replaced_string = replace_invalid_byte(raw_value)
+            @parser.parse(replaced_string) do |t, values|
+              if values
+                t ||= time
+                r = handle_parsed(tag, record, t, values)
+                new_es.add(t, r)
+              else
+                log.warn "pattern not match with data '#{raw_value}'" unless @suppress_parse_error_log
+                if @reserve_data
+                  t = time
+                  r = handle_parsed(tag, record, time, {})
+                  new_es.add(t, r)
+                end
+              end
+            end
+          else
+            raise
+          end
+        rescue => e
+          log.warn "parse failed #{e.message}" unless @suppress_parse_error_log
+        end
+      end
+      new_es
+    end
+
+    private
+
+    def handle_parsed(tag, record, t, values)
+      if values && @inject_key_prefix
+        values = Hash[values.map { |k, v| [@inject_key_prefix + k, v] }]
+      end
+      r = @hash_value_field ? {@hash_value_field => values} : values
+      if @reserve_data
+        r = r ? record.merge(r) : record
+      end
+      r
+    end
+
+    def replace_invalid_byte(string)
+      replace_options = {invalid: :replace, undef: :replace, replace: '?'}
+      original_encoding = string.encoding
+      temporal_encoding = (original_encoding == Encoding::UTF_8 ? Encoding::UTF_16BE : Encoding::UTF_8)
+      string.encode(temporal_encoding, original_encoding, replace_options).encode(original_encoding)
+    end
+  end
+end
diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
new file mode 100644
index 0000000000..c5815f0cf0
--- /dev/null
+++ b/test/plugin/test_filter_parser.rb
@@ -0,0 +1,713 @@
+require_relative '../helper'
+require 'fluent/test/driver/filter'
+require 'fluent/plugin/filter_parser'
+
+class ParserFilterTest < Test::Unit::TestCase
+  def setup
+    Fluent::Test.setup
+  end
+
+  CONFIG = %[
+    key_name      message
+    format        /^(?<x>.)(?<y>.) (?<time>.+)$/
+    time_format   %Y%m%d%H%M%S
+    reserve_data  true
+  ]
+
+  def create_driver(conf=CONFIG,tag='test')
+    Fluent::Test::FilterTestDriver.new(Fluent::Plugin::ParserFilter, tag).configure(conf)
+  end
+
+  def test_configure
+    assert_raise(Fluent::ConfigError) {
+      create_driver('')
+    }
+    assert_raise(Fluent::ConfigError) {
+      create_driver %[
+        format unknown_format_that_will_never_be_implemented
+        key_name foo
+      ]
+    }
+    assert_nothing_raised {
+      create_driver %[
+        format /(?<x>.)/
+        key_name foo
+      ]
+    }
+    assert_nothing_raised {
+      create_driver %[
+        format /(?<x>.)/
+        key_name foo
+      ]
+    }
+    assert_nothing_raised {
+      create_driver %[
+        format /(?<x>.)/
+        key_name foo
+      ]
+    }
+    assert_nothing_raised {
+      create_driver %[
+        format /(?<x>.)/
+        key_name foo
+      ]
+    }
+    assert_nothing_raised {
+      create_driver %[
+        format json
+        key_name foo
+      ]
+    }
+    assert_nothing_raised {
+      create_driver %[
+        format ltsv
+        key_name foo
+      ]
+    }
+    assert_nothing_raised {
+      create_driver %[
+        format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+        key_name message
+        suppress_parse_error_log true
+      ]
+    }
+    assert_nothing_raised {
+      create_driver %[
+        format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+        key_name message
+        suppress_parse_error_log false
+      ]
+    }
+    d = create_driver %[
+      key_name foo
+      format /(?<x>.)/
+    ]
+    assert_equal false, d.instance.reserve_data
+  end
+
+  # CONFIG = %[
+  #   remove_prefix test
+  #   add_prefix    parsed
+  #   key_name      message
+  #   format        /^(?<x>.)(?<y>.) (?<time>.+)$/
+  #   time_format   %Y%m%d%H%M%S
+  #   reserve_data  true
+  # ]
+  def test_filter
+    d1 = create_driver(CONFIG, 'test.no.change')
+    time = Time.parse("2012-01-02 13:14:15").to_i
+    d1.run do
+      d1.filter({'message' => '12 20120402182059'}, time)
+      d1.filter({'message' => '34 20120402182100'}, time)
+      d1.filter({'message' => '56 20120402182100'}, time)
+      d1.filter({'message' => '78 20120402182101'}, time)
+      d1.filter({'message' => '90 20120402182100'}, time)
+    end
+    filtered = d1.filtered_as_array
+    assert_equal 5, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal Time.parse("2012-04-02 18:20:59").to_i, first[1]
+    assert_equal '1', first[2]['x']
+    assert_equal '2', first[2]['y']
+    assert_equal '12 20120402182059', first[2]['message']
+
+    second = filtered[1]
+    assert_equal 'test.no.change', second[0]
+    assert_equal Time.parse("2012-04-02 18:21:00").to_i, second[1]
+    assert_equal '3', second[2]['x']
+    assert_equal '4', second[2]['y']
+
+    third = filtered[2]
+    assert_equal 'test.no.change', third[0]
+    assert_equal Time.parse("2012-04-02 18:21:00").to_i, third[1]
+    assert_equal '5', third[2]['x']
+    assert_equal '6', third[2]['y']
+
+    fourth = filtered[3]
+    assert_equal 'test.no.change', fourth[0]
+    assert_equal Time.parse("2012-04-02 18:21:01").to_i, fourth[1]
+    assert_equal '7', fourth[2]['x']
+    assert_equal '8', fourth[2]['y']
+
+    fifth = filtered[4]
+    assert_equal 'test.no.change', fifth[0]
+    assert_equal Time.parse("2012-04-02 18:21:00").to_i, fifth[1]
+    assert_equal '9', fifth[2]['x']
+    assert_equal '0', fifth[2]['y']
+
+    d2 = create_driver(%[
+      tag parsed
+      key_name      data
+      format        /^(?<x>.)(?<y>.) (?<t>.+)$/
+    ], 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d2.run do
+      d2.filter({'data' => '12 20120402182059'}, time)
+      d2.filter({'data' => '34 20120402182100'}, time)
+    end
+    filtered = d2.filtered_as_array
+    assert_equal 2, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal time, first[1]
+    assert_nil first[2]['data']
+    assert_equal '1', first[2]['x']
+    assert_equal '2', first[2]['y']
+    assert_equal '20120402182059', first[2]['t']
+
+    second = filtered[1]
+    assert_equal 'test.no.change', second[0]
+    assert_equal time, second[1]
+    assert_nil second[2]['data']
+    assert_equal '3', second[2]['x']
+    assert_equal '4', second[2]['y']
+    assert_equal '20120402182100', second[2]['t']
+
+    d3 = create_driver(%[
+      tag parsed
+      key_name      data
+      format        /^(?<x>[0-9])(?<y>[0-9]) (?<t>.+)$/
+    ], 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d3.run do
+      d3.filter({'data' => '12 20120402182059'}, time)
+      d3.filter({'data' => '34 20120402182100'}, time)
+      d3.filter({'data' => 'xy 20120402182101'}, time)
+    end
+    filtered = d3.filtered_as_array
+    assert_equal 2, filtered.length
+
+    d3x = create_driver(%[
+      tag parsed
+      key_name      data
+      format        /^(?<x>\\d)(?<y>\\d) (?<t>.+)$/
+      reserve_data  yes
+    ], 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d3x.run do
+      d3x.filter({'data' => '12 20120402182059'}, time)
+      d3x.filter({'data' => '34 20120402182100'}, time)
+      d3x.filter({'data' => 'xy 20120402182101'}, time)
+    end
+    filtered = d3x.filtered_as_array
+    assert_equal 3, filtered.length
+
+    d4 = create_driver(%[
+      tag parsed
+      key_name      data
+      format        json
+    ], 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d4.run do
+      d4.filter({'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'}, time)
+      d4.filter({'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'}, time)
+    end
+    filtered = d4.filtered_as_array
+    assert_equal 1, filtered.length
+
+    d4x = create_driver(%[
+      tag parsed
+      key_name      data
+      format        json
+      reserve_data  yes
+    ], 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d4x.run do
+      d4x.filter({'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'}, time)
+      d4x.filter({'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'}, time)
+    end
+    filtered = d4x.filtered_as_array
+    assert_equal 2, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal time, first[1]
+    assert_equal '{"xxx":"first","yyy":"second"}', first[2]['data']
+    assert_equal 'first', first[2]['xxx']
+    assert_equal 'second', first[2]['yyy']
+
+    second = filtered[1]
+    assert_equal 'test.no.change', second[0]
+    assert_equal time, second[1]
+    assert_equal 'foobar', second[2]['data']
+    assert_equal 'x', second[2]['xxx']
+    assert_equal 'y', second[2]['yyy']
+  end
+
+  CONFIG_LTSV =  %[
+    format ltsv
+    key_name data
+  ]
+  def test_filter_ltsv
+    d = create_driver(CONFIG_LTSV, 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d.run do
+      d.filter({'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'}, time)
+      d.filter({'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'}, time)
+    end
+    filtered = d.filtered_as_array
+    assert_equal 2, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal time, first[1]
+    assert_nil first[2]['data']
+    assert_equal 'first', first[2]['xxx']
+    assert_equal 'second', first[2]['yyy']
+
+    second = filtered[1]
+    assert_equal 'test.no.change', second[0]
+    assert_equal time, second[1]
+    assert_nil first[2]['data']
+    assert_equal 'first', second[2]['xxx']
+    assert_equal 'second2', second[2]['yyy']
+
+    d = create_driver(CONFIG_LTSV + %[
+      reserve_data yes
+    ], 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d.run do
+      d.filter({'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'}, time)
+      d.filter({'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'}, time)
+    end
+    filtered = d.filtered_as_array
+    assert_equal 2, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal time, first[1]
+    assert_equal "xxx:first\tyyy:second", first[2]['data']
+    assert_equal 'first', first[2]['xxx']
+    assert_equal 'second', first[2]['yyy']
+
+    second = filtered[1]
+    assert_equal 'test.no.change', second[0]
+    assert_equal time, second[1]
+    assert_equal "xxx:first\tyyy:second", first[2]['data']
+    assert_equal 'first', second[2]['xxx']
+    assert_equal 'second2', second[2]['yyy']
+
+    # convert types
+    d = create_driver(CONFIG_LTSV + %[
+      types i:integer,s:string,f:float,b:bool
+    ], 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d.run do
+      d.filter({'data' => "i:1\ts:2\tf:3\tb:true\tx:123"}, time)
+    end
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal time, first[1]
+    assert_equal 1, first[2]['i']
+    assert_equal '2', first[2]['s']
+    assert_equal 3.0, first[2]['f']
+    assert_equal true, first[2]['b']
+    assert_equal '123', first[2]['x']
+  end
+
+  CONFIG_TSV =  %[
+    format tsv
+    key_name data
+    keys key1,key2,key3
+  ]
+  def test_filter_tsv
+    d = create_driver(CONFIG_TSV, 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d.run do
+      d.filter({'data' => "value1\tvalue2\tvalueThree", 'xxx' => 'x', 'yyy' => 'y'}, time)
+    end
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal time, first[1]
+    assert_nil first[2]['data']
+    assert_equal 'value1', first[2]['key1']
+    assert_equal 'value2', first[2]['key2']
+    assert_equal 'valueThree', first[2]['key3']
+  end
+
+  CONFIG_CSV =  %[
+    format csv
+    key_name data
+    keys key1,key2,key3
+  ]
+  def test_filter_csv
+    d = create_driver(CONFIG_CSV, 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d.run do
+      d.filter({'data' => 'value1,"value2","value""ThreeYes!"', 'xxx' => 'x', 'yyy' => 'y'}, time)
+    end
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal time, first[1]
+    assert_nil first[2]['data']
+    assert_equal 'value1', first[2]['key1']
+    assert_equal 'value2', first[2]['key2']
+    assert_equal 'value"ThreeYes!', first[2]['key3']
+  end
+
+  CONFIG_HASH_VALUE_FIELD = %[
+    format       json
+    key_name     data
+    hash_value_field parsed
+  ]
+  CONFIG_HASH_VALUE_FIELD_RESERVE_DATA = %[
+    format       json
+    key_name     data
+    reserve_data yes
+    hash_value_field parsed
+  ]
+  CONFIG_HASH_VALUE_FIELD_WITH_INJECT_KEY_PREFIX = %[
+    format       json
+    key_name     data
+    hash_value_field parsed
+    inject_key_prefix data.
+  ]
+  def test_filter_inject_hash_value_field
+    original = {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'}
+
+    d = create_driver(CONFIG_HASH_VALUE_FIELD, 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d.run do
+      d.filter(original, time)
+    end
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal time, first[1]
+
+    record = first[2]
+    assert_equal 1, record.keys.size
+    assert_equal({"xxx"=>"first","yyy"=>"second"}, record['parsed'])
+
+    d = create_driver(CONFIG_HASH_VALUE_FIELD_RESERVE_DATA, 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d.run do
+      d.filter(original, time)
+    end
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal time, first[1]
+
+    record = first[2]
+    assert_equal 4, record.keys.size
+    assert_equal original['data'], record['data']
+    assert_equal original['xxx'], record['xxx']
+    assert_equal original['yyy'], record['yyy']
+    assert_equal({"xxx"=>"first","yyy"=>"second"}, record['parsed'])
+
+    d = create_driver(CONFIG_HASH_VALUE_FIELD_WITH_INJECT_KEY_PREFIX, 'test.no.change')
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d.run do
+      d.filter(original, time)
+    end
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+
+    first = filtered[0]
+    assert_equal 'test.no.change', first[0]
+    assert_equal time, first[1]
+
+    record = first[2]
+    assert_equal 1, record.keys.size
+    assert_equal({"data.xxx"=>"first","data.yyy"=>"second"}, record['parsed'])
+  end
+
+  CONFIG_DONT_PARSE_TIME = %[
+    key_name data
+    format json
+    time_parse no
+  ]
+  def test_time_should_be_reserved
+    t = Time.now.to_i
+    d = create_driver(CONFIG_DONT_PARSE_TIME, 'test.no.change')
+
+    d.run do
+      d.filter({'data' => '{"time":1383190430, "f1":"v1"}'}, t)
+      d.filter({'data' => '{"time":"1383190430", "f1":"v1"}'}, t)
+      d.filter({'data' => '{"time":"2013-10-31 12:34:03 +0900", "f1":"v1"}'}, t)
+    end
+    filtered = d.filtered_as_array
+    assert_equal 3, filtered.length
+
+    assert_equal 'test.no.change', filtered[0][0]
+    assert_equal 'v1', filtered[0][2]['f1']
+    assert_equal 1383190430, filtered[0][2]['time']
+    assert_equal t, filtered[0][1]
+
+    assert_equal 'test.no.change', filtered[1][0]
+    assert_equal 'v1', filtered[1][2]['f1']
+    assert_equal "1383190430", filtered[1][2]['time']
+    assert_equal t, filtered[1][1]
+
+    assert_equal 'test.no.change', filtered[2][0]
+    assert_equal 'v1', filtered[2][2]['f1']
+    assert_equal '2013-10-31 12:34:03 +0900', filtered[2][2]['time']
+    assert_equal t, filtered[2][1]
+  end
+
+  CONFIG_INVALID_TIME_VALUE = %[
+    remove_prefix test
+    key_name data
+    format json
+  ] # 'time' is implicit @time_key
+  def test_filter_invalid_time_data
+    # should not raise errors
+    t = Time.now.to_i
+    d = create_driver(CONFIG_INVALID_TIME_VALUE, 'test.no.change')
+    assert_nothing_raised {
+      d.run do
+        d.filter({'data' => '{"time":[], "f1":"v1"}'}, t)
+        d.filter({'data' => '{"time":"thisisnottime", "f1":"v1"}'}, t)
+      end
+    }
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+
+    assert_equal 'test.no.change', filtered[0][0]
+    assert_equal 0, filtered[0][1]
+    assert_equal 'v1', filtered[0][2]['f1']
+    assert_equal 0, filtered[0][2]['time'].to_i
+  end
+
+  # REGEXP = /^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$/
+
+  CONFIG_NOT_REPLACE = %[
+    remove_prefix test
+    key_name      data
+    format        /^(?<message>.*)$/
+  ]
+  CONFIG_INVALID_BYTE = CONFIG_NOT_REPLACE + %[
+    replace_invalid_sequence true
+  ]
+  def test_filter_invalid_byte
+    invalid_utf8 = "\xff".force_encoding('UTF-8')
+
+    d = create_driver(CONFIG_NOT_REPLACE, 'test.no.change')
+    assert_raise(ArgumentError) {
+      d.run do
+        d.filter({'data' => invalid_utf8}, Time.now.to_i)
+      end
+    }
+
+    d = create_driver(CONFIG_INVALID_BYTE, 'test.in')
+    assert_nothing_raised {
+      d.run do
+        d.emit({'data' => invalid_utf8}, Time.now.to_i)
+      end
+    }
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+    assert_nil filtered[0][2]['data']
+    assert_equal '?'.force_encoding('UTF-8'), filtered[0][2]['message']
+
+    d = create_driver(CONFIG_INVALID_BYTE + %[
+      reserve_data yes
+    ], 'test.no.change')
+    assert_nothing_raised {
+      d.run do
+        d.filter({'data' => invalid_utf8}, Time.now.to_i)
+      end
+    }
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+    assert_equal invalid_utf8, filtered[0][2]['data']
+    assert_equal '?'.force_encoding('UTF-8'), filtered[0][2]['message']
+
+    invalid_ascii = "\xff".force_encoding('US-ASCII')
+    d = create_driver(CONFIG_INVALID_BYTE, 'test.no.change')
+    assert_nothing_raised {
+      d.run do
+        d.filter({'data' => invalid_ascii}, Time.now.to_i)
+      end
+    }
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+    assert_nil filtered[0][2]['data']
+    assert_equal '?'.force_encoding('US-ASCII'), filtered[0][2]['message']
+  end
+
+  CONFIG_NOT_IGNORE = %[
+    remove_prefix    test
+    key_name         data
+    format           json
+    hash_value_field parsed
+  ]
+  CONFIG_IGNORE = CONFIG_NOT_IGNORE + %[
+    ignore_key_not_exist true
+  ]
+  CONFIG_PASS_SAME_RECORD = CONFIG_IGNORE + %[
+    reserve_data true
+  ]
+  def test_filter_key_not_exist
+    d = create_driver(CONFIG_NOT_IGNORE, 'test.no.ignore')
+    assert_nothing_raised {
+      d.run do
+        d.filter({'foo' => 'bar'}, Time.now.to_i)
+      end
+    }
+    assert_match(/data does not exist/, d.instance.log.out.logs.first)
+
+    d = create_driver(CONFIG_IGNORE, 'test.ignore')
+    assert_nothing_raised {
+      d.run do
+        d.filter({'foo' => 'bar'}, Time.now.to_i)
+      end
+    }
+    assert_not_match(/data does not exist/, d.instance.log.out.logs.first)
+
+    d = create_driver(CONFIG_PASS_SAME_RECORD, 'test.pass_same_record')
+    assert_nothing_raised {
+      d.run do
+        d.filter({'foo' => 'bar'}, Time.now.to_i)
+      end
+    }
+    filtered = d.filtered_as_array
+    assert_equal 1, filtered.length
+    assert_nil filtered[0][2]['data']
+    assert_equal 'bar', filtered[0][2]['foo']
+  end
+
+  # suppress_parse_error_log test
+  CONFIG_DISABELED_SUPPRESS_PARSE_ERROR_LOG = %[
+    tag hogelog
+    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+    key_name message
+    suppress_parse_error_log false
+  ]
+  CONFIG_ENABELED_SUPPRESS_PARSE_ERROR_LOG = %[
+    tag hogelog
+    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+    key_name message
+    suppress_parse_error_log true
+  ]
+  CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG = %[
+    tag hogelog
+    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+    key_name message
+  ]
+
+  INVALID_MESSAGE = 'foo bar'
+  VALID_MESSAGE   = 'col1=foo col2=bar'
+
+  # if call warn() raise exception
+  class DummyLoggerWarnedException < StandardError; end
+  class DummyLogger
+    def reset
+    end
+    def warn(message)
+      raise DummyLoggerWarnedException
+    end
+  end
+
+  def swap_logger(instance)
+    raise "use with block" unless block_given?
+    dummy = DummyLogger.new
+    saved_logger = instance.log
+    instance.log = dummy
+    restore = lambda { instance.log = saved_logger }
+
+    yield
+
+    restore.call
+  end
+
+  def test_parser_error_warning
+    d = create_driver(CONFIG_INVALID_TIME_VALUE, 'test.no.change')
+    swap_logger(d.instance) do
+      assert_raise(DummyLoggerWarnedException) {
+        d.run do
+          d.filter({'data' => '{"time":[], "f1":"v1"}'}, Time.now.to_i)
+        end
+      }
+    end
+  end
+
+  class DefaultSuppressParseErrorLogTest < self
+    def setup
+      # default(disabled) 'suppress_parse_error_log' is not specify
+      @d = create_driver(CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG, 'test.no.change')
+    end
+
+    def test_raise_exception
+      swap_logger(@d.instance) do
+        assert_raise(DummyLoggerWarnedException) {
+          @d.run do
+            @d.filter({'message' => INVALID_MESSAGE}, Time.now.to_i)
+          end
+        }
+      end
+    end
+
+    def test_nothing_raised
+      swap_logger(@d.instance) do
+      assert_nothing_raised {
+        @d.run do
+          @d.filter({'message' => VALID_MESSAGE}, Time.now.to_i)
+        end
+      }
+      end
+    end
+  end
+
+  class DisabledSuppressParseErrorLogTest < self
+    def setup
+      # disabled 'suppress_parse_error_log'
+      @d = create_driver(CONFIG_DISABELED_SUPPRESS_PARSE_ERROR_LOG, 'test.no.change')
+    end
+
+    def test_raise_exception
+      swap_logger(@d.instance) do
+        assert_raise(DummyLoggerWarnedException) {
+          @d.run do
+            @d.filter({'message' => INVALID_MESSAGE}, Time.now.to_i)
+          end
+        }
+      end
+    end
+
+    def test_nothing_raised
+      swap_logger(@d.instance) do
+      assert_nothing_raised {
+        @d.run do
+          @d.filter({'message' => VALID_MESSAGE}, Time.now.to_i)
+        end
+      }
+      end
+    end
+  end
+
+  class EnabledSuppressParseErrorLogTest < self
+    def setup
+      # enabled 'suppress_parse_error_log'
+      @d = create_driver(CONFIG_ENABELED_SUPPRESS_PARSE_ERROR_LOG, 'test.no.change')
+    end
+
+    def test_nothing_raised
+      swap_logger(@d.instance) do
+      assert_nothing_raised {
+        @d.run do
+          @d.filter({'message' => INVALID_MESSAGE}, Time.now.to_i)
+          @d.filter({'message' => VALID_MESSAGE},   Time.now.to_i)
+        end
+      }
+      end
+    end
+  end
+end

From 0f14fa6d4d2136cc10a7b7c3602c9498b77057fa Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Wed, 31 Aug 2016 02:33:59 +0900
Subject: [PATCH 02/19] Apply v0.14 style. Use plugin helpers.

---
 lib/fluent/plugin/filter_parser.rb |  45 ++-
 test/plugin/test_filter_parser.rb  | 463 ++++++++++++++---------------
 2 files changed, 253 insertions(+), 255 deletions(-)

diff --git a/lib/fluent/plugin/filter_parser.rb b/lib/fluent/plugin/filter_parser.rb
index 7faa9022d8..f2b4426bf0 100644
--- a/lib/fluent/plugin/filter_parser.rb
+++ b/lib/fluent/plugin/filter_parser.rb
@@ -1,10 +1,31 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+require 'fluent/time'
+require 'fluent/config/error'
 require 'fluent/plugin/filter'
-require 'fluent/plugin/parser'
+require 'fluent/plugin_helper/parser'
+require 'fluent/plugin_helper/compat_parameters'
 
 module Fluent::Plugin
   class ParserFilter < Filter
     Fluent::Plugin.register_filter('parser', self)
 
+    helpers :parser, :compat_parameters
+
     config_param :key_name, :string
     config_param :reserve_data, :bool, default: false
     config_param :inject_key_prefix, :string, default: nil
@@ -16,20 +37,22 @@ class ParserFilter < Filter
 
     attr_reader :parser
 
-    def initialize
-      super
-      require 'time'
-    end
-
     def configure(conf)
+      compat_parameters_convert(conf, :parser)
+      parser_config = conf.elements('parse').first
+      unless parser_config
+        raise Fluent::ConfigError, "<parse> section is required"
+      end
+      unless parser_config["@type"]
+        raise Fluent::ConfigError, "parse/@type is required"
+      end
+
       super
 
-      @parser = Fluent::TextParser.new
-      @parser.estimate_current_event = false
-      @parser.configure(conf)
-      if !@time_parse && @parser.parser.respond_to?("time_key=".to_sym)
+      @parser = parser_create(usage: 'filter_parser', conf: parser_config)
+      if !@time_parse && @parser.respond_to?("time_key=".to_sym)
         # disable parse time
-        @parser.parser.time_key = nil
+        @parser.time_key = nil
       end
 
       self
diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index c5815f0cf0..61a09976a4 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -1,21 +1,25 @@
 require_relative '../helper'
+require 'timecop'
 require 'fluent/test/driver/filter'
 require 'fluent/plugin/filter_parser'
 
 class ParserFilterTest < Test::Unit::TestCase
   def setup
     Fluent::Test.setup
+    @tag = 'test'
+    @default_time = Time.parse('2010-05-04 03:02:01 UTC')
+    Timecop.freeze(@default_time)
   end
 
   CONFIG = %[
-    key_name      message
-    format        /^(?<x>.)(?<y>.) (?<time>.+)$/
-    time_format   %Y%m%d%H%M%S
-    reserve_data  true
+    key_name     message
+    format       /^(?<x>.)(?<y>.) (?<time>.+)$/
+    time_format  %Y%m%d%H%M%S
+    reserve_data true
   ]
 
-  def create_driver(conf=CONFIG,tag='test')
-    Fluent::Test::FilterTestDriver.new(Fluent::Plugin::ParserFilter, tag).configure(conf)
+  def create_driver(conf=CONFIG)
+    Fluent::Test::Driver::Filter.new(Fluent::Plugin::ParserFilter).configure(conf)
   end
 
   def test_configure
@@ -82,7 +86,7 @@ def test_configure
       key_name foo
       format /(?<x>.)/
     ]
-    assert_equal false, d.instance.reserve_data
+    assert_false d.instance.reserve_data
   end
 
   # CONFIG = %[
@@ -94,90 +98,82 @@ def test_configure
   #   reserve_data  true
   # ]
   def test_filter
-    d1 = create_driver(CONFIG, 'test.no.change')
+    d1 = create_driver(CONFIG)
     time = Time.parse("2012-01-02 13:14:15").to_i
     d1.run do
-      d1.filter({'message' => '12 20120402182059'}, time)
-      d1.filter({'message' => '34 20120402182100'}, time)
-      d1.filter({'message' => '56 20120402182100'}, time)
-      d1.filter({'message' => '78 20120402182101'}, time)
-      d1.filter({'message' => '90 20120402182100'}, time)
+      d1.feed(@tag, time, {'message' => '12 20120402182059'})
+      d1.feed(@tag, time, {'message' => '34 20120402182100'})
+      d1.feed(@tag, time, {'message' => '56 20120402182100'})
+      d1.feed(@tag, time, {'message' => '78 20120402182101'})
+      d1.feed(@tag, time, {'message' => '90 20120402182100'})
     end
-    filtered = d1.filtered_as_array
+    filtered = d1.filtered
     assert_equal 5, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal Time.parse("2012-04-02 18:20:59").to_i, first[1]
-    assert_equal '1', first[2]['x']
-    assert_equal '2', first[2]['y']
-    assert_equal '12 20120402182059', first[2]['message']
+    assert_equal Time.parse("2012-04-02 18:20:59").to_i, first[0]
+    assert_equal '1', first[1]['x']
+    assert_equal '2', first[1]['y']
+    assert_equal '12 20120402182059', first[1]['message']
 
     second = filtered[1]
-    assert_equal 'test.no.change', second[0]
-    assert_equal Time.parse("2012-04-02 18:21:00").to_i, second[1]
-    assert_equal '3', second[2]['x']
-    assert_equal '4', second[2]['y']
+    assert_equal Time.parse("2012-04-02 18:21:00").to_i, second[0]
+    assert_equal '3', second[1]['x']
+    assert_equal '4', second[1]['y']
 
     third = filtered[2]
-    assert_equal 'test.no.change', third[0]
-    assert_equal Time.parse("2012-04-02 18:21:00").to_i, third[1]
-    assert_equal '5', third[2]['x']
-    assert_equal '6', third[2]['y']
+    assert_equal Time.parse("2012-04-02 18:21:00").to_i, third[0]
+    assert_equal '5', third[1]['x']
+    assert_equal '6', third[1]['y']
 
     fourth = filtered[3]
-    assert_equal 'test.no.change', fourth[0]
-    assert_equal Time.parse("2012-04-02 18:21:01").to_i, fourth[1]
-    assert_equal '7', fourth[2]['x']
-    assert_equal '8', fourth[2]['y']
+    assert_equal Time.parse("2012-04-02 18:21:01").to_i, fourth[0]
+    assert_equal '7', fourth[1]['x']
+    assert_equal '8', fourth[1]['y']
 
     fifth = filtered[4]
-    assert_equal 'test.no.change', fifth[0]
-    assert_equal Time.parse("2012-04-02 18:21:00").to_i, fifth[1]
-    assert_equal '9', fifth[2]['x']
-    assert_equal '0', fifth[2]['y']
+    assert_equal Time.parse("2012-04-02 18:21:00").to_i, fifth[0]
+    assert_equal '9', fifth[1]['x']
+    assert_equal '0', fifth[1]['y']
 
     d2 = create_driver(%[
-      tag parsed
-      key_name      data
-      format        /^(?<x>.)(?<y>.) (?<t>.+)$/
-    ], 'test.no.change')
-    time = Time.parse("2012-04-02 18:20:59").to_i
+      key_name data
+      format   /^(?<x>.)(?<y>.) (?<t>.+)$/
+    ])
+    time = Fluent::EventTime.from_time(@default_time) # EventTime emit test
     d2.run do
-      d2.filter({'data' => '12 20120402182059'}, time)
-      d2.filter({'data' => '34 20120402182100'}, time)
+      d2.feed(@tag, time, {'data' => '12 20120402182059'})
+      d2.feed(@tag, time, {'data' => '34 20120402182100'})
     end
-    filtered = d2.filtered_as_array
+    filtered = d2.filtered
     assert_equal 2, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal time, first[1]
-    assert_nil first[2]['data']
-    assert_equal '1', first[2]['x']
-    assert_equal '2', first[2]['y']
-    assert_equal '20120402182059', first[2]['t']
+    assert_equal time, first[0]
+    assert_nil first[1]['data']
+    assert_equal '1', first[1]['x']
+    assert_equal '2', first[1]['y']
+    assert_equal '20120402182059', first[1]['t']
 
     second = filtered[1]
-    assert_equal 'test.no.change', second[0]
-    assert_equal time, second[1]
-    assert_nil second[2]['data']
-    assert_equal '3', second[2]['x']
-    assert_equal '4', second[2]['y']
-    assert_equal '20120402182100', second[2]['t']
+    assert_equal time, second[0]
+    assert_nil second[1]['data']
+    assert_equal '3', second[1]['x']
+    assert_equal '4', second[1]['y']
+    assert_equal '20120402182100', second[1]['t']
 
     d3 = create_driver(%[
       tag parsed
       key_name      data
       format        /^(?<x>[0-9])(?<y>[0-9]) (?<t>.+)$/
-    ], 'test.no.change')
+    ])
     time = Time.parse("2012-04-02 18:20:59").to_i
     d3.run do
-      d3.filter({'data' => '12 20120402182059'}, time)
-      d3.filter({'data' => '34 20120402182100'}, time)
-      d3.filter({'data' => 'xy 20120402182101'}, time)
+      d3.feed(@tag, time, {'data' => '12 20120402182059'})
+      d3.feed(@tag, time, {'data' => '34 20120402182100'})
+      d3.feed(@tag, time, {'data' => 'xy 20120402182101'})
     end
-    filtered = d3.filtered_as_array
+    filtered = d3.filtered
     assert_equal 2, filtered.length
 
     d3x = create_driver(%[
@@ -185,27 +181,27 @@ def test_filter
       key_name      data
       format        /^(?<x>\\d)(?<y>\\d) (?<t>.+)$/
       reserve_data  yes
-    ], 'test.no.change')
+    ])
     time = Time.parse("2012-04-02 18:20:59").to_i
     d3x.run do
-      d3x.filter({'data' => '12 20120402182059'}, time)
-      d3x.filter({'data' => '34 20120402182100'}, time)
-      d3x.filter({'data' => 'xy 20120402182101'}, time)
+      d3x.feed(@tag, time, {'data' => '12 20120402182059'})
+      d3x.feed(@tag, time, {'data' => '34 20120402182100'})
+      d3x.feed(@tag, time, {'data' => 'xy 20120402182101'})
     end
-    filtered = d3x.filtered_as_array
+    filtered = d3x.filtered
     assert_equal 3, filtered.length
 
     d4 = create_driver(%[
       tag parsed
       key_name      data
       format        json
-    ], 'test.no.change')
+    ])
     time = Time.parse("2012-04-02 18:20:59").to_i
     d4.run do
-      d4.filter({'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'}, time)
-      d4.filter({'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'}, time)
+      d4.feed(@tag, time, {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'})
+      d4.feed(@tag, time, {'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'})
     end
-    filtered = d4.filtered_as_array
+    filtered = d4.filtered
     assert_equal 1, filtered.length
 
     d4x = create_driver(%[
@@ -213,28 +209,26 @@ def test_filter
       key_name      data
       format        json
       reserve_data  yes
-    ], 'test.no.change')
-    time = Time.parse("2012-04-02 18:20:59").to_i
+    ])
+    time = @default_time.to_i
     d4x.run do
-      d4x.filter({'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'}, time)
-      d4x.filter({'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'}, time)
+      d4x.feed(@tag, time, {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'})
+      d4x.feed(@tag, time, {'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'})
     end
-    filtered = d4x.filtered_as_array
+    filtered = d4x.filtered
     assert_equal 2, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal time, first[1]
-    assert_equal '{"xxx":"first","yyy":"second"}', first[2]['data']
-    assert_equal 'first', first[2]['xxx']
-    assert_equal 'second', first[2]['yyy']
+    assert_equal time, first[0]
+    assert_equal '{"xxx":"first","yyy":"second"}', first[1]['data']
+    assert_equal 'first', first[1]['xxx']
+    assert_equal 'second', first[1]['yyy']
 
     second = filtered[1]
-    assert_equal 'test.no.change', second[0]
-    assert_equal time, second[1]
-    assert_equal 'foobar', second[2]['data']
-    assert_equal 'x', second[2]['xxx']
-    assert_equal 'y', second[2]['yyy']
+    assert_equal time, second[0]
+    assert_equal 'foobar', second[1]['data']
+    assert_equal 'x', second[1]['xxx']
+    assert_equal 'y', second[1]['yyy']
   end
 
   CONFIG_LTSV =  %[
@@ -242,73 +236,64 @@ def test_filter
     key_name data
   ]
   def test_filter_ltsv
-    d = create_driver(CONFIG_LTSV, 'test.no.change')
-    time = Time.parse("2012-04-02 18:20:59").to_i
+    d = create_driver(CONFIG_LTSV)
+    time = @default_time.to_i
     d.run do
-      d.filter({'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'}, time)
-      d.filter({'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'}, time)
+      d.feed(@tag, time, {'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'})
+      d.feed(@tag, time, {'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'})
     end
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 2, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal time, first[1]
-    assert_nil first[2]['data']
-    assert_equal 'first', first[2]['xxx']
-    assert_equal 'second', first[2]['yyy']
+    assert_equal time, first[0]
+    assert_nil first[1]['data']
+    assert_equal 'first', first[1]['xxx']
+    assert_equal 'second', first[1]['yyy']
 
     second = filtered[1]
-    assert_equal 'test.no.change', second[0]
-    assert_equal time, second[1]
-    assert_nil first[2]['data']
-    assert_equal 'first', second[2]['xxx']
-    assert_equal 'second2', second[2]['yyy']
-
-    d = create_driver(CONFIG_LTSV + %[
-      reserve_data yes
-    ], 'test.no.change')
-    time = Time.parse("2012-04-02 18:20:59").to_i
+    assert_equal time, second[0]
+    assert_nil first[1]['data']
+    assert_equal 'first', second[1]['xxx']
+    assert_equal 'second2', second[1]['yyy']
+
+    d = create_driver(CONFIG_LTSV + %[reserve_data yes])
+    time = @default_time.to_i
     d.run do
-      d.filter({'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'}, time)
-      d.filter({'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'}, time)
+      d.feed(@tag, time, {'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'})
+      d.feed(@tag, time, {'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'})
     end
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 2, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal time, first[1]
-    assert_equal "xxx:first\tyyy:second", first[2]['data']
-    assert_equal 'first', first[2]['xxx']
-    assert_equal 'second', first[2]['yyy']
+    assert_equal time, first[0]
+    assert_equal "xxx:first\tyyy:second", first[1]['data']
+    assert_equal 'first', first[1]['xxx']
+    assert_equal 'second', first[1]['yyy']
 
     second = filtered[1]
-    assert_equal 'test.no.change', second[0]
-    assert_equal time, second[1]
-    assert_equal "xxx:first\tyyy:second", first[2]['data']
-    assert_equal 'first', second[2]['xxx']
-    assert_equal 'second2', second[2]['yyy']
+    assert_equal time, second[0]
+    assert_equal "xxx:first\tyyy:second", first[1]['data']
+    assert_equal 'first', second[1]['xxx']
+    assert_equal 'second2', second[1]['yyy']
 
     # convert types
-    d = create_driver(CONFIG_LTSV + %[
-      types i:integer,s:string,f:float,b:bool
-    ], 'test.no.change')
-    time = Time.parse("2012-04-02 18:20:59").to_i
+    d = create_driver(CONFIG_LTSV + %[types i:integer,s:string,f:float,b:bool])
+    time = @default_time.to_i
     d.run do
-      d.filter({'data' => "i:1\ts:2\tf:3\tb:true\tx:123"}, time)
+      d.feed(@tag, time, {'data' => "i:1\ts:2\tf:3\tb:true\tx:123"})
     end
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal time, first[1]
-    assert_equal 1, first[2]['i']
-    assert_equal '2', first[2]['s']
-    assert_equal 3.0, first[2]['f']
-    assert_equal true, first[2]['b']
-    assert_equal '123', first[2]['x']
+    assert_equal time, first[0]
+    assert_equal 1, first[1]['i']
+    assert_equal '2', first[1]['s']
+    assert_equal 3.0, first[1]['f']
+    assert_equal true, first[1]['b']
+    assert_equal '123', first[1]['x']
   end
 
   CONFIG_TSV =  %[
@@ -317,21 +302,20 @@ def test_filter_ltsv
     keys key1,key2,key3
   ]
   def test_filter_tsv
-    d = create_driver(CONFIG_TSV, 'test.no.change')
-    time = Time.parse("2012-04-02 18:20:59").to_i
+    d = create_driver(CONFIG_TSV)
+    time = @default_time.to_i
     d.run do
-      d.filter({'data' => "value1\tvalue2\tvalueThree", 'xxx' => 'x', 'yyy' => 'y'}, time)
+      d.feed(@tag, time, {'data' => "value1\tvalue2\tvalueThree", 'xxx' => 'x', 'yyy' => 'y'})
     end
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal time, first[1]
-    assert_nil first[2]['data']
-    assert_equal 'value1', first[2]['key1']
-    assert_equal 'value2', first[2]['key2']
-    assert_equal 'valueThree', first[2]['key3']
+    assert_equal time, first[0]
+    assert_nil first[1]['data']
+    assert_equal 'value1', first[1]['key1']
+    assert_equal 'value2', first[1]['key2']
+    assert_equal 'valueThree', first[1]['key3']
   end
 
   CONFIG_CSV =  %[
@@ -340,21 +324,20 @@ def test_filter_tsv
     keys key1,key2,key3
   ]
   def test_filter_csv
-    d = create_driver(CONFIG_CSV, 'test.no.change')
-    time = Time.parse("2012-04-02 18:20:59").to_i
+    d = create_driver(CONFIG_CSV)
+    time = @default_time.to_i
     d.run do
-      d.filter({'data' => 'value1,"value2","value""ThreeYes!"', 'xxx' => 'x', 'yyy' => 'y'}, time)
+      d.feed(@tag, time, {'data' => 'value1,"value2","value""ThreeYes!"', 'xxx' => 'x', 'yyy' => 'y'})
     end
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal time, first[1]
-    assert_nil first[2]['data']
-    assert_equal 'value1', first[2]['key1']
-    assert_equal 'value2', first[2]['key2']
-    assert_equal 'value"ThreeYes!', first[2]['key3']
+    assert_equal time, first[0]
+    assert_nil first[1]['data']
+    assert_equal 'value1', first[1]['key1']
+    assert_equal 'value2', first[1]['key2']
+    assert_equal 'value"ThreeYes!', first[1]['key3']
   end
 
   CONFIG_HASH_VALUE_FIELD = %[
@@ -377,54 +360,51 @@ def test_filter_csv
   def test_filter_inject_hash_value_field
     original = {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'}
 
-    d = create_driver(CONFIG_HASH_VALUE_FIELD, 'test.no.change')
-    time = Time.parse("2012-04-02 18:20:59").to_i
+    d = create_driver(CONFIG_HASH_VALUE_FIELD)
+    time = @default_time.to_i
     d.run do
-      d.filter(original, time)
+      d.feed(@tag, time, original)
     end
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal time, first[1]
+    assert_equal time, first[0]
 
-    record = first[2]
+    record = first[1]
     assert_equal 1, record.keys.size
     assert_equal({"xxx"=>"first","yyy"=>"second"}, record['parsed'])
 
-    d = create_driver(CONFIG_HASH_VALUE_FIELD_RESERVE_DATA, 'test.no.change')
-    time = Time.parse("2012-04-02 18:20:59").to_i
+    d = create_driver(CONFIG_HASH_VALUE_FIELD_RESERVE_DATA)
+    time = @default_time.to_i
     d.run do
-      d.filter(original, time)
+      d.feed(@tag, time, original)
     end
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal time, first[1]
+    assert_equal time, first[0]
 
-    record = first[2]
+    record = first[1]
     assert_equal 4, record.keys.size
     assert_equal original['data'], record['data']
     assert_equal original['xxx'], record['xxx']
     assert_equal original['yyy'], record['yyy']
     assert_equal({"xxx"=>"first","yyy"=>"second"}, record['parsed'])
 
-    d = create_driver(CONFIG_HASH_VALUE_FIELD_WITH_INJECT_KEY_PREFIX, 'test.no.change')
-    time = Time.parse("2012-04-02 18:20:59").to_i
+    d = create_driver(CONFIG_HASH_VALUE_FIELD_WITH_INJECT_KEY_PREFIX)
+    time = @default_time.to_i
     d.run do
-      d.filter(original, time)
+      d.feed(@tag, time, original)
     end
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal 'test.no.change', first[0]
-    assert_equal time, first[1]
+    assert_equal time, first[0]
 
-    record = first[2]
+    record = first[1]
     assert_equal 1, record.keys.size
     assert_equal({"data.xxx"=>"first","data.yyy"=>"second"}, record['parsed'])
   end
@@ -436,30 +416,26 @@ def test_filter_inject_hash_value_field
   ]
   def test_time_should_be_reserved
     t = Time.now.to_i
-    d = create_driver(CONFIG_DONT_PARSE_TIME, 'test.no.change')
-
+    d = create_driver(CONFIG_DONT_PARSE_TIME)
     d.run do
-      d.filter({'data' => '{"time":1383190430, "f1":"v1"}'}, t)
-      d.filter({'data' => '{"time":"1383190430", "f1":"v1"}'}, t)
-      d.filter({'data' => '{"time":"2013-10-31 12:34:03 +0900", "f1":"v1"}'}, t)
+      d.feed(@tag, t, {'data' => '{"time":1383190430, "f1":"v1"}'})
+      d.feed(@tag, t, {'data' => '{"time":"1383190430", "f1":"v1"}'})
+      d.feed(@tag, t, {'data' => '{"time":"2013-10-31 12:34:03 +0900", "f1":"v1"}'})
     end
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 3, filtered.length
 
-    assert_equal 'test.no.change', filtered[0][0]
-    assert_equal 'v1', filtered[0][2]['f1']
-    assert_equal 1383190430, filtered[0][2]['time']
-    assert_equal t, filtered[0][1]
+    assert_equal 'v1', filtered[0][1]['f1']
+    assert_equal 1383190430, filtered[0][1]['time']
+    assert_equal t, filtered[0][0]
 
-    assert_equal 'test.no.change', filtered[1][0]
-    assert_equal 'v1', filtered[1][2]['f1']
-    assert_equal "1383190430", filtered[1][2]['time']
-    assert_equal t, filtered[1][1]
+    assert_equal 'v1', filtered[1][1]['f1']
+    assert_equal "1383190430", filtered[1][1]['time']
+    assert_equal t, filtered[1][0]
 
-    assert_equal 'test.no.change', filtered[2][0]
-    assert_equal 'v1', filtered[2][2]['f1']
-    assert_equal '2013-10-31 12:34:03 +0900', filtered[2][2]['time']
-    assert_equal t, filtered[2][1]
+    assert_equal 'v1', filtered[2][1]['f1']
+    assert_equal '2013-10-31 12:34:03 +0900', filtered[2][1]['time']
+    assert_equal t, filtered[2][0]
   end
 
   CONFIG_INVALID_TIME_VALUE = %[
@@ -469,21 +445,20 @@ def test_time_should_be_reserved
   ] # 'time' is implicit @time_key
   def test_filter_invalid_time_data
     # should not raise errors
-    t = Time.now.to_i
-    d = create_driver(CONFIG_INVALID_TIME_VALUE, 'test.no.change')
+    time = Time.now.to_i
+    d = create_driver(CONFIG_INVALID_TIME_VALUE)
     assert_nothing_raised {
       d.run do
-        d.filter({'data' => '{"time":[], "f1":"v1"}'}, t)
-        d.filter({'data' => '{"time":"thisisnottime", "f1":"v1"}'}, t)
+        d.feed(@tag, time, {'data' => '{"time":[], "f1":"v1"}'})
+        d.feed(@tag, time, {'data' => '{"time":"thisisnottime", "f1":"v1"}'})
       end
     }
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
 
-    assert_equal 'test.no.change', filtered[0][0]
-    assert_equal 0, filtered[0][1]
-    assert_equal 'v1', filtered[0][2]['f1']
-    assert_equal 0, filtered[0][2]['time'].to_i
+    assert_equal 0, filtered[0][0]
+    assert_equal 'v1', filtered[0][1]['f1']
+    assert_equal 0, filtered[0][1]['time'].to_i
   end
 
   # REGEXP = /^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$/
@@ -499,48 +474,46 @@ def test_filter_invalid_time_data
   def test_filter_invalid_byte
     invalid_utf8 = "\xff".force_encoding('UTF-8')
 
-    d = create_driver(CONFIG_NOT_REPLACE, 'test.no.change')
+    d = create_driver(CONFIG_NOT_REPLACE)
     assert_raise(ArgumentError) {
       d.run do
-        d.filter({'data' => invalid_utf8}, Time.now.to_i)
+        d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => invalid_utf8})
       end
     }
 
-    d = create_driver(CONFIG_INVALID_BYTE, 'test.in')
+    d = create_driver(CONFIG_INVALID_BYTE)
     assert_nothing_raised {
       d.run do
-        d.emit({'data' => invalid_utf8}, Time.now.to_i)
+        d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => invalid_utf8})
       end
     }
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
-    assert_nil filtered[0][2]['data']
-    assert_equal '?'.force_encoding('UTF-8'), filtered[0][2]['message']
+    assert_nil filtered[0][1]['data']
+    assert_equal '?'.force_encoding('UTF-8'), filtered[0][1]['message']
 
-    d = create_driver(CONFIG_INVALID_BYTE + %[
-      reserve_data yes
-    ], 'test.no.change')
+    d = create_driver(CONFIG_INVALID_BYTE + %[reserve_data yes])
     assert_nothing_raised {
       d.run do
-        d.filter({'data' => invalid_utf8}, Time.now.to_i)
+        d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => invalid_utf8})
       end
     }
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
-    assert_equal invalid_utf8, filtered[0][2]['data']
-    assert_equal '?'.force_encoding('UTF-8'), filtered[0][2]['message']
+    assert_equal invalid_utf8, filtered[0][1]['data']
+    assert_equal '?'.force_encoding('UTF-8'), filtered[0][1]['message']
 
     invalid_ascii = "\xff".force_encoding('US-ASCII')
-    d = create_driver(CONFIG_INVALID_BYTE, 'test.no.change')
+    d = create_driver(CONFIG_INVALID_BYTE)
     assert_nothing_raised {
       d.run do
-        d.filter({'data' => invalid_ascii}, Time.now.to_i)
+        d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => invalid_ascii})
       end
     }
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
-    assert_nil filtered[0][2]['data']
-    assert_equal '?'.force_encoding('US-ASCII'), filtered[0][2]['message']
+    assert_nil filtered[0][1]['data']
+    assert_equal '?'.force_encoding('US-ASCII'), filtered[0][1]['message']
   end
 
   CONFIG_NOT_IGNORE = %[
@@ -556,32 +529,34 @@ def test_filter_invalid_byte
     reserve_data true
   ]
   def test_filter_key_not_exist
-    d = create_driver(CONFIG_NOT_IGNORE, 'test.no.ignore')
+    d = create_driver(CONFIG_NOT_IGNORE)
     assert_nothing_raised {
-      d.run do
-        d.filter({'foo' => 'bar'}, Time.now.to_i)
+      d.run(shutdown: false) do
+        d.feed(@tag, Fluent::EventTime.now.to_i, {'foo' => 'bar'})
       end
     }
-    assert_match(/data does not exist/, d.instance.log.out.logs.first)
+    assert_match(/data does not exist/, d.instance.log.logs.first)
+    d.instance_shutdown
 
-    d = create_driver(CONFIG_IGNORE, 'test.ignore')
+    d = create_driver(CONFIG_IGNORE)
     assert_nothing_raised {
-      d.run do
-        d.filter({'foo' => 'bar'}, Time.now.to_i)
+      d.run(shutdown: false) do
+        d.feed(@tag, Fluent::EventTime.now.to_i, {'foo' => 'bar'})
       end
     }
-    assert_not_match(/data does not exist/, d.instance.log.out.logs.first)
+    assert_not_match(/data does not exist/, d.instance.log.logs.first)
+    d.instance_shutdown
 
-    d = create_driver(CONFIG_PASS_SAME_RECORD, 'test.pass_same_record')
+    d = create_driver(CONFIG_PASS_SAME_RECORD)
     assert_nothing_raised {
       d.run do
-        d.filter({'foo' => 'bar'}, Time.now.to_i)
+        d.feed(@tag, Fluent::EventTime.now.to_i, {'foo' => 'bar'})
       end
     }
-    filtered = d.filtered_as_array
+    filtered = d.filtered
     assert_equal 1, filtered.length
-    assert_nil filtered[0][2]['data']
-    assert_equal 'bar', filtered[0][2]['foo']
+    assert_nil filtered[0][1]['data']
+    assert_equal 'bar', filtered[0][1]['foo']
   end
 
   # suppress_parse_error_log test
@@ -629,27 +604,27 @@ def swap_logger(instance)
   end
 
   def test_parser_error_warning
-    d = create_driver(CONFIG_INVALID_TIME_VALUE, 'test.no.change')
+    d = create_driver(CONFIG_INVALID_TIME_VALUE)
     swap_logger(d.instance) do
       assert_raise(DummyLoggerWarnedException) {
         d.run do
-          d.filter({'data' => '{"time":[], "f1":"v1"}'}, Time.now.to_i)
+          d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => '{"time":[], "f1":"v1"}'})
         end
       }
     end
   end
 
   class DefaultSuppressParseErrorLogTest < self
-    def setup
+    setup do
       # default(disabled) 'suppress_parse_error_log' is not specify
-      @d = create_driver(CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG, 'test.no.change')
+      @d = create_driver(CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG)
     end
 
     def test_raise_exception
       swap_logger(@d.instance) do
         assert_raise(DummyLoggerWarnedException) {
           @d.run do
-            @d.filter({'message' => INVALID_MESSAGE}, Time.now.to_i)
+            @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
           end
         }
       end
@@ -659,7 +634,7 @@ def test_nothing_raised
       swap_logger(@d.instance) do
       assert_nothing_raised {
         @d.run do
-          @d.filter({'message' => VALID_MESSAGE}, Time.now.to_i)
+          @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
         end
       }
       end
@@ -667,16 +642,16 @@ def test_nothing_raised
   end
 
   class DisabledSuppressParseErrorLogTest < self
-    def setup
+    setup do
       # disabled 'suppress_parse_error_log'
-      @d = create_driver(CONFIG_DISABELED_SUPPRESS_PARSE_ERROR_LOG, 'test.no.change')
+      @d = create_driver(CONFIG_DISABELED_SUPPRESS_PARSE_ERROR_LOG)
     end
 
     def test_raise_exception
       swap_logger(@d.instance) do
         assert_raise(DummyLoggerWarnedException) {
           @d.run do
-            @d.filter({'message' => INVALID_MESSAGE}, Time.now.to_i)
+            @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
           end
         }
       end
@@ -686,7 +661,7 @@ def test_nothing_raised
       swap_logger(@d.instance) do
       assert_nothing_raised {
         @d.run do
-          @d.filter({'message' => VALID_MESSAGE}, Time.now.to_i)
+          @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
         end
       }
       end
@@ -694,17 +669,17 @@ def test_nothing_raised
   end
 
   class EnabledSuppressParseErrorLogTest < self
-    def setup
+    setup do
       # enabled 'suppress_parse_error_log'
-      @d = create_driver(CONFIG_ENABELED_SUPPRESS_PARSE_ERROR_LOG, 'test.no.change')
+      @d = create_driver(CONFIG_ENABELED_SUPPRESS_PARSE_ERROR_LOG)
     end
 
     def test_nothing_raised
       swap_logger(@d.instance) do
       assert_nothing_raised {
         @d.run do
-          @d.filter({'message' => INVALID_MESSAGE}, Time.now.to_i)
-          @d.filter({'message' => VALID_MESSAGE},   Time.now.to_i)
+          @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
+          @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
         end
       }
       end

From a11d4fb6692c5b76e23586c960514e4cdf0c5c91 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Wed, 31 Aug 2016 14:18:21 +0900
Subject: [PATCH 03/19] Add missing parser parameters to compat_parameters

---
 lib/fluent/plugin_helper/compat_parameters.rb | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/lib/fluent/plugin_helper/compat_parameters.rb b/lib/fluent/plugin_helper/compat_parameters.rb
index c5b2f3a3c6..12faf24226 100644
--- a/lib/fluent/plugin_helper/compat_parameters.rb
+++ b/lib/fluent/plugin_helper/compat_parameters.rb
@@ -58,6 +58,9 @@ module CompatParameters
         "time_key"    => "time_key",
         "time_format" => "time_format",
         "delimiter"   => "delimiter",
+        "keep_time_key" => "keep_time_key",
+        "null_empty_string" => "null_empty_string", # ValuesParser
+        "null_value_pattern" => "null_value_pattern", # ValuesParser
         "json_parser"      => "json_parser", # JSONParser
         "label_delimiter"  => "label_delimiter", # LabeledTSVParser
         "format_firstline" => "format_firstline", # MultilineParser

From a0c8f7eb02708672db2af44f0425cba12c9a3d83 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 1 Sep 2016 18:05:21 +0900
Subject: [PATCH 04/19] Call Timecop.return in teardown to restore Time

---
 test/plugin/test_filter_parser.rb | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index 61a09976a4..37cb336816 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -11,6 +11,10 @@ def setup
     Timecop.freeze(@default_time)
   end
 
+  def teardown
+    Timecop.return
+  end
+
   CONFIG = %[
     key_name     message
     format       /^(?<x>.)(?<y>.) (?<time>.+)$/

From b166787d9258946574e35d0afff094d86ee4eea9 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Fri, 2 Sep 2016 18:34:23 +0900
Subject: [PATCH 05/19] Use filter_with_time API

Default filter_stream implementation catches exceptions and
exceptions are routed to error stream.
So warning tests are changed to log check way.
---
 lib/fluent/plugin/filter_parser.rb |  84 +++++++++---------
 test/plugin/test_filter_parser.rb  | 134 +++++++++++------------------
 2 files changed, 97 insertions(+), 121 deletions(-)

diff --git a/lib/fluent/plugin/filter_parser.rb b/lib/fluent/plugin/filter_parser.rb
index f2b4426bf0..3c73f724c5 100644
--- a/lib/fluent/plugin/filter_parser.rb
+++ b/lib/fluent/plugin/filter_parser.rb
@@ -58,60 +58,66 @@ def configure(conf)
       self
     end
 
-    def filter_stream(tag, es)
-      new_es = Fluent::MultiEventStream.new
-      es.each do |time, record|
-        raw_value = record[@key_name]
-        if raw_value.nil?
-          log.warn "#{@key_name} does not exist" unless @ignore_key_not_exist
-          new_es.add(time, handle_parsed(tag, record, time, {})) if @reserve_data
-          next
+    FAILED_RESULT = [nil, nil].freeze # reduce allocation cost 
+
+    def filter_with_time(tag, time, record)
+      raw_value = record[@key_name]
+      if raw_value.nil?
+        log.warn "#{@key_name} does not exist" unless @ignore_key_not_exist
+        if @reserve_data
+          return time, handle_parsed(tag, record, time, {})
+        else
+          return FAILED_RESULT
+        end
+      end
+      begin
+        @parser.parse(raw_value) do |t, values|
+          if values
+            t ||= time
+            r = handle_parsed(tag, record, t, values)
+            return t, r
+          else
+            log.warn "pattern not match with data '#{raw_value}'" unless @suppress_parse_error_log
+            if @reserve_data
+              t = time
+              r = handle_parsed(tag, record, time, {})
+              return t, r
+            else
+              return FAILED_RESULT
+            end
+          end
         end
-        begin
-          @parser.parse(raw_value) do |t, values|
+      rescue Fluent::TextParser::ParserError => e
+        log.warn e.message unless @suppress_parse_error_log
+      rescue ArgumentError => e
+        if @replace_invalid_sequence
+          unless e.message.index("invalid byte sequence in") == 0
+            raise
+          end
+          replaced_string = replace_invalid_byte(raw_value)
+          @parser.parse(replaced_string) do |t, values|
             if values
               t ||= time
               r = handle_parsed(tag, record, t, values)
-              new_es.add(t, r)
+              return t, r
             else
               log.warn "pattern not match with data '#{raw_value}'" unless @suppress_parse_error_log
               if @reserve_data
                 t = time
                 r = handle_parsed(tag, record, time, {})
-                new_es.add(t, r)
-              end
-            end
-          end
-        rescue Fluent::TextParser::ParserError => e
-          log.warn e.message unless @suppress_parse_error_log
-        rescue ArgumentError => e
-          if @replace_invalid_sequence
-            unless e.message.index("invalid byte sequence in") == 0
-              raise
-            end
-            replaced_string = replace_invalid_byte(raw_value)
-            @parser.parse(replaced_string) do |t, values|
-              if values
-                t ||= time
-                r = handle_parsed(tag, record, t, values)
-                new_es.add(t, r)
+                return t, r
               else
-                log.warn "pattern not match with data '#{raw_value}'" unless @suppress_parse_error_log
-                if @reserve_data
-                  t = time
-                  r = handle_parsed(tag, record, time, {})
-                  new_es.add(t, r)
-                end
+                return FAILED_RESULT
               end
             end
-          else
-            raise
           end
-        rescue => e
-          log.warn "parse failed #{e.message}" unless @suppress_parse_error_log
+        else
+          raise
         end
+      rescue => e
+        log.warn "parse failed #{e.message}" unless @suppress_parse_error_log
+        return FAILED_RESULT
       end
-      new_es
     end
 
     private
diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index 37cb336816..55f10011f6 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -479,11 +479,13 @@ def test_filter_invalid_byte
     invalid_utf8 = "\xff".force_encoding('UTF-8')
 
     d = create_driver(CONFIG_NOT_REPLACE)
-    assert_raise(ArgumentError) {
-      d.run do
-        d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => invalid_utf8})
-      end
-    }
+    d.run(shutdown: false) do
+      d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => invalid_utf8})
+    end
+    error_event = d.error_events.first
+    assert_equal "test", error_event[0]
+    assert_instance_of ArgumentError, error_event[3]
+    d.instance_shutdown      
 
     d = create_driver(CONFIG_INVALID_BYTE)
     assert_nothing_raised {
@@ -564,60 +566,24 @@ def test_filter_key_not_exist
   end
 
   # suppress_parse_error_log test
-  CONFIG_DISABELED_SUPPRESS_PARSE_ERROR_LOG = %[
-    tag hogelog
-    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
-    key_name message
-    suppress_parse_error_log false
-  ]
-  CONFIG_ENABELED_SUPPRESS_PARSE_ERROR_LOG = %[
-    tag hogelog
-    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
-    key_name message
-    suppress_parse_error_log true
-  ]
-  CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG = %[
-    tag hogelog
-    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
-    key_name message
-  ]
-
   INVALID_MESSAGE = 'foo bar'
   VALID_MESSAGE   = 'col1=foo col2=bar'
 
-  # if call warn() raise exception
-  class DummyLoggerWarnedException < StandardError; end
-  class DummyLogger
-    def reset
-    end
-    def warn(message)
-      raise DummyLoggerWarnedException
-    end
-  end
-
-  def swap_logger(instance)
-    raise "use with block" unless block_given?
-    dummy = DummyLogger.new
-    saved_logger = instance.log
-    instance.log = dummy
-    restore = lambda { instance.log = saved_logger }
-
-    yield
-
-    restore.call
-  end
-
   def test_parser_error_warning
     d = create_driver(CONFIG_INVALID_TIME_VALUE)
-    swap_logger(d.instance) do
-      assert_raise(DummyLoggerWarnedException) {
-        d.run do
-          d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => '{"time":[], "f1":"v1"}'})
-        end
-      }
+    d.run(shutdown: false) do
+      d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => '{"time":[], "f1":"v1"}'})
     end
+    assert_match(/\[warn\]: invalid time value:/, d.instance.log.logs.first)
+    d.instance_shutdown
   end
 
+  CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG = %[
+    tag hogelog
+    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+    key_name message
+  ]
+
   class DefaultSuppressParseErrorLogTest < self
     setup do
       # default(disabled) 'suppress_parse_error_log' is not specify
@@ -625,26 +591,29 @@ class DefaultSuppressParseErrorLogTest < self
     end
 
     def test_raise_exception
-      swap_logger(@d.instance) do
-        assert_raise(DummyLoggerWarnedException) {
-          @d.run do
-            @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
-          end
-        }
+      @d.run(shutdown: false) do
+        @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
       end
+      assert_match(/\[warn\]: pattern not match with data/, @d.instance.log.logs.first)
+      @d.instance_shutdown
     end
 
     def test_nothing_raised
-      swap_logger(@d.instance) do
-      assert_nothing_raised {
-        @d.run do
-          @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
-        end
-      }
+      @d.run(shutdown: false) do
+        @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
       end
+      assert_empty @d.instance.log.logs
+      @d.instance_shutdown
     end
   end
 
+  CONFIG_DISABELED_SUPPRESS_PARSE_ERROR_LOG = %[
+    tag hogelog
+    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+    key_name message
+    suppress_parse_error_log false
+  ]
+
   class DisabledSuppressParseErrorLogTest < self
     setup do
       # disabled 'suppress_parse_error_log'
@@ -652,26 +621,29 @@ class DisabledSuppressParseErrorLogTest < self
     end
 
     def test_raise_exception
-      swap_logger(@d.instance) do
-        assert_raise(DummyLoggerWarnedException) {
-          @d.run do
-            @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
-          end
-        }
+      @d.run(shutdown: false) do
+        @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
       end
+      assert_match(/\[warn\]: pattern not match with data/, @d.instance.log.logs.first)
+      @d.instance_shutdown
     end
 
     def test_nothing_raised
-      swap_logger(@d.instance) do
-      assert_nothing_raised {
-        @d.run do
-          @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
-        end
-      }
+      @d.run(shutdown: false) do
+        @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
       end
+      assert_empty @d.instance.log.logs
+      @d.instance_shutdown
     end
   end
 
+  CONFIG_ENABELED_SUPPRESS_PARSE_ERROR_LOG = %[
+    tag hogelog
+    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+    key_name message
+    suppress_parse_error_log true
+  ]
+
   class EnabledSuppressParseErrorLogTest < self
     setup do
       # enabled 'suppress_parse_error_log'
@@ -679,14 +651,12 @@ class EnabledSuppressParseErrorLogTest < self
     end
 
     def test_nothing_raised
-      swap_logger(@d.instance) do
-      assert_nothing_raised {
-        @d.run do
-          @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
-          @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
-        end
-      }
+      @d.run(shutdown: false) do
+        @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
+        @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
       end
+      assert_empty @d.instance.log.logs
+      @d.instance_shutdown
     end
   end
 end

From 9359eeade339d8cf068e2794411e1f86ab5cd74a Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Fri, 4 Nov 2016 15:47:35 +0900
Subject: [PATCH 06/19] Remove duplicated compat parameters

---
 lib/fluent/plugin_helper/compat_parameters.rb | 2 --
 1 file changed, 2 deletions(-)

diff --git a/lib/fluent/plugin_helper/compat_parameters.rb b/lib/fluent/plugin_helper/compat_parameters.rb
index 12faf24226..198b62bf9e 100644
--- a/lib/fluent/plugin_helper/compat_parameters.rb
+++ b/lib/fluent/plugin_helper/compat_parameters.rb
@@ -52,8 +52,6 @@ module CompatParameters
         "types" => nil,
         "types_delimiter" => nil,
         "types_label_delimiter" => nil,
-        "null_value_pattern" => "null_value_pattern",
-        "null_empty_string" => "null_empty_string",
         "keys" => "keys", # CSVParser, TSVParser (old ValuesParser)
         "time_key"    => "time_key",
         "time_format" => "time_format",

From 8e8b92d385dd18e56cddf027c77511ee15e2de10 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Fri, 4 Nov 2016 15:52:20 +0900
Subject: [PATCH 07/19] Use error stream instead of warning logs

---
 lib/fluent/plugin/filter_parser.rb | 15 +++++------
 test/plugin/test_filter_parser.rb  | 43 +++++++++++++++++-------------
 2 files changed, 31 insertions(+), 27 deletions(-)

diff --git a/lib/fluent/plugin/filter_parser.rb b/lib/fluent/plugin/filter_parser.rb
index 3c73f724c5..deca4ecd9d 100644
--- a/lib/fluent/plugin/filter_parser.rb
+++ b/lib/fluent/plugin/filter_parser.rb
@@ -31,9 +31,7 @@ class ParserFilter < Filter
     config_param :inject_key_prefix, :string, default: nil
     config_param :replace_invalid_sequence, :bool, default: false
     config_param :hash_value_field, :string, default: nil
-    config_param :suppress_parse_error_log, :bool, default: false
     config_param :time_parse, :bool, default: true
-    config_param :ignore_key_not_exist, :bool, default: false
 
     attr_reader :parser
 
@@ -63,7 +61,7 @@ def configure(conf)
     def filter_with_time(tag, time, record)
       raw_value = record[@key_name]
       if raw_value.nil?
-        log.warn "#{@key_name} does not exist" unless @ignore_key_not_exist
+        router.emit_error_event(tag, time, record, ArgumentError.new("#{@key_name} does not exist"))
         if @reserve_data
           return time, handle_parsed(tag, record, time, {})
         else
@@ -77,7 +75,7 @@ def filter_with_time(tag, time, record)
             r = handle_parsed(tag, record, t, values)
             return t, r
           else
-            log.warn "pattern not match with data '#{raw_value}'" unless @suppress_parse_error_log
+            router.emit_error_event(tag, time, record, Fluent::Plugin::Parser::ParserError.new("pattern not match with data '#{raw_value}'"))
             if @reserve_data
               t = time
               r = handle_parsed(tag, record, time, {})
@@ -87,8 +85,9 @@ def filter_with_time(tag, time, record)
             end
           end
         end
-      rescue Fluent::TextParser::ParserError => e
-        log.warn e.message unless @suppress_parse_error_log
+      rescue Fluent::Plugin::Parser::ParserError => e
+        router.emit_error_event(tag, time, record, e)
+        return FAILED_RESULT
       rescue ArgumentError => e
         if @replace_invalid_sequence
           unless e.message.index("invalid byte sequence in") == 0
@@ -101,7 +100,7 @@ def filter_with_time(tag, time, record)
               r = handle_parsed(tag, record, t, values)
               return t, r
             else
-              log.warn "pattern not match with data '#{raw_value}'" unless @suppress_parse_error_log
+              router.emit_error_event(tag, time, record, Fluent::Plugin::Parser::ParserError.new("pattern not match with data '#{raw_value}'"))
               if @reserve_data
                 t = time
                 r = handle_parsed(tag, record, time, {})
@@ -115,7 +114,7 @@ def filter_with_time(tag, time, record)
           raise
         end
       rescue => e
-        log.warn "parse failed #{e.message}" unless @suppress_parse_error_log
+        router.emit_error_event(tag, time, record, Fluent::Plugin::Parser::ParserError.new("parse failed #{e.message}"))
         return FAILED_RESULT
       end
     end
diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index 55f10011f6..6ace21459e 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -2,8 +2,11 @@
 require 'timecop'
 require 'fluent/test/driver/filter'
 require 'fluent/plugin/filter_parser'
+require 'flexmock/test_unit'
 
 class ParserFilterTest < Test::Unit::TestCase
+  include FlexMock::TestCase
+
   def setup
     Fluent::Test.setup
     @tag = 'test'
@@ -12,9 +15,11 @@ def setup
   end
 
   def teardown
+    super
     Timecop.return
   end
 
+  ParserError = Fluent::Plugin::Parser::ParserError
   CONFIG = %[
     key_name     message
     format       /^(?<x>.)(?<y>.) (?<time>.+)$/
@@ -536,12 +541,14 @@ def test_filter_invalid_byte
   ]
   def test_filter_key_not_exist
     d = create_driver(CONFIG_NOT_IGNORE)
+    flexmock(d.instance.router).should_receive(:emit_error_event).
+      with(String, Integer, Hash, ArgumentError.new("data does not exist")).once
     assert_nothing_raised {
       d.run(shutdown: false) do
         d.feed(@tag, Fluent::EventTime.now.to_i, {'foo' => 'bar'})
       end
     }
-    assert_match(/data does not exist/, d.instance.log.logs.first)
+    #assert_match(/data does not exist/, d.instance.log.logs.first)
     d.instance_shutdown
 
     d = create_driver(CONFIG_IGNORE)
@@ -571,11 +578,11 @@ def test_filter_key_not_exist
 
   def test_parser_error_warning
     d = create_driver(CONFIG_INVALID_TIME_VALUE)
-    d.run(shutdown: false) do
+    flexmock(d.instance.router).should_receive(:emit_error_event).
+      with(String, Integer, Hash, ParserError).once
+    d.run do
       d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => '{"time":[], "f1":"v1"}'})
     end
-    assert_match(/\[warn\]: invalid time value:/, d.instance.log.logs.first)
-    d.instance_shutdown
   end
 
   CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG = %[
@@ -591,19 +598,18 @@ class DefaultSuppressParseErrorLogTest < self
     end
 
     def test_raise_exception
-      @d.run(shutdown: false) do
+      flexmock(@d.instance.router).should_receive(:emit_error_event).
+        with(String, Integer, Hash, ParserError.new("pattern not match with data '#{INVALID_MESSAGE}'")).once
+      @d.run do
         @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
       end
-      assert_match(/\[warn\]: pattern not match with data/, @d.instance.log.logs.first)
-      @d.instance_shutdown
     end
 
     def test_nothing_raised
-      @d.run(shutdown: false) do
+      flexmock(@d.instance.router).should_receive(:emit_error_event).never
+      @d.run do
         @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
       end
-      assert_empty @d.instance.log.logs
-      @d.instance_shutdown
     end
   end
 
@@ -621,19 +627,18 @@ class DisabledSuppressParseErrorLogTest < self
     end
 
     def test_raise_exception
-      @d.run(shutdown: false) do
+      flexmock(@d.instance.router).should_receive(:emit_error_event).
+        with(String, Integer, Hash, ParserError.new("pattern not match with data '#{INVALID_MESSAGE}'")).once
+      @d.run do
         @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
       end
-      assert_match(/\[warn\]: pattern not match with data/, @d.instance.log.logs.first)
-      @d.instance_shutdown
     end
 
     def test_nothing_raised
-      @d.run(shutdown: false) do
+      flexmock(@d.instance.router).should_receive(:emit_error_event).never
+      @d.run do
         @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
       end
-      assert_empty @d.instance.log.logs
-      @d.instance_shutdown
     end
   end
 
@@ -651,12 +656,12 @@ class EnabledSuppressParseErrorLogTest < self
     end
 
     def test_nothing_raised
-      @d.run(shutdown: false) do
+      flexmock(@d.instance.router).should_receive(:emit_error_event).
+        with(String, Integer, Hash, ParserError).once
+      @d.run do
         @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
         @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
       end
-      assert_empty @d.instance.log.logs
-      @d.instance_shutdown
     end
   end
 end

From c993be5806030395f39f596adfc26e4f3e3850c3 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Fri, 4 Nov 2016 16:40:31 +0900
Subject: [PATCH 08/19] Replace time_parse with reserve_time

The drawback is parser tries to parse time field even if
users don't need parsed time.
If this is a critical, we should revive time_parse option.
---
 lib/fluent/plugin/filter_parser.rb | 26 ++++++++++----------------
 test/plugin/test_filter_parser.rb  |  7 ++++---
 2 files changed, 14 insertions(+), 19 deletions(-)

diff --git a/lib/fluent/plugin/filter_parser.rb b/lib/fluent/plugin/filter_parser.rb
index deca4ecd9d..7897783e25 100644
--- a/lib/fluent/plugin/filter_parser.rb
+++ b/lib/fluent/plugin/filter_parser.rb
@@ -28,32 +28,22 @@ class ParserFilter < Filter
 
     config_param :key_name, :string
     config_param :reserve_data, :bool, default: false
+    config_param :reserve_time, :bool, default: false
     config_param :inject_key_prefix, :string, default: nil
     config_param :replace_invalid_sequence, :bool, default: false
     config_param :hash_value_field, :string, default: nil
-    config_param :time_parse, :bool, default: true
+
+    config_section :parse do
+    end
 
     attr_reader :parser
 
     def configure(conf)
       compat_parameters_convert(conf, :parser)
-      parser_config = conf.elements('parse').first
-      unless parser_config
-        raise Fluent::ConfigError, "<parse> section is required"
-      end
-      unless parser_config["@type"]
-        raise Fluent::ConfigError, "parse/@type is required"
-      end
 
       super
 
-      @parser = parser_create(usage: 'filter_parser', conf: parser_config)
-      if !@time_parse && @parser.respond_to?("time_key=".to_sym)
-        # disable parse time
-        @parser.time_key = nil
-      end
-
-      self
+      @parser = parser_create
     end
 
     FAILED_RESULT = [nil, nil].freeze # reduce allocation cost 
@@ -71,7 +61,11 @@ def filter_with_time(tag, time, record)
       begin
         @parser.parse(raw_value) do |t, values|
           if values
-            t ||= time
+            t = if @reserve_time
+                  time
+                else
+                  t.nil? ? time : t
+                end
             r = handle_parsed(tag, record, t, values)
             return t, r
           else
diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index 6ace21459e..aa50e28a47 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -421,7 +421,8 @@ def test_filter_inject_hash_value_field
   CONFIG_DONT_PARSE_TIME = %[
     key_name data
     format json
-    time_parse no
+    keep_time_key true
+    reserve_time true
   ]
   def test_time_should_be_reserved
     t = Time.now.to_i
@@ -465,9 +466,9 @@ def test_filter_invalid_time_data
     filtered = d.filtered
     assert_equal 1, filtered.length
 
-    assert_equal 0, filtered[0][0]
+    assert_equal 0, filtered[0][0].to_i
     assert_equal 'v1', filtered[0][1]['f1']
-    assert_equal 0, filtered[0][1]['time'].to_i
+    assert_equal nil, filtered[0][1]['time']
   end
 
   # REGEXP = /^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$/

From 5ffcb8784f65af876f1024fc905fa651d8f15d8e Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Fri, 4 Nov 2016 16:59:39 +0900
Subject: [PATCH 09/19] Update test to use new configuration format

Several tests are executed by new and compat configurations
for checking backward compatibility
---
 test/plugin/test_filter_parser.rb | 192 ++++++++++++++++++++----------
 1 file changed, 127 insertions(+), 65 deletions(-)

diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index aa50e28a47..f93a68a552 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -22,9 +22,12 @@ def teardown
   ParserError = Fluent::Plugin::Parser::ParserError
   CONFIG = %[
     key_name     message
-    format       /^(?<x>.)(?<y>.) (?<time>.+)$/
-    time_format  %Y%m%d%H%M%S
     reserve_data true
+    <parse>
+      @type regexp
+      expression /^(?<x>.)(?<y>.) (?<time>.+)$/
+      time_format %Y%m%d%H%M%S
+    </parse>
   ]
 
   def create_driver(conf=CONFIG)
@@ -37,63 +40,69 @@ def test_configure
     }
     assert_raise(Fluent::ConfigError) {
       create_driver %[
-        format unknown_format_that_will_never_be_implemented
         key_name foo
+        <parse>
+          @type unknown_format_that_will_never_be_implemented
+        </parse>
       ]
     }
     assert_nothing_raised {
       create_driver %[
-        format /(?<x>.)/
         key_name foo
+        <parse>
+          @type regexp
+          expression /(?<x>.)/
+        </parse>
       ]
     }
     assert_nothing_raised {
       create_driver %[
-        format /(?<x>.)/
         key_name foo
+        <parse>
+          @type json
+        </parse>
       ]
     }
     assert_nothing_raised {
       create_driver %[
-        format /(?<x>.)/
         key_name foo
-      ]
-    }
-    assert_nothing_raised {
-      create_driver %[
-        format /(?<x>.)/
-        key_name foo
-      ]
-    }
-    assert_nothing_raised {
-      create_driver %[
         format json
-        key_name foo
       ]
     }
     assert_nothing_raised {
       create_driver %[
-        format ltsv
         key_name foo
+        <parse>
+          @type ltsv
+        </parse>
       ]
     }
     assert_nothing_raised {
       create_driver %[
-        format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
         key_name message
         suppress_parse_error_log true
+        <parse>
+          @type regexp
+          expression /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+        </parse>
       ]
     }
     assert_nothing_raised {
       create_driver %[
-        format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
         key_name message
         suppress_parse_error_log false
+        <parse>
+          @type regexp
+          expression /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+        </parse>
       ]
     }
     d = create_driver %[
       key_name foo
-      format /(?<x>.)/
+      <parse>
+        @type regexp
+        expression /(?<x>.)/
+      </parse>
     ]
     assert_false d.instance.reserve_data
   end
@@ -172,9 +181,11 @@ def test_filter
     assert_equal '20120402182100', second[1]['t']
 
     d3 = create_driver(%[
-      tag parsed
-      key_name      data
-      format        /^(?<x>[0-9])(?<y>[0-9]) (?<t>.+)$/
+      key_name data
+      <parse>
+        @type regexp
+        expression /^(?<x>[0-9])(?<y>[0-9]) (?<t>.+)$/
+      </parse>
     ])
     time = Time.parse("2012-04-02 18:20:59").to_i
     d3.run do
@@ -186,10 +197,12 @@ def test_filter
     assert_equal 2, filtered.length
 
     d3x = create_driver(%[
-      tag parsed
-      key_name      data
-      format        /^(?<x>\\d)(?<y>\\d) (?<t>.+)$/
-      reserve_data  yes
+      key_name data
+      reserve_data yes
+      <parse>
+        @type regexp
+        expression /^(?<x>\\d)(?<y>\\d) (?<t>.+)$/
+      </parse>
     ])
     time = Time.parse("2012-04-02 18:20:59").to_i
     d3x.run do
@@ -201,9 +214,10 @@ def test_filter
     assert_equal 3, filtered.length
 
     d4 = create_driver(%[
-      tag parsed
-      key_name      data
-      format        json
+      key_name data
+      <parse>
+        @type json
+      </parse>
     ])
     time = Time.parse("2012-04-02 18:20:59").to_i
     d4.run do
@@ -214,10 +228,11 @@ def test_filter
     assert_equal 1, filtered.length
 
     d4x = create_driver(%[
-      tag parsed
-      key_name      data
-      format        json
-      reserve_data  yes
+      key_name data
+      reserve_data yes
+      <parse>
+        @type json
+      </parse>
     ])
     time = @default_time.to_i
     d4x.run do
@@ -241,8 +256,17 @@ def test_filter
   end
 
   CONFIG_LTSV =  %[
-    format ltsv
     key_name data
+    <parse>
+      @type ltsv
+    </parse>
+  ]
+  CONFIG_LTSV_WITH_TYPES =  %[
+    key_name data
+    <parse>
+      @type ltsv
+      types i:integer,s:string,f:float,b:bool
+    </parse>
   ]
   def test_filter_ltsv
     d = create_driver(CONFIG_LTSV)
@@ -288,7 +312,8 @@ def test_filter_ltsv
     assert_equal 'second2', second[1]['yyy']
 
     # convert types
-    d = create_driver(CONFIG_LTSV + %[types i:integer,s:string,f:float,b:bool])
+    #d = create_driver(CONFIG_LTSV + %[
+    d = create_driver(CONFIG_LTSV_WITH_TYPES)
     time = @default_time.to_i
     d.run do
       d.feed(@tag, time, {'data' => "i:1\ts:2\tf:3\tb:true\tx:123"})
@@ -306,9 +331,11 @@ def test_filter_ltsv
   end
 
   CONFIG_TSV =  %[
-    format tsv
     key_name data
-    keys key1,key2,key3
+    <parse>
+      @type tsv
+      keys key1,key2,key3
+    </parse>
   ]
   def test_filter_tsv
     d = create_driver(CONFIG_TSV)
@@ -328,12 +355,21 @@ def test_filter_tsv
   end
 
   CONFIG_CSV =  %[
-    format csv
     key_name data
+    <parse>
+      @type csv
+      keys key1,key2,key3
+    </parse>
+  ]
+  CONFIG_CSV_COMPAT =  %[
+    key_name data
+    format csv
     keys key1,key2,key3
   ]
-  def test_filter_csv
-    d = create_driver(CONFIG_CSV)
+  data(new_conf: CONFIG_CSV,
+       compat_conf: CONFIG_CSV_COMPAT)
+  def test_filter_csv(conf)
+    d = create_driver(conf)
     time = @default_time.to_i
     d.run do
       d.feed(@tag, time, {'data' => 'value1,"value2","value""ThreeYes!"', 'xxx' => 'x', 'yyy' => 'y'})
@@ -350,21 +386,27 @@ def test_filter_csv
   end
 
   CONFIG_HASH_VALUE_FIELD = %[
-    format       json
-    key_name     data
+    key_name data
     hash_value_field parsed
+    <parse>
+      @type json
+    </parse>
   ]
   CONFIG_HASH_VALUE_FIELD_RESERVE_DATA = %[
-    format       json
-    key_name     data
+    key_name data
     reserve_data yes
     hash_value_field parsed
+    <parse>
+      @type json
+    </parse>
   ]
   CONFIG_HASH_VALUE_FIELD_WITH_INJECT_KEY_PREFIX = %[
-    format       json
-    key_name     data
+    key_name data
     hash_value_field parsed
     inject_key_prefix data.
+    <parse>
+      @type json
+    </parse>
   ]
   def test_filter_inject_hash_value_field
     original = {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'}
@@ -420,13 +462,23 @@ def test_filter_inject_hash_value_field
 
   CONFIG_DONT_PARSE_TIME = %[
     key_name data
+    reserve_time true
+    <parse>
+      @type json
+      keep_time_key true
+    </parse>
+  ]
+  CONFIG_DONT_PARSE_TIME_COMPAT = %[
+    key_name data
+    reserve_time true
     format json
     keep_time_key true
-    reserve_time true
   ]
-  def test_time_should_be_reserved
+  data(new_conf: CONFIG_DONT_PARSE_TIME,
+       compat_conf: CONFIG_DONT_PARSE_TIME_COMPAT)
+  def test_time_should_be_reserved(conf)
     t = Time.now.to_i
-    d = create_driver(CONFIG_DONT_PARSE_TIME)
+    d = create_driver(conf)
     d.run do
       d.feed(@tag, t, {'data' => '{"time":1383190430, "f1":"v1"}'})
       d.feed(@tag, t, {'data' => '{"time":"1383190430", "f1":"v1"}'})
@@ -449,9 +501,10 @@ def test_time_should_be_reserved
   end
 
   CONFIG_INVALID_TIME_VALUE = %[
-    remove_prefix test
     key_name data
-    format json
+    <parse>
+      @type json
+    </parse>
   ] # 'time' is implicit @time_key
   def test_filter_invalid_time_data
     # should not raise errors
@@ -474,9 +527,11 @@ def test_filter_invalid_time_data
   # REGEXP = /^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$/
 
   CONFIG_NOT_REPLACE = %[
-    remove_prefix test
-    key_name      data
-    format        /^(?<message>.*)$/
+    key_name data
+    <parse>
+      @type regexp
+      expression /^(?<message>.*)$/
+    </parse>
   ]
   CONFIG_INVALID_BYTE = CONFIG_NOT_REPLACE + %[
     replace_invalid_sequence true
@@ -529,10 +584,11 @@ def test_filter_invalid_byte
   end
 
   CONFIG_NOT_IGNORE = %[
-    remove_prefix    test
-    key_name         data
-    format           json
+    key_name data
     hash_value_field parsed
+    <parse>
+      @type json
+    </parse>
   ]
   CONFIG_IGNORE = CONFIG_NOT_IGNORE + %[
     ignore_key_not_exist true
@@ -587,9 +643,11 @@ def test_parser_error_warning
   end
 
   CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG = %[
-    tag hogelog
-    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
     key_name message
+    <parse>
+      @type regexp
+      expression /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+    </parse>
   ]
 
   class DefaultSuppressParseErrorLogTest < self
@@ -615,10 +673,12 @@ def test_nothing_raised
   end
 
   CONFIG_DISABELED_SUPPRESS_PARSE_ERROR_LOG = %[
-    tag hogelog
-    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
     key_name message
     suppress_parse_error_log false
+    <parse>
+      @type regexp
+      expression /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+    </parse>
   ]
 
   class DisabledSuppressParseErrorLogTest < self
@@ -644,10 +704,12 @@ def test_nothing_raised
   end
 
   CONFIG_ENABELED_SUPPRESS_PARSE_ERROR_LOG = %[
-    tag hogelog
-    format /^col1=(?<col1>.+) col2=(?<col2>.+)$/
     key_name message
     suppress_parse_error_log true
+    <parse>
+      @type regexp
+      expression /^col1=(?<col1>.+) col2=(?<col2>.+)$/
+    </parse>
   ]
 
   class EnabledSuppressParseErrorLogTest < self

From 2c82887aed6e1b52699ca9dde6b2fe1a0b662395 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 10 Nov 2016 09:06:08 +0900
Subject: [PATCH 10/19] Use String#scrub instead of own routine

---
 lib/fluent/plugin/filter_parser.rb | 12 +++---------
 1 file changed, 3 insertions(+), 9 deletions(-)

diff --git a/lib/fluent/plugin/filter_parser.rb b/lib/fluent/plugin/filter_parser.rb
index 7897783e25..5d23f54c92 100644
--- a/lib/fluent/plugin/filter_parser.rb
+++ b/lib/fluent/plugin/filter_parser.rb
@@ -46,7 +46,8 @@ def configure(conf)
       @parser = parser_create
     end
 
-    FAILED_RESULT = [nil, nil].freeze # reduce allocation cost 
+    FAILED_RESULT = [nil, nil].freeze # reduce allocation cost
+    REPLACE_CHAR = '?'.freeze
 
     def filter_with_time(tag, time, record)
       raw_value = record[@key_name]
@@ -87,7 +88,7 @@ def filter_with_time(tag, time, record)
           unless e.message.index("invalid byte sequence in") == 0
             raise
           end
-          replaced_string = replace_invalid_byte(raw_value)
+          replaced_string = raw_value.scrub(REPLACE_CHAR)
           @parser.parse(replaced_string) do |t, values|
             if values
               t ||= time
@@ -125,12 +126,5 @@ def handle_parsed(tag, record, t, values)
       end
       r
     end
-
-    def replace_invalid_byte(string)
-      replace_options = {invalid: :replace, undef: :replace, replace: '?'}
-      original_encoding = string.encoding
-      temporal_encoding = (original_encoding == Encoding::UTF_8 ? Encoding::UTF_16BE : Encoding::UTF_8)
-      string.encode(temporal_encoding, original_encoding, replace_options).encode(original_encoding)
-    end
   end
 end

From 4aef481e0abcaabd340dbab21e6ffa5e5fe7e9ff Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 10 Nov 2016 09:06:39 +0900
Subject: [PATCH 11/19] Remove useless config_section definition

---
 lib/fluent/plugin/filter_parser.rb | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/lib/fluent/plugin/filter_parser.rb b/lib/fluent/plugin/filter_parser.rb
index 5d23f54c92..7961bbf8a4 100644
--- a/lib/fluent/plugin/filter_parser.rb
+++ b/lib/fluent/plugin/filter_parser.rb
@@ -33,9 +33,6 @@ class ParserFilter < Filter
     config_param :replace_invalid_sequence, :bool, default: false
     config_param :hash_value_field, :string, default: nil
 
-    config_section :parse do
-    end
-
     attr_reader :parser
 
     def configure(conf)

From 33cac26885b0db1fc4989e15be99e551990c241f Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 10 Nov 2016 09:15:20 +0900
Subject: [PATCH 12/19] Improve invalid string handling to reduce duplicated
 code

---
 lib/fluent/plugin/filter_parser.rb | 29 +++++------------------------
 1 file changed, 5 insertions(+), 24 deletions(-)

diff --git a/lib/fluent/plugin/filter_parser.rb b/lib/fluent/plugin/filter_parser.rb
index 7961bbf8a4..cc5d785b4a 100644
--- a/lib/fluent/plugin/filter_parser.rb
+++ b/lib/fluent/plugin/filter_parser.rb
@@ -81,30 +81,11 @@ def filter_with_time(tag, time, record)
         router.emit_error_event(tag, time, record, e)
         return FAILED_RESULT
       rescue ArgumentError => e
-        if @replace_invalid_sequence
-          unless e.message.index("invalid byte sequence in") == 0
-            raise
-          end
-          replaced_string = raw_value.scrub(REPLACE_CHAR)
-          @parser.parse(replaced_string) do |t, values|
-            if values
-              t ||= time
-              r = handle_parsed(tag, record, t, values)
-              return t, r
-            else
-              router.emit_error_event(tag, time, record, Fluent::Plugin::Parser::ParserError.new("pattern not match with data '#{raw_value}'"))
-              if @reserve_data
-                t = time
-                r = handle_parsed(tag, record, time, {})
-                return t, r
-              else
-                return FAILED_RESULT
-              end
-            end
-          end
-        else
-          raise
-        end
+        raise unless @replace_invalid_sequence
+        raise unless e.message.index("invalid byte sequence in") == 0
+
+        raw_value = raw_value.scrub(REPLACE_CHAR)
+        retry
       rescue => e
         router.emit_error_event(tag, time, record, Fluent::Plugin::Parser::ParserError.new("parse failed #{e.message}"))
         return FAILED_RESULT

From d08a34d167f6cb21fd976268ef3da37ce44fd5fa Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 10 Nov 2016 09:19:43 +0900
Subject: [PATCH 13/19] Recent update removes ValuesParser from v0.14 parsers

---
 lib/fluent/plugin_helper/compat_parameters.rb | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/fluent/plugin_helper/compat_parameters.rb b/lib/fluent/plugin_helper/compat_parameters.rb
index 198b62bf9e..879bff8f8e 100644
--- a/lib/fluent/plugin_helper/compat_parameters.rb
+++ b/lib/fluent/plugin_helper/compat_parameters.rb
@@ -57,8 +57,8 @@ module CompatParameters
         "time_format" => "time_format",
         "delimiter"   => "delimiter",
         "keep_time_key" => "keep_time_key",
-        "null_empty_string" => "null_empty_string", # ValuesParser
-        "null_value_pattern" => "null_value_pattern", # ValuesParser
+        "null_empty_string" => "null_empty_string",
+        "null_value_pattern" => "null_value_pattern",
         "json_parser"      => "json_parser", # JSONParser
         "label_delimiter"  => "label_delimiter", # LabeledTSVParser
         "format_firstline" => "format_firstline", # MultilineParser

From 5cb93a4244864e029547e26514d9a0de28dd8af7 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 10 Nov 2016 09:24:34 +0900
Subject: [PATCH 14/19] Use default_tag for driver run

---
 test/plugin/test_filter_parser.rb | 78 +++++++++++++++----------------
 1 file changed, 39 insertions(+), 39 deletions(-)

diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index f93a68a552..aa40a349f8 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -118,12 +118,12 @@ def test_configure
   def test_filter
     d1 = create_driver(CONFIG)
     time = Time.parse("2012-01-02 13:14:15").to_i
-    d1.run do
-      d1.feed(@tag, time, {'message' => '12 20120402182059'})
-      d1.feed(@tag, time, {'message' => '34 20120402182100'})
-      d1.feed(@tag, time, {'message' => '56 20120402182100'})
-      d1.feed(@tag, time, {'message' => '78 20120402182101'})
-      d1.feed(@tag, time, {'message' => '90 20120402182100'})
+    d1.run(default_tag: @tag) do
+      d1.feed(time, {'message' => '12 20120402182059'})
+      d1.feed(time, {'message' => '34 20120402182100'})
+      d1.feed(time, {'message' => '56 20120402182100'})
+      d1.feed(time, {'message' => '78 20120402182101'})
+      d1.feed(time, {'message' => '90 20120402182100'})
     end
     filtered = d1.filtered
     assert_equal 5, filtered.length
@@ -159,9 +159,9 @@ def test_filter
       format   /^(?<x>.)(?<y>.) (?<t>.+)$/
     ])
     time = Fluent::EventTime.from_time(@default_time) # EventTime emit test
-    d2.run do
-      d2.feed(@tag, time, {'data' => '12 20120402182059'})
-      d2.feed(@tag, time, {'data' => '34 20120402182100'})
+    d2.run(default_tag: @tag) do
+      d2.feed(time, {'data' => '12 20120402182059'})
+      d2.feed(time, {'data' => '34 20120402182100'})
     end
     filtered = d2.filtered
     assert_equal 2, filtered.length
@@ -188,10 +188,10 @@ def test_filter
       </parse>
     ])
     time = Time.parse("2012-04-02 18:20:59").to_i
-    d3.run do
-      d3.feed(@tag, time, {'data' => '12 20120402182059'})
-      d3.feed(@tag, time, {'data' => '34 20120402182100'})
-      d3.feed(@tag, time, {'data' => 'xy 20120402182101'})
+    d3.run(default_tag: @tag) do
+      d3.feed(time, {'data' => '12 20120402182059'})
+      d3.feed(time, {'data' => '34 20120402182100'})
+      d3.feed(time, {'data' => 'xy 20120402182101'})
     end
     filtered = d3.filtered
     assert_equal 2, filtered.length
@@ -205,10 +205,10 @@ def test_filter
       </parse>
     ])
     time = Time.parse("2012-04-02 18:20:59").to_i
-    d3x.run do
-      d3x.feed(@tag, time, {'data' => '12 20120402182059'})
-      d3x.feed(@tag, time, {'data' => '34 20120402182100'})
-      d3x.feed(@tag, time, {'data' => 'xy 20120402182101'})
+    d3x.run(default_tag: @tag) do
+      d3x.feed(time, {'data' => '12 20120402182059'})
+      d3x.feed(time, {'data' => '34 20120402182100'})
+      d3x.feed(time, {'data' => 'xy 20120402182101'})
     end
     filtered = d3x.filtered
     assert_equal 3, filtered.length
@@ -220,9 +220,9 @@ def test_filter
       </parse>
     ])
     time = Time.parse("2012-04-02 18:20:59").to_i
-    d4.run do
-      d4.feed(@tag, time, {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'})
-      d4.feed(@tag, time, {'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'})
+    d4.run(default_tag: @tag) do
+      d4.feed(time, {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'})
+      d4.feed(time, {'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'})
     end
     filtered = d4.filtered
     assert_equal 1, filtered.length
@@ -235,9 +235,9 @@ def test_filter
       </parse>
     ])
     time = @default_time.to_i
-    d4x.run do
-      d4x.feed(@tag, time, {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'})
-      d4x.feed(@tag, time, {'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'})
+    d4x.run(default_tag: @tag) do
+      d4x.feed(time, {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'})
+      d4x.feed(time, {'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'})
     end
     filtered = d4x.filtered
     assert_equal 2, filtered.length
@@ -271,9 +271,9 @@ def test_filter
   def test_filter_ltsv
     d = create_driver(CONFIG_LTSV)
     time = @default_time.to_i
-    d.run do
-      d.feed(@tag, time, {'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'})
-      d.feed(@tag, time, {'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'})
+    d.run(default_tag: @tag) do
+      d.feed(time, {'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'})
+      d.feed(time, {'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'})
     end
     filtered = d.filtered
     assert_equal 2, filtered.length
@@ -292,9 +292,9 @@ def test_filter_ltsv
 
     d = create_driver(CONFIG_LTSV + %[reserve_data yes])
     time = @default_time.to_i
-    d.run do
-      d.feed(@tag, time, {'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'})
-      d.feed(@tag, time, {'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'})
+    d.run(default_tag: @tag) do
+      d.feed(time, {'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'})
+      d.feed(time, {'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'})
     end
     filtered = d.filtered
     assert_equal 2, filtered.length
@@ -479,10 +479,10 @@ def test_filter_inject_hash_value_field
   def test_time_should_be_reserved(conf)
     t = Time.now.to_i
     d = create_driver(conf)
-    d.run do
-      d.feed(@tag, t, {'data' => '{"time":1383190430, "f1":"v1"}'})
-      d.feed(@tag, t, {'data' => '{"time":"1383190430", "f1":"v1"}'})
-      d.feed(@tag, t, {'data' => '{"time":"2013-10-31 12:34:03 +0900", "f1":"v1"}'})
+    d.run(default_tag: @tag) do
+      d.feed(t, {'data' => '{"time":1383190430, "f1":"v1"}'})
+      d.feed(t, {'data' => '{"time":"1383190430", "f1":"v1"}'})
+      d.feed(t, {'data' => '{"time":"2013-10-31 12:34:03 +0900", "f1":"v1"}'})
     end
     filtered = d.filtered
     assert_equal 3, filtered.length
@@ -511,9 +511,9 @@ def test_filter_invalid_time_data
     time = Time.now.to_i
     d = create_driver(CONFIG_INVALID_TIME_VALUE)
     assert_nothing_raised {
-      d.run do
-        d.feed(@tag, time, {'data' => '{"time":[], "f1":"v1"}'})
-        d.feed(@tag, time, {'data' => '{"time":"thisisnottime", "f1":"v1"}'})
+      d.run(default_tag: @tag) do
+        d.feed(time, {'data' => '{"time":[], "f1":"v1"}'})
+        d.feed(time, {'data' => '{"time":"thisisnottime", "f1":"v1"}'})
       end
     }
     filtered = d.filtered
@@ -721,9 +721,9 @@ class EnabledSuppressParseErrorLogTest < self
     def test_nothing_raised
       flexmock(@d.instance.router).should_receive(:emit_error_event).
         with(String, Integer, Hash, ParserError).once
-      @d.run do
-        @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
-        @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
+      @d.run(default_tag: @tag) do
+        @d.feed(Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
+        @d.feed(Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
       end
     end
   end

From cb3dee61a2fcd4404d227fdd30cd7ea8d7ab0c0a Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 10 Nov 2016 09:29:48 +0900
Subject: [PATCH 15/19] Remove deleted paramters tests

---
 test/plugin/test_filter_parser.rb | 83 +------------------------------
 1 file changed, 1 insertion(+), 82 deletions(-)

diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index aa40a349f8..b5f4056661 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -80,17 +80,6 @@ def test_configure
     assert_nothing_raised {
       create_driver %[
         key_name message
-        suppress_parse_error_log true
-        <parse>
-          @type regexp
-          expression /^col1=(?<col1>.+) col2=(?<col2>.+)$/
-        </parse>
-      ]
-    }
-    assert_nothing_raised {
-      create_driver %[
-        key_name message
-        suppress_parse_error_log false
         <parse>
           @type regexp
           expression /^col1=(?<col1>.+) col2=(?<col2>.+)$/
@@ -590,10 +579,7 @@ def test_filter_invalid_byte
       @type json
     </parse>
   ]
-  CONFIG_IGNORE = CONFIG_NOT_IGNORE + %[
-    ignore_key_not_exist true
-  ]
-  CONFIG_PASS_SAME_RECORD = CONFIG_IGNORE + %[
+  CONFIG_PASS_SAME_RECORD = CONFIG_NOT_IGNORE + %[
     reserve_data true
   ]
   def test_filter_key_not_exist
@@ -605,16 +591,6 @@ def test_filter_key_not_exist
         d.feed(@tag, Fluent::EventTime.now.to_i, {'foo' => 'bar'})
       end
     }
-    #assert_match(/data does not exist/, d.instance.log.logs.first)
-    d.instance_shutdown
-
-    d = create_driver(CONFIG_IGNORE)
-    assert_nothing_raised {
-      d.run(shutdown: false) do
-        d.feed(@tag, Fluent::EventTime.now.to_i, {'foo' => 'bar'})
-      end
-    }
-    assert_not_match(/data does not exist/, d.instance.log.logs.first)
     d.instance_shutdown
 
     d = create_driver(CONFIG_PASS_SAME_RECORD)
@@ -652,7 +628,6 @@ def test_parser_error_warning
 
   class DefaultSuppressParseErrorLogTest < self
     setup do
-      # default(disabled) 'suppress_parse_error_log' is not specify
       @d = create_driver(CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG)
     end
 
@@ -671,60 +646,4 @@ def test_nothing_raised
       end
     end
   end
-
-  CONFIG_DISABELED_SUPPRESS_PARSE_ERROR_LOG = %[
-    key_name message
-    suppress_parse_error_log false
-    <parse>
-      @type regexp
-      expression /^col1=(?<col1>.+) col2=(?<col2>.+)$/
-    </parse>
-  ]
-
-  class DisabledSuppressParseErrorLogTest < self
-    setup do
-      # disabled 'suppress_parse_error_log'
-      @d = create_driver(CONFIG_DISABELED_SUPPRESS_PARSE_ERROR_LOG)
-    end
-
-    def test_raise_exception
-      flexmock(@d.instance.router).should_receive(:emit_error_event).
-        with(String, Integer, Hash, ParserError.new("pattern not match with data '#{INVALID_MESSAGE}'")).once
-      @d.run do
-        @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
-      end
-    end
-
-    def test_nothing_raised
-      flexmock(@d.instance.router).should_receive(:emit_error_event).never
-      @d.run do
-        @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
-      end
-    end
-  end
-
-  CONFIG_ENABELED_SUPPRESS_PARSE_ERROR_LOG = %[
-    key_name message
-    suppress_parse_error_log true
-    <parse>
-      @type regexp
-      expression /^col1=(?<col1>.+) col2=(?<col2>.+)$/
-    </parse>
-  ]
-
-  class EnabledSuppressParseErrorLogTest < self
-    setup do
-      # enabled 'suppress_parse_error_log'
-      @d = create_driver(CONFIG_ENABELED_SUPPRESS_PARSE_ERROR_LOG)
-    end
-
-    def test_nothing_raised
-      flexmock(@d.instance.router).should_receive(:emit_error_event).
-        with(String, Integer, Hash, ParserError).once
-      @d.run(default_tag: @tag) do
-        @d.feed(Fluent::EventTime.now.to_i, {'message' => INVALID_MESSAGE})
-        @d.feed(Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})
-      end
-    end
-  end
 end

From 163dc5c5157970bf9d0f851b816a46eebd0a0fad Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 10 Nov 2016 09:33:12 +0900
Subject: [PATCH 16/19] Rename suppress warning test to unmached pattern

---
 test/plugin/test_filter_parser.rb | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index b5f4056661..88dc4ea336 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -605,11 +605,11 @@ def test_filter_key_not_exist
     assert_equal 'bar', filtered[0][1]['foo']
   end
 
-  # suppress_parse_error_log test
+  # emit_error_event test
   INVALID_MESSAGE = 'foo bar'
   VALID_MESSAGE   = 'col1=foo col2=bar'
 
-  def test_parser_error_warning
+  def test_call_emit_error_event_when_parser_error
     d = create_driver(CONFIG_INVALID_TIME_VALUE)
     flexmock(d.instance.router).should_receive(:emit_error_event).
       with(String, Integer, Hash, ParserError).once
@@ -618,7 +618,7 @@ def test_parser_error_warning
     end
   end
 
-  CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG = %[
+  CONFIG_UNMATCHED_PATTERN_LOG = %[
     key_name message
     <parse>
       @type regexp
@@ -626,12 +626,12 @@ def test_parser_error_warning
     </parse>
   ]
 
-  class DefaultSuppressParseErrorLogTest < self
+  class UnmatchedPatternLogTest < self
     setup do
-      @d = create_driver(CONFIG_DEFAULT_SUPPRESS_PARSE_ERROR_LOG)
+      @d = create_driver(CONFIG_UNMATCHED_PATTERN_LOG)
     end
 
-    def test_raise_exception
+    def test_call_emit_error_event_when_pattern_is_mismached
       flexmock(@d.instance.router).should_receive(:emit_error_event).
         with(String, Integer, Hash, ParserError.new("pattern not match with data '#{INVALID_MESSAGE}'")).once
       @d.run do
@@ -639,7 +639,7 @@ def test_raise_exception
       end
     end
 
-    def test_nothing_raised
+    def test_not_call_emit_error_event_when_pattern_is_mached
       flexmock(@d.instance.router).should_receive(:emit_error_event).never
       @d.run do
         @d.feed(@tag, Fluent::EventTime.now.to_i, {'message' => VALID_MESSAGE})

From 71d700c8a56251b7891ac2f7274e4dcde62ede04 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 10 Nov 2016 10:00:37 +0900
Subject: [PATCH 17/19] Improve tests

- Use proper assertion
- Add data to EventTime and Integer time tests
- Extract some assertions for better test cases
---
 test/plugin/test_filter_parser.rb | 108 +++++++++++++++++-------------
 1 file changed, 63 insertions(+), 45 deletions(-)

diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index 88dc4ea336..8467d35ebf 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -19,6 +19,14 @@ def teardown
     Timecop.return
   end
 
+  def assert_equal_parsed_time(expected, actual)
+    if expected.is_a?(Integer)
+      assert_equal(expected, actual.to_i)
+    else
+      assert_equal_event_time(expected, actual)
+    end
+  end
+
   ParserError = Fluent::Plugin::Parser::ParserError
   CONFIG = %[
     key_name     message
@@ -106,7 +114,7 @@ def test_configure
   # ]
   def test_filter
     d1 = create_driver(CONFIG)
-    time = Time.parse("2012-01-02 13:14:15").to_i
+    time = event_time("2012-01-02 13:14:15")
     d1.run(default_tag: @tag) do
       d1.feed(time, {'message' => '12 20120402182059'})
       d1.feed(time, {'message' => '34 20120402182100'})
@@ -118,28 +126,28 @@ def test_filter
     assert_equal 5, filtered.length
 
     first = filtered[0]
-    assert_equal Time.parse("2012-04-02 18:20:59").to_i, first[0]
+    assert_equal_event_time event_time("2012-04-02 18:20:59"), first[0]
     assert_equal '1', first[1]['x']
     assert_equal '2', first[1]['y']
     assert_equal '12 20120402182059', first[1]['message']
 
     second = filtered[1]
-    assert_equal Time.parse("2012-04-02 18:21:00").to_i, second[0]
+    assert_equal_event_time event_time("2012-04-02 18:21:00"), second[0]
     assert_equal '3', second[1]['x']
     assert_equal '4', second[1]['y']
 
     third = filtered[2]
-    assert_equal Time.parse("2012-04-02 18:21:00").to_i, third[0]
+    assert_equal_event_time event_time("2012-04-02 18:21:00"), third[0]
     assert_equal '5', third[1]['x']
     assert_equal '6', third[1]['y']
 
     fourth = filtered[3]
-    assert_equal Time.parse("2012-04-02 18:21:01").to_i, fourth[0]
+    assert_equal_event_time event_time("2012-04-02 18:21:01"), fourth[0]
     assert_equal '7', fourth[1]['x']
     assert_equal '8', fourth[1]['y']
 
     fifth = filtered[4]
-    assert_equal Time.parse("2012-04-02 18:21:00").to_i, fifth[0]
+    assert_equal_event_time event_time("2012-04-02 18:21:00"), fifth[0]
     assert_equal '9', fifth[1]['x']
     assert_equal '0', fifth[1]['y']
 
@@ -156,14 +164,14 @@ def test_filter
     assert_equal 2, filtered.length
 
     first = filtered[0]
-    assert_equal time, first[0]
+    assert_equal_event_time time, first[0]
     assert_nil first[1]['data']
     assert_equal '1', first[1]['x']
     assert_equal '2', first[1]['y']
     assert_equal '20120402182059', first[1]['t']
 
     second = filtered[1]
-    assert_equal time, second[0]
+    assert_equal_event_time time, second[0]
     assert_nil second[1]['data']
     assert_equal '3', second[1]['x']
     assert_equal '4', second[1]['y']
@@ -185,23 +193,6 @@ def test_filter
     filtered = d3.filtered
     assert_equal 2, filtered.length
 
-    d3x = create_driver(%[
-      key_name data
-      reserve_data yes
-      <parse>
-        @type regexp
-        expression /^(?<x>\\d)(?<y>\\d) (?<t>.+)$/
-      </parse>
-    ])
-    time = Time.parse("2012-04-02 18:20:59").to_i
-    d3x.run(default_tag: @tag) do
-      d3x.feed(time, {'data' => '12 20120402182059'})
-      d3x.feed(time, {'data' => '34 20120402182100'})
-      d3x.feed(time, {'data' => 'xy 20120402182101'})
-    end
-    filtered = d3x.filtered
-    assert_equal 3, filtered.length
-
     d4 = create_driver(%[
       key_name data
       <parse>
@@ -216,7 +207,27 @@ def test_filter
     filtered = d4.filtered
     assert_equal 1, filtered.length
 
-    d4x = create_driver(%[
+  end
+
+  def test_filter_with_reserved_data
+    d1 = create_driver(%[
+      key_name data
+      reserve_data yes
+      <parse>
+        @type regexp
+        expression /^(?<x>\\d)(?<y>\\d) (?<t>.+)$/
+      </parse>
+    ])
+    time = Time.parse("2012-04-02 18:20:59").to_i
+    d1.run(default_tag: @tag) do
+      d1.feed(time, {'data' => '12 20120402182059'})
+      d1.feed(time, {'data' => '34 20120402182100'})
+      d1.feed(time, {'data' => 'xy 20120402182101'})
+    end
+    filtered = d1.filtered
+    assert_equal 3, filtered.length
+
+    d2 = create_driver(%[
       key_name data
       reserve_data yes
       <parse>
@@ -224,11 +235,11 @@ def test_filter
       </parse>
     ])
     time = @default_time.to_i
-    d4x.run(default_tag: @tag) do
-      d4x.feed(time, {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'})
-      d4x.feed(time, {'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'})
+    d2.run(default_tag: @tag) do
+      d2.feed(time, {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'})
+      d2.feed(time, {'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'})
     end
-    filtered = d4x.filtered
+    filtered = d2.filtered
     assert_equal 2, filtered.length
 
     first = filtered[0]
@@ -244,6 +255,7 @@ def test_filter
     assert_equal 'y', second[1]['yyy']
   end
 
+
   CONFIG_LTSV =  %[
     key_name data
     <parse>
@@ -257,9 +269,11 @@ def test_filter
       types i:integer,s:string,f:float,b:bool
     </parse>
   ]
-  def test_filter_ltsv
+  data(:event_time => lambda { |time| Fluent::EventTime.from_time(time) },
+       :int_time => lambda { |time| time.to_i })
+  def test_filter_ltsv(time_parse)
     d = create_driver(CONFIG_LTSV)
-    time = @default_time.to_i
+    time = time_parse.call(@default_time)
     d.run(default_tag: @tag) do
       d.feed(time, {'data' => "xxx:first\tyyy:second", 'xxx' => 'x', 'yyy' => 'y'})
       d.feed(time, {'data' => "xxx:first\tyyy:second2", 'xxx' => 'x', 'yyy' => 'y'})
@@ -268,13 +282,13 @@ def test_filter_ltsv
     assert_equal 2, filtered.length
 
     first = filtered[0]
-    assert_equal time, first[0]
+    assert_equal_parsed_time time, first[0]
     assert_nil first[1]['data']
     assert_equal 'first', first[1]['xxx']
     assert_equal 'second', first[1]['yyy']
 
     second = filtered[1]
-    assert_equal time, second[0]
+    assert_equal_parsed_time time, second[0]
     assert_nil first[1]['data']
     assert_equal 'first', second[1]['xxx']
     assert_equal 'second2', second[1]['yyy']
@@ -289,13 +303,13 @@ def test_filter_ltsv
     assert_equal 2, filtered.length
 
     first = filtered[0]
-    assert_equal time, first[0]
+    assert_equal_parsed_time time, first[0]
     assert_equal "xxx:first\tyyy:second", first[1]['data']
     assert_equal 'first', first[1]['xxx']
     assert_equal 'second', first[1]['yyy']
 
     second = filtered[1]
-    assert_equal time, second[0]
+    assert_equal_parsed_time time, second[0]
     assert_equal "xxx:first\tyyy:second", first[1]['data']
     assert_equal 'first', second[1]['xxx']
     assert_equal 'second2', second[1]['yyy']
@@ -311,7 +325,7 @@ def test_filter_ltsv
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal time, first[0]
+    assert_equal_parsed_time time, first[0]
     assert_equal 1, first[1]['i']
     assert_equal '2', first[1]['s']
     assert_equal 3.0, first[1]['f']
@@ -326,9 +340,11 @@ def test_filter_ltsv
       keys key1,key2,key3
     </parse>
   ]
-  def test_filter_tsv
+  data(:event_time => lambda { |time| Fluent::EventTime.from_time(time) },
+       :int_time => lambda { |time| time.to_i })
+  def test_filter_tsv(time_parse)
     d = create_driver(CONFIG_TSV)
-    time = @default_time.to_i
+    time = time_parse.call(@default_time)
     d.run do
       d.feed(@tag, time, {'data' => "value1\tvalue2\tvalueThree", 'xxx' => 'x', 'yyy' => 'y'})
     end
@@ -336,7 +352,7 @@ def test_filter_tsv
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal time, first[0]
+    assert_equal_parsed_time time, first[0]
     assert_nil first[1]['data']
     assert_equal 'value1', first[1]['key1']
     assert_equal 'value2', first[1]['key2']
@@ -397,11 +413,13 @@ def test_filter_csv(conf)
       @type json
     </parse>
   ]
-  def test_filter_inject_hash_value_field
+  data(:event_time => lambda { |time| Fluent::EventTime.from_time(time) },
+       :int_time => lambda { |time| time.to_i })
+  def test_filter_inject_hash_value_field(time_parse)
     original = {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'}
 
     d = create_driver(CONFIG_HASH_VALUE_FIELD)
-    time = @default_time.to_i
+    time = time_parse.call(@default_time)
     d.run do
       d.feed(@tag, time, original)
     end
@@ -409,7 +427,7 @@ def test_filter_inject_hash_value_field
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal time, first[0]
+    assert_equal_parsed_time time, first[0]
 
     record = first[1]
     assert_equal 1, record.keys.size
@@ -424,7 +442,7 @@ def test_filter_inject_hash_value_field
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal time, first[0]
+    assert_equal_parsed_time time, first[0]
 
     record = first[1]
     assert_equal 4, record.keys.size
@@ -442,7 +460,7 @@ def test_filter_inject_hash_value_field
     assert_equal 1, filtered.length
 
     first = filtered[0]
-    assert_equal time, first[0]
+    assert_equal_parsed_time time, first[0]
 
     record = first[1]
     assert_equal 1, record.keys.size

From 82953c52d7b57eea48b514e08140ad953d2c8c57 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Thu, 10 Nov 2016 10:09:06 +0900
Subject: [PATCH 18/19] Forget to update test to use event_time instead of to_i

---
 test/plugin/test_filter_parser.rb | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index 8467d35ebf..d04477e5e0 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -218,7 +218,7 @@ def test_filter_with_reserved_data
         expression /^(?<x>\\d)(?<y>\\d) (?<t>.+)$/
       </parse>
     ])
-    time = Time.parse("2012-04-02 18:20:59").to_i
+    time = event_time("2012-04-02 18:20:59")
     d1.run(default_tag: @tag) do
       d1.feed(time, {'data' => '12 20120402182059'})
       d1.feed(time, {'data' => '34 20120402182100'})
@@ -234,7 +234,7 @@ def test_filter_with_reserved_data
         @type json
       </parse>
     ])
-    time = @default_time.to_i
+    time = Fluent::EventTime.from_time(@default_time)
     d2.run(default_tag: @tag) do
       d2.feed(time, {'data' => '{"xxx":"first","yyy":"second"}', 'xxx' => 'x', 'yyy' => 'y'})
       d2.feed(time, {'data' => 'foobar', 'xxx' => 'x', 'yyy' => 'y'})
@@ -243,13 +243,13 @@ def test_filter_with_reserved_data
     assert_equal 2, filtered.length
 
     first = filtered[0]
-    assert_equal time, first[0]
+    assert_equal_event_time time, first[0]
     assert_equal '{"xxx":"first","yyy":"second"}', first[1]['data']
     assert_equal 'first', first[1]['xxx']
     assert_equal 'second', first[1]['yyy']
 
     second = filtered[1]
-    assert_equal time, second[0]
+    assert_equal_event_time time, second[0]
     assert_equal 'foobar', second[1]['data']
     assert_equal 'x', second[1]['xxx']
     assert_equal 'y', second[1]['yyy']

From f64bb7700b5c2fda3bc4f918b7ac0e812e4244c1 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Fri, 11 Nov 2016 02:05:08 +0900
Subject: [PATCH 19/19] Remove useless shutdown option from run

---
 test/plugin/test_filter_parser.rb | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/test/plugin/test_filter_parser.rb b/test/plugin/test_filter_parser.rb
index d04477e5e0..0e54989478 100644
--- a/test/plugin/test_filter_parser.rb
+++ b/test/plugin/test_filter_parser.rb
@@ -547,13 +547,12 @@ def test_filter_invalid_byte
     invalid_utf8 = "\xff".force_encoding('UTF-8')
 
     d = create_driver(CONFIG_NOT_REPLACE)
-    d.run(shutdown: false) do
+    d.run do
       d.feed(@tag, Fluent::EventTime.now.to_i, {'data' => invalid_utf8})
     end
     error_event = d.error_events.first
     assert_equal "test", error_event[0]
     assert_instance_of ArgumentError, error_event[3]
-    d.instance_shutdown      
 
     d = create_driver(CONFIG_INVALID_BYTE)
     assert_nothing_raised {
@@ -605,11 +604,10 @@ def test_filter_key_not_exist
     flexmock(d.instance.router).should_receive(:emit_error_event).
       with(String, Integer, Hash, ArgumentError.new("data does not exist")).once
     assert_nothing_raised {
-      d.run(shutdown: false) do
+      d.run do
         d.feed(@tag, Fluent::EventTime.now.to_i, {'foo' => 'bar'})
       end
     }
-    d.instance_shutdown
 
     d = create_driver(CONFIG_PASS_SAME_RECORD)
     assert_nothing_raised {