logstash-plugins#73: added support for script_fields

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 4.0.4
+  - Add support for retrieving script fields
+
 ## 4.0.3
   - Docs: Add requirement to use version 4.0.2 or higher to support sending Content-Type headers
   - Fix scrolling to use json bodies in the requests (this makes scrolling not work in ES 1.x)

lib/logstash/inputs/elasticsearch.rb

@@ -102,6 +102,9 @@ class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
   # http://www.elasticsearch.org/guide/en/elasticsearch/guide/current/_document_metadata.html
   config :docinfo_fields, :validate => :array, :default => ['_index', '_type', '_id']
 
+  # This parameter enumerates which script fields to retrieve
+  config :script_fields, :validate => :array
+
   # Basic Auth - username
   config :user, :validate => :string
 
@@ -190,6 +193,13 @@ def push_hit(hit, output_queue)
       event.set(@docinfo_target, docinfo_target)
     end
 
+    # go through the list of script fields to include in the event
+    if @script_fields && hit['fields']
+      @script_fields.each do |field|
+        event.set(field, hit['fields'][field])
+      end
+    end
+
     output_queue << event
   end
 

logstash-input-elasticsearch.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-elasticsearch'
-  s.version         = '4.0.3'
+  s.version         = '4.0.4'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Read from an Elasticsearch cluster, based on search query results"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/inputs/elasticsearch_spec.rb

@@ -16,7 +16,8 @@
         "_type" => "logs",
         "_id" => "C5b2xLQwTZa76jBmHIbwHQ",
         "_score" => 1.0,
-        "_source" => { "message" => ["ohayo"] }
+        "_source" => { "message" => ["ohayo"] },
+        "fields" => { "message_copy" => ["ohayo"] }
       }
       allow(esclient).to receive(:search) { { "hits" => { "hits" => [hit] } } }
       allow(esclient).to receive(:scroll) { { "hits" => { "hits" => [hit] } } }
@@ -50,7 +51,8 @@
           "_type" => "logs",
           "_id" => "C5b2xLQwTZa76jBmHIbwHQ",
           "_score" => 1.0,
-          "_source" => { "message" => ["ohayo"] }
+          "_source" => { "message" => ["ohayo"] },
+          "fields" => { "message_copy" => ["ohayo"] }
         } ]
       }
     }
@@ -96,7 +98,8 @@
               "message" => ["ohayo"],
               "metadata_with_hash" => { "awesome" => "logstash" },
               "metadata_with_string" => "a string"
-            }
+            },
+            "fields" => { "message_copy" => ["ohayo"] }
           } ]
         }
       }
@@ -249,5 +252,41 @@
         expect(event.get("[@metadata][_id]")).to eq(nil)
       end
     end
-  end
+
+    context "when query contains script fields but not enumerating script fields" do
+      it 'should not include the script fields at the root of the event' do
+        config = %q[
+          input {
+            elasticsearch {
+              hosts => ["localhost"]
+              query => '{ "query": { "match": { "message": "ohayo" } }, "script_fields": { "message_copy": {"script": "doc.message.values"} } }'
+            }
+          }
+        ]
+        event = input(config) do |pipeline, queue|
+          queue.pop
+        end
+
+        expect(event.get("message_copy")).to eq(nil)
+      end
+    end
+
+    context "when query contains script fields and enumerating script fields" do
+      it 'should include the script fields at the root of the event' do
+        config = %q[
+          input {
+            elasticsearch {
+              hosts => ["localhost"]
+              query => '{ "query": { "match": { "message": "ohayo" } }, "script_fields": { "message_copy": {"script": "doc.message.values"} } }'
+              script_fields => ["message_copy"]
+            }
+          }
+        ]
+        event = input(config) do |pipeline, queue|
+          queue.pop
+        end
+
+        expect(event.get("message_copy")).to eq(["ohayo"])
+      end
+    end  end
 end