1 # -*- coding: utf-8 -*-
4 # Copyright 2020 Telefonica Investigacion y Desarrollo, S.A.U.
6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
7 # not use this file except in compliance with the License. You may obtain
8 # a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15 # License for the specific language governing permissions and limitations
21 This is thread that interacts with a VIM. It processes TASKs sequentially against a single VIM.
22 The tasks are stored at database in table ro_tasks
23 A single ro_task refers to a VIM element (flavor, image, network, ...).
24 A ro_task can contain several 'tasks', each one with a target, where to store the results
32 from copy
import deepcopy
33 from http
import HTTPStatus
35 from pkg_resources
import iter_entry_points
36 from shutil
import rmtree
37 from unittest
.mock
import Mock
39 # from osm_common import dbmongo, dbmemory, fslocal, fsmongo, msglocal, msgkafka, version as common_version
40 from osm_common
.dbbase
import DbException
41 from osm_ro_plugin
.vim_dummy
import VimDummyConnector
42 from osm_ro_plugin
.sdn_dummy
import SdnDummyConnector
43 from osm_ro_plugin
import vimconn
, sdnconn
44 from osm_ng_ro
.vim_admin
import LockRenew
47 __author__
= "Alfonso Tierno"
48 __date__
= "$28-Sep-2017 12:07:15$"
51 def deep_get(target_dict
, *args
, **kwargs
):
53 Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
54 Example target_dict={a: {b: 5}}; key_list=[a,b] returns 5; both key_list=[a,b,c] and key_list=[f,h] return None
55 :param target_dict: dictionary to be read
56 :param args: list of keys to read from target_dict
57 :param kwargs: only can contain default=value to return if key is not present in the nested dictionary
58 :return: The wanted value if exist, None or default otherwise
61 if not isinstance(target_dict
, dict) or key
not in target_dict
:
62 return kwargs
.get("default")
63 target_dict
= target_dict
[key
]
67 class NsWorkerException(Exception):
71 class FailingConnector
:
72 def __init__(self
, error_msg
):
73 self
.error_msg
= error_msg
75 for method
in dir(vimconn
.VimConnector
):
78 self
, method
, Mock(side_effect
=vimconn
.VimConnException(error_msg
))
81 for method
in dir(sdnconn
.SdnConnectorBase
):
84 self
, method
, Mock(side_effect
=sdnconn
.SdnConnectorError(error_msg
))
88 class NsWorkerExceptionNotFound(NsWorkerException
):
92 class VimInteractionBase
:
93 """Base class to call VIM/SDN for creating, deleting and refresh networks, VMs, flavors, ...
94 It implements methods that does nothing and return ok"""
96 def __init__(self
, db
, my_vims
, db_vims
, logger
):
99 self
.my_vims
= my_vims
100 self
.db_vims
= db_vims
102 def new(self
, ro_task
, task_index
, task_depends
):
105 def refresh(self
, ro_task
):
106 """skip calling VIM to get image, flavor status. Assumes ok"""
107 if ro_task
["vim_info"]["vim_status"] == "VIM_ERROR":
112 def delete(self
, ro_task
, task_index
):
113 """skip calling VIM to delete image. Assumes ok"""
116 def exec(self
, ro_task
, task_index
, task_depends
):
117 return "DONE", None, None
120 class VimInteractionNet(VimInteractionBase
):
121 def new(self
, ro_task
, task_index
, task_depends
):
123 task
= ro_task
["tasks"][task_index
]
124 task_id
= task
["task_id"]
127 target_vim
= self
.my_vims
[ro_task
["target_id"]]
131 if task
.get("find_params"):
132 # if management, get configuration of VIM
133 if task
["find_params"].get("filter_dict"):
134 vim_filter
= task
["find_params"]["filter_dict"]
136 elif task
["find_params"].get("mgmt"):
138 self
.db_vims
[ro_task
["target_id"]],
140 "management_network_id",
143 "id": self
.db_vims
[ro_task
["target_id"]]["config"][
144 "management_network_id"
148 self
.db_vims
[ro_task
["target_id"]],
150 "management_network_name",
153 "name": self
.db_vims
[ro_task
["target_id"]]["config"][
154 "management_network_name"
158 vim_filter
= {"name": task
["find_params"]["name"]}
160 raise NsWorkerExceptionNotFound(
161 "Invalid find_params for new_net {}".format(task
["find_params"])
164 vim_nets
= target_vim
.get_network_list(vim_filter
)
165 if not vim_nets
and not task
.get("params"):
166 raise NsWorkerExceptionNotFound(
167 "Network not found with this criteria: '{}'".format(
168 task
.get("find_params")
171 elif len(vim_nets
) > 1:
172 raise NsWorkerException(
173 "More than one network found with this criteria: '{}'".format(
179 vim_net_id
= vim_nets
[0]["id"]
182 params
= task
["params"]
183 vim_net_id
, created_items
= target_vim
.new_network(**params
)
186 ro_vim_item_update
= {
187 "vim_id": vim_net_id
,
188 "vim_status": "BUILD",
190 "created_items": created_items
,
194 "task={} {} new-net={} created={}".format(
195 task_id
, ro_task
["target_id"], vim_net_id
, created
199 return "BUILD", ro_vim_item_update
200 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
202 "task={} vim={} new-net: {}".format(task_id
, ro_task
["target_id"], e
)
204 ro_vim_item_update
= {
205 "vim_status": "VIM_ERROR",
207 "vim_details": str(e
),
210 return "FAILED", ro_vim_item_update
212 def refresh(self
, ro_task
):
213 """Call VIM to get network status"""
214 ro_task_id
= ro_task
["_id"]
215 target_vim
= self
.my_vims
[ro_task
["target_id"]]
216 vim_id
= ro_task
["vim_info"]["vim_id"]
217 net_to_refresh_list
= [vim_id
]
220 vim_dict
= target_vim
.refresh_nets_status(net_to_refresh_list
)
221 vim_info
= vim_dict
[vim_id
]
223 if vim_info
["status"] == "ACTIVE":
225 elif vim_info
["status"] == "BUILD":
226 task_status
= "BUILD"
228 task_status
= "FAILED"
229 except vimconn
.VimConnException
as e
:
230 # Mark all tasks at VIM_ERROR status
232 "ro_task={} vim={} get-net={}: {}".format(
233 ro_task_id
, ro_task
["target_id"], vim_id
, e
236 vim_info
= {"status": "VIM_ERROR", "error_msg": str(e
)}
237 task_status
= "FAILED"
239 ro_vim_item_update
= {}
240 if ro_task
["vim_info"]["vim_status"] != vim_info
["status"]:
241 ro_vim_item_update
["vim_status"] = vim_info
["status"]
243 if ro_task
["vim_info"]["vim_name"] != vim_info
.get("name"):
244 ro_vim_item_update
["vim_name"] = vim_info
.get("name")
246 if vim_info
["status"] in ("ERROR", "VIM_ERROR"):
247 if ro_task
["vim_info"]["vim_details"] != vim_info
.get("error_msg"):
248 ro_vim_item_update
["vim_details"] = vim_info
.get("error_msg")
249 elif vim_info
["status"] == "DELETED":
250 ro_vim_item_update
["vim_id"] = None
251 ro_vim_item_update
["vim_details"] = "Deleted externally"
253 if ro_task
["vim_info"]["vim_details"] != vim_info
["vim_info"]:
254 ro_vim_item_update
["vim_details"] = vim_info
["vim_info"]
256 if ro_vim_item_update
:
258 "ro_task={} {} get-net={}: status={} {}".format(
260 ro_task
["target_id"],
262 ro_vim_item_update
.get("vim_status"),
263 ro_vim_item_update
.get("vim_details")
264 if ro_vim_item_update
.get("vim_status") != "ACTIVE"
269 return task_status
, ro_vim_item_update
271 def delete(self
, ro_task
, task_index
):
272 task
= ro_task
["tasks"][task_index
]
273 task_id
= task
["task_id"]
274 net_vim_id
= ro_task
["vim_info"]["vim_id"]
275 ro_vim_item_update_ok
= {
276 "vim_status": "DELETED",
278 "vim_details": "DELETED",
283 if net_vim_id
or ro_task
["vim_info"]["created_items"]:
284 target_vim
= self
.my_vims
[ro_task
["target_id"]]
285 target_vim
.delete_network(
286 net_vim_id
, ro_task
["vim_info"]["created_items"]
288 except vimconn
.VimConnNotFoundException
:
289 ro_vim_item_update_ok
["vim_details"] = "already deleted"
290 except vimconn
.VimConnException
as e
:
292 "ro_task={} vim={} del-net={}: {}".format(
293 ro_task
["_id"], ro_task
["target_id"], net_vim_id
, e
296 ro_vim_item_update
= {
297 "vim_status": "VIM_ERROR",
298 "vim_details": "Error while deleting: {}".format(e
),
301 return "FAILED", ro_vim_item_update
304 "task={} {} del-net={} {}".format(
306 ro_task
["target_id"],
308 ro_vim_item_update_ok
.get("vim_details", ""),
312 return "DONE", ro_vim_item_update_ok
315 class VimInteractionVdu(VimInteractionBase
):
316 max_retries_inject_ssh_key
= 20 # 20 times
317 time_retries_inject_ssh_key
= 30 # wevery 30 seconds
319 def new(self
, ro_task
, task_index
, task_depends
):
320 task
= ro_task
["tasks"][task_index
]
321 task_id
= task
["task_id"]
324 target_vim
= self
.my_vims
[ro_task
["target_id"]]
328 params
= task
["params"]
329 params_copy
= deepcopy(params
)
330 net_list
= params_copy
["net_list"]
333 # change task_id into network_id
334 if "net_id" in net
and net
["net_id"].startswith("TASK-"):
335 network_id
= task_depends
[net
["net_id"]]
338 raise NsWorkerException(
339 "Cannot create VM because depends on a network not created or found "
340 "for {}".format(net
["net_id"])
343 net
["net_id"] = network_id
345 if params_copy
["image_id"].startswith("TASK-"):
346 params_copy
["image_id"] = task_depends
[params_copy
["image_id"]]
348 if params_copy
["flavor_id"].startswith("TASK-"):
349 params_copy
["flavor_id"] = task_depends
[params_copy
["flavor_id"]]
351 vim_vm_id
, created_items
= target_vim
.new_vminstance(**params_copy
)
352 interfaces
= [iface
["vim_id"] for iface
in params_copy
["net_list"]]
354 ro_vim_item_update
= {
356 "vim_status": "BUILD",
358 "created_items": created_items
,
360 "interfaces_vim_ids": interfaces
,
364 "task={} {} new-vm={} created={}".format(
365 task_id
, ro_task
["target_id"], vim_vm_id
, created
369 return "BUILD", ro_vim_item_update
370 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
372 "task={} {} new-vm: {}".format(task_id
, ro_task
["target_id"], e
)
374 ro_vim_item_update
= {
375 "vim_status": "VIM_ERROR",
377 "vim_details": str(e
),
380 return "FAILED", ro_vim_item_update
382 def delete(self
, ro_task
, task_index
):
383 task
= ro_task
["tasks"][task_index
]
384 task_id
= task
["task_id"]
385 vm_vim_id
= ro_task
["vim_info"]["vim_id"]
386 ro_vim_item_update_ok
= {
387 "vim_status": "DELETED",
389 "vim_details": "DELETED",
394 if vm_vim_id
or ro_task
["vim_info"]["created_items"]:
395 target_vim
= self
.my_vims
[ro_task
["target_id"]]
396 target_vim
.delete_vminstance(
397 vm_vim_id
, ro_task
["vim_info"]["created_items"]
399 except vimconn
.VimConnNotFoundException
:
400 ro_vim_item_update_ok
["vim_details"] = "already deleted"
401 except vimconn
.VimConnException
as e
:
403 "ro_task={} vim={} del-vm={}: {}".format(
404 ro_task
["_id"], ro_task
["target_id"], vm_vim_id
, e
407 ro_vim_item_update
= {
408 "vim_status": "VIM_ERROR",
409 "vim_details": "Error while deleting: {}".format(e
),
412 return "FAILED", ro_vim_item_update
415 "task={} {} del-vm={} {}".format(
417 ro_task
["target_id"],
419 ro_vim_item_update_ok
.get("vim_details", ""),
423 return "DONE", ro_vim_item_update_ok
425 def refresh(self
, ro_task
):
426 """Call VIM to get vm status"""
427 ro_task_id
= ro_task
["_id"]
428 target_vim
= self
.my_vims
[ro_task
["target_id"]]
429 vim_id
= ro_task
["vim_info"]["vim_id"]
434 vm_to_refresh_list
= [vim_id
]
436 vim_dict
= target_vim
.refresh_vms_status(vm_to_refresh_list
)
437 vim_info
= vim_dict
[vim_id
]
439 if vim_info
["status"] == "ACTIVE":
441 elif vim_info
["status"] == "BUILD":
442 task_status
= "BUILD"
444 task_status
= "FAILED"
446 # try to load and parse vim_information
448 vim_info_info
= yaml
.safe_load(vim_info
["vim_info"])
449 if vim_info_info
.get("name"):
450 vim_info
["name"] = vim_info_info
["name"]
453 except vimconn
.VimConnException
as e
:
454 # Mark all tasks at VIM_ERROR status
456 "ro_task={} vim={} get-vm={}: {}".format(
457 ro_task_id
, ro_task
["target_id"], vim_id
, e
460 vim_info
= {"status": "VIM_ERROR", "error_msg": str(e
)}
461 task_status
= "FAILED"
463 ro_vim_item_update
= {}
465 # Interfaces cannot be present if e.g. VM is not present, that is status=DELETED
467 if vim_info
.get("interfaces"):
468 for vim_iface_id
in ro_task
["vim_info"]["interfaces_vim_ids"]:
472 for iface
in vim_info
["interfaces"]
473 if vim_iface_id
== iface
["vim_interface_id"]
478 # iface.pop("vim_info", None)
479 vim_interfaces
.append(iface
)
483 for t
in ro_task
["tasks"]
484 if t
and t
["action"] == "CREATE" and t
["status"] != "FINISHED"
486 if vim_interfaces
and task_create
.get("mgmt_vnf_interface") is not None:
487 vim_interfaces
[task_create
["mgmt_vnf_interface"]][
491 mgmt_vdu_iface
= task_create
.get(
492 "mgmt_vdu_interface", task_create
.get("mgmt_vnf_interface", 0)
495 vim_interfaces
[mgmt_vdu_iface
]["mgmt_vdu_interface"] = True
497 if ro_task
["vim_info"]["interfaces"] != vim_interfaces
:
498 ro_vim_item_update
["interfaces"] = vim_interfaces
500 if ro_task
["vim_info"]["vim_status"] != vim_info
["status"]:
501 ro_vim_item_update
["vim_status"] = vim_info
["status"]
503 if ro_task
["vim_info"]["vim_name"] != vim_info
.get("name"):
504 ro_vim_item_update
["vim_name"] = vim_info
.get("name")
506 if vim_info
["status"] in ("ERROR", "VIM_ERROR"):
507 if ro_task
["vim_info"]["vim_details"] != vim_info
.get("error_msg"):
508 ro_vim_item_update
["vim_details"] = vim_info
.get("error_msg")
509 elif vim_info
["status"] == "DELETED":
510 ro_vim_item_update
["vim_id"] = None
511 ro_vim_item_update
["vim_details"] = "Deleted externally"
513 if ro_task
["vim_info"]["vim_details"] != vim_info
["vim_info"]:
514 ro_vim_item_update
["vim_details"] = vim_info
["vim_info"]
516 if ro_vim_item_update
:
518 "ro_task={} {} get-vm={}: status={} {}".format(
520 ro_task
["target_id"],
522 ro_vim_item_update
.get("vim_status"),
523 ro_vim_item_update
.get("vim_details")
524 if ro_vim_item_update
.get("vim_status") != "ACTIVE"
529 return task_status
, ro_vim_item_update
531 def exec(self
, ro_task
, task_index
, task_depends
):
532 task
= ro_task
["tasks"][task_index
]
533 task_id
= task
["task_id"]
534 target_vim
= self
.my_vims
[ro_task
["target_id"]]
535 db_task_update
= {"retries": 0}
536 retries
= task
.get("retries", 0)
539 params
= task
["params"]
540 params_copy
= deepcopy(params
)
541 params_copy
["ro_key"] = self
.db
.decrypt(
542 params_copy
.pop("private_key"),
543 params_copy
.pop("schema_version"),
544 params_copy
.pop("salt"),
546 params_copy
["ip_addr"] = params_copy
.pop("ip_address")
547 target_vim
.inject_user_key(**params_copy
)
549 "task={} {} action-vm=inject_key".format(task_id
, ro_task
["target_id"])
556 ) # params_copy["key"]
557 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
560 if retries
< self
.max_retries_inject_ssh_key
:
566 "next_retry": self
.time_retries_inject_ssh_key
,
571 "task={} {} inject-ssh-key: {}".format(task_id
, ro_task
["target_id"], e
)
573 ro_vim_item_update
= {"vim_details": str(e
)}
575 return "FAILED", ro_vim_item_update
, db_task_update
578 class VimInteractionImage(VimInteractionBase
):
579 def new(self
, ro_task
, task_index
, task_depends
):
580 task
= ro_task
["tasks"][task_index
]
581 task_id
= task
["task_id"]
584 target_vim
= self
.my_vims
[ro_task
["target_id"]]
588 if task
.get("find_params"):
589 vim_images
= target_vim
.get_image_list(**task
["find_params"])
592 raise NsWorkerExceptionNotFound(
593 "Image not found with this criteria: '{}'".format(
597 elif len(vim_images
) > 1:
598 raise NsWorkerException(
599 "More than one image found with this criteria: '{}'".format(
604 vim_image_id
= vim_images
[0]["id"]
606 ro_vim_item_update
= {
607 "vim_id": vim_image_id
,
608 "vim_status": "DONE",
610 "created_items": created_items
,
614 "task={} {} new-image={} created={}".format(
615 task_id
, ro_task
["target_id"], vim_image_id
, created
619 return "DONE", ro_vim_item_update
620 except (NsWorkerException
, vimconn
.VimConnException
) as e
:
622 "task={} {} new-image: {}".format(task_id
, ro_task
["target_id"], e
)
624 ro_vim_item_update
= {
625 "vim_status": "VIM_ERROR",
627 "vim_details": str(e
),
630 return "FAILED", ro_vim_item_update
633 class VimInteractionFlavor(VimInteractionBase
):
634 def delete(self
, ro_task
, task_index
):
635 task
= ro_task
["tasks"][task_index
]
636 task_id
= task
["task_id"]
637 flavor_vim_id
= ro_task
["vim_info"]["vim_id"]
638 ro_vim_item_update_ok
= {
639 "vim_status": "DELETED",
641 "vim_details": "DELETED",
647 target_vim
= self
.my_vims
[ro_task
["target_id"]]
648 target_vim
.delete_flavor(flavor_vim_id
)
649 except vimconn
.VimConnNotFoundException
:
650 ro_vim_item_update_ok
["vim_details"] = "already deleted"
651 except vimconn
.VimConnException
as e
:
653 "ro_task={} vim={} del-flavor={}: {}".format(
654 ro_task
["_id"], ro_task
["target_id"], flavor_vim_id
, e
657 ro_vim_item_update
= {
658 "vim_status": "VIM_ERROR",
659 "vim_details": "Error while deleting: {}".format(e
),
662 return "FAILED", ro_vim_item_update
665 "task={} {} del-flavor={} {}".format(
667 ro_task
["target_id"],
669 ro_vim_item_update_ok
.get("vim_details", ""),
673 return "DONE", ro_vim_item_update_ok
675 def new(self
, ro_task
, task_index
, task_depends
):
676 task
= ro_task
["tasks"][task_index
]
677 task_id
= task
["task_id"]
680 target_vim
= self
.my_vims
[ro_task
["target_id"]]
686 if task
.get("find_params"):
688 flavor_data
= task
["find_params"]["flavor_data"]
689 vim_flavor_id
= target_vim
.get_flavor_id_from_data(flavor_data
)
690 except vimconn
.VimConnNotFoundException
:
693 if not vim_flavor_id
and task
.get("params"):
695 flavor_data
= task
["params"]["flavor_data"]
696 vim_flavor_id
= target_vim
.new_flavor(flavor_data
)
699 ro_vim_item_update
= {
700 "vim_id": vim_flavor_id
,
701 "vim_status": "DONE",
703 "created_items": created_items
,
707 "task={} {} new-flavor={} created={}".format(
708 task_id
, ro_task
["target_id"], vim_flavor_id
, created
712 return "DONE", ro_vim_item_update
713 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
715 "task={} vim={} new-flavor: {}".format(task_id
, ro_task
["target_id"], e
)
717 ro_vim_item_update
= {
718 "vim_status": "VIM_ERROR",
720 "vim_details": str(e
),
723 return "FAILED", ro_vim_item_update
726 class VimInteractionSdnNet(VimInteractionBase
):
728 def _match_pci(port_pci
, mapping
):
730 Check if port_pci matches with mapping
731 mapping can have brackets to indicate that several chars are accepted. e.g
732 pci '0000:af:10.1' matches with '0000:af:1[01].[1357]'
733 :param port_pci: text
734 :param mapping: text, can contain brackets to indicate several chars are available
735 :return: True if matches, False otherwise
737 if not port_pci
or not mapping
:
739 if port_pci
== mapping
:
745 bracket_start
= mapping
.find("[", mapping_index
)
747 if bracket_start
== -1:
750 bracket_end
= mapping
.find("]", bracket_start
)
751 if bracket_end
== -1:
754 length
= bracket_start
- mapping_index
757 and port_pci
[pci_index
: pci_index
+ length
]
758 != mapping
[mapping_index
:bracket_start
]
763 port_pci
[pci_index
+ length
]
764 not in mapping
[bracket_start
+ 1 : bracket_end
]
768 pci_index
+= length
+ 1
769 mapping_index
= bracket_end
+ 1
771 if port_pci
[pci_index
:] != mapping
[mapping_index
:]:
776 def _get_interfaces(self
, vlds_to_connect
, vim_account_id
):
778 :param vlds_to_connect: list with format vnfrs:<id>:vld.<vld_id> or nsrs:<id>:vld.<vld_id>
779 :param vim_account_id:
784 for vld
in vlds_to_connect
:
785 table
, _
, db_id
= vld
.partition(":")
786 db_id
, _
, vld
= db_id
.partition(":")
787 _
, _
, vld_id
= vld
.partition(".")
790 q_filter
= {"vim-account-id": vim_account_id
, "_id": db_id
}
791 iface_key
= "vnf-vld-id"
792 else: # table == "nsrs"
793 q_filter
= {"vim-account-id": vim_account_id
, "nsr-id-ref": db_id
}
794 iface_key
= "ns-vld-id"
796 db_vnfrs
= self
.db
.get_list("vnfrs", q_filter
=q_filter
)
798 for db_vnfr
in db_vnfrs
:
799 for vdu_index
, vdur
in enumerate(db_vnfr
.get("vdur", ())):
800 for iface_index
, interface
in enumerate(vdur
["interfaces"]):
801 if interface
.get(iface_key
) == vld_id
and interface
.get(
803 ) in ("SR-IOV", "PCI-PASSTHROUGH"):
805 interface_
= interface
.copy()
806 interface_
["id"] = "vnfrs:{}:vdu.{}.interfaces.{}".format(
807 db_vnfr
["_id"], vdu_index
, iface_index
810 if vdur
.get("status") == "ERROR":
811 interface_
["status"] = "ERROR"
813 interfaces
.append(interface_
)
817 def refresh(self
, ro_task
):
818 # look for task create
819 task_create_index
, _
= next(
821 for i_t
in enumerate(ro_task
["tasks"])
823 and i_t
[1]["action"] == "CREATE"
824 and i_t
[1]["status"] != "FINISHED"
827 return self
.new(ro_task
, task_create_index
, None)
829 def new(self
, ro_task
, task_index
, task_depends
):
831 task
= ro_task
["tasks"][task_index
]
832 task_id
= task
["task_id"]
833 target_vim
= self
.my_vims
[ro_task
["target_id"]]
835 sdn_net_id
= ro_task
["vim_info"]["vim_id"]
837 created_items
= ro_task
["vim_info"].get("created_items")
838 connected_ports
= ro_task
["vim_info"].get("connected_ports", [])
839 new_connected_ports
= []
840 last_update
= ro_task
["vim_info"].get("last_update", 0)
841 sdn_status
= ro_task
["vim_info"].get("vim_status", "BUILD") or "BUILD"
843 created
= ro_task
["vim_info"].get("created", False)
847 params
= task
["params"]
848 vlds_to_connect
= params
["vlds"]
849 associated_vim
= params
["target_vim"]
850 # external additional ports
851 additional_ports
= params
.get("sdn-ports") or ()
852 _
, _
, vim_account_id
= associated_vim
.partition(":")
856 if associated_vim
not in self
.db_vims
:
857 self
.db_vims
[associated_vim
] = self
.db
.get_one(
858 "vim_accounts", {"_id": vim_account_id
}
861 db_vim
= self
.db_vims
[associated_vim
]
863 # look for ports to connect
864 ports
= self
._get
_interfaces
(vlds_to_connect
, vim_account_id
)
868 pending_ports
= error_ports
= 0
870 sdn_need_update
= False
873 vlan_used
= port
.get("vlan") or vlan_used
875 # TODO. Do not connect if already done
876 if not port
.get("compute_node") or not port
.get("pci"):
877 if port
.get("status") == "ERROR":
884 compute_node_mappings
= next(
887 for c
in db_vim
["config"].get("sdn-port-mapping", ())
888 if c
and c
["compute_node"] == port
["compute_node"]
893 if compute_node_mappings
:
894 # process port_mapping pci of type 0000:af:1[01].[1357]
898 for p
in compute_node_mappings
["ports"]
899 if self
._match
_pci
(port
["pci"], p
.get("pci"))
905 if not db_vim
["config"].get("mapping_not_needed"):
907 "Port mapping not found for compute_node={} pci={}".format(
908 port
["compute_node"], port
["pci"]
915 service_endpoint_id
= "{}:{}".format(port
["compute_node"], port
["pci"])
917 "service_endpoint_id": pmap
.get("service_endpoint_id")
918 or service_endpoint_id
,
919 "service_endpoint_encapsulation_type": "dot1q"
920 if port
["type"] == "SR-IOV"
922 "service_endpoint_encapsulation_info": {
923 "vlan": port
.get("vlan"),
924 "mac": port
.get("mac-address"),
925 "device_id": pmap
.get("device_id") or port
["compute_node"],
926 "device_interface_id": pmap
.get("device_interface_id")
928 "switch_dpid": pmap
.get("switch_id") or pmap
.get("switch_dpid"),
929 "switch_port": pmap
.get("switch_port"),
930 "service_mapping_info": pmap
.get("service_mapping_info"),
935 # if port["modified_at"] > last_update:
936 # sdn_need_update = True
937 new_connected_ports
.append(port
["id"]) # TODO
938 sdn_ports
.append(new_port
)
942 "{} interfaces have not been created as VDU is on ERROR status".format(
947 # connect external ports
948 for index
, additional_port
in enumerate(additional_ports
):
949 additional_port_id
= additional_port
.get(
950 "service_endpoint_id"
951 ) or "external-{}".format(index
)
954 "service_endpoint_id": additional_port_id
,
955 "service_endpoint_encapsulation_type": additional_port
.get(
956 "service_endpoint_encapsulation_type", "dot1q"
958 "service_endpoint_encapsulation_info": {
959 "vlan": additional_port
.get("vlan") or vlan_used
,
960 "mac": additional_port
.get("mac_address"),
961 "device_id": additional_port
.get("device_id"),
962 "device_interface_id": additional_port
.get(
963 "device_interface_id"
965 "switch_dpid": additional_port
.get("switch_dpid")
966 or additional_port
.get("switch_id"),
967 "switch_port": additional_port
.get("switch_port"),
968 "service_mapping_info": additional_port
.get(
969 "service_mapping_info"
974 new_connected_ports
.append(additional_port_id
)
977 # if there are more ports to connect or they have been modified, call create/update
980 sdn_info
= "; ".join(error_list
)
981 elif set(connected_ports
) != set(new_connected_ports
) or sdn_need_update
:
982 last_update
= time
.time()
985 if len(sdn_ports
) < 2:
986 sdn_status
= "ACTIVE"
988 if not pending_ports
:
990 "task={} {} new-sdn-net done, less than 2 ports".format(
991 task_id
, ro_task
["target_id"]
995 net_type
= params
.get("type") or "ELAN"
999 ) = target_vim
.create_connectivity_service(net_type
, sdn_ports
)
1002 "task={} {} new-sdn-net={} created={}".format(
1003 task_id
, ro_task
["target_id"], sdn_net_id
, created
1007 created_items
= target_vim
.edit_connectivity_service(
1008 sdn_net_id
, conn_info
=created_items
, connection_points
=sdn_ports
1012 "task={} {} update-sdn-net={} created={}".format(
1013 task_id
, ro_task
["target_id"], sdn_net_id
, created
1017 connected_ports
= new_connected_ports
1019 wim_status_dict
= target_vim
.get_connectivity_service_status(
1020 sdn_net_id
, conn_info
=created_items
1022 sdn_status
= wim_status_dict
["sdn_status"]
1024 if wim_status_dict
.get("sdn_info"):
1025 sdn_info
= str(wim_status_dict
.get("sdn_info")) or ""
1027 if wim_status_dict
.get("error_msg"):
1028 sdn_info
= wim_status_dict
.get("error_msg") or ""
1031 if sdn_status
!= "ERROR":
1032 sdn_info
= "Waiting for getting interfaces location from VIM. Obtained '{}' of {}".format(
1033 len(ports
) - pending_ports
, len(ports
)
1036 if sdn_status
== "ACTIVE":
1037 sdn_status
= "BUILD"
1039 ro_vim_item_update
= {
1040 "vim_id": sdn_net_id
,
1041 "vim_status": sdn_status
,
1043 "created_items": created_items
,
1044 "connected_ports": connected_ports
,
1045 "vim_details": sdn_info
,
1046 "last_update": last_update
,
1049 return sdn_status
, ro_vim_item_update
1050 except Exception as e
:
1052 "task={} vim={} new-net: {}".format(task_id
, ro_task
["target_id"], e
),
1053 exc_info
=not isinstance(
1054 e
, (sdnconn
.SdnConnectorError
, vimconn
.VimConnException
)
1057 ro_vim_item_update
= {
1058 "vim_status": "VIM_ERROR",
1060 "vim_details": str(e
),
1063 return "FAILED", ro_vim_item_update
1065 def delete(self
, ro_task
, task_index
):
1066 task
= ro_task
["tasks"][task_index
]
1067 task_id
= task
["task_id"]
1068 sdn_vim_id
= ro_task
["vim_info"].get("vim_id")
1069 ro_vim_item_update_ok
= {
1070 "vim_status": "DELETED",
1072 "vim_details": "DELETED",
1078 target_vim
= self
.my_vims
[ro_task
["target_id"]]
1079 target_vim
.delete_connectivity_service(
1080 sdn_vim_id
, ro_task
["vim_info"].get("created_items")
1083 except Exception as e
:
1085 isinstance(e
, sdnconn
.SdnConnectorError
)
1086 and e
.http_code
== HTTPStatus
.NOT_FOUND
.value
1088 ro_vim_item_update_ok
["vim_details"] = "already deleted"
1091 "ro_task={} vim={} del-sdn-net={}: {}".format(
1092 ro_task
["_id"], ro_task
["target_id"], sdn_vim_id
, e
1094 exc_info
=not isinstance(
1095 e
, (sdnconn
.SdnConnectorError
, vimconn
.VimConnException
)
1098 ro_vim_item_update
= {
1099 "vim_status": "VIM_ERROR",
1100 "vim_details": "Error while deleting: {}".format(e
),
1103 return "FAILED", ro_vim_item_update
1106 "task={} {} del-sdn-net={} {}".format(
1108 ro_task
["target_id"],
1110 ro_vim_item_update_ok
.get("vim_details", ""),
1114 return "DONE", ro_vim_item_update_ok
1117 class NsWorker(threading
.Thread
):
1118 REFRESH_BUILD
= 5 # 5 seconds
1119 REFRESH_ACTIVE
= 60 # 1 minute
1121 REFRESH_IMAGE
= 3600 * 10
1122 REFRESH_DELETE
= 3600 * 10
1126 def __init__(self
, worker_index
, config
, plugins
, db
):
1129 :param worker_index: thread index
1130 :param config: general configuration of RO, among others the process_id with the docker id where it runs
1131 :param plugins: global shared dict with the loaded plugins
1132 :param db: database class instance to use
1134 threading
.Thread
.__init
__(self
)
1135 self
.config
= config
1136 self
.plugins
= plugins
1137 self
.plugin_name
= "unknown"
1138 self
.logger
= logging
.getLogger("ro.worker{}".format(worker_index
))
1139 self
.worker_index
= worker_index
1140 self
.task_queue
= queue
.Queue(self
.QUEUE_SIZE
)
1141 # targetvim: vimplugin class
1143 # targetvim: vim information from database
1146 self
.vim_targets
= []
1147 self
.my_id
= config
["process_id"] + ":" + str(worker_index
)
1150 "net": VimInteractionNet(self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
),
1151 "vdu": VimInteractionVdu(self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
),
1152 "image": VimInteractionImage(
1153 self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
1155 "flavor": VimInteractionFlavor(
1156 self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
1158 "sdn_net": VimInteractionSdnNet(
1159 self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
1162 self
.time_last_task_processed
= None
1163 # lists of tasks to delete because nsrs or vnfrs has been deleted from db
1164 self
.tasks_to_delete
= []
1165 # it is idle when there are not vim_targets associated
1167 self
.task_locked_time
= config
["global"]["task_locked_time"]
1169 def insert_task(self
, task
):
1171 self
.task_queue
.put(task
, False)
1174 raise NsWorkerException("timeout inserting a task")
1176 def terminate(self
):
1177 self
.insert_task("exit")
1179 def del_task(self
, task
):
1180 with self
.task_lock
:
1181 if task
["status"] == "SCHEDULED":
1182 task
["status"] = "SUPERSEDED"
1184 else: # task["status"] == "processing"
1185 self
.task_lock
.release()
1188 def _process_vim_config(self
, target_id
, db_vim
):
1190 Process vim config, creating vim configuration files as ca_cert
1191 :param target_id: vim/sdn/wim + id
1192 :param db_vim: Vim dictionary obtained from database
1193 :return: None. Modifies vim. Creates a folder target_id:worker_index and several files
1195 if not db_vim
.get("config"):
1201 if db_vim
["config"].get("ca_cert_content"):
1202 file_name
= "{}:{}".format(target_id
, self
.worker_index
)
1206 except FileExistsError
:
1209 file_name
= file_name
+ "/ca_cert"
1211 with
open(file_name
, "w") as f
:
1212 f
.write(db_vim
["config"]["ca_cert_content"])
1213 del db_vim
["config"]["ca_cert_content"]
1214 db_vim
["config"]["ca_cert"] = file_name
1215 except Exception as e
:
1216 raise NsWorkerException(
1217 "Error writing to file '{}': {}".format(file_name
, e
)
1220 def _load_plugin(self
, name
, type="vim"):
1221 # type can be vim or sdn
1222 if "rovim_dummy" not in self
.plugins
:
1223 self
.plugins
["rovim_dummy"] = VimDummyConnector
1225 if "rosdn_dummy" not in self
.plugins
:
1226 self
.plugins
["rosdn_dummy"] = SdnDummyConnector
1228 if name
in self
.plugins
:
1229 return self
.plugins
[name
]
1232 for v
in iter_entry_points("osm_ro{}.plugins".format(type), name
):
1233 self
.plugins
[name
] = v
.load()
1234 except Exception as e
:
1235 raise NsWorkerException("Cannot load plugin osm_{}: {}".format(name
, e
))
1237 if name
and name
not in self
.plugins
:
1238 raise NsWorkerException(
1239 "Plugin 'osm_{n}' has not been installed".format(n
=name
)
1242 return self
.plugins
[name
]
1244 def _unload_vim(self
, target_id
):
1246 Unload a vim_account. Removes it from self db_vims dictionary, my_vims dictionary and vim_targets list
1247 :param target_id: Contains type:_id; where type can be 'vim', ...
1251 self
.db_vims
.pop(target_id
, None)
1252 self
.my_vims
.pop(target_id
, None)
1254 if target_id
in self
.vim_targets
:
1255 self
.vim_targets
.remove(target_id
)
1257 self
.logger
.info("Unloaded {}".format(target_id
))
1258 rmtree("{}:{}".format(target_id
, self
.worker_index
))
1259 except FileNotFoundError
:
1260 pass # this is raised by rmtree if folder does not exist
1261 except Exception as e
:
1262 self
.logger
.error("Cannot unload {}: {}".format(target_id
, e
))
1264 def _check_vim(self
, target_id
):
1266 Load a VIM/SDN/WIM (if not loaded) and check connectivity, updating database with ENABLE or ERROR
1267 :param target_id: Contains type:_id; type can be 'vim', 'sdn' or 'wim'
1270 target
, _
, _id
= target_id
.partition(":")
1276 loaded
= target_id
in self
.vim_targets
1286 step
= "Getting {} from db".format(target_id
)
1287 db_vim
= self
.db
.get_one(target_database
, {"_id": _id
})
1289 for op_index
, operation
in enumerate(
1290 db_vim
["_admin"].get("operations", ())
1292 if operation
["operationState"] != "PROCESSING":
1295 locked_at
= operation
.get("locked_at")
1297 if locked_at
is not None and locked_at
>= now
- self
.task_locked_time
:
1298 # some other thread is doing this operation
1302 op_text
= "_admin.operations.{}.".format(op_index
)
1304 if not self
.db
.set_one(
1308 op_text
+ "operationState": "PROCESSING",
1309 op_text
+ "locked_at": locked_at
,
1312 op_text
+ "locked_at": now
,
1313 "admin.current_operation": op_index
,
1315 fail_on_empty
=False,
1319 unset_dict
[op_text
+ "locked_at"] = None
1320 unset_dict
["current_operation"] = None
1321 step
= "Loading " + target_id
1322 error_text
= self
._load
_vim
(target_id
)
1325 step
= "Checking connectivity"
1328 self
.my_vims
[target_id
].check_vim_connectivity()
1330 self
.my_vims
[target_id
].check_credentials()
1332 update_dict
["_admin.operationalState"] = "ENABLED"
1333 update_dict
["_admin.detailed-status"] = ""
1334 unset_dict
[op_text
+ "detailed-status"] = None
1335 update_dict
[op_text
+ "operationState"] = "COMPLETED"
1339 except Exception as e
:
1340 error_text
= "{}: {}".format(step
, e
)
1341 self
.logger
.error("{} for {}: {}".format(step
, target_id
, e
))
1344 if update_dict
or unset_dict
:
1346 update_dict
[op_text
+ "operationState"] = "FAILED"
1347 update_dict
[op_text
+ "detailed-status"] = error_text
1348 unset_dict
.pop(op_text
+ "detailed-status", None)
1349 update_dict
["_admin.operationalState"] = "ERROR"
1350 update_dict
["_admin.detailed-status"] = error_text
1353 update_dict
[op_text
+ "statusEnteredTime"] = now
1357 q_filter
={"_id": _id
},
1358 update_dict
=update_dict
,
1360 fail_on_empty
=False,
1364 self
._unload
_vim
(target_id
)
1366 def _reload_vim(self
, target_id
):
1367 if target_id
in self
.vim_targets
:
1368 self
._load
_vim
(target_id
)
1370 # if the vim is not loaded, but database information of VIM is cached at self.db_vims,
1371 # just remove it to force load again next time it is needed
1372 self
.db_vims
.pop(target_id
, None)
1374 def _load_vim(self
, target_id
):
1376 Load or reload a vim_account, sdn_controller or wim_account.
1377 Read content from database, load the plugin if not loaded.
1378 In case of error loading the plugin, it load a failing VIM_connector
1379 It fills self db_vims dictionary, my_vims dictionary and vim_targets list
1380 :param target_id: Contains type:_id; where type can be 'vim', ...
1381 :return: None if ok, descriptive text if error
1383 target
, _
, _id
= target_id
.partition(":")
1395 step
= "Getting {}={} from db".format(target
, _id
)
1396 # TODO process for wim, sdnc, ...
1397 vim
= self
.db
.get_one(target_database
, {"_id": _id
})
1399 # if deep_get(vim, "config", "sdn-controller"):
1400 # step = "Getting sdn-controller-id='{}' from db".format(vim["config"]["sdn-controller"])
1401 # db_sdn = self.db.get_one("sdns", {"_id": vim["config"]["sdn-controller"]})
1403 step
= "Decrypting password"
1404 schema_version
= vim
.get("schema_version")
1405 self
.db
.encrypt_decrypt_fields(
1408 fields
=("password", "secret"),
1409 schema_version
=schema_version
,
1412 self
._process
_vim
_config
(target_id
, vim
)
1415 plugin_name
= "rovim_" + vim
["vim_type"]
1416 step
= "Loading plugin '{}'".format(plugin_name
)
1417 vim_module_conn
= self
._load
_plugin
(plugin_name
)
1418 step
= "Loading {}'".format(target_id
)
1419 self
.my_vims
[target_id
] = vim_module_conn(
1422 tenant_id
=vim
.get("vim_tenant_id"),
1423 tenant_name
=vim
.get("vim_tenant_name"),
1426 user
=vim
["vim_user"],
1427 passwd
=vim
["vim_password"],
1428 config
=vim
.get("config") or {},
1432 plugin_name
= "rosdn_" + vim
["type"]
1433 step
= "Loading plugin '{}'".format(plugin_name
)
1434 vim_module_conn
= self
._load
_plugin
(plugin_name
, "sdn")
1435 step
= "Loading {}'".format(target_id
)
1437 wim_config
= wim
.pop("config", {}) or {}
1438 wim
["uuid"] = wim
["_id"]
1439 wim
["wim_url"] = wim
["url"]
1442 wim_config
["dpid"] = wim
.pop("dpid")
1444 if wim
.get("switch_id"):
1445 wim_config
["switch_id"] = wim
.pop("switch_id")
1447 # wim, wim_account, config
1448 self
.my_vims
[target_id
] = vim_module_conn(wim
, wim
, wim_config
)
1449 self
.db_vims
[target_id
] = vim
1450 self
.error_status
= None
1453 "Connector loaded for {}, plugin={}".format(target_id
, plugin_name
)
1455 except Exception as e
:
1457 "Cannot load {} plugin={}: {} {}".format(
1458 target_id
, plugin_name
, step
, e
1462 self
.db_vims
[target_id
] = vim
or {}
1463 self
.db_vims
[target_id
] = FailingConnector(str(e
))
1464 error_status
= "{} Error: {}".format(step
, e
)
1468 if target_id
not in self
.vim_targets
:
1469 self
.vim_targets
.append(target_id
)
1471 def _get_db_task(self
):
1473 Read actions from database and reload them at memory. Fill self.refresh_list, pending_list, vim_actions
1478 if not self
.time_last_task_processed
:
1479 self
.time_last_task_processed
= now
1483 locked
= self
.db
.set_one(
1486 "target_id": self
.vim_targets
,
1487 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
1488 "locked_at.lt": now
- self
.task_locked_time
,
1489 "to_check_at.lt": self
.time_last_task_processed
,
1491 update_dict
={"locked_by": self
.my_id
, "locked_at": now
},
1492 fail_on_empty
=False,
1497 ro_task
= self
.db
.get_one(
1500 "target_id": self
.vim_targets
,
1501 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
1507 if self
.time_last_task_processed
== now
:
1508 self
.time_last_task_processed
= None
1511 self
.time_last_task_processed
= now
1512 # self.time_last_task_processed = min(self.time_last_task_processed + 1000, now)
1514 except DbException
as e
:
1515 self
.logger
.error("Database exception at _get_db_task: {}".format(e
))
1516 except Exception as e
:
1517 self
.logger
.critical(
1518 "Unexpected exception at _get_db_task: {}".format(e
), exc_info
=True
1523 def _delete_task(self
, ro_task
, task_index
, task_depends
, db_update
):
1525 Determine if this task need to be done or superseded
1528 my_task
= ro_task
["tasks"][task_index
]
1529 task_id
= my_task
["task_id"]
1530 needed_delete
= ro_task
["vim_info"]["created"] or ro_task
["vim_info"].get(
1531 "created_items", False
1534 if my_task
["status"] == "FAILED":
1535 return None, None # TODO need to be retry??
1538 for index
, task
in enumerate(ro_task
["tasks"]):
1539 if index
== task_index
or not task
:
1543 my_task
["target_record"] == task
["target_record"]
1544 and task
["action"] == "CREATE"
1547 db_update
["tasks.{}.status".format(index
)] = task
[
1550 elif task
["action"] == "CREATE" and task
["status"] not in (
1554 needed_delete
= False
1557 return self
.item2class
[my_task
["item"]].delete(ro_task
, task_index
)
1559 return "SUPERSEDED", None
1560 except Exception as e
:
1561 if not isinstance(e
, NsWorkerException
):
1562 self
.logger
.critical(
1563 "Unexpected exception at _delete_task task={}: {}".format(
1569 return "FAILED", {"vim_status": "VIM_ERROR", "vim_details": str(e
)}
1571 def _create_task(self
, ro_task
, task_index
, task_depends
, db_update
):
1573 Determine if this task need to create something at VIM
1576 my_task
= ro_task
["tasks"][task_index
]
1577 task_id
= my_task
["task_id"]
1580 if my_task
["status"] == "FAILED":
1581 return None, None # TODO need to be retry??
1582 elif my_task
["status"] == "SCHEDULED":
1583 # check if already created by another task
1584 for index
, task
in enumerate(ro_task
["tasks"]):
1585 if index
== task_index
or not task
:
1588 if task
["action"] == "CREATE" and task
["status"] not in (
1593 return task
["status"], "COPY_VIM_INFO"
1596 task_status
, ro_vim_item_update
= self
.item2class
[my_task
["item"]].new(
1597 ro_task
, task_index
, task_depends
1599 # TODO update other CREATE tasks
1600 except Exception as e
:
1601 if not isinstance(e
, NsWorkerException
):
1603 "Error executing task={}: {}".format(task_id
, e
), exc_info
=True
1606 task_status
= "FAILED"
1607 ro_vim_item_update
= {"vim_status": "VIM_ERROR", "vim_details": str(e
)}
1608 # TODO update ro_vim_item_update
1610 return task_status
, ro_vim_item_update
1614 def _get_dependency(self
, task_id
, ro_task
=None, target_id
=None):
1616 Look for dependency task
1617 :param task_id: Can be one of
1618 1. target_vim+blank+task.target_record_id: "(vim|sdn|wim):<id> (vnfrs|nsrs):(vld|vdu|flavor|image).<id>"
1619 2. task.target_record_id: "(vnfrs|nsrs):(vld|vdu|flavor|image).<id>"
1620 3. task.task_id: "<action_id>:number"
1623 :return: database ro_task plus index of task
1626 task_id
.startswith("vim:")
1627 or task_id
.startswith("sdn:")
1628 or task_id
.startswith("wim:")
1630 target_id
, _
, task_id
= task_id
.partition(" ")
1632 if task_id
.startswith("nsrs:") or task_id
.startswith("vnfrs:"):
1633 ro_task_dependency
= self
.db
.get_one(
1635 q_filter
={"target_id": target_id
, "tasks.target_record_id": task_id
},
1636 fail_on_empty
=False,
1639 if ro_task_dependency
:
1640 for task_index
, task
in enumerate(ro_task_dependency
["tasks"]):
1641 if task
["target_record_id"] == task_id
:
1642 return ro_task_dependency
, task_index
1646 for task_index
, task
in enumerate(ro_task
["tasks"]):
1647 if task
and task
["task_id"] == task_id
:
1648 return ro_task
, task_index
1650 ro_task_dependency
= self
.db
.get_one(
1653 "tasks.ANYINDEX.task_id": task_id
,
1654 "tasks.ANYINDEX.target_record.ne": None,
1656 fail_on_empty
=False,
1659 if ro_task_dependency
:
1660 for task_index
, task
in ro_task_dependency
["tasks"]:
1661 if task
["task_id"] == task_id
:
1662 return ro_task_dependency
, task_index
1663 raise NsWorkerException("Cannot get depending task {}".format(task_id
))
1665 def _process_pending_tasks(self
, ro_task
):
1666 ro_task_id
= ro_task
["_id"]
1669 next_check_at
= now
+ (24 * 60 * 60)
1670 db_ro_task_update
= {}
1672 def _update_refresh(new_status
):
1673 # compute next_refresh
1675 nonlocal next_check_at
1676 nonlocal db_ro_task_update
1679 next_refresh
= time
.time()
1681 if task
["item"] in ("image", "flavor"):
1682 next_refresh
+= self
.REFRESH_IMAGE
1683 elif new_status
== "BUILD":
1684 next_refresh
+= self
.REFRESH_BUILD
1685 elif new_status
== "DONE":
1686 next_refresh
+= self
.REFRESH_ACTIVE
1688 next_refresh
+= self
.REFRESH_ERROR
1690 next_check_at
= min(next_check_at
, next_refresh
)
1691 db_ro_task_update
["vim_info.refresh_at"] = next_refresh
1692 ro_task
["vim_info"]["refresh_at"] = next_refresh
1695 # 0: get task_status_create
1697 task_status_create
= None
1701 for t
in ro_task
["tasks"]
1703 and t
["action"] == "CREATE"
1704 and t
["status"] in ("BUILD", "DONE")
1710 task_status_create
= task_create
["status"]
1712 # 1: look for tasks in status SCHEDULED, or in status CREATE if action is DONE or BUILD
1713 for task_action
in ("DELETE", "CREATE", "EXEC"):
1714 db_vim_update
= None
1717 for task_index
, task
in enumerate(ro_task
["tasks"]):
1719 continue # task deleted
1722 target_update
= None
1726 task_action
in ("DELETE", "EXEC")
1727 and task
["status"] not in ("SCHEDULED", "BUILD")
1729 or task
["action"] != task_action
1731 task_action
== "CREATE"
1732 and task
["status"] in ("FINISHED", "SUPERSEDED")
1737 task_path
= "tasks.{}.status".format(task_index
)
1739 db_vim_info_update
= None
1741 if task
["status"] == "SCHEDULED":
1742 # check if tasks that this depends on have been completed
1743 dependency_not_completed
= False
1745 for dependency_task_id
in task
.get("depends_on") or ():
1748 dependency_task_index
,
1749 ) = self
._get
_dependency
(
1750 dependency_task_id
, target_id
=ro_task
["target_id"]
1752 dependency_task
= dependency_ro_task
["tasks"][
1753 dependency_task_index
1756 if dependency_task
["status"] == "SCHEDULED":
1757 dependency_not_completed
= True
1758 next_check_at
= min(
1759 next_check_at
, dependency_ro_task
["to_check_at"]
1761 # must allow dependent task to be processed first
1762 # to do this set time after last_task_processed
1763 next_check_at
= max(
1764 self
.time_last_task_processed
, next_check_at
1767 elif dependency_task
["status"] == "FAILED":
1768 error_text
= "Cannot {} {} because depends on failed {} {} id={}): {}".format(
1771 dependency_task
["action"],
1772 dependency_task
["item"],
1774 dependency_ro_task
["vim_info"].get(
1779 "task={} {}".format(task
["task_id"], error_text
)
1781 raise NsWorkerException(error_text
)
1783 task_depends
[dependency_task_id
] = dependency_ro_task
[
1787 "TASK-{}".format(dependency_task_id
)
1788 ] = dependency_ro_task
["vim_info"]["vim_id"]
1790 if dependency_not_completed
:
1791 # TODO set at vim_info.vim_details that it is waiting
1794 # before calling VIM-plugin as it can take more than task_locked_time, insert to LockRenew
1795 # the task of renew this locking. It will update database locket_at periodically
1797 lock_object
= LockRenew
.add_lock_object(
1798 "ro_tasks", ro_task
, self
1801 if task
["action"] == "DELETE":
1802 (new_status
, db_vim_info_update
,) = self
._delete
_task
(
1803 ro_task
, task_index
, task_depends
, db_ro_task_update
1806 "FINISHED" if new_status
== "DONE" else new_status
1808 # ^with FINISHED instead of DONE it will not be refreshing
1810 if new_status
in ("FINISHED", "SUPERSEDED"):
1811 target_update
= "DELETE"
1812 elif task
["action"] == "EXEC":
1817 ) = self
.item2class
[task
["item"]].exec(
1818 ro_task
, task_index
, task_depends
1821 "FINISHED" if new_status
== "DONE" else new_status
1823 # ^with FINISHED instead of DONE it will not be refreshing
1826 # load into database the modified db_task_update "retries" and "next_retry"
1827 if db_task_update
.get("retries"):
1829 "tasks.{}.retries".format(task_index
)
1830 ] = db_task_update
["retries"]
1832 next_check_at
= time
.time() + db_task_update
.get(
1835 target_update
= None
1836 elif task
["action"] == "CREATE":
1837 if task
["status"] == "SCHEDULED":
1838 if task_status_create
:
1839 new_status
= task_status_create
1840 target_update
= "COPY_VIM_INFO"
1842 new_status
, db_vim_info_update
= self
.item2class
[
1844 ].new(ro_task
, task_index
, task_depends
)
1845 # self._create_task(ro_task, task_index, task_depends, db_ro_task_update)
1846 _update_refresh(new_status
)
1849 ro_task
["vim_info"]["refresh_at"]
1850 and now
> ro_task
["vim_info"]["refresh_at"]
1852 new_status
, db_vim_info_update
= self
.item2class
[
1855 _update_refresh(new_status
)
1857 except Exception as e
:
1858 new_status
= "FAILED"
1859 db_vim_info_update
= {
1860 "vim_status": "VIM_ERROR",
1861 "vim_details": str(e
),
1865 e
, (NsWorkerException
, vimconn
.VimConnException
)
1868 "Unexpected exception at _delete_task task={}: {}".format(
1875 if db_vim_info_update
:
1876 db_vim_update
= db_vim_info_update
.copy()
1877 db_ro_task_update
.update(
1880 for k
, v
in db_vim_info_update
.items()
1883 ro_task
["vim_info"].update(db_vim_info_update
)
1886 if task_action
== "CREATE":
1887 task_status_create
= new_status
1888 db_ro_task_update
[task_path
] = new_status
1890 if target_update
or db_vim_update
:
1891 if target_update
== "DELETE":
1892 self
._update
_target
(task
, None)
1893 elif target_update
== "COPY_VIM_INFO":
1894 self
._update
_target
(task
, ro_task
["vim_info"])
1896 self
._update
_target
(task
, db_vim_update
)
1898 except Exception as e
:
1900 isinstance(e
, DbException
)
1901 and e
.http_code
== HTTPStatus
.NOT_FOUND
1903 # if the vnfrs or nsrs has been removed from database, this task must be removed
1905 "marking to delete task={}".format(task
["task_id"])
1907 self
.tasks_to_delete
.append(task
)
1910 "Unexpected exception at _update_target task={}: {}".format(
1916 locked_at
= ro_task
["locked_at"]
1920 lock_object
["locked_at"],
1921 lock_object
["locked_at"] + self
.task_locked_time
,
1923 # locked_at contains two times to avoid race condition. In case the lock has been renew, it will
1924 # contain exactly locked_at + self.task_locked_time
1925 LockRenew
.remove_lock_object(lock_object
)
1928 "_id": ro_task
["_id"],
1929 "to_check_at": ro_task
["to_check_at"],
1930 "locked_at": locked_at
,
1932 # modify own task. Try filtering by to_next_check. For race condition if to_check_at has been modified,
1933 # outside this task (by ro_nbi) do not update it
1934 db_ro_task_update
["locked_by"] = None
1935 # locked_at converted to int only for debugging. When has not decimals it means it has been unlocked
1936 db_ro_task_update
["locked_at"] = int(now
- self
.task_locked_time
)
1937 db_ro_task_update
["modified_at"] = now
1938 db_ro_task_update
["to_check_at"] = next_check_at
1940 if not self
.db
.set_one(
1942 update_dict
=db_ro_task_update
,
1944 fail_on_empty
=False,
1946 del db_ro_task_update
["to_check_at"]
1947 del q_filter
["to_check_at"]
1951 update_dict
=db_ro_task_update
,
1954 except DbException
as e
:
1956 "ro_task={} Error updating database {}".format(ro_task_id
, e
)
1958 except Exception as e
:
1960 "Error executing ro_task={}: {}".format(ro_task_id
, e
), exc_info
=True
1963 def _update_target(self
, task
, ro_vim_item_update
):
1964 table
, _
, temp
= task
["target_record"].partition(":")
1965 _id
, _
, path_vim_status
= temp
.partition(":")
1966 path_item
= path_vim_status
[: path_vim_status
.rfind(".")]
1967 path_item
= path_item
[: path_item
.rfind(".")]
1968 # path_vim_status: dot separated list targeting vim information, e.g. "vdur.10.vim_info.vim:id"
1969 # path_item: dot separated list targeting record information, e.g. "vdur.10"
1971 if ro_vim_item_update
:
1973 path_vim_status
+ "." + k
: v
1974 for k
, v
in ro_vim_item_update
.items()
1976 in ("vim_id", "vim_details", "vim_name", "vim_status", "interfaces")
1979 if path_vim_status
.startswith("vdur."):
1980 # for backward compatibility, add vdur.name apart from vdur.vim_name
1981 if ro_vim_item_update
.get("vim_name"):
1982 update_dict
[path_item
+ ".name"] = ro_vim_item_update
["vim_name"]
1984 # for backward compatibility, add vdur.vim-id apart from vdur.vim_id
1985 if ro_vim_item_update
.get("vim_id"):
1986 update_dict
[path_item
+ ".vim-id"] = ro_vim_item_update
["vim_id"]
1988 # update general status
1989 if ro_vim_item_update
.get("vim_status"):
1990 update_dict
[path_item
+ ".status"] = ro_vim_item_update
[
1994 if ro_vim_item_update
.get("interfaces"):
1995 path_interfaces
= path_item
+ ".interfaces"
1997 for i
, iface
in enumerate(ro_vim_item_update
.get("interfaces")):
2001 path_interfaces
+ ".{}.".format(i
) + k
: v
2002 for k
, v
in iface
.items()
2003 if k
in ("vlan", "compute_node", "pci")
2007 # put ip_address and mac_address with ip-address and mac-address
2008 if iface
.get("ip_address"):
2010 path_interfaces
+ ".{}.".format(i
) + "ip-address"
2011 ] = iface
["ip_address"]
2013 if iface
.get("mac_address"):
2015 path_interfaces
+ ".{}.".format(i
) + "mac-address"
2016 ] = iface
["mac_address"]
2018 if iface
.get("mgmt_vnf_interface") and iface
.get("ip_address"):
2019 update_dict
["ip-address"] = iface
.get("ip_address").split(
2023 if iface
.get("mgmt_vdu_interface") and iface
.get("ip_address"):
2024 update_dict
[path_item
+ ".ip-address"] = iface
.get(
2028 self
.db
.set_one(table
, q_filter
={"_id": _id
}, update_dict
=update_dict
)
2030 update_dict
= {path_item
+ ".status": "DELETED"}
2033 q_filter
={"_id": _id
},
2034 update_dict
=update_dict
,
2035 unset
={path_vim_status
: None},
2038 def _process_delete_db_tasks(self
):
2040 Delete task from database because vnfrs or nsrs or both have been deleted
2041 :return: None. Uses and modify self.tasks_to_delete
2043 while self
.tasks_to_delete
:
2044 task
= self
.tasks_to_delete
[0]
2045 vnfrs_deleted
= None
2046 nsr_id
= task
["nsr_id"]
2048 if task
["target_record"].startswith("vnfrs:"):
2049 # check if nsrs is present
2050 if self
.db
.get_one("nsrs", {"_id": nsr_id
}, fail_on_empty
=False):
2051 vnfrs_deleted
= task
["target_record"].split(":")[1]
2054 self
.delete_db_tasks(self
.db
, nsr_id
, vnfrs_deleted
)
2055 except Exception as e
:
2057 "Error deleting task={}: {}".format(task
["task_id"], e
)
2059 self
.tasks_to_delete
.pop(0)
2062 def delete_db_tasks(db
, nsr_id
, vnfrs_deleted
):
2064 Static method because it is called from osm_ng_ro.ns
2065 :param db: instance of database to use
2066 :param nsr_id: affected nsrs id
2067 :param vnfrs_deleted: only tasks with this vnfr id. If None, all affected by nsr_id
2068 :return: None, exception is fails
2071 for retry
in range(retries
):
2072 ro_tasks
= db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2076 for ro_task
in ro_tasks
:
2078 to_delete_ro_task
= True
2080 for index
, task
in enumerate(ro_task
["tasks"]):
2083 elif (not vnfrs_deleted
and task
["nsr_id"] == nsr_id
) or (
2085 and task
["target_record"].startswith("vnfrs:" + vnfrs_deleted
)
2087 db_update
["tasks.{}".format(index
)] = None
2089 # used by other nsr, ro_task cannot be deleted
2090 to_delete_ro_task
= False
2092 # delete or update if nobody has changed ro_task meanwhile. Used modified_at for known if changed
2093 if to_delete_ro_task
:
2097 "_id": ro_task
["_id"],
2098 "modified_at": ro_task
["modified_at"],
2100 fail_on_empty
=False,
2104 db_update
["modified_at"] = now
2108 "_id": ro_task
["_id"],
2109 "modified_at": ro_task
["modified_at"],
2111 update_dict
=db_update
,
2112 fail_on_empty
=False,
2118 raise NsWorkerException("Exceeded {} retries".format(retries
))
2122 self
.logger
.info("Starting")
2124 # step 1: get commands from queue
2126 if self
.vim_targets
:
2127 task
= self
.task_queue
.get(block
=False)
2130 self
.logger
.debug("enters in idle state")
2132 task
= self
.task_queue
.get(block
=True)
2135 if task
[0] == "terminate":
2137 elif task
[0] == "load_vim":
2138 self
.logger
.info("order to load vim {}".format(task
[1]))
2139 self
._load
_vim
(task
[1])
2140 elif task
[0] == "unload_vim":
2141 self
.logger
.info("order to unload vim {}".format(task
[1]))
2142 self
._unload
_vim
(task
[1])
2143 elif task
[0] == "reload_vim":
2144 self
._reload
_vim
(task
[1])
2145 elif task
[0] == "check_vim":
2146 self
.logger
.info("order to check vim {}".format(task
[1]))
2147 self
._check
_vim
(task
[1])
2149 except Exception as e
:
2150 if isinstance(e
, queue
.Empty
):
2153 self
.logger
.critical(
2154 "Error processing task: {}".format(e
), exc_info
=True
2157 # step 2: process pending_tasks, delete not needed tasks
2159 if self
.tasks_to_delete
:
2160 self
._process
_delete
_db
_tasks
()
2162 ro_task
= self
._get
_db
_task
()
2164 self
._process
_pending
_tasks
(ro_task
)
2168 except Exception as e
:
2169 self
.logger
.critical(
2170 "Unexpected exception at run: " + str(e
), exc_info
=True
2173 self
.logger
.info("Finishing")