tarantool · rosik · Jul 27, 2021 · Jul 26, 2021
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -44,6 +44,8 @@ Added
 
 - New script ``rst/BuildUML.cmake`` for rendering UML diagrams from the doc.
 
+- New vshard option ``rebalancer_max_sending``
+
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Changed
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

diff --git a/cartridge/vshard-utils.lua b/cartridge/vshard-utils.lua
@@ -22,6 +22,7 @@ vars:new('known_groups', nil
     --    [group_name] = {
     --        bucket_count = number,
     --        rebalancer_max_receiving = number,
+    --        rebalancer_max_sending = number,
     --        collect_lua_garbage = boolean,
     --        collect_bucket_garbage_interval = number,
     --        sync_timeout = number,
@@ -179,6 +180,16 @@ local function validate_vshard_group(field, vsgroup_new, vsgroup_old)
             '%s.rebalancer_max_receiving must be positive', field
         )
     end
+    if vsgroup_new.rebalancer_max_sending ~= nil then
+        ValidateConfigError:assert(
+            type(vsgroup_new.rebalancer_max_sending) == 'number',
+            '%s.rebalancer_max_sending must be a number', field
+        )
+        ValidateConfigError:assert(
+            vsgroup_new.rebalancer_max_sending > 0,
+            '%s.rebalancer_max_sending must be positive', field
+        )
+    end
     if vsgroup_new.collect_lua_garbage ~= nil then
         ValidateConfigError:assert(
             type(vsgroup_new.collect_lua_garbage) == 'boolean',
@@ -249,6 +260,7 @@ local function validate_vshard_group(field, vsgroup_new, vsgroup_old)
         ['bucket_count'] = true,
         ['bootstrapped'] = true,
         ['rebalancer_max_receiving'] = true,
+        ['rebalancer_max_sending'] = true,
         ['collect_lua_garbage'] = true,
         ['sync_timeout'] = true,
         ['collect_bucket_garbage_interval'] = true,
@@ -405,6 +417,10 @@ local function get_known_groups()
             g.rebalancer_max_receiving = vshard_consts.DEFAULT_REBALANCER_MAX_RECEIVING
         end
 
+        if g.rebalancer_max_sending == nil then
+            g.rebalancer_max_sending = vshard_consts.DEFAULT_REBALANCER_MAX_SENDING
+        end
+
         if g.collect_lua_garbage == nil then
             g.collect_lua_garbage = false
         end
@@ -501,6 +517,7 @@ local function get_vshard_config(group_name, conf)
     return {
         bucket_count = vshard_groups[group_name].bucket_count,
         rebalancer_max_receiving = vshard_groups[group_name].rebalancer_max_receiving,
+        rebalancer_max_sending = vshard_groups[group_name].rebalancer_max_sending,
         sched_ref_quota = vshard_groups[group_name].sched_ref_quota,
         sched_move_quota = vshard_groups[group_name].sched_move_quota,
         collect_lua_garbage = vshard_groups[group_name].collect_lua_garbage,
@@ -561,6 +578,7 @@ local function edit_vshard_options(group_name, vshard_options)
         'string',
         {
             rebalancer_max_receiving = '?number',
+            rebalancer_max_sending = '?number',
             collect_lua_garbage = '?boolean',
             sync_timeout = '?number',
             collect_bucket_garbage_interval = '?number',

diff --git a/cartridge/webui/api-vshard.lua b/cartridge/webui/api-vshard.lua
@@ -25,6 +25,12 @@ local gql_type_vsgroup = gql_types.object({
                 'The maximum number of buckets that can be received in parallel by a single replica set ' ..
                 'in the storage group'
         },
+        rebalancer_max_sending = {
+            kind = gql_types.int.nonNull,
+            description =
+                'The maximum number of buckets that can be sent in parallel by a single replica set ' ..
+                'in the storage group'
+        },
         collect_lua_garbage = {
             kind = gql_types.boolean.nonNull,
             description = 'If set to true, the Lua collectgarbage() function is called periodically'
@@ -159,6 +165,7 @@ local function init(graphql)
         args = {
             name = gql_types.string.nonNull,
             rebalancer_max_receiving = gql_types.int,
+            rebalancer_max_sending = gql_types.int,
             collect_lua_garbage = gql_types.boolean,
             sync_timeout = gql_types.float,
             collect_bucket_garbage_interval = gql_types.float,

diff --git a/doc/schema.graphql b/doc/schema.graphql
@@ -1,5 +1,5 @@
 # source: http://localhost:8081/admin/api
-# timestamp: Mon Jul 12 2021 03:57:45 GMT+0300 (Moscow Standard Time)
+# timestamp: Mon Jul 26 2021 16:55:54 GMT+0300 (Москва, стандартное время)
 
 """Cluster management"""
 type Apicluster {
@@ -249,7 +249,7 @@ type MutationApicluster {
 
   """Restart replication on specified by uuid servers"""
   restart_replication(uuids: [String!]): Boolean
-  edit_vshard_options(rebalancer_max_receiving: Int, sched_ref_quota: Long, collect_bucket_garbage_interval: Float, sync_timeout: Float, collect_lua_garbage: Boolean, rebalancer_disbalance_threshold: Float, name: String!, sched_move_quota: Long): VshardGroup!
+  edit_vshard_options(rebalancer_max_receiving: Int, collect_bucket_garbage_interval: Float, collect_lua_garbage: Boolean, sched_ref_quota: Long, rebalancer_max_sending: Int, sync_timeout: Float, rebalancer_disbalance_threshold: Float, name: String!, sched_move_quota: Long): VshardGroup!
   auth_params(cookie_max_age: Long, enabled: Boolean, cookie_renew_age: Long): UserManagementAPI!
 
   """Create a new user"""
@@ -607,11 +607,16 @@ type VshardGroup {
   """
   sync_timeout: Float!
 
+  """Whether the group is ready to operate"""
+  bootstrapped: Boolean!
+
   """Scheduler storage ref quota"""
   sched_ref_quota: Long!
 
-  """Whether the group is ready to operate"""
-  bootstrapped: Boolean!
+  """
+  The maximum number of buckets that can be sent in parallel by a single replica set in the storage group
+  """
+  rebalancer_max_sending: Int!
 
   """A maximum bucket disbalance threshold, in percent"""
   rebalancer_disbalance_threshold: Float!

diff --git a/test/integration/multisharding_one_test.lua b/test/integration/multisharding_one_test.lua
@@ -57,6 +57,7 @@ local function get_vshard_groups(cluster)
                     bucket_count
                     bootstrapped
                     rebalancer_max_receiving
+                    rebalancer_max_sending
                     collect_lua_garbage
                     sync_timeout
                     collect_bucket_garbage_interval
@@ -72,6 +73,7 @@ local function edit_vshard_group(cluster, kv_args)
     local res = cluster.main_server:graphql({query = [[
         mutation(
             $rebalancer_max_receiving: Int
+            $rebalancer_max_sending: Int
             $group: String!
             $collect_lua_garbage: Boolean
             $sync_timeout: Float
@@ -82,6 +84,7 @@ local function edit_vshard_group(cluster, kv_args)
                 edit_vshard_options(
                     name: $group
                     rebalancer_max_receiving: $rebalancer_max_receiving
+                    rebalancer_max_sending: $rebalancer_max_sending
                     collect_lua_garbage: $collect_lua_garbage
                     sync_timeout: $sync_timeout
                     collect_bucket_garbage_interval: $collect_bucket_garbage_interval
@@ -91,6 +94,7 @@ local function edit_vshard_group(cluster, kv_args)
                     bucket_count
                     bootstrapped
                     rebalancer_max_receiving
+                    rebalancer_max_sending
                     collect_lua_garbage
                     sync_timeout
                     collect_bucket_garbage_interval
@@ -116,6 +120,7 @@ function g.test_api()
                 bucket_count
                 bootstrapped
                 rebalancer_max_receiving
+                rebalancer_max_sending
                 collect_lua_garbage
                 sync_timeout
                 collect_bucket_garbage_interval
@@ -151,6 +156,7 @@ function g.test_api()
             ['collect_lua_garbage'] = false,
             ['rebalancer_disbalance_threshold'] = 1,
             ['rebalancer_max_receiving'] = 100,
+            ['rebalancer_max_sending'] = 1,
             ['sync_timeout'] = 1,
             ['name'] = 'default',
             ['bucket_count'] = 3000,
@@ -187,6 +193,7 @@ function g.test_set_vshard_options_positive()
     local res = edit_vshard_group(g.cluster, {
         group = "default",
         rebalancer_max_receiving = 42,
+        rebalancer_max_sending = 1,
         collect_lua_garbage = true,
         sync_timeout = 24,
         rebalancer_disbalance_threshold = 14
@@ -196,6 +203,7 @@ function g.test_set_vshard_options_positive()
         ['collect_lua_garbage'] = true,
         ['rebalancer_disbalance_threshold'] = 14,
         ['rebalancer_max_receiving'] = 42,
+        ['rebalancer_max_sending'] = 1,
         ['sync_timeout'] = 24,
         ['name'] = 'default',
         ['bucket_count'] = 3000,
@@ -205,13 +213,15 @@ function g.test_set_vshard_options_positive()
     local res = edit_vshard_group(g.cluster, {
         group = "default",
         rebalancer_max_receiving = nil,
+        rebalancer_max_sending = nil,
         sync_timeout = 25,
     })
     t.assert_equals(res['data']['cluster']['edit_vshard_options'], {
         ['collect_bucket_garbage_interval'] = box.NULL,
         ['collect_lua_garbage'] = true,
         ['rebalancer_disbalance_threshold'] = 14,
         ['rebalancer_max_receiving'] = 42,
+        ['rebalancer_max_sending'] = 1,
         ['sync_timeout'] = 25,
         ['name'] = 'default',
         ['bucket_count'] = 3000,
@@ -225,6 +235,7 @@ function g.test_set_vshard_options_positive()
             ['collect_lua_garbage'] = true,
             ['rebalancer_disbalance_threshold'] = 14,
             ['rebalancer_max_receiving'] = 42,
+            ['rebalancer_max_sending'] = 1,
             ['sync_timeout'] = 25,
             ['name'] = 'default',
             ['bucket_count'] = 3000,
@@ -245,6 +256,11 @@ function g.test_set_vshard_options_negative()
         edit_vshard_group, g.cluster, {group = "default", rebalancer_max_receiving = -42}
     )
 
+    t.assert_error_msg_contains(
+        [[vshard_groups["default"].rebalancer_max_sending must be positive]],
+        edit_vshard_group, g.cluster, {group = "default", rebalancer_max_sending = -42}
+    )
+
     t.assert_error_msg_contains(
         [[vshard_groups["default"].sync_timeout must be non-negative]],
         edit_vshard_group, g.cluster, {group = "default", sync_timeout = -24}

diff --git a/test/integration/multisharding_two_test.lua b/test/integration/multisharding_two_test.lua
@@ -81,6 +81,7 @@ local function get_vshard_groups(cluster)
                     bucket_count
                     bootstrapped
                     rebalancer_max_receiving
+                    rebalancer_max_sending
                     collect_lua_garbage
                     sync_timeout
                     collect_bucket_garbage_interval
@@ -96,6 +97,7 @@ local function edit_vshard_group(cluster, kv_args)
     local res = cluster.main_server:graphql({query = [[
         mutation(
             $rebalancer_max_receiving: Int
+            $rebalancer_max_sending: Int
             $group: String!
             $collect_lua_garbage: Boolean
             $sync_timeout: Float
@@ -106,6 +108,7 @@ local function edit_vshard_group(cluster, kv_args)
                 edit_vshard_options(
                     name: $group
                     rebalancer_max_receiving: $rebalancer_max_receiving
+                    rebalancer_max_sending: $rebalancer_max_sending
                     collect_lua_garbage: $collect_lua_garbage
                     sync_timeout: $sync_timeout
                     collect_bucket_garbage_interval: $collect_bucket_garbage_interval
@@ -115,6 +118,7 @@ local function edit_vshard_group(cluster, kv_args)
                     bucket_count
                     bootstrapped
                     rebalancer_max_receiving
+                    rebalancer_max_sending
                     collect_lua_garbage
                     sync_timeout
                     collect_bucket_garbage_interval
@@ -141,6 +145,7 @@ function g.test_api()
                 bucket_count
                 bootstrapped
                 rebalancer_max_receiving
+                rebalancer_max_sending
                 collect_lua_garbage
                 sync_timeout
                 collect_bucket_garbage_interval
@@ -183,6 +188,7 @@ function g.test_api()
             ['collect_lua_garbage'] = false,
             ['rebalancer_disbalance_threshold'] = 1,
             ['rebalancer_max_receiving'] = 100,
+            ['rebalancer_max_sending'] = 1,
             ['sync_timeout'] = 1,
             ['name'] = 'cold',
             ['bucket_count'] = 2000,
@@ -192,6 +198,7 @@ function g.test_api()
             ['collect_lua_garbage'] = false,
             ['rebalancer_disbalance_threshold'] = 1,
             ['rebalancer_max_receiving'] = 100,
+            ['rebalancer_max_sending'] = 1,
             ['sync_timeout'] = 1,
             ['name'] = 'hot',
             ['bucket_count'] = 30000,
@@ -288,6 +295,7 @@ function g.test_set_vshard_options_positive()
         ['collect_lua_garbage'] = false,
         ['rebalancer_disbalance_threshold'] = 1,
         ['rebalancer_max_receiving'] = 42,
+        ['rebalancer_max_sending'] = 1,
         ['sync_timeout'] = 1,
         ['name'] = 'cold',
         ['bucket_count'] = 2000,
@@ -296,13 +304,15 @@ function g.test_set_vshard_options_positive()
 
     local res = edit_vshard_group(g.cluster, {
         group = 'hot',
-        rebalancer_max_receiving = 44
+        rebalancer_max_receiving = 44,
+        rebalancer_max_sending = 2,
     })
     t.assert_equals(res['data']['cluster']['edit_vshard_options'], {
         ['collect_bucket_garbage_interval'] = box.NULL,
         ['collect_lua_garbage'] = false,
         ['rebalancer_disbalance_threshold'] = 1,
         ['rebalancer_max_receiving'] = 44,
+        ['rebalancer_max_sending'] = 2,
         ['sync_timeout'] = 1,
         ['name'] = 'hot',
         ['bucket_count'] = 30000,
@@ -316,6 +326,7 @@ function g.test_set_vshard_options_positive()
             ['collect_lua_garbage'] = false,
             ['rebalancer_disbalance_threshold'] = 1,
             ['rebalancer_max_receiving'] = 42,
+            ['rebalancer_max_sending'] = 1,
             ['sync_timeout'] = 1,
             ['name'] = 'cold',
             ['bucket_count'] = 2000,
@@ -326,6 +337,7 @@ function g.test_set_vshard_options_positive()
             ['collect_lua_garbage'] = false,
             ['rebalancer_disbalance_threshold'] = 1,
             ['rebalancer_max_receiving'] = 44,
+            ['rebalancer_max_sending'] = 2,
             ['sync_timeout'] = 1,
             ['name'] = 'hot',
             ['bucket_count'] = 30000,

diff --git a/test/unit/vshard_config_test.lua b/test/unit/vshard_config_test.lua
@@ -165,6 +165,15 @@ vshard_groups:
     rebalancer_max_receiving: 0
 ...]])
 
+check_config('vshard_groups["global"].rebalancer_max_sending must be positive',
+[[---
+vshard_groups:
+  global:
+    bucket_count: 200
+    bootstrapped: false
+    rebalancer_max_sending: 0
+...]])
+
 check_config('vshard_groups["global"].collect_lua_garbage must be a boolean',
 [[---
 vshard_groups:

diff --git a/test/upgrade/data_1.10.5/localhost-13301/config/vshard_groups.yml b/test/upgrade/data_1.10.5/localhost-13301/config/vshard_groups.yml
@@ -1,6 +1,7 @@
 ---
 default:
   rebalancer_max_receiving: 100
+  rebalancer_max_sending: 1
   bootstrapped: true
   collect_bucket_garbage_interval: 0.5
   collect_lua_garbage: false

diff --git a/test/upgrade/data_1.10.5/localhost-13303/config/vshard_groups.yml b/test/upgrade/data_1.10.5/localhost-13303/config/vshard_groups.yml
@@ -1,6 +1,7 @@
 ---
 default:
   rebalancer_max_receiving: 100
+  rebalancer_max_sending: 1
   bootstrapped: true
   collect_bucket_garbage_interval: 0.5
   collect_lua_garbage: false

diff --git a/test/upgrade/data_1.10.5/localhost-13305/config/vshard_groups.yml b/test/upgrade/data_1.10.5/localhost-13305/config/vshard_groups.yml
@@ -1,6 +1,7 @@
 ---
 default:
   rebalancer_max_receiving: 100
+  rebalancer_max_sending: 1
   bootstrapped: true
   collect_bucket_garbage_interval: 0.5
   collect_lua_garbage: false