Refactored code sharing to play nicer with java plugin

Fixes #302

Author: andrewvc

lib/logstash/outputs/elasticsearch.rb

@@ -62,6 +62,17 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
 
   config_name "elasticsearch"
 
+  # The Elasticsearch action to perform. Valid actions are:
+  #
+  # - index: indexes a document (an event from Logstash).
+  # - delete: deletes a document by id (An id is required for this action)
+  # - create: indexes a document, fails if a document by that id already exists in the index.
+  # - update: updates a document by id. Update has a special case where you can upsert -- update a
+  #   document if not already present. See the `upsert` option
+  #
+  # For more details on actions, check out the http://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html[Elasticsearch bulk API documentation]
+  config :action, :validate => %w(index delete create update), :default => "index"
+
   # Username to authenticate to a secure Elasticsearch cluster
   config :user, :validate => :string
   # Password to authenticate to a secure Elasticsearch cluster
@@ -106,27 +117,12 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   # How long to wait, in seconds, between sniffing attempts
   config :sniffing_delay, :validate => :number, :default => 5
 
-  # Set max retry for each event. The total time spent blocked on retries will be
-  # (max_retries * retry_max_interval). This may vary a bit if Elasticsearch is very slow to respond
-  config :max_retries, :validate => :number, :default => 3
-
-  # Set max interval between bulk retries.
-  config :retry_max_interval, :validate => :number, :default => 2
-
-  # DEPRECATED This setting no longer does anything. If you need to change the number of retries in flight
-  # try increasing the total number of workers to better handle this.
-  config :retry_max_items, :validate => :number, :default => 500, :deprecated => true
-
   # Set the address of a forward HTTP proxy.
   # Can be either a string, such as `http://localhost:123` or a hash in the form
   # of `{host: 'proxy.org' port: 80 scheme: 'http'}`.
   # Note, this is NOT a SOCKS proxy, but a plain HTTP proxy
   config :proxy
 
-  # Enable `doc_as_upsert` for update mode.
-  # Create a new document with source if `document_id` doesn't exist in Elasticsearch
-  config :doc_as_upsert, :validate => :boolean, :default => false
-
   # Set the timeout for network operations and requests sent Elasticsearch. If
   # a timeout occurs, the request will be retried.
   config :timeout, :validate => :number
@@ -135,6 +131,12 @@ def build_client
     @client = ::LogStash::Outputs::ElasticSearch::HttpClientBuilder.build(@logger, @hosts, params)
   end
 
+  def close
+    @stopping.make_true
+    @client.stop_sniffing!
+    @buffer.stop
+  end
+
   @@plugins = Gem::Specification.find_all{|spec| spec.name =~ /logstash-output-elasticsearch-/ }
 
   @@plugins.each do |plugin|

lib/logstash/outputs/elasticsearch/common.rb

@@ -19,23 +19,25 @@ def register
     end
 
     def receive(event)
+      @buffer << event_action_tuple(event)
+    end
+
+    # Receive an array of events and immediately attempt to index them (no buffering)
+    def multi_receive(events)
+      retrying_submit(events.map {|e| event_action_tuple(e) })
+    end
+
+    # Convert the event into a 3-tuple of action, params, and event
+    def event_action_tuple(event)
       params = event_action_params(event)
       action = event.sprintf(@action)
-
-      @buffer << [action, params, event]
+      [action, params, event]
     end
 
     def flush
       @buffer.flush
     end
 
-    def close
-      @stopping.make_true
-      @client.stop_sniffing!
-      @buffer.stop
-    end
-
-
     def setup_hosts
       @hosts = Array(@hosts)
       if @hosts.empty?
@@ -61,7 +63,7 @@ def retrying_submit(actions)
       submit_actions = actions
 
       while submit_actions && submit_actions.length > 0 && retries_left > 0
-        return if !submit_actions # If everything's a success we move along
+        return if !submit_actions || submit_actions.empty? # If everything's a success we move along
         # We retry with whatever is didn't succeed
         begin
           submit_actions = submit(submit_actions)
@@ -76,7 +78,7 @@ def retrying_submit(actions)
     end
 
     def submit(actions)
-      es_actions = actions.map { |a, doc, event| [a, doc, event.to_hash] }
+      es_actions = actions.map { |a, doc, event| [a, doc, event.to_hash]}
 
       bulk_response = safe_bulk(es_actions,actions)
 

lib/logstash/outputs/elasticsearch/common_configs.rb

@@ -11,7 +11,7 @@ def self.included(mod)
       # The index type to write events to. Generally you should try to write only
       # similar events to the same 'type'. String expansion `%{foo}` works here.
       #
-      # Deprecated in favor of `document_type` field.
+      # Deprecated in favor of `docoument_type` field.
       mod.config :index_type, :validate => :string, :obsolete => "Please use the 'document_type' setting instead. It has the same effect, but is more appropriately named."
 
       # The document type to write events to. Generally you should try to write only
@@ -62,8 +62,7 @@ def self.included(mod)
       #     `["127.0.0.1:9200","127.0.0.2:9200"]`
       # It is important to exclude http://www.elastic.co/guide/en/elasticsearch/reference/current/modules-node.html[dedicated master nodes] from the `hosts` list
       # to prevent LS from sending bulk requests to the master nodes.  So this parameter should only reference either data or client nodes in Elasticsearch.
-
-      mod.config :hosts, :validate => :array
+      mod.config :hosts, :validate => :array, :default => ["127.0.0.1"]
 
       mod.config :host, :obsolete => "Please use the 'hosts' setting instead. You can specify multiple entries separated by comma in 'host:port' format."
 
@@ -91,20 +90,24 @@ def self.included(mod)
       # near-real-time.
       mod.config :idle_flush_time, :validate => :number, :default => 1
 
-      # The Elasticsearch action to perform. Valid actions are:
-      #
-      # - index: indexes a document (an event from Logstash).
-      # - delete: deletes a document by id (An id is required for this action)
-      # - create: indexes a document, fails if a document by that id already exists in the index.
-      # - update: updates a document by id. Update has a special case where you can upsert -- update a
-      #   document if not already present. See the `upsert` option
-      #
-      # For more details on actions, check out the http://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html[Elasticsearch bulk API documentation]
-      mod.config :action, :validate => %w(index delete create update), :default => "index"
-
-      # Set upsert content for update mode.
+      # Set upsert content for update mode.s
       # Create a new document with this parameter as json string if `document_id` doesn't exists
       mod.config :upsert, :validate => :string, :default => ""
+
+      # Enable `doc_as_upsert` for update mode.
+      # Create a new document with source if `document_id` doesn't exist in Elasticsearch
+      mod.config :doc_as_upsert, :validate => :boolean, :default => false
+
+      # Set max retry for each event. The total time spent blocked on retries will be
+      # (max_retries * retry_max_interval). This may vary a bit if Elasticsearch is very slow to respond
+      mod.config :max_retries, :validate => :number, :default => 3
+
+      # Set max interval between bulk retries.
+      mod.config :retry_max_interval, :validate => :number, :default => 2
+
+      # DEPRECATED This setting no longer does anything. If you need to change the number of retries in flight
+      # try increasing the total number of workers to better handle this.
+      mod.config :retry_max_items, :validate => :number, :default => 500, :deprecated => true
     end
   end
-end; end; end
+end end end