diff --git a/app/lib/katello/resources/candlepin.rb b/app/lib/katello/resources/candlepin.rb index 68bf03ddfc9..f84b626a950 100644 --- a/app/lib/katello/resources/candlepin.rb +++ b/app/lib/katello/resources/candlepin.rb @@ -4,6 +4,8 @@ module Candlepin TOTAL_COUNT_HEADER = :x_total_count # as parsed by rest_client class CandlepinResource < HttpResource + include Candlepin::PersistentConnection + cfg = SETTINGS[:katello][:candlepin] url = cfg[:url] uri = URI.parse(url) @@ -83,6 +85,7 @@ def self.fetch_paged(page_size = -1) class UpstreamCandlepinResource < CandlepinResource extend ::Katello::Util::HttpProxy + self.use_persistent_connection = false self.prefix = '/subscription' class << self diff --git a/app/lib/katello/resources/candlepin/persistent_connection.rb b/app/lib/katello/resources/candlepin/persistent_connection.rb new file mode 100644 index 00000000000..5df55a9c871 --- /dev/null +++ b/app/lib/katello/resources/candlepin/persistent_connection.rb @@ -0,0 +1,129 @@ +require 'net/http/persistent' +require 'ostruct' + +module Katello + module Resources + module Candlepin + # Provides persistent HTTP connection pooling for CandlepinResource. + # + # Net::HTTP::Persistent maintains one TCP+TLS connection per thread, + # reused across all Candlepin calls within that thread. Stale connections + # are detected and re-established automatically by the library. + # + # Include in CandlepinResource; set `self.use_persistent_connection = false` + # in subclasses that connect to different servers (e.g. UpstreamCandlepinResource). + module PersistentConnection + extend ActiveSupport::Concern + + # Response wrapper that preserves the RestClient::Response contract. + # + # Inherits from String (like RestClient::Response) so that + # MultiJson.load(exception.response) works in CandlepinError.from_exception. + class PooledResponse < String + attr_reader :code, :headers, :request + + def initialize(net_response, url:) + super(net_response.body || '') + @code = net_response.code.to_i + @headers = {} + net_response.each_header do |k, v| + @headers[k.downcase.tr('-', '_').to_sym] = v + end + @request = OpenStruct.new(url: url) + end + + def body + to_s + end + end + + # Translate RestClient-style symbol values to proper MIME types. + # RestClient converts {:accept => :json} to "Accept: application/json", + # but Net::HTTP passes symbols literally ("Accept: json") which Candlepin rejects. + CONTENT_TYPE_MAP = {json: 'application/json', xml: 'application/xml'}.freeze + + included do + class_attribute :use_persistent_connection, default: true + end + + class_methods do + def persistent_http + @persistent_http ||= begin + http = Net::HTTP::Persistent.new(name: 'candlepin') + http.idle_timeout = 30 + http.max_requests = 500 + http.ca_file = ssl_ca_file if ssl_ca_file + http.open_timeout = SETTINGS[:katello][:rest_client_timeout] + http.read_timeout = SETTINGS[:katello][:rest_client_timeout] + http + end + end + + def issue_request(method:, path:, headers: {}, payload: nil) + return super unless use_persistent_connection + + url = self.site + path + uri = URI.parse(url) + + request = Katello::HttpResource::REQUEST_MAP[method].new(uri) + + # OAuth signing — same consumer/secret as rest_client(), applied to the actual request + consumer = OAuth::Consumer.new( + self.consumer_key, self.consumer_secret, + site: self.site, http_method: method, + request_token_path: '', authorize_path: '', access_token_path: '' + ) + consumer.sign!(request) + + (headers || {}).each do |k, v| + request[k.to_s] = CONTENT_TYPE_MAP[v] || v.to_s + end + + if payload + request.body = payload + request.content_type = 'application/json' unless request['Content-Type'] + end + + logger.debug("Pooled Candlepin #{method.upcase} request: #{path}") + begin + logger.debug "Body: #{filter_sensitive_data(payload.to_json)}" + rescue JSON::GeneratorError, Encoding::UndefinedConversionError + logger.debug "Body: Error: could not render payload as json" + end if payload + + net_response = persistent_http.request(uri, request) + response = PooledResponse.new(net_response, url: url) + + if response.code >= 400 + raise_pooled_error(response, path, method) + end + + process_response(response) + rescue Errno::ECONNREFUSED + service = path.split("/").second + raise Errors::ConnectionRefusedException, + _("A backend service [ %s ] is unreachable") % service.capitalize + rescue Net::HTTP::Persistent::Error => e + raise Errors::ConnectionRefusedException, + _("A backend service [ Candlepin ] is unreachable: %s") % e.message + end + + private + + def raise_pooled_error(response, path, method) + unless response.headers[:x_version] + fail ::Katello::Errors::CandlepinNotRunning + end + + exception_class = RestClient::Exceptions::EXCEPTIONS_MAP.fetch( + response.code, RestClient::ExceptionWithResponse + ) + exception = exception_class.new(response, response.code) + + raise_rest_client_exception(exception, path, method.to_s.upcase) + end + end + end + end + end +end diff --git a/app/lib/katello/resources/candlepin/proxy.rb b/app/lib/katello/resources/candlepin/proxy.rb index 847bf80c87f..feccda63019 100644 --- a/app/lib/katello/resources/candlepin/proxy.rb +++ b/app/lib/katello/resources/candlepin/proxy.rb @@ -8,30 +8,32 @@ def self.logger def self.post(path, body) logger.debug "Sending POST request to Candlepin: #{path}" - client = CandlepinResource.rest_client(Net::HTTP::Post, :post, path_with_cp_prefix(path)) - client.post body, {:accept => :json, :content_type => :json}.merge(User.cp_oauth_header) + CandlepinResource.post(path_with_cp_prefix(path), body, + {:accept => :json, :content_type => :json}.merge(User.cp_oauth_header)) end def self.delete(path, body = nil) logger.debug "Sending DELETE request to Candlepin: #{path}" - client = CandlepinResource.rest_client(Net::HTTP::Delete, :delete, path_with_cp_prefix(path)) - # Some candlepin calls will set the body in DELETE requests. - client.options[:payload] = body unless body.nil? - client.delete({:accept => :json, :content_type => :json}.merge(User.cp_oauth_header)) + headers = {:accept => :json, :content_type => :json}.merge(User.cp_oauth_header) + if body + CandlepinResource.delete(path_with_cp_prefix(path), body, headers) + else + CandlepinResource.delete(path_with_cp_prefix(path), headers) + end end def self.get(path, extra_headers = {}) logger.debug "Sending GET request to Candlepin: #{path}" - client = CandlepinResource.rest_client(Net::HTTP::Get, :get, path_with_cp_prefix(path)) - client.get(extra_headers.merge!(default_request_headers)) + CandlepinResource.get(path_with_cp_prefix(path), + extra_headers.merge!(default_request_headers)) rescue RestClient::NotModified => e e.response end def self.put(path, body) logger.debug "Sending PUT request to Candlepin: #{path}" - client = CandlepinResource.rest_client(Net::HTTP::Put, :put, path_with_cp_prefix(path)) - client.put body, {:accept => :json, :content_type => :json}.merge(User.cp_oauth_header) + CandlepinResource.put(path_with_cp_prefix(path), body, + {:accept => :json, :content_type => :json}.merge(User.cp_oauth_header)) end def self.path_with_cp_prefix(path) diff --git a/katello.gemspec b/katello.gemspec index 1ae8a13f8b3..22a7acba916 100644 --- a/katello.gemspec +++ b/katello.gemspec @@ -31,6 +31,7 @@ Gem::Specification.new do |gem| gem.add_dependency "json" gem.add_dependency "oauth" gem.add_dependency "rest-client" + gem.add_dependency "net-http-persistent", ">= 4.0" gem.add_dependency "rabl" gem.add_dependency "foreman-tasks", ">= 12.0.0" diff --git a/test/lib/resources/persistent_connection_test.rb b/test/lib/resources/persistent_connection_test.rb new file mode 100644 index 00000000000..cc18fa2aff0 --- /dev/null +++ b/test/lib/resources/persistent_connection_test.rb @@ -0,0 +1,143 @@ +require 'katello_test_helper' + +module Katello + module Resources + module Candlepin + class PooledResponseTest < ActiveSupport::TestCase + def setup + @net_response = stub( + :body => '{"id":"abc","name":"test"}', + :code => '200' + ) + @net_response.stubs(:each_header).multiple_yields( + ['X-Candlepin-Request-Uuid', 'req-123'], + ['X-Version', '4.3.1'], + ['Content-Type', 'application/json'] + ) + @response = PersistentConnection::PooledResponse.new(@net_response, url: 'https://localhost:23443/candlepin/consumers/uuid-1') + end + + def test_body_returns_response_body + assert_equal '{"id":"abc","name":"test"}', @response.body + end + + def test_code_returns_integer + assert_equal 200, @response.code + assert_kind_of Integer, @response.code + end + + def test_headers_uses_symbol_keys_with_underscores + assert_equal 'req-123', @response.headers[:x_candlepin_request_uuid] + assert_equal '4.3.1', @response.headers[:x_version] + assert_equal 'application/json', @response.headers[:content_type] + end + + def test_request_url + assert_equal 'https://localhost:23443/candlepin/consumers/uuid-1', @response.request.url + end + + def test_inherits_from_string + assert_kind_of String, @response + assert_equal '{"id":"abc","name":"test"}', @response.to_s + end + + def test_json_parseable_as_string + parsed = JSON.parse(@response) + assert_equal 'abc', parsed['id'] + end + + def test_code_supports_integer_comparison + assert @response.code < 400 + error_response = PersistentConnection::PooledResponse.new( + stub(:body => '{}', :code => '404').tap { |r| r.stubs(:each_header) }, + url: 'https://localhost/test' + ) + assert error_response.code >= 400 + end + end + + class PersistentConnectionIssueRequestTest < ActiveSupport::TestCase + def setup + @success_response = stub(:body => '{"status":"ok"}', :code => '200') + @success_response.stubs(:each_header).multiple_yields( + ['X-Candlepin-Request-Uuid', 'req-456'], + ['X-Version', '4.3.1'] + ) + end + + def test_successful_request_returns_pooled_response + mock_http = mock('persistent_http') + mock_http.expects(:request).returns(@success_response) + CandlepinResource.stubs(:persistent_http).returns(mock_http) + + result = CandlepinResource.issue_request(method: :get, path: '/candlepin/status', headers: {}) + + assert_kind_of PersistentConnection::PooledResponse, result + assert_equal 200, result.code + assert_equal '{"status":"ok"}', result.body + end + + def test_4xx_raises_restclient_exception + error_response = stub(:body => '{"displayMessage":"Not found"}', :code => '404') + error_response.stubs(:each_header).multiple_yields(['X-Version', '4.3.1']) + mock_http = mock('persistent_http') + mock_http.expects(:request).returns(error_response) + CandlepinResource.stubs(:persistent_http).returns(mock_http) + + assert_raises(RestClient::Exception) do + CandlepinResource.issue_request(method: :get, path: '/candlepin/missing', headers: {}) + end + end + + def test_404_without_x_version_raises_candlepin_not_running + error_response = stub(:body => '404', :code => '404') + error_response.stubs(:each_header) + mock_http = mock('persistent_http') + mock_http.expects(:request).returns(error_response) + CandlepinResource.stubs(:persistent_http).returns(mock_http) + + assert_raises(Katello::Errors::CandlepinNotRunning) do + CandlepinResource.issue_request(method: :get, path: '/candlepin/status', headers: {}) + end + end + + def test_connection_refused_raises_connection_refused_exception + mock_http = mock('persistent_http') + mock_http.expects(:request).raises(Errno::ECONNREFUSED) + CandlepinResource.stubs(:persistent_http).returns(mock_http) + + assert_raises(Katello::Errors::ConnectionRefusedException) do + CandlepinResource.issue_request(method: :get, path: '/candlepin/status', headers: {}) + end + end + + def test_persistent_error_raises_connection_refused_exception + mock_http = mock('persistent_http') + mock_http.expects(:request).raises(Net::HTTP::Persistent::Error.new('too many connection resets')) + CandlepinResource.stubs(:persistent_http).returns(mock_http) + + assert_raises(Katello::Errors::ConnectionRefusedException) do + CandlepinResource.issue_request(method: :get, path: '/candlepin/status', headers: {}) + end + end + + def test_symbol_header_values_are_translated + mock_http = mock('persistent_http') + mock_http.expects(:request).with do |_uri, req| + req['accept'] == 'application/json' && req['content_type'] == 'application/json' + end.returns(@success_response) + CandlepinResource.stubs(:persistent_http).returns(mock_http) + + CandlepinResource.issue_request( + method: :get, path: '/candlepin/status', + headers: {accept: :json, content_type: :json} + ) + end + + def test_upstream_resource_bypasses_pooling + assert_equal false, UpstreamCandlepinResource.use_persistent_connection + end + end + end + end +end