community.general/plugins/modules/aerospike_migrations.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
"""short_description: Check or wait for migrations between nodes"""

# Copyright (c) 2018, Albert Autin
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import (absolute_import, division, print_function)

__metaclass__ = type

DOCUMENTATION = '''
---
module: aerospike_migrations
short_description: Check or wait for migrations between nodes
description:
    - This can be used to check for migrations in a cluster.
      This makes it easy to do a rolling upgrade/update on Aerospike nodes.
    - If waiting for migrations is not desired, simply just poll until
      port 3000 if available or asinfo -v status returns ok
author: "Albert Autin (@Alb0t)"
extends_documentation_fragment:
  - community.general.attributes
attributes:
  check_mode:
    support: full
  diff_mode:
    support: none
options:
    host:
        description:
            - Which host do we use as seed for info connection
        required: false
        type: str
        default: localhost
    port:
        description:
            - Which port to connect to Aerospike on (service port)
        required: false
        type: int
        default: 3000
    connect_timeout:
        description:
            - How long to try to connect before giving up (milliseconds)
        required: false
        type: int
        default: 1000
    consecutive_good_checks:
        description:
            - How many times should the cluster report "no migrations"
              consecutively before returning OK back to ansible?
        required: false
        type: int
        default: 3
    sleep_between_checks:
        description:
            - How long to sleep between each check (seconds).
        required: false
        type: int
        default: 60
    tries_limit:
        description:
            - How many times do we poll before giving up and failing?
        default: 300
        required: false
        type: int
    local_only:
        description:
            - Do you wish to only check for migrations on the local node
              before returning, or do you want all nodes in the cluster
              to finish before returning?
        required: true
        type: bool
    min_cluster_size:
        description:
            - Check will return bad until cluster size is met
              or until tries is exhausted
        required: false
        type: int
        default: 1
    fail_on_cluster_change:
        description:
            - Fail if the cluster key changes
              if something else is changing the cluster, we may want to fail
        required: false
        type: bool
        default: true
    migrate_tx_key:
        description:
            - The metric key used to determine if we have tx migrations
              remaining. Changeable due to backwards compatibility.
        required: false
        type: str
        default: migrate_tx_partitions_remaining
    migrate_rx_key:
        description:
            - The metric key used to determine if we have rx migrations
              remaining. Changeable due to backwards compatibility.
        required: false
        type: str
        default: migrate_rx_partitions_remaining
    target_cluster_size:
        description:
            - When all aerospike builds in the cluster are greater than
              version 4.3, then the C(cluster-stable) info command will be used.
              Inside this command, you can optionally specify what the target
              cluster size is - but it is not necessary. You can still rely on
              min_cluster_size if you don't want to use this option.
            - If this option is specified on a cluster that has at least 1
              host <4.3 then it will be ignored until the min version reaches
              4.3.
        required: false
        type: int
'''
EXAMPLES = '''
# check for migrations on local node
- name: Wait for migrations on local node before proceeding
  community.general.aerospike_migrations:
    host: "localhost"
    connect_timeout: 2000
    consecutive_good_checks: 5
    sleep_between_checks: 15
    tries_limit: 600
    local_only: false

# example playbook:
- name: Upgrade aerospike
  hosts: all
  become: true
  serial: 1
  tasks:
    - name: Install dependencies
      ansible.builtin.apt:
        name:
            - python
            - python-pip
            - python-setuptools
        state: latest
    - name: Setup aerospike
      ansible.builtin.pip:
          name: aerospike
# check for migrations every (sleep_between_checks)
# If at least (consecutive_good_checks) checks come back OK in a row, then return OK.
# Will exit if any exception, which can be caused by bad nodes,
# nodes not returning data, or other reasons.
# Maximum runtime before giving up in this case will be:
# Tries Limit * Sleep Between Checks * delay * retries
    - name: Wait for aerospike migrations
      community.general.aerospike_migrations:
          local_only: true
          sleep_between_checks: 1
          tries_limit: 5
          consecutive_good_checks: 3
          fail_on_cluster_change: true
          min_cluster_size: 3
          target_cluster_size: 4
      register: migrations_check
      until: migrations_check is succeeded
      changed_when: false
      delay: 60
      retries: 120
    - name: Another thing
      ansible.builtin.shell: |
          echo foo
    - name: Reboot
      ansible.builtin.reboot:
'''

RETURN = '''
# Returns only a success/failure result. Changed is always false.
'''

import traceback

from ansible.module_utils.basic import AnsibleModule, missing_required_lib

LIB_FOUND_ERR = None
try:
    import aerospike
    from time import sleep
    import re
except ImportError as ie:
    LIB_FOUND = False
    LIB_FOUND_ERR = traceback.format_exc()
else:
    LIB_FOUND = True


def run_module():
    """run ansible module"""
    module_args = dict(
        host=dict(type='str', required=False, default='localhost'),
        port=dict(type='int', required=False, default=3000),
        connect_timeout=dict(type='int', required=False, default=1000),
        consecutive_good_checks=dict(type='int', required=False, default=3),
        sleep_between_checks=dict(type='int', required=False, default=60),
        tries_limit=dict(type='int', required=False, default=300),
        local_only=dict(type='bool', required=True),
        min_cluster_size=dict(type='int', required=False, default=1),
        target_cluster_size=dict(type='int', required=False, default=None),
        fail_on_cluster_change=dict(type='bool', required=False, default=True),
        migrate_tx_key=dict(type='str', required=False, no_log=False,
                            default="migrate_tx_partitions_remaining"),
        migrate_rx_key=dict(type='str', required=False, no_log=False,
                            default="migrate_rx_partitions_remaining")
    )

    result = dict(
        changed=False,
    )

    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True
    )
    if not LIB_FOUND:
        module.fail_json(msg=missing_required_lib('aerospike'),
                         exception=LIB_FOUND_ERR)

    try:
        if module.check_mode:
            has_migrations, skip_reason = False, None
        else:
            migrations = Migrations(module)
            has_migrations, skip_reason = migrations.has_migs(
                module.params['local_only']
            )

        if has_migrations:
            module.fail_json(msg="Failed.", skip_reason=skip_reason)
    except Exception as e:
        module.fail_json(msg="Error: {0}".format(e))

    module.exit_json(**result)


class Migrations:
    """ Check or wait for migrations between nodes """

    def __init__(self, module):
        self.module = module
        self._client = self._create_client().connect()
        self._nodes = {}
        self._update_nodes_list()
        self._cluster_statistics = {}
        self._update_cluster_statistics()
        self._namespaces = set()
        self._update_cluster_namespace_list()
        self._build_list = set()
        self._update_build_list()
        self._start_cluster_key = \
            self._cluster_statistics[self._nodes[0]]['cluster_key']

    def _create_client(self):
        """ TODO: add support for auth, tls, and other special features
         I won't use those features, so I'll wait until somebody complains
         or does it for me (Cross fingers)
         create the client object"""
        config = {
            'hosts': [
                (self.module.params['host'], self.module.params['port'])
            ],
            'policies': {
                'timeout': self.module.params['connect_timeout']
            }
        }
        return aerospike.client(config)

    def _info_cmd_helper(self, cmd, node=None, delimiter=';'):
        """delimiter is for separate stats that come back, NOT for kv
        separation which is ="""
        if node is None:  # If no node passed, use the first one (local)
            node = self._nodes[0]
        data = self._client.info_node(cmd, node)
        data = data.split("\t")
        if len(data) != 1 and len(data) != 2:
            self.module.fail_json(
                msg="Unexpected number of values returned in info command: " +
                str(len(data))
            )
        # data will be in format 'command\touput'
        data = data[-1]
        data = data.rstrip("\n\r")
        data_arr = data.split(delimiter)

        # some commands don't return in kv format
        # so we dont want a dict from those.
        if '=' in data:
            retval = dict(
                metric.split("=", 1) for metric in data_arr
            )
        else:
            # if only 1 element found, and not kv, return just the value.
            if len(data_arr) == 1:
                retval = data_arr[0]
            else:
                retval = data_arr
        return retval

    def _update_build_list(self):
        """creates self._build_list which is a unique list
        of build versions."""
        self._build_list = set()
        for node in self._nodes:
            build = self._info_cmd_helper('build', node)
            self._build_list.add(build)

    # just checks to see if the version is 4.3 or greater
    def _can_use_cluster_stable(self):
        # if version <4.3 we can't use cluster-stable info cmd
        # regex hack to check for versions beginning with 0-3 or
        # beginning with 4.0,4.1,4.2
        if re.search(R'^([0-3]\.|4\.[0-2])', min(self._build_list)):
            return False
        return True

    def _update_cluster_namespace_list(self):
        """ make a unique list of namespaces
        TODO: does this work on a rolling namespace add/deletion?
        thankfully if it doesn't, we dont need this on builds >=4.3"""
        self._namespaces = set()
        for node in self._nodes:
            namespaces = self._info_cmd_helper('namespaces', node)
            for namespace in namespaces:
                self._namespaces.add(namespace)

    def _update_cluster_statistics(self):
        """create a dict of nodes with their related stats """
        self._cluster_statistics = {}
        for node in self._nodes:
            self._cluster_statistics[node] = \
                self._info_cmd_helper('statistics', node)

    def _update_nodes_list(self):
        """get a fresh list of all the nodes"""
        self._nodes = self._client.get_nodes()
        if not self._nodes:
            self.module.fail_json("Failed to retrieve at least 1 node.")

    def _namespace_has_migs(self, namespace, node=None):
        """returns a True or False.
        Does the namespace have migrations for the node passed?
        If no node passed, uses the local node or the first one in the list"""
        namespace_stats = self._info_cmd_helper("namespace/" + namespace, node)
        try:
            namespace_tx = \
                int(namespace_stats[self.module.params['migrate_tx_key']])
            namespace_rx = \
                int(namespace_stats[self.module.params['migrate_rx_key']])
        except KeyError:
            self.module.fail_json(
                msg="Did not find partition remaining key:" +
                self.module.params['migrate_tx_key'] +
                " or key:" +
                self.module.params['migrate_rx_key'] +
                " in 'namespace/" +
                namespace +
                "' output."
            )
        except TypeError:
            self.module.fail_json(
                msg="namespace stat returned was not numerical"
            )
        return namespace_tx != 0 or namespace_rx != 0

    def _node_has_migs(self, node=None):
        """just calls namespace_has_migs and
        if any namespace has migs returns true"""
        migs = 0
        self._update_cluster_namespace_list()
        for namespace in self._namespaces:
            if self._namespace_has_migs(namespace, node):
                migs += 1
        return migs != 0

    def _cluster_key_consistent(self):
        """create a dictionary to store what each node
        returns the cluster key as. we should end up with only 1 dict key,
        with the key being the cluster key."""
        cluster_keys = {}
        for node in self._nodes:
            cluster_key = self._cluster_statistics[node][
                'cluster_key']
            if cluster_key not in cluster_keys:
                cluster_keys[cluster_key] = 1
            else:
                cluster_keys[cluster_key] += 1
        if len(cluster_keys.keys()) == 1 and \
                self._start_cluster_key in cluster_keys:
            return True
        return False

    def _cluster_migrates_allowed(self):
        """ensure all nodes have 'migrate_allowed' in their stats output"""
        for node in self._nodes:
            node_stats = self._info_cmd_helper('statistics', node)
            allowed = node_stats['migrate_allowed']
            if allowed == "false":
                return False
        return True

    def _cluster_has_migs(self):
        """calls node_has_migs for each node"""
        migs = 0
        for node in self._nodes:
            if self._node_has_migs(node):
                migs += 1
        if migs == 0:
            return False
        return True

    def _has_migs(self, local):
        if local:
            return self._local_node_has_migs()
        return self._cluster_has_migs()

    def _local_node_has_migs(self):
        return self._node_has_migs(None)

    def _is_min_cluster_size(self):
        """checks that all nodes in the cluster are returning the
        minimum cluster size specified in their statistics output"""
        sizes = set()
        for node in self._cluster_statistics:
            sizes.add(int(self._cluster_statistics[node]['cluster_size']))

        if (len(sizes)) > 1:  # if we are getting more than 1 size, lets say no
            return False
        if (min(sizes)) >= self.module.params['min_cluster_size']:
            return True
        return False

    def _cluster_stable(self):
        """Added 4.3:
        cluster-stable:size=<target-cluster-size>;ignore-migrations=<yes/no>;namespace=<namespace-name>
        Returns the current 'cluster_key' when the following are satisfied:

         If 'size' is specified then the target node's 'cluster-size'
         must match size.
         If 'ignore-migrations' is either unspecified or 'false' then
         the target node's migrations counts must be zero for the provided
         'namespace' or all namespaces if 'namespace' is not provided."""
        cluster_key = set()
        cluster_key.add(self._info_cmd_helper('statistics')['cluster_key'])
        cmd = "cluster-stable:"
        target_cluster_size = self.module.params['target_cluster_size']
        if target_cluster_size is not None:
            cmd = cmd + "size=" + str(target_cluster_size) + ";"
        for node in self._nodes:
            try:
                cluster_key.add(self._info_cmd_helper(cmd, node))
            except aerospike.exception.ServerError as e:  # unstable-cluster is returned in form of Exception
                if 'unstable-cluster' in e.msg:
                    return False
                raise e
        if len(cluster_key) == 1:
            return True
        return False

    def _cluster_good_state(self):
        """checks a few things to make sure we're OK to say the cluster
        has no migs. It could be in a unhealthy condition that does not allow
        migs, or a split brain"""
        if self._cluster_key_consistent() is not True:
            return False, "Cluster key inconsistent."
        if self._is_min_cluster_size() is not True:
            return False, "Cluster min size not reached."
        if self._cluster_migrates_allowed() is not True:
            return False, "migrate_allowed is false somewhere."
        return True, "OK."

    def has_migs(self, local=True):
        """returns a boolean, False if no migrations otherwise True"""
        consecutive_good = 0
        try_num = 0
        skip_reason = list()
        while \
            try_num < int(self.module.params['tries_limit']) and \
                consecutive_good < \
                int(self.module.params['consecutive_good_checks']):

            self._update_nodes_list()
            self._update_cluster_statistics()

            # These checks are outside of the while loop because
            # we probably want to skip & sleep instead of failing entirely
            stable, reason = self._cluster_good_state()
            if stable is not True:
                skip_reason.append(
                    "Skipping on try#" + str(try_num) +
                    " for reason:" + reason
                )
            else:
                if self._can_use_cluster_stable():
                    if self._cluster_stable():
                        consecutive_good += 1
                    else:
                        consecutive_good = 0
                        skip_reason.append(
                            "Skipping on try#" + str(try_num) +
                            " for reason:" + " cluster_stable"
                        )
                elif self._has_migs(local):
                    # print("_has_migs")
                    skip_reason.append(
                        "Skipping on try#" + str(try_num) +
                        " for reason:" + " migrations"
                    )
                    consecutive_good = 0
                else:
                    consecutive_good += 1
                    if consecutive_good == self.module.params[
                            'consecutive_good_checks']:
                        break
            try_num += 1
            sleep(self.module.params['sleep_between_checks'])
            # print(skip_reason)
        if consecutive_good == self.module.params['consecutive_good_checks']:
            return False, None
        return True, skip_reason


def main():
    """main method for ansible module"""
    run_module()


if __name__ == '__main__':
    main()