lcm/lcm.py

   1 #!/usr/bin/python3
   2 # -*- coding: utf-8 -*-
   3
   4 import asyncio
   5 import aiohttp
   6 import yaml
   7 import ROclient
   8 import time
   9 import dbmemory
  10 import logging
  11
  12 from copy import deepcopy
  13 from uuid import uuid4
  14
  15 #streamformat = "%(asctime)s %(name)s %(levelname)s: %(message)s"
  16 streamformat = "%(name)s %(levelname)s: %(message)s"
  17 logging.basicConfig(format=streamformat, level=logging.DEBUG)
  18 logger = logging.getLogger('lcm')
  19
  20 ro_account = {
  21     "url": "http://localhost:9090/openmano",
  22     "tenant": "osm"
  23 }
  24
  25 vca_account = {
  26     # TODO
  27 }
  28
  29 # conains created tasks/futures to be able to cancel
  30 lcm_tasks = {}
  31
  32 headers_req = {'Accept': 'application/yaml', 'content-type': 'application/yaml'}
  33 ns_status = ("CREATION-SCHEDULED", "DEPLOYING", "CONFIGURING", "DELETION-SCHEDULED", "UN-CONFIGURING", "UNDEPLOYING")
  34
  35 # TODO replace with database calls
  36 db = dbmemory.dbmemory()
  37
  38
  39
  40 async def CreateNS(loop, nsr_id):
  41     logger.debug("CreateNS task nsr_id={} Enter".format(nsr_id))
  42     nsr_lcm = {
  43         "id": nsr_id,
  44         "RO": {"vnfd_id": {}, "nsd_id": None, "nsr_id": None, "nsr_status": "SCHEDULED"},
  45         "nsr_ip": {},
  46         "VCA": {"TODO"},
  47         "status": "BUILD",
  48         "status_detailed": "",
  49     }
  50
  51     deloyment_timeout = 120
  52     try:
  53         ns_request = db.get_one("ns_request", {"id": nsr_id})
  54         nsd = db.get_one("nsd", {"id": ns_request["nsd_id"]})
  55         RO = ROclient.ROClient(loop, endpoint_url=ro_account["url"], tenant=ro_account["tenant"],
  56                                datacenter=ns_request["vim"])
  57         nsr_lcm["status_detailed"] = "Creating vnfd at RO"
  58         # ns_request["constituent-vnfr-ref"] = []
  59
  60         db.create("nsr_lcm", nsr_lcm)
  61
  62         # get vnfds, instantiate at RO
  63         logger.debug("CreateNS task nsr_id={} RO VNFD".format(nsr_id))
  64         for c_vnf in nsd["constituent-vnfd"]:
  65             vnfd_id = c_vnf["vnfd-id-ref"]
  66             vnfd = db.get_one("vnfd", {"id": vnfd_id})
  67             vnfd.pop("_admin", None)
  68             vnfd.pop("_id", None)
  69             # vnfr = deepcopy(vnfd)
  70             # vnfr["member-vnf-index"] = c_vnf["member-vnf-index"]
  71             # vnfr["nsr-id"] = nsr_id
  72             # vnfr["id"] = uuid4()
  73             # vnfr["vnf-id"] = vnfd["id"]
  74             # ns_request["constituent-vnfr-ref"],append(vnfd_id)
  75
  76             # TODO change id for RO in case it is present
  77             try:
  78                 desc = await RO.create("vnfd", descriptor=vnfd)
  79                 nsr_lcm["RO"]["vnfd_id"][vnfd_id] = desc["uuid"]
  80                 db.replace("nsr_lcm", {"id": nsr_id}, nsr_lcm)
  81             except ROclient.ROClientException as e:
  82                 if e.http_code == 409:  # conflict, vnfd already present
  83                     print("debug", e)
  84                 else:
  85                     raise
  86
  87             # db_new("vnfr", vnfr)
  88             # db_update("ns_request", nsr_id, ns_request)
  89
  90         # create nsd at RO
  91         logger.debug("CreateNS task nsr_id={} RO NSD".format(nsr_id))
  92         nsr_lcm["status_detailed"] = "Creating nsd at RO"
  93         nsd_id = ns_request["nsd_id"]
  94         nsd = db.get_one("nsd", {"id": nsd_id})
  95         nsd.pop("_admin", None)
  96         nsd.pop("_id", None)
  97         try:
  98             desc = await RO.create("nsd", descriptor=nsd)
  99             nsr_lcm["RO"]["nsd_id"] = desc["uuid"]
 100             db.replace("nsr_lcm", {"id": nsr_id}, nsr_lcm)
 101         except ROclient.ROClientException as e:
 102             if e.http_code == 409:  # conflict, nsd already present
 103                 print("debug", e)
 104             else:
 105                 raise
 106
 107         # Crate ns at RO
 108         logger.debug("CreateNS task nsr_id={} RO NS".format(nsr_id))
 109         nsr_lcm["status_detailed"] = "Creating ns at RO"
 110         desc = await RO.create("ns", name=ns_request["name"], datacenter=ns_request["vim"], scenario=nsr_lcm["RO"]["nsd_id"])
 111         RO_nsr_id = desc["uuid"]
 112         nsr_lcm["RO"]["nsr_id"] = RO_nsr_id
 113         nsr_lcm["RO"]["nsr_status"] = "BUILD"
 114         db.replace("nsr_lcm", {"id": nsr_id}, nsr_lcm)
 115
 116         # wait until NS is ready
 117         deloyment_timeout = 600
 118         while deloyment_timeout > 0:
 119             ns_status_detailed = "Waiting ns ready at RO"
 120             nsr_lcm["status_detailed"] = ns_status_detailed
 121             desc = await RO.show("ns", RO_nsr_id)
 122             ns_status, ns_status_info = RO.check_ns_status(desc)
 123             nsr_lcm["RO"]["nsr_status"] = ns_status
 124             if ns_status == "ERROR":
 125                 raise ROclient.ROClientException(ns_status_info)
 126             elif ns_status == "BUILD":
 127                 nsr_lcm["status_detailed"] = ns_status_detailed + "; nsr_id: '{}', {}".format(nsr_id, ns_status_info)
 128             elif ns_status == "ACTIVE":
 129                 nsr_lcm["nsr_ip"] = RO.get_ns_vnf_ip(desc)
 130                 break
 131             else:
 132                 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status)
 133
 134             await asyncio.sleep(5, loop=loop)
 135             deloyment_timeout -= 5
 136         if deloyment_timeout <= 0:
 137             raise ROclient.ROClientException("Timeot wating ns to be ready")
 138         nsr_lcm["status_detailed"] = "Configuring vnfr"
 139         db.replace("nsr_lcm", {"id": nsr_id}, nsr_lcm)
 140
 141         #for nsd in nsr_lcm["descriptors"]["nsd"]:
 142
 143         logger.debug("CreateNS task nsr_id={} VCA look for".format(nsr_id))
 144         for c_vnf in nsd["constituent-vnfd"]:
 145             vnfd_id = c_vnf["vnfd-id-ref"]
 146             vnfd_index = int(c_vnf["member-vnf-index"])
 147             vnfd = db.get_one("vnfd", {"id": vnfd_id})
 148             if vnfd.get("vnf-configuration") and vnfd["vnf-configuration"].get("juju"):
 149                 proxy_charm = vnfd["vnf-configuration"]["juju"]["charm"]
 150                 config_primitive = vnfd["vnf-configuration"].get("config-primitive")
 151                 # get parameters for juju charm
 152                 base_folder = vnfd["_admin"]["storage"]
 153                 path = base_folder + "/charms/" + proxy_charm
 154                 mgmt_ip = nsr_lcm['nsr_ip'][vnfd_index]
 155                 # TODO launch VCA charm
 156                 # task = asyncio.ensure_future(DeployCharm(loop, path, mgmt_ip, config_primitive))
 157         nsr_lcm["status"] = "DONE"
 158         db.replace("nsr_lcm", {"id": nsr_id}, nsr_lcm)
 159
 160         return nsr_lcm
 161
 162     except (ROclient.ROClientException, Exception) as e:
 163         logger.debug("CreateNS nsr_id={} Exception {}".format(nsr_id, e), exc_info=True)
 164         nsr_lcm["status"] = "ERROR"
 165         nsr_lcm["status_detailed"] += ": ERROR {}".format(e)
 166     finally:
 167         logger.debug("CreateNS task nsr_id={} Exit".format(nsr_id))
 168
 169
 170 async def DestroyNS(loop, nsr_id):
 171     logger.debug("DestroyNS task nsr_id={} Enter".format(nsr_id))
 172     nsr_lcm = db.get_one("nsr_lcm", {"id": nsr_id})
 173     ns_request = db.get_one("ns_request", {"id": nsr_id})
 174
 175     nsr_lcm["status"] = "DELETING"
 176     nsr_lcm["status_detailed"] = "Deleting charms"
 177     db.replace("nsr_lcm", {"id": nsr_id}, nsr_lcm)
 178     # TODO destroy VCA charm
 179
 180     # remove from RO
 181     RO = ROclient.ROClient(loop, endpoint_url=ro_account["url"], tenant=ro_account["tenant"],
 182                            datacenter=ns_request["vim"])
 183     # Delete ns
 184     try:
 185         RO_nsr_id = nsr_lcm["RO"]["nsr_id"]
 186         if RO_nsr_id:
 187             nsr_lcm["status_detailed"] = "Deleting ns at RO"
 188             desc = await RO.delete("ns", RO_nsr_id)
 189             print("debug", "deleted RO ns {}".format(RO_nsr_id))
 190             nsr_lcm["RO"]["nsr_id"] = None
 191             nsr_lcm["RO"]["nsr_status"] = "DELETED"
 192             db.replace("nsr_lcm", {"id": nsr_id}, nsr_lcm)
 193     except ROclient.ROClientException as e:
 194         if e.http_code == 404:
 195             nsr_lcm["RO"]["nsr_id"] = None
 196             nsr_lcm["RO"]["nsr_status"] = "DELETED"
 197             db.replace("nsr_lcm", {"id": nsr_id}, nsr_lcm)
 198             print("warning", e)
 199         else:
 200             print("error", e)
 201
 202     # Delete nsd
 203     try:
 204         RO_nsd_id = nsr_lcm["RO"]["nsd_id"]
 205         if RO_nsd_id:
 206             nsr_lcm["status_detailed"] = "Deleting nsd at RO"
 207             desc = await RO.delete("nsd", RO_nsd_id)
 208             print("debug", "deleted RO nsd {}".format(RO_nsd_id))
 209             nsr_lcm["RO"]["nsd_id"] = None
 210             db.replace("nsr_lcm", {"id": nsr_id}, nsr_lcm)
 211     except ROclient.ROClientException as e:
 212         if e.http_code == 404:
 213             nsr_lcm["RO"]["nsd_id"] = None
 214             print("warning", e)
 215         else:
 216             print("error", e)
 217
 218     for vnf_id, RO_vnfd_id in nsr_lcm["RO"]["vnfd_id"].items():
 219         try:
 220             if RO_vnfd_id:
 221                 nsr_lcm["status_detailed"] = "Deleting vnfd at RO"
 222                 desc = await RO.delete("vnfd", RO_vnfd_id)
 223                 print("debug", "deleted RO vnfd {}".format(RO_vnfd_id))
 224                 nsr_lcm["RO"]["vnfd_id"][vnf_id] = None
 225                 db.replace("nsr_lcm", {"id": nsr_id}, nsr_lcm)
 226         except ROclient.ROClientException as e:
 227             if e.http_code == 404:
 228                 nsr_lcm["RO"]["vnfd_id"][vnf_id] = None
 229                 print("warning", e)
 230             else:
 231                 print("error", e)
 232     logger.debug("DestroyNS task nsr_id={} Exit".format(nsr_id))
 233
 234
 235 async def test(loop, param=None):
 236     logger.debug("Starting/Ending test task: {}".format(param))
 237
 238
 239 def cancel_tasks(loop, nsr_id):
 240     """
 241     Cancel all active tasks of a concrete nsr identified for nsr_id
 242     :param loop: loop
 243     :param nsr_id:  nsr identity
 244     :return: None, or raises an exception if not possible
 245     """
 246     global lcm_tasks
 247     if not lcm_tasks.get(nsr_id):
 248         return
 249     for order_id, tasks_set in lcm_tasks[nsr_id].items():
 250         for task_name, task in tasks_set.items():
 251             result = task.cancel()
 252             if result:
 253                 logger.debug("nsr_id={} order_id={} task={} cancelled".format(nsr_id, order_id, task_name))
 254     lcm_tasks[nsr_id] = {}
 255
 256
 257
 258 async def read_kafka(loop, bus_info):
 259     global lcm_tasks
 260     logger.debug("kafka task Enter")
 261     order_id = 1
 262     # future = asyncio.Future()
 263     with open(bus_info["file"]) as f:
 264
 265         # ignore old orders. Read file
 266         command = "fake"
 267         while command:
 268             command = f.read()
 269
 270         while True:
 271             command = f.read()
 272             if not command:
 273                 await asyncio.sleep(2, loop=loop)
 274                 continue
 275             order_id += 1
 276             command = command.strip()
 277             command, _, params = command.partition(" ")
 278             if command == "exit":
 279                 print("Bye!")
 280                 break
 281             elif command.startswith("#"):
 282                 continue
 283             elif command == "echo":
 284                 print(params)
 285             elif command == "test":
 286                 asyncio.Task(test(loop, params), loop=loop)
 287             elif command == "break":
 288                 print("put a break in this line of code")
 289             elif command == "new-ns":
 290                 nsr_id = params.strip()
 291                 logger.debug("Deploying NS {}".format(nsr_id))
 292                 task = asyncio.ensure_future(CreateNS(loop, nsr_id))
 293                 if nsr_id not in lcm_tasks:
 294                     lcm_tasks[nsr_id] = {}
 295                 lcm_tasks[nsr_id][order_id] = {"CreateNS": task}
 296             elif command == "del-ns":
 297                 nsr_id = params.strip()
 298                 logger.debug("Deleting NS {}".format(nsr_id))
 299                 cancel_tasks(loop, nsr_id)
 300                 task = asyncio.ensure_future(DestroyNS(loop, nsr_id))
 301                 if nsr_id not in lcm_tasks:
 302                     lcm_tasks[nsr_id] = {}
 303                 lcm_tasks[nsr_id][order_id] = {"DestroyNS": task}
 304             elif command == "get-ns":
 305                 nsr_id = params.strip()
 306                 nsr_lcm = db.get_one("nsr_lcm", {"id": nsr_id})
 307                 print("nsr_lcm", nsr_lcm)
 308                 print("lcm_tasks", lcm_tasks.get(nsr_id))
 309             else:
 310                 logger.debug("unknown command '{}'".format(command))
 311                 print("Usage:\n  echo <>\n  new-ns <ns1|ns2>\n  del-ns <ns1|ns2>\n  get-ns <ns1|ns2>")
 312     logger.debug("kafka task Exit")
 313
 314
 315 def lcm():
 316     loop = asyncio.get_event_loop()
 317     loop.run_until_complete(read_kafka(loop, {"file": "/home/atierno/OSM/osm/NBI/kafka"}))
 318     return
 319
 320
 321 def lcm2():
 322     loop = asyncio.get_event_loop()
 323     # asyncio.ensure_future(CreateNS, loop)
 324     try:
 325         content = loop.run_until_complete(CreateNS(loop, "ns1"))
 326         print("Done: {}".format(content))
 327     except ROclient.ROClientException as e:
 328         print("Error {}".format(e))
 329
 330     time.sleep(10)
 331
 332     content = loop.run_until_complete(DestroyNS(loop, "ns1"))
 333     print(content)
 334
 335     loop.close()
 336
 337
 338 if __name__ == '__main__':
 339
 340     # FOR TEST
 341     RO_VIM = "OST2_MRT"
 342
 343     #FILL DATABASE
 344     with open("/home/atierno/OSM/osm/devops/descriptor-packages/vnfd/ping_vnf/src/ping_vnfd.yaml") as f:
 345         vnfd = yaml.load(f)
 346         vnfd_clean, _ = ROclient.remove_envelop("vnfd", vnfd)
 347         vnfd_clean["_admin"] = {"storage": "/home/atierno/OSM/osm/devops/descriptor-packages/vnfd/ping_vnf"}
 348         db.create("vnfd", vnfd_clean)
 349     with open("/home/atierno/OSM/osm/devops/descriptor-packages/vnfd/pong_vnf/src/pong_vnfd.yaml") as f:
 350         vnfd = yaml.load(f)
 351         vnfd_clean, _ = ROclient.remove_envelop("vnfd", vnfd)
 352         vnfd_clean["_admin"] = {"storage": "/home/atierno/OSM/osm/devops/descriptor-packages/vnfd/pong_vnf"}
 353         db.create("vnfd", vnfd_clean)
 354     with open("/home/atierno/OSM/osm/devops/descriptor-packages/nsd/ping_pong_ns/src/ping_pong_nsd.yaml") as f:
 355         nsd = yaml.load(f)
 356         nsd_clean, _ = ROclient.remove_envelop("nsd", nsd)
 357         nsd_clean["_admin"] = {"storage": "/home/atierno/OSM/osm/devops/descriptor-packages/nsd/ping_pong_ns"}
 358         db.create("nsd", nsd_clean)
 359
 360     ns_request = {
 361         "id": "ns1",
 362         "nsr_id": "ns1",
 363         "name": "pingpongOne",
 364         "vim": RO_VIM,
 365         "nsd_id": nsd_clean["id"],  # nsd_ping_pong
 366     }
 367     db.create("ns_request", ns_request)
 368     ns_request = {
 369         "id": "ns2",
 370         "nsr_id": "ns2",
 371         "name": "pingpongTwo",
 372         "vim": RO_VIM,
 373         "nsd_id": nsd_clean["id"],  # nsd_ping_pong
 374     }
 375     db.create("ns_request", ns_request)
 376     # lcm2()
 377     lcm()
 378
 379
 380