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