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/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

@@ -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
 
 

spec/integration/outputs/parent_spec.rb

@@ -55,10 +55,7 @@ def mock_actions_with_response(*resp)
     mock_actions_with_response({"errors" => false})
     expect(subject).to receive(:submit).with([action1, action2]).once.and_call_original
     subject.register
-    subject.receive(event1)
-    subject.receive(event2)
-    subject.flush
-    sleep(2)
+    subject.multi_receive([event1, event2])
   end
 
   it "retry exceptions within the submit body" do
@@ -73,8 +70,7 @@ def mock_actions_with_response(*resp)
       end
     end
 
-    subject.receive(event1)
-    subject.flush
+    subject.multi_receive([event1])
   end
 
   it "should retry actions with response status of 503" do
@@ -86,12 +82,7 @@ def mock_actions_with_response(*resp)
     expect(subject).to receive(:submit).with([action2]).ordered.once.and_call_original
 
     subject.register
-    subject.receive(event1)
-    subject.receive(event1)
-    subject.receive(event1)
-    subject.receive(event2)
-    subject.flush
-    sleep(3)
+    subject.multi_receive([event1, event1, event1, event2])
   end
 
   it "should retry actions with response status of 429" do
@@ -101,9 +92,7 @@ def mock_actions_with_response(*resp)
                                {"errors" => false})
     expect(subject).to receive(:submit).with([action1]).twice.and_call_original
 
-    subject.receive(event1)
-    subject.flush
-    sleep(3)
+    subject.multi_receive([event1])
   end
 
   it "should retry an event infinitely until a non retryable status occurs" do
@@ -115,49 +104,38 @@ def mock_actions_with_response(*resp)
                                {"errors" => true, "statuses" => [500]})
     expect(subject).to receive(:submit).with([action1]).exactly(6).times.and_call_original
     subject.register
-    subject.receive(event1)
-    subject.flush
-    sleep(5)
+    subject.multi_receive([event1])
   end
 
   it "non-retryable errors like mapping errors (400) should be dropped and not be retried (unfortunately)" do
     subject.register
-    subject.receive(invalid_event)
     expect(subject).to receive(:submit).once.and_call_original
+    subject.multi_receive([invalid_event])
     subject.close
 
     @es.indices.refresh
-    sleep(5)
-    Stud::try(10.times) do
-      r = @es.search
-      insist { r["hits"]["total"] } == 0
-    end
+    r = @es.search
+    expect(r["hits"]["total"]).to eql(0)
   end
 
   it "successful requests should not be appended to retry queue" do
-    subject.register
-    subject.receive(event1)
     expect(subject).to receive(:submit).once.and_call_original
+
+    subject.register
+    subject.multi_receive([event1])
     subject.close
     @es.indices.refresh
-    sleep(5)
-    Stud::try(10.times) do
-      r = @es.search
-      insist { r["hits"]["total"] } == 1
-    end
+    r = @es.search
+    expect(r["hits"]["total"]).to eql(1)
   end
 
   it "should only index proper events" do
     subject.register
-    subject.receive(invalid_event)
-    subject.receive(event1)
+    subject.multi_receive([invalid_event, event1])
     subject.close
 
     @es.indices.refresh
-    sleep(5)
-    Stud::try(10.times) do
-      r = @es.search
-      insist { r["hits"]["total"] } == 1
-    end
+    r = @es.search
+    expect(r["hits"]["total"]).to eql(1)
   end
 end

spec/integration/outputs/retry_spec.rb

@@ -12,7 +12,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