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