Support routing option for documents

The bulk API supports a specified `_routing` value for document
placement within a shard in an index. This should be exposed for custom
routing for aliases, filtering, etc.

http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/docs-bulk.html#bulk-routing

This also respects the move away from the routing parameter within the
event as defined in:
elastic/elasticsearch#6730

Author: masteinhauser

lib/logstash/outputs/elasticsearch.rb

@@ -49,6 +49,10 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   # similar events to the same 'type'. String expansion `%{foo}` works here.
   config :index_type, :validate => :string
 
+  # A routing override to be applied to all processed events.
+  # This can be dynamic using the `%{foo}` syntax.
+  config :routing, :validate => :string
+
   # Starting in Logstash 1.3 (unless you set option `manage_template` to false)
   # a default mapping template for Elasticsearch will be applied, if you do not
   # already have one set to match the index pattern defined (default of
@@ -117,7 +121,7 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   # Run the Elasticsearch server embedded in this process.
   # This option is useful if you want to run a single Logstash process that
   # handles log processing and indexing; it saves you from needing to run
-  # a separate Elasticsearch process. An example use case is 
+  # a separate Elasticsearch process. An example use case is
   # proof-of-concept testing.
   # WARNING: This is not recommended for production use!
   config :embedded, :validate => :boolean, :default => false
@@ -250,7 +254,7 @@ def register
       client_settings["network.host"] = @bind_host if @bind_host
       client_settings["transport.tcp.port"] = @bind_port if @bind_port
       client_settings["client.transport.sniff"] = @sniffing
- 
+
       if @node_name
         client_settings["node.name"] = @node_name
       else
@@ -382,7 +386,7 @@ def get_template
   def receive(event)
     return unless output?(event)
 
-    # block until we have not maxed out our 
+    # block until we have not maxed out our
     # retry queue. This is applying back-pressure
     # to slow down the receive-rate
     @retry_flush_mutex.synchronize {
@@ -393,11 +397,11 @@ def receive(event)
 
     # Set the 'type' value for the index.
     type = @index_type ? event.sprintf(@index_type) : (event["type"] || "logs")
-
     index = event.sprintf(@index)
-
     document_id = @document_id ? event.sprintf(@document_id) : nil
-    buffer_receive([event.sprintf(@action), { :_id => document_id, :_index => index, :_type => type }, event])
+    routing = @routing ? event.sprintf(@routing) : nil
+
+    buffer_receive([event.sprintf(@action), { :_id => document_id, :_index => index, :_type => type, :_routing => routing }, event])
   end # def receive
 
   public
@@ -447,19 +451,19 @@ def teardown
 
     @retry_teardown_requested.make_true
     # First, make sure retry_timer_thread is stopped
-    # to ensure we do not signal a retry based on 
+    # to ensure we do not signal a retry based on
     # the retry interval.
     Thread.kill(@retry_timer_thread)
     @retry_timer_thread.join
-    # Signal flushing in the case that #retry_flush is in 
+    # Signal flushing in the case that #retry_flush is in
     # the process of waiting for a signal.
     @retry_flush_mutex.synchronize { @retry_queue_needs_flushing.signal }
-    # Now, #retry_flush is ensured to not be in a state of 
+    # Now, #retry_flush is ensured to not be in a state of
     # waiting and can be safely joined into the main thread
     # for further final execution of an in-process remaining call.
     @retry_thread.join
 
-    # execute any final actions along with a proceeding retry for any 
+    # execute any final actions along with a proceeding retry for any
     # final actions that did not succeed.
     buffer_flush(:final => true)
     retry_flush
@@ -544,11 +548,11 @@ def generate_jks cert_path
   end
 
   private
-  # in charge of submitting any actions in @retry_queue that need to be 
+  # in charge of submitting any actions in @retry_queue that need to be
   # retried
   #
   # This method is not called concurrently. It is only called by @retry_thread
-  # and once that thread is ended during the teardown process, a final call 
+  # and once that thread is ended during the teardown process, a final call
   # to this method is done upon teardown in the main thread.
   def retry_flush()
     unless @retry_queue.empty?