Fixes #39291 - Add persistent HTTP connection pooling for Candlepin

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

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

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)

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"

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 => '<html>404</html>', :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