switch to manticore and improve logging during connection failures

Author: jsvd

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 3.5.0
+  - Added connection check during register to avoid failures during processing
+  - Changed Elasticsearch Client transport to use Manticore
+  - Changed amount of logging details during connection failure
+
 ## 3.4.0
   - Adds `[@metadata][total_hits]` with total hits returned from the query ([#106](https://github.com/logstash-plugins/logstash-filter-elasticsearch/pull/106))
   - Improves error logging to fully inspect caught exceptions ([#105](https://github.com/logstash-plugins/logstash-filter-elasticsearch/pull/105))

lib/logstash/filters/elasticsearch.rb

@@ -71,6 +71,7 @@ def register
       @query_dsl = file.read
     end
 
+    test_connection!
   end # def register
 
   def filter(event)
@@ -123,7 +124,13 @@ def filter(event)
       end
 
     rescue => e
-      @logger.warn("Failed to query elasticsearch for previous event", :index => @index, :query => query, :event => event, :error => e.inspect)
+      if @logger.trace?
+        @logger.warn("Failed to query elasticsearch for previous event", :index => @index, :query => query, :event => event.to_hash, :error => e.message, :backtrace => e.backtrace)
+      elsif @logger.debug?
+        @logger.warn("Failed to query elasticsearch for previous event", :index => @index, :error => e.message, :backtrace => e.backtrace)
+      else
+        @logger.warn("Failed to query elasticsearch for previous event", :index => @index, :error => e.message)
+      end
       @tag_on_failure.each{|tag| event.tag(tag)}
     else
       filter_matched(event) if matched
@@ -165,4 +172,8 @@ def extract_value(source, path)
       memo[old_key_fragment]
     end
   end
+
+  def test_connection!
+    get_client.client.ping
+  end
 end #class LogStash::Filters::Elasticsearch

lib/logstash/filters/elasticsearch/client.rb

@@ -1,6 +1,7 @@
 # encoding: utf-8
 require "elasticsearch"
 require "base64"
+require "elasticsearch/transport/transport/http/manticore"
 
 
 module LogStash
@@ -22,10 +23,11 @@ def initialize(user, password, options={})
 
         hosts.map! {|h| { host: h, scheme: 'https' } } if ssl
         # set ca_file even if ssl isn't on, since the host can be an https url
-        transport_options[:ssl] = { ca_file: options[:ca_file] } if options[:ca_file]
+        ssl_options = { ssl: true, ca_file: options[:ca_file] } if options[:ca_file]
+        ssl_options ||= {}
 
         @logger.info("New ElasticSearch filter client", :hosts => hosts)
-        @client = ::Elasticsearch::Client.new(hosts: hosts, transport_options: transport_options)
+        @client = ::Elasticsearch::Client.new(hosts: hosts, transport_options: transport_options, transport_class: ::Elasticsearch::Transport::Transport::HTTP::Manticore, :ssl => ssl_options)
       end
 
       def search(params)

logstash-filter-elasticsearch.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-filter-elasticsearch'
-  s.version         = '3.4.0'
+  s.version         = '3.5.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Copies fields from previous log events in Elasticsearch to current events "
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
@@ -22,6 +22,7 @@ Gem::Specification.new do |s|
   # Gem dependencies
   s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
   s.add_runtime_dependency 'elasticsearch', ">= 5.0.3", " <6.0.0"
+  s.add_runtime_dependency 'manticore', "~> 0.6"
 
   s.add_development_dependency 'logstash-devutils'
 end

spec/filters/elasticsearch_spec.rb

@@ -9,6 +9,9 @@
   context "registration" do
 
     let(:plugin) { LogStash::Plugin.lookup("filter", "elasticsearch").new({}) }
+    before do
+      allow(plugin).to receive(:test_connection!)
+    end
 
     it "should not raise an exception" do
       expect {plugin.register}.to_not raise_error
@@ -37,6 +40,7 @@
     before(:each) do
       allow(LogStash::Filters::ElasticsearchClient).to receive(:new).and_return(client)
       allow(client).to receive(:search).and_return(response)
+      allow(plugin).to receive(:test_connection!)
       plugin.register
     end