Add Retries to the Elasticsearch Client

.ci/setup.sh

@@ -0,0 +1,11 @@
+# user_agent requires /etc/protocols, which is provided by netbase.
+# https://github.com/jruby/jruby/issues/3955
+if [ ! -f "/etc/protocols" ]; then
+  if [ $(command -v apt-get) ]; then
+    echo "installing netbase with apt-get"
+    sudo apt-get install -y netbase
+  else
+    echo "installing netbase with yum"
+    sudo yum install -y netbase
+  fi
+fi

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.14.0
+  - Added support for configurable retries with new `retry_on_failure` and `retry_on_status` options [#160](https://github.com/logstash-plugins/logstash-filter-elasticsearch/pull/160)
+
 ## 3.13.0
   - Added support for this plugin identifying itself to Elasticsearch with an SSL/TLS client certificate using a new `keystore` option [#162](https://github.com/logstash-plugins/logstash-filter-elasticsearch/pull/162)
 

docs/index.asciidoc

@@ -142,6 +142,8 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-query>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-query_template>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-result_size>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-retry_on_failure>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-result_on_status_>> |<<number,number list>>|No
 | <<plugins-{type}s-{plugin}-sort>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-ssl>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-keystore>> |a valid filesystem path|No
@@ -330,11 +332,30 @@ the {ref}/query-dsl.html[Elasticsearch query documentation].
 [id="plugins-{type}s-{plugin}-result_size"]
 ===== `result_size` 
 
-  * Value type is <<number,number>>
-  * Default value is `1`
+* Value type is <<number,number>>
+* Default value is `1`
 
 How many results to return
 
+[id="plugins-{type}s-{plugin}-retry_on_failure"]
+===== `retry_on_failure`
+
+* Value type is <<number,number>>
+* Default value is `0` (retries disabled)
+
+How many times to retry an individual failed request.
+
+When enabled, retry requests that result in connection errors or an HTTP status code included in <<plugins-{type}s-{plugin}-retry_on_status>>
+
+[id="plugins-{type}s-{plugin}-retry_on_status"]
+===== `retry_on_status`
+
+* Value type is <<number,number list>>
+* Default value is an empty list `[]`
+
+Which HTTP Status codes to consider for retries (in addition to connection errors) when using <<plugins-{type}s-{plugin}-retry_on_failure>>,
+
+
 [id="plugins-{type}s-{plugin}-sort"]
 ===== `sort` 
 

lib/logstash/filters/elasticsearch.rb

@@ -82,6 +82,12 @@ class LogStash::Filters::Elasticsearch < LogStash::Filters::Base
   # Tags the event on failure to look up geo information. This can be used in later analysis.
   config :tag_on_failure, :validate => :array, :default => ["_elasticsearch_lookup_failure"]
 
+  # How many times to retry on failure?
+  config :retry_on_failure, :validate => :number, :default => 0
+
+  # What status codes to retry on?
+  config :retry_on_status, :validate => :number, :list => true, :default => [500, 502, 503, 504]
+
   # config :ca_trusted_fingerprint, :validate => :sha_256_hex
   include LogStash::PluginMixins::CATrustedFingerprintSupport
 
@@ -215,6 +221,8 @@ def client_options
       :proxy => @proxy,
       :ssl => @ssl,
       :ca_file => @ca_file,
+      :retry_on_failure => @retry_on_failure,
+      :retry_on_status => @retry_on_status,
       :keystore => @keystore,
       :keystore_password => @keystore_password,
       :ssl_trust_strategy => trust_strategy_for_ca_trusted_fingerprint

lib/logstash/filters/elasticsearch/client.rb

@@ -34,15 +34,23 @@ def initialize(logger, hosts, options = {})
         # set ca_file even if ssl isn't on, since the host can be an https url
         ssl_options.update(ssl: true, ca_file: options[:ca_file]) if options[:ca_file]
         ssl_options.update(ssl: true, trust_strategy: options[:ssl_trust_strategy]) if options[:ssl_trust_strategy]
-
         if keystore
           ssl_options[:keystore] = keystore
           logger.debug("Keystore for client certificate", :keystore => keystore)
           ssl_options[:keystore_password] = keystore_password.value if keystore_password
         end
 
+        client_options = {
+                       hosts: hosts,
+             transport_class: ::Elasticsearch::Transport::Transport::HTTP::Manticore,
+           transport_options: transport_options,
+                         ssl: ssl_options,
+            retry_on_failure: options[:retry_on_failure],
+             retry_on_status: options[:retry_on_status]
+        }
+
         logger.info("New ElasticSearch filter client", :hosts => hosts)
-        @client = ::Elasticsearch::Client.new(hosts: hosts, transport_options: transport_options, transport_class: ::Elasticsearch::Transport::Transport::HTTP::Manticore, :ssl => ssl_options)
+        @client = ::Elasticsearch::Client.new(client_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.13.0'
+  s.version         = '3.14.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"
@@ -29,4 +29,3 @@ Gem::Specification.new do |s|
 
   s.add_development_dependency 'logstash-devutils'
 end
-

spec/filters/elasticsearch_spec.rb

@@ -301,6 +301,15 @@
       end
     end
 
+    context 'with client-level retries' do
+      let(:config) do
+        super().merge(
+          "retry_on_failure" => 3,
+          "retry_on_status" => [500]
+        )
+      end
+    end
+
     context "if query is on nested field" do
       let(:config) do
         {
@@ -559,6 +568,28 @@ def wait_receive_request
         end
       end
     end
+
+    describe "retry_on_failure" do
+      let(:config) { super().merge("retry_on_failure" => 3) }
+
+      it 'propagates to the client' do
+        plugin.register
+
+        client = plugin.send(:get_client).client
+        expect( extract_transport(client).options[:retry_on_failure] ).to eq(3)
+      end
+    end
+
+    describe "retry_on_status" do
+      let(:config) { super().merge("retry_on_status" => [500, 502, 503, 504]) }
+
+      it 'propagates to the client' do
+        plugin.register
+
+        client = plugin.send(:get_client).client
+        expect( extract_transport(client).options[:retry_on_status] ).to eq([500, 502, 503, 504])
+      end
+    end
   end
 
   describe "ca_trusted_fingerprint" do