osm_lcm/lcm_helm_conn.py

   1 ##
   2 # Copyright 2020 Telefonica Investigacion y Desarrollo, S.A.U.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #    http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  13 # implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #
  17 ##
  18 import functools
  19 import yaml
  20 import asyncio
  21 import socket
  22 import uuid
  23
  24 from grpclib.client import Channel
  25
  26 from osm_lcm.frontend_pb2 import PrimitiveRequest
  27 from osm_lcm.frontend_pb2 import SshKeyRequest, SshKeyReply
  28 from osm_lcm.frontend_grpc import FrontendExecutorStub
  29
  30 from n2vc.n2vc_conn import N2VCConnector
  31 from n2vc.k8s_helm_conn import K8sHelmConnector
  32 from n2vc.exceptions import N2VCBadArgumentsException, N2VCException, N2VCExecutionException
  33
  34 from osm_lcm.lcm_utils import deep_get
  35
  36
  37 def retryer(max_wait_time=60, delay_time=10):
  38     def wrapper(func):
  39         retry_exceptions = (
  40             ConnectionRefusedError
  41         )
  42
  43         @functools.wraps(func)
  44         async def wrapped(*args, **kwargs):
  45             wait_time = max_wait_time
  46             while wait_time > 0:
  47                 try:
  48                     return await func(*args, **kwargs)
  49                 except retry_exceptions:
  50                     wait_time = wait_time - delay_time
  51                     await asyncio.sleep(delay_time)
  52                     continue
  53             else:
  54                 return ConnectionRefusedError
  55         return wrapped
  56     return wrapper
  57
  58
  59 class LCMHelmConn(N2VCConnector):
  60     _KUBECTL_OSM_NAMESPACE = "osm"
  61     _KUBECTL_OSM_CLUSTER_NAME = "_system-osm-k8s"
  62     _EE_SERVICE_PORT = 50050
  63
  64     # Time beetween retries
  65     _EE_RETRY_DELAY = 10
  66     # Initial max retry time
  67     _MAX_INITIAL_RETRY_TIME = 300
  68     # Other retry time
  69     _MAX_RETRY_TIME = 30
  70
  71     def __init__(self,
  72                  db: object,
  73                  fs: object,
  74                  log: object = None,
  75                  loop: object = None,
  76                  url: str = None,
  77                  username: str = None,
  78                  vca_config: dict = None,
  79                  on_update_db=None, ):
  80         """
  81         Initialize EE helm connector.
  82         """
  83
  84         # parent class constructor
  85         N2VCConnector.__init__(
  86             self,
  87             db=db,
  88             fs=fs,
  89             log=log,
  90             loop=loop,
  91             url=url,
  92             username=username,
  93             vca_config=vca_config,
  94             on_update_db=on_update_db,
  95         )
  96
  97         self.log.debug("Initialize helm N2VC connector")
  98
  99         # TODO - Obtain data from configuration
 100         self._ee_service_port = self._EE_SERVICE_PORT
 101
 102         self._retry_delay = self._EE_RETRY_DELAY
 103         self._max_retry_time = self._MAX_RETRY_TIME
 104         self._initial_retry_time = self._MAX_INITIAL_RETRY_TIME
 105
 106         # initialize helm connector
 107         self._k8sclusterhelm = K8sHelmConnector(
 108             kubectl_command=self.vca_config.get("kubectlpath"),
 109             helm_command=self.vca_config.get("helmpath"),
 110             fs=self.fs,
 111             log=self.log,
 112             db=self.db,
 113             on_update_db=None,
 114         )
 115
 116         self._system_cluster_id = None
 117         self.log.info("Helm N2VC connector initialized")
 118
 119     # TODO - ¿reuse_ee_id?
 120     async def create_execution_environment(self,
 121                                            namespace: str,
 122                                            db_dict: dict,
 123                                            reuse_ee_id: str = None,
 124                                            progress_timeout: float = None,
 125                                            total_timeout: float = None,
 126                                            artifact_path: str = None,
 127                                            vca_type: str = None) -> (str, dict):
 128         """
 129         Creates a new helm execution environment deploying the helm-chat indicated in the
 130         attifact_path
 131         :param str namespace: This param is not used, all helm charts are deployed in the osm
 132         system namespace
 133         :param dict db_dict: where to write to database when the status changes.
 134             It contains a dictionary with {collection: str, filter: {},  path: str},
 135                 e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path:
 136                 "_admin.deployed.VCA.3"}
 137         :param str reuse_ee_id: ee id from an older execution. TODO - right now this params is not used
 138         :param float progress_timeout:
 139         :param float total_timeout:
 140         :param str artifact_path  path of package content
 141         :param str vca_type  Type of vca, not used as assumed of type helm
 142         :returns str, dict: id of the new execution environment including namespace.helm_id
 143         and credentials object set to None as all credentials should be osm kubernetes .kubeconfig
 144         """
 145
 146         self.log.info(
 147             "create_execution_environment: namespace: {}, artifact_path: {}, db_dict: {}, "
 148             "reuse_ee_id: {}".format(
 149                 namespace, artifact_path, db_dict, reuse_ee_id)
 150         )
 151
 152         # Validate artifact-path is provided
 153         if artifact_path is None or len(artifact_path) == 0:
 154             raise N2VCBadArgumentsException(
 155                 message="artifact_path is mandatory", bad_args=["artifact_path"]
 156             )
 157
 158         # Validate artifact-path exists
 159
 160         # remove / in charm path
 161         while artifact_path.find("//") >= 0:
 162             artifact_path = artifact_path.replace("//", "/")
 163
 164         # check charm path
 165         if self.fs.file_exists(artifact_path):
 166             helm_chart_path = artifact_path
 167         else:
 168             msg = "artifact path does not exist: {}".format(artifact_path)
 169             raise N2VCBadArgumentsException(message=msg, bad_args=["artifact_path"])
 170
 171         if artifact_path.startswith("/"):
 172             full_path = self.fs.path + helm_chart_path
 173         else:
 174             full_path = self.fs.path + "/" + helm_chart_path
 175
 176         try:
 177             # Call helm conn install
 178             # Obtain system cluster id from database
 179             system_cluster_uuid = self._get_system_cluster_id()
 180
 181             self.log.debug("install helm chart: {}".format(full_path))
 182             helm_id = await self._k8sclusterhelm.install(system_cluster_uuid, kdu_model=full_path,
 183                                                          namespace=self._KUBECTL_OSM_NAMESPACE,
 184                                                          db_dict=db_dict,
 185                                                          timeout=progress_timeout)
 186
 187             ee_id = "{}.{}".format(self._KUBECTL_OSM_NAMESPACE, helm_id)
 188             return ee_id, None
 189         except Exception as e:
 190             self.log.error("Error deploying chart ee: {}".format(e), exc_info=True)
 191             raise N2VCException("Error deploying chart ee: {}".format(e))
 192
 193     async def register_execution_environment(self, namespace: str, credentials: dict, db_dict: dict,
 194                                              progress_timeout: float = None, total_timeout: float = None) -> str:
 195         # nothing to do
 196         pass
 197
 198     async def install_configuration_sw(self,
 199                                        ee_id: str,
 200                                        artifact_path: str,
 201                                        db_dict: dict,
 202                                        progress_timeout: float = None,
 203                                        total_timeout: float = None,
 204                                        config: dict = None,
 205                                        ):
 206         # nothing to do
 207         pass
 208
 209     async def add_relation(self, ee_id_1: str, ee_id_2: str, endpoint_1: str, endpoint_2: str):
 210         # nothing to do
 211         pass
 212
 213     async def remove_relation(self):
 214         # nothing to to
 215         pass
 216
 217     async def get_status(self, namespace: str, yaml_format: bool = True):
 218         # not used for this connector
 219         pass
 220
 221     async def get_ee_ssh_public__key(self, ee_id: str, db_dict: dict, progress_timeout: float = None,
 222                                      total_timeout: float = None) -> str:
 223         """
 224         Obtains ssh-public key from ee executing GetSShKey method from the ee.
 225
 226         :param str ee_id: the id of the execution environment returned by
 227             create_execution_environment or register_execution_environment
 228         :param dict db_dict:
 229         :param float progress_timeout:
 230         :param float total_timeout:
 231         :returns: public key of the execution environment
 232         """
 233
 234         self.log.info(
 235             "get_ee_ssh_public_key: ee_id: {}, db_dict: {}".format(
 236                 ee_id, db_dict)
 237         )
 238
 239         # check arguments
 240         if ee_id is None or len(ee_id) == 0:
 241             raise N2VCBadArgumentsException(
 242                 message="ee_id is mandatory", bad_args=["ee_id"]
 243             )
 244
 245         try:
 246             # Obtain ip_addr for the ee service, it is resolved by dns from the ee name by kubernetes
 247             namespace, helm_id = self._get_ee_id_parts(ee_id)
 248             ip_addr = socket.gethostbyname(helm_id)
 249
 250             # Obtain ssh_key from the ee, this method will implement retries to allow the ee
 251             # install libraries and start successfully
 252             ssh_key = await self._get_ssh_key(ip_addr)
 253             return ssh_key
 254         except Exception as e:
 255             self.log.error("Error obtaining ee ssh_key: {}".format(e), exc_info=True)
 256             raise N2VCException("Error obtaining ee ssh_ke: {}".format(e))
 257
 258     async def exec_primitive(self, ee_id: str, primitive_name: str, params_dict: dict, db_dict: dict = None,
 259                              progress_timeout: float = None, total_timeout: float = None) -> str:
 260         """
 261         Execute a primitive in the execution environment
 262
 263         :param str ee_id: the one returned by create_execution_environment or
 264             register_execution_environment with the format namespace.helm_id
 265         :param str primitive_name: must be one defined in the software. There is one
 266             called 'config', where, for the proxy case, the 'credentials' of VM are
 267             provided
 268         :param dict params_dict: parameters of the action
 269         :param dict db_dict: where to write into database when the status changes.
 270                         It contains a dict with
 271                             {collection: <str>, filter: {},  path: <str>},
 272                             e.g. {collection: "nslcmops", filter:
 273                                 {_id: <nslcmop_id>, path: "_admin.VCA"}
 274                         It will be used to store information about intermediate notifications
 275         :param float progress_timeout:
 276         :param float total_timeout:
 277         :returns str: primitive result, if ok. It raises exceptions in case of fail
 278         """
 279
 280         self.log.info("exec primitive for ee_id : {}, primitive_name: {}, params_dict: {}, db_dict: {}".format(
 281             ee_id, primitive_name, params_dict, db_dict
 282         ))
 283
 284         # check arguments
 285         if ee_id is None or len(ee_id) == 0:
 286             raise N2VCBadArgumentsException(
 287                 message="ee_id is mandatory", bad_args=["ee_id"]
 288             )
 289         if primitive_name is None or len(primitive_name) == 0:
 290             raise N2VCBadArgumentsException(
 291                 message="action_name is mandatory", bad_args=["action_name"]
 292             )
 293         if params_dict is None:
 294             params_dict = dict()
 295
 296         try:
 297             namespace, helm_id = self._get_ee_id_parts(ee_id)
 298             ip_addr = socket.gethostbyname(helm_id)
 299         except Exception as e:
 300             self.log.error("Error getting ee ip ee: {}".format(e))
 301             raise N2VCException("Error getting ee ip ee: {}".format(e))
 302
 303         if primitive_name == "config":
 304             try:
 305                 # Execute config primitive, higher timeout to check the case ee is starting
 306                 status, detailed_message = await self._execute_config_primitive(ip_addr, params_dict, db_dict=db_dict)
 307                 self.log.debug("Executed config primitive ee_id_ {}, status: {}, message: {}".format(
 308                     ee_id, status, detailed_message))
 309                 if status != "OK":
 310                     self.log.error("Error configuring helm ee, status: {}, message: {}".format(
 311                         status, detailed_message))
 312                     raise N2VCExecutionException(
 313                         message="Error configuring helm ee_id: {}, status: {}, message: {}: ".format(
 314                             ee_id, status, detailed_message
 315                         ),
 316                         primitive_name=primitive_name,
 317                     )
 318             except Exception as e:
 319                 self.log.error("Error configuring helm ee: {}".format(e))
 320                 raise N2VCExecutionException(
 321                     message="Error configuring helm ee_id: {}, {}".format(
 322                         ee_id, e
 323                     ),
 324                     primitive_name=primitive_name,
 325                 )
 326             return "CONFIG OK"
 327         else:
 328             try:
 329                 # Execute primitive
 330                 status, detailed_message = await self._execute_primitive(ip_addr, primitive_name,
 331                                                                          params_dict, db_dict=db_dict)
 332                 self.log.debug("Executed primitive {} ee_id_ {}, status: {}, message: {}".format(
 333                     primitive_name, ee_id, status, detailed_message))
 334                 if status != "OK" and status != "PROCESSING":
 335                     self.log.error(
 336                         "Execute primitive {} returned not ok status: {}, message: {}".format(
 337                             primitive_name, status, detailed_message)
 338                     )
 339                     raise N2VCExecutionException(
 340                         message="Execute primitive {} returned not ok status: {}, message: {}".format(
 341                             primitive_name, status, detailed_message
 342                         ),
 343                         primitive_name=primitive_name,
 344                     )
 345             except Exception as e:
 346                 self.log.error(
 347                     "Error executing primitive {}: {}".format(primitive_name, e)
 348                 )
 349                 raise N2VCExecutionException(
 350                     message="Error executing primitive {} into ee={} : {}".format(
 351                         primitive_name, ee_id, e
 352                     ),
 353                     primitive_name=primitive_name,
 354                 )
 355             return detailed_message
 356
 357     async def deregister_execution_environments(self):
 358         # nothing to be done
 359         pass
 360
 361     async def delete_execution_environment(self, ee_id: str, db_dict: dict = None, total_timeout: float = None):
 362         """
 363         Delete an execution environment
 364         :param str ee_id: id of the execution environment to delete, included namespace.helm_id
 365         :param dict db_dict: where to write into database when the status changes.
 366                         It contains a dict with
 367                             {collection: <str>, filter: {},  path: <str>},
 368                             e.g. {collection: "nsrs", filter:
 369                                 {_id: <nsd-id>, path: "_admin.deployed.VCA.3"}
 370         :param float total_timeout:
 371         """
 372
 373         self.log.info("ee_id: {}".format(ee_id))
 374
 375         # check arguments
 376         if ee_id is None:
 377             raise N2VCBadArgumentsException(
 378                 message="ee_id is mandatory", bad_args=["ee_id"]
 379             )
 380
 381         try:
 382
 383             # Obtain cluster_uuid
 384             system_cluster_uuid = self._get_system_cluster_id()
 385
 386             # Get helm_id
 387             namespace, helm_id = self._get_ee_id_parts(ee_id)
 388
 389             # Uninstall chart
 390             await self._k8sclusterhelm.uninstall(system_cluster_uuid, helm_id)
 391             self.log.info("ee_id: {} deleted".format(ee_id))
 392         except Exception as e:
 393             self.log.error("Error deleting ee id: {}: {}".format(ee_id, e), exc_info=True)
 394             raise N2VCException("Error deleting ee id {}: {}".format(ee_id, e))
 395
 396     async def delete_namespace(self, namespace: str, db_dict: dict = None, total_timeout: float = None):
 397         # method not implemented for this connector, execution environments must be deleted individually
 398         pass
 399
 400     async def install_k8s_proxy_charm(
 401         self,
 402         charm_name: str,
 403         namespace: str,
 404         artifact_path: str,
 405         db_dict: dict,
 406         progress_timeout: float = None,
 407         total_timeout: float = None,
 408         config: dict = None,
 409     ) -> str:
 410         pass
 411
 412     @retryer(max_wait_time=_MAX_INITIAL_RETRY_TIME, delay_time=_EE_RETRY_DELAY)
 413     async def _get_ssh_key(self, ip_addr):
 414         channel = Channel(ip_addr, self._ee_service_port)
 415         try:
 416             stub = FrontendExecutorStub(channel)
 417             self.log.debug("get ssh key, ip_addr: {}".format(ip_addr))
 418             reply: SshKeyReply = await stub.GetSshKey(SshKeyRequest())
 419             return reply.message
 420         finally:
 421             channel.close()
 422
 423     @retryer(max_wait_time=_MAX_INITIAL_RETRY_TIME, delay_time=_EE_RETRY_DELAY)
 424     async def _execute_config_primitive(self, ip_addr, params, db_dict=None):
 425         return await self._execute_primitive_internal(ip_addr, "config", params, db_dict=db_dict)
 426
 427     @retryer(max_wait_time=_MAX_RETRY_TIME, delay_time=_EE_RETRY_DELAY)
 428     async def _execute_primitive(self, ip_addr, primitive_name, params, db_dict=None):
 429         return await  self._execute_primitive_internal(ip_addr, primitive_name, params, db_dict=db_dict)
 430
 431     async def _execute_primitive_internal(self, ip_addr, primitive_name, params, db_dict=None):
 432
 433         channel = Channel(ip_addr, self._ee_service_port)
 434         try:
 435             stub = FrontendExecutorStub(channel)
 436             async with stub.RunPrimitive.open() as stream:
 437                 primitive_id = str(uuid.uuid1())
 438                 result = None
 439                 self.log.debug("Execute primitive internal: id:{}, name:{}, params: {}".
 440                                format(primitive_id, primitive_name, params))
 441                 await stream.send_message(
 442                     PrimitiveRequest(id=primitive_id, name=primitive_name, params=yaml.dump(params)), end=True)
 443                 async for reply in stream:
 444                     self.log.debug("Received reply: {}".format(reply))
 445                     result = reply
 446                     # If db_dict provided write notifs in database
 447                     if db_dict:
 448                         self._write_op_detailed_status(db_dict, reply.status, reply.detailed_message)
 449                 if result:
 450                     return reply.status, reply.detailed_message
 451                 else:
 452                     return "ERROR", "No result received"
 453         finally:
 454             channel.close()
 455
 456     def _write_op_detailed_status(self, db_dict, status, detailed_message):
 457
 458         # write ee_id to database: _admin.deployed.VCA.x
 459         try:
 460             the_table = db_dict["collection"]
 461             the_filter = db_dict["filter"]
 462             update_dict = {"detailed-status": "{}: {}".format(status, detailed_message)}
 463             # self.log.debug('Writing ee_id to database: {}'.format(the_path))
 464             self.db.set_one(
 465                 table=the_table,
 466                 q_filter=the_filter,
 467                 update_dict=update_dict,
 468                 fail_on_empty=True,
 469             )
 470         except asyncio.CancelledError:
 471             raise
 472         except Exception as e:
 473             self.log.error("Error writing detailedStatus to database: {}".format(e))
 474
 475     def _get_system_cluster_id(self):
 476         if not self._system_cluster_id:
 477             db_k8cluster = self.db.get_one("k8sclusters", {"name": self._KUBECTL_OSM_CLUSTER_NAME})
 478             k8s_hc_id = deep_get(db_k8cluster, ("_admin", "helm-chart", "id"))
 479             self._system_cluster_id = k8s_hc_id
 480         return self._system_cluster_id
 481
 482     def _get_ee_id_parts(self, ee_id):
 483         namespace, _, helm_id = ee_id.partition('.')
 484         return namespace, helm_id