Cloudpods/ocboot/lib/service.py

#!/usr/bin/env python3
# encoding: utf-8

from __future__ import unicode_literals

from .ocboot import KEY_DISK_PATHS, KEY_ENABLE_CONTAINERD, KEY_HOST_NETWORKS, KEY_IMAGE_REPOSITORY, KEY_K8S_CONTROLPLANE_HOST, ClickhouseConfig, NodeConfig, Config, get_ansible_global_vars_by_cluster, KEY_PRIMARY_MASTER_NODE_IP
from .cmd import run_ansible_playbook
from .ansible import get_inventory_config
from .parser import help_d, inject_add_hostagent_options, inject_primary_node_options, inject_ssh_options
from .parser import inject_add_nodes_options
from .parser import inject_auto_backup_options
from .parser import inject_ssh_hosts_options
from . import utils
from . import ansible
from . import consts
from .cluster import construct_cluster
from .ocboot import WorkerConfig, Config
from .ocboot import get_ansible_global_vars
from .ocboot import KEY_ONECLOUD_VERSION
from .ssh import SSHClient
from .color import RB as Red


class BaseService(object):

    def __init__(self, subparsers, action, help_text):
        self.action = action
        parser = subparsers.add_parser(
            action, help=help_text)
        self.inject_options(parser)
        self._set_parser_defaults(parser)

    def inject_options(self, parser):
        pass

    def _set_parser_defaults(self, parser):
        parser.set_defaults(func=self.do_action)

    def do_action(self, args):
        pass


class Service(BaseService):

    def __init__(self, subparsers, action):
        super(Service, self).__init__(subparsers,
                                      action, "%s services of hosts" % action)

    def inject_options(self, parser):
        inject_ssh_hosts_options(parser)

    def do_action(self, args):
        config = NodesConfig(args.target_node_hosts,
                             args.ssh_user,
                             args.ssh_private_file,
                             args.ssh_port)
        return config.run(self.action)


class NodesConfig(object):

    def __init__(self, target_nodes, ssh_user, ssh_private_file, ssh_port):
        target_nodes = list(set(target_nodes))
        conf = [{'hostname': target_node, 'user': ssh_user, 'port': ssh_port}
                for target_node in target_nodes]
        conf_dict = {
            'hosts': conf,
        }
        self.nodes_config = NodeConfig(Config(conf_dict))

    def run(self, action, vars=None):
        inventory_content = get_inventory_config(self.nodes_config)
        yaml_content = utils.to_yaml(inventory_content)
        filepath = '/tmp/cluster_%s_node_services_inventory.yml' % action
        with open(filepath, 'w') as f:
            f.write(yaml_content)
        # start run upgrade playbook
        run_ansible_playbook(
            filepath,
            './onecloud/%s-services.yml' % action,
            vars=vars,
        )


class PrimaryMasterService(BaseService):

    def __init__(self, subparsers, action):
        super().__init__(subparsers, action, f"{action} service of primary_master_host")

    def inject_options(self, parser):
        inject_primary_node_options(parser)
        inject_ssh_options(parser)

    def get_ansible_vars(self, args, cluster, primary_master_host):
        vars = get_ansible_global_vars(
            cluster.get_current_version(),
            cluster.is_using_k3s())
        vars[KEY_K8S_CONTROLPLANE_HOST] = primary_master_host
        return vars


    def do_action(self, args):
        cluster = construct_cluster(
            args.primary_master_host,
            args.ssh_user,
            args.ssh_private_file,
            args.ssh_port)
        vars = self.get_ansible_vars(args, cluster, args.primary_master_host)
        config = NodesConfig(
            [args.primary_master_host],
            args.ssh_user,
            args.ssh_private_file,
            args.ssh_port,
        )
        return config.run(self.action, vars=vars)


class AddNodeBaseService(BaseService):

    def __init__(self, subparsers, action, help_text):
        super(AddNodeBaseService, self).__init__(subparsers, action, help_text)

    def inject_options(self, parser):
        inject_add_nodes_options(parser)


class AddNodeService(AddNodeBaseService):

    def __init__(self, subparsers, action, help_text):
        super(AddNodeService, self).__init__(subparsers, action, help_text)

    def inject_options(self, parser):
        super(AddNodeService, self).inject_options(parser)
        inject_add_hostagent_options(parser)

    def do_action(self, args):
        cluster = construct_cluster(
            args.primary_master_host,
            args.ssh_user,
            args.ssh_private_file,
            args.ssh_port)

        config = AddNodesConfig(cluster,
                                args.target_node_hosts,
                                args.ssh_user,
                                args.ssh_private_file,
                                args.ssh_port,
                                args.ssh_node_port,
                                args.enable_host_on_vm)
        return config.run()


class AddLBAgentService(AddNodeBaseService):

    def __init__(self, subparsers, action, help_text):
        super(AddLBAgentService, self).__init__(subparsers, action, help_text)

    def inject_options(self, parser):
        super(AddLBAgentService, self).inject_options(parser)

    def do_action(self, args):
        cluster = construct_cluster(
            args.primary_master_host,
            args.ssh_user,
            args.ssh_private_file,
            args.ssh_port)

        if args.ip_type == '':
            if utils.is_ipv4(args.primary_master_host):
                args.ip_type = consts.IP_TYPE_IPV4
            elif utils.is_ipv6(args.primary_master_host):
                args.ip_type = consts.IP_TYPE_IPV6
            else:
                raise ValueError("ip type is not set and cannot be determined from primary master host")

        # 处理双栈配置
        kwargs = {
            'ip_dual_conf': getattr(args, 'ip_dual_conf', None),
            'ip_type': args.ip_type,
            'offline_data_path': args.offline_data_path,
        }

        # 如果是双栈配置，需要处理IPv4和IPv6地址
        if args.ip_type == consts.IP_TYPE_DUAL_STACK and hasattr(args, 'ip_dual_conf') and args.ip_dual_conf:
            # 确定哪个是IPv4，哪个是IPv6
            if utils.is_ipv4(args.target_node_hosts[0]):
                # 主IP是IPv4，ip_dual_conf是IPv6
                kwargs['node_ip_v4'] = args.target_node_hosts[0]
                kwargs['node_ip_v6'] = args.ip_dual_conf
            else:
                # 主IP是IPv6，ip_dual_conf是IPv4
                kwargs['node_ip_v4'] = args.ip_dual_conf
                kwargs['node_ip_v6'] = args.target_node_hosts[0]

        config = AddNodesConfig(cluster,
                                args.target_node_hosts,
                                args.ssh_user,
                                args.ssh_private_file,
                                args.ssh_port,
                                args.ssh_node_port,
                                enable_host_on_vm=False, enable_lbagent=True, **kwargs)
        return config.run()


class AddNodesConfig(object):

    def __init__(self, cluster, target_nodes, ssh_user, ssh_private_file,
                 controlplane_ssh_port, ssh_port,
                 enable_host_on_vm=False,
                 enable_lbagent=False,
                 **kwargs):
        target_nodes = list(set(target_nodes))
        target_hostnames = [node.get_hostname() for node in cluster.k8s_nodes]
        self.enable_containerd = kwargs.get('runtime') == 'containerd'

        for target_node in target_nodes:
            # check IP:
            node = cluster.find_node_by_ip_or_hostname(target_node)
            if node is not None:
                raise Exception(Red("Node %s(%s) already exists in cluster (By IP Check). " % (
                    node.get_hostname(), node.get_ip())))

            # check Hostname:
            cli = SSHClient(
                target_node,
                ssh_user,
                ssh_private_file,
                ssh_port
            )
            target_hostname = cli.get_hostname()
            if target_hostname in target_hostnames:
                raise Exception(Red(f"Node {target_hostname}[{target_node}] already exists in cluster (By Hostname Check). "))

        self.current_version = cluster.get_current_version()
        controlplane_host = cluster.get_cluster_controlplane_host()
        primary_master_node_ip = cluster.get_primary_master_node_ip()
        nodes_conf = [{'hostname': target_node, 'user': ssh_user,
                       'port': ssh_port} for target_node in target_nodes]
        as_host = True
        if enable_lbagent:
            # can't enable lbagent and host at same time
            as_host = False
        woker_config_dict = {
            KEY_ONECLOUD_VERSION: self.current_version,
            'hosts': nodes_conf,
            'controlplane_host': controlplane_host,
            'as_controller': False,
            'as_host': as_host,
            'as_host_on_vm': enable_host_on_vm,
            'controlplane_ssh_port': controlplane_ssh_port,
            'enable_lbagent': enable_lbagent,
            KEY_ENABLE_CONTAINERD: self.enable_containerd,
            KEY_HOST_NETWORKS: kwargs.get(KEY_HOST_NETWORKS, None),
            KEY_DISK_PATHS: kwargs.get(KEY_DISK_PATHS, None),
            KEY_PRIMARY_MASTER_NODE_IP: primary_master_node_ip,
        }
        (repo, is_insecure) = cluster.get_repository()
        if is_insecure:
            woker_config_dict['insecure_registries'] = [repo]
        self.worker_config = WorkerConfig(Config(woker_config_dict))

        self.image_repository = cluster.get_image_repository()
        self.is_using_k3s = cluster.is_using_k3s()

        self.offline_data_path = kwargs.get('offline_data_path', None)
        self.ip_type = kwargs.get('ip_type', None)

        self.ip_dual_conf = kwargs.get("ip_dual_conf", None)
        self.node_ip_v4 = kwargs.get("node_ip_v4", None)
        self.node_ip_v6 = kwargs.get("node_ip_v6", None)

        self.enable_ai_env = kwargs.get('enable_ai_env', False)
        self.gpu_device_virtual_number = kwargs.get('gpu_device_virtual_number', 2)
        self.nvidia_driver_installer_path = kwargs.get('nvidia_driver_installer_path')
        self.cuda_installer_path = kwargs.get('cuda_installer_path')

        utils.pr_green(f"Get current cluster: {controlplane_host}, primary_master_node_ip: {primary_master_node_ip}, version: {self.current_version}, is_using_k3s: {self.is_using_k3s}")

    def run(self):
        inventory_content = ansible.get_inventory_config(self.worker_config)
        yaml_content = utils.to_yaml(inventory_content)
        filepath = './cluster_add_node_inventory.yml'
        with open(filepath, 'w') as f:
            f.write(yaml_content)
        # start run upgrade playbook
        return_code = run_ansible_playbook(
            filepath,
            './onecloud/add-node.yml',
            vars=self.get_vars(),
        )
        return return_code

    def get_vars(self):
        vars = get_ansible_global_vars(self.current_version, self.is_using_k3s)
        vars[KEY_IMAGE_REPOSITORY] = self.image_repository

        if self.offline_data_path:
            vars['offline_data_path'] = self.offline_data_path
            vars['iso_install_mode'] = True
            vars['docker_insecure_registries'] = ['private-registry.onecloud:5000']
            vars[KEY_IMAGE_REPOSITORY] = 'private-registry.onecloud:5000/yunion'
        if self.ip_type:
            vars['ip_type'] = self.ip_type

        if self.node_ip_v4:
            vars['node_ip_v4'] = self.node_ip_v4
        if self.node_ip_v6:
            vars['node_ip_v6'] = self.node_ip_v6

        if self.enable_ai_env:
            vars['enable_ai_env'] = True
            vars['gpu_device_virtual_number'] = self.gpu_device_virtual_number
            if self.nvidia_driver_installer_path:
                vars['nvidia_driver_installer_path'] = self.nvidia_driver_installer_path
            if self.cuda_installer_path:
                vars['cuda_installer_path'] = self.cuda_installer_path

        return vars


class AutoBackupConfig(NodesConfig):

    def __init__(self, target_nodes, ssh_user, ssh_private_file, ssh_port,
                 backup_path,
                 light_backup,
                 max_backups,
                 max_disck_percentage):
        target_nodes = list(set(target_nodes))
        conf = [{'hostname': target_node, 'user': ssh_user, 'port': ssh_port}
                for target_node in target_nodes]
        conf_dict = {
            'hosts': conf,
        }
        self.vars = {
            'backup_path': backup_path,
            'light_backup': 'true' if light_backup else 'false',
            'max_backups': max_backups,
            'max_disck_percentage': max_disck_percentage,
        }
        self.nodes_config = NodeConfig(Config(conf_dict))

    def run(self, action):
        inventory_content = get_inventory_config(self.nodes_config)
        yaml_content = utils.to_yaml(inventory_content)
        filepath = '/tmp/cluster_%s_node_services_inventory.yml' % action
        with open(filepath, 'w') as f:
            f.write(yaml_content)
        # start run upgrade playbook
        run_ansible_playbook(
            filepath,
            './onecloud/%s-services.yml' % action,
            vars=self.vars,
        )


class AutoBackupService(Service):

    def __init__(self, subparsers, action):
        super(AutoBackupService, self).__init__(subparsers, action)

    def inject_options(self, parser):
        super(AutoBackupService, self).inject_options(parser)
        inject_auto_backup_options(parser)

    def do_action(self, args):
        config = AutoBackupConfig(args.target_node_hosts,
                                  args.ssh_user,
                                  args.ssh_private_file,
                                  args.ssh_port,
                                  args.backup_path,
                                  args.light,
                                  args.max_backups,
                                  args.max_disk_percentage,
                                  )
        return config.run(self.action)


class ClickhouseServiceConfig(object):

    def __init__(self, cluster, primary_host,
                 ch_pwd, ch_port, offline_data_path,
                 ssh_user='root', ssh_port=22):
        cfg = {
            'host': primary_host,
            'user': ssh_user,
            'port': ssh_port,
            'ch_password': ch_pwd,
            'ch_port': ch_port,
        }
        self.cluster = cluster
        self.primary_host = primary_host
        self.ch_config = ClickhouseConfig(Config(cfg))
        self.offline_data_path = offline_data_path

    def run(self):
        inventory_content = ansible.get_inventory_config(self.ch_config)
        yaml_content = utils.to_yaml(inventory_content)
        filepath = './cluster_clickhouse_inventory.yml'
        with open(filepath, 'w') as f:
            f.write(yaml_content)
        # start run upgrade playbook
        return_code = run_ansible_playbook(
            filepath,
            './onecloud/clickhouse-services.yml',
            vars=self.get_vars(),
        )

    def get_vars(self):
        vars = self.ch_config.ansible_vars()
        vars['offline_data_path'] = self.offline_data_path
        vars[KEY_K8S_CONTROLPLANE_HOST] = self.primary_host
        global_vars = get_ansible_global_vars_by_cluster(self.cluster)
        vars.update(global_vars)
        return vars


class ClickhouseService(BaseService):

    def __init__(self, subparsers, action):
        super().__init__(subparsers, action, 'deploy clickhouse')

    def inject_options(self, parser):
        inject_primary_node_options(parser)
        inject_ssh_options(parser)
        parser.add_argument("offline_data_path",
                            metavar='OFFLINE_DATA_PATH',
                            help="offline ISO mount point, e.g., /mnt")
        parser.add_argument("--ch-password", dest="ch_password",
                            help=help_d("clickhouse password"))
        parser.add_argument("--ch-port", dest="ch_port",
                            default=9000, type=int,
                            help=help_d("clickhouse port"))

    def do_action(self, args):
        # config = 
        if args.ch_password is None:
            args.ch_password = utils.generage_random_string(12)
            print(f'generated clickhouse password: {args.ch_password}')
        cluster = construct_cluster(
            args.primary_master_host,
            args.ssh_user,
            args.ssh_private_file,
            args.ssh_port)
        print(f'found cluster {cluster.get_current_version()}')
        config = ClickhouseServiceConfig(cluster,
            args.primary_master_host, args.ch_password, args.ch_port, args.offline_data_path, args.ssh_user, args.ssh_port)
        return config.run()