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