Merge pull request #197 from datastax/RUBY-256

RUBY-256: Execution profiles

Author: stamhankar999

CHANGELOG.md

@@ -3,6 +3,7 @@ Features:
 * Do not mark a host as down if there are active connections.
 * Update Keyspace metadata to include collection of indexes defined in the keyspace.
 * Update Table metadata to include trigger-collection and view-collection metadata.
+* Added execution profiles to encapsulate a group of request execution options.
 
 Bug Fixes:
 * [RUBY-255](https://datastax-oss.atlassian.net/browse/RUBY-255) ControlConnection.peer_ip ignores peers that are missing critical information in system.peers.

features/basics/execution_profiles.feature

@@ -0,0 +1,65 @@
+Feature: Execution profiles
+
+  Execution profiles allow a user to group various execution options into a 'profile'. A user can then execute
+  statements with different profiles by specifying the profile name.
+
+  Background:
+    Given a running cassandra cluster
+
+  Scenario: Configure different load balancing policies with profiles.
+    Given the following example:
+      """ruby
+      require 'cassandra'
+
+      include Cassandra::LoadBalancing::Policies
+      profiles = {
+          p1: Cassandra::Execution::Profile.new(load_balancing_policy: WhiteList.new(['127.0.0.1'], RoundRobin.new)),
+          p2: Cassandra::Execution::Profile.new(load_balancing_policy: WhiteList.new(['127.0.0.2'], RoundRobin.new))
+      }
+
+      cluster = Cassandra.cluster(execution_profiles: profiles)
+      session = cluster.connect
+
+      puts "Running with default profile"
+
+      # By default, the driver uses a dc-aware, token-aware round-robin load balancing policy that
+      # is notified of which nodes are available in random order. To make this test's output
+      # deterministic, we sort the results by ip address.
+      ip_list = []
+      3.times do
+        rs = session.execute('select rpc_address from system.local')
+        ip_list << rs.first['rpc_address'].to_s
+      end
+      puts ip_list.sort.join("\n")
+
+      # p2 and p3 set up load-balancing policies that will match only one node, so there's no
+      # issue of hitting nodes in random order.
+      puts "Running with profile p1"
+      3.times do
+        rs = session.execute('select rpc_address from system.local', execution_profile: :p1)
+        puts rs.first['rpc_address']
+      end
+
+      puts "Running with profile p2"
+      3.times do
+        rs = session.execute('select rpc_address from system.local', execution_profile: :p2)
+        puts rs.first['rpc_address']
+      end
+      """
+    When it is executed
+    Then its output should contain:
+      """
+      Running with default profile
+      127.0.0.1
+      127.0.0.2
+      127.0.0.3
+      Running with profile p1
+      127.0.0.1
+      127.0.0.1
+      127.0.0.1
+      Running with profile p2
+      127.0.0.2
+      127.0.0.2
+      127.0.0.2
+      """
+

lib/cassandra.rb

@@ -68,6 +68,7 @@ module Cassandra
     :credentials,
     :custom_types,
     :datacenter,
+    :execution_profiles,
     :futures_factory,
     :heartbeat_interval,
     :hosts,
@@ -117,10 +118,14 @@ module Cassandra
   #   found by the default Token-Aware Load Balancing Policy should be
   #   shuffled. See {Cassandra::LoadBalancing::Policies::TokenAware#initialize Token-Aware Load Balancing Policy}.
   #
+  # @option options [Hash<String|Symbol, ExecutionProfile>] :execution_profiles (nil)
+  #   Hash of {Cassandra::Execution::Profile}s that are available for client use (e.g.
+  #   {Session#execute}, {Session#execute_async}, {Session#prepare}, and {Session#prepare_async}).
+  #
   # @option options [Numeric] :connect_timeout (10) connection timeout in
   #   seconds. Setting value to `nil` will reset it to 5 seconds.
   #
-  # @option options [Numeric] :timeout (10) request execution timeout in
+  # @option options [Numeric] :timeout (12) request execution timeout in
   #   seconds. Setting value to `nil` will remove request timeout.
   #
   # @option options [Numeric] :heartbeat_interval (30) how often should a
@@ -503,14 +508,28 @@ def self.validate_and_massage_options(options)
       end
     end
 
+    if options.key?(:execution_profiles)
+      Util.assert_instance_of(::Hash, options[:execution_profiles],
+                              ':execution_profiles must be a hash of <name,ExecutionProfile> entries.')
+    end
+
     if options.key?(:timeout)
       timeout = options[:timeout]
 
       unless timeout.nil?
-        Util.assert_instance_of(::Numeric, timeout) do
-          ":timeout must be a number of seconds, #{timeout.inspect} given"
+        Util.assert_instance_of(::Numeric, timeout, ":timeout must be a number of seconds, #{timeout.inspect} given")
+        Util.assert(timeout > 0, ":timeout must be greater than 0, #{timeout} given")
+      end
+    end
+    if options.key?(:execution_profiles)
+      options[:execution_profiles].each do |name, profile|
+        timeout = profile.timeout
+        unless timeout.nil?
+          Util.assert_instance_of(::Numeric, timeout,
+                                  ":timeout of execution profile #{name} must be a number of seconds, " \
+                                "#{timeout.inspect} given")
+          Util.assert(timeout > 0, ":timeout of execution profile #{name} must be greater than 0, #{timeout} given")
         end
-        Util.assert(timeout > 0) { ":timeout must be greater than 0, #{timeout} given" }
       end
     end
 
@@ -566,12 +585,24 @@ def self.validate_and_massage_options(options)
       load_balancing_policy = options[:load_balancing_policy]
       methods = [:host_up, :host_down, :host_found, :host_lost, :setup, :teardown,
                  :distance, :plan]
-
       Util.assert_responds_to_all(methods, load_balancing_policy) do
         ":load_balancing_policy #{load_balancing_policy.inspect} must respond " \
             "to #{methods.inspect}, but doesn't"
       end
     end
+    if options.key?(:execution_profiles)
+      methods = [:host_up, :host_down, :host_found, :host_lost, :setup, :teardown,
+                 :distance, :plan]
+      options[:execution_profiles].each do |name, profile|
+        load_balancing_policy = profile.load_balancing_policy
+        unless load_balancing_policy.nil?
+          Util.assert_responds_to_all(methods, load_balancing_policy,
+            ":load_balancing_policy in execution profile #{name} #{load_balancing_policy.inspect} must respond " \
+            "to #{methods.inspect}, but doesn't"
+          )
+        end
+      end
+    end
 
     if options.key?(:reconnection_policy)
       reconnection_policy = options[:reconnection_policy]
@@ -585,21 +616,42 @@ def self.validate_and_massage_options(options)
     if options.key?(:retry_policy)
       retry_policy = options[:retry_policy]
       methods = [:read_timeout, :write_timeout, :unavailable]
-
       Util.assert_responds_to_all(methods, retry_policy) do
         ":retry_policy #{retry_policy.inspect} must respond to #{methods.inspect}, " \
             "but doesn't"
       end
     end
+    if options.key?(:execution_profiles)
+      methods = [:read_timeout, :write_timeout, :unavailable]
+      options[:execution_profiles].each do |name, profile|
+        retry_policy = profile.retry_policy
+        unless retry_policy.nil?
+          Util.assert_responds_to_all(methods, retry_policy,
+                                      ":retry_policy in execution profile #{name} #{retry_policy.inspect} must " \
+                                      "respond to #{methods.inspect}, but doesn't"
+          )
+        end
+      end
+    end
 
     options[:listeners] = Array(options[:listeners]) if options.key?(:listeners)
 
     if options.key?(:consistency)
       consistency = options[:consistency]
-
-      Util.assert_one_of(CONSISTENCIES, consistency) do
+      Util.assert_one_of(CONSISTENCIES, consistency,
         ":consistency must be one of #{CONSISTENCIES.inspect}, " \
             "#{consistency.inspect} given"
+      )
+    end
+    if options.key?(:execution_profiles)
+      options[:execution_profiles].each do |name, profile|
+        consistency = profile.consistency
+        unless consistency.nil?
+          Util.assert_one_of(CONSISTENCIES, consistency,
+            ":consistency in execution profile #{name} must be one of #{CONSISTENCIES.inspect}, " \
+            "#{consistency.inspect} given"
+          )
+        end
       end
     end
 
@@ -824,6 +876,8 @@ def self.validate_and_massage_options(options)
 
 require 'cassandra/execution/info'
 require 'cassandra/execution/options'
+require 'cassandra/execution/profile'
+require 'cassandra/execution/profile_manager'
 require 'cassandra/execution/trace'
 
 require 'cassandra/load_balancing'
@@ -846,7 +900,9 @@ module Cassandra
   # @private
   VOID_STATEMENT = Statements::Void.new
   # @private
-  VOID_OPTIONS   = Execution::Options.new(consistency: :one)
+  VOID_OPTIONS   = Execution::Options.new(consistency: :one,
+                                          load_balancing_policy: LoadBalancing::Policies::RoundRobin.new,
+                                          retry_policy: Retry::Policies::Default.new)
   # @private
   NO_HOSTS       = Errors::NoHostsAvailable.new
 end

lib/cassandra/cluster.rb

@@ -36,9 +36,8 @@ def initialize(logger,
                    cluster_metadata,
                    execution_options,
                    connection_options,
-                   load_balancing_policy,
+                   profile_manager,
                    reconnection_policy,
-                   retry_policy,
                    address_resolution_policy,
                    connector,
                    futures_factory,
@@ -52,17 +51,16 @@ def initialize(logger,
       @metadata              = cluster_metadata
       @execution_options     = execution_options
       @connection_options    = connection_options
-      @load_balancing_policy = load_balancing_policy
+      @profile_manager       = profile_manager
       @reconnection_policy   = reconnection_policy
-      @retry_policy          = retry_policy
       @address_resolver      = address_resolution_policy
       @connector             = connector
       @futures               = futures_factory
       @timestamp_generator   = timestamp_generator
 
       @control_connection.on_close do |_cause|
         begin
-          @load_balancing_policy.teardown(self)
+          @profile_manager.teardown(self)
         rescue
           nil
         end
@@ -201,14 +199,13 @@ def connect_async(keyspace = nil)
                            @schema,
                            @io_reactor,
                            @connector,
-                           @load_balancing_policy,
+                           @profile_manager,
                            @reconnection_policy,
-                           @retry_policy,
                            @address_resolver,
                            @connection_options,
                            @futures,
                            @timestamp_generator)
-      session = Session.new(client, @execution_options, @futures)
+      session = Session.new(client, @execution_options, @futures, @profile_manager)
       promise = @futures.promise
 
       client.connect.on_complete do |f|
@@ -282,9 +279,7 @@ def inspect
       "name=#{name.inspect}, " \
       "port=#{@connection_options.port}, " \
       "protocol_version=#{@connection_options.protocol_version}, " \
-      "load_balancing_policy=#{@load_balancing_policy.inspect}, " \
-      "consistency=#{@execution_options.consistency.inspect}, " \
-      "timeout=#{@execution_options.timeout.inspect}, " \
+      "profile_manager=#{@profile_manager.inspect}, " \
       "hosts=#{hosts.inspect}, " \
       "keyspaces=#{keyspaces.inspect}>"
     end

lib/cassandra/cluster/client.rb

@@ -29,9 +29,8 @@ def initialize(logger,
                      cluster_schema,
                      io_reactor,
                      connector,
-                     load_balancing_policy,
+                     profile_manager,
                      reconnection_policy,
-                     retry_policy,
                      address_resolution_policy,
                      connection_options,
                      futures_factory,
@@ -41,9 +40,8 @@ def initialize(logger,
         @schema                      = cluster_schema
         @reactor                     = io_reactor
         @connector                   = connector
-        @load_balancing_policy       = load_balancing_policy
+        @profile_manager             = profile_manager
         @reconnection_policy         = reconnection_policy
-        @retry_policy                = retry_policy
         @address_resolver            = address_resolution_policy
         @connection_options          = connection_options
         @futures                     = futures_factory
@@ -67,7 +65,7 @@ def connect
 
           @state = :connecting
           @registry.each_host do |host|
-            distance = @load_balancing_policy.distance(host)
+            distance = @profile_manager.distance(host)
 
             case distance
             when :ignore
@@ -168,7 +166,7 @@ def host_up(host)
         pool_size = 0
 
         synchronize do
-          distance = @load_balancing_policy.distance(host)
+          distance = @profile_manager.distance(host)
           case distance
           when :ignore
             return Ione::Future.resolved
@@ -248,7 +246,7 @@ def query(statement, options)
         promise   = @futures.promise
 
         keyspace = @keyspace
-        plan     = @load_balancing_policy.plan(keyspace, statement, options)
+        plan = options.load_balancing_policy.plan(keyspace, statement, options)
 
         send_request_by_plan(promise,
                              keyspace,
@@ -270,7 +268,7 @@ def prepare(cql, options)
 
         keyspace  = @keyspace
         statement = VOID_STATEMENT
-        plan      = @load_balancing_policy.plan(keyspace, statement, options)
+        plan      = options.load_balancing_policy.plan(keyspace, statement, options)
 
         send_request_by_plan(promise,
                              keyspace,
@@ -303,7 +301,7 @@ def execute(statement, options)
         promise         = @futures.promise
 
         keyspace = @keyspace
-        plan     = @load_balancing_policy.plan(keyspace, statement, options)
+        plan     = options.load_balancing_policy.plan(keyspace, statement, options)
 
         execute_by_plan(promise, keyspace, statement, options, request, plan, timeout)
 
@@ -329,7 +327,7 @@ def batch(statement, options)
                                                timestamp,
                                                payload)
         keyspace  = @keyspace
-        plan      = @load_balancing_policy.plan(keyspace, statement, options)
+        plan      = options.load_balancing_policy.plan(keyspace, statement, options)
         promise   = @futures.promise
 
         batch_by_plan(promise, keyspace, statement, options, request, plan, timeout)
@@ -1075,20 +1073,20 @@ def handle_response(response_future,
 
             case r
             when Protocol::UnavailableErrorResponse
-              decision = @retry_policy.unavailable(statement,
+              decision = options.retry_policy.unavailable(statement,
                                                    r.consistency,
                                                    r.required,
                                                    r.alive,
                                                    retries)
             when Protocol::WriteTimeoutErrorResponse
-              decision = @retry_policy.write_timeout(statement,
+              decision = options.retry_policy.write_timeout(statement,
                                                      r.consistency,
                                                      r.write_type,
                                                      r.blockfor,
                                                      r.received,
                                                      retries)
             when Protocol::ReadTimeoutErrorResponse
-              decision = @retry_policy.read_timeout(statement,
+              decision = options.retry_policy.read_timeout(statement,
                                                     r.consistency,
                                                     r.blockfor,
                                                     r.received,
@@ -1445,15 +1443,15 @@ def wait_for_schema_agreement(connection, schedule)
           unless local.empty?
             host = @registry.host(connection.host)
 
-            if host && @load_balancing_policy.distance(host) != :ignore
+            if host && @profile_manager.distance(host) != :ignore
               versions << version = local.first['schema_version']
               @logger.debug("Host #{host.ip} schema version is #{version}")
             end
           end
 
           peers.each do |row|
             host = @registry.host(peer_ip(row))
-            next unless host && @load_balancing_policy.distance(host) != :ignore
+            next unless host && @profile_manager.distance(host) != :ignore
 
             versions << version = row['schema_version']
             @logger.debug("Host #{host.ip} schema version is #{version}")