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