Merge pull request #1654 from fluent/filter-parser-with-record-accessor

filter_parser: use record_accessor

lib/fluent/plugin/filter_parser.rb

@@ -24,7 +24,7 @@ module Fluent::Plugin
   class ParserFilter < Filter
     Fluent::Plugin.register_filter('parser', self)
 
-    helpers :parser, :compat_parameters
+    helpers :parser, :record_accessor, :compat_parameters
 
     config_param :key_name, :string
     config_param :reserve_data, :bool, default: false
@@ -41,14 +41,15 @@ def configure(conf)
 
       super
 
+      @accessor = record_accessor_create(@key_name)
       @parser = parser_create
     end
 
     FAILED_RESULT = [nil, nil].freeze # reduce allocation cost
     REPLACE_CHAR = '?'.freeze
 
     def filter_with_time(tag, time, record)
-      raw_value = record[@key_name]
+      raw_value = @accessor.call(record)
       if raw_value.nil?
         if @emit_invalid_record_to_error
           router.emit_error_event(tag, time, record, ArgumentError.new("#{@key_name} does not exist"))

test/plugin/test_filter_parser.rb

@@ -390,6 +390,29 @@ def test_filter_csv(conf)
     assert_equal 'value"ThreeYes!', first[1]['key3']
   end
 
+  def test_filter_with_nested_record
+    d = create_driver(%[
+      key_name $.data.log
+      <parse>
+        @type csv
+        keys key1,key2,key3
+      </parse>
+    ])
+    time = @default_time.to_i
+    d.run do
+      d.feed(@tag, time, {'data' => {'log' => 'value1,"value2","value""ThreeYes!"'}, 'xxx' => 'x', 'yyy' => 'y'})
+    end
+    filtered = d.filtered
+    assert_equal 1, filtered.length
+
+    first = filtered[0]
+    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 = %[
     key_name data
     hash_value_field parsed