out_exec_filter: separated {tag_key,time_key,time_format} to {in_,out_}{tag_key,time_key,time_format}

Author: frsyuki

lib/fluent/plugin/out_exec_filter.rb

@@ -44,6 +44,9 @@ def initialize
   config_param :in_keys, :default => [] do |val|
     val.split(',')
   end
+  config_param :in_tag_key, :default => nil
+  config_param :in_time_key, :default => nil
+  config_param :in_time_format, :default => nil
 
   config_param :out_format, :default => :tsv do |val|
     f = SUPPORTED_FORMAT[val]
@@ -53,9 +56,11 @@ def initialize
   config_param :out_keys, :default => [] do |val|  # for tsv format
     val.split(',')
   end
+  config_param :out_tag_key, :default => nil
+  config_param :out_time_key, :default => nil
+  config_param :out_time_format, :default => nil
 
   config_param :tag, :string, :default => nil
-  config_param :tag_key, :string, :default => nil
 
   config_param :time_key, :string, :default => nil
   config_param :time_format, :string, :default => nil
@@ -66,6 +71,24 @@ def initialize
   config_set_default :flush_interval, 1
 
   def configure(conf)
+    if tag_key = conf['tag_key']
+      # TODO obsoleted?
+      @in_tag_key = tag_key
+      @out_tag_key = tag_key
+    end
+
+    if time_key = conf['time_key']
+      # TODO obsoleted?
+      @in_time_key = time_key
+      @out_time_key = time_key
+    end
+
+    if time_format = conf['time_format']
+      # TODO obsoleted?
+      @in_time_format = time_format
+      @out_time_format = time_format
+    end
+
     super
 
     if localtime = conf['localtime']
@@ -74,20 +97,29 @@ def configure(conf)
       @localtime = false
     end
 
-    if !@tag && !@tag_key
-      raise ConfigError, "'tag' or 'tag_key' option is required on exec_filter output"
+    if !@tag && !@out_tag_key
+      raise ConfigError, "'tag' or 'out_tag_key' option is required on exec_filter output"
     end
 
-    if @time_key
-      if @time_format
-        f = @time_format
+    if @in_time_key
+      if f = @in_time_format
         tf = TimeFormatter.new(f, @localtime)
         @time_format_proc = tf.method(:format)
-        @time_parse_proc = Proc.new {|str| Time.strptime(str, f).to_i }
       else
         @time_format_proc = Proc.new {|time| time.to_s }
+      end
+    elsif @in_time_format
+      $log.warn "in_time_format effects nothing when in_time_key is not specified: #{conf}"
+    end
+
+    if @out_time_key
+      if f = @out_time_format
+        @time_parse_proc = Proc.new {|str| Time.strptime(str, f).to_i }
+      else
         @time_parse_proc = Proc.new {|str| str.to_i }
       end
+    elsif @out_time_format
+      $log.warn "out_time_format effects nothing when out_time_key is not specified: #{conf}"
     end
 
     if @remove_prefix
@@ -164,11 +196,11 @@ def format_stream(tag, es)
     out = ''
 
     es.each {|time,record|
-      if @time_key
-        record[@time_key] = @time_format_proc.call(time)
+      if @in_time_key
+        record[@in_time_key] = @time_format_proc.call(time)
       end
-      if @tag_key
-        record[@tag_key] = tag
+      if @in_tag_key
+        record[@in_tag_key] = tag
       end
       @formatter.call(record, out)
     }
@@ -267,13 +299,13 @@ def call(record, out)
   end
 
   def on_message(record)
-    if val = record.delete(@time_key)
+    if val = record.delete(@out_time_key)
       time = @time_parse_proc.call(val)
     else
       time = Engine.now
     end
 
-    if val = record.delete(@tag_key)
+    if val = record.delete(@out_tag_key)
       tag = if @add_prefix
               @added_prefix_string + val
             else

test/plugin/out_exec_filter.rb

@@ -8,10 +8,11 @@ def setup
 
   CONFIG = %[
     command cat
-    in_keys time,tag,k1
-    out_keys time,tag,k2
+    in_keys time_in,tag,k1
+    out_keys time_out,tag,k2
     tag_key tag
-    time_key time
+    in_time_key time_in
+    out_time_key time_out
     time_format %Y-%m-%d %H:%M:%S
     localtime
     num_children 3
@@ -24,11 +25,14 @@ def create_driver(conf = CONFIG, tag = 'test')
   def test_configure
     d = create_driver
 
-    assert_equal ["time","tag","k1"], d.instance.in_keys
-    assert_equal ["time","tag","k2"], d.instance.out_keys
-    assert_equal "tag", d.instance.tag_key
-    assert_equal "time", d.instance.time_key
-    assert_equal "%Y-%m-%d %H:%M:%S", d.instance.time_format
+    assert_equal ["time_in","tag","k1"], d.instance.in_keys
+    assert_equal ["time_out","tag","k2"], d.instance.out_keys
+    assert_equal "tag", d.instance.out_tag_key
+    assert_equal "tag", d.instance.in_tag_key
+    assert_equal "time_in", d.instance.in_time_key
+    assert_equal "time_out", d.instance.out_time_key
+    assert_equal "%Y-%m-%d %H:%M:%S", d.instance.in_time_format
+    assert_equal "%Y-%m-%d %H:%M:%S", d.instance.out_time_format
     assert_equal true, d.instance.localtime
     assert_equal 3, d.instance.num_children