cfg_checker/common/kube_utils.py

"""
Module to handle interaction with Kube
"""
import base64
import os
import urllib3
import yaml

from kubernetes import client as kclient, config as kconfig
from kubernetes.stream import stream

from cfg_checker.common import logger, logger_cli
from cfg_checker.common.exception import InvalidReturnException, KubeException
from cfg_checker.common.file_utils import create_temp_file_with_content
from cfg_checker.common.other import utils, shell
from cfg_checker.common.ssh_utils import ssh_shell_p
from cfg_checker.common.const import ENV_LOCAL

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def _init_kube_conf_local(config):
    # Init kube library locally
    _path = "local:{}".format(config.kube_config_path)
    try:
        kconfig.load_kube_config()
        if config.insecure:
            kconfig.assert_hostname = False
            kconfig.client_side_validation = False
        logger_cli.debug(
            "...found Kube env: core, {}". format(
                ",".join(
                    kclient.CoreApi().get_api_versions().versions
                )
            )
        )
        return kconfig, kclient.ApiClient()
    except Exception as e:
        logger.warn("Failed to init local Kube client: {}".format(
                str(e)
            )
        )
        return None, None, _path


def _init_kube_conf_remote(config):
    # init remote client
    # Preload Kube token
    """
    APISERVER=$(kubectl config view --minify |
            grep server | cut -f 2- -d ":" | tr -d " ")
    SECRET_NAME=$(kubectl get secrets |
            grep ^default | cut -f1 -d ' ')
    TOKEN=$(kubectl describe secret $SECRET_NAME |
            grep -E '^token' | cut -f2 -d':' | tr -d " ")

    echo "Detected API Server at: '${APISERVER}'"
    echo "Got secret: '${SECRET_NAME}'"
    echo "Loaded token: '${TOKEN}'"

    curl $APISERVER/api
        --header "Authorization: Bearer $TOKEN" --insecure
    """
    import yaml
    _path = ''
    if not config.env_name == ENV_LOCAL:
        _path = "{}@{}:{}".format(
            config.ssh_user,
            config.ssh_host,
            config.kube_config_path
        )
        _c_data = ssh_shell_p(
            "sudo cat " + config.kube_config_path,
            config.ssh_host,
            username=config.ssh_user,
            keypath=config.ssh_key,
            piped=False,
            use_sudo=config.ssh_uses_sudo,
        )
    else:
        _path = "local:{}".format(config.kube_config_path)
        with open(config.kube_config_path, 'r') as ff:
            _c_data = ff.read()

    if len(_c_data) < 1:
        return None, None, _path

    _conf = yaml.load(_c_data, Loader=yaml.SafeLoader)

    _kube_conf = kclient.Configuration()
    # A remote host configuration

    # To work with remote cluster, we need to extract these
    # keys = ['host', 'ssl_ca_cert', 'cert_file', 'key_file', 'verify_ssl']
    # When v12 of the client will be release, we will use load_from_dict

    _kube_conf.ssl_ca_cert = create_temp_file_with_content(
        base64.standard_b64decode(
            _conf['clusters'][0]['cluster']['certificate-authority-data']
        )
    )
    _host = _conf['clusters'][0]['cluster']['server']
    _kube_conf.cert_file = create_temp_file_with_content(
        base64.standard_b64decode(
            _conf['users'][0]['user']['client-certificate-data']
        )
    )
    _kube_conf.key_file = create_temp_file_with_content(
        base64.standard_b64decode(
            _conf['users'][0]['user']['client-key-data']
        )
    )
    if "http" not in _host or "443" not in _host:
        logger_cli.error(
            "Failed to extract Kube host: '{}'".format(_host)
        )
    else:
        logger_cli.debug(
            "...'context' host extracted: '{}' via SSH@{}".format(
                _host,
                config.ssh_host
            )
        )

    # Substitute context host to ours
    _tmp = _host.split(':')
    _kube_conf.host = \
        _tmp[0] + "://" + config.mcp_host + ":" + _tmp[2]
    config.kube_port = _tmp[2]
    logger_cli.debug(
        "...kube remote host updated to {}".format(
            _kube_conf.host
        )
    )
    _kube_conf.verify_ssl = False
    _kube_conf.debug = config.debug
    if config.insecure:
        _kube_conf.assert_hostname = False
        _kube_conf.client_side_validation = False

    # Nevertheless if you want to do it
    # you can with these 2 parameters
    # configuration.verify_ssl=True
    # ssl_ca_cert is the filepath
    # to the file that contains the certificate.
    # configuration.ssl_ca_cert="certificate"

    # _kube_conf.api_key = {
    #     "authorization": "Bearer " + config.kube_token
    # }

    # Create a ApiClient with our config
    _kube_api = kclient.ApiClient(_kube_conf)

    return _kube_conf, _kube_api, _path


class KubeApi(object):
    def __init__(self, config):
        self.config = config
        self.initialized = self._init_kclient()
        self.last_response = None

    def _init_kclient(self):
        # if there is no password - try to get local, if this available
        logger_cli.debug("... init kube config")
        if self.config.env_name == "local":
            self.kConf, self.kApi, self.kConfigPath = _init_kube_conf_local(
                self.config
            )
            self.is_local = True
            # Load local config data
            if os.path.exists(self.config.kube_config_path):
                _c_data = shell("sudo cat " + self.config.kube_config_path)
                _conf = yaml.load(_c_data, Loader=yaml.SafeLoader)
                self.user_keypath = create_temp_file_with_content(
                    base64.standard_b64decode(
                        _conf['users'][0]['user']['client-key-data']
                    )
                )
                self.yaml_conf = _c_data
        else:
            self.kConf, self.kApi, self.kConfigPath = _init_kube_conf_remote(
                self.config
            )
            self.is_local = False

        if self.kConf is None or self.kApi is None:
            return False
        else:
            return True

    def get_versions_api(self):
        # client.CoreApi().get_api_versions().versions
        return kclient.VersionApi(self.kApi)


class KubeRemote(KubeApi):
    def __init__(self, config):
        super(KubeRemote, self).__init__(config)
        self._coreV1 = None

    @property
    def CoreV1(self):
        if not self._coreV1:
            self._coreV1 = kclient.CoreV1Api(self.kApi)
        return self._coreV1

    @staticmethod
    def _typed_list_to_dict(i_list):
        _dict = {}
        for _item in i_list:
            _d = _item.to_dict()
            _type = _d.pop("type")
            _dict[_type.lower()] = _d

        return _dict

    @staticmethod
    def _get_listed_attrs(items, _path):
        _list = []
        for _n in items:
            _list.append(utils.rgetattr(_n, _path))

        return _list

    def get_node_info(self, http=False):
        # Query API for the nodes and do some presorting
        _nodes = {}
        if http:
            _raw_nodes = self.CoreV1.list_node_with_http_info()
        else:
            _raw_nodes = self.CoreV1.list_node()

        if not isinstance(_raw_nodes, kclient.models.v1_node_list.V1NodeList):
            raise InvalidReturnException(
                "Invalid return type: '{}'".format(type(_raw_nodes))
            )

        for _n in _raw_nodes.items:
            _name = _n.metadata.name
            _d = _n.to_dict()
            # parse inner data classes as dicts
            _d['addresses'] = self._typed_list_to_dict(_n.status.addresses)
            _d['conditions'] = self._typed_list_to_dict(_n.status.conditions)
            # Update 'status' type
            if isinstance(_d['conditions']['ready']['status'], str):
                _d['conditions']['ready']['status'] = utils.to_bool(
                    _d['conditions']['ready']['status']
                )
            # Parse image names?
            # TODO: Here is the place where we can parse each node image names

            # Parse roles
            _d['labels'] = {}
            for _label, _data in _d["metadata"]["labels"].items():
                if _data.lower() in ["true", "false"]:
                    _d['labels'][_label] = utils.to_bool(_data)
                else:
                    _d['labels'][_label] = _data

            # Save
            _nodes[_name] = _d

        # debug report on how many nodes detected
        logger_cli.debug("...node items returned '{}'".format(len(_nodes)))

        return _nodes

    def exec_on_target_pod(
        self,
        cmd,
        pod_name,
        namespace,
        strict=False,
        _request_timeout=120,
        **kwargs
    ):
        logger_cli.debug(
            "...searching for pods with the name '{}'".format(pod_name)
        )
        _pods = {}
        _pods = self._coreV1.list_namespaced_pod(namespace)
        _names = self._get_listed_attrs(_pods.items, "metadata.name")

        _pname = ""
        if not strict:
            _pnames = [n for n in _names if n.startswith(pod_name)]
            if len(_pnames) > 1:
                logger_cli.debug(
                    "...more than one pod found for '{}': {}\n"
                    "...using first one".format(
                        pod_name,
                        ", ".join(_pnames)
                    )
                )
                _pname = _pnames[0]
            elif len(_pname) < 1:
                raise KubeException("No pods found for '{}'".format(pod_name))
        else:
            _pname = pod_name
        logger_cli.debug(
            "...cmd: [CoreV1] exec {} -n {} -- {}".format(
                _pname,
                namespace,
                cmd
            )
        )
        _r = stream(
            self.CoreV1.connect_get_namespaced_pod_exec,
            _pname,
            namespace,
            command=cmd.split(),
            stderr=True,
            stdin=False,
            stdout=True,
            tty=False,
            _request_timeout=_request_timeout,
            **kwargs
        )

        return _r