Module: Elasticsearch::Transport::Transport::Base Abstract
- Included in:
- HTTP::Curb, HTTP::Faraday, HTTP::Manticore
- Defined in:
- lib/elasticsearch/transport/transport/base.rb
Overview
Module with common functionality for transport implementations.
Constant Summary collapse
- DEFAULT_PORT =
9200
- DEFAULT_PROTOCOL =
'http'
- DEFAULT_RELOAD_AFTER =
Requests
10_000
- DEFAULT_RESURRECT_AFTER =
Seconds
60
- DEFAULT_MAX_RETRIES =
Requests
3
- DEFAULT_SERIALIZER_CLASS =
Serializer::MultiJson
- SANITIZED_PASSWORD =
'*' * (rand(14)+1)
Instance Attribute Summary collapse
-
#connections ⇒ Object
readonly
Returns the value of attribute connections.
-
#counter ⇒ Object
readonly
Returns the value of attribute counter.
-
#hosts ⇒ Object
readonly
Returns the value of attribute hosts.
-
#last_request_at ⇒ Object
readonly
Returns the value of attribute last_request_at.
-
#logger ⇒ Object
Returns the value of attribute logger.
-
#max_retries ⇒ Object
Returns the value of attribute max_retries.
-
#options ⇒ Object
readonly
Returns the value of attribute options.
-
#protocol ⇒ Object
readonly
Returns the value of attribute protocol.
-
#reload_after ⇒ Object
Returns the value of attribute reload_after.
-
#reload_connections ⇒ Object
Returns the value of attribute reload_connections.
-
#resurrect_after ⇒ Object
Returns the value of attribute resurrect_after.
-
#serializer ⇒ Object
Returns the value of attribute serializer.
-
#sniffer ⇒ Object
Returns the value of attribute sniffer.
-
#tracer ⇒ Object
Returns the value of attribute tracer.
Instance Method Summary collapse
- #__build_connection(host, options = {}, block = nil) ⇒ Connections::Connection abstract private
-
#__build_connections ⇒ Connections::Collection
private
Builds and returns a collection of connections.
-
#__close_connections ⇒ Object
private
Closes the connections collection.
-
#__convert_to_json(o = nil, options = {}) ⇒ Object
private
Converts any non-String object to JSON.
-
#__full_url(host) ⇒ Object
private
Returns a full URL based on information from host.
-
#__log(method, path, params, body, url, response, json, took, duration) ⇒ Object
private
Log request and response information.
-
#__log_failed(response) ⇒ Object
private
Log failed request.
-
#__raise_transport_error(response) ⇒ Object
private
Raise error specific for the HTTP response status or a generic server error.
-
#__rebuild_connections(arguments = {}) ⇒ Connections::Collection
private
Rebuilds the connections collection in the transport.
-
#__trace(method, path, params, body, url, response, json, took, duration) ⇒ Object
private
Trace the request in the ‘curl` format.
-
#get_connection(options = {}) ⇒ Connections::Connection
Returns a connection from the connection pool by delegating to Connections::Collection#get_connection.
- #host_unreachable_exceptions ⇒ Array abstract
-
#initialize(arguments = {}, &block) ⇒ Object
Creates a new transport object.
-
#perform_request(method, path, params = {}, body = nil, &block) ⇒ Response
abstract
Performs a request to Elasticsearch, while handling logging, tracing, marking dead connections, retrying the request and reloading the connections.
-
#reload_connections! ⇒ Object
Reloads and replaces the connection collection based on cluster information.
-
#resurrect_dead_connections! ⇒ Object
Tries to “resurrect” all eligible dead connections.
Instance Attribute Details
#connections ⇒ Object (readonly)
Returns the value of attribute connections.
16 17 18 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 16 def connections @connections end |
#counter ⇒ Object (readonly)
Returns the value of attribute counter.
16 17 18 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 16 def counter @counter end |
#hosts ⇒ Object (readonly)
Returns the value of attribute hosts.
16 17 18 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 16 def hosts @hosts end |
#last_request_at ⇒ Object (readonly)
Returns the value of attribute last_request_at.
16 17 18 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 16 def last_request_at @last_request_at end |
#logger ⇒ Object
Returns the value of attribute logger.
17 18 19 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 17 def logger @logger end |
#max_retries ⇒ Object
Returns the value of attribute max_retries.
17 18 19 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 17 def max_retries @max_retries end |
#options ⇒ Object (readonly)
Returns the value of attribute options.
16 17 18 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 16 def @options end |
#protocol ⇒ Object (readonly)
Returns the value of attribute protocol.
16 17 18 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 16 def protocol @protocol end |
#reload_after ⇒ Object
Returns the value of attribute reload_after.
17 18 19 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 17 def reload_after @reload_after end |
#reload_connections ⇒ Object
Returns the value of attribute reload_connections.
17 18 19 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 17 def reload_connections @reload_connections end |
#resurrect_after ⇒ Object
Returns the value of attribute resurrect_after.
17 18 19 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 17 def resurrect_after @resurrect_after end |
#serializer ⇒ Object
Returns the value of attribute serializer.
17 18 19 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 17 def serializer @serializer end |
#sniffer ⇒ Object
Returns the value of attribute sniffer.
17 18 19 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 17 def sniffer @sniffer end |
#tracer ⇒ Object
Returns the value of attribute tracer.
17 18 19 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 17 def tracer @tracer end |
Instance Method Details
#__build_connection(host, options = {}, block = nil) ⇒ Connections::Connection
This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.
Build and return a connection. A transport implementation must implement this method. See HTTP::Faraday#__build_connection for an example.
150 151 152 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 150 def __build_connection(host, ={}, block=nil) raise NoMethodError, "Implement this method in your class" end |
#__build_connections ⇒ Connections::Collection
This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.
Builds and returns a collection of connections
The adapters have to implement the #__build_connection method.
127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 127 def __build_connections Connections::Collection.new \ :connections => hosts.map { |host| host[:protocol] = host[:scheme] || [:scheme] || [:http][:scheme] || DEFAULT_PROTOCOL host[:port] ||= [:port] || [:http][:scheme] || DEFAULT_PORT if ([:user] || [:http][:user]) && !host[:user] host[:user] ||= [:user] || [:http][:user] host[:password] ||= [:password] || [:http][:password] end __build_connection(host, ([:transport_options] || {}), @block) }, :selector_class => [:selector_class], :selector => [:selector] end |
#__close_connections ⇒ Object
This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.
Closes the connections collection
158 159 160 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 158 def __close_connections # A hook point for specific adapters when they need to close connections end |
#__convert_to_json(o = nil, options = {}) ⇒ Object
This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.
Converts any non-String object to JSON
208 209 210 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 208 def __convert_to_json(o=nil, ={}) o = o.is_a?(String) ? o : serializer.dump(o, ) end |
#__full_url(host) ⇒ Object
This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.
Returns a full URL based on information from host
217 218 219 220 221 222 223 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 217 def __full_url(host) url = "#{host[:protocol]}://" url += "#{CGI.escape(host[:user])}:#{CGI.escape(host[:password])}@" if host[:user] url += "#{host[:host]}:#{host[:port]}" url += "#{host[:path]}" if host[:path] url end |
#__log(method, path, params, body, url, response, json, took, duration) ⇒ Object
This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.
Log request and response information
166 167 168 169 170 171 172 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 166 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 |
#__log_failed(response) ⇒ Object
This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.
Log failed request
178 179 180 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 178 def __log_failed(response) logger.fatal "[#{response.status}] #{response.body}" end |
#__raise_transport_error(response) ⇒ Object
This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.
Raise error specific for the HTTP response status or a generic server error
199 200 201 202 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 199 def __raise_transport_error(response) error = ERRORS[response.status] || ServerError raise error.new "[#{response.status}] #{response.body}" end |
#__rebuild_connections(arguments = {}) ⇒ Connections::Collection
This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.
Rebuilds the connections collection in the transport.
The methods adds new connections from the passed hosts to the collection, and removes all connections not contained in the passed hosts.
103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 103 def __rebuild_connections(arguments={}) @state_mutex.synchronize do @hosts = arguments[:hosts] || [] @options = arguments[:options] || {} __close_connections new_connections = __build_connections stale_connections = @connections.all.select { |c| ! new_connections.include?(c) } new_connections = new_connections.reject { |c| @connections.include?(c) } @connections.remove(stale_connections) @connections.add(new_connections) @connections end end |
#__trace(method, path, params, body, url, response, json, took, duration) ⇒ Object
This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.
Trace the request in the ‘curl` format
186 187 188 189 190 191 192 193 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 186 def __trace(method, path, params, body, url, response, json, took, duration) trace_url = "http://localhost:9200/#{path}?pretty" + ( params.empty? ? '' : "&#{::Faraday::Utils::ParamsHash[params].to_query}" ) trace_body = body ? " -d '#{__convert_to_json(body, :pretty => true)}'" : '' tracer.info "curl -X #{method.to_s.upcase} '#{trace_url}'#{trace_body}\n" tracer.debug "# #{Time.now.iso8601} [#{response.status}] (#{format('%.3f', duration)}s)\n#" tracer.debug json ? serializer.dump(json, :pretty => true).gsub(/^/, '# ').sub(/\}$/, "\n# }")+"\n" : "# #{response.body}\n" end |
#get_connection(options = {}) ⇒ Connections::Connection
Returns a connection from the connection pool by delegating to Connections::Collection#get_connection.
Resurrects dead connection if the ‘resurrect_after` timeout has passed. Increments the counter and performs connection reloading if the `reload_connections` option is set.
66 67 68 69 70 71 72 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 66 def get_connection(={}) resurrect_dead_connections! if Time.now > @last_request_at + @resurrect_after @counter_mtx.synchronize { @counter += 1 } reload_connections! if reload_connections && counter % reload_after == 0 connections.get_connection() end |
#host_unreachable_exceptions ⇒ Array
Returns an Array of connection errors specific to the transport implementation. See HTTP::Faraday#host_unreachable_exceptions for an example.
331 332 333 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 331 def host_unreachable_exceptions [Errno::ECONNREFUSED] end |
#initialize(arguments = {}, &block) ⇒ Object
Creates a new transport object
31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 31 def initialize(arguments={}, &block) @state_mutex = Mutex.new @hosts = arguments[:hosts] || [] @options = arguments[:options] || {} @options[:http] ||= {} @block = block @connections = __build_connections @serializer = [:serializer] || ( [:serializer_class] ? [:serializer_class].new(self) : DEFAULT_SERIALIZER_CLASS.new(self) ) @protocol = [:protocol] || DEFAULT_PROTOCOL @logger = [:logger] @tracer = [:tracer] @sniffer = [:sniffer_class] ? [:sniffer_class].new(self) : Sniffer.new(self) @counter = 0 @counter_mtx = Mutex.new @last_request_at = Time.now @reload_connections = [:reload_connections] @reload_after = [:reload_connections].is_a?(Fixnum) ? [:reload_connections] : DEFAULT_RELOAD_AFTER @resurrect_after = [:resurrect_after] || DEFAULT_RESURRECT_AFTER @max_retries = [:retry_on_failure].is_a?(Fixnum) ? [:retry_on_failure] : DEFAULT_MAX_RETRIES @retry_on_status = Array([:retry_on_status]).map { |d| d.to_i } end |
#perform_request(method, path, params = {}, body = nil, &block) ⇒ Response
The transport implementation has to implement this method either in full, or by invoking this method with a block. See HTTP::Faraday#perform_request for an example.
Performs a request to Elasticsearch, while handling logging, tracing, marking dead connections, retrying the request and reloading the connections.
242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 242 def perform_request(method, path, params={}, body=nil, &block) raise NoMethodError, "Implement this method in your transport class" unless block_given? start = Time.now if logger || tracer tries = 0 begin tries += 1 connection = get_connection or raise Error.new("Cannot get new connection from pool.") if connection.connection.respond_to?(:params) && connection.connection.params.respond_to?(:to_hash) params = connection.connection.params.merge(params.to_hash) end url = connection.full_url(path, params) response = block.call(connection, url) connection.healthy! if connection.failures > 0 # 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 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 if tries <= max_retries retry else logger.fatal "[#{e.class}] Cannot get response from #{url} after #{tries} tries" if logger raise e end else raise e end rescue *host_unreachable_exceptions => e logger.error "[#{e.class}] #{e.} #{connection.host.inspect}" if logger connection.dead! if @options[:reload_on_failure] and tries < connections.all.size logger.warn "[#{e.class}] Reloading connections (attempt #{tries} of #{connections.all.size})" if logger reload_connections! and retry end if @options[:retry_on_failure] logger.warn "[#{e.class}] Attempt #{tries} connecting to #{connection.host.inspect}" if logger if tries <= max_retries retry else logger.fatal "[#{e.class}] Cannot connect to #{connection.host.inspect} after #{tries} tries" if logger raise e end else raise e end rescue Exception => e logger.fatal "[#{e.class}] #{e.} (#{connection.host.inspect if connection})" if logger raise e end #/begin 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 = serializer.load(response.body) if response.body && !response.body.empty? && response.headers && response.headers["content-type"] =~ /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 Response.new response.status, json || response.body, response.headers ensure @last_request_at = Time.now end |
#reload_connections! ⇒ Object
Reloads and replaces the connection collection based on cluster information
78 79 80 81 82 83 84 85 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 78 def reload_connections! hosts = sniffer.hosts __rebuild_connections :hosts => hosts, :options => self rescue SnifferTimeoutError logger.error "[SnifferTimeoutError] Timeout when reloading connections." if logger self end |
#resurrect_dead_connections! ⇒ Object
Tries to “resurrect” all eligible dead connections
91 92 93 |
# File 'lib/elasticsearch/transport/transport/base.rb', line 91 def resurrect_dead_connections! connections.dead.each { |c| c.resurrect! } end |