storage: allow to choose rebalancer location

test/storage-luatest/storage_2_2_test.lua

@@ -0,0 +1,182 @@
+local t = require('luatest')
+local vtest = require('test.luatest_helpers.vtest')
+local vutil = require('vshard.util')
+
+local group_config = {{engine = 'memtx'}, {engine = 'vinyl'}}
+
+if vutil.feature.memtx_mvcc then
+ table.insert(group_config, {
+ engine = 'memtx', memtx_use_mvcc_engine = true
+ })
+ table.insert(group_config, {
+ engine = 'vinyl', memtx_use_mvcc_engine = true
+ })
+end
+
+local test_group = t.group('storage', group_config)
+
+local cfg_template = {
+ sharding = {
+ {
+ replicas = {
+ replica_1_a = {
+ master = true,
+ },
+ replica_1_b = {},
+ },
+ },
+ {
+ replicas = {
+ replica_2_a = {
+ master = true,
+ },
+ replica_2_b = {},
+ },
+ },
+ },
+ bucket_count = 16,
+ rebalancer_mode = 'off',
+}
+local global_cfg
+
+test_group.before_all(function(g)
+ cfg_template.memtx_use_mvcc_engine = g.params.memtx_use_mvcc_engine
+ global_cfg = vtest.config_new(cfg_template)
+
+ vtest.cluster_new(g, global_cfg)
+ vtest.cluster_bootstrap(g, global_cfg)
+end)
+
+test_group.after_all(function(g)
+ g.cluster:drop()
+end)
+
+local function find_rebalancer_name(g)
+ local map, err = vtest.cluster_exec_each(g, function()
+ return ivshard.storage.internal.rebalancer_fiber ~= nil
+ end)
+ t.assert_equals(err, nil)
+ local rebalancer_name
+ for name, has_rebalancer in pairs(map) do
+ if has_rebalancer then
+ t.assert_equals(rebalancer_name, nil)
+ rebalancer_name = name
+ end
+ end
+ return rebalancer_name
+end
+
+test_group.test_rebalancer_mode_off_no_marks = function(g)
+ local _, err = vtest.cluster_exec_each(g, function()
+ ilt.assert(not ivshard.storage.internal.rebalancer_fiber)
+ end)
+ t.assert_equals(err, nil)
+end
+
+test_group.test_rebalancer_mode_off_ignore_marks = function(g)
+ local new_cfg_template = table.deepcopy(cfg_template)
+ new_cfg_template.sharding[1].replicas.replica_1_a.rebalancer = true
+ local new_cfg = vtest.config_new(new_cfg_template)
+ vtest.cluster_cfg(g, new_cfg)
+ local _, err = vtest.cluster_exec_each(g, function()
+ ilt.assert(not ivshard.storage.internal.rebalancer_fiber)
+ end)
+ t.assert_equals(err, nil)
+ -- Revert.
+ vtest.cluster_cfg(g, global_cfg)
+end
+
+test_group.test_rebalancer_mode_auto_ignore_marks = function(g)
+ local new_cfg_template = table.deepcopy(cfg_template)
+ new_cfg_template.rebalancer_mode = 'auto'
+ local new_cfg = vtest.config_new(new_cfg_template)
+ local map, err = vtest.cluster_exec_each(g, function(cfg)
+ -- Tell each instance it should be the rebalancer. But it must be
+ -- ignored when mode = 'auto'.
+ local info = box.info
+ cfg.sharding[ivutil.replicaset_uuid(info)]
+ .replicas[info.uuid].rebalancer = true
+ local _, err = ivshard.storage.cfg(cfg, box.info.uuid)
+ ilt.assert_equals(err, nil)
+ return {
+ has_rebalancer = ivshard.storage.internal.rebalancer_fiber ~= nil,
+ uuid = box.info.uuid,
+ }
+ end, {new_cfg})
+ t.assert_equals(err, nil)
+ local min_uuid
+ local rebalancer_uuid
+ for _, res in pairs(map) do
+ if not min_uuid or min_uuid > res.uuid then
+ min_uuid = res.uuid
+ end
+ if res.has_rebalancer then
+ t.assert_equals(rebalancer_uuid, nil)
+ rebalancer_uuid = res.uuid
+ end
+ end
+ t.assert_equals(min_uuid, rebalancer_uuid)
+ -- Revert.
+ vtest.cluster_cfg(g, global_cfg)
+end
+
+test_group.test_rebalancer_mode_manual_no_marks = function(g)
+ local new_cfg_template = table.deepcopy(cfg_template)
+ new_cfg_template.rebalancer_mode = 'manual'
+ local new_cfg = vtest.config_new(new_cfg_template)
+ vtest.cluster_cfg(g, new_cfg)
+ local _, err = vtest.cluster_exec_each(g, function()
+ ilt.assert(not ivshard.storage.internal.rebalancer_fiber)
+ end)
+ t.assert_equals(err, nil)
+ -- Revert.
+ vtest.cluster_cfg(g, global_cfg)
+end
+
+test_group.test_rebalancer_mode_manual_with_mark = function(g)
+ local new_cfg_template = table.deepcopy(cfg_template)
+ new_cfg_template.rebalancer_mode = 'manual'
+
+ -- Try a master.
+ new_cfg_template.sharding[1].replicas.replica_1_a.rebalancer = true
+ local new_cfg = vtest.config_new(new_cfg_template)
+ vtest.cluster_cfg(g, new_cfg)
+ t.assert_equals(find_rebalancer_name(g), 'replica_1_a')
+
+ -- Try a replica.
+ new_cfg_template.sharding[1].replicas.replica_1_a.rebalancer = nil
+ new_cfg_template.sharding[2].replicas.replica_2_b.rebalancer = true
+ new_cfg = vtest.config_new(new_cfg_template)
+ vtest.cluster_cfg(g, new_cfg)
+ t.assert_equals(find_rebalancer_name(g), 'replica_2_b')
+
+ -- Ensure it works even on the replica.
+ vtest.cluster_rebalancer_disable(g)
+ g.replica_1_a:exec(function(rs2_uuid)
+ while true do
+ local bid = _G.get_first_bucket()
+ if not bid then
+ break
+ end
+ local _, err = ivshard.storage.bucket_send(
+ bid, rs2_uuid, {timeout = iwait_timeout})
+ ilt.assert_equals(err, nil)
+ end
+ end, {g.replica_2_a:replicaset_uuid()})
+ vtest.cluster_rebalancer_enable(g)
+
+ g.replica_2_b:exec(function()
+ ivtest.service_wait_for_new_ok(
+ ivshard.storage.internal.rebalancer_service, {
+ on_yield = ivshard.storage.rebalancer_wakeup
+ })
+ end)
+ local _, err = vtest.cluster_exec_each_master(g, function(bucket_count)
+ _G.bucket_gc_wait()
+ ilt.assert_equals(ivshard.storage.buckets_count(), bucket_count)
+ end, {new_cfg_template.bucket_count / #new_cfg_template.sharding})
+ t.assert_equals(err, nil)
+
+ -- Revert.
+ vtest.cluster_cfg(g, global_cfg)
+end

test/unit-luatest/config_test.lua

@@ -91,3 +91,49 @@ g.test_replica_listen = function()
  t.assert_equals(rep_1_a.uri, url1, 'uri value')
  t.assert_equals(rep_1_a.listen, {url1, url2}, 'listen value')
 end
+
+g.test_rebalancer_mode = function()
+ local config = {
+ sharding = {
+ storage_1_uuid = {
+ replicas = {
+ storage_1_a_uuid = {
+ uri = 'storage:storage@127.0.0.1:3301',
+ name = 'storage_1_a',
+ },
+ },
+ },
+ storage_2_uuid = {
+ replicas = {
+ storage_2_a_uuid = {
+ uri = 'storage:storage@127.0.0.1:3302',
+ name = 'storage_2_a',
+ },
+ }
+ }
+ },
+ rebalancer_mode = 'off',
+ }
+ t.assert(vcfg.check(config), 'mode = off')
+ local sharding = config.sharding
+ local storage_1_a = sharding.storage_1_uuid.replicas.storage_1_a_uuid
+ storage_1_a.rebalancer = true
+ t.assert(vcfg.check(config), 'mode = off with one marked')
+ storage_1_a.rebalancer = nil
+
+ config.rebalancer_mode = 'auto'
+ t.assert(vcfg.check(config), 'mode = auto')
+ storage_1_a.rebalancer = true
+ t.assert(vcfg.check(config), 'mode = auto with one marked')
+ storage_1_a.rebalancer = nil
+
+ config.rebalancer_mode = 'manual'
+ t.assert(vcfg.check(config), 'mode = manual')
+ storage_1_a.rebalancer = true
+ t.assert(vcfg.check(config), 'mode = manual with one marked')
+
+ local storage_2_a = sharding.storage_2_uuid.replicas.storage_2_a_uuid
+ storage_2_a.rebalancer = true
+ t.assert_error_msg_contains('More than one rebalancer is found',
+ vcfg.check, config)
+end

test/unit/config.result

@@ -462,9 +462,9 @@ lcfg.check(cfg)['sharding']
 - replicaset_uuid:
  replicas:
  replica_uuid:
- master: true
  uri: 127.0.0.1
  name: storage
+ master: true
  weight: 100000
 ...
 replica.uri = 'user:password@localhost'
@@ -475,9 +475,9 @@ lcfg.check(cfg)['sharding']
 - replicaset_uuid:
  replicas:
  replica_uuid:
- master: true
  uri: user:password@localhost
  name: storage
+ master: true
  weight: 100000
 ...
 replica.url = old_uri

vshard/cfg.lua

@@ -44,6 +44,12 @@ local function check_replicaset_master(master)
  end
 end
 
+local function cfg_check_rebalancer_mode(mode)
+ if mode ~= 'auto' and mode ~= 'manual' and mode ~= 'off' then
+ error('Only "auto", "manual", and "off" are supported')
+ end
+end
+
 local function is_number(v)
  return type(v) == 'number' and v == v
 end
@@ -148,6 +154,7 @@ local replica_template = {
  type = 'boolean', name = "Master", is_optional = true,
  check = check_replica_master
  },
+ rebalancer = {type = 'boolean', name = 'Rebalancer', is_optional = true},
 }
 
 local function check_replicas(replicas)
@@ -209,6 +216,7 @@ local function check_sharding(sharding)
  local uuids = {}
  local uris = {}
  local names = {}
+ local rebalancer_uuid
  local is_all_weights_zero = true
  for replicaset_uuid, replicaset in pairs(sharding) do
  if uuids[replicaset_uuid] then
@@ -252,6 +260,14 @@ local function check_sharding(sharding)
  names[name] = 2
  end
  end
+ if replica.rebalancer then
+ if rebalancer_uuid then
+ error(string.format('More than one rebalancer is found: '..
+ 'on replicas %s and %s', replica_uuid,
+ rebalancer_uuid))
+ end
+ rebalancer_uuid = replica_uuid
+ end
  end
  is_all_weights_zero = is_all_weights_zero and replicaset.weight == 0
  end
@@ -278,6 +294,10 @@ local cfg_template = {
  type = 'positive integer', name = 'Bucket count', is_optional = true,
  default = consts.DEFAULT_BUCKET_COUNT
  },
+ rebalancer_mode = {
+ type = 'string', name = 'Rebalancer mode', is_optional = true,
+ default = 'auto', check = cfg_check_rebalancer_mode,
+ },
  rebalancer_disbalance_threshold = {
  type = 'non-negative number', name = 'Rebalancer disbalance threshold',
  is_optional = true,

vshard/storage/init.lua

@@ -3315,6 +3315,32 @@ end
 -- Configuration
 --------------------------------------------------------------------------------
 
+local function storage_cfg_find_rebalancer(cfg)
+ local mode = cfg.rebalancer_mode
+ if mode == 'off' then
+ return
+ end
+ local min_master_uuid
+ local rebalancer_uuid
+ for _, rs in pairs(cfg.sharding) do
+ for replica_uuid, replica in pairs(rs.replicas) do
+ if replica.rebalancer then
+ assert(not rebalancer_uuid)
+ rebalancer_uuid = replica_uuid
+ end
+ if replica.master and (not min_master_uuid or
+ replica_uuid < min_master_uuid) then
+ min_master_uuid = replica_uuid
+ end
+ end
+ end
+ if mode == 'auto' then
+ return min_master_uuid
+ end
+ assert(mode == 'manual')
+ return rebalancer_uuid
+end
+
 local function storage_cfg(cfg, this_replica_uuid, is_reload)
  if this_replica_uuid == nil then
  error('Usage: cfg(configuration, this_replica_uuid)')
@@ -3337,13 +3363,8 @@ local function storage_cfg(cfg, this_replica_uuid, is_reload)
  local this_replicaset
  local this_replica
  local new_replicasets = lreplicaset.buildall(vshard_cfg)
- local min_master
  for _, rs in pairs(new_replicasets) do
  for replica_uuid, replica in pairs(rs.replicas) do
- if (min_master == nil or replica_uuid < min_master.uuid) and
- rs.master == replica then
- min_master = replica
- end
  if replica_uuid == this_replica_uuid then
  assert(this_replicaset == nil)
  this_replicaset = rs
@@ -3481,7 +3502,8 @@ local function storage_cfg(cfg, this_replica_uuid, is_reload)
  local_on_master_enable()
  end
 
- if min_master == this_replica then
+ local rebalancer_uuid = storage_cfg_find_rebalancer(vshard_cfg)
+ if rebalancer_uuid == this_replica.uuid then
  if not M.rebalancer_fiber then
  M.rebalancer_fiber =
  util.reloadable_fiber_create('vshard.rebalancer', M,
@@ -3492,7 +3514,12 @@ local function storage_cfg(cfg, this_replica_uuid, is_reload)
  M.rebalancer_fiber:wakeup()
  end
  elseif M.rebalancer_fiber then
- log.info('Rebalancer location has changed to %s', min_master)
+ if rebalancer_uuid then
+ log.info('Rebalancer location has changed to instance %s',
+ rebalancer_uuid)
+ else
+ log.info('Rebalancer is turned off')
+ end
  M.rebalancer_fiber:cancel()
  M.rebalancer_fiber = nil
  end