Dynamic include_keys and exclude_keys for kv filter

Closes #1618

lib/logstash/filters/kv.rb

@@ -206,13 +206,18 @@ class LogStash::Filters::KV < LogStash::Filters::Base
     if !event =~ /[@field_split]/
       return kv_keys
     end
+    
+    # Interpret dynamic keys for @include_keys and @exclude_keys
+    include_keys = @include_keys.map{|key| event.sprintf(key)}
+    exclude_keys = @exclude_keys.map{|key| event.sprintf(key)}
+    
     text.scan(@scan_re) do |key, v1, v2, v3|
       value = v1 || v2 || v3
       key = @trimkey.nil? ? key : key.gsub(@trimkey_re, "")
-
-      # Bail out as per the values of @include_keys and @exclude_keys
-      next if not @include_keys.empty? and not @include_keys.include?(key)
-      next if @exclude_keys.include?(key)
+      
+      # Bail out as per the values of include_keys and exclude_keys
+      next if not include_keys.empty? and not include_keys.include?(key)
+      next if exclude_keys.include?(key)
 
       key = event.sprintf(@prefix) + key
 

spec/filters/kv.rb

@@ -346,6 +346,38 @@ describe LogStash::Filters::KV do
       insist { subject["__doublequoted"] } == "hello world"
     end
   end
+  
+  describe "test include_keys with dynamic key" do
+    config <<-CONFIG
+      filter {
+        kv {
+          source => "data"
+          include_keys => [ "%{key}"]
+        }
+      }
+    CONFIG
+    
+    sample({"data" => "foo=bar baz=fizz", "key" => "foo"}) do
+      insist { subject["foo"] } == "bar"
+      insist { subject["baz"] } == nil
+    end
+  end
+  
+  describe "test exclude_keys with dynamic key" do
+    config <<-CONFIG
+      filter {
+        kv {
+          source => "data"
+          exclude_keys => [ "%{key}"]
+        }
+      }
+    CONFIG
+    
+    sample({"data" => "foo=bar baz=fizz", "key" => "foo"}) do
+      insist { subject["foo"] } == nil
+      insist { subject["baz"] } == "fizz"
+    end
+  end
 
   describe "test include_keys and exclude_keys" do
     config <<-CONFIG