Work toward threadsafety

Fix off-by-one in comparison for resurrecton (fixes jruby test failures)

Use jruby in test matrix

Don't test on 1.8.7 anymore

Fixed retry logic / added tests for manticore

Work toward pool tests

Round out tests for manticore sniffer

Clarify code around normalizing host parameters

Fix healtcheck path option

Partial fixes based on PR review

Incorporate more PR feedback

Author: andrewvc

.travis.yml

@@ -14,13 +14,10 @@ branches:
     - travis
 
 rvm:
-  - 1.8.7
   - 1.9.3
   - 2.1
   - 2.2
-
-jdk:
-  - openjdk7
+  - jruby-1.7.24
 
 env:
   - TEST_SUITE=unit

elasticsearch-transport/lib/elasticsearch/transport/client.rb

@@ -128,6 +128,11 @@ def perform_request(method, path, params={}, body=nil)
         transport.perform_request method, path, params, body
       end
 
+      # Shuts down the client
+      def close
+        @transport.__close_connections
+      end
+
       # Normalizes and returns hosts configuration.
       #
       # Arrayifies the `hosts_config` argument and extracts `host` and `port` info from strings.

elasticsearch-transport/lib/elasticsearch/transport/transport/base.rb

@@ -44,7 +44,8 @@ def initialize(arguments={}, &block)
           @logger      = options[:logger]
           @tracer      = options[:tracer]
 
-          @sniffer     = options[:sniffer_class] ? options[:sniffer_class].new(self) : Sniffer.new(self)
+          sniffer_class = options.fetch(:sniffer_class, Sniffer)
+          @sniffer     = sniffer_class.new(self, @logger)
           @counter     = 0
           @counter_mtx = Mutex.new
           @last_request_at = Time.now
@@ -165,8 +166,10 @@ def __close_connections
         #
         def __log(method, path, params, body, url, response, json, took, duration)
           sanitized_url = url.to_s.gsub(/\/\/(.+):(.+)@/, '//' + '\1:' + SANITIZED_PASSWORD +  '@')
+
           logger.info  "#{method.to_s.upcase} #{sanitized_url} " +
                        "[status:#{response.status}, request:#{sprintf('%.3fs', duration)}, query:#{took}]"
+
           logger.debug "> #{__convert_to_json(body)}" if body
           logger.debug "< #{response.body}"
         end
@@ -197,8 +200,8 @@ def __trace(method, path, params, body, url, response, json, took, duration)
         # @api private
         #
         def __raise_transport_error(response)
-          error = ERRORS[response.status] || ServerError
-          raise error.new "[#{response.status}] #{response.body}"
+          error_class = ERRORS.fetch(response.status, ServerError)
+          raise error_class.new(response), "[#{response.status}] #{response.body}"
         end
 
         # Converts any non-String object to JSON
@@ -260,8 +263,7 @@ def perform_request(method, path, params={}, body=nil, &block)
 
             # Raise an exception so we can catch it for `retry_on_status`
             __raise_transport_error(response) if response.status.to_i >= 300 && @retry_on_status.include?(response.status.to_i)
-
-          rescue Elasticsearch::Transport::Transport::ServerError => e
+          rescue ::Elasticsearch::Transport::Transport::ServerError => e
             if @retry_on_status.include?(response.status)
               logger.warn "[#{e.class}] Attempt #{tries} to get response from #{url}" if logger
               logger.debug "[#{e.class}] Attempt #{tries} to get response from #{url}" if logger
@@ -323,6 +325,11 @@ def perform_request(method, path, params={}, body=nil, &block)
           @last_request_at = Time.now
         end
 
+        def __deserialize_response(response)
+          serializer.load(response.body) if response.headers && response.headers["content-type"] =~ /json/
+
+        end
+
         # @abstract Returns an Array of connection errors specific to the transport implementation.
         #           See {HTTP::Faraday#host_unreachable_exceptions} for an example.
         #

elasticsearch-transport/lib/elasticsearch/transport/transport/connections/connection.rb

@@ -114,7 +114,7 @@ def resurrect!
           #
           def resurrectable?
             @state_mutex.synchronize {
-              Time.now > @dead_since + ( @options[:resurrect_timeout] * 2 ** (@failures-1) )
+              Time.now >= @dead_since + ( @options[:resurrect_timeout] * 2 ** (@failures-1) )
             }
           end
 

elasticsearch-transport/lib/elasticsearch/transport/transport/errors.rb

@@ -12,7 +12,26 @@ class SnifferTimeoutError < Timeout::Error; end
 
       # Elasticsearch server error (HTTP status 5xx)
       #
-      class ServerError < Error; end
+      class ServerError < Error;
+        attr_reader :response
+
+        def initialize(response)
+          @response = response
+        end
+      end
+
+      class HostUnreachableError < Error;
+        attr_reader :original_error, :url
+
+        def initialize(original_error, url)
+          @original_error = original_error
+          @url = url
+        end
+
+        def message
+          "[#{original_error.class}] #{original_error.message}"
+        end
+      end
 
       module Errors; end
 

elasticsearch-transport/lib/elasticsearch/transport/transport/http/manticore.rb

@@ -1,4 +1,7 @@
 require 'manticore'
+require "elasticsearch/transport/transport/http/manticore/pool"
+require "elasticsearch/transport/transport/http/manticore/adapter"
+require "elasticsearch/transport/transport/http/manticore/manticore_sniffer"
 
 module Elasticsearch
   module Transport
@@ -43,97 +46,159 @@ module HTTP
         # @see Transport::Base
         #
         class Manticore
+          attr_reader :pool, :adapter, :options
           include Base
 
           def initialize(arguments={}, &block)
-            @manticore = build_client(arguments[:options] || {})
-            super(arguments, &block)
-          end
-
-          # Should just be run once at startup
-          def build_client(options={})
-            client_options = options[:transport_options] || {}
-            client_options[:ssl] = options[:ssl] || {}
+            @options = arguments[:options] || {}
+            @logger = options[:logger]
+            @adapter = Adapter.new(logger, options)
+            @healthcheck_path = options[:healthcheck_path] || "/"
+            normalized_hosts = (arguments[:hosts] || []).map {|h| normalize_host(h)}
+            @pool = Manticore::Pool.new(logger, @adapter, @healthcheck_path, normalized_hosts)
+            @protocol    = options[:protocol] || DEFAULT_PROTOCOL
+            @serializer = options[:serializer] || ( options[:serializer_class] ? options[:serializer_class].new(self) : DEFAULT_SERIALIZER_CLASS.new(self) )
+            @max_retries     = options[:retry_on_failure].is_a?(Fixnum)   ? options[:retry_on_failure]   : DEFAULT_MAX_RETRIES
+            @retry_on_status = Array(options[:retry_on_status]).map { |d| d.to_i }
 
-            @manticore = ::Manticore::Client.new(client_options)
+            setup_sniffing!
           end
 
-          # Performs the request by invoking {Transport::Base#perform_request} with a block.
-          #
-          # @return [Response]
-          # @see    Transport::Base#perform_request
-          #
-          def perform_request(method, path, params={}, body=nil)
-            super do |connection, url|
-              params[:body] = __convert_to_json(body) if body
-              params = params.merge @request_options
-              case method
-              when "GET"
-                resp = connection.connection.get(url, params)
-              when "HEAD"
-                resp = connection.connection.head(url, params)
-              when "PUT"
-                resp = connection.connection.put(url, params)
-              when "POST"
-                resp = connection.connection.post(url, params)
-              when "DELETE"
-                resp = connection.connection.delete(url, params)
+          def normalize_host(host)
+            case host
+            when URI
+              host
+            when String
+              URI.parse(host)
+            when Hash
+              host = host.clone
+              host[:scheme] ||= (host[:scheme] || host[:protocol] || "http").to_s
+              if host[:scheme] == 'http'
+                URI::HTTP.build(host)
+              elsif scheme == 'https'
+                URI::HTTPS.build(host)
               else
-                raise ArgumentError.new "Method #{method} not supported"
+                raise ArgumentError, "Unrecognized scheme for host #{host}"
               end
-              Response.new resp.code, resp.read_body, resp.headers
+            else
+              raise ArgumentError, "Host parameter #{host} is not valid! Try something like 'http://localhost:9200'!"
             end
           end
 
-          # Builds and returns a collection of connections.
-          # Each connection is a Manticore::Client
-          #
-          # @return [Connections::Collection]
-          #
-          def __build_connections
-            @request_options = {}
+          def setup_sniffing!
+            if options[:sniffing] || options[:reload_connections]
+              # We don't support sniffers that aren't threadsafe with timers here!
+              sniffer_class = options[:sniffer_class] ? options[:sniffer_class] : ::Elasticsearch::Transport::Transport::HTTP::Manticore::ManticoreSniffer
+              raise ArgumentError, "Sniffer class #{sniffer_class} must be a ManticoreSniffer!" if sniffer_class.nil? || !sniffer_class.ancestors.include?(::Elasticsearch::Transport::Transport::HTTP::Manticore::ManticoreSniffer)
+              @sniffer = sniffer_class.new(self, logger)
+              @sniffer.sniff_every(options[:sniffer_delay] || 5) do |urls|
+                logger.info("Will update internal host pool with #{urls.inspect}")
+                @pool.update_urls(urls)
+              end
+            end
+          end
 
-            if options.key?(:headers)
-              @request_options[:headers] = options[:headers]
+          # Sniff (if enabled) to get the newest list of hosts
+          # then attempt to resurrect any dead URLs
+          def reload_connections!
+            if options[:sniffing]
+              @pool.update_urls(@sniffer.hosts)
             end
+            @pool.resurrect_dead!
+          end
 
-            Connections::Collection.new \
-              :connections => hosts.map { |host|
-                host[:protocol]   = host[:scheme] || DEFAULT_PROTOCOL
-                host[:port]     ||= DEFAULT_PORT
+          def perform_request(method, path, params={}, body=nil)
+            body = __convert_to_json(body) if body
+            url, response = with_request_retries do
+              url, response = @pool.perform_request(method, path, params, body)
+              # Raise an exception so we can catch it for `retry_on_status`
+              __raise_transport_error(response) if response.status.to_i >= 300 && @retry_on_status.include?(response.status.to_i)
+              [url, response]
+            end
+
+            enrich_response(method, url, path, params, body, response)
+          end
+
+          # This takes a host string to aid in debug logging
+          def with_request_retries
+            tries = 0
+            begin
+              tries += 1
+              yield
+            rescue ::Elasticsearch::Transport::Transport::ServerError => e
+              if @retry_on_status.include?(e.response.status)
+                logger.warn "[#{e.class}] Attempt #{tries} to get response from #{url}" if logger
+                logger.debug "[#{e.class}] Attempt #{tries} to get response from #{url}" if logger
+                if tries <= max_retries
+                  retry
+                else
+                  logger.error "[#{e.class}] Cannot get response from #{url} after #{tries} tries" if logger
+                  raise e
+                end
+              else
+                raise e
+              end
+            rescue ::Elasticsearch::Transport::Transport::HostUnreachableError => e
+              logger.error "[#{e.class}] #{e.message} #{e.url}" if logger
 
-                host.delete(:user)     # auth is not supported here.
-                host.delete(:password) # use the headers
+              if @options[:retry_on_failure]
+                logger.warn "[#{e.class}] Attempt #{tries} connecting to #{connection.host.inspect}" if logger
+                if tries <= max_retries
+                  if @options[:reload_on_failure] && pool.alive_urls_count == 0
+                    logger.warn "[#{e.class}] Reloading connections (attempt #{tries} of #{max_retries})" if logger
+                    reload_connections!
+                  end
 
-                Connections::Connection.new \
-                  :host => host,
-                  :connection => @manticore
-              },
-              :selector_class => options[:selector_class],
-              :selector => options[:selector]
+                  retry
+                else
+                  logger.fatal "[#{e.class}] Cannot connect to #{connection.host.inspect} after #{tries} tries" if logger
+                  raise e
+                end
+              end
+            rescue Exception => e
+              logger.fatal "[#{e.class}] #{e.message} ()" if logger
+              raise e
+            end
           end
 
-          # Closes all connections by marking them as dead
-          # and closing the underlying HttpClient instances
-          #
-          # @return [Connections::Collection]
-          #
           def __close_connections
-            # The Manticore adapter uses a single long-lived instance
-            # of Manticore::Client, so we don't close the connections.
+            if @sniffer
+              logger.info("Closing sniffer...") if logger
+              @sniffer.close
+            end
+            logger.info("Sniffer closed.") if logger
+            logger.info("Closing pool") if logger
+            @pool.close # closes adapter as well
+            logger.info("Pool closed") if logger
+          end
+
+          def enrich_response(method, url, path, params, body, response)
+            start = Time.now if logger || tracer
+
+            duration = Time.now-start if logger || tracer
+
+            if response.status.to_i >= 300
+              __log    method, path, params, body, url, response, nil, 'N/A', duration if logger
+              __trace  method, path, params, body, url, response, nil, 'N/A', duration if tracer
+              __log_failed response if logger
+              __raise_transport_error response
+            end
+
+            json = __deserialize_response(response)
+            if json
+              took     = (json['took'] ? sprintf('%.3fs', json['took']/1000.0) : 'n/a') rescue 'n/a' if logger || tracer
+
+              __log   method, path, params, body, url, response, json, took, duration if logger
+              __trace method, path, params, body, url, response, json, took, duration if tracer
+            end
+
+            # If the response wasn't JSON we just return it as a string
+            data = json || response.body
+            ::Elasticsearch::Transport::Transport::Response.new response.status, data, response.headers
           end
 
-          # Returns an array of implementation specific connection errors.
-          #
-          # @return [Array]
-          #
           def host_unreachable_exceptions
-            [
-              ::Manticore::Timeout,
-              ::Manticore::SocketException,
-              ::Manticore::ClientProtocolException,
-              ::Manticore::ResolutionFailure
-            ]
+            [::Manticore::Timeout,::Manticore::SocketException, ::Manticore::ClientProtocolException, ::Manticore::ResolutionFailure]
           end
         end
       end