Threadsafe Manticore

.travis.yml

@@ -6,21 +6,18 @@ sudo: false
 
 language: ruby
 
-cache: bundler
-
 branches:
   only:
     - master
     - travis
 
 rvm:
-  - 1.8.7
-  - 1.9.3
-  - 2.1
-  - 2.2
+  - jruby-1.7.23
+#  - 1.8.7
+#  - 1.9.3
+#  - 2.1
+#  - 2.2
 
-jdk:
-  - openjdk7
 
 env:
   - TEST_SUITE=unit
@@ -39,8 +36,7 @@ before_script:
   - gem install bundler -v 1.11.2
   - rake setup
   - rake elasticsearch:update
-  - rake bundle:clean
-  - rake bundle:install
+  - echo elasticsearch-* elasticsearch | xargs -n1 -IFF bash -c 'echo processing bundle for FF && cd FF && bundle && cd ..'
 
 script:
   - SERVER=start TEST_CLUSTER_LOGS=/tmp/log TEST_BUILD_REF=tags/v$ES_VERSION TEST_CLUSTER_COMMAND=/tmp/elasticsearch-$ES_VERSION/bin/elasticsearch rake test:$TEST_SUITE

elasticsearch-extensions/Gemfile

@@ -3,18 +3,8 @@ source 'https://rubygems.org'
 # Specify your gem's dependencies in elasticsearch-extensions.gemspec
 gemspec
 
-if File.exists? File.expand_path("../../elasticsearch-api/elasticsearch-api.gemspec", __FILE__)
-  gem 'elasticsearch-api', :path => File.expand_path("../../elasticsearch-api", __FILE__), :require => false
-end
+gem 'elasticsearch-api', :path => "../elasticsearch-api", :require => false
+gem 'elasticsearch-transport', :path => "../elasticsearch-transport", :require => false
+gem 'elasticsearch-extensions', :path => "../elasticsearch-extensions", :require => false
 
-if File.exists? File.expand_path("../../elasticsearch-transport/elasticsearch-transport.gemspec", __FILE__)
-  gem 'elasticsearch-transport', :path => File.expand_path("../../elasticsearch-transport", __FILE__), :require => false
-end
-
-if File.exists? File.expand_path("../../elasticsearch-extensions", __FILE__)
-  gem 'elasticsearch-extensions', :path => File.expand_path("../../elasticsearch-extensions", __FILE__), :require => false
-end
-
-if File.exists? File.expand_path("../../elasticsearch/elasticsearch.gemspec", __FILE__)
-  gem 'elasticsearch', :path => File.expand_path("../../elasticsearch/", __FILE__)
-end
+gem 'elasticsearch', :path => "../elasticsearch/"

elasticsearch-extensions/elasticsearch-extensions.gemspec

@@ -56,6 +56,6 @@ Gem::Specification.new do |s|
   end
 
   # Gems for testing integrations
-  s.add_development_dependency "patron"
-  s.add_development_dependency "oj"
+  s.add_development_dependency "patron" unless defined?(JRUBY_VERSION) || defined?(Rubinius)
+  s.add_development_dependency "oj" unless defined?(JRUBY_VERSION) || defined?(Rubinius)
 end

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,138 @@ 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)
+            @options = arguments[:options] || {}
+            @options[:http] ||= {}
+            @logger = options[:logger]
+            @adapter = Adapter.new(logger, options)
+            @healthcheck_path = options[:healthcheck_path] || "/"
+            @pool = Manticore::Pool.new(logger, @adapter, @healthcheck_path, (arguments[:hosts] || []), 5, self.host_unreachable_exceptions, options)
+            @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 }
+
+            setup_sniffing!
           end
 
-          # Should just be run once at startup
-          def build_client(options={})
-            client_options = options[:transport_options] || {}
-            client_options[:ssl] = options[:ssl] || {}
+          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
 
-            @manticore = ::Manticore::Client.new(client_options)
+          # 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
 
-          # 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)
+            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 ArgumentError.new "Method #{method} not supported"
+                raise e
               end
-              Response.new resp.code, resp.read_body, resp.headers
+            rescue ::Elasticsearch::Transport::Transport::HostUnreachableError => e
+              logger.error "[#{e.class}] #{e.message} #{e.url}" if logger
+
+              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
+
+                  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
 
-          # Builds and returns a collection of connections.
-          # Each connection is a Manticore::Client
-          #
-          # @return [Connections::Collection]
-          #
-          def __build_connections
-            @request_options = {}
-
-            if options.key?(:headers)
-              @request_options[:headers] = options[:headers]
+          def __close_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
 
-            Connections::Collection.new \
-              :connections => hosts.map { |host|
-                host[:protocol]   = host[:scheme] || DEFAULT_PROTOCOL
-                host[:port]     ||= DEFAULT_PORT
+          def enrich_response(method, url, path, params, body, response)
+            start = Time.now if logger || tracer
 
-                host.delete(:user)     # auth is not supported here.
-                host.delete(:password) # use the headers
+            duration = Time.now-start if logger || tracer
 
-                Connections::Connection.new \
-                  :host => host,
-                  :connection => @manticore
-              },
-              :selector_class => options[:selector_class],
-              :selector => options[:selector]
-          end
+            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
 
-          # 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.
+            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