# contact with: nfvlabs@tid.es
##
-import paramiko
import subprocess
import os
import shutil
from uuid import uuid4
import random
from n2vc.k8s_conn import K8sConnector
+from n2vc.exceptions import K8sException
class K8sHelmConnector(K8sConnector):
self._helm_command = helm_command
self._check_file_exists(filename=helm_command, exception_if_not_exists=True)
+ # initialize helm client-only
+ self.debug('Initializing helm client-only...')
+ command = '{} init --client-only'.format(self._helm_command)
+ try:
+ asyncio.ensure_future(self._local_async_exec(command=command, raise_exception_on_error=False))
+ # loop = asyncio.get_event_loop()
+ # loop.run_until_complete(self._local_async_exec(command=command, raise_exception_on_error=False))
+ except Exception as e:
+ self.warning(msg='helm init failed (it was already initialized): {}'.format(e))
+
self.info('K8S Helm connector initialized')
async def init_env(
namespace: str = 'kube-system',
reuse_cluster_uuid=None
) -> (str, bool):
+ """
+ It prepares a given K8s cluster environment to run Charts on both sides:
+ client (OSM)
+ server (Tiller)
+
+ :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid '.kube/config'
+ :param namespace: optional namespace to be used for helm. By default, 'kube-system' will be used
+ :param reuse_cluster_uuid: existing cluster uuid for reuse
+ :return: uuid of the K8s cluster and True if connector has installed some software in the cluster
+ (on error, an exception will be raised)
+ """
cluster_uuid = reuse_cluster_uuid
if not cluster_uuid:
kube_dir, helm_dir, config_filename, cluster_dir = \
self._get_paths(cluster_name=cluster_uuid, create_if_not_exist=True)
- command = '{} --kubeconfig={} --home={} repo list --output yaml'.format(self._helm_command, config_filename, helm_dir)
+ command = '{} --kubeconfig={} --home={} repo list --output yaml'\
+ .format(self._helm_command, config_filename, helm_dir)
output, rc = await self._local_async_exec(command=command, raise_exception_on_error=True)
if output and len(output) > 0:
msg = 'Cluster has releases and not force. Cannot reset K8s environment. Cluster uuid: {}'\
.format(cluster_uuid)
self.error(msg)
- raise Exception(msg)
+ raise K8sException(msg)
if uninstall_sw:
else:
msg = 'Tiller deployment not found in cluster {}'.format(cluster_uuid)
self.error(msg)
- # raise Exception(msg)
self.debug('namespace for tiller: {}'.format(namespace))
atomic: bool = True,
timeout: float = 300,
params: dict = None,
- db_dict: dict = None
+ db_dict: dict = None,
+ kdu_name: str = None
):
self.debug('installing {} in cluster {}'.format(kdu_model, cluster_uuid))
- start = time.time()
- end = start + timeout
-
# config filename
kube_dir, helm_dir, config_filename, cluster_dir = \
self._get_paths(cluster_name=cluster_uuid, create_if_not_exist=True)
version_str = '--version {}'.format(parts[1])
kdu_model = parts[0]
- # generate a name for the releas. Then, check if already exists
+ # generate a name for the release. Then, check if already exists
kdu_instance = None
while kdu_instance is None:
kdu_instance = K8sHelmConnector._generate_release_name(kdu_model)
if result is not None:
# instance already exists: generate a new one
kdu_instance = None
- except:
- kdu_instance = None
+ except K8sException:
+ pass
# helm repo install
command = '{} install {} --output yaml --kubeconfig={} --home={} {} {} --name={} {} {}'\
if rc != 0:
msg = 'Error executing command: {}\nOutput: {}'.format(command, output)
self.error(msg)
- raise Exception(msg)
+ raise K8sException(msg)
self.debug('Returning kdu_instance {}'.format(kdu_instance))
return kdu_instance
self.debug('upgrading {} in cluster {}'.format(kdu_model, cluster_uuid))
- start = time.time()
- end = start + timeout
-
# config filename
kube_dir, helm_dir, config_filename, cluster_dir = \
self._get_paths(cluster_name=cluster_uuid, create_if_not_exist=True)
)
# wait for execution task
- await asyncio.wait([ exec_task ])
+ await asyncio.wait([exec_task])
# cancel status task
status_task.cancel()
if rc != 0:
msg = 'Error executing command: {}\nOutput: {}'.format(command, output)
self.error(msg)
- raise Exception(msg)
+ raise K8sException(msg)
# return new revision number
instance = await self.get_instance_info(cluster_uuid=cluster_uuid, kdu_instance=kdu_instance)
if rc != 0:
msg = 'Error executing command: {}\nOutput: {}'.format(command, output)
self.error(msg)
- raise Exception(msg)
+ raise K8sException(msg)
# return new revision number
instance = await self.get_instance_info(cluster_uuid=cluster_uuid, kdu_instance=kdu_instance)
async def inspect_kdu(
self,
- kdu_model: str
+ kdu_model: str,
+ repo_url: str = None
) -> str:
- self.debug('inspect kdu_model {}'.format(kdu_model))
+ self.debug('inspect kdu_model {} from (optional) repo: {}'.format(kdu_model, repo_url))
- command = '{} inspect values {}'\
- .format(self._helm_command, kdu_model)
+ return await self._exec_inspect_comand(inspect_command='', kdu_model=kdu_model, repo_url=repo_url)
- output, rc = await self._local_async_exec(command=command)
+ async def values_kdu(
+ self,
+ kdu_model: str,
+ repo_url: str = None
+ ) -> str:
- return output
+ self.debug('inspect kdu_model values {} from (optional) repo: {}'.format(kdu_model, repo_url))
+
+ return await self._exec_inspect_comand(inspect_command='values', kdu_model=kdu_model, repo_url=repo_url)
async def help_kdu(
self,
- kdu_model: str
- ):
-
- self.debug('help kdu_model {}'.format(kdu_model))
-
- command = '{} inspect readme {}'\
- .format(self._helm_command, kdu_model)
+ kdu_model: str,
+ repo_url: str = None
+ ) -> str:
- output, rc = await self._local_async_exec(command=command, raise_exception_on_error=True)
+ self.debug('inspect kdu_model {} readme.md from repo: {}'.format(kdu_model, repo_url))
- return output
+ return await self._exec_inspect_comand(inspect_command='readme', kdu_model=kdu_model, repo_url=repo_url)
async def status_kdu(
self,
cluster_uuid: str,
kdu_instance: str
- ):
-
- return await self._status_kdu(cluster_uuid=cluster_uuid, kdu_instance=kdu_instance, show_error_log=True)
+ ) -> str:
+ # call internal function
+ return await self._status_kdu(
+ cluster_uuid=cluster_uuid,
+ kdu_instance=kdu_instance,
+ show_error_log=True,
+ return_text=True
+ )
"""
##################################################################################################
##################################################################################################
"""
+ async def _exec_inspect_comand(
+ self,
+ inspect_command: str,
+ kdu_model: str,
+ repo_url: str = None
+ ):
+
+ repo_str = ''
+ if repo_url:
+ repo_str = ' --repo {}'.format(repo_url)
+ idx = kdu_model.find('/')
+ if idx >= 0:
+ idx += 1
+ kdu_model = kdu_model[idx:]
+
+ inspect_command = '{} inspect {} {}{}'.format(self._helm_command, inspect_command, kdu_model, repo_str)
+ output, rc = await self._local_async_exec(command=inspect_command, encode_utf8=True)
+
+ return output
+
async def _status_kdu(
self,
cluster_uuid: str,
kdu_instance: str,
- show_error_log: bool = False
+ show_error_log: bool = False,
+ return_text: bool = False
):
self.debug('status of kdu_instance {}'.format(kdu_instance))
show_error_log=show_error_log
)
+ if return_text:
+ return str(output)
+
if rc != 0:
return None
def _generate_release_name(
chart_name: str
):
+ # check embeded chart (file or dir)
+ if chart_name.startswith('/'):
+ # extract file or directory name
+ chart_name = chart_name[chart_name.rfind('/')+1:]
+ # check URL
+ elif '://' in chart_name:
+ # extract last portion of URL
+ chart_name = chart_name[chart_name.rfind('/')+1:]
+
name = ''
for c in chart_name:
if c.isalpha() or c.isnumeric():
def get_random_number():
r = random.randrange(start=1, stop=99999999)
s = str(r)
- s = s.rjust(width=10, fillchar=' ')
+ s = s.rjust(10, '0')
return s
name = name + get_random_number()
kdu_instance: str
) -> bool:
- status = await self.status_kdu(cluster_uuid=cluster_uuid, kdu_instance=kdu_instance)
+ status = await self._status_kdu(cluster_uuid=cluster_uuid, kdu_instance=kdu_instance, return_text=False)
# extract info.status.resources-> str
# format:
# 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):
- params_str = ''
if params and len(params) > 0:
kube_dir, helm_dir, config_filename, cluster_dir = \
if not os.path.exists(cluster_dir):
msg = 'Base cluster dir {} does not exist'.format(cluster_dir)
self.error(msg)
- raise Exception(msg)
+ raise K8sException(msg)
# kube dir
kube_dir = cluster_dir + '/' + '.kube'
if not os.path.exists(kube_dir):
msg = 'Kube config dir {} does not exist'.format(kube_dir)
self.error(msg)
- raise Exception(msg)
+ raise K8sException(msg)
# helm home dir
helm_dir = cluster_dir + '/' + '.helm'
if not os.path.exists(helm_dir):
msg = 'Helm config dir {} does not exist'.format(helm_dir)
self.error(msg)
- raise Exception(msg)
+ raise K8sException(msg)
config_filename = kube_dir + '/config'
return kube_dir, helm_dir, config_filename, cluster_dir
self,
command: str,
raise_exception_on_error: bool = False,
- show_error_log: bool = True
+ show_error_log: bool = True,
+ encode_utf8: bool = False
) -> (str, int):
command = K8sHelmConnector._remove_multiple_spaces(command)
output = ''
if stdout:
output = stdout.decode('utf-8').strip()
+ # output = stdout.decode()
if stderr:
output = stderr.decode('utf-8').strip()
+ # output = stderr.decode()
if return_code != 0 and show_error_log:
self.debug('Return code (FAIL): {}\nOutput:\n{}'.format(return_code, output))
self.debug('Return code: {}'.format(return_code))
if raise_exception_on_error and return_code != 0:
- raise Exception(output)
+ raise K8sException(output)
+
+ if encode_utf8:
+ output = output.encode('utf-8').strip()
+ output = str(output).replace('\\n', '\n')
return output, return_code
+ except asyncio.CancelledError:
+ raise
+ except K8sException:
+ raise
except Exception as e:
msg = 'Exception executing command: {} -> {}'.format(command, e)
- if show_error_log:
- self.error(msg)
- return '', -1
-
- def _remote_exec(
- self,
- hostname: str,
- username: str,
- password: str,
- command: str,
- timeout: int = 10
- ) -> (str, int):
-
- command = K8sHelmConnector._remove_multiple_spaces(command)
- self.debug('Executing sync remote ssh command: {}'.format(command))
-
- ssh = paramiko.SSHClient()
- ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
- ssh.connect(hostname=hostname, username=username, password=password)
- ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command(command=command, timeout=timeout)
- output = ssh_stdout.read().decode('utf-8')
- error = ssh_stderr.read().decode('utf-8')
- if error:
- self.error('ERROR: {}'.format(error))
- return_code = 1
- else:
- return_code = 0
- output = output.replace('\\n', '\n')
- self.debug('OUTPUT: {}'.format(output))
-
- return output, return_code
+ self.error(msg)
+ if raise_exception_on_error:
+ raise K8sException(e) from e
+ else:
+ return '', -1
def _check_file_exists(self, filename: str, exception_if_not_exists: bool = False):
self.debug('Checking if file {} exists...'.format(filename))
msg = 'File {} does not exist'.format(filename)
if exception_if_not_exists:
self.error(msg)
- raise Exception(msg)
-
+ raise K8sException(msg)
projects / osm / N2VC.git / blobdiff