osm_lcm/lcm.py

   1 #!/usr/bin/python3
   2 # -*- coding: utf-8 -*-
   3
   4 import asyncio
   5 import yaml
   6 import ROclient
   7 import logging
   8 import logging.handlers
   9 import getopt
  10 import functools
  11 import sys
  12 from osm_common import dbmemory
  13 from osm_common import dbmongo
  14 from osm_common import fslocal
  15 from osm_common import msglocal
  16 from osm_common import msgkafka
  17 from osm_common.dbbase import DbException
  18 from osm_common.fsbase import FsException
  19 from osm_common.msgbase import MsgException
  20 from os import environ, path
  21 from n2vc.vnf import N2VC
  22 from n2vc import version as N2VC_version
  23
  24 from copy import deepcopy
  25 from http import HTTPStatus
  26 from time import time
  27
  28
  29 __author__ = "Alfonso Tierno"
  30
  31
  32 class LcmException(Exception):
  33     pass
  34
  35
  36 class Lcm:
  37
  38     def __init__(self, config_file):
  39         """
  40         Init, Connect to database, filesystem storage, and messaging
  41         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
  42         :return: None
  43         """
  44
  45         self.db = None
  46         self.msg = None
  47         self.fs = None
  48         self.pings_not_received = 1
  49
  50         # contains created tasks/futures to be able to cancel
  51         self.lcm_ns_tasks = {}
  52         self.lcm_vim_tasks = {}
  53         self.lcm_sdn_tasks = {}
  54         # logging
  55         self.logger = logging.getLogger('lcm')
  56         # load configuration
  57         config = self.read_config_file(config_file)
  58         self.config = config
  59         self.ro_config={
  60             "endpoint_url": "http://{}:{}/openmano".format(config["RO"]["host"], config["RO"]["port"]),
  61             "tenant":  config.get("tenant", "osm"),
  62             "logger_name": "lcm.ROclient",
  63             "loglevel": "ERROR",
  64         }
  65
  66         self.vca = config["VCA"]  # TODO VCA
  67         self.loop = None
  68
  69         # logging
  70         log_format_simple = "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)s %(message)s"
  71         log_formatter_simple = logging.Formatter(log_format_simple, datefmt='%Y-%m-%dT%H:%M:%S')
  72         config["database"]["logger_name"] = "lcm.db"
  73         config["storage"]["logger_name"] = "lcm.fs"
  74         config["message"]["logger_name"] = "lcm.msg"
  75         if "logfile" in config["global"]:
  76             file_handler = logging.handlers.RotatingFileHandler(config["global"]["logfile"],
  77                                                                 maxBytes=100e6, backupCount=9, delay=0)
  78             file_handler.setFormatter(log_formatter_simple)
  79             self.logger.addHandler(file_handler)
  80         else:
  81             str_handler = logging.StreamHandler()
  82             str_handler.setFormatter(log_formatter_simple)
  83             self.logger.addHandler(str_handler)
  84
  85         if config["global"].get("loglevel"):
  86             self.logger.setLevel(config["global"]["loglevel"])
  87
  88         # logging other modules
  89         for k1, logname in {"message": "lcm.msg", "database": "lcm.db", "storage": "lcm.fs"}.items():
  90             config[k1]["logger_name"] = logname
  91             logger_module = logging.getLogger(logname)
  92             if "logfile" in config[k1]:
  93                 file_handler = logging.handlers.RotatingFileHandler(config[k1]["logfile"],
  94                                                                     maxBytes=100e6, backupCount=9, delay=0)
  95                 file_handler.setFormatter(log_formatter_simple)
  96                 logger_module.addHandler(file_handler)
  97             if "loglevel" in config[k1]:
  98                 logger_module.setLevel(config[k1]["loglevel"])
  99
 100         self.n2vc = N2VC(
 101             log=self.logger,
 102             server=config['VCA']['host'],
 103             port=config['VCA']['port'],
 104             user=config['VCA']['user'],
 105             secret=config['VCA']['secret'],
 106             # TODO: This should point to the base folder where charms are stored,
 107             # if there is a common one (like object storage). Otherwise, leave
 108             # it unset and pass it via DeployCharms
 109             # artifacts=config['VCA'][''],
 110             artifacts=None,
 111         )
 112         # check version of N2VC
 113         # TODO enhance with int conversion or from distutils.version import LooseVersion
 114         # or with list(map(int, version.split(".")))
 115         if N2VC_version < "0.0.2":
 116             raise LcmException("Not compatible osm/N2VC version '{}'. Needed '0.0.2' or higher".format(N2VC_version))
 117         try:
 118             if config["database"]["driver"] == "mongo":
 119                 self.db = dbmongo.DbMongo()
 120                 self.db.db_connect(config["database"])
 121             elif config["database"]["driver"] == "memory":
 122                 self.db = dbmemory.DbMemory()
 123                 self.db.db_connect(config["database"])
 124             else:
 125                 raise LcmException("Invalid configuration param '{}' at '[database]':'driver'".format(
 126                     config["database"]["driver"]))
 127
 128             if config["storage"]["driver"] == "local":
 129                 self.fs = fslocal.FsLocal()
 130                 self.fs.fs_connect(config["storage"])
 131             else:
 132                 raise LcmException("Invalid configuration param '{}' at '[storage]':'driver'".format(
 133                     config["storage"]["driver"]))
 134
 135             if config["message"]["driver"] == "local":
 136                 self.msg = msglocal.MsgLocal()
 137                 self.msg.connect(config["message"])
 138             elif config["message"]["driver"] == "kafka":
 139                 self.msg = msgkafka.MsgKafka()
 140                 self.msg.connect(config["message"])
 141             else:
 142                 raise LcmException("Invalid configuration param '{}' at '[message]':'driver'".format(
 143                     config["storage"]["driver"]))
 144         except (DbException, FsException, MsgException) as e:
 145             self.logger.critical(str(e), exc_info=True)
 146             raise LcmException(str(e))
 147
 148     def update_db(self, item, _id, _desc):
 149         try:
 150             self.db.replace(item, _id, _desc)
 151         except DbException as e:
 152             self.logger.error("Updating {} _id={}: {}".format(item, _id, e))
 153
 154     def update_db_2(self, item, _id, _desc):
 155         try:
 156             self.db.set_one(item, {"_id": _id}, _desc)
 157         except DbException as e:
 158             self.logger.error("Updating {} _id={}: {}".format(item, _id, e))
 159
 160     async def vim_create(self, vim_content, order_id):
 161         vim_id = vim_content["_id"]
 162         logging_text = "Task vim_create={} ".format(vim_id)
 163         self.logger.debug(logging_text + "Enter")
 164         db_vim = None
 165         exc = None
 166         try:
 167             step = "Getting vim from db"
 168             db_vim = self.db.get_one("vim_accounts", {"_id": vim_id})
 169             if "_admin" not in db_vim:
 170                 db_vim["_admin"] = {}
 171             if "deployed" not in db_vim["_admin"]:
 172                 db_vim["_admin"]["deployed"] = {}
 173             db_vim["_admin"]["deployed"]["RO"] = None
 174
 175             step = "Creating vim at RO"
 176             RO = ROclient.ROClient(self.loop, **self.ro_config)
 177             vim_RO = deepcopy(vim_content)
 178             vim_RO.pop("_id", None)
 179             vim_RO.pop("_admin", None)
 180             vim_RO.pop("schema_version", None)
 181             vim_RO.pop("schema_type", None)
 182             vim_RO.pop("vim_tenant_name", None)
 183             vim_RO["type"] = vim_RO.pop("vim_type")
 184             vim_RO.pop("vim_user", None)
 185             vim_RO.pop("vim_password", None)
 186             desc = await RO.create("vim", descriptor=vim_RO)
 187             RO_vim_id = desc["uuid"]
 188             db_vim["_admin"]["deployed"]["RO"] = RO_vim_id
 189             self.update_db("vim_accounts", vim_id, db_vim)
 190
 191             step = "Attach vim to RO tenant"
 192             vim_RO = {"vim_tenant_name": vim_content["vim_tenant_name"],
 193                 "vim_username": vim_content["vim_user"],
 194                 "vim_password": vim_content["vim_password"],
 195                 "config": vim_content["config"]
 196             }
 197             desc = await RO.attach_datacenter(RO_vim_id , descriptor=vim_RO)
 198             db_vim["_admin"]["operationalState"] = "ENABLED"
 199             self.update_db("vim_accounts", vim_id, db_vim)
 200
 201             self.logger.debug(logging_text + "Exit Ok RO_vim_id".format(RO_vim_id))
 202             return RO_vim_id
 203
 204         except (ROclient.ROClientException, DbException) as e:
 205             self.logger.error(logging_text + "Exit Exception {}".format(e))
 206             exc = e
 207         except Exception as e:
 208             self.logger.critical(logging_text + "Exit Exception {}".format(e), exc_info=True)
 209             exc = e
 210         finally:
 211             if exc and db_vim:
 212                 db_vim["_admin"]["operationalState"] = "ERROR"
 213                 db_vim["_admin"]["detailed-status"] = "ERROR {}: {}".format(step , exc)
 214                 self.update_db("vim_accounts", vim_id, db_vim)
 215
 216     async def vim_edit(self, vim_content, order_id):
 217         vim_id = vim_content["_id"]
 218         logging_text = "Task vim_edit={} ".format(vim_id)
 219         self.logger.debug(logging_text + "Enter")
 220         db_vim = None
 221         exc = None
 222         step = "Getting vim from db"
 223         try:
 224             db_vim = self.db.get_one("vim_accounts", {"_id": vim_id})
 225             if db_vim.get("_admin") and db_vim["_admin"].get("deployed") and db_vim["_admin"]["deployed"].get("RO"):
 226                 RO_vim_id = db_vim["_admin"]["deployed"]["RO"]
 227                 step = "Editing vim at RO"
 228                 RO = ROclient.ROClient(self.loop, **self.ro_config)
 229                 vim_RO = deepcopy(vim_content)
 230                 vim_RO.pop("_id", None)
 231                 vim_RO.pop("_admin", None)
 232                 vim_RO.pop("schema_version", None)
 233                 vim_RO.pop("schema_type", None)
 234                 vim_RO.pop("vim_tenant_name", None)
 235                 vim_RO["type"] = vim_RO.pop("vim_type")
 236                 vim_RO.pop("vim_user", None)
 237                 vim_RO.pop("vim_password", None)
 238                 if vim_RO:
 239                     desc = await RO.edit("vim", RO_vim_id, descriptor=vim_RO)
 240
 241                 step = "Editing vim-account at RO tenant"
 242                 vim_RO = {}
 243                 for k in ("vim_tenant_name", "vim_password", "config"):
 244                     if k in vim_content:
 245                         vim_RO[k] = vim_content[k]
 246                 if "vim_user" in vim_content:
 247                     vim_content["vim_username"] = vim_content["vim_user"]
 248                 if vim_RO:
 249                     desc = await RO.edit("vim_account", RO_vim_id, descriptor=vim_RO)
 250                 db_vim["_admin"]["operationalState"] = "ENABLED"
 251                 self.update_db("vim_accounts", vim_id, db_vim)
 252
 253             self.logger.debug(logging_text + "Exit Ok RO_vim_id".format(RO_vim_id))
 254             return RO_vim_id
 255
 256         except (ROclient.ROClientException, DbException) as e:
 257             self.logger.error(logging_text + "Exit Exception {}".format(e))
 258             exc = e
 259         except Exception as e:
 260             self.logger.critical(logging_text + "Exit Exception {}".format(e), exc_info=True)
 261             exc = e
 262         finally:
 263             if exc and db_vim:
 264                 db_vim["_admin"]["operationalState"] = "ERROR"
 265                 db_vim["_admin"]["detailed-status"] = "ERROR {}: {}".format(step , exc)
 266                 self.update_db("vim_accounts", vim_id, db_vim)
 267
 268     async def vim_delete(self, vim_id, order_id):
 269         logging_text = "Task vim_delete={} ".format(vim_id)
 270         self.logger.debug(logging_text + "Enter")
 271         db_vim = None
 272         exc = None
 273         step = "Getting vim from db"
 274         try:
 275             db_vim = self.db.get_one("vim_accounts", {"_id": vim_id})
 276             if db_vim.get("_admin") and db_vim["_admin"].get("deployed") and db_vim["_admin"]["deployed"].get("RO"):
 277                 RO_vim_id = db_vim["_admin"]["deployed"]["RO"]
 278                 RO = ROclient.ROClient(self.loop, **self.ro_config)
 279                 step = "Detaching vim from RO tenant"
 280                 try:
 281                     await RO.detach_datacenter(RO_vim_id)
 282                 except ROclient.ROClientException as e:
 283                     if e.http_code == 404:  # not found
 284                         self.logger.debug(logging_text + "RO_vim_id={} already detached".format(RO_vim_id))
 285                     else:
 286                         raise
 287
 288                 step = "Deleting vim from RO"
 289                 try:
 290                     await RO.delete("vim", RO_vim_id)
 291                 except ROclient.ROClientException as e:
 292                     if e.http_code == 404:  # not found
 293                         self.logger.debug(logging_text + "RO_vim_id={} already deleted".format(RO_vim_id))
 294                     else:
 295                         raise
 296             else:
 297                 # nothing to delete
 298                 self.logger.error(logging_text + "Skipping. There is not RO information at database")
 299             self.db.del_one("vim_accounts", {"_id": vim_id})
 300             self.logger.debug("vim_delete task vim_id={} Exit Ok".format(vim_id))
 301             return None
 302
 303         except (ROclient.ROClientException, DbException) as e:
 304             self.logger.error(logging_text + "Exit Exception {}".format(e))
 305             exc = e
 306         except Exception as e:
 307             self.logger.critical(logging_text + "Exit Exception {}".format(e), exc_info=True)
 308             exc = e
 309         finally:
 310             if exc and db_vim:
 311                 db_vim["_admin"]["operationalState"] = "ERROR"
 312                 db_vim["_admin"]["detailed-status"] = "ERROR {}: {}".format(step , exc)
 313                 self.update_db("vim_accounts", vim_id, db_vim)
 314
 315     async def sdn_create(self, sdn_content, order_id):
 316         sdn_id = sdn_content["_id"]
 317         logging_text = "Task sdn_create={} ".format(sdn_id)
 318         self.logger.debug(logging_text + "Enter")
 319         db_sdn = None
 320         exc = None
 321         try:
 322             step = "Getting sdn from db"
 323             db_sdn = self.db.get_one("sdns", {"_id": sdn_id})
 324             if "_admin" not in db_sdn:
 325                 db_sdn["_admin"] = {}
 326             if "deployed" not in db_sdn["_admin"]:
 327                 db_sdn["_admin"]["deployed"] = {}
 328             db_sdn["_admin"]["deployed"]["RO"] =  None
 329
 330             step = "Creating sdn at RO"
 331             RO = ROclient.ROClient(self.loop, **self.ro_config)
 332             sdn_RO = deepcopy(sdn_content)
 333             sdn_RO.pop("_id", None)
 334             sdn_RO.pop("_admin", None)
 335             sdn_RO.pop("schema_version", None)
 336             sdn_RO.pop("schema_type", None)
 337             sdn_RO.pop("description", None)
 338             desc = await RO.create("sdn", descriptor=sdn_RO)
 339             RO_sdn_id = desc["uuid"]
 340             db_sdn["_admin"]["deployed"]["RO"] = RO_sdn_id
 341             db_sdn["_admin"]["operationalState"] = "ENABLED"
 342             self.update_db("sdns", sdn_id, db_sdn)
 343             self.logger.debug(logging_text + "Exit Ok RO_sdn_id".format(RO_sdn_id))
 344             return RO_sdn_id
 345
 346         except (ROclient.ROClientException, DbException) as e:
 347             self.logger.error(logging_text + "Exit Exception {}".format(e))
 348             exc = e
 349         except Exception as e:
 350             self.logger.critical(logging_text + "Exit Exception {}".format(e), exc_info=True)
 351             exc = e
 352         finally:
 353             if exc and db_sdn:
 354                 db_sdn["_admin"]["operationalState"] = "ERROR"
 355                 db_sdn["_admin"]["detailed-status"] = "ERROR {}: {}".format(step , exc)
 356                 self.update_db("sdns", sdn_id, db_sdn)
 357
 358     async def sdn_edit(self, sdn_content, order_id):
 359         sdn_id = sdn_content["_id"]
 360         logging_text = "Task sdn_edit={} ".format(sdn_id)
 361         self.logger.debug(logging_text + "Enter")
 362         db_sdn = None
 363         exc = None
 364         step = "Getting sdn from db"
 365         try:
 366             db_sdn = self.db.get_one("sdns", {"_id": sdn_id})
 367             if db_sdn.get("_admin") and db_sdn["_admin"].get("deployed") and db_sdn["_admin"]["deployed"].get("RO"):
 368                 RO_sdn_id = db_sdn["_admin"]["deployed"]["RO"]
 369                 RO = ROclient.ROClient(self.loop, **self.ro_config)
 370                 step = "Editing sdn at RO"
 371                 sdn_RO = deepcopy(sdn_content)
 372                 sdn_RO.pop("_id", None)
 373                 sdn_RO.pop("_admin", None)
 374                 sdn_RO.pop("schema_version", None)
 375                 sdn_RO.pop("schema_type", None)
 376                 sdn_RO.pop("description", None)
 377                 if sdn_RO:
 378                     desc = await RO.edit("sdn", RO_sdn_id, descriptor=sdn_RO)
 379                 db_sdn["_admin"]["operationalState"] = "ENABLED"
 380                 self.update_db("sdns", sdn_id, db_sdn)
 381
 382             self.logger.debug(logging_text + "Exit Ok RO_sdn_id".format(RO_sdn_id))
 383             return RO_sdn_id
 384
 385         except (ROclient.ROClientException, DbException) as e:
 386             self.logger.error(logging_text + "Exit Exception {}".format(e))
 387             exc = e
 388         except Exception as e:
 389             self.logger.critical(logging_text + "Exit Exception {}".format(e), exc_info=True)
 390             exc = e
 391         finally:
 392             if exc and db_sdn:
 393                 db_sdn["_admin"]["operationalState"] = "ERROR"
 394                 db_sdn["_admin"]["detailed-status"] = "ERROR {}: {}".format(step , exc)
 395                 self.update_db("sdns", sdn_id, db_sdn)
 396
 397     async def sdn_delete(self, sdn_id, order_id):
 398         logging_text = "Task sdn_delete={} ".format(sdn_id)
 399         self.logger.debug(logging_text + "Enter")
 400         db_sdn = None
 401         exc = None
 402         step = "Getting sdn from db"
 403         try:
 404             db_sdn = self.db.get_one("sdns", {"_id": sdn_id})
 405             if db_sdn.get("_admin") and db_sdn["_admin"].get("deployed") and db_sdn["_admin"]["deployed"].get("RO"):
 406                 RO_sdn_id = db_sdn["_admin"]["deployed"]["RO"]
 407                 RO = ROclient.ROClient(self.loop, **self.ro_config)
 408                 step = "Deleting sdn from RO"
 409                 try:
 410                     await RO.delete("sdn", RO_sdn_id)
 411                 except ROclient.ROClientException as e:
 412                     if e.http_code == 404:  # not found
 413                         self.logger.debug(logging_text + "RO_sdn_id={} already deleted".format(RO_sdn_id))
 414                     else:
 415                         raise
 416             else:
 417                 # nothing to delete
 418                 self.logger.error(logging_text + "Skipping. There is not RO information at database")
 419             self.db.del_one("sdns", {"_id": sdn_id})
 420             self.logger.debug("sdn_delete task sdn_id={} Exit Ok".format(sdn_id))
 421             return None
 422
 423         except (ROclient.ROClientException, DbException) as e:
 424             self.logger.error(logging_text + "Exit Exception {}".format(e))
 425             exc = e
 426         except Exception as e:
 427             self.logger.critical(logging_text + "Exit Exception {}".format(e), exc_info=True)
 428             exc = e
 429         finally:
 430             if exc and db_sdn:
 431                 db_sdn["_admin"]["operationalState"] = "ERROR"
 432                 db_sdn["_admin"]["detailed-status"] = "ERROR {}: {}".format(step , exc)
 433                 self.update_db("sdns", sdn_id, db_sdn)
 434
 435     def vnfd2RO(self, vnfd, new_id=None):
 436         """
 437         Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
 438         :param vnfd: input vnfd
 439         :param new_id: overrides vnf id if provided
 440         :return: copy of vnfd
 441         """
 442         ci_file = None
 443         try:
 444             vnfd_RO = deepcopy(vnfd)
 445             vnfd_RO.pop("_id", None)
 446             vnfd_RO.pop("_admin", None)
 447             if new_id:
 448                 vnfd_RO["id"] = new_id
 449             for vdu in vnfd_RO["vdu"]:
 450                 if "cloud-init-file" in vdu:
 451                     base_folder = vnfd["_admin"]["storage"]
 452                     clout_init_file = "{}/{}/cloud_init/{}".format(
 453                         base_folder["folder"],
 454                         base_folder["pkg-dir"],
 455                         vdu["cloud-init-file"]
 456                     )
 457                     ci_file = self.fs.file_open(clout_init_file, "r")
 458                     # TODO: detect if binary or text. Propose to read as binary and try to decode to utf8. If fails convert to base 64 or similar
 459                     clout_init_content = ci_file.read()
 460                     ci_file.close()
 461                     ci_file = None
 462                     vdu.pop("cloud-init-file", None)
 463                     vdu["cloud-init"] = clout_init_content
 464             return vnfd_RO
 465         except FsException as e:
 466             raise LcmException("Error reading file at vnfd {}: {} ".format(vnfd["_id"], e))
 467         finally:
 468             if ci_file:
 469                 ci_file.close()
 470
 471     def n2vc_callback(self, model_name, application_name, status, message, db_nsr, db_nslcmop, member_vnf_index, task=None):
 472         """
 473         Callback both for charm status change and task completion
 474         :param model_name: Charm model name
 475         :param application_name: Charm application name
 476         :param status: Can be
 477             - blocked: The unit needs manual intervention
 478             - maintenance: The unit is actively deploying/configuring
 479             - waiting: The unit is waiting for another charm to be ready
 480             - active: The unit is deployed, configured, and ready
 481             - error: The charm has failed and needs attention.
 482             - terminated: The charm has been destroyed
 483             - removing,
 484             - removed
 485         :param message: detailed message error
 486         :param db_nsr: nsr database content
 487         :param db_nslcmop: nslcmop database content
 488         :param member_vnf_index: NSD member-vnf-index
 489         :param task: None for charm status change, or task for completion task callback
 490         :return:
 491         """
 492         nsr_id = None
 493         nslcmop_id = None
 494         update_nsr = update_nslcmop = False
 495         try:
 496             nsr_id = db_nsr["_id"]
 497             nslcmop_id = db_nslcmop["_id"]
 498             nsr_lcm = db_nsr["_admin"]["deployed"]
 499             ns_action = db_nslcmop["lcmOperationType"]
 500             logging_text = "Task ns={} {}={} [n2vc_callback] vnf_index={}".format(nsr_id, ns_action, nslcmop_id,
 501                                                                                   member_vnf_index)
 502
 503             if task:
 504                 if task.cancelled():
 505                     self.logger.debug(logging_text + " task Cancelled")
 506                     # TODO update db_nslcmop
 507                     return
 508
 509                 if task.done():
 510                     exc = task.exception()
 511                     if exc:
 512                         self.logger.error(logging_text + " task Exception={}".format(exc))
 513                         if ns_action in ("instantiate", "terminate"):
 514                             nsr_lcm["VCA"][member_vnf_index]['operational-status'] = "error"
 515                             nsr_lcm["VCA"][member_vnf_index]['detailed-status'] = str(exc)
 516                         elif ns_action == "action":
 517                             db_nslcmop["operationState"] = "FAILED"
 518                             db_nslcmop["detailed-status"] = str(exc)
 519                             db_nslcmop["statusEnteredTime"] = time()
 520                             update_nslcmop = True
 521                             return
 522
 523                     else:
 524                         self.logger.debug(logging_text + " task Done")
 525                         # TODO revise with Adam if action is finished and ok when task is done
 526                         if ns_action == "action":
 527                             db_nslcmop["operationState"] = "COMPLETED"
 528                             db_nslcmop["detailed-status"] = "Done"
 529                             db_nslcmop["statusEnteredTime"] = time()
 530                             update_nslcmop = True
 531                         # task is Done, but callback is still ongoing. So ignore
 532                         return
 533             elif status:
 534                 self.logger.debug(logging_text + " Enter status={}".format(status))
 535                 if nsr_lcm["VCA"][member_vnf_index]['operational-status'] == status:
 536                     return  # same status, ignore
 537                 nsr_lcm["VCA"][member_vnf_index]['operational-status'] = status
 538                 nsr_lcm["VCA"][member_vnf_index]['detailed-status'] = str(message)
 539             else:
 540                 self.logger.critical(logging_text + " Enter with bad parameters", exc_info=True)
 541                 return
 542
 543             all_active = True
 544             status_map = {}
 545             n2vc_error_text = []   # contain text error list. If empty no one is in error status
 546             for vnf_index, vca_info in nsr_lcm["VCA"].items():
 547                 vca_status = vca_info["operational-status"]
 548                 if vca_status not in status_map:
 549                     # Initialize it
 550                     status_map[vca_status] = 0
 551                 status_map[vca_status] += 1
 552
 553                 if vca_status != "active":
 554                     all_active = False
 555                 elif vca_status in ("error", "blocked"):
 556                     n2vc_error_text.append("member_vnf_index={} {}: {}".format(member_vnf_index, vca_status,
 557                                                                            vca_info["detailed-status"]))
 558
 559             if all_active:
 560                 self.logger.debug("[n2vc_callback] ns_instantiate={} vnf_index={} All active".format(nsr_id, member_vnf_index))
 561                 db_nsr["config-status"] = "configured"
 562                 db_nsr["detailed-status"] = "done"
 563                 db_nslcmop["operationState"] = "COMPLETED"
 564                 db_nslcmop["detailed-status"] = "Done"
 565                 db_nslcmop["statusEnteredTime"] = time()
 566             elif n2vc_error_text:
 567                 db_nsr["config-status"] = "failed"
 568                 error_text = "fail configuring " + ";".join(n2vc_error_text)
 569                 db_nsr["detailed-status"] = error_text
 570                 db_nslcmop["operationState"] = "FAILED_TEMP"
 571                 db_nslcmop["detailed-status"] = error_text
 572                 db_nslcmop["statusEnteredTime"] = time()
 573             else:
 574                 cs = "configuring: "
 575                 separator = ""
 576                 for status, num in status_map.items():
 577                     cs += separator + "{}: {}".format(status, num)
 578                     separator = ", "
 579                 db_nsr["config-status"] = cs
 580                 db_nsr["detailed-status"] = cs
 581                 db_nslcmop["detailed-status"] = cs
 582             update_nsr = update_nslcmop = True
 583
 584         except Exception as e:
 585             self.logger.critical("[n2vc_callback] vnf_index={} Exception {}".format(member_vnf_index, e), exc_info=True)
 586         finally:
 587             try:
 588                 if update_nslcmop:
 589                     self.update_db("nslcmops", nslcmop_id, db_nslcmop)
 590                 if update_nsr:
 591                     self.update_db("nsrs", nsr_id, db_nsr)
 592             except Exception as e:
 593                 self.logger.critical("[n2vc_callback] vnf_index={} Update database Exception {}".format(
 594                     member_vnf_index, e), exc_info=True)
 595
 596     def ns_params_2_RO(self, ns_params):
 597         """
 598         Creates a RO ns descriptor from OSM ns_instantite params
 599         :param ns_params: OSM instantiate params
 600         :return: The RO ns descriptor
 601         """
 602         vim_2_RO = {}
 603         def vim_account_2_RO(vim_account):
 604             if vim_account in vim_2_RO:
 605                 return vim_2_RO[vim_account]
 606             db_vim = self.db.get_one("vim_accounts", {"_id": vim_account})
 607             # if db_vim["_admin"]["operationalState"] == "PROCESSING":
 608             #     #TODO check if VIM is creating and wait
 609             if db_vim["_admin"]["operationalState"] != "ENABLED":
 610                 raise LcmException("VIM={} is not available. operationalState={}".format(
 611                     vim_account, db_vim["_admin"]["operationalState"]))
 612             RO_vim_id = db_vim["_admin"]["deployed"]["RO"]
 613             vim_2_RO[vim_account] = RO_vim_id
 614             return RO_vim_id
 615
 616         if not ns_params:
 617             return None
 618         RO_ns_params = {
 619             # "name": ns_params["nsName"],
 620             # "description": ns_params.get("nsDescription"),
 621             "datacenter": vim_account_2_RO(ns_params["vimAccountId"]),
 622             # "scenario": ns_params["nsdId"],
 623             "vnfs": {},
 624             "networks": {},
 625         }
 626         if ns_params.get("ssh-authorized-key"):
 627             RO_ns_params["cloud-config"] = {"key-pairs": ns_params["ssh-authorized-key"]}
 628         if ns_params.get("vnf"):
 629             for vnf in ns_params["vnf"]:
 630                 RO_vnf = {}
 631                 if "vimAccountId" in vnf:
 632                     RO_vnf["datacenter"] = vim_account_2_RO(vnf["vimAccountId"])
 633                 if RO_vnf:
 634                     RO_ns_params["vnfs"][vnf["member-vnf-index"]] = RO_vnf
 635         if ns_params.get("vld"):
 636             for vld in ns_params["vld"]:
 637                 RO_vld = {}
 638                 if "ip-profile" in vld:
 639                     RO_vld["ip-profile"] = vld["ip-profile"]
 640                 if "vim-network-name" in vld:
 641                     RO_vld["sites"] = []
 642                     if isinstance(vld["vim-network-name"], dict):
 643                         for vim_account, vim_net in vld["vim-network-name"].items():
 644                             RO_vld["sites"].append({
 645                                 "netmap-use": vim_net,
 646                                 "datacenter": vim_account_2_RO(vim_account)
 647                             })
 648                     else:  #isinstance str
 649                         RO_vld["sites"].append({"netmap-use":  vld["vim-network-name"]})
 650                 if RO_vld:
 651                     RO_ns_params["networks"][vld["name"]] = RO_vld
 652         return RO_ns_params
 653
 654     async def ns_instantiate(self, nsr_id, nslcmop_id):
 655         logging_text = "Task ns={} instantiate={} ".format(nsr_id, nslcmop_id)
 656         self.logger.debug(logging_text + "Enter")
 657         # get all needed from database
 658         db_nsr = None
 659         db_nslcmop = None
 660         db_vnfr = {}
 661         exc = None
 662         step = "Getting nsr, nslcmop, RO_vims from db"
 663         try:
 664             db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 665             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 666             nsd = db_nsr["nsd"]
 667             nsr_name = db_nsr["name"]   # TODO short-name??
 668             needed_vnfd = {}
 669             vnfr_filter = {"nsr-id-ref": nsr_id, "member-vnf-index-ref": None}
 670             for c_vnf in nsd["constituent-vnfd"]:
 671                 vnfd_id = c_vnf["vnfd-id-ref"]
 672                 vnfr_filter["member-vnf-index-ref"] = c_vnf["member-vnf-index"]
 673                 db_vnfr[c_vnf["member-vnf-index"]] = self.db.get_one("vnfrs", vnfr_filter)
 674                 if vnfd_id not in needed_vnfd:
 675                     step = "Getting vnfd={} from db".format(vnfd_id)
 676                     needed_vnfd[vnfd_id] = self.db.get_one("vnfds", {"id": vnfd_id})
 677
 678             nsr_lcm = db_nsr["_admin"].get("deployed")
 679             if not nsr_lcm:
 680                 nsr_lcm = db_nsr["_admin"]["deployed"] = {
 681                     "id": nsr_id,
 682                     "RO": {"vnfd_id": {}, "nsd_id": None, "nsr_id": None, "nsr_status": "SCHEDULED"},
 683                     "nsr_ip": {},
 684                     "VCA": {},
 685                 }
 686             db_nsr["detailed-status"] = "creating"
 687             db_nsr["operational-status"] = "init"
 688
 689             RO = ROclient.ROClient(self.loop, **self.ro_config)
 690
 691             # get vnfds, instantiate at RO
 692             for vnfd_id, vnfd in needed_vnfd.items():
 693                 step = db_nsr["detailed-status"] = "Creating vnfd={} at RO".format(vnfd_id)
 694                 self.logger.debug(logging_text + step)
 695                 vnfd_id_RO = nsr_id + "." + vnfd_id[:200]
 696
 697                 # look if present
 698                 vnfd_list = await RO.get_list("vnfd", filter_by={"osm_id": vnfd_id_RO})
 699                 if vnfd_list:
 700                     nsr_lcm["RO"]["vnfd_id"][vnfd_id] = vnfd_list[0]["uuid"]
 701                     self.logger.debug(logging_text + "RO vnfd={} exist. Using RO_id={}".format(
 702                         vnfd_id, vnfd_list[0]["uuid"]))
 703                 else:
 704                     vnfd_RO = self.vnfd2RO(vnfd, vnfd_id_RO)
 705                     desc = await RO.create("vnfd", descriptor=vnfd_RO)
 706                     nsr_lcm["RO"]["vnfd_id"][vnfd_id] = desc["uuid"]
 707                     db_nsr["_admin"]["nsState"] = "INSTANTIATED"
 708                 self.update_db("nsrs", nsr_id, db_nsr)
 709
 710             # create nsd at RO
 711             nsd_id = nsd["id"]
 712             step = db_nsr["detailed-status"] = "Creating nsd={} at RO".format(nsd_id)
 713             self.logger.debug(logging_text + step)
 714
 715             nsd_id_RO = nsd_id + "." + nsd_id[:200]
 716             nsd_list = await RO.get_list("nsd", filter_by={"osm_id": nsd_id_RO})
 717             if nsd_list:
 718                 nsr_lcm["RO"]["nsd_id"] = nsd_list[0]["uuid"]
 719                 self.logger.debug(logging_text + "RO nsd={} exist. Using RO_id={}".format(
 720                     nsd_id, nsd_list[0]["uuid"]))
 721             else:
 722                 nsd_RO = deepcopy(nsd)
 723                 nsd_RO["id"] = nsd_id_RO
 724                 nsd_RO.pop("_id", None)
 725                 nsd_RO.pop("_admin", None)
 726                 for c_vnf in nsd_RO["constituent-vnfd"]:
 727                     vnfd_id = c_vnf["vnfd-id-ref"]
 728                     c_vnf["vnfd-id-ref"] = nsr_id + "." + vnfd_id[:200]
 729                 desc = await RO.create("nsd", descriptor=nsd_RO)
 730                 db_nsr["_admin"]["nsState"] = "INSTANTIATED"
 731                 nsr_lcm["RO"]["nsd_id"] = desc["uuid"]
 732             self.update_db("nsrs", nsr_id, db_nsr)
 733
 734             # Crate ns at RO
 735             # if present use it unless in error status
 736             RO_nsr_id = nsr_lcm["RO"].get("nsr_id")
 737             if RO_nsr_id:
 738                 try:
 739                     step = db_nsr["detailed-status"] = "Looking for existing ns at RO"
 740                     self.logger.debug(logging_text + step + " RO_ns_id={}".format(RO_nsr_id))
 741                     desc = await RO.show("ns", RO_nsr_id)
 742                 except ROclient.ROClientException as e:
 743                     if e.http_code != HTTPStatus.NOT_FOUND:
 744                         raise
 745                     RO_nsr_id = nsr_lcm["RO"]["nsr_id"] = None
 746                 if RO_nsr_id:
 747                     ns_status, ns_status_info = RO.check_ns_status(desc)
 748                     nsr_lcm["RO"]["nsr_status"] = ns_status
 749                     if ns_status == "ERROR":
 750                         step = db_nsr["detailed-status"] = "Deleting ns at RO"
 751                         self.logger.debug(logging_text + step + " RO_ns_id={}".format(RO_nsr_id))
 752                         await RO.delete("ns", RO_nsr_id)
 753                         RO_nsr_id = nsr_lcm["RO"]["nsr_id"] = None
 754             if not RO_nsr_id:
 755                 step = db_nsr["detailed-status"] = "Creating ns at RO"
 756                 self.logger.debug(logging_text + step)
 757                 RO_ns_params = self.ns_params_2_RO(db_nsr.get("instantiate_params"))
 758                 desc = await RO.create("ns", descriptor=RO_ns_params,
 759                                        name=db_nsr["name"],
 760                                        scenario=nsr_lcm["RO"]["nsd_id"])
 761                 RO_nsr_id = nsr_lcm["RO"]["nsr_id"] = desc["uuid"]
 762                 db_nsr["_admin"]["nsState"] = "INSTANTIATED"
 763                 nsr_lcm["RO"]["nsr_status"] = "BUILD"
 764
 765             self.update_db("nsrs", nsr_id, db_nsr)
 766             # update VNFR vimAccount
 767             step = "Updating VNFR vimAcccount"
 768             for vnf_index, vnfr in db_vnfr.items():
 769                 if vnfr.get("vim-account-id"):
 770                     continue
 771                 if db_nsr["instantiate_params"].get("vnf") and db_nsr["instantiate_params"]["vnf"].get(vnf_index) \
 772                         and db_nsr["instantiate_params"]["vnf"][vnf_index].get("vimAccountId"):
 773                     vnfr["vim-account-id"] = db_nsr["instantiate_params"]["vnf"][vnf_index]["vimAccountId"]
 774                 else:
 775                     vnfr["vim-account-id"] = db_nsr["instantiate_params"]["vimAccountId"]
 776                 self.update_db("vnfrs", vnfr["_id"], vnfr)
 777
 778             # wait until NS is ready
 779             step = ns_status_detailed = "Waiting ns ready at RO"
 780             db_nsr["detailed-status"] = ns_status_detailed
 781             self.logger.debug(logging_text + step + " RO_ns_id={}".format(RO_nsr_id))
 782             deployment_timeout = 2*3600   # Two hours
 783             while deployment_timeout > 0:
 784                 desc = await RO.show("ns", RO_nsr_id)
 785                 ns_status, ns_status_info = RO.check_ns_status(desc)
 786                 nsr_lcm["RO"]["nsr_status"] = ns_status
 787                 if ns_status == "ERROR":
 788                     raise ROclient.ROClientException(ns_status_info)
 789                 elif ns_status == "BUILD":
 790                     db_nsr_detailed_status_old = db_nsr["detailed-status"]
 791                     db_nsr["detailed-status"] = ns_status_detailed + "; {}".format(ns_status_info)
 792                     if db_nsr_detailed_status_old != db_nsr["detailed-status"]:
 793                         self.update_db("nsrs", nsr_id, db_nsr)
 794                 elif ns_status == "ACTIVE":
 795                     step = "Waiting for management IP address from VIM"
 796                     try:
 797                         ns_RO_info = nsr_lcm["nsr_ip"] = RO.get_ns_vnf_info(desc)
 798                         break
 799                     except ROclient.ROClientException as e:
 800                         if e.http_code != 409:  # IP address is not ready return code is 409 CONFLICT
 801                             raise e
 802                 else:
 803                     assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status)
 804                 await asyncio.sleep(5, loop=self.loop)
 805                 deployment_timeout -= 5
 806             if deployment_timeout <= 0:
 807                 raise ROclient.ROClientException("Timeout waiting ns to be ready")
 808             step = "Updating VNFRs"
 809             for vnf_index, vnfr_deployed in ns_RO_info.items():
 810                 vnfr = db_vnfr[vnf_index]
 811                 vnfr["ip-address"] = vnfr_deployed.get("ip_address")
 812                 for vdu_id, vdu_deployed in vnfr_deployed["vdur"].items():
 813                     for vdur in vnfr["vdur"]:
 814                         if vdur["vdu-id-ref"] == vdu_id:
 815                             vdur["vim-id"] = vdu_deployed.get("vim_id")
 816                             vdur["ip-address"] = vdu_deployed.get("ip_address")
 817                             break
 818                 self.update_db("vnfrs", vnfr["_id"], vnfr)
 819
 820             db_nsr["detailed-status"] = "Configuring vnfr"
 821             self.update_db("nsrs", nsr_id, db_nsr)
 822
 823             # The parameters we'll need to deploy a charm
 824             number_to_configure = 0
 825
 826             def deploy():
 827                 """An inner function to deploy the charm from either vnf or vdu
 828                 """
 829
 830                 # Login to the VCA.
 831                 # if number_to_configure == 0:
 832                 #     self.logger.debug("Logging into N2VC...")
 833                 #     task = asyncio.ensure_future(self.n2vc.login())
 834                 #     yield from asyncio.wait_for(task, 30.0)
 835                 #     self.logger.debug("Logged into N2VC!")
 836
 837                 ## await self.n2vc.login()
 838
 839                 # Note: The charm needs to exist on disk at the location
 840                 # specified by charm_path.
 841                 base_folder = vnfd["_admin"]["storage"]
 842                 storage_params = self.fs.get_params()
 843                 charm_path = "{}{}/{}/charms/{}".format(
 844                     storage_params["path"],
 845                     base_folder["folder"],
 846                     base_folder["pkg-dir"],
 847                     proxy_charm
 848                 )
 849
 850                 # Setup the runtime parameters for this VNF
 851                 params['rw_mgmt_ip'] = db_vnfr[vnf_index]["ip-address"]
 852
 853                 # ns_name will be ignored in the current version of N2VC
 854                 # but will be implemented for the next point release.
 855                 model_name = 'default'
 856                 application_name = self.n2vc.FormatApplicationName(
 857                     nsr_name,
 858                     vnf_index,
 859                     vnfd['name'],
 860                 )
 861
 862                 nsr_lcm["VCA"][vnf_index] = {
 863                     "model": model_name,
 864                     "application": application_name,
 865                     "operational-status": "init",
 866                     "detailed-status": "",
 867                     "vnfd_id": vnfd_id,
 868                 }
 869
 870                 self.logger.debug("Task create_ns={} Passing artifacts path '{}' for {}".format(nsr_id, charm_path, proxy_charm))
 871                 task = asyncio.ensure_future(
 872                     self.n2vc.DeployCharms(
 873                         model_name,          # The network service name
 874                         application_name,    # The application name
 875                         vnfd,                # The vnf descriptor
 876                         charm_path,          # Path to charm
 877                         params,              # Runtime params, like mgmt ip
 878                         {},                  # for native charms only
 879                         self.n2vc_callback,  # Callback for status changes
 880                         db_nsr,              # Callback parameter
 881                         db_nslcmop,
 882                         vnf_index,           # Callback parameter
 883                         None,                # Callback parameter (task)
 884                     )
 885                 )
 886                 task.add_done_callback(functools.partial(self.n2vc_callback, model_name, application_name, None, None,
 887                                                          db_nsr, db_nslcmop, vnf_index))
 888                 self.lcm_ns_tasks[nsr_id][nslcmop_id]["create_charm:" + vnf_index] = task
 889
 890             # TODO: Make this call inside deploy()
 891             # Login to the VCA. If there are multiple calls to login(),
 892             # subsequent calls will be a nop and return immediately.
 893             await self.n2vc.login()
 894
 895             step = "Looking for needed vnfd to configure"
 896             self.logger.debug(logging_text + step)
 897             for c_vnf in nsd["constituent-vnfd"]:
 898                 vnfd_id = c_vnf["vnfd-id-ref"]
 899                 vnf_index = str(c_vnf["member-vnf-index"])
 900                 vnfd = needed_vnfd[vnfd_id]
 901
 902                 # Check if this VNF has a charm configuration
 903                 vnf_config = vnfd.get("vnf-configuration")
 904
 905                 if vnf_config and vnf_config.get("juju"):
 906                     proxy_charm = vnf_config["juju"]["charm"]
 907                     params = {}
 908
 909                     if proxy_charm:
 910                         if 'initial-config-primitive' in vnf_config:
 911                             params['initial-config-primitive'] = vnf_config['initial-config-primitive']
 912
 913                         deploy()
 914                         number_to_configure += 1
 915
 916                 # Deploy charms for each VDU that supports one.
 917                 for vdu in vnfd['vdu']:
 918                     vdu_config = vdu.get('vdu-configuration')
 919                     proxy_charm = None
 920                     params = {}
 921
 922                     if vdu_config and vdu_config.get("juju"):
 923                         proxy_charm = vdu_config["juju"]["charm"]
 924
 925                         if 'initial-config-primitive' in vdu_config:
 926                             params['initial-config-primitive'] = vdu_config['initial-config-primitive']
 927
 928                         if proxy_charm:
 929                             deploy()
 930                             number_to_configure += 1
 931
 932             if number_to_configure:
 933                 db_nsr["config-status"] = "configuring"
 934                 db_nsr["detailed-status"] = "configuring: init: {}".format(number_to_configure)
 935                 db_nslcmop["detailed-status"] = "configuring: init: {}".format(number_to_configure)
 936             else:
 937                 db_nslcmop["operationState"] = "COMPLETED"
 938                 db_nslcmop["detailed-status"] = "done"
 939                 db_nsr["config-status"] = "configured"
 940                 db_nsr["detailed-status"] = "done"
 941             db_nsr["operational-status"] = "running"
 942             self.update_db("nsrs", nsr_id, db_nsr)
 943             self.update_db("nslcmops", nslcmop_id, db_nslcmop)
 944             self.logger.debug("Task ns_instantiate={} Exit Ok".format(nsr_id))
 945             return nsr_lcm
 946
 947         except (ROclient.ROClientException, DbException, LcmException) as e:
 948             self.logger.error(logging_text + "Exit Exception {}".format(e))
 949             exc = e
 950         except Exception as e:
 951             self.logger.critical(logging_text + "Exit Exception {} {}".format(type(e).__name__, e), exc_info=True)
 952             exc = e
 953         finally:
 954             if exc:
 955                 if db_nsr:
 956                     db_nsr["detailed-status"] = "ERROR {}: {}".format(step, exc)
 957                     db_nsr["operational-status"] = "failed"
 958                     self.update_db("nsrs", nsr_id, db_nsr)
 959                 if db_nslcmop:
 960                     db_nslcmop["detailed-status"] = "FAILED {}: {}".format(step, exc)
 961                     db_nslcmop["operationState"] = "FAILED"
 962                     db_nslcmop["statusEnteredTime"] = time()
 963                     self.update_db("nslcmops", nslcmop_id, db_nslcmop)
 964
 965     async def ns_terminate(self, nsr_id, nslcmop_id):
 966         logging_text = "Task ns={} terminate={} ".format(nsr_id, nslcmop_id)
 967         self.logger.debug(logging_text + "Enter")
 968         db_nsr = None
 969         db_nslcmop = None
 970         exc = None
 971         step = "Getting nsr, nslcmop from db"
 972         failed_detail = []   # annotates all failed error messages
 973         vca_task_list = []
 974         vca_task_dict = {}
 975         try:
 976             db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
 977             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
 978             # nsd = db_nsr["nsd"]
 979             nsr_lcm = deepcopy(db_nsr["_admin"]["deployed"])
 980             if db_nsr["_admin"]["nsState"] == "NOT_INSTANTIATED":
 981                 return
 982             # TODO ALF remove
 983             # db_vim = self.db.get_one("vim_accounts", {"_id":  db_nsr["datacenter"]})
 984             # #TODO check if VIM is creating and wait
 985             # RO_vim_id = db_vim["_admin"]["deployed"]["RO"]
 986
 987             db_nsr_update = {
 988                 "operational-status": "terminating",
 989                 "config-status": "terminating",
 990                 "detailed-status": "Deleting charms",
 991             }
 992             self.update_db_2("nsrs", nsr_id, db_nsr_update)
 993
 994             try:
 995                 self.logger.debug(logging_text + step)
 996                 for vnf_index, deploy_info in nsr_lcm["VCA"].items():
 997                     if deploy_info and deploy_info.get("application"):
 998                         task = asyncio.ensure_future(
 999                             self.n2vc.RemoveCharms(
1000                                 deploy_info['model'],
1001                                 deploy_info['application'],
1002                                 # self.n2vc_callback,
1003                                 # db_nsr,
1004                                 # db_nslcmop,
1005                                 # vnf_index,
1006                             )
1007                         )
1008                         vca_task_list.append(task)
1009                         vca_task_dict[vnf_index] = task
1010                         # task.add_done_callback(functools.partial(self.n2vc_callback, deploy_info['model'],
1011                         #                                          deploy_info['application'], None, db_nsr,
1012                         #                                          db_nslcmop, vnf_index))
1013                         self.lcm_ns_tasks[nsr_id][nslcmop_id]["delete_charm:" + vnf_index] = task
1014             except Exception as e:
1015                 self.logger.debug(logging_text + "Failed while deleting charms: {}".format(e))
1016             # remove from RO
1017
1018             RO = ROclient.ROClient(self.loop, **self.ro_config)
1019             # Delete ns
1020             RO_nsr_id = nsr_lcm["RO"].get("nsr_id")
1021             if RO_nsr_id:
1022                 try:
1023                     step = db_nsr["detailed-status"] = "Deleting ns at RO"
1024                     self.logger.debug(logging_text + step)
1025                     desc = await RO.delete("ns", RO_nsr_id)
1026                     nsr_lcm["RO"]["nsr_id"] = None
1027                     nsr_lcm["RO"]["nsr_status"] = "DELETED"
1028                 except ROclient.ROClientException as e:
1029                     if e.http_code == 404:  # not found
1030                         nsr_lcm["RO"]["nsr_id"] = None
1031                         nsr_lcm["RO"]["nsr_status"] = "DELETED"
1032                         self.logger.debug(logging_text + "RO_ns_id={} already deleted".format(RO_nsr_id))
1033                     elif e.http_code == 409:   #conflict
1034                         failed_detail.append("RO_ns_id={} delete conflict: {}".format(RO_nsr_id, e))
1035                         self.logger.debug(logging_text + failed_detail[-1])
1036                     else:
1037                         failed_detail.append("RO_ns_id={} delete error: {}".format(RO_nsr_id, e))
1038                         self.logger.error(logging_text + failed_detail[-1])
1039
1040             # Delete nsd
1041             RO_nsd_id = nsr_lcm["RO"]["nsd_id"]
1042             if RO_nsd_id:
1043                 try:
1044                     step = db_nsr["detailed-status"] = "Deleting nsd at RO"
1045                     desc = await RO.delete("nsd", RO_nsd_id)
1046                     self.logger.debug(logging_text + "RO_nsd_id={} deleted".format(RO_nsd_id))
1047                     nsr_lcm["RO"]["nsd_id"] = None
1048                 except ROclient.ROClientException as e:
1049                     if e.http_code == 404:  # not found
1050                         nsr_lcm["RO"]["nsd_id"] = None
1051                         self.logger.debug(logging_text + "RO_nsd_id={} already deleted".format(RO_nsd_id))
1052                     elif e.http_code == 409:   #conflict
1053                         failed_detail.append("RO_nsd_id={} delete conflict: {}".format(RO_nsd_id, e))
1054                         self.logger.debug(logging_text + failed_detail[-1])
1055                     else:
1056                         failed_detail.append("RO_nsd_id={} delete error: {}".format(RO_nsd_id, e))
1057                         self.logger.error(logging_text + failed_detail[-1])
1058
1059             for vnf_id, RO_vnfd_id in nsr_lcm["RO"]["vnfd_id"].items():
1060                 if not RO_vnfd_id:
1061                     continue
1062                 try:
1063                     step = db_nsr["detailed-status"] = "Deleting vnfd={} at RO".format(vnf_id)
1064                     desc = await RO.delete("vnfd", RO_vnfd_id)
1065                     self.logger.debug(logging_text + "RO_vnfd_id={} deleted".format(RO_vnfd_id))
1066                     nsr_lcm["RO"]["vnfd_id"][vnf_id] = None
1067                 except ROclient.ROClientException as e:
1068                     if e.http_code == 404:  # not found
1069                         nsr_lcm["RO"]["vnfd_id"][vnf_id] = None
1070                         self.logger.debug(logging_text + "RO_vnfd_id={} already deleted ".format(RO_vnfd_id))
1071                     elif e.http_code == 409:   #conflict
1072                         failed_detail.append("RO_vnfd_id={} delete conflict: {}".format(RO_vnfd_id, e))
1073                         self.logger.debug(logging_text + failed_detail[-1])
1074                     else:
1075                         failed_detail.append("RO_vnfd_id={} delete error: {}".format(RO_vnfd_id, e))
1076                         self.logger.error(logging_text + failed_detail[-1])
1077
1078             if vca_task_list:
1079                 await asyncio.wait(vca_task_list, timeout=300)
1080             for vnf_index, task in vca_task_dict.items():
1081                 if task.cancelled():
1082                     failed_detail.append("VCA[{}] Deletion has been cancelled".format(vnf_index))
1083                 elif task.done():
1084                     exc = task.exception()
1085                     if exc:
1086                         failed_detail.append("VCA[{}] Deletion exception: {}".format(vnf_index, exc))
1087                     else:
1088                         nsr_lcm["VCA"][vnf_index] = None
1089                 else:  # timeout
1090                     # TODO Should it be cancelled?!!
1091                     task.cancel()
1092                     failed_detail.append("VCA[{}] Deletion timeout".format(vnf_index))
1093
1094             if failed_detail:
1095                 self.logger.error(logging_text + " ;".join(failed_detail))
1096                 db_nsr_update = {
1097                     "operational-status": "failed",
1098                     "detailed-status": "Deletion errors " + "; ".join(failed_detail),
1099                     "_admin": {"deployed": nsr_lcm, }
1100                 }
1101                 db_nslcmop_update = {
1102                     "detailed-status": "; ".join(failed_detail),
1103                     "operationState": "FAILED",
1104                     "statusEnteredTime": time()
1105                 }
1106             elif db_nslcmop["operationParams"].get("autoremove"):
1107                 self.db.del_one("nsrs", {"_id": nsr_id})
1108                 self.db.del_list("nslcmops", {"nsInstanceId": nsr_id})
1109                 self.db.del_list("vnfrs", {"nsr-id-ref": nsr_id})
1110             else:
1111                 db_nsr_update = {
1112                     "operational-status": "terminated",
1113                     "detailed-status": "Done",
1114                     "_admin": {"deployed": nsr_lcm, "nsState": "NOT_INSTANTIATED"}
1115                 }
1116                 db_nslcmop_update = {
1117                     "detailed-status": "Done",
1118                     "operationState": "COMPLETED",
1119                     "statusEnteredTime": time()
1120                 }
1121             self.logger.debug(logging_text + "Exit")
1122
1123         except (ROclient.ROClientException, DbException) as e:
1124             self.logger.error(logging_text + "Exit Exception {}".format(e))
1125             exc = e
1126         except Exception as e:
1127             self.logger.critical(logging_text + "Exit Exception {}".format(e), exc_info=True)
1128             exc = e
1129         finally:
1130             if exc and db_nslcmop:
1131                 db_nslcmop_update = {
1132                     "detailed-status": "FAILED {}: {}".format(step, exc),
1133                     "operationState": "FAILED",
1134                     "statusEnteredTime": time(),
1135                 }
1136             if db_nslcmop_update:
1137                 self.update_db_2("nslcmops", nslcmop_id, db_nslcmop_update)
1138             if db_nsr_update:
1139                 self.update_db_2("nsrs", nsr_id, db_nsr_update)
1140
1141     async def ns_action(self, nsr_id, nslcmop_id):
1142         logging_text = "Task ns={} action={} ".format(nsr_id, nslcmop_id)
1143         self.logger.debug(logging_text + "Enter")
1144         # get all needed from database
1145         db_nsr = None
1146         db_nslcmop = None
1147         db_nslcmop_update = None
1148         exc = None
1149         try:
1150             step = "Getting information from database"
1151             db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
1152             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
1153             nsr_lcm = db_nsr["_admin"].get("deployed")
1154             vnf_index = db_nslcmop["operationParams"]["member_vnf_index"]
1155
1156             #TODO check if ns is in a proper status
1157             vca_deployed = nsr_lcm["VCA"].get(vnf_index)
1158             if not vca_deployed:
1159                 raise LcmException("charm for member_vnf_index={} is not deployed".format(vnf_index))
1160             model_name = vca_deployed.get("model")
1161             application_name = vca_deployed.get("application")
1162             if not model_name or not application_name:
1163                 raise LcmException("charm for member_vnf_index={} is not properly deployed".format(vnf_index))
1164             if vca_deployed["operational-status"] != "active":
1165                 raise LcmException("charm for member_vnf_index={} operational_status={} not 'active'".format(
1166                     vnf_index, vca_deployed["operational-status"]))
1167             primitive = db_nslcmop["operationParams"]["primitive"]
1168             primitive_params = db_nslcmop["operationParams"]["primitive_params"]
1169             callback = None     # self.n2vc_callback
1170             callback_args = ()  # [db_nsr, db_nslcmop, vnf_index, None]
1171             await self.n2vc.login()
1172             task = asyncio.ensure_future(
1173                 self.n2vc.ExecutePrimitive(
1174                     model_name,
1175                     application_name,
1176                     primitive, callback,
1177                     *callback_args,
1178                     **primitive_params
1179                 )
1180             )
1181             # task.add_done_callback(functools.partial(self.n2vc_callback, model_name, application_name, None,
1182             #                                          db_nsr, db_nslcmop, vnf_index))
1183             # self.lcm_ns_tasks[nsr_id][nslcmop_id]["action: " + primitive] = task
1184             # wait until completed with timeout
1185             await asyncio.wait((task,), timeout=300)
1186
1187             result = "FAILED"  # by default
1188             result_detail = ""
1189             if task.cancelled():
1190                 db_nslcmop["detailed-status"] = "Task has been cancelled"
1191             elif task.done():
1192                 exc = task.exception()
1193                 if exc:
1194                     result_detail = str(exc)
1195                 else:
1196                     self.logger.debug(logging_text + " task Done")
1197                     # TODO revise with Adam if action is finished and ok when task is done or callback is needed
1198                     result = "COMPLETED"
1199                     result_detail = "Done"
1200             else:   # timeout
1201                 # TODO Should it be cancelled?!!
1202                 task.cancel()
1203                 result_detail = "timeout"
1204
1205             db_nslcmop_update = {
1206                 "detailed-status": result_detail,
1207                 "operationState": result,
1208                 "statusEnteredTime": time()
1209             }
1210             self.logger.debug(logging_text + " task Done with result {} {}".format(result, result_detail))
1211             return  # database update is called inside finally
1212
1213         except (DbException, LcmException) as e:
1214             self.logger.error(logging_text + "Exit Exception {}".format(e))
1215             exc = e
1216         except Exception as e:
1217             self.logger.critical(logging_text + "Exit Exception {} {}".format(type(e).__name__, e), exc_info=True)
1218             exc = e
1219         finally:
1220             if exc and db_nslcmop:
1221                 db_nslcmop_update = {
1222                     "detailed-status": "FAILED {}: {}".format(step, exc),
1223                     "operationState": "FAILED",
1224                     "statusEnteredTime": time(),
1225                 }
1226             if db_nslcmop_update:
1227                 self.update_db_2("nslcmops", nslcmop_id, db_nslcmop_update)
1228
1229     async def test(self, param=None):
1230         self.logger.debug("Starting/Ending test task: {}".format(param))
1231
1232     def cancel_tasks(self, topic, _id):
1233         """
1234         Cancel all active tasks of a concrete nsr or vim identified for _id
1235         :param topic: can be ns or vim_account
1236         :param _id:  nsr or vim identity
1237         :return: None, or raises an exception if not possible
1238         """
1239         if topic == "ns":
1240             lcm_tasks = self.lcm_ns_tasks
1241         elif topic== "vim_account":
1242             lcm_tasks = self.lcm_vim_tasks
1243         elif topic== "sdn":
1244             lcm_tasks = self.lcm_sdn_tasks
1245
1246         if not lcm_tasks.get(_id):
1247             return
1248         for order_id, tasks_set in lcm_tasks[_id].items():
1249             for task_name, task in tasks_set.items():
1250                 result = task.cancel()
1251                 if result:
1252                     self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, order_id, task_name))
1253         lcm_tasks[_id] = {}
1254
1255     async def kafka_ping(self):
1256         self.logger.debug("Task kafka_ping Enter")
1257         consecutive_errors = 0
1258         first_start = True
1259         kafka_has_received = False
1260         self.pings_not_received = 1
1261         while True:
1262             try:
1263                 await self.msg.aiowrite("admin", "ping", {"from": "lcm", "to": "lcm"},  self.loop)
1264                 # time between pings are low when it is not received and at starting
1265                 wait_time = 5 if not kafka_has_received else 120
1266                 if not self.pings_not_received:
1267                     kafka_has_received = True
1268                 self.pings_not_received += 1
1269                 await asyncio.sleep(wait_time, loop=self.loop)
1270                 if self.pings_not_received > 10:
1271                     raise LcmException("It is not receiving pings from Kafka bus")
1272                 consecutive_errors = 0
1273                 first_start = False
1274             except LcmException:
1275                 raise
1276             except Exception as e:
1277                 # if not first_start is the first time after starting. So leave more time and wait
1278                 # to allow kafka starts
1279                 if consecutive_errors == 8 if not first_start else 30:
1280                     self.logger.error("Task kafka_read task exit error too many errors. Exception: {}".format(e))
1281                     raise
1282                 consecutive_errors += 1
1283                 self.logger.error("Task kafka_read retrying after Exception {}".format(e))
1284                 wait_time = 1 if not first_start else 5
1285                 await asyncio.sleep(wait_time, loop=self.loop)
1286
1287     async def kafka_read(self):
1288         self.logger.debug("Task kafka_read Enter")
1289         order_id = 1
1290         # future = asyncio.Future()
1291         consecutive_errors = 0
1292         first_start = True
1293         while consecutive_errors < 10:
1294             try:
1295                 topics = ("admin", "ns", "vim_account", "sdn")
1296                 topic, command, params = await self.msg.aioread(topics, self.loop)
1297                 self.logger.debug("Task kafka_read receives {} {}: {}".format(topic, command, params))
1298                 consecutive_errors = 0
1299                 first_start = False
1300                 order_id += 1
1301                 if command == "exit":
1302                     print("Bye!")
1303                     break
1304                 elif command.startswith("#"):
1305                     continue
1306                 elif command == "echo":
1307                     # just for test
1308                     print(params)
1309                     sys.stdout.flush()
1310                     continue
1311                 elif command == "test":
1312                     asyncio.Task(self.test(params), loop=self.loop)
1313                     continue
1314
1315                 if topic == "admin":
1316                     if command == "ping" and params["to"] == "lcm" and params["from"] == "lcm":
1317                         self.pings_not_received = 0
1318                     continue
1319                 elif topic == "ns":
1320                     if command == "instantiate":
1321                         # self.logger.debug("Deploying NS {}".format(nsr_id))
1322                         nslcmop = params
1323                         nslcmop_id = nslcmop["_id"]
1324                         nsr_id = nslcmop["nsInstanceId"]
1325                         task = asyncio.ensure_future(self.ns_instantiate(nsr_id, nslcmop_id))
1326                         if nsr_id not in self.lcm_ns_tasks:
1327                             self.lcm_ns_tasks[nsr_id] = {}
1328                         self.lcm_ns_tasks[nsr_id][nslcmop_id] = {"ns_instantiate": task}
1329                         continue
1330                     elif command == "terminate":
1331                         # self.logger.debug("Deleting NS {}".format(nsr_id))
1332                         nslcmop = params
1333                         nslcmop_id = nslcmop["_id"]
1334                         nsr_id = nslcmop["nsInstanceId"]
1335                         self.cancel_tasks(topic, nsr_id)
1336                         task = asyncio.ensure_future(self.ns_terminate(nsr_id, nslcmop_id))
1337                         if nsr_id not in self.lcm_ns_tasks:
1338                             self.lcm_ns_tasks[nsr_id] = {}
1339                         self.lcm_ns_tasks[nsr_id][nslcmop_id] = {"ns_terminate": task}
1340                         continue
1341                     elif command == "action":
1342                         # self.logger.debug("Update NS {}".format(nsr_id))
1343                         nslcmop = params
1344                         nslcmop_id = nslcmop["_id"]
1345                         nsr_id = nslcmop["nsInstanceId"]
1346                         task = asyncio.ensure_future(self.ns_action(nsr_id, nslcmop_id))
1347                         if nsr_id not in self.lcm_ns_tasks:
1348                             self.lcm_ns_tasks[nsr_id] = {}
1349                         self.lcm_ns_tasks[nsr_id][nslcmop_id] = {"ns_action": task}
1350                         continue
1351                     elif command == "show":
1352                         try:
1353                             db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
1354                             print(
1355                             "nsr:\n    _id={}\n    operational-status: {}\n    config-status: {}\n    detailed-status: "
1356                             "{}\n    deploy: {}\n    tasks: {}".format(
1357                                 nsr_id, db_nsr["operational-status"],
1358                                 db_nsr["config-status"], db_nsr["detailed-status"],
1359                                 db_nsr["_admin"]["deployed"], self.lcm_ns_tasks.get(nsr_id)))
1360                         except Exception as e:
1361                             print("nsr {} not found: {}".format(nsr_id, e))
1362                         sys.stdout.flush()
1363                         continue
1364                     elif command == "deleted":
1365                         continue  # TODO cleaning of task just in case should be done
1366                 elif topic == "vim_account":
1367                     vim_id = params["_id"]
1368                     if command == "create":
1369                         task = asyncio.ensure_future(self.vim_create(params, order_id))
1370                         if vim_id not in self.lcm_vim_tasks:
1371                             self.lcm_vim_tasks[vim_id] = {}
1372                         self.lcm_vim_tasks[vim_id][order_id] = {"vim_create": task}
1373                         continue
1374                     elif command == "delete":
1375                         self.cancel_tasks(topic, vim_id)
1376                         task = asyncio.ensure_future(self.vim_delete(vim_id, order_id))
1377                         if vim_id not in self.lcm_vim_tasks:
1378                             self.lcm_vim_tasks[vim_id] = {}
1379                         self.lcm_vim_tasks[vim_id][order_id] = {"vim_delete": task}
1380                         continue
1381                     elif command == "show":
1382                         print("not implemented show with vim_account")
1383                         sys.stdout.flush()
1384                         continue
1385                     elif command == "edit":
1386                         task = asyncio.ensure_future(self.vim_edit(vim_id, order_id))
1387                         if vim_id not in self.lcm_vim_tasks:
1388                             self.lcm_vim_tasks[vim_id] = {}
1389                         self.lcm_vim_tasks[vim_id][order_id] = {"vim_edit": task}
1390                         continue
1391                 elif topic == "sdn":
1392                     _sdn_id = params["_id"]
1393                     if command == "create":
1394                         task = asyncio.ensure_future(self.sdn_create(params, order_id))
1395                         if _sdn_id not in self.lcm_sdn_tasks:
1396                             self.lcm_sdn_tasks[_sdn_id] = {}
1397                         self.lcm_sdn_tasks[_sdn_id][order_id] = {"sdn_create": task}
1398                         continue
1399                     elif command == "delete":
1400                         self.cancel_tasks(topic, _sdn_id)
1401                         task = asyncio.ensure_future(self.sdn_delete(_sdn_id, order_id))
1402                         if _sdn_id not in self.lcm_sdn_tasks:
1403                             self.lcm_sdn_tasks[_sdn_id] = {}
1404                         self.lcm_sdn_tasks[_sdn_id][order_id] = {"sdn_delete": task}
1405                         continue
1406                     elif command == "edit":
1407                         task = asyncio.ensure_future(self.sdn_edit(_sdn_id, order_id))
1408                         if _sdn_id not in self.lcm_sdn_tasks:
1409                             self.lcm_sdn_tasks[_sdn_id] = {}
1410                         self.lcm_sdn_tasks[_sdn_id][order_id] = {"sdn_edit": task}
1411                         continue
1412                 self.logger.critical("unknown topic {} and command '{}'".format(topic, command))
1413             except Exception as e:
1414                 # if not first_start is the first time after starting. So leave more time and wait
1415                 # to allow kafka starts
1416                 if consecutive_errors == 8 if not first_start else 30:
1417                     self.logger.error("Task kafka_read task exit error too many errors. Exception: {}".format(e))
1418                     raise
1419                 consecutive_errors += 1
1420                 self.logger.error("Task kafka_read retrying after Exception {}".format(e))
1421                 wait_time = 2 if not first_start else 5
1422                 await asyncio.sleep(wait_time, loop=self.loop)
1423
1424         # self.logger.debug("Task kafka_read terminating")
1425         self.logger.debug("Task kafka_read exit")
1426
1427     def start(self):
1428         self.loop = asyncio.get_event_loop()
1429         self.loop.run_until_complete(asyncio.gather(
1430             self.kafka_read(),
1431             self.kafka_ping()
1432         ))
1433         # TODO
1434         # self.logger.debug("Terminating cancelling creation tasks")
1435         # self.cancel_tasks("ALL", "create")
1436         # timeout = 200
1437         # while self.is_pending_tasks():
1438         #     self.logger.debug("Task kafka_read terminating. Waiting for tasks termination")
1439         #     await asyncio.sleep(2, loop=self.loop)
1440         #     timeout -= 2
1441         #     if not timeout:
1442         #         self.cancel_tasks("ALL", "ALL")
1443         self.loop.close()
1444         self.loop = None
1445         if self.db:
1446             self.db.db_disconnect()
1447         if self.msg:
1448             self.msg.disconnect()
1449         if self.fs:
1450             self.fs.fs_disconnect()
1451
1452     def read_config_file(self, config_file):
1453         # TODO make a [ini] + yaml inside parser
1454         # the configparser library is not suitable, because it does not admit comments at the end of line,
1455         # and not parse integer or boolean
1456         try:
1457             with open(config_file) as f:
1458                 conf = yaml.load(f)
1459             for k, v in environ.items():
1460                 if not k.startswith("OSMLCM_"):
1461                     continue
1462                 k_items = k.lower().split("_")
1463                 c = conf
1464                 try:
1465                     for k_item in k_items[1:-1]:
1466                         if k_item in ("ro", "vca"):
1467                             # put in capital letter
1468                             k_item = k_item.upper()
1469                         c = c[k_item]
1470                     if k_items[-1] == "port":
1471                         c[k_items[-1]] = int(v)
1472                     else:
1473                         c[k_items[-1]] = v
1474                 except Exception as e:
1475                     self.logger.warn("skipping environ '{}' on exception '{}'".format(k, e))
1476
1477             return conf
1478         except Exception as e:
1479             self.logger.critical("At config file '{}': {}".format(config_file, e))
1480             exit(1)
1481
1482
1483 def usage():
1484     print("""Usage: {} [options]
1485         -c|--config [configuration_file]: loads the configuration file (default: ./nbi.cfg)
1486         -h|--help: shows this help
1487         """.format(sys.argv[0]))
1488         # --log-socket-host HOST: send logs to this host")
1489         # --log-socket-port PORT: send logs using this port (default: 9022)")
1490
1491
1492 if __name__ == '__main__':
1493     try:
1494         # load parameters and configuration
1495         opts, args = getopt.getopt(sys.argv[1:], "hc:", ["config=", "help"])
1496         # TODO add  "log-socket-host=", "log-socket-port=", "log-file="
1497         config_file = None
1498         for o, a in opts:
1499             if o in ("-h", "--help"):
1500                 usage()
1501                 sys.exit()
1502             elif o in ("-c", "--config"):
1503                 config_file = a
1504             # elif o == "--log-socket-port":
1505             #     log_socket_port = a
1506             # elif o == "--log-socket-host":
1507             #     log_socket_host = a
1508             # elif o == "--log-file":
1509             #     log_file = a
1510             else:
1511                 assert False, "Unhandled option"
1512         if config_file:
1513             if not path.isfile(config_file):
1514                 print("configuration file '{}' that not exist".format(config_file), file=sys.stderr)
1515                 exit(1)
1516         else:
1517             for config_file in (__file__[:__file__.rfind(".")] + ".cfg", "./lcm.cfg", "/etc/osm/lcm.cfg"):
1518                 if path.isfile(config_file):
1519                     break
1520             else:
1521                 print("No configuration file 'nbi.cfg' found neither at local folder nor at /etc/osm/", file=sys.stderr)
1522                 exit(1)
1523         lcm = Lcm(config_file)
1524         lcm.start()
1525     except getopt.GetoptError as e:
1526         print(str(e), file=sys.stderr)
1527         # usage()
1528         exit(1)