rebalancer: introduce rebalancer flag

The flags allows to explicitly assign the rebalancer role to a
specific instance or a replicaset. Or forbid its automatic
assignment to instances or whole replicasets.

Part of #432

NO_DOC=in the final commit with rebalancer options

test/storage-luatest/rebalancer_test.lua

@@ -157,3 +157,79 @@ test_group.test_rebalancer_location = function(g)
     vtest.cluster_cfg(g, global_cfg)
     wait_rebalancer_on_instance(g, 'replica_1_a')
 end
+
+test_group.test_locate_with_flag = function(g)
+    t.assert_equals(vtest.cluster_rebalancer_find(g), 'replica_1_a')
+    --
+    -- Assign to another replicaset, with a non-minimal UUID.
+    --
+    local new_cfg_template = table.deepcopy(cfg_template)
+    new_cfg_template.sharding[1].rebalancer = false
+    new_cfg_template.sharding[2].rebalancer = true
+    local new_global_cfg = vtest.config_new(new_cfg_template)
+    vtest.cluster_cfg(g, new_global_cfg)
+    wait_rebalancer_on_instance(g, 'replica_2_a')
+    --
+    -- Automatically move the rebalancer together with the master role.
+    --
+    new_cfg_template.sharding[2].replicas.replica_2_a.read_only = true
+    new_cfg_template.sharding[2].replicas.replica_2_b.read_only = false
+    new_global_cfg = vtest.config_new(new_cfg_template)
+    vtest.cluster_cfg(g, new_global_cfg)
+    wait_rebalancer_on_instance(g, 'replica_2_b')
+    --
+    -- Assign to the replicaset with the maximal UUID.
+    --
+    new_cfg_template.sharding[2].rebalancer = false
+    new_cfg_template.sharding[3].rebalancer = true
+    new_global_cfg = vtest.config_new(new_cfg_template)
+    vtest.cluster_cfg(g, new_global_cfg)
+    wait_rebalancer_on_instance(g, 'replica_3_a')
+    --
+    -- Assign explicitly to a read-only replica.
+    --
+    new_cfg_template.sharding[3].master = nil
+    new_cfg_template.sharding[3].replicas.replica_3_b.master = true
+    new_global_cfg = vtest.config_new(new_cfg_template)
+    vtest.cluster_cfg(g, new_global_cfg)
+    wait_rebalancer_on_instance(g, 'replica_3_b')
+    --
+    -- Forbid the rebalancer on the min-UUID replicaset. Then the replicaset
+    -- with the next UUID is used.
+    --
+    new_cfg_template = table.deepcopy(cfg_template)
+    new_cfg_template.sharding[1].rebalancer = false
+    new_global_cfg = vtest.config_new(new_cfg_template)
+    vtest.cluster_cfg(g, new_global_cfg)
+    wait_rebalancer_on_instance(g, 'replica_2_a')
+    --
+    -- Forbid to run on the current master. But on a replica it won't run
+    -- without an explicit flag. Hence no rebalancer at all.
+    --
+    new_cfg_template.sharding[2].replicas.replica_2_a.rebalancer = false
+    new_global_cfg = vtest.config_new(new_cfg_template)
+    vtest.cluster_cfg(g, new_global_cfg)
+    wait_rebalancer_on_instance(g)
+    --
+    -- The master appears on another instance. The rebalancer can finally start.
+    --
+    new_cfg_template.sharding[2].replicas.replica_2_b.read_only = false
+    new_global_cfg = vtest.config_new(new_cfg_template)
+    vtest.cluster_cfg(g, new_global_cfg)
+    wait_rebalancer_on_instance(g, 'replica_2_b')
+    --
+    -- Explicitly assign to a replica in a non-min-UUID replicaset. Without
+    -- setting this flag for any replicaset.
+    --
+    new_cfg_template = table.deepcopy(cfg_template)
+    new_cfg_template.sharding[1].rebalancer = nil
+    new_cfg_template.sharding[2].replicas.replica_2_b.rebalancer = true
+    new_global_cfg = vtest.config_new(new_cfg_template)
+    vtest.cluster_cfg(g, new_global_cfg)
+    wait_rebalancer_on_instance(g, 'replica_2_b')
+    --
+    -- Cleanup.
+    --
+    vtest.cluster_cfg(g, global_cfg)
+    wait_rebalancer_on_instance(g, 'replica_1_a')
+end

test/unit-luatest/config_test.lua

@@ -128,3 +128,86 @@ g.test_extract_vshard = function()
         replication_timeout = 10,
     })
 end
+
+g.test_rebalancer_flag = function()
+    local storage_1_a = {
+        uri = 'storage:[email protected]:3301',
+        name = 'storage_1_a',
+    }
+    local replicaset_1 = {
+        replicas = {
+            storage_1_a_uuid = storage_1_a,
+        },
+    }
+    local storage_2_a = {
+        uri = 'storage:[email protected]:3302',
+        name = 'storage_2_a',
+    }
+    local replicaset_2 = {
+        replicas = {
+            storage_2_a_uuid = storage_2_a,
+        },
+    }
+    local config = {
+        sharding = {
+            storage_1_uuid = replicaset_1,
+            storage_2_uuid = replicaset_2,
+        },
+    }
+    t.assert(vcfg.check(config))
+    --
+    -- Bad replica-rebalancer flag.
+    --
+    storage_1_a.rebalancer = 'test'
+    t.assert_error_msg_content_equals(
+        'Rebalancer flag must be boolean', vcfg.check, config)
+    storage_1_a.rebalancer = nil
+    --
+    -- Bad replicaset-rebalancer flag.
+    --
+    replicaset_1.rebalancer = 'test'
+    t.assert_error_msg_content_equals(
+        'Rebalancer flag must be boolean', vcfg.check, config)
+    replicaset_1.rebalancer = nil
+    --
+    -- Rebalancer flag for a replicaset and an instance.
+    --
+    storage_1_a.rebalancer = true
+    replicaset_1.rebalancer = true
+    t.assert_error_msg_content_equals(
+        'Found 2 rebalancer flags at storage_1_uuid and storage_1_a_uuid',
+        vcfg.check, config)
+    storage_1_a.rebalancer = nil
+    replicaset_1.rebalancer = nil
+    --
+    -- Rebalancer flag for 2 replicasets.
+    --
+    replicaset_1.rebalancer = true
+    replicaset_2.rebalancer = true
+    t.assert_error_msg_content_equals(
+        'Found 2 rebalancer flags at storage_1_uuid and storage_2_uuid',
+        vcfg.check, config)
+    replicaset_1.rebalancer = nil
+    replicaset_2.rebalancer = nil
+    --
+    -- Rebalancer flag for 2 instances.
+    --
+    storage_1_a.rebalancer = true
+    storage_2_a.rebalancer = true
+    t.assert_error_msg_content_equals(
+        'Found 2 rebalancer flags at storage_1_a_uuid and storage_2_a_uuid',
+        vcfg.check, config)
+    storage_1_a.rebalancer = nil
+    storage_2_a.rebalancer = nil
+    --
+    -- Conflicting rebalancer flag in one replicaset.
+    --
+    replicaset_1.rebalancer = false
+    storage_1_a.rebalancer = true
+    t.assert_error_msg_content_equals(
+        'Replicaset storage_1_uuid can\'t run the rebalancer, and yet it was '..
+        'explicitly assigned to its instance storage_1_a_uuid',
+        vcfg.check, config)
+    replicaset_1.rebalancer = nil
+    storage_1_a.rebalancer = nil
+end

test/unit/config.result

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

vshard/cfg.lua

@@ -149,6 +149,7 @@ local replica_template = {
         type = 'boolean', name = "Master", is_optional = true,
         check = check_replica_master
     },
+    rebalancer = {type = 'boolean', name = 'Rebalancer flag', is_optional = true},
 }
 
 local function check_replicas(replicas)
@@ -169,6 +170,7 @@ local replicaset_template = {
         type = 'string', name = 'Master search mode', is_optional = true,
         check = check_replicaset_master
     },
+    rebalancer = {type = 'boolean', name = 'Rebalancer flag', is_optional = true},
 }
 
 --
@@ -211,6 +213,7 @@ local function check_sharding(sharding)
     local uris = {}
     local names = {}
     local is_all_weights_zero = true
+    local rebalancer_uuid
     for replicaset_uuid, replicaset in pairs(sharding) do
         if uuids[replicaset_uuid] then
             error(string.format('Duplicate uuid %s', replicaset_uuid))
@@ -223,7 +226,15 @@ local function check_sharding(sharding)
         if w == math.huge or w == -math.huge then
             error('Replicaset weight can not be Inf')
         end
+        if replicaset.rebalancer then
+            if rebalancer_uuid then
+                error(('Found 2 rebalancer flags at %s and %s'):format(
+                      rebalancer_uuid, replicaset_uuid))
+            end
+            rebalancer_uuid = replicaset_uuid
+        end
         validate_config(replicaset, replicaset_template)
+        local no_rebalancer = replicaset.rebalancer == false
         local is_master_auto = replicaset.master == 'auto'
         for replica_uuid, replica in pairs(replicaset.replicas) do
             if uris[replica.uri] then
@@ -240,6 +251,18 @@ local function check_sharding(sharding)
                                     'master flag in replica uuid %s',
                                     replica_uuid))
             end
+            if replica.rebalancer then
+                if rebalancer_uuid then
+                    error(('Found 2 rebalancer flags at %s and %s'):format(
+                          rebalancer_uuid, replica_uuid))
+                end
+                if no_rebalancer then
+                    error(('Replicaset %s can\'t run the rebalancer, and yet '..
+                           'it was explicitly assigned to its instance '..
+                           '%s'):format(replicaset_uuid, replica_uuid))
+                end
+                rebalancer_uuid = replica_uuid
+            end
             -- Log warning in case replica.name duplicate is
             -- found. Message appears once for each unique
             -- duplicate.

vshard/storage/init.lua

@@ -3224,27 +3224,50 @@ end
 --
 local function rebalancer_cfg_find_instance(cfg)
     local target_uuid
+    local is_assigned
     for _, rs in pairs(cfg.sharding) do
+        if rs.rebalancer == false then
+            goto next_rs
+        end
         for replica_uuid, replica in pairs(rs.replicas) do
+            local is_rebalancer = rs.rebalancer or replica.rebalancer
+            local no_rebalancer = replica.rebalancer == false
+            if is_rebalancer and not is_assigned then
+                is_assigned = true
+                target_uuid = nil
+            end
             local ok = true
-            ok = ok and replica.master
+            ok = ok and not no_rebalancer
+            ok = ok and (replica.master or replica.rebalancer)
             ok = ok and (not target_uuid or replica_uuid < target_uuid)
+            ok = ok and (not is_assigned or is_rebalancer)
             if ok then
                 target_uuid = replica_uuid
             end
         end
+    ::next_rs::
     end
     return target_uuid
 end
 
 local function rebalancer_cfg_find_replicaset(cfg)
     local target_uuid
+    local is_assigned
     for rs_uuid, rs in pairs(cfg.sharding) do
+        local is_rebalancer = rs.rebalancer
+        local no_rebalancer = rs.rebalancer == false
+        if is_rebalancer and not is_assigned then
+            is_assigned = true
+            target_uuid = nil
+        end
         local ok = true
+        ok = ok and not no_rebalancer
         ok = ok and (rs.master == 'auto')
         ok = ok and (not target_uuid or rs_uuid < target_uuid)
+        ok = ok and (not is_assigned or is_rebalancer)
         if ok then
             target_uuid = rs_uuid
+            is_assigned = is_rebalancer
         end
     end
     return target_uuid
@@ -3255,6 +3278,15 @@ local function rebalancer_is_needed()
     local this_replica_uuid = M.this_replica.uuid
     local this_replicaset_uuid = M.this_replicaset.uuid
 
+    local this_replicaset_cfg = cfg.sharding[this_replicaset_uuid]
+    if this_replicaset_cfg.rebalancer == false then
+        return false
+    end
+    local this_replica_cfg = this_replicaset_cfg.replicas[this_replica_uuid]
+    if this_replica_cfg.rebalancer == false then
+        return false
+    end
+
     local uuid = rebalancer_cfg_find_instance(cfg)
     if uuid then
         return this_replica_uuid == uuid