Add the ability to choose a control instance among the leaders (#400)

Closes #396

Before the patch, the control instance was chosen among all live instances.

Now added the ability to select a control instance only among leaders
using the `cartridge_force_leader_control_instance` flag

Author: hackallcode

CHANGELOG.md

@@ -13,6 +13,8 @@ README.md to use the newest tag with new release
 ### Added
 
 - Add `cartridge_log_dir_parent` to configure directory of logs
+- Add `cartridge_force_leader_control_instance` variable to choose a control
+  instance among the leaders
 
 ### Fixed
 

defaults/main.yml

@@ -8,6 +8,7 @@ cartridge_not_save_cookie_in_app_config: false
 cartridge_remove_temporary_files: false
 cartridge_ignore_split_brain: false
 cartridge_paths_to_keep_on_cleanup: []
+cartridge_force_leader_control_instance: false
 
 # Role scenario configuration
 

doc/steps.md

@@ -236,7 +236,8 @@ This is instance that is:
   otherwise, any instance that should be joined during current play is chosen;
 * control instance should have minimal Cartridge version across all suitable
   instances (because Cartridge two-phase commit should be called by instance
-  that has lowest version).
+  that has lowest version);
+* control instance is a leader if `cartridge_force_leader_control_instance` is set.
 
 Steps that require control instance (such as [`edit_topology`](#step-edit_topology))
 call `set_control_instance` implicitly if `control_instance` variable isn't set.
@@ -251,6 +252,7 @@ Input variables from config:
 - `expelled` - indicates if instance must be expelled from topology;
 - `stateboard` - indicates that the instance is a stateboard;
 - `replicaset_alias` - replicaset alias, will be displayed in Web UI;
+- `cartridge_force_leader_control_instance` - indicates that only a leader can be selected as a control instance.
 
 Output variables:
 

doc/variables.md

@@ -17,7 +17,9 @@ failover.
 - `cartridge_paths_to_keep_on_cleanup` (`list-of-strings`, default: `[]`): list of paths that are
   absolute or relative to `work/memtx/vinyl/wal` directory that should be kept on instance
   cleanup (`config` and` .tarantool.cookie` will be kept independently of this variable); it's
-  possible to use bash patterns, e.g. `*.control`.
+  possible to use bash patterns, e.g. `*.control`;
+- `cartridge_force_leader_control_instance` (`boolean`, default: `false`): flag indicates that only a leader
+  can be selected as a control instance.
 
 ## Role scenario configuration
 

library/cartridge_failover_promote.py

@@ -34,14 +34,6 @@ def check_leaders_promotion_is_possible(control_console):
     return None
 
 
-def get_active_leaders(control_console):
-    active_leaders, _ = control_console.eval_res_err('''
-        return require('cartridge.failover').get_active_leaders()
-    ''')
-
-    return active_leaders
-
-
 def call_failover_promote(control_console, replicaset_leaders, force_inconsistency):
     opts = {
         'force_inconsistency': force_inconsistency,
@@ -184,7 +176,7 @@ def failover_promote(params):
     # set two-phase commit opts
     helpers.set_twophase_options_from_params(control_console, params)
 
-    active_leaders = get_active_leaders(control_console)
+    active_leaders, _ = helpers.get_active_leaders(control_console)
 
     _, err = call_failover_promote(control_console, replicaset_leaders, force_inconsistency)
     if err is not None:
@@ -194,7 +186,7 @@ def failover_promote(params):
             warnings=critical_warnings,
         )
 
-    new_active_leaders = get_active_leaders(control_console)
+    new_active_leaders, _ = helpers.get_active_leaders(control_console)
 
     if critical_warnings:
         return helpers.ModuleRes(

library/cartridge_get_control_instance.py

@@ -8,6 +8,7 @@
     'play_hosts': {'required': True, 'type': 'list'},
     'console_sock': {'required': True, 'type': 'str'},
     'app_name': {'required': True, 'type': 'str'},
+    'leader_only': {'required': False, 'type': 'bool', 'default': False},
 }
 
 GET_TWOPHASE_COMMIT_VERSION_TIMEOUT = 60
@@ -94,14 +95,26 @@ def candidate_is_ok(uri, names_by_uris, module_hostvars, cluster_disabled_instan
     return helpers.is_enabled(instance_vars)
 
 
-def get_control_instance_name(module_hostvars, cluster_disabled_instances, play_hosts, control_console):
+def get_control_instance_name(
+    module_hostvars,
+    cluster_disabled_instances,
+    play_hosts,
+    control_console,
+    leader_only=False,
+):
     members, err = get_membership_members(control_console)
+    if err is not None:
+        return None, err
+    leaders, err = helpers.get_active_leaders(control_console)
     if err is not None:
         return None, err
 
+    leaders_uuid = set(leaders.values())
+
     alien_members_uris = []
     members_without_uuid = []
     members_by_uuid = {}
+    leaders_uris = []
 
     for uri, member in sorted(members.items()):
         if not member:
@@ -130,6 +143,8 @@ def get_control_instance_name(module_hostvars, cluster_disabled_instances, play_
             continue
 
         members_by_uuid[member_uuid] = member
+        if member_uuid in leaders_uuid:
+            leaders_uris.append(uri)
 
     if alien_members_uris:
         helpers.warn('Incorrect members with the following URIs ignored: %s' % ', '.join(alien_members_uris))
@@ -206,9 +221,17 @@ def get_control_instance_name(module_hostvars, cluster_disabled_instances, play_
     if err is not None:
         return None, "Failed to check instances two-phase commit version: %s" % err
 
-    idx = twophase_commit_versions.index(min(twophase_commit_versions))
-    control_instance_uri = candidates_uris[idx]
+    min_version = min(twophase_commit_versions)
+    min_version_candidates = filter(lambda c: twophase_commit_versions[c[0]] == min_version, enumerate(candidates_uris))
+    candidates_uris = list(map(lambda c: c[1], min_version_candidates))
+
+    if leader_only:
+        leader_candidates_uris = list(filter(lambda uri: uri in leaders_uris, candidates_uris))
+        if len(leader_candidates_uris) == 0:
+            return None, "Not found any leader instance between the candidates: %s" % ', '.join(candidates_uris)
+        candidates_uris = leader_candidates_uris
 
+    control_instance_uri = candidates_uris[0]
     control_instance_name = names_by_uris[control_instance_uri]
 
     return control_instance_name, None
@@ -220,11 +243,12 @@ def get_control_instance(params):
     play_hosts = params['play_hosts']
     console_sock = params['console_sock']
     app_name = params['app_name']
+    leader_only = params.get('leader_only', False)
 
     control_console = helpers.get_control_console(console_sock)
 
     control_instance_name, err = get_control_instance_name(
-        module_hostvars, cluster_disabled_instances, play_hosts, control_console
+        module_hostvars, cluster_disabled_instances, play_hosts, control_console, leader_only
     )
     if err is not None:
         return helpers.ModuleRes(failed=True, msg=err)

module_utils/helpers.py

@@ -647,10 +647,17 @@ def patch_clusterwide_config(control_console, new_sections):
     return True, None
 
 
-def enrich_replicasets_with_leaders(control_console, replicasets):
+def get_active_leaders(control_console):
     leaders, err = control_console.eval_res_err('''
         return require('cartridge.failover').get_active_leaders()
     ''')
+    if err is None and not leaders:
+        leaders = {}
+    return leaders, err
+
+
+def enrich_replicasets_with_leaders(control_console, replicasets):
+    leaders, err = get_active_leaders(control_console)
     if err:
         return err
 
@@ -754,6 +761,7 @@ class Helpers:
     read_yaml_file = staticmethod(read_yaml_file)
     get_clusterwide_config = staticmethod(get_clusterwide_config)
     patch_clusterwide_config = staticmethod(patch_clusterwide_config)
+    get_active_leaders = staticmethod(get_active_leaders)
     enrich_replicasets_with_leaders = staticmethod(enrich_replicasets_with_leaders)
     get_disabled_instances = staticmethod(get_disabled_instances)
     get_topology_checksum = staticmethod(get_topology_checksum)

tasks/set_instance_facts.yml

@@ -150,6 +150,7 @@
 
       # Edit topology check
 
+      cartridge_force_leader_control_instance: '{{ cartridge_force_leader_control_instance }}'
       cartridge_force_advertise_uris_change: '{{ cartridge_force_advertise_uris_change }}'
       cartridge_ignore_extra_cluster_instances: '{{ cartridge_ignore_extra_cluster_instances }}'
       cartridge_ignore_extra_cluster_replicasets: '{{ cartridge_ignore_extra_cluster_replicasets }}'

tasks/steps/blocks/set_control_instance.yml

@@ -10,6 +10,7 @@
     play_hosts: '{{ play_hosts }}'
     console_sock: '{{ alive_not_expelled_instance.console_sock }}'
     app_name: '{{ cartridge_app_name }}'
+    leader_only: '{{ cartridge_force_leader_control_instance }}'
   run_once: true
   delegate_to: '{{ alive_not_expelled_instance.name }}'
   register: control_instance_res

unit/test_get_control_instance.py

@@ -28,6 +28,7 @@ def call_get_control_instance(
     module_hostvars=None,
     play_hosts=None,
     cluster_disabled_instances='default',
+    leader_only=False,
 ):
     if module_hostvars is None:
         module_hostvars = {}
@@ -41,6 +42,7 @@ def call_get_control_instance(
         'play_hosts': play_hosts,
         'console_sock': console_sock,
         'app_name': app_name,
+        'leader_only': leader_only,
     })
 
 
@@ -458,6 +460,49 @@ def test_dead_instances(self):
             'http_port': 8083,
         })
 
+    def test_leaders_only(self):
+        self.instance.add_replicaset(alias='r1', instances=['instance-1', 'instance-2'])
+        self.instance.add_replicaset(alias='r2', instances=['instance-3', 'instance-4'])
+
+        self.instance.set_failover_params(mode='stateful', state_provider='stateboard')
+        self.instance.set_variable('active_leaders', {
+            'r1-uuid': 'instance-2-uuid',
+            'r2-uuid': 'instance-4-uuid',
+        })
+
+        hostvars = {}
+        hostvars.update(get_instance_hostvars('instance-1', run_dir='run-dir-1', http_port=8081))
+        hostvars.update(get_instance_hostvars('instance-2', run_dir='run-dir-2', http_port=8082))
+        hostvars.update(get_instance_hostvars('instance-3', run_dir='run-dir-3', http_port=8083))
+        hostvars.update(get_instance_hostvars('instance-4', run_dir='run-dir-4', http_port=8084))
+
+        # instance-2 is selected since it's URI is
+        # first lexicographically
+        self.instance.set_membership_members([
+            utils.get_member('instance-1', with_uuid=True),
+            utils.get_member('instance-2', with_uuid=True),
+            utils.get_member('instance-3', with_uuid=True),
+            utils.get_member('instance-4', with_uuid=True),
+        ])
+        res = call_get_control_instance('myapp', self.console_sock, hostvars, leader_only=True)
+        self.assertFalse(res.failed, msg=res.msg)
+        self.assertEqual(res.fact, {
+            'name': 'instance-2',
+            'console_sock': 'run-dir-2/myapp.instance-2.control',
+            'http_port': 8082,
+        })
+
+        # both leaders without UUID and dead
+        self.instance.set_membership_members([
+            utils.get_member('instance-1', with_uuid=True),
+            utils.get_member('instance-2', with_uuid=False, status='dead'),
+            utils.get_member('instance-3', with_uuid=True),
+            utils.get_member('instance-4', with_uuid=False, status='dead'),
+        ])
+        res = call_get_control_instance('myapp', self.console_sock, hostvars, leader_only=True)
+        self.assertTrue(res.failed)
+        self.assertIn("Not found any leader instance between the candidates: instance-1-uri, instance-3-uri", res.msg)
+
     def tearDown(self):
         self.instance.stop()
         del self.instance