Partial fixes based on PR review

Author: andrewvc

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

@@ -42,8 +42,8 @@ def initialize(arguments={}, &block)
           @logger      = options[:logger]
           @tracer      = options[:tracer]
 
-          sniffer_class = options[:sniffer_class] ? options[:sniffer_class] : Sniffer
-          @sniffer     = sniffer_class.new(self, logger)
+          sniffer_class = options.fetch(:sniffer_class, Sniffer)
+          @sniffer     = sniffer_class.new(self, @logger)
           @counter     = 0
           @counter_mtx = Mutex.new
           @last_request_at = Time.now
@@ -151,8 +151,8 @@ def __trace(method, path, params, body, url, response, json, took, duration)
         #
         # @api private
         def __raise_transport_error(response)
-          error = ERRORS[response.status].new(response) || ServerError.new(response)
-          raise error, "[#{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
@@ -275,8 +275,9 @@ def perform_request(method, path, params={}, body=nil, &block)
           @last_request_at = Time.now
         end
 
-        def deserialize_response(response)
+        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.

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

@@ -54,7 +54,6 @@ def initialize(arguments={}, &block)
             @logger = options[:logger]
             @adapter = Adapter.new(logger, options)
             @healthcheck_path = options[:healthcheck_path] || "/"
-            # TODO handle HTTPS
             normalized_hosts = arguments[:hosts].map {|h| normalize_host(h)}
             @pool = Manticore::Pool.new(logger, @adapter, @healthcheck_path, normalized_hosts)
             @protocol    = options[:protocol] || DEFAULT_PROTOCOL
@@ -67,22 +66,22 @@ def initialize(arguments={}, &block)
 
           def normalize_host(host)
             case host
-              when URI
-                host.to_s
-              when String
-                URI.parse(host).to_s
-              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, "Unrecognized scheme for host #{host}"
-                end
+            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 "Host parameter #{host} is not valid! Try something like 'http://localhost:9200'!"
+                raise ArgumentError, "Unrecognized scheme for host #{host}"
+              end
+            else
+              raise ArgumentError, "Host parameter #{host} is not valid! Try something like 'http://localhost:9200'!"
             end
           end
 
@@ -109,16 +108,15 @@ def reload_connections!
           end
 
           def perform_request(method, path, params={}, body=nil)
-            with_request_retries do
-              body = __convert_to_json(body) if body
-              enriching_response(method, path, params, body) 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
+            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
@@ -134,7 +132,7 @@ def with_request_retries
                 if tries <= max_retries
                   retry
                 else
-                  logger.fatal "[#{e.class}] Cannot get response from #{url} after #{tries} tries" if logger
+                  logger.error "[#{e.class}] Cannot get response from #{url} after #{tries} tries" if logger
                   raise e
                 end
               else
@@ -147,7 +145,7 @@ def with_request_retries
                 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 #{connections.all.size})" if logger
+                    logger.warn "[#{e.class}] Reloading connections (attempt #{tries} of #{max_retries})" if logger
                     reload_connections!
                   end
 
@@ -171,13 +169,12 @@ def __close_connections
             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 enriching_response(method, path, params, body)
+          def enrich_response(method, url, path, params, body, response)
             start = Time.now if logger || tracer
 
-            url, response = yield
-
             duration = Time.now-start if logger || tracer
 
             if response.status.to_i >= 300
@@ -187,15 +184,17 @@ def enriching_response(method, path, params, body)
               __raise_transport_error response
             end
 
-            json = deserialize_response(response)
-
-            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
+            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
 
-            ::Elasticsearch::Transport::Transport::Response.new response.status, json || response.body, response.headers
+            # 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
         end
       end

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

@@ -3,6 +3,9 @@ module Transport
     module Transport
       module HTTP
         class Manticore
+          # While the Manticore Pool only supports Manticore today, one day other adapters
+          # may use it. This class provides an abstraction around the actual HTTP client for
+          # the pool.
           class Adapter
             attr_reader :manticore, :logger
 
@@ -31,18 +34,18 @@ def perform_request(url, method, path, params={}, body=nil)
               params[:body] = body if body
               url_and_path = (url + path).to_s # Convert URI object to string
               case method
-                when "GET"
-                  resp = @manticore.get(url_and_path, params)
-                when "HEAD"
-                  resp = @manticore.head(url_and_path, params)
-                when "PUT"
-                  resp = @manticore.put(url_and_path, params)
-                when "POST"
-                  resp = @manticore.post(url_and_path, params)
-                when "DELETE"
-                  resp = @manticore.delete(url_and_path, params)
-                else
-                  raise ArgumentError.new "Method #{method} not supported"
+              when "GET"
+                resp = @manticore.get(url_and_path, params)
+              when "HEAD"
+                resp = @manticore.head(url_and_path, params)
+              when "PUT"
+                resp = @manticore.put(url_and_path, params)
+              when "POST"
+                resp = @manticore.post(url_and_path, params)
+              when "DELETE"
+                resp = @manticore.delete(url_and_path, params)
+              else
+                raise ArgumentError.new "Method #{method} not supported"
               end
               Response.new resp.code, resp.read_body, resp.headers
             end

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

@@ -6,16 +6,13 @@ class Manticore
           # Handles node discovery ("sniffing")
           #
           class ManticoreSniffer < ::Elasticsearch::Transport::Transport::Sniffer
-            ES1_RE_URL  = /\[([^\/]*)?\/?([^:]*):([0-9]+)\]/
-            ES2_RE_URL  = /([^\/]*)?\/?([^:]*):([0-9]+)/
-
             attr_reader   :transport
             attr_accessor :timeout
 
             # @param transport [Object] A transport instance
             #
             def initialize(*args)
-              @timeout = 1
+              @timeout = 1 # in seconds
               @state_mutex = Mutex.new
               @stopping = false
               super
@@ -42,15 +39,15 @@ def sniff_every(interval_seconds, &block)
                         last_sniff = Time.now
                       end
                     rescue Exception => e
-                      logger.warn("Error while sniffing Elasticsearch Nodes! [#{e.class.name}][#{e.message}][#{e.backtrace.to_a}]") if logger
+                      logger.warn("Error while sniffing Elasticsearch Nodes! [#{e.class.name}][#{e.message}][#{e.backtrace.join("\n")}]") if logger
                     end
                   end
                 end
               end
             end
 
             def hosts
-              nodes = transport.perform_request('GET', '_nodes/http', :timeout => timeout).body
+              nodes = transport.perform_request('GET', '_nodes/http', :request_timeout => timeout).body
               urls = hosts_from_nodes(nodes).map {|n| "#{transport.protocol}://#{n["http_address"]}"}
               urls
             end

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

@@ -19,17 +19,17 @@ def initialize(logger, adapter, healthcheck_path="/", urls=[], resurrect_interva
             end
 
             def close
-              logger.info  "Stopping resurrectionist" if logger
+              logger.debug  "Stopping resurrectionist" if logger
               stop_resurrectionist
-              logger.info  "Waiting for open manticore connections" if logger
+              logger.debug  "Waiting for open manticore connections" if logger
               wait_for_open_connections
-              logger.info("Closing adapter #{@adapter}") if logger
+              logger.debug("Closing adapter #{@adapter}") if logger
               @adapter.close
             end
 
             def wait_for_open_connections
               until open_connections.empty?
-                logger.info "Blocked on shutdown to to open connections #{@state_mutex.synchronize {@url_info}}" if logger
+                logger.info "Blocked on shutdown to open connections #{@state_mutex.synchronize {@url_info}}" if logger
                 sleep 1
               end
             end
@@ -58,7 +58,7 @@ def resurrect_dead!
               # Try to keep locking granularity low such that we don't affect IO...
               @state_mutex.synchronize { @url_info.select {|url,meta| meta[:dead] } }.each do |url,meta|
                 begin
-                  @logger.info("Checking url #{url} with path #{@healthcheck_path} to see if node resurrected! ")
+                  @logger.info("Checking url #{url} with path #{@healthcheck_path} to see if node resurrected")
                   perform_request_to_url(url, "GET", @healthcheck_path)
                   # If no exception was raised it must have succeeded!
                   logger.warn("Resurrected connection to dead ES instance at #{url}")
@@ -81,7 +81,7 @@ def perform_request(method, path, params={}, body=nil)
             end
 
             def perform_request_to_url(url, method, path, params={}, body=nil)
-              raise ArgumentError, "No URL specified!" unless url
+              raise ArgumentError, "No URL specified" unless url
 
               @adapter.perform_request(url, method, path, params, body)
             rescue ::Manticore::Timeout,::Manticore::SocketException, ::Manticore::ClientProtocolException, ::Manticore::ResolutionFailure => e
@@ -131,7 +131,7 @@ def empty_url_meta
             def with_connection
               url, url_meta = get_connection
 
-              raise Error, "No Available connections!" unless url
+              raise Error, "No Available connections" unless url
               yield url
             rescue ::Elasticsearch::Transport::Transport::HostUnreachableError => e
               mark_dead(url, e)
@@ -143,7 +143,7 @@ def with_connection
             def mark_dead(url, error)
               @state_mutex.synchronize do
                 url_meta = @url_info[url]
-                logger.warn("Marking url #{url} as dead! Last error: [#{error.class}] #{error.message}")
+                logger.warn("Marking url #{url} as dead. Last error: [#{error.class}] #{error.message}")
                 url_meta[:dead] = true
                 url_meta[:last_error] = error
               end
@@ -160,7 +160,7 @@ def get_connection
                 # The goal here is to pick a random connection from the least-in-use connections
                 # We want some randomness so that we don't hit the same node over and over, but
                 # we also want more 'fair' behavior in the event of high concurrency
-                raise "No available connections!" if @url_info.empty?
+                raise "No available connections." if @url_info.empty?
 
                 eligible_set = nil
                 lowest_value_seen = nil

elasticsearch-transport/lib/elasticsearch/transport/transport/sniffer.rb

@@ -35,8 +35,6 @@ def hosts
           end
         end
 
-        protected
-
         def hosts_from_nodes(nodes)
           hosts = nodes['nodes'].map do |id,info|
             addr_str = info["#{transport.protocol}_address"].to_s

elasticsearch-transport/lib/elasticsearch/transport/version.rb

@@ -1,6 +1,5 @@
 module Elasticsearch
   module Transport
     VERSION = "1.0.16.pre"
-    #VERSION = "1.0.16.pre"
   end
 end