Make this plugin a 'threadsafe' output (requires LS >= 2.2.0)

Fixes logstash-plugins#377
Before acceptance this will need elastic/elasticsearch-ruby#281
Merged in and used for the ruby client

Author: andrewvc

lib/logstash/outputs/elasticsearch.rb

@@ -4,7 +4,6 @@
 require "logstash/outputs/base"
 require "logstash/json"
 require "concurrent"
-require "stud/buffer"
 require "socket" # for Socket.gethostname
 require "thread" # for safe queueing
 require "uri" # for escaping user input
@@ -51,6 +50,9 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
   require "logstash/outputs/elasticsearch/common_configs"
   require "logstash/outputs/elasticsearch/common"
 
+  # This is needed to properly pool connections
+  declare_threadsafe!
+
   # Protocol agnostic (i.e. non-http, non-java specific) configs go here
   include(LogStash::Outputs::ElasticSearch::CommonConfigs)
 
@@ -135,7 +137,6 @@ def build_client
   def close
     @stopping.make_true
     @client.stop_sniffing!
-    @buffer.stop
   end
 
   @@plugins = Gem::Specification.find_all{|spec| spec.name =~ /logstash-output-elasticsearch-/ }

lib/logstash/outputs/elasticsearch/buffer.rb

@@ -1,5 +1,4 @@
 require "logstash/outputs/elasticsearch/template_manager"
-require "logstash/outputs/elasticsearch/buffer"
 
 module LogStash; module Outputs; class ElasticSearch;
   module Common
@@ -13,16 +12,11 @@ def register
       setup_hosts # properly sets @hosts
       build_client
       install_template
-      setup_buffer_and_handler
       check_action_validity
 
       @logger.info("New Elasticsearch output", :class => self.class.name, :hosts => @hosts)
     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)
       events.each_slice(@flush_size) do |slice|
@@ -37,10 +31,6 @@ def event_action_tuple(event)
       [action, params, event]
     end
 
-    def flush
-      @buffer.flush
-    end
-
     def setup_hosts
       @hosts = Array(@hosts)
       if @hosts.empty?
@@ -53,12 +43,6 @@ def install_template
       TemplateManager.install_template(self)
     end
 
-    def setup_buffer_and_handler
-      @buffer = ::LogStash::Outputs::ElasticSearch::Buffer.new(@logger, @flush_size, @idle_flush_time) do |actions|
-        retrying_submit(actions)
-      end
-    end
-
     def check_action_validity
       raise LogStash::ConfigurationError, "No action specified!" unless @action
 

lib/logstash/outputs/elasticsearch/common.rb

@@ -88,21 +88,11 @@ def self.included(mod)
       mod.config :port, :obsolete => "Please use the 'hosts' setting instead. Hosts entries can be in 'host:port' format."
 
       # This plugin uses the bulk index API for improved indexing performance.
-      # In Logstashes >= 2.2 this setting defines the maximum sized bulk request Logstash will make
+      # This setting defines the maximum sized bulk request Logstash will make
       # You you may want to increase this to be in line with your pipeline's batch size.
       # If you specify a number larger than the batch size of your pipeline it will have no effect,
       # save for the case where a filter increases the size of an inflight batch by outputting
       # events.
-      #
-      # In Logstashes <= 2.1 this plugin uses its own internal buffer of events.
-      # This config option sets that size. In these older logstashes this size may
-      # have a significant impact on heap usage, whereas in 2.2+ it will never increase it.
-      # To make efficient bulk API calls, we will buffer a certain number of
-      # events before flushing that out to Elasticsearch. This setting
-      # controls how many events will be buffered before sending a batch
-      # of events. Increasing the `flush_size` has an effect on Logstash's heap size.
-      # Remember to also increase the heap size using `LS_HEAP_SIZE` if you are sending big documents
-      # or have increased the `flush_size` to a higher value.
       mod.config :flush_size, :validate => :number, :default => 500
 
       # The amount of time since last flush before a flush is forced.

lib/logstash/outputs/elasticsearch/common_configs.rb

@@ -57,7 +57,7 @@ def start_sniffing!
       if options[:sniffing]
         @sniffer_thread = Thread.new do
           loop do
-            @request_mutex.synchronize { sniff! }
+            sniff!
             sleep (options[:sniffing_delay].to_f || 30)
           end
         end
@@ -70,8 +70,12 @@ def stop_sniffing!
 
     def sniff!
       client.transport.reload_connections! if options[:sniffing]
-      hosts_by_name = client.transport.hosts.map {|h| h["name"]}.sort
-      @logger.debug({"count" => hosts_by_name.count, "hosts" => hosts_by_name})
+
+      if @logger.debug?
+        hosts_by_name = client.transport.hosts.map {|h| h["name"]}.sort
+        @logger.debug({"count" => hosts_by_name.count, "hosts" => hosts_by_name})
+      end
+
     rescue StandardError => e
       @logger.error("Error while sniffing connection",
                     :message => e.message,

lib/logstash/outputs/elasticsearch/http_client.rb

@@ -24,7 +24,7 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency 'elasticsearch', ['>= 1.0.13', '~> 1.0']
   s.add_runtime_dependency 'stud', ['>= 0.0.17', '~> 0.0']
   s.add_runtime_dependency 'cabin', ['~> 0.6']
-  s.add_runtime_dependency "logstash-core", ">= 2.0.0", "< 3.0.0"
+  s.add_runtime_dependency "logstash-core", ">= 2.2.0", "< 3.0.0"
 
   s.add_development_dependency 'ftw', '~> 0.0.42'
   s.add_development_dependency 'logstash-codec-plain'

logstash-output-elasticsearch.gemspec

@@ -29,8 +29,7 @@ def get_es_output(action, id = nil)
     it "should create new documents with or without id" do
       subject = get_es_output("create", "id123")
       subject.register
-      subject.receive(LogStash::Event.new("message" => "sample message here"))
-      subject.flush
+      subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
       @es.indices.refresh
       # Wait or fail until everything's indexed.
       Stud::try(3.times) do
@@ -42,8 +41,7 @@ def get_es_output(action, id = nil)
     it "should create new documents without id" do
       subject = get_es_output("create")
       subject.register
-      subject.receive(LogStash::Event.new("message" => "sample message here"))
-      subject.flush
+      subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
       @es.indices.refresh
       # Wait or fail until everything's indexed.
       Stud::try(3.times) do

spec/integration/outputs/create_spec.rb

@@ -11,7 +11,7 @@
     before do
       subject.register
       event_count.times do
-        subject.receive(LogStash::Event.new("message" => "Hello World!", "type" => type))
+        subject.multi_receive([LogStash::Event.new("message" => "Hello World!", "type" => type)])
       end
     end
 

spec/integration/outputs/index_spec.rb

@@ -19,9 +19,7 @@
       ftw.put!("#{index_url}/#{type}_parent/test", :body => pdoc.to_json)
 
       subject.register
-      event_count.times do
-        subject.receive(LogStash::Event.new("link_to" => "test", "message" => "Hello World!", "type" => type))
-      end
+      subject.multi_receive(event_count.times.map { LogStash::Event.new("link_to" => "test", "message" => "Hello World!", "type" => type) })
     end