[Client-618] max connections enforced

.github/workflows/development.yml

@@ -6,19 +6,19 @@ jobs:
   test:
     runs-on: ${{matrix.os}}-latest
     continue-on-error: ${{matrix.experimental}}
-    
+
     strategy:
       matrix:
         os:
           - ubuntu
-        
+
         ruby:
           - 2.6
           - 2.7
-        
+
         experimental: [false]
         env: [""]
-        
+
         include:
           - os: ubuntu
             ruby: head
@@ -30,14 +30,14 @@ jobs:
       with:
         ruby-version: ${{matrix.ruby}}
         bundler-cache: true
-    
+
     - name: Start server
       timeout-minutes: 5
       env:
         TERM: dumb
       run:
         .github/workflows/start_cluster.sh 2
-    
+
     - name: Run tests
       timeout-minutes: 30
       env:

aerospike.gemspec

@@ -8,7 +8,7 @@ Gem::Specification.new do |s|
   s.name        = "aerospike"
   s.version     = Aerospike::VERSION
   s.platform    = Gem::Platform::RUBY
-  s.authors     = [ "Khosrow Afroozeh", "Jan Hecking" ]
+  s.authors     = ["Khosrow Afroozeh", "Jan Hecking", "Sachin Venkatesha Murthy"]
   s.email       = [ "[email protected]", "[email protected]" ]
   s.homepage    = "http://www.github.com/aerospike/aerospike-client-ruby"
   s.summary     = "An Aerospike driver for Ruby."

lib/aerospike/aerospike_exception.rb

@@ -96,5 +96,11 @@ def initialize(msg=nil)
         super(ResultCode::INVALID_NAMESPACE, msg)
       end
     end
+
+    class MaxConnectionsExceeded < Aerospike
+      def initialize(msg = nil)
+        super(ResultCode::MAX_CONNECTION_EXCEEDED, msg)
+      end
+    end
   end
 end

lib/aerospike/client.rb

@@ -225,7 +225,6 @@ def truncate(namespace, set_name = nil, before_last_update = nil, options = {})
       policy = create_policy(options, Policy, default_info_policy)
 
       node = @cluster.random_node
-      conn = node.get_connection(policy.timeout)
 
       if set_name && !set_name.to_s.strip.empty?
         str_cmd = "truncate:namespace=#{namespace}"

lib/aerospike/cluster.rb

@@ -321,14 +321,24 @@ def update_partitions(parser)
 
     def request_info(policy, *commands)
       node = random_node
-      conn = node.get_connection(policy.timeout)
+      begin
+        conn = node.get_connection(policy.timeout)
+      rescue => e
+        Aerospike.logger.error("Get connection failed with exception: #{e}")
+        raise e
+      end
       Info.request(conn, *commands).tap do
         node.put_connection(conn)
       end
     end
 
     def request_node_info(node, policy, *commands)
-      conn = node.get_connection(policy.timeout)
+      begin
+        conn = node.get_connection(policy.timeout)
+      rescue => e
+        Aerospike.logger.error("Get connection failed with exception: #{e}")
+        raise e
+      end
       Info.request(conn, *commands).tap do
         node.put_connection(conn)
       end

lib/aerospike/command/admin_command.rb

@@ -345,14 +345,13 @@ def execute_command(cluster, policy)
       timeout = 1
       timeout = policy.timeout if policy && policy.timeout > 0
 
-      conn = node.get_connection(timeout)
-
       begin
+        conn = node.get_connection(timeout)
         conn.write(@data_buffer, @data_offset)
         conn.read(@data_buffer, HEADER_SIZE)
         node.put_connection(conn)
       rescue => e
-        conn.close if conn
+        node.close_connection(conn) if conn
         raise e
       end
 
@@ -377,13 +376,11 @@ def read_users(cluster, policy)
         status, list = read_user_blocks(conn)
         node.put_connection(conn)
       rescue => e
-        conn.close if conn
+        node.close_connection(conn) if conn
         raise e
       end
-
       raise Exceptions::Aerospike.new(status) if status > 0
-
-      return list
+      list
     end
 
     def read_user_blocks(conn)
@@ -512,13 +509,13 @@ def read_roles(cluster, policy)
         status, list = read_role_blocks(conn)
         node.put_connection(conn)
       rescue => e
-        conn.close if conn
+        node.close_connection(conn) if conn
         raise e
       end
 
       raise Exceptions::Aerospike.new(status) if status > 0
 
-      return list
+      list
     end
 
     def read_role_blocks(conn)

lib/aerospike/command/command.rb

@@ -487,6 +487,9 @@ def execute
           @node = get_node
           @conn = @node.get_connection(@policy.timeout)
         rescue => e
+          if e.is_a?(Aerospike::Exceptions::MaxConnectionsExceeded)
+            Aerospike.logger.error("Maximum connections established. No new connection can be created. #{e}")
+          end
           if @node
             # Socket connection error has occurred. Decrease health and retry.
             @node.decrease_health
@@ -510,7 +513,7 @@ def execute
 
             # All runtime exceptions are considered fatal. Do not retry.
             # Close socket to flush out possible garbage. Do not put back in pool.
-            @conn.close if @conn
+            @node.close_connection(@conn) if @conn
             raise e
           end
 
@@ -523,7 +526,7 @@ def execute
           rescue => e
             # IO errors are considered temporary anomalies. Retry.
             # Close socket to flush out possible garbage. Do not put back in pool.
-            @conn.close if @conn
+            @node.close_connection(@conn) if @conn
 
             Aerospike.logger.error("Node #{@node.to_s}: #{e}")
             # IO error means connection to server @node is unhealthy.
@@ -548,7 +551,7 @@ def execute
             # cancelling/closing the batch/multi commands will return an error, which will
             # close the connection to throw away its data and signal the server about the
             # situation. We will not put back the connection in the buffer.
-            @conn.close if @conn
+            @node.close_connection(@conn) if @conn
             raise e
           end
 

lib/aerospike/connection/authenticate.rb

@@ -31,10 +31,11 @@ def call(conn, user, hashed_pass)
         end
       end
     end
+
     module AuthenticateNew
       class << self
         INVALID_SESSION_ERR = [ResultCode::INVALID_CREDENTIAL,
-          ResultCode::EXPIRED_SESSION]
+                               ResultCode::EXPIRED_SESSION]
 
         def call(conn, cluster)
           command = LoginCommand.new
@@ -65,4 +66,3 @@ def call(conn, cluster)
     end
   end
 end
-

lib/aerospike/node.rb

@@ -84,14 +84,14 @@ def get_connection(timeout)
     # Put back a connection to the cache. If cache is full, the connection will be
     # closed and discarded
     def put_connection(conn)
-      conn.close if !active?
+      @connections.cleanup(conn) unless active?
       @connections.offer(conn)
     end
 
     # Separate connection for refreshing
     def tend_connection
       if @tend_connection.nil? || @tend_connection.closed?
-        @tend_connection = Cluster::CreateConnection.(cluster, host)
+        @tend_connection = @connections.create
       end
       @tend_connection
     end
@@ -177,7 +177,7 @@ def aliases
       @aliases.value
     end
 
-    # Marks node as inactice and closes all cached connections
+    # Marks node as inactive and closes all cached connections
     def close
       inactive!
       close_connections
@@ -224,14 +224,18 @@ def refresh_reset
       Node::Refresh::Reset.(self)
     end
 
+    def close_connection(conn)
+      @connections.cleanup(conn)
+    end
+
     private
 
     def close_connections
-      @tend_connection.close if @tend_connection
+      @connections.cleanup(@tend_connection) if @tend_connection
       # drain connections and close all of them
       # non-blocking, does not call create_block when passed false
-      while conn = @connections.poll(false)
-        conn.close if conn
+      while (conn = @connections.poll(create_new: false))
+        @connections.cleanup(conn)
       end
     end
   end # class Node

lib/aerospike/node/refresh/info.rb

@@ -51,7 +51,7 @@ def call(node, peers)
             peers.refresh_count += 1
             node.reset_failures!
           rescue ::Aerospike::Exceptions::Aerospike => e
-            conn.close if conn
+            node.close_connection(conn) if conn
             node.decrease_health
             peers.generation_changed = true if peers.use_peers?
             Refresh::Failed.(node, e)

lib/aerospike/node/refresh/partitions.rb

@@ -30,7 +30,7 @@ def call(node, peers)
             parser = PartitionParser.new(node, conn)
             node.cluster.update_partitions(parser)
           rescue ::Aerospike::Exceptions::Aerospike => e
-            conn.close
+            node.close_connection(conn)
             Refresh::Failed.(node, e)
           end
 

lib/aerospike/node/refresh/racks.rb

@@ -30,7 +30,7 @@ def call(node)
             parser = RackParser.new(node, conn)
             node.update_racks(parser)
           rescue ::Aerospike::Exceptions::Aerospike => e
-            conn.close
+            node.close_connection(conn)
             Refresh::Failed.(node, e)
           end
 

lib/aerospike/node_validator.rb

@@ -101,4 +101,4 @@ def is_ip?(hostname)
     end
 
   end # class
-end # module
+end # module

lib/aerospike/result_code.rb

@@ -20,6 +20,9 @@ module ResultCode
 
     attr_reader :code
 
+    #Maximum connections established. New connections cannot be created.
+    MAX_CONNECTION_EXCEEDED = -21
+
     # One or more keys failed in a batch.
     BATCH_FAILED = -20
 
@@ -299,6 +302,7 @@ module ResultCode
     # Internal error.
     QUERY_DUPLICATE = 215
 
+
     def self.message(code)
       case code
       when BATCH_FAILED
@@ -577,6 +581,9 @@ def self.message(code)
       when QUERY_DUPLICATE
         "Internal query error"
 
+      when MAX_CONNECTION_EXCEEDED
+        "Maximum new connections exceeded."
+
       else
         "ResultCode #{code} unknown in the client. Please file a github issue."
       end # case

lib/aerospike/task/execute_task.rb

@@ -45,8 +45,12 @@ def all_nodes_done?
         elsif node.supports_feature?(Aerospike::Features::QUERY_SHOW)
           command = cmd2
         end
-
-        conn = node.get_connection(0)
+        begin
+          conn = node.get_connection(0)
+        rescue => e
+          Aerospike.logger.error("Get connection failed with exception: #{e}")
+          raise e
+        end
         responseMap, _ = Info.request(conn, command)
         node.put_connection(conn)
 

lib/aerospike/task/index_task.rb

@@ -42,7 +42,12 @@ def all_nodes_done?
       nodes = @cluster.nodes
 
       nodes.each do |node|
-        conn = node.get_connection(1)
+        begin
+          conn = node.get_connection(1)
+        rescue => e
+          Aerospike.logger.error("Get connection failed with exception: #{e}")
+          raise e
+        end
         response_map = Info.request(conn, command)
         _, response = response_map.first
         match = response.to_s.match(MATCHER)

lib/aerospike/task/udf_register_task.rb

@@ -39,15 +39,20 @@ def all_nodes_done?
       nodes = @cluster.nodes
 
       nodes.each do |node|
-        conn = node.get_connection(1)
+        begin
+          conn = node.get_connection(1)
+        rescue => e
+          Aerospike.logger.error("Get connection failed with exception: #{e}")
+          raise e
+        end
         response_map = Info.request(conn, command)
         _, response = response_map.first
         index = response.to_s.index("filename=#{@package_name}")
 
         return false if index.nil?
       end
 
-      return true
+      true
     end
 
   end # class

lib/aerospike/task/udf_remove_task.rb

@@ -39,7 +39,12 @@ def all_nodes_done?
       nodes = @cluster.nodes
 
       nodes.each do |node|
-        conn = node.get_connection(1)
+        begin
+          conn = node.get_connection(1)
+        rescue => e
+          Aerospike.logger.error("Get connection failed with exception: #{e}")
+          raise e
+        end
         response_map = Info.request(conn, command)
         _, response = response_map.first
         index = response.to_s.index("filename=#{@package_name}")