- # find key:value in several lines
- @staticmethod
- def _find_in_lines(p_lines: list, p_key: str) -> str:
- for line in p_lines:
- try:
- if line.startswith(p_key + ':'):
- parts = line.split(':')
- the_value = parts[1].strip()
- return the_value
- except Exception as e:
- # ignore it
- pass
- return None
-
- # params for use in -f file
- # returns values file option and filename (in order to delete it at the end)
- def _params_to_file_option(self, cluster_uuid: str, params: dict) -> (str, str):
-
- if params and len(params) > 0:
- kube_dir, helm_dir, config_filename, cluster_dir = \
- self._get_paths(cluster_name=cluster_uuid, create_if_not_exist=True)
-
- def get_random_number():
- r = random.randrange(start=1, stop=99999999)
- s = str(r)
- while len(s) < 10:
- s = '0' + s
- return s
-
- params2 = dict()
- for key in params:
- value = params.get(key)
- if '!!yaml' in str(value):
- value = yaml.load(value[7:])
- params2[key] = value
-
- values_file = get_random_number() + '.yaml'
- with open(values_file, 'w') as stream:
- yaml.dump(params2, stream, indent=4, default_flow_style=False)
-
- return '-f {}'.format(values_file), values_file
-
- return '', None
-
- # params for use in --set option
- @staticmethod
- def _params_to_set_option(params: dict) -> str:
- params_str = ''
- if params and len(params) > 0:
- start = True
- for key in params:
- value = params.get(key, None)
- if value is not None:
- if start:
- params_str += '--set '
- start = False
- else:
- params_str += ','
- params_str += '{}={}'.format(key, value)
- return params_str
-
- @staticmethod
- def _output_to_lines(output: str) -> list:
- output_lines = list()
- lines = output.splitlines(keepends=False)
- for line in lines:
- line = line.strip()
- if len(line) > 0:
- output_lines.append(line)
- return output_lines
-
- @staticmethod
- def _output_to_table(output: str) -> list:
- output_table = list()
- lines = output.splitlines(keepends=False)
- for line in lines:
- line = line.replace('\t', ' ')
- line_list = list()
- output_table.append(line_list)
- cells = line.split(sep=' ')
- for cell in cells:
- cell = cell.strip()
- if len(cell) > 0:
- line_list.append(cell)
- return output_table
-
- def _get_paths(self, cluster_name: str, create_if_not_exist: bool = False) -> (str, str, str, str):
- """
- Returns kube and helm directories