n2vc/n2vc_conn.py

   1 ##
   2 # Copyright 2019 Telefonica Investigacion y Desarrollo, S.A.U.
   3 # This file is part of OSM
   4 # All Rights Reserved.
   5 #
   6 # Licensed under the Apache License, Version 2.0 (the "License");
   7 # you may not use this file except in compliance with the License.
   8 # You may obtain a copy of the License at
   9 #
  10 #    http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS,
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  15 # implied.
  16 # See the License for the specific language governing permissions and
  17 # limitations under the License.
  18 #
  19 # For those usages not covered by the Apache License, Version 2.0 please
  20 # contact with: nfvlabs@tid.es
  21 ##
  22
  23
  24 import abc
  25 import asyncio
  26 import os
  27 import subprocess
  28 import shlex
  29 import time
  30 from enum import Enum
  31 from http import HTTPStatus
  32 from n2vc.loggable import Loggable
  33 from n2vc.exceptions import N2VCBadArgumentsException
  34 import yaml
  35
  36 from osm_common.dbmongo import DbException
  37
  38
  39 class N2VCDeploymentStatus(Enum):
  40     PENDING = 'pending'
  41     RUNNING = 'running'
  42     COMPLETED = 'completed'
  43     FAILED = 'failed'
  44     UNKNOWN = 'unknown'
  45
  46
  47 class N2VCConnector(abc.ABC, Loggable):
  48     """Generic N2VC connector
  49
  50     Abstract class
  51     """
  52
  53     """
  54     ##################################################################################################
  55     ########################################## P U B L I C ###########################################
  56     ##################################################################################################
  57     """
  58
  59     def __init__(
  60             self,
  61             db: object,
  62             fs: object,
  63             log: object,
  64             loop: object,
  65             url: str,
  66             username: str,
  67             vca_config: dict,
  68             on_update_db=None
  69     ):
  70         """Initialize N2VC abstract connector. It defines de API for VCA connectors
  71
  72         :param object db: Mongo object managing the MongoDB (repo common DbBase)
  73         :param object fs: FileSystem object managing the package artifacts (repo common FsBase)
  74         :param object log: the logging object to log to
  75         :param object loop: the loop to use for asyncio (default current thread loop)
  76         :param str url: a string that how to connect to the VCA (if needed, IP and port can be obtained from there)
  77         :param str username: the username to authenticate with VCA
  78         :param dict vca_config: Additional parameters for the specific VCA. For example, for juju it will contain:
  79             secret: The password to authenticate with
  80             public_key: The contents of the juju public SSH key
  81             ca_cert str: The CA certificate used to authenticate
  82         :param on_update_db: callback called when n2vc connector updates database. Received arguments:
  83             table: e.g. "nsrs"
  84             filter: e.g. {_id: <nsd-id> }
  85             path: e.g. "_admin.deployed.VCA.3."
  86             updated_data: e.g. , "{ _admin.deployed.VCA.3.status: 'xxx', etc }"
  87         """
  88
  89         # parent class
  90         Loggable.__init__(self, log=log, log_to_console=True, prefix='\nN2VC')
  91
  92         # check arguments
  93         if db is None:
  94             raise N2VCBadArgumentsException('Argument db is mandatory', ['db'])
  95         if fs is None:
  96             raise N2VCBadArgumentsException('Argument fs is mandatory', ['fs'])
  97
  98         self.info('url={}, username={}, vca_config={}'.format(url, username, vca_config))
  99
 100         # store arguments into self
 101         self.db = db
 102         self.fs = fs
 103         self.loop = loop or asyncio.get_event_loop()
 104         self.url = url
 105         self.username = username
 106         self.vca_config = vca_config
 107         self.on_update_db = on_update_db
 108
 109         # generate private/public key-pair
 110         self.private_key_path = None
 111         self.public_key_path = None
 112         self.get_public_key()
 113
 114     @abc.abstractmethod
 115     async def get_status(self, namespace: str, yaml_format: bool = True):
 116         """Get namespace status
 117
 118         :param namespace: we obtain ns from namespace
 119         :param yaml_format: returns a yaml string
 120         """
 121
 122     # TODO: review which public key
 123     def get_public_key(self) -> str:
 124         """Get the VCA ssh-public-key
 125
 126         Returns the SSH public key from local mahine, to be injected into virtual machines to
 127         be managed by the VCA.
 128         First run, a ssh keypair will be created.
 129         The public key is injected into a VM so that we can provision the
 130         machine with Juju, after which Juju will communicate with the VM
 131         directly via the juju agent.
 132         """
 133
 134         public_key = ''
 135
 136         # Find the path where we expect our key lives (~/.ssh)
 137         homedir = os.environ.get('HOME')
 138         if not homedir:
 139             self.warning('No HOME environment variable, using /tmp')
 140             homedir = '/tmp'
 141         sshdir = "{}/.ssh".format(homedir)
 142         if not os.path.exists(sshdir):
 143             os.mkdir(sshdir)
 144
 145         self.private_key_path = "{}/id_n2vc_rsa".format(sshdir)
 146         self.public_key_path = "{}.pub".format(self.private_key_path)
 147
 148         # If we don't have a key generated, then we have to generate it using ssh-keygen
 149         if not os.path.exists(self.private_key_path):
 150             cmd = "ssh-keygen -t {} -b {} -N '' -f {}".format(
 151                 "rsa",
 152                 "4096",
 153                 self.private_key_path
 154             )
 155             # run command with arguments
 156             subprocess.check_output(shlex.split(cmd))
 157
 158         # Read the public key. Only one public key (one line) in the file
 159         with open(self.public_key_path, "r") as file:
 160             public_key = file.readline()
 161
 162         return public_key
 163
 164     @abc.abstractmethod
 165     async def create_execution_environment(
 166         self,
 167         namespace: str,
 168         db_dict: dict,
 169         reuse_ee_id: str = None,
 170         progress_timeout: float = None,
 171         total_timeout: float = None
 172     ) -> (str, dict):
 173         """Create an Execution Environment. Returns when it is created or raises an exception on failing
 174
 175         :param str namespace: Contains a dot separate string.
 176                     LCM will use: [<nsi-id>].<ns-id>.<vnf-id>.<vdu-id>[-<count>]
 177         :param dict db_dict: where to write to database when the status changes.
 178             It contains a dictionary with {collection: str, filter: {},  path: str},
 179                 e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.VCA.3"}
 180         :param str reuse_ee_id: ee id from an older execution. It allows us to reuse an older environment
 181         :param float progress_timeout:
 182         :param float total_timeout:
 183         :returns str, dict: id of the new execution environment and credentials for it
 184                     (credentials can contains hostname, username, etc depending on underlying cloud)
 185         """
 186
 187     @abc.abstractmethod
 188     async def register_execution_environment(
 189         self,
 190         namespace: str,
 191         credentials: dict,
 192         db_dict: dict,
 193         progress_timeout: float = None,
 194         total_timeout: float = None
 195     ) -> str:
 196         """
 197         Register an existing execution environment at the VCA
 198
 199         :param str namespace: same as create_execution_environment method
 200         :param dict credentials: credentials to access the existing execution environment
 201                     (it can contains hostname, username, path to private key, etc depending on underlying cloud)
 202         :param dict db_dict: where to write to database when the status changes.
 203             It contains a dictionary with {collection: str, filter: {},  path: str},
 204                 e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.VCA.3"}
 205         :param float progress_timeout:
 206         :param float total_timeout:
 207         :returns str: id of the execution environment
 208         """
 209
 210     @abc.abstractmethod
 211     async def install_configuration_sw(
 212         self,
 213         ee_id: str,
 214         artifact_path: str,
 215         db_dict: dict,
 216         progress_timeout: float = None,
 217         total_timeout: float = None
 218     ):
 219         """
 220         Install the software inside the execution environment identified by ee_id
 221
 222         :param str ee_id: the id of the execution environment returned by create_execution_environment
 223                         or register_execution_environment
 224         :param str artifact_path: where to locate the artifacts (parent folder) using the self.fs
 225             the final artifact path will be a combination of this artifact_path and additional string from
 226             the config_dict (e.g. charm name)
 227         :param dict db_dict: where to write into database when the status changes.
 228                         It contains a dict with {collection: <str>, filter: {},  path: <str>},
 229                             e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.VCA.3"}
 230         :param float progress_timeout:
 231         :param float total_timeout:
 232         """
 233
 234     @abc.abstractmethod
 235     async def get_ee_ssh_public__key(
 236         self,
 237         ee_id: str,
 238         db_dict: dict,
 239         progress_timeout: float = None,
 240         total_timeout: float = None
 241     ) -> str:
 242         """
 243         Generate a priv/pub key pair in the execution environment and return the public key
 244
 245         :param str ee_id: the id of the execution environment returned by create_execution_environment
 246                         or register_execution_environment
 247         :param dict db_dict: where to write into database when the status changes.
 248                         It contains a dict with {collection: <str>, filter: {},  path: <str>},
 249                             e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.VCA.3"}
 250         :param float progress_timeout:
 251         :param float total_timeout:
 252         :returns: public key of the execution environment
 253                     For the case of juju proxy charm ssh-layered, it is the one returned by 'get-ssh-public-key'
 254                     primitive.
 255                     It raises a N2VC exception if fails
 256         """
 257
 258     @abc.abstractmethod
 259     async def add_relation(
 260         self,
 261         ee_id_1: str,
 262         ee_id_2: str,
 263         endpoint_1: str,
 264         endpoint_2: str
 265     ):
 266         """
 267         Add a relation between two Execution Environments (using their associated endpoints).
 268
 269         :param str ee_id_1: The id of the first execution environment
 270         :param str ee_id_2: The id of the second execution environment
 271         :param str endpoint_1: The endpoint in the first execution environment
 272         :param str endpoint_2: The endpoint in the second execution environment
 273         """
 274
 275     # TODO
 276     @abc.abstractmethod
 277     async def remove_relation(
 278         self
 279     ):
 280         """
 281         """
 282
 283     # TODO
 284     @abc.abstractmethod
 285     async def deregister_execution_environments(
 286         self
 287     ):
 288         """
 289         """
 290
 291     @abc.abstractmethod
 292     async def delete_namespace(
 293         self,
 294         namespace: str,
 295         db_dict: dict = None,
 296         total_timeout: float = None
 297     ):
 298         """
 299         Remove a network scenario and its execution environments
 300         :param namespace: [<nsi-id>].<ns-id>
 301         :param dict db_dict: where to write into database when the status changes.
 302                         It contains a dict with {collection: <str>, filter: {},  path: <str>},
 303                             e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.VCA.3"}
 304         :param float total_timeout:
 305         """
 306
 307     @abc.abstractmethod
 308     async def delete_execution_environment(
 309         self,
 310         ee_id: str,
 311         db_dict: dict = None,
 312         total_timeout: float = None
 313     ):
 314         """
 315         Delete an execution environment
 316         :param str ee_id: id of the execution environment to delete
 317         :param dict db_dict: where to write into database when the status changes.
 318                         It contains a dict with {collection: <str>, filter: {},  path: <str>},
 319                             e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.VCA.3"}
 320         :param float total_timeout:
 321         """
 322
 323     @abc.abstractmethod
 324     async def exec_primitive(
 325         self,
 326         ee_id: str,
 327         primitive_name: str,
 328         params_dict: dict,
 329         db_dict: dict = None,
 330         progress_timeout: float = None,
 331         total_timeout: float = None
 332     ) -> str:
 333         """
 334         Execute a primitive in the execution environment
 335
 336         :param str ee_id: the one returned by create_execution_environment or register_execution_environment
 337         :param str primitive_name: must be one defined in the software. There is one called 'config',
 338             where, for the proxy case, the 'credentials' of VM are provided
 339         :param dict params_dict: parameters of the action
 340         :param dict db_dict: where to write into database when the status changes.
 341                         It contains a dict with {collection: <str>, filter: {},  path: <str>},
 342                             e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path: "_admin.deployed.VCA.3"}
 343         :param float progress_timeout:
 344         :param float total_timeout:
 345         :returns str: primitive result, if ok. It raises exceptions in case of fail
 346         """
 347
 348     async def disconnect(self):
 349         """
 350         Disconnect from VCA
 351         """
 352
 353     """
 354     ##################################################################################################
 355     ########################################## P R I V A T E #########################################
 356     ##################################################################################################
 357     """
 358
 359     def _get_namespace_components(self, namespace: str) -> (str, str, str, str, str):
 360         """
 361         Split namespace components
 362
 363         :param namespace: [<nsi-id>].<ns-id>.<vnf-id>.<vdu-id>[-<count>]
 364         :return: nsi_id, ns_id, vnf_id, vdu_id, vdu_count
 365         """
 366
 367         # check parameters
 368         if namespace is None or len(namespace) == 0:
 369             raise N2VCBadArgumentsException('Argument namespace is mandatory', ['namespace'])
 370
 371         # split namespace components
 372         parts = namespace.split('.')
 373         nsi_id = None
 374         ns_id = None
 375         vnf_id = None
 376         vdu_id = None
 377         vdu_count = None
 378         if len(parts) > 0 and len(parts[0]) > 0:
 379             nsi_id = parts[0]
 380         if len(parts) > 1 and len(parts[1]) > 0:
 381             ns_id = parts[1]
 382         if len(parts) > 2 and len(parts[2]) > 0:
 383             vnf_id = parts[2]
 384         if len(parts) > 3 and len(parts[3]) > 0:
 385             vdu_id = parts[3]
 386             vdu_parts = parts[3].split('-')
 387             if len(vdu_parts) > 1:
 388                 vdu_id = vdu_parts[0]
 389                 vdu_count = vdu_parts[1]
 390
 391         return nsi_id, ns_id, vnf_id, vdu_id, vdu_count
 392
 393     async def write_app_status_to_db(
 394             self,
 395             db_dict: dict,
 396             status: N2VCDeploymentStatus,
 397             detailed_status: str,
 398             vca_status: str,
 399             entity_type: str
 400     ):
 401         if not db_dict:
 402             self.debug('No db_dict => No database write')
 403             return
 404
 405         # self.debug('status={} / detailed-status={} / VCA-status={} / entity_type={}'
 406         #            .format(str(status.value), detailed_status, vca_status, entity_type))
 407
 408         try:
 409
 410             the_table = db_dict['collection']
 411             the_filter = db_dict['filter']
 412             the_path = db_dict['path']
 413             if not the_path[-1] == '.':
 414                 the_path = the_path + '.'
 415             update_dict = {
 416                 the_path + 'status': str(status.value),
 417                 the_path + 'detailed-status': detailed_status,
 418                 the_path + 'VCA-status': vca_status,
 419                 the_path + 'entity-type': entity_type,
 420                 the_path + 'status-time': str(time.time()),
 421             }
 422
 423             self.db.set_one(
 424                 table=the_table,
 425                 q_filter=the_filter,
 426                 update_dict=update_dict,
 427                 fail_on_empty=True
 428             )
 429
 430             # database callback
 431             if self.on_update_db:
 432                 if asyncio.iscoroutinefunction(self.on_update_db):
 433                     await self.on_update_db(the_table, the_filter, the_path, update_dict)
 434                 else:
 435                     self.on_update_db(the_table, the_filter, the_path, update_dict)
 436
 437         except DbException as e:
 438             if e.http_code == HTTPStatus.NOT_FOUND:
 439                 self.error('NOT_FOUND error: Exception writing status to database: {}'.format(e))
 440             else:
 441                 self.info('Exception writing status to database: {}'.format(e))
 442
 443
 444 def juju_status_2_osm_status(type: str, status: str) -> N2VCDeploymentStatus:
 445     if type == 'application' or type == 'unit':
 446         if status in ['waiting', 'maintenance']:
 447             return N2VCDeploymentStatus.RUNNING
 448         if status in ['error']:
 449                 return N2VCDeploymentStatus.FAILED
 450         elif status in ['active']:
 451             return N2VCDeploymentStatus.COMPLETED
 452         elif status in ['blocked']:
 453             return N2VCDeploymentStatus.RUNNING
 454         else:
 455             return N2VCDeploymentStatus.UNKNOWN
 456     elif type == 'action':
 457         if status in ['running']:
 458             return N2VCDeploymentStatus.RUNNING
 459         elif status in ['completed']:
 460             return N2VCDeploymentStatus.COMPLETED
 461         else:
 462             return N2VCDeploymentStatus.UNKNOWN
 463     elif type == 'machine':
 464         if status in ['pending']:
 465             return N2VCDeploymentStatus.PENDING
 466         elif status in ['started']:
 467             return N2VCDeploymentStatus.COMPLETED
 468         else:
 469             return N2VCDeploymentStatus.UNKNOWN
 470
 471     return N2VCDeploymentStatus.FAILED
 472
 473
 474 def obj_to_yaml(obj: object) -> str:
 475     # dump to yaml
 476     dump_text = yaml.dump(obj, default_flow_style=False, indent=2)
 477     # split lines
 478     lines = dump_text.splitlines()
 479     # remove !!python/object tags
 480     yaml_text = ''
 481     for line in lines:
 482         index = line.find('!!python/object')
 483         if index >= 0:
 484             line = line[:index]
 485         yaml_text += line + '\n'
 486     return yaml_text
 487
 488
 489 def obj_to_dict(obj: object) -> dict:
 490     # convert obj to yaml
 491     yaml_text = obj_to_yaml(obj)
 492     # parse to dict
 493     return yaml.load(yaml_text, Loader=yaml.Loader)