Added the processing when did not match the rules.

lib/fluent/plugin/out_rewrite.rb

@@ -5,7 +5,8 @@ class RewriteOutput < Output
     config_param :remove_prefix, :string, :default => nil
     config_param :add_prefix,    :string, :default => nil
 
-    attr_reader  :rules
+    attr_reader   :rules
+    attr_accessor :warn_msg
 
     def configure(conf)
       super
@@ -18,18 +19,41 @@ def configure(conf)
         @added_prefix_string = @add_prefix + '.'
       end
 
+      invalid_keys = Array.new
+
       @rules = conf.elements.select { |element|
         element.name == 'rule'
       }.each { |element|
-        if element.has_key?("pattern")
+
+        if element["tag"] || element["fallback"]
+          element["append_to_tag"] = true
+        end
+
+        if element.has_key?("pattern") && element.has_key?("key")
           element["regex"] = Regexp.new(element["pattern"])
+        else
+          raise ConfigError, "out_rewrite: In the rules section, both 'key' and 'pattern' must be set."
+        end
+
+        if !@remove_prefix && !@add_prefix
+          unless element.has_key?("ignore")
+            invalid_keys << element["key"]
+          end
+
+          if invalid_keys.include?(element["key"]) && element.has_key?("append_to_tag")
+            invalid_keys.delete(element["key"])
+          end
         end
 
         element.keys.each do |k|
           # read and throw away to supress unread configuration warning
           element[k]
         end
       }
+
+      if invalid_keys.size != 0
+        raise ConfigError, "out_rewrite: 'add_prefix' or 'remove_prefix' option has been set. or finally 'append_to_tag' must be set in the rule section. Your invalid key(s) is(are) #{invalid_keys.uniq}"
+      end
     end
 
     def start
@@ -41,6 +65,9 @@ def shutdown
     end
 
     def emit(tag, es, chain)
+      @warn_msg = nil
+      _tag = tag.clone
+
       if @remove_prefix and
         ((tag.start_with?(@removed_prefix_string) && tag.length > @removed_length) || tag == @remove_prefix)
         tag = tag[@removed_length..-1] || ''
@@ -52,7 +79,11 @@ def emit(tag, es, chain)
 
       es.each do |time, record|
         filtered_tag, record = rewrite(tag, record)
-        Engine.emit(filtered_tag, time, record) if filtered_tag && record
+        if _tag != tag && filtered_tag && record
+          Engine.emit(filtered_tag, time, record)
+        else
+          $log.warn "out_rewrite: #{@warn_msg}"
+        end
       end
 
       chain.next
@@ -75,8 +106,10 @@ def apply_rule(rule, tag, record)
       pattern    = rule["pattern"]
       last       = nil
 
-      return [tag, record] if !key || !record.has_key?(key)
-      return [tag, record] unless pattern
+      if !record.has_key?(key)
+        @warn_msg = "Since there is no matching JSON key \"#{key}\", can't emit record #{record}, cause infinity looping. Check the rules of the setting where the pattern has become \"#{pattern}\""
+        return [tag, record]
+      end
 
       if matched = record[key].match(rule["regex"])
         return if rule["ignore"]
@@ -102,6 +135,8 @@ def apply_rule(rule, tag, record)
       else
         if rule["append_to_tag"] && rule["fallback"]
           tag += (tag_prefix + rule["fallback"])
+        else
+          @warn_msg = "Since there is no rule matches, can't emit record #{record}, cause infinity looping. If you want to emit even if do not match a rules, set the 'fallback' rule."
         end
       end
 

test/plugin/test_out_rewrite.rb

@@ -16,22 +16,55 @@ def test_configure
 
       <rule>
         key foo
+        pattern \\?.+$
       </rule>
       <to_be_ignored>
         key bar
+        pattern \\?.+$
       </to_be_ignored>
       <rule>
         key baz
+        pattern \\?.+$
       </rule>
     ])
 
     assert_equal "test",     d.instance.remove_prefix
     assert_equal "filtered", d.instance.add_prefix
     assert_equal 2, d.instance.rules.size
+
+    begin
+      create_driver(%[
+        <rule>
+          key foo
+        </rule>
+      ])
+    rescue => error
+    ensure
+      assert_equal(
+        error.message,
+        "out_rewrite: In the rules section, both 'key' and 'pattern' must be set."
+      )
+    end
+
+    begin
+      create_driver(%[
+        <rule>
+          key foo
+          pattern \\?.+$
+        </rule>
+      ])
+    rescue => error
+    ensure
+      assert_equal(
+        error.message,
+        "out_rewrite: 'add_prefix' or 'remove_prefix' option has been set. or finally 'append_to_tag' must be set in the rule section. Your invalid key(s) is(are) [\"foo\"]"
+      )
+    end
   end
 
   def test_rewrite_replace
     d1 = create_driver(%[
+      add_prefix rewrited
       <rule>
         key     path
         pattern \\?.+$
@@ -45,6 +78,7 @@ def test_rewrite_replace
     )
 
     d2 = create_driver(%[
+      add_prefix rewrited
       <rule>
         key     path
         pattern (/[^/]+)\\?([^=]+)=(\\d)
@@ -189,6 +223,7 @@ def test_rewrite_append_tag
 
   def test_last
     d = create_driver(%[
+      add_prefix rewrited
       <rule>
         key     path
         pattern ^/foo$
@@ -289,6 +324,8 @@ def test_emit
     assert_equal({ "path" => "/entries/1" }, emits[3][2])
 
     d2 = create_driver(%[
+      add_prefix filtered
+
       <rule>
         key     path
         pattern \\?.+$
@@ -297,11 +334,47 @@ def test_emit
     ])
     d2.run do
       d2.emit({ "path" => "/foo?bar=1" })
-    end
+    end 
     emits = d2.emits
 
     assert_equal 1, emits.size
-    assert_equal('test', emits[0][0])
+    assert_equal('filtered.test', emits[0][0])
     assert_equal({ "path" => "/foo" }, emits[0][2])
+
+    d3 = create_driver(%[
+      <rule>
+        key     status
+        pattern ^500
+        tag     internal
+      </rule>
+    ])
+    d3.run do
+      d3.emit({ "path" => "/foo?bar=1" })
+    end
+    emits = d3.emits
+
+    assert_equal 0, emits.size
+    assert_equal(
+      "Since there is no matching JSON key \"status\", can't emit record {\"path\"=>\"/foo?bar=1\"}, cause infinity looping. Check the rules of the setting where the pattern has become \"^500\"",
+      d3.instance.warn_msg
+    )
+
+    d4 = create_driver(%[
+      <rule>
+        key     path
+        pattern ^\/(users|entries)
+        append_to_tag true
+      </rule>
+    ])
+    d4.run do
+      d4.emit({ "path" => "/pull-requester/studio3104" })
+    end
+    emits = d4.emits
+
+    assert_equal 0, emits.size
+    assert_equal(
+      "Since there is no rule matches, can't emit record \{\"path\"=>\"/pull-requester/studio3104\"\}, cause infinity looping. If you want to emit even if do not match a rules, set the 'fallback' rule.",
+      d4.instance.warn_msg
+    )
   end
 end