Network check for MCC/MOS
- Network info gathering using DaemonSet with 'hostNetwork=True'
- DaemonSet handling routines
- Mapper and Checker refactoring for Kube
Fixes
- SSH timeouts handling using env vars
MCP_SSH_TIMEOUT when connecting
MCP_SCRIPT_RUN_TIMEOUT when running command
- Progress class supports 0 as an index
Related-PROD: PROD-36575
Change-Id: Ie03a9051007eeb788901acae3696ea2bfdfe33e2
diff --git a/cfg_checker/common/kube_utils.py b/cfg_checker/common/kube_utils.py
index 315f5ee..86e59a5 100644
--- a/cfg_checker/common/kube_utils.py
+++ b/cfg_checker/common/kube_utils.py
@@ -202,6 +202,8 @@
def __init__(self, config):
super(KubeRemote, self).__init__(config)
self._coreV1 = None
+ self._appsV1 = None
+ self._podV1 = None
@property
def CoreV1(self):
@@ -209,6 +211,18 @@
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 = {}
@@ -227,6 +241,14 @@
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 = {}
@@ -279,15 +301,14 @@
_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:
+ 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(
@@ -309,7 +330,11 @@
cmd
)
)
- _r = stream(
+ # 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,
@@ -319,7 +344,146 @@
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()
- return _r
+ 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 []
diff --git a/cfg_checker/common/settings.py b/cfg_checker/common/settings.py
index dac917e..eac81c1 100644
--- a/cfg_checker/common/settings.py
+++ b/cfg_checker/common/settings.py
@@ -194,10 +194,16 @@
self.ssh_key = os.environ.get('MCP_SSH_KEY', None)
self.ssh_user = os.environ.get('MCP_SSH_USER', None)
self.ssh_host = os.environ.get('MCP_SSH_HOST', None)
+ self.ssh_connect_timeout = int(
+ os.environ.get('MCP_SSH_TIMEOUT', "15")
+ )
self.mcp_host = os.environ.get('MCP_ENV_HOST', None)
self.salt_port = os.environ.get('MCP_SALT_PORT', '6969')
self.threads = int(os.environ.get('MCP_THREADS', "5"))
+ self.script_execution_timeout = int(
+ os.environ.get('MCP_SCRIPT_RUN_TIMEOUT', "300")
+ )
self.skip_nodes = utils.node_string_to_list(os.environ.get(
'CFG_SKIP_NODES',