Merge pull request #3170 from sap-contributions/bbs-failover

Use Net::HTTP::Persistent in Diego client

Author: philippthun

Gemfile

@@ -21,6 +21,7 @@ gem 'membrane', '~> 1.0'
 gem 'mime-types', '~> 3.4'
 gem 'multi_json'
 gem 'multipart-parser'
+gem 'net-http-persistent'
 gem 'net-ssh'
 gem 'netaddr', '>= 2.0.4'
 gem 'newrelic_rpm'

Gemfile.lock

@@ -120,6 +120,7 @@ GEM
       simplecov (<= 0.13)
     coderay (1.1.3)
     concurrent-ruby (1.2.0)
+    connection_pool (2.3.0)
     cookiejar (0.3.3)
     crack (0.4.5)
       rexml
@@ -326,6 +327,8 @@ GEM
     mustermann (3.0.0)
       ruby2_keywords (~> 0.0.1)
     mysql2 (0.5.5)
+    net-http-persistent (4.0.1)
+      connection_pool (~> 2.2)
     net-ssh (7.0.1)
     netaddr (2.0.6)
     netrc (0.11.0)
@@ -584,6 +587,7 @@ DEPENDENCIES
   multi_json
   multipart-parser
   mysql2 (~> 0.5.5)
+  net-http-persistent
   net-ssh
   netaddr (>= 2.0.4)
   newrelic_rpm

lib/cloud_controller/deployment_updater/dispatcher.rb

@@ -15,7 +15,7 @@ def dispatch
           deployments_to_cancel = DeploymentModel.where(state: DeploymentModel::CANCELING_STATE).all
 
           begin
-            workpool = WorkPool.new(50)
+            workpool = WorkPool.new
 
             logger.info("scaling #{deployments_to_scale.size} deployments")
             deployments_to_scale.each do |deployment|

lib/cloud_controller/diego/processes_sync.rb

@@ -12,7 +12,7 @@ class BBSFetchError < Error
 
       def initialize(config:, statsd_updater: VCAP::CloudController::Metrics::StatsdUpdater.new)
         @config   = config
-        @workpool = WorkPool.new(50, store_exceptions: true)
+        @workpool = WorkPool.new(store_exceptions: true)
         @statsd_updater = statsd_updater
       end
 

lib/cloud_controller/diego/reporters/instances_reporter.rb

@@ -14,7 +14,7 @@ def initialize(bbs_instances_client)
       end
 
       def self.singleton_workpool
-        @singleton_workpool ||= WorkPool.new(50)
+        @singleton_workpool ||= WorkPool.new
       end
 
       def all_instances_for_app(process)

lib/cloud_controller/diego/tasks_sync.rb

@@ -10,7 +10,7 @@ class BBSFetchError < Error
 
       def initialize(config:)
         @config   = config
-        @workpool = WorkPool.new(50, store_exceptions: true)
+        @workpool = WorkPool.new(store_exceptions: true)
       end
 
       def sync

lib/diego/client.rb

@@ -1,16 +1,16 @@
 require 'diego/bbs/bbs'
 require 'diego/errors'
 require 'diego/routes'
+require 'net/http/persistent'
+require 'uri'
 
 module Diego
   class Client
-    PROTOBUF_HEADER = { 'Content-Type'.freeze => 'application/x-protobuf'.freeze }.freeze
-
     def initialize(url:, ca_cert_file:, client_cert_file:, client_key_file:,
       connect_timeout:, send_timeout:, receive_timeout:)
       ENV['PB_IGNORE_DEPRECATIONS'] ||= 'true'
-      @client = build_client(
-        url,
+      @bbs_url = URI(url)
+      @http_client = new_http_client(
         ca_cert_file,
         client_cert_file,
         client_key_file,
@@ -20,157 +20,121 @@ def initialize(url:, ca_cert_file:, client_cert_file:, client_key_file:,
     end
 
     def ping
-      response = with_request_error_handling do
-        client.post(Routes::PING)
-      end
+      req = post_request(path: Routes::PING)
+      response = request_with_error_handling(req)
 
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::PingResponse)
     end
 
     def upsert_domain(domain:, ttl:)
-      request = protobuf_encode!({ domain: domain, ttl: ttl.to_i }, Bbs::Models::UpsertDomainRequest)
-
-      response = with_request_error_handling do
-        client.post(Routes::UPSERT_DOMAIN, request, PROTOBUF_HEADER)
-      end
+      req = post_request(body: protobuf_encode!({ domain: domain, ttl: ttl.to_i }, Bbs::Models::UpsertDomainRequest), path: Routes::UPSERT_DOMAIN)
+      response = request_with_error_handling(req)
 
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::UpsertDomainResponse)
     end
 
     def desire_task(task_definition:, domain:, task_guid:)
-      request = protobuf_encode!({ task_definition: task_definition, domain: domain, task_guid: task_guid }, Bbs::Models::DesireTaskRequest)
+      req = post_request(body: protobuf_encode!({ task_definition: task_definition, domain: domain, task_guid: task_guid }, Bbs::Models::DesireTaskRequest),
+path: Routes::DESIRE_TASK)
+      response = request_with_error_handling(req)
 
-      response = with_request_error_handling do
-        client.post(Routes::DESIRE_TASK, request, PROTOBUF_HEADER)
-      end
-
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::TaskLifecycleResponse)
     end
 
     def task_by_guid(task_guid)
-      request = protobuf_encode!({ task_guid: task_guid }, Bbs::Models::TaskByGuidRequest)
-
-      response = with_request_error_handling do
-        client.post(Routes::TASK_BY_GUID, request, PROTOBUF_HEADER)
-      end
+      req = post_request(body: protobuf_encode!({ task_guid: task_guid }, Bbs::Models::TaskByGuidRequest), path: Routes::TASK_BY_GUID)
+      response = request_with_error_handling(req)
 
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::TaskResponse)
     end
 
     def tasks(domain: '', cell_id: '')
-      request = protobuf_encode!({ domain: domain, cell_id: cell_id }, Bbs::Models::TasksRequest)
+      req = post_request(body: protobuf_encode!({ domain: domain, cell_id: cell_id }, Bbs::Models::TasksRequest), path: Routes::LIST_TASKS)
+      response = request_with_error_handling(req)
 
-      response = with_request_error_handling do
-        client.post(Routes::LIST_TASKS, request, PROTOBUF_HEADER)
-      end
-
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::TasksResponse)
     end
 
     def cancel_task(task_guid)
-      request = protobuf_encode!({ task_guid: task_guid }, Bbs::Models::TaskGuidRequest)
-
-      response = with_request_error_handling do
-        client.post(Routes::CANCEL_TASK, request, PROTOBUF_HEADER)
-      end
+      req = post_request(body: protobuf_encode!({ task_guid: task_guid }, Bbs::Models::TaskGuidRequest), path: Routes::CANCEL_TASK)
+      response = request_with_error_handling(req)
 
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::TaskLifecycleResponse)
     end
 
     def desire_lrp(lrp)
-      request = protobuf_encode!({ desired_lrp: lrp }, Bbs::Models::DesireLRPRequest)
+      req = post_request(body: protobuf_encode!({ desired_lrp: lrp }, Bbs::Models::DesireLRPRequest), path: Routes::DESIRE_LRP)
+      response = request_with_error_handling(req)
 
-      response = with_request_error_handling do
-        client.post(Routes::DESIRE_LRP, request, PROTOBUF_HEADER)
-      end
-
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::DesiredLRPLifecycleResponse)
     end
 
     def desired_lrp_by_process_guid(process_guid)
-      request = protobuf_encode!({ process_guid: process_guid }, Bbs::Models::DesiredLRPByProcessGuidRequest)
-
-      response = with_request_error_handling do
-        client.post(Routes::DESIRED_LRP_BY_PROCESS_GUID, request, PROTOBUF_HEADER)
-      end
+      req = post_request(body: protobuf_encode!({ process_guid: process_guid }, Bbs::Models::DesiredLRPByProcessGuidRequest), path: Routes::DESIRED_LRP_BY_PROCESS_GUID)
+      response = request_with_error_handling(req)
 
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::DesiredLRPResponse)
     end
 
     def update_desired_lrp(process_guid, lrp_update)
-      request = protobuf_encode!({ process_guid: process_guid, update: lrp_update }, Bbs::Models::UpdateDesiredLRPRequest)
+      req = post_request(body: protobuf_encode!({ process_guid: process_guid, update: lrp_update }, Bbs::Models::UpdateDesiredLRPRequest), path: Routes::UPDATE_DESIRED_LRP)
+      response = request_with_error_handling(req)
 
-      response = with_request_error_handling do
-        client.post(Routes::UPDATE_DESIRED_LRP, request, PROTOBUF_HEADER)
-      end
-
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::DesiredLRPLifecycleResponse)
     end
 
     def remove_desired_lrp(process_guid)
-      request = protobuf_encode!({ process_guid: process_guid }, Bbs::Models::RemoveDesiredLRPRequest)
-
-      response = with_request_error_handling do
-        client.post(Routes::REMOVE_DESIRED_LRP, request, PROTOBUF_HEADER)
-      end
+      req = post_request(body: protobuf_encode!({ process_guid: process_guid }, Bbs::Models::RemoveDesiredLRPRequest), path: Routes::REMOVE_DESIRED_LRP)
+      response = request_with_error_handling(req)
 
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::DesiredLRPLifecycleResponse)
     end
 
     def retire_actual_lrp(actual_lrp_key)
-      request = protobuf_encode!({ actual_lrp_key: actual_lrp_key }, Bbs::Models::RetireActualLRPRequest)
+      req = post_request(body: protobuf_encode!({ actual_lrp_key: actual_lrp_key }, Bbs::Models::RetireActualLRPRequest), path: Routes::RETIRE_ACTUAL_LRP)
+      response = request_with_error_handling(req)
 
-      response = with_request_error_handling do
-        client.post(Routes::RETIRE_ACTUAL_LRP, request, PROTOBUF_HEADER)
-      end
-
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::ActualLRPLifecycleResponse)
     end
 
     def desired_lrp_scheduling_infos(domain)
-      request = protobuf_encode!({ domain: domain }, Bbs::Models::DesiredLRPsRequest)
-
-      response = with_request_error_handling do
-        client.post(Routes::DESIRED_LRP_SCHEDULING_INFOS, request, PROTOBUF_HEADER)
-      end
+      req = post_request(body: protobuf_encode!({ domain: domain }, Bbs::Models::DesiredLRPsRequest), path: Routes::DESIRED_LRP_SCHEDULING_INFOS)
+      response = request_with_error_handling(req)
 
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::DesiredLRPSchedulingInfosResponse)
     end
 
     def actual_lrps_by_process_guid(process_guid)
-      request = protobuf_encode!({ process_guid: process_guid }, Bbs::Models::ActualLRPsRequest)
+      req = post_request(body: protobuf_encode!({ process_guid: process_guid }, Bbs::Models::ActualLRPsRequest), path: Routes::ACTUAL_LRPS)
+      response = request_with_error_handling(req)
 
-      response = with_request_error_handling do
-        client.post(Routes::ACTUAL_LRPS, request, PROTOBUF_HEADER)
-      end
-
-      validate_status!(response: response, statuses: [200])
+      validate_status!(response)
       protobuf_decode!(response.body, Bbs::Models::ActualLRPsResponse)
     end
 
-    def with_request_error_handling(&blk)
-      tries ||= 3
-      yield
+    def request_with_error_handling(req)
+      attempt ||= 1
+      http_client.request(bbs_url + req.path, req)
     rescue => e
-      retry unless (tries -= 1).zero?
+      retry unless (attempt += 1) > 3
       raise RequestError.new(e.message)
     end
 
     private
 
-    attr_reader :client
+    attr_reader :http_client, :bbs_url
 
     def protobuf_encode!(hash, protobuf_message_class)
       # See below link to understand proto3 message encoding
@@ -180,8 +144,15 @@ def protobuf_encode!(hash, protobuf_message_class)
       raise EncodeError.new(e.message)
     end
 
-    def validate_status!(response:, statuses:)
-      raise ResponseError.new("failed with status: #{response.status}, body: #{response.body}") unless statuses.include?(response.status)
+    def post_request(body: nil, path:)
+      req = Net::HTTP::Post.new(path)
+      req.body = body if body
+      req['Content-Type'.freeze] = 'application/x-protobuf'.freeze
+      req
+    end
+
+    def validate_status!(response)
+      raise ResponseError.new("failed with status: #{response.code}, body: #{response.body}") unless response.code == '200'
     end
 
     def protobuf_decode!(message, protobuf_decoder)
@@ -190,14 +161,16 @@ def protobuf_decode!(message, protobuf_decoder)
       raise DecodeError.new(e.message)
     end
 
-    def build_client(url, ca_cert_file, client_cert_file, client_key_file,
-      connect_timeout, send_timeout, receive_timeout)
-      client                 = HTTPClient.new(base_url: url)
-      client.connect_timeout = connect_timeout
-      client.send_timeout    = send_timeout
-      client.receive_timeout = receive_timeout
-      client.ssl_config.set_client_cert_file(client_cert_file, client_key_file)
-      client.ssl_config.set_trust_ca(ca_cert_file)
+    def new_http_client(ca_cert_file, client_cert_file, client_key_file,
+                        connect_timeout, send_timeout, receive_timeout)
+      client = Net::HTTP::Persistent.new(pool_size: WorkPool::SIZE)
+      client.verify_mode = OpenSSL::SSL::VERIFY_PEER
+      client.private_key = OpenSSL::PKey::RSA.new(File.read(client_key_file))
+      client.certificate = OpenSSL::X509::Certificate.new(File.read(client_cert_file))
+      client.ca_file = ca_cert_file
+      client.open_timeout = connect_timeout
+      client.read_timeout = receive_timeout
+      client.write_timeout = send_timeout
       client
     end
   end

lib/utils/workpool.rb

@@ -1,7 +1,9 @@
 class WorkPool
+  SIZE = 50
+
   attr_reader :exceptions, :threads
 
-  def initialize(size, store_exceptions: false)
+  def initialize(size: SIZE, store_exceptions: false)
     @size = size
     @store_exceptions = store_exceptions