| """ |
| 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(config_file=config.kube_config_path) |
| 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(), _path |
| 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 = '' |
| # Try to load remote config only if it was not detected already |
| if not config.kube_config_detected and not config.env_name == ENV_LOCAL: |
| _path = "{}@{}:{}".format( |
| config.ssh_user, |
| config.ssh_host, |
| config.kube_config_path |
| ) |
| _c_data = ssh_shell_p( |
| "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 |
| # Try to load local config data |
| if self.config.kube_config_path and \ |
| os.path.exists(self.config.kube_config_path): |
| _cmd = "cat " + self.config.kube_config_path |
| _c_data = shell(_cmd) |
| _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 |
| self._appsV1 = None |
| self._podV1 = None |
| |
| @property |
| def CoreV1(self): |
| if not self._coreV1: |
| self._coreV1 = kclient.CoreV1Api(self.kApi) |
| return self._coreV1 |
| |
| @property |
| def AppsV1(self): |
| if not self._appsV1: |
| self._appsV1 = kclient.AppsV1Api(self.kApi) |
| return self._appsV1 |
| |
| @property |
| def PodsV1(self): |
| if not self._podsV1: |
| self._podsV1 = kclient.V1Pod(self.kApi) |
| return self._podsV1 |
| |
| @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 |
| |
| @staticmethod |
| def safe_get_item_by_name(api_resource, _name): |
| for item in api_resource.items: |
| if item.metadata.name == _name: |
| return item |
| |
| return None |
| |
| 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 |
| ): |
| if not strict: |
| 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 = "" |
| _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 |
| ) |
| ) |
| # Set preload_content to False to preserve JSON |
| # If not, output gets converted to str |
| # Which causes to change " to ' |
| # After that json.loads(...) fail |
| _pod_stream = 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, |
| _preload_content=False, |
| **kwargs |
| ) |
| # run for timeout |
| _pod_stream.run_forever(timeout=_request_timeout) |
| # read the output |
| return _pod_stream.read_stdout() |
| |
| def ensure_namespace(self, ns): |
| """ |
| Ensure that given namespace exists |
| """ |
| # list active namespaces |
| _v1NamespaceList = self.CoreV1.list_namespace() |
| _ns = self.safe_get_item_by_name(_v1NamespaceList, ns) |
| |
| if _ns is None: |
| logger_cli.debug("... creating namespace '{}'".format(ns)) |
| _r = self.CoreV1.create_namespace(ns) |
| # TODO: check return on fail |
| if not _r: |
| return False |
| else: |
| logger_cli.debug("... found existing namespace '{}'".format(ns)) |
| |
| return True |
| |
| def get_daemon_set_by_name(self, ns, name): |
| return self.safe_get_item_by_name( |
| self.AppsV1.list_namespaced_daemon_set(ns), |
| name |
| ) |
| |
| def create_config_map(self, ns, name, source, recreate=True): |
| """ |
| Creates/Overwrites ConfigMap in working namespace |
| """ |
| # Prepare source |
| logger_cli.debug( |
| "... preparing config map '{}/{}' with files from '{}'".format( |
| ns, |
| name, |
| source |
| ) |
| ) |
| _data = {} |
| if os.path.isfile(source): |
| # populate data with one file |
| with open(source, 'rt') as fS: |
| _data[os.path.split(source)[1]] = fS.read() |
| elif os.path.isdir(source): |
| # walk dirs and populate all 'py' files |
| for path, dirs, files in os.walk(source): |
| _e = ('.py') |
| _subfiles = (_fl for _fl in files |
| if _fl.endswith(_e) and not _fl.startswith('.')) |
| for _file in _subfiles: |
| with open(os.path.join(path, _file), 'rt') as fS: |
| _data[_file] = fS.read() |
| |
| _cm = kclient.V1ConfigMap() |
| _cm.metadata = kclient.V1ObjectMeta(name=name, namespace=ns) |
| _cm.data = _data |
| logger_cli.debug( |
| "... prepared config map with {} scripts".format(len(_data)) |
| ) |
| # Query existing configmap, delete if needed |
| _existing_cm = self.safe_get_item_by_name( |
| self.CoreV1.list_namespaced_config_map(namespace=ns), |
| name |
| ) |
| if _existing_cm is not None: |
| self.CoreV1.replace_namespaced_config_map( |
| namespace=ns, |
| name=name, |
| body=_cm |
| ) |
| logger_cli.debug( |
| "... replaced existing config map '{}/{}'".format( |
| ns, |
| name |
| ) |
| ) |
| else: |
| # Create it |
| self.CoreV1.create_namespaced_config_map( |
| namespace=ns, |
| body=_cm |
| ) |
| logger_cli.debug("... created config map '{}/{}'".format( |
| ns, |
| name |
| )) |
| |
| return _data.keys() |
| |
| def prepare_daemonset_from_yaml(self, ns, ds_yaml): |
| _name = ds_yaml['metadata']['name'] |
| _ds = self.get_daemon_set_by_name(ns, _name) |
| |
| if _ds is not None: |
| logger_cli.debug( |
| "... found existing daemonset '{}'".format(_name) |
| ) |
| _r = self.AppsV1.replace_namespaced_daemon_set( |
| _ds.metadata.name, |
| _ds.metadata.namespace, |
| body=ds_yaml |
| ) |
| logger_cli.debug( |
| "... replacing existing daemonset '{}'".format(_name) |
| ) |
| return _r |
| else: |
| logger_cli.debug( |
| "... creating daemonset '{}'".format(_name) |
| ) |
| _r = self.AppsV1.create_namespaced_daemon_set(ns, body=ds_yaml) |
| return _r |
| |
| def delete_daemon_set_by_name(self, ns, name): |
| return self.AppsV1.delete_namespaced_daemon_set(name, ns) |
| |
| def exec_on_all_pods(self, pods): |
| """ |
| Create multiple threads to execute script on all target pods |
| """ |
| # Create map for threads: [[node_name, ns, pod_name]...] |
| _pod_list = [] |
| for item in pods.items: |
| _pod_list.append( |
| [ |
| item.spec.nodeName, |
| item.metadata.namespace, |
| item.metadata.name |
| ] |
| ) |
| |
| # map func and cmd |
| |
| # create result list |
| |
| return [] |