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