elasticsearch.rb in logstash-input-elasticsearch-4.9.0

- old
+ new

@@ -1,10 +1,11 @@
 # encoding: utf-8
 require "logstash/inputs/base"
 require "logstash/namespace"
 require "logstash/json"
 require "logstash/util/safe_uri"
+require 'logstash/plugin_mixins/validator_support/field_reference_validation_adapter'
 require "base64"
 require_relative "patch"
 
 
 # .Compatibility Note
@@ -60,10 +61,12 @@
 #
 # Further documentation describing this syntax can be found https://github.com/jmettraux/rufus-scheduler#parsing-cronlines-and-time-strings[here].
 #
 #
 class LogStash::Inputs::Elasticsearch < LogStash::Inputs::Base
+  extend LogStash::PluginMixins::ValidatorSupport::FieldReferenceValidationAdapter
+
   config_name "elasticsearch"
 
   default :codec, "json"
 
   # List of elasticsearch hosts to use for querying.
@@ -173,10 +176,13 @@
   #
   # There is no schedule by default. If no schedule is given, then the statement is run
   # exactly once.
   config :schedule, :validate => :string
 
+  # If set, the _source of each hit will be added nested under the target instead of at the top-level
+  config :target, :validate => :field_reference
+
   def register
     require "elasticsearch"
     require "rufus/scheduler"
     require "elasticsearch/transport/transport/http/manticore"
 
@@ -296,10 +302,15 @@
     # return no hits and original scroll_id so we can try to clear it
     [false, scroll_id]
   end
 
   def push_hit(hit, output_queue)
-    event = LogStash::Event.new(hit['_source'])
+    if @target.nil?
+      event = LogStash::Event.new(hit['_source'])
+    else
+      event = LogStash::Event.new
+      event.set(@target, hit['_source'])
+    end
 
     if @docinfo
       # do not assume event[@docinfo_target] to be in-place updatable. first get it, update it, then at the end set it in the event.
       docinfo_target = event.get(@docinfo_target) || {}
 