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
27 from copy
import deepcopy
28 from http
import HTTPStatus
32 from shutil
import rmtree
35 from unittest
.mock
import Mock
37 from importlib_metadata
import entry_points
38 from osm_common
.dbbase
import DbException
39 from osm_ng_ro
.vim_admin
import LockRenew
40 from osm_ro_plugin
import sdnconn
, vimconn
41 from osm_ro_plugin
.sdn_dummy
import SdnDummyConnector
42 from osm_ro_plugin
.vim_dummy
import VimDummyConnector
45 __author__
= "Alfonso Tierno"
46 __date__
= "$28-Sep-2017 12:07:15$"
49 def deep_get(target_dict
, *args
, **kwargs
):
51 Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
52 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
53 :param target_dict: dictionary to be read
54 :param args: list of keys to read from target_dict
55 :param kwargs: only can contain default=value to return if key is not present in the nested dictionary
56 :return: The wanted value if exist, None or default otherwise
59 if not isinstance(target_dict
, dict) or key
not in target_dict
:
60 return kwargs
.get("default")
61 target_dict
= target_dict
[key
]
65 class NsWorkerException(Exception):
69 class FailingConnector
:
70 def __init__(self
, error_msg
):
71 self
.error_msg
= error_msg
73 for method
in dir(vimconn
.VimConnector
):
76 self
, method
, Mock(side_effect
=vimconn
.VimConnException(error_msg
))
79 for method
in dir(sdnconn
.SdnConnectorBase
):
82 self
, method
, Mock(side_effect
=sdnconn
.SdnConnectorError(error_msg
))
86 class NsWorkerExceptionNotFound(NsWorkerException
):
90 class VimInteractionBase
:
91 """Base class to call VIM/SDN for creating, deleting and refresh networks, VMs, flavors, ...
92 It implements methods that does nothing and return ok"""
94 def __init__(self
, db
, my_vims
, db_vims
, logger
):
97 self
.my_vims
= my_vims
98 self
.db_vims
= db_vims
100 def new(self
, ro_task
, task_index
, task_depends
):
103 def refresh(self
, ro_task
):
104 """skip calling VIM to get image, flavor status. Assumes ok"""
105 if ro_task
["vim_info"]["vim_status"] == "VIM_ERROR":
110 def delete(self
, ro_task
, task_index
):
111 """skip calling VIM to delete image. Assumes ok"""
114 def exec(self
, ro_task
, task_index
, task_depends
):
115 return "DONE", None, None
118 class VimInteractionNet(VimInteractionBase
):
119 def new(self
, ro_task
, task_index
, task_depends
):
121 task
= ro_task
["tasks"][task_index
]
122 task_id
= task
["task_id"]
125 target_vim
= self
.my_vims
[ro_task
["target_id"]]
127 mgmtnet_defined_in_vim
= False
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"):
139 self
.db_vims
[ro_task
["target_id"]],
141 "management_network_id",
143 mgmtnet_defined_in_vim
= True
145 "id": self
.db_vims
[ro_task
["target_id"]]["config"][
146 "management_network_id"
150 self
.db_vims
[ro_task
["target_id"]],
152 "management_network_name",
154 mgmtnet_defined_in_vim
= True
156 "name": self
.db_vims
[ro_task
["target_id"]]["config"][
157 "management_network_name"
161 vim_filter
= {"name": task
["find_params"]["name"]}
163 raise NsWorkerExceptionNotFound(
164 "Invalid find_params for new_net {}".format(task
["find_params"])
167 vim_nets
= target_vim
.get_network_list(vim_filter
)
168 if not vim_nets
and not task
.get("params"):
169 # If there is mgmt-network in the descriptor,
170 # there is no mapping of that network to a VIM network in the descriptor,
171 # also there is no mapping in the "--config" parameter or at VIM creation;
172 # that mgmt-network will be created.
173 if mgmtnet
and not mgmtnet_defined_in_vim
:
175 vim_filter
.get("name")
176 if vim_filter
.get("name")
177 else vim_filter
.get("id")[:16]
179 vim_net_id
, created_items
= target_vim
.new_network(
183 "Created mgmt network vim_net_id: {}".format(vim_net_id
)
187 raise NsWorkerExceptionNotFound(
188 "Network not found with this criteria: '{}'".format(
189 task
.get("find_params")
192 elif len(vim_nets
) > 1:
193 raise NsWorkerException(
194 "More than one network found with this criteria: '{}'".format(
200 vim_net_id
= vim_nets
[0]["id"]
203 params
= task
["params"]
204 vim_net_id
, created_items
= target_vim
.new_network(**params
)
207 ro_vim_item_update
= {
208 "vim_id": vim_net_id
,
209 "vim_status": "BUILD",
211 "created_items": created_items
,
215 "task={} {} new-net={} created={}".format(
216 task_id
, ro_task
["target_id"], vim_net_id
, created
220 return "BUILD", ro_vim_item_update
221 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
223 "task={} vim={} new-net: {}".format(task_id
, ro_task
["target_id"], e
)
225 ro_vim_item_update
= {
226 "vim_status": "VIM_ERROR",
228 "vim_details": str(e
),
231 return "FAILED", ro_vim_item_update
233 def refresh(self
, ro_task
):
234 """Call VIM to get network status"""
235 ro_task_id
= ro_task
["_id"]
236 target_vim
= self
.my_vims
[ro_task
["target_id"]]
237 vim_id
= ro_task
["vim_info"]["vim_id"]
238 net_to_refresh_list
= [vim_id
]
241 vim_dict
= target_vim
.refresh_nets_status(net_to_refresh_list
)
242 vim_info
= vim_dict
[vim_id
]
244 if vim_info
["status"] == "ACTIVE":
246 elif vim_info
["status"] == "BUILD":
247 task_status
= "BUILD"
249 task_status
= "FAILED"
250 except vimconn
.VimConnException
as e
:
251 # Mark all tasks at VIM_ERROR status
253 "ro_task={} vim={} get-net={}: {}".format(
254 ro_task_id
, ro_task
["target_id"], vim_id
, e
257 vim_info
= {"status": "VIM_ERROR", "error_msg": str(e
)}
258 task_status
= "FAILED"
260 ro_vim_item_update
= {}
261 if ro_task
["vim_info"]["vim_status"] != vim_info
["status"]:
262 ro_vim_item_update
["vim_status"] = vim_info
["status"]
264 if ro_task
["vim_info"]["vim_name"] != vim_info
.get("name"):
265 ro_vim_item_update
["vim_name"] = vim_info
.get("name")
267 if vim_info
["status"] in ("ERROR", "VIM_ERROR"):
268 if ro_task
["vim_info"]["vim_details"] != vim_info
.get("error_msg"):
269 ro_vim_item_update
["vim_details"] = vim_info
.get("error_msg")
270 elif vim_info
["status"] == "DELETED":
271 ro_vim_item_update
["vim_id"] = None
272 ro_vim_item_update
["vim_details"] = "Deleted externally"
274 if ro_task
["vim_info"]["vim_details"] != vim_info
["vim_info"]:
275 ro_vim_item_update
["vim_details"] = vim_info
["vim_info"]
277 if ro_vim_item_update
:
279 "ro_task={} {} get-net={}: status={} {}".format(
281 ro_task
["target_id"],
283 ro_vim_item_update
.get("vim_status"),
284 ro_vim_item_update
.get("vim_details")
285 if ro_vim_item_update
.get("vim_status") != "ACTIVE"
290 return task_status
, ro_vim_item_update
292 def delete(self
, ro_task
, task_index
):
293 task
= ro_task
["tasks"][task_index
]
294 task_id
= task
["task_id"]
295 net_vim_id
= ro_task
["vim_info"]["vim_id"]
296 ro_vim_item_update_ok
= {
297 "vim_status": "DELETED",
299 "vim_details": "DELETED",
304 if net_vim_id
or ro_task
["vim_info"]["created_items"]:
305 target_vim
= self
.my_vims
[ro_task
["target_id"]]
306 target_vim
.delete_network(
307 net_vim_id
, ro_task
["vim_info"]["created_items"]
309 except vimconn
.VimConnNotFoundException
:
310 ro_vim_item_update_ok
["vim_details"] = "already deleted"
311 except vimconn
.VimConnException
as e
:
313 "ro_task={} vim={} del-net={}: {}".format(
314 ro_task
["_id"], ro_task
["target_id"], net_vim_id
, e
317 ro_vim_item_update
= {
318 "vim_status": "VIM_ERROR",
319 "vim_details": "Error while deleting: {}".format(e
),
322 return "FAILED", ro_vim_item_update
325 "task={} {} del-net={} {}".format(
327 ro_task
["target_id"],
329 ro_vim_item_update_ok
.get("vim_details", ""),
333 return "DONE", ro_vim_item_update_ok
336 class VimInteractionVdu(VimInteractionBase
):
337 max_retries_inject_ssh_key
= 20 # 20 times
338 time_retries_inject_ssh_key
= 30 # wevery 30 seconds
340 def new(self
, ro_task
, task_index
, task_depends
):
341 task
= ro_task
["tasks"][task_index
]
342 task_id
= task
["task_id"]
345 target_vim
= self
.my_vims
[ro_task
["target_id"]]
349 params
= task
["params"]
350 params_copy
= deepcopy(params
)
351 net_list
= params_copy
["net_list"]
354 # change task_id into network_id
355 if "net_id" in net
and net
["net_id"].startswith("TASK-"):
356 network_id
= task_depends
[net
["net_id"]]
359 raise NsWorkerException(
360 "Cannot create VM because depends on a network not created or found "
361 "for {}".format(net
["net_id"])
364 net
["net_id"] = network_id
366 if params_copy
["image_id"].startswith("TASK-"):
367 params_copy
["image_id"] = task_depends
[params_copy
["image_id"]]
369 if params_copy
["flavor_id"].startswith("TASK-"):
370 params_copy
["flavor_id"] = task_depends
[params_copy
["flavor_id"]]
372 affinity_group_list
= params_copy
["affinity_group_list"]
373 for affinity_group
in affinity_group_list
:
374 # change task_id into affinity_group_id
375 if "affinity_group_id" in affinity_group
and affinity_group
[
377 ].startswith("TASK-"):
378 affinity_group_id
= task_depends
[
379 affinity_group
["affinity_group_id"]
382 if not affinity_group_id
:
383 raise NsWorkerException(
384 "found for {}".format(affinity_group
["affinity_group_id"])
387 affinity_group
["affinity_group_id"] = affinity_group_id
389 vim_vm_id
, created_items
= target_vim
.new_vminstance(**params_copy
)
390 interfaces
= [iface
["vim_id"] for iface
in params_copy
["net_list"]]
392 ro_vim_item_update
= {
394 "vim_status": "BUILD",
396 "created_items": created_items
,
398 "interfaces_vim_ids": interfaces
,
402 "task={} {} new-vm={} created={}".format(
403 task_id
, ro_task
["target_id"], vim_vm_id
, created
407 return "BUILD", ro_vim_item_update
408 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
410 "task={} {} new-vm: {}".format(task_id
, ro_task
["target_id"], e
)
412 ro_vim_item_update
= {
413 "vim_status": "VIM_ERROR",
415 "vim_details": str(e
),
418 return "FAILED", ro_vim_item_update
420 def delete(self
, ro_task
, task_index
):
421 task
= ro_task
["tasks"][task_index
]
422 task_id
= task
["task_id"]
423 vm_vim_id
= ro_task
["vim_info"]["vim_id"]
424 ro_vim_item_update_ok
= {
425 "vim_status": "DELETED",
427 "vim_details": "DELETED",
432 if vm_vim_id
or ro_task
["vim_info"]["created_items"]:
433 target_vim
= self
.my_vims
[ro_task
["target_id"]]
434 target_vim
.delete_vminstance(
435 vm_vim_id
, ro_task
["vim_info"]["created_items"]
437 except vimconn
.VimConnNotFoundException
:
438 ro_vim_item_update_ok
["vim_details"] = "already deleted"
439 except vimconn
.VimConnException
as e
:
441 "ro_task={} vim={} del-vm={}: {}".format(
442 ro_task
["_id"], ro_task
["target_id"], vm_vim_id
, e
445 ro_vim_item_update
= {
446 "vim_status": "VIM_ERROR",
447 "vim_details": "Error while deleting: {}".format(e
),
450 return "FAILED", ro_vim_item_update
453 "task={} {} del-vm={} {}".format(
455 ro_task
["target_id"],
457 ro_vim_item_update_ok
.get("vim_details", ""),
461 return "DONE", ro_vim_item_update_ok
463 def refresh(self
, ro_task
):
464 """Call VIM to get vm status"""
465 ro_task_id
= ro_task
["_id"]
466 target_vim
= self
.my_vims
[ro_task
["target_id"]]
467 vim_id
= ro_task
["vim_info"]["vim_id"]
472 vm_to_refresh_list
= [vim_id
]
474 vim_dict
= target_vim
.refresh_vms_status(vm_to_refresh_list
)
475 vim_info
= vim_dict
[vim_id
]
477 if vim_info
["status"] == "ACTIVE":
479 elif vim_info
["status"] == "BUILD":
480 task_status
= "BUILD"
482 task_status
= "FAILED"
484 # try to load and parse vim_information
486 vim_info_info
= yaml
.safe_load(vim_info
["vim_info"])
487 if vim_info_info
.get("name"):
488 vim_info
["name"] = vim_info_info
["name"]
491 except vimconn
.VimConnException
as e
:
492 # Mark all tasks at VIM_ERROR status
494 "ro_task={} vim={} get-vm={}: {}".format(
495 ro_task_id
, ro_task
["target_id"], vim_id
, e
498 vim_info
= {"status": "VIM_ERROR", "error_msg": str(e
)}
499 task_status
= "FAILED"
501 ro_vim_item_update
= {}
503 # Interfaces cannot be present if e.g. VM is not present, that is status=DELETED
505 if vim_info
.get("interfaces"):
506 for vim_iface_id
in ro_task
["vim_info"]["interfaces_vim_ids"]:
510 for iface
in vim_info
["interfaces"]
511 if vim_iface_id
== iface
["vim_interface_id"]
516 # iface.pop("vim_info", None)
517 vim_interfaces
.append(iface
)
521 for t
in ro_task
["tasks"]
522 if t
and t
["action"] == "CREATE" and t
["status"] != "FINISHED"
524 if vim_interfaces
and task_create
.get("mgmt_vnf_interface") is not None:
525 vim_interfaces
[task_create
["mgmt_vnf_interface"]][
529 mgmt_vdu_iface
= task_create
.get(
530 "mgmt_vdu_interface", task_create
.get("mgmt_vnf_interface", 0)
533 vim_interfaces
[mgmt_vdu_iface
]["mgmt_vdu_interface"] = True
535 if ro_task
["vim_info"]["interfaces"] != vim_interfaces
:
536 ro_vim_item_update
["interfaces"] = vim_interfaces
538 if ro_task
["vim_info"]["vim_status"] != vim_info
["status"]:
539 ro_vim_item_update
["vim_status"] = vim_info
["status"]
541 if ro_task
["vim_info"]["vim_name"] != vim_info
.get("name"):
542 ro_vim_item_update
["vim_name"] = vim_info
.get("name")
544 if vim_info
["status"] in ("ERROR", "VIM_ERROR"):
545 if ro_task
["vim_info"]["vim_details"] != vim_info
.get("error_msg"):
546 ro_vim_item_update
["vim_details"] = vim_info
.get("error_msg")
547 elif vim_info
["status"] == "DELETED":
548 ro_vim_item_update
["vim_id"] = None
549 ro_vim_item_update
["vim_details"] = "Deleted externally"
551 if ro_task
["vim_info"]["vim_details"] != vim_info
["vim_info"]:
552 ro_vim_item_update
["vim_details"] = vim_info
["vim_info"]
554 if ro_vim_item_update
:
556 "ro_task={} {} get-vm={}: status={} {}".format(
558 ro_task
["target_id"],
560 ro_vim_item_update
.get("vim_status"),
561 ro_vim_item_update
.get("vim_details")
562 if ro_vim_item_update
.get("vim_status") != "ACTIVE"
567 return task_status
, ro_vim_item_update
569 def exec(self
, ro_task
, task_index
, task_depends
):
570 task
= ro_task
["tasks"][task_index
]
571 task_id
= task
["task_id"]
572 target_vim
= self
.my_vims
[ro_task
["target_id"]]
573 db_task_update
= {"retries": 0}
574 retries
= task
.get("retries", 0)
577 params
= task
["params"]
578 params_copy
= deepcopy(params
)
579 params_copy
["ro_key"] = self
.db
.decrypt(
580 params_copy
.pop("private_key"),
581 params_copy
.pop("schema_version"),
582 params_copy
.pop("salt"),
584 params_copy
["ip_addr"] = params_copy
.pop("ip_address")
585 target_vim
.inject_user_key(**params_copy
)
587 "task={} {} action-vm=inject_key".format(task_id
, ro_task
["target_id"])
594 ) # params_copy["key"]
595 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
598 if retries
< self
.max_retries_inject_ssh_key
:
604 "next_retry": self
.time_retries_inject_ssh_key
,
609 "task={} {} inject-ssh-key: {}".format(task_id
, ro_task
["target_id"], e
)
611 ro_vim_item_update
= {"vim_details": str(e
)}
613 return "FAILED", ro_vim_item_update
, db_task_update
616 class VimInteractionImage(VimInteractionBase
):
617 def new(self
, ro_task
, task_index
, task_depends
):
618 task
= ro_task
["tasks"][task_index
]
619 task_id
= task
["task_id"]
622 target_vim
= self
.my_vims
[ro_task
["target_id"]]
626 if task
.get("find_params"):
627 vim_images
= target_vim
.get_image_list(**task
["find_params"])
630 raise NsWorkerExceptionNotFound(
631 "Image not found with this criteria: '{}'".format(
635 elif len(vim_images
) > 1:
636 raise NsWorkerException(
637 "More than one image found with this criteria: '{}'".format(
642 vim_image_id
= vim_images
[0]["id"]
644 ro_vim_item_update
= {
645 "vim_id": vim_image_id
,
646 "vim_status": "DONE",
648 "created_items": created_items
,
652 "task={} {} new-image={} created={}".format(
653 task_id
, ro_task
["target_id"], vim_image_id
, created
657 return "DONE", ro_vim_item_update
658 except (NsWorkerException
, vimconn
.VimConnException
) as e
:
660 "task={} {} new-image: {}".format(task_id
, ro_task
["target_id"], e
)
662 ro_vim_item_update
= {
663 "vim_status": "VIM_ERROR",
665 "vim_details": str(e
),
668 return "FAILED", ro_vim_item_update
671 class VimInteractionFlavor(VimInteractionBase
):
672 def delete(self
, ro_task
, task_index
):
673 task
= ro_task
["tasks"][task_index
]
674 task_id
= task
["task_id"]
675 flavor_vim_id
= ro_task
["vim_info"]["vim_id"]
676 ro_vim_item_update_ok
= {
677 "vim_status": "DELETED",
679 "vim_details": "DELETED",
685 target_vim
= self
.my_vims
[ro_task
["target_id"]]
686 target_vim
.delete_flavor(flavor_vim_id
)
687 except vimconn
.VimConnNotFoundException
:
688 ro_vim_item_update_ok
["vim_details"] = "already deleted"
689 except vimconn
.VimConnException
as e
:
691 "ro_task={} vim={} del-flavor={}: {}".format(
692 ro_task
["_id"], ro_task
["target_id"], flavor_vim_id
, e
695 ro_vim_item_update
= {
696 "vim_status": "VIM_ERROR",
697 "vim_details": "Error while deleting: {}".format(e
),
700 return "FAILED", ro_vim_item_update
703 "task={} {} del-flavor={} {}".format(
705 ro_task
["target_id"],
707 ro_vim_item_update_ok
.get("vim_details", ""),
711 return "DONE", ro_vim_item_update_ok
713 def new(self
, ro_task
, task_index
, task_depends
):
714 task
= ro_task
["tasks"][task_index
]
715 task_id
= task
["task_id"]
718 target_vim
= self
.my_vims
[ro_task
["target_id"]]
724 if task
.get("find_params"):
726 flavor_data
= task
["find_params"]["flavor_data"]
727 vim_flavor_id
= target_vim
.get_flavor_id_from_data(flavor_data
)
728 except vimconn
.VimConnNotFoundException
:
731 if not vim_flavor_id
and task
.get("params"):
733 flavor_data
= task
["params"]["flavor_data"]
734 vim_flavor_id
= target_vim
.new_flavor(flavor_data
)
737 ro_vim_item_update
= {
738 "vim_id": vim_flavor_id
,
739 "vim_status": "DONE",
741 "created_items": created_items
,
745 "task={} {} new-flavor={} created={}".format(
746 task_id
, ro_task
["target_id"], vim_flavor_id
, created
750 return "DONE", ro_vim_item_update
751 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
753 "task={} vim={} new-flavor: {}".format(task_id
, ro_task
["target_id"], e
)
755 ro_vim_item_update
= {
756 "vim_status": "VIM_ERROR",
758 "vim_details": str(e
),
761 return "FAILED", ro_vim_item_update
764 class VimInteractionAffinityGroup(VimInteractionBase
):
765 def delete(self
, ro_task
, task_index
):
766 task
= ro_task
["tasks"][task_index
]
767 task_id
= task
["task_id"]
768 affinity_group_vim_id
= ro_task
["vim_info"]["vim_id"]
769 ro_vim_item_update_ok
= {
770 "vim_status": "DELETED",
772 "vim_details": "DELETED",
777 if affinity_group_vim_id
:
778 target_vim
= self
.my_vims
[ro_task
["target_id"]]
779 target_vim
.delete_affinity_group(affinity_group_vim_id
)
780 except vimconn
.VimConnNotFoundException
:
781 ro_vim_item_update_ok
["vim_details"] = "already deleted"
782 except vimconn
.VimConnException
as e
:
784 "ro_task={} vim={} del-affinity-or-anti-affinity-group={}: {}".format(
785 ro_task
["_id"], ro_task
["target_id"], affinity_group_vim_id
, e
788 ro_vim_item_update
= {
789 "vim_status": "VIM_ERROR",
790 "vim_details": "Error while deleting: {}".format(e
),
793 return "FAILED", ro_vim_item_update
796 "task={} {} del-affinity-or-anti-affinity-group={} {}".format(
798 ro_task
["target_id"],
799 affinity_group_vim_id
,
800 ro_vim_item_update_ok
.get("vim_details", ""),
804 return "DONE", ro_vim_item_update_ok
806 def new(self
, ro_task
, task_index
, task_depends
):
807 task
= ro_task
["tasks"][task_index
]
808 task_id
= task
["task_id"]
811 target_vim
= self
.my_vims
[ro_task
["target_id"]]
814 affinity_group_vim_id
= None
816 if task
.get("params"):
817 affinity_group_data
= task
["params"]["affinity_group_data"]
818 affinity_group_vim_id
= target_vim
.new_affinity_group(
823 ro_vim_item_update
= {
824 "vim_id": affinity_group_vim_id
,
825 "vim_status": "DONE",
827 "created_items": created_items
,
831 "task={} {} new-affinity-or-anti-affinity-group={} created={}".format(
832 task_id
, ro_task
["target_id"], affinity_group_vim_id
, created
836 return "DONE", ro_vim_item_update
837 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
839 "task={} vim={} new-affinity-or-anti-affinity-group:"
840 " {}".format(task_id
, ro_task
["target_id"], e
)
842 ro_vim_item_update
= {
843 "vim_status": "VIM_ERROR",
845 "vim_details": str(e
),
848 return "FAILED", ro_vim_item_update
851 class VimInteractionSdnNet(VimInteractionBase
):
853 def _match_pci(port_pci
, mapping
):
855 Check if port_pci matches with mapping
856 mapping can have brackets to indicate that several chars are accepted. e.g
857 pci '0000:af:10.1' matches with '0000:af:1[01].[1357]'
858 :param port_pci: text
859 :param mapping: text, can contain brackets to indicate several chars are available
860 :return: True if matches, False otherwise
862 if not port_pci
or not mapping
:
864 if port_pci
== mapping
:
870 bracket_start
= mapping
.find("[", mapping_index
)
872 if bracket_start
== -1:
875 bracket_end
= mapping
.find("]", bracket_start
)
876 if bracket_end
== -1:
879 length
= bracket_start
- mapping_index
882 and port_pci
[pci_index
: pci_index
+ length
]
883 != mapping
[mapping_index
:bracket_start
]
888 port_pci
[pci_index
+ length
]
889 not in mapping
[bracket_start
+ 1 : bracket_end
]
893 pci_index
+= length
+ 1
894 mapping_index
= bracket_end
+ 1
896 if port_pci
[pci_index
:] != mapping
[mapping_index
:]:
901 def _get_interfaces(self
, vlds_to_connect
, vim_account_id
):
903 :param vlds_to_connect: list with format vnfrs:<id>:vld.<vld_id> or nsrs:<id>:vld.<vld_id>
904 :param vim_account_id:
909 for vld
in vlds_to_connect
:
910 table
, _
, db_id
= vld
.partition(":")
911 db_id
, _
, vld
= db_id
.partition(":")
912 _
, _
, vld_id
= vld
.partition(".")
915 q_filter
= {"vim-account-id": vim_account_id
, "_id": db_id
}
916 iface_key
= "vnf-vld-id"
917 else: # table == "nsrs"
918 q_filter
= {"vim-account-id": vim_account_id
, "nsr-id-ref": db_id
}
919 iface_key
= "ns-vld-id"
921 db_vnfrs
= self
.db
.get_list("vnfrs", q_filter
=q_filter
)
923 for db_vnfr
in db_vnfrs
:
924 for vdu_index
, vdur
in enumerate(db_vnfr
.get("vdur", ())):
925 for iface_index
, interface
in enumerate(vdur
["interfaces"]):
926 if interface
.get(iface_key
) == vld_id
and interface
.get(
928 ) in ("SR-IOV", "PCI-PASSTHROUGH"):
930 interface_
= interface
.copy()
931 interface_
["id"] = "vnfrs:{}:vdu.{}.interfaces.{}".format(
932 db_vnfr
["_id"], vdu_index
, iface_index
935 if vdur
.get("status") == "ERROR":
936 interface_
["status"] = "ERROR"
938 interfaces
.append(interface_
)
942 def refresh(self
, ro_task
):
943 # look for task create
944 task_create_index
, _
= next(
946 for i_t
in enumerate(ro_task
["tasks"])
948 and i_t
[1]["action"] == "CREATE"
949 and i_t
[1]["status"] != "FINISHED"
952 return self
.new(ro_task
, task_create_index
, None)
954 def new(self
, ro_task
, task_index
, task_depends
):
956 task
= ro_task
["tasks"][task_index
]
957 task_id
= task
["task_id"]
958 target_vim
= self
.my_vims
[ro_task
["target_id"]]
960 sdn_net_id
= ro_task
["vim_info"]["vim_id"]
962 created_items
= ro_task
["vim_info"].get("created_items")
963 connected_ports
= ro_task
["vim_info"].get("connected_ports", [])
964 new_connected_ports
= []
965 last_update
= ro_task
["vim_info"].get("last_update", 0)
966 sdn_status
= ro_task
["vim_info"].get("vim_status", "BUILD") or "BUILD"
968 created
= ro_task
["vim_info"].get("created", False)
972 params
= task
["params"]
973 vlds_to_connect
= params
["vlds"]
974 associated_vim
= params
["target_vim"]
975 # external additional ports
976 additional_ports
= params
.get("sdn-ports") or ()
977 _
, _
, vim_account_id
= associated_vim
.partition(":")
981 if associated_vim
not in self
.db_vims
:
982 self
.db_vims
[associated_vim
] = self
.db
.get_one(
983 "vim_accounts", {"_id": vim_account_id
}
986 db_vim
= self
.db_vims
[associated_vim
]
988 # look for ports to connect
989 ports
= self
._get
_interfaces
(vlds_to_connect
, vim_account_id
)
993 pending_ports
= error_ports
= 0
995 sdn_need_update
= False
998 vlan_used
= port
.get("vlan") or vlan_used
1000 # TODO. Do not connect if already done
1001 if not port
.get("compute_node") or not port
.get("pci"):
1002 if port
.get("status") == "ERROR":
1009 compute_node_mappings
= next(
1012 for c
in db_vim
["config"].get("sdn-port-mapping", ())
1013 if c
and c
["compute_node"] == port
["compute_node"]
1018 if compute_node_mappings
:
1019 # process port_mapping pci of type 0000:af:1[01].[1357]
1023 for p
in compute_node_mappings
["ports"]
1024 if self
._match
_pci
(port
["pci"], p
.get("pci"))
1030 if not db_vim
["config"].get("mapping_not_needed"):
1032 "Port mapping not found for compute_node={} pci={}".format(
1033 port
["compute_node"], port
["pci"]
1040 service_endpoint_id
= "{}:{}".format(port
["compute_node"], port
["pci"])
1042 "service_endpoint_id": pmap
.get("service_endpoint_id")
1043 or service_endpoint_id
,
1044 "service_endpoint_encapsulation_type": "dot1q"
1045 if port
["type"] == "SR-IOV"
1047 "service_endpoint_encapsulation_info": {
1048 "vlan": port
.get("vlan"),
1049 "mac": port
.get("mac-address"),
1050 "device_id": pmap
.get("device_id") or port
["compute_node"],
1051 "device_interface_id": pmap
.get("device_interface_id")
1053 "switch_dpid": pmap
.get("switch_id") or pmap
.get("switch_dpid"),
1054 "switch_port": pmap
.get("switch_port"),
1055 "service_mapping_info": pmap
.get("service_mapping_info"),
1060 # if port["modified_at"] > last_update:
1061 # sdn_need_update = True
1062 new_connected_ports
.append(port
["id"]) # TODO
1063 sdn_ports
.append(new_port
)
1067 "{} interfaces have not been created as VDU is on ERROR status".format(
1072 # connect external ports
1073 for index
, additional_port
in enumerate(additional_ports
):
1074 additional_port_id
= additional_port
.get(
1075 "service_endpoint_id"
1076 ) or "external-{}".format(index
)
1079 "service_endpoint_id": additional_port_id
,
1080 "service_endpoint_encapsulation_type": additional_port
.get(
1081 "service_endpoint_encapsulation_type", "dot1q"
1083 "service_endpoint_encapsulation_info": {
1084 "vlan": additional_port
.get("vlan") or vlan_used
,
1085 "mac": additional_port
.get("mac_address"),
1086 "device_id": additional_port
.get("device_id"),
1087 "device_interface_id": additional_port
.get(
1088 "device_interface_id"
1090 "switch_dpid": additional_port
.get("switch_dpid")
1091 or additional_port
.get("switch_id"),
1092 "switch_port": additional_port
.get("switch_port"),
1093 "service_mapping_info": additional_port
.get(
1094 "service_mapping_info"
1099 new_connected_ports
.append(additional_port_id
)
1102 # if there are more ports to connect or they have been modified, call create/update
1104 sdn_status
= "ERROR"
1105 sdn_info
= "; ".join(error_list
)
1106 elif set(connected_ports
) != set(new_connected_ports
) or sdn_need_update
:
1107 last_update
= time
.time()
1110 if len(sdn_ports
) < 2:
1111 sdn_status
= "ACTIVE"
1113 if not pending_ports
:
1115 "task={} {} new-sdn-net done, less than 2 ports".format(
1116 task_id
, ro_task
["target_id"]
1120 net_type
= params
.get("type") or "ELAN"
1124 ) = target_vim
.create_connectivity_service(net_type
, sdn_ports
)
1127 "task={} {} new-sdn-net={} created={}".format(
1128 task_id
, ro_task
["target_id"], sdn_net_id
, created
1132 created_items
= target_vim
.edit_connectivity_service(
1133 sdn_net_id
, conn_info
=created_items
, connection_points
=sdn_ports
1137 "task={} {} update-sdn-net={} created={}".format(
1138 task_id
, ro_task
["target_id"], sdn_net_id
, created
1142 connected_ports
= new_connected_ports
1144 wim_status_dict
= target_vim
.get_connectivity_service_status(
1145 sdn_net_id
, conn_info
=created_items
1147 sdn_status
= wim_status_dict
["sdn_status"]
1149 if wim_status_dict
.get("sdn_info"):
1150 sdn_info
= str(wim_status_dict
.get("sdn_info")) or ""
1152 if wim_status_dict
.get("error_msg"):
1153 sdn_info
= wim_status_dict
.get("error_msg") or ""
1156 if sdn_status
!= "ERROR":
1157 sdn_info
= "Waiting for getting interfaces location from VIM. Obtained '{}' of {}".format(
1158 len(ports
) - pending_ports
, len(ports
)
1161 if sdn_status
== "ACTIVE":
1162 sdn_status
= "BUILD"
1164 ro_vim_item_update
= {
1165 "vim_id": sdn_net_id
,
1166 "vim_status": sdn_status
,
1168 "created_items": created_items
,
1169 "connected_ports": connected_ports
,
1170 "vim_details": sdn_info
,
1171 "last_update": last_update
,
1174 return sdn_status
, ro_vim_item_update
1175 except Exception as e
:
1177 "task={} vim={} new-net: {}".format(task_id
, ro_task
["target_id"], e
),
1178 exc_info
=not isinstance(
1179 e
, (sdnconn
.SdnConnectorError
, vimconn
.VimConnException
)
1182 ro_vim_item_update
= {
1183 "vim_status": "VIM_ERROR",
1185 "vim_details": str(e
),
1188 return "FAILED", ro_vim_item_update
1190 def delete(self
, ro_task
, task_index
):
1191 task
= ro_task
["tasks"][task_index
]
1192 task_id
= task
["task_id"]
1193 sdn_vim_id
= ro_task
["vim_info"].get("vim_id")
1194 ro_vim_item_update_ok
= {
1195 "vim_status": "DELETED",
1197 "vim_details": "DELETED",
1203 target_vim
= self
.my_vims
[ro_task
["target_id"]]
1204 target_vim
.delete_connectivity_service(
1205 sdn_vim_id
, ro_task
["vim_info"].get("created_items")
1208 except Exception as e
:
1210 isinstance(e
, sdnconn
.SdnConnectorError
)
1211 and e
.http_code
== HTTPStatus
.NOT_FOUND
.value
1213 ro_vim_item_update_ok
["vim_details"] = "already deleted"
1216 "ro_task={} vim={} del-sdn-net={}: {}".format(
1217 ro_task
["_id"], ro_task
["target_id"], sdn_vim_id
, e
1219 exc_info
=not isinstance(
1220 e
, (sdnconn
.SdnConnectorError
, vimconn
.VimConnException
)
1223 ro_vim_item_update
= {
1224 "vim_status": "VIM_ERROR",
1225 "vim_details": "Error while deleting: {}".format(e
),
1228 return "FAILED", ro_vim_item_update
1231 "task={} {} del-sdn-net={} {}".format(
1233 ro_task
["target_id"],
1235 ro_vim_item_update_ok
.get("vim_details", ""),
1239 return "DONE", ro_vim_item_update_ok
1242 class NsWorker(threading
.Thread
):
1243 REFRESH_BUILD
= 5 # 5 seconds
1244 REFRESH_ACTIVE
= 60 # 1 minute
1246 REFRESH_IMAGE
= 3600 * 10
1247 REFRESH_DELETE
= 3600 * 10
1251 def __init__(self
, worker_index
, config
, plugins
, db
):
1254 :param worker_index: thread index
1255 :param config: general configuration of RO, among others the process_id with the docker id where it runs
1256 :param plugins: global shared dict with the loaded plugins
1257 :param db: database class instance to use
1259 threading
.Thread
.__init
__(self
)
1260 self
.config
= config
1261 self
.plugins
= plugins
1262 self
.plugin_name
= "unknown"
1263 self
.logger
= logging
.getLogger("ro.worker{}".format(worker_index
))
1264 self
.worker_index
= worker_index
1265 self
.task_queue
= queue
.Queue(self
.QUEUE_SIZE
)
1266 # targetvim: vimplugin class
1268 # targetvim: vim information from database
1271 self
.vim_targets
= []
1272 self
.my_id
= config
["process_id"] + ":" + str(worker_index
)
1275 "net": VimInteractionNet(self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
),
1276 "vdu": VimInteractionVdu(self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
),
1277 "image": VimInteractionImage(
1278 self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
1280 "flavor": VimInteractionFlavor(
1281 self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
1283 "sdn_net": VimInteractionSdnNet(
1284 self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
1286 "affinity-or-anti-affinity-group": VimInteractionAffinityGroup(
1287 self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
1290 self
.time_last_task_processed
= None
1291 # lists of tasks to delete because nsrs or vnfrs has been deleted from db
1292 self
.tasks_to_delete
= []
1293 # it is idle when there are not vim_targets associated
1295 self
.task_locked_time
= config
["global"]["task_locked_time"]
1297 def insert_task(self
, task
):
1299 self
.task_queue
.put(task
, False)
1302 raise NsWorkerException("timeout inserting a task")
1304 def terminate(self
):
1305 self
.insert_task("exit")
1307 def del_task(self
, task
):
1308 with self
.task_lock
:
1309 if task
["status"] == "SCHEDULED":
1310 task
["status"] = "SUPERSEDED"
1312 else: # task["status"] == "processing"
1313 self
.task_lock
.release()
1316 def _process_vim_config(self
, target_id
, db_vim
):
1318 Process vim config, creating vim configuration files as ca_cert
1319 :param target_id: vim/sdn/wim + id
1320 :param db_vim: Vim dictionary obtained from database
1321 :return: None. Modifies vim. Creates a folder target_id:worker_index and several files
1323 if not db_vim
.get("config"):
1329 if db_vim
["config"].get("ca_cert_content"):
1330 file_name
= "{}:{}".format(target_id
, self
.worker_index
)
1334 except FileExistsError
:
1337 file_name
= file_name
+ "/ca_cert"
1339 with
open(file_name
, "w") as f
:
1340 f
.write(db_vim
["config"]["ca_cert_content"])
1341 del db_vim
["config"]["ca_cert_content"]
1342 db_vim
["config"]["ca_cert"] = file_name
1343 except Exception as e
:
1344 raise NsWorkerException(
1345 "Error writing to file '{}': {}".format(file_name
, e
)
1348 def _load_plugin(self
, name
, type="vim"):
1349 # type can be vim or sdn
1350 if "rovim_dummy" not in self
.plugins
:
1351 self
.plugins
["rovim_dummy"] = VimDummyConnector
1353 if "rosdn_dummy" not in self
.plugins
:
1354 self
.plugins
["rosdn_dummy"] = SdnDummyConnector
1356 if name
in self
.plugins
:
1357 return self
.plugins
[name
]
1360 for ep
in entry_points(group
="osm_ro{}.plugins".format(type), name
=name
):
1361 self
.plugins
[name
] = ep
.load()
1362 except Exception as e
:
1363 raise NsWorkerException("Cannot load plugin osm_{}: {}".format(name
, e
))
1365 if name
and name
not in self
.plugins
:
1366 raise NsWorkerException(
1367 "Plugin 'osm_{n}' has not been installed".format(n
=name
)
1370 return self
.plugins
[name
]
1372 def _unload_vim(self
, target_id
):
1374 Unload a vim_account. Removes it from self db_vims dictionary, my_vims dictionary and vim_targets list
1375 :param target_id: Contains type:_id; where type can be 'vim', ...
1379 self
.db_vims
.pop(target_id
, None)
1380 self
.my_vims
.pop(target_id
, None)
1382 if target_id
in self
.vim_targets
:
1383 self
.vim_targets
.remove(target_id
)
1385 self
.logger
.info("Unloaded {}".format(target_id
))
1386 rmtree("{}:{}".format(target_id
, self
.worker_index
))
1387 except FileNotFoundError
:
1388 pass # this is raised by rmtree if folder does not exist
1389 except Exception as e
:
1390 self
.logger
.error("Cannot unload {}: {}".format(target_id
, e
))
1392 def _check_vim(self
, target_id
):
1394 Load a VIM/SDN/WIM (if not loaded) and check connectivity, updating database with ENABLE or ERROR
1395 :param target_id: Contains type:_id; type can be 'vim', 'sdn' or 'wim'
1398 target
, _
, _id
= target_id
.partition(":")
1404 loaded
= target_id
in self
.vim_targets
1414 step
= "Getting {} from db".format(target_id
)
1415 db_vim
= self
.db
.get_one(target_database
, {"_id": _id
})
1417 for op_index
, operation
in enumerate(
1418 db_vim
["_admin"].get("operations", ())
1420 if operation
["operationState"] != "PROCESSING":
1423 locked_at
= operation
.get("locked_at")
1425 if locked_at
is not None and locked_at
>= now
- self
.task_locked_time
:
1426 # some other thread is doing this operation
1430 op_text
= "_admin.operations.{}.".format(op_index
)
1432 if not self
.db
.set_one(
1436 op_text
+ "operationState": "PROCESSING",
1437 op_text
+ "locked_at": locked_at
,
1440 op_text
+ "locked_at": now
,
1441 "admin.current_operation": op_index
,
1443 fail_on_empty
=False,
1447 unset_dict
[op_text
+ "locked_at"] = None
1448 unset_dict
["current_operation"] = None
1449 step
= "Loading " + target_id
1450 error_text
= self
._load
_vim
(target_id
)
1453 step
= "Checking connectivity"
1456 self
.my_vims
[target_id
].check_vim_connectivity()
1458 self
.my_vims
[target_id
].check_credentials()
1460 update_dict
["_admin.operationalState"] = "ENABLED"
1461 update_dict
["_admin.detailed-status"] = ""
1462 unset_dict
[op_text
+ "detailed-status"] = None
1463 update_dict
[op_text
+ "operationState"] = "COMPLETED"
1467 except Exception as e
:
1468 error_text
= "{}: {}".format(step
, e
)
1469 self
.logger
.error("{} for {}: {}".format(step
, target_id
, e
))
1472 if update_dict
or unset_dict
:
1474 update_dict
[op_text
+ "operationState"] = "FAILED"
1475 update_dict
[op_text
+ "detailed-status"] = error_text
1476 unset_dict
.pop(op_text
+ "detailed-status", None)
1477 update_dict
["_admin.operationalState"] = "ERROR"
1478 update_dict
["_admin.detailed-status"] = error_text
1481 update_dict
[op_text
+ "statusEnteredTime"] = now
1485 q_filter
={"_id": _id
},
1486 update_dict
=update_dict
,
1488 fail_on_empty
=False,
1492 self
._unload
_vim
(target_id
)
1494 def _reload_vim(self
, target_id
):
1495 if target_id
in self
.vim_targets
:
1496 self
._load
_vim
(target_id
)
1498 # if the vim is not loaded, but database information of VIM is cached at self.db_vims,
1499 # just remove it to force load again next time it is needed
1500 self
.db_vims
.pop(target_id
, None)
1502 def _load_vim(self
, target_id
):
1504 Load or reload a vim_account, sdn_controller or wim_account.
1505 Read content from database, load the plugin if not loaded.
1506 In case of error loading the plugin, it load a failing VIM_connector
1507 It fills self db_vims dictionary, my_vims dictionary and vim_targets list
1508 :param target_id: Contains type:_id; where type can be 'vim', ...
1509 :return: None if ok, descriptive text if error
1511 target
, _
, _id
= target_id
.partition(":")
1523 step
= "Getting {}={} from db".format(target
, _id
)
1524 # TODO process for wim, sdnc, ...
1525 vim
= self
.db
.get_one(target_database
, {"_id": _id
})
1527 # if deep_get(vim, "config", "sdn-controller"):
1528 # step = "Getting sdn-controller-id='{}' from db".format(vim["config"]["sdn-controller"])
1529 # db_sdn = self.db.get_one("sdns", {"_id": vim["config"]["sdn-controller"]})
1531 step
= "Decrypting password"
1532 schema_version
= vim
.get("schema_version")
1533 self
.db
.encrypt_decrypt_fields(
1536 fields
=("password", "secret"),
1537 schema_version
=schema_version
,
1540 self
._process
_vim
_config
(target_id
, vim
)
1543 plugin_name
= "rovim_" + vim
["vim_type"]
1544 step
= "Loading plugin '{}'".format(plugin_name
)
1545 vim_module_conn
= self
._load
_plugin
(plugin_name
)
1546 step
= "Loading {}'".format(target_id
)
1547 self
.my_vims
[target_id
] = vim_module_conn(
1550 tenant_id
=vim
.get("vim_tenant_id"),
1551 tenant_name
=vim
.get("vim_tenant_name"),
1554 user
=vim
["vim_user"],
1555 passwd
=vim
["vim_password"],
1556 config
=vim
.get("config") or {},
1560 plugin_name
= "rosdn_" + vim
["type"]
1561 step
= "Loading plugin '{}'".format(plugin_name
)
1562 vim_module_conn
= self
._load
_plugin
(plugin_name
, "sdn")
1563 step
= "Loading {}'".format(target_id
)
1565 wim_config
= wim
.pop("config", {}) or {}
1566 wim
["uuid"] = wim
["_id"]
1567 wim
["wim_url"] = wim
["url"]
1570 wim_config
["dpid"] = wim
.pop("dpid")
1572 if wim
.get("switch_id"):
1573 wim_config
["switch_id"] = wim
.pop("switch_id")
1575 # wim, wim_account, config
1576 self
.my_vims
[target_id
] = vim_module_conn(wim
, wim
, wim_config
)
1577 self
.db_vims
[target_id
] = vim
1578 self
.error_status
= None
1581 "Connector loaded for {}, plugin={}".format(target_id
, plugin_name
)
1583 except Exception as e
:
1585 "Cannot load {} plugin={}: {} {}".format(
1586 target_id
, plugin_name
, step
, e
1590 self
.db_vims
[target_id
] = vim
or {}
1591 self
.db_vims
[target_id
] = FailingConnector(str(e
))
1592 error_status
= "{} Error: {}".format(step
, e
)
1596 if target_id
not in self
.vim_targets
:
1597 self
.vim_targets
.append(target_id
)
1599 def _get_db_task(self
):
1601 Read actions from database and reload them at memory. Fill self.refresh_list, pending_list, vim_actions
1606 if not self
.time_last_task_processed
:
1607 self
.time_last_task_processed
= now
1612 # Log RO tasks only when loglevel is DEBUG
1613 if self.logger.getEffectiveLevel() == logging.DEBUG:
1620 + str(self.task_locked_time)
1622 + "time_last_task_processed="
1623 + str(self.time_last_task_processed)
1629 locked
= self
.db
.set_one(
1632 "target_id": self
.vim_targets
,
1633 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
1634 "locked_at.lt": now
- self
.task_locked_time
,
1635 "to_check_at.lt": self
.time_last_task_processed
,
1637 update_dict
={"locked_by": self
.my_id
, "locked_at": now
},
1638 fail_on_empty
=False,
1643 ro_task
= self
.db
.get_one(
1646 "target_id": self
.vim_targets
,
1647 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
1653 if self
.time_last_task_processed
== now
:
1654 self
.time_last_task_processed
= None
1657 self
.time_last_task_processed
= now
1658 # self.time_last_task_processed = min(self.time_last_task_processed + 1000, now)
1660 except DbException
as e
:
1661 self
.logger
.error("Database exception at _get_db_task: {}".format(e
))
1662 except Exception as e
:
1663 self
.logger
.critical(
1664 "Unexpected exception at _get_db_task: {}".format(e
), exc_info
=True
1669 def _get_db_all_tasks(self
):
1671 Read all content of table ro_tasks to log it
1675 # Checking the content of the BD:
1678 ro_task
= self
.db
.get_list("ro_tasks")
1680 self
._log
_ro
_task
(rt
, None, None, "TASK_WF", "GET_ALL_TASKS")
1683 except DbException
as e
:
1684 self
.logger
.error("Database exception at _get_db_all_tasks: {}".format(e
))
1685 except Exception as e
:
1686 self
.logger
.critical(
1687 "Unexpected exception at _get_db_all_tasks: {}".format(e
), exc_info
=True
1692 def _log_ro_task(self
, ro_task
, db_ro_task_update
, db_ro_task_delete
, mark
, event
):
1694 Generate a log with the following format:
1696 Mark;Event;ro_task_id;locked_at;modified_at;created_at;to_check_at;locked_by;
1697 target_id;vim_info.refresh_at;vim_info;no_of_tasks;task_status;action_id;
1698 task_array_index;task_id;task_action;task_item;task_args
1702 TASK_WF;GET_TASK;888f1864-749a-4fc2-bc1a-97c0fffd6a6f:2;1642158724.8210013;
1703 1642158640.7986135;1642158640.7986135;1642158640.7986135;b134c9494e75:0a
1704 ;vim:b7ff9e24-8868-4d68-8a57-a59dc11d0327;None;{'created': False,
1705 'created_items': None, 'vim_id': None, 'vim_name': None, 'vim_status': None,
1706 'vim_details': None, 'refresh_at': None};1;SCHEDULED;
1707 888f1864-749a-4fc2-bc1a-97c0fffd6a6f;0;888f1864-749a-4fc2-bc1a-97c0fffd6a6f:2;
1708 CREATE;image;{'filter_dict': {'name': 'ubuntu-os-cloud:image-family:ubuntu-1804-lts'}}
1713 if ro_task
is not None and isinstance(ro_task
, dict):
1714 for t
in ro_task
["tasks"]:
1718 line
.append(ro_task
.get("_id", ""))
1719 line
.append(str(ro_task
.get("locked_at", "")))
1720 line
.append(str(ro_task
.get("modified_at", "")))
1721 line
.append(str(ro_task
.get("created_at", "")))
1722 line
.append(str(ro_task
.get("to_check_at", "")))
1723 line
.append(str(ro_task
.get("locked_by", "")))
1724 line
.append(str(ro_task
.get("target_id", "")))
1725 line
.append(str(ro_task
.get("vim_info", {}).get("refresh_at", "")))
1726 line
.append(str(ro_task
.get("vim_info", "")))
1727 line
.append(str(ro_task
.get("tasks", "")))
1728 if isinstance(t
, dict):
1729 line
.append(str(t
.get("status", "")))
1730 line
.append(str(t
.get("action_id", "")))
1732 line
.append(str(t
.get("task_id", "")))
1733 line
.append(str(t
.get("action", "")))
1734 line
.append(str(t
.get("item", "")))
1735 line
.append(str(t
.get("find_params", "")))
1736 line
.append(str(t
.get("params", "")))
1738 line
.extend([""] * 2)
1740 line
.extend([""] * 5)
1743 self
.logger
.debug(";".join(line
))
1744 elif db_ro_task_update
is not None and isinstance(db_ro_task_update
, dict):
1747 st
= "tasks.{}.status".format(i
)
1748 if st
not in db_ro_task_update
:
1753 line
.append(db_ro_task_update
.get("_id", ""))
1754 line
.append(str(db_ro_task_update
.get("locked_at", "")))
1755 line
.append(str(db_ro_task_update
.get("modified_at", "")))
1757 line
.append(str(db_ro_task_update
.get("to_check_at", "")))
1758 line
.append(str(db_ro_task_update
.get("locked_by", "")))
1760 line
.append(str(db_ro_task_update
.get("vim_info.refresh_at", "")))
1762 line
.append(str(db_ro_task_update
.get("vim_info", "")))
1763 line
.append(str(str(db_ro_task_update
).count(".status")))
1764 line
.append(db_ro_task_update
.get(st
, ""))
1767 line
.extend([""] * 3)
1769 self
.logger
.debug(";".join(line
))
1771 elif db_ro_task_delete
is not None and isinstance(db_ro_task_delete
, dict):
1775 line
.append(db_ro_task_delete
.get("_id", ""))
1777 line
.append(db_ro_task_delete
.get("modified_at", ""))
1778 line
.extend([""] * 13)
1779 self
.logger
.debug(";".join(line
))
1785 line
.extend([""] * 16)
1786 self
.logger
.debug(";".join(line
))
1788 except Exception as e
:
1789 self
.logger
.error("Error logging ro_task: {}".format(e
))
1791 def _delete_task(self
, ro_task
, task_index
, task_depends
, db_update
):
1793 Determine if this task need to be done or superseded
1796 my_task
= ro_task
["tasks"][task_index
]
1797 task_id
= my_task
["task_id"]
1798 needed_delete
= ro_task
["vim_info"]["created"] or ro_task
["vim_info"].get(
1799 "created_items", False
1802 if my_task
["status"] == "FAILED":
1803 return None, None # TODO need to be retry??
1806 for index
, task
in enumerate(ro_task
["tasks"]):
1807 if index
== task_index
or not task
:
1811 my_task
["target_record"] == task
["target_record"]
1812 and task
["action"] == "CREATE"
1815 db_update
["tasks.{}.status".format(index
)] = task
[
1818 elif task
["action"] == "CREATE" and task
["status"] not in (
1822 needed_delete
= False
1825 return self
.item2class
[my_task
["item"]].delete(ro_task
, task_index
)
1827 return "SUPERSEDED", None
1828 except Exception as e
:
1829 if not isinstance(e
, NsWorkerException
):
1830 self
.logger
.critical(
1831 "Unexpected exception at _delete_task task={}: {}".format(
1837 return "FAILED", {"vim_status": "VIM_ERROR", "vim_details": str(e
)}
1839 def _create_task(self
, ro_task
, task_index
, task_depends
, db_update
):
1841 Determine if this task need to create something at VIM
1844 my_task
= ro_task
["tasks"][task_index
]
1845 task_id
= my_task
["task_id"]
1848 if my_task
["status"] == "FAILED":
1849 return None, None # TODO need to be retry??
1850 elif my_task
["status"] == "SCHEDULED":
1851 # check if already created by another task
1852 for index
, task
in enumerate(ro_task
["tasks"]):
1853 if index
== task_index
or not task
:
1856 if task
["action"] == "CREATE" and task
["status"] not in (
1861 return task
["status"], "COPY_VIM_INFO"
1864 task_status
, ro_vim_item_update
= self
.item2class
[my_task
["item"]].new(
1865 ro_task
, task_index
, task_depends
1867 # TODO update other CREATE tasks
1868 except Exception as e
:
1869 if not isinstance(e
, NsWorkerException
):
1871 "Error executing task={}: {}".format(task_id
, e
), exc_info
=True
1874 task_status
= "FAILED"
1875 ro_vim_item_update
= {"vim_status": "VIM_ERROR", "vim_details": str(e
)}
1876 # TODO update ro_vim_item_update
1878 return task_status
, ro_vim_item_update
1882 def _get_dependency(self
, task_id
, ro_task
=None, target_id
=None):
1884 Look for dependency task
1885 :param task_id: Can be one of
1886 1. target_vim+blank+task.target_record_id: "(vim|sdn|wim):<id> (vnfrs|nsrs):(vld|vdu|flavor|image).<id>"
1887 2. task.target_record_id: "(vnfrs|nsrs):(vld|vdu|flavor|image).<id>"
1888 3. task.task_id: "<action_id>:number"
1891 :return: database ro_task plus index of task
1894 task_id
.startswith("vim:")
1895 or task_id
.startswith("sdn:")
1896 or task_id
.startswith("wim:")
1898 target_id
, _
, task_id
= task_id
.partition(" ")
1900 if task_id
.startswith("nsrs:") or task_id
.startswith("vnfrs:"):
1901 ro_task_dependency
= self
.db
.get_one(
1903 q_filter
={"target_id": target_id
, "tasks.target_record_id": task_id
},
1904 fail_on_empty
=False,
1907 if ro_task_dependency
:
1908 for task_index
, task
in enumerate(ro_task_dependency
["tasks"]):
1909 if task
["target_record_id"] == task_id
:
1910 return ro_task_dependency
, task_index
1914 for task_index
, task
in enumerate(ro_task
["tasks"]):
1915 if task
and task
["task_id"] == task_id
:
1916 return ro_task
, task_index
1918 ro_task_dependency
= self
.db
.get_one(
1921 "tasks.ANYINDEX.task_id": task_id
,
1922 "tasks.ANYINDEX.target_record.ne": None,
1924 fail_on_empty
=False,
1927 if ro_task_dependency
:
1928 for task_index
, task
in ro_task_dependency
["tasks"]:
1929 if task
["task_id"] == task_id
:
1930 return ro_task_dependency
, task_index
1931 raise NsWorkerException("Cannot get depending task {}".format(task_id
))
1933 def _process_pending_tasks(self
, ro_task
):
1934 ro_task_id
= ro_task
["_id"]
1937 next_check_at
= now
+ (24 * 60 * 60)
1938 db_ro_task_update
= {}
1940 def _update_refresh(new_status
):
1941 # compute next_refresh
1943 nonlocal next_check_at
1944 nonlocal db_ro_task_update
1947 next_refresh
= time
.time()
1949 if task
["item"] in ("image", "flavor"):
1950 next_refresh
+= self
.REFRESH_IMAGE
1951 elif new_status
== "BUILD":
1952 next_refresh
+= self
.REFRESH_BUILD
1953 elif new_status
== "DONE":
1954 next_refresh
+= self
.REFRESH_ACTIVE
1956 next_refresh
+= self
.REFRESH_ERROR
1958 next_check_at
= min(next_check_at
, next_refresh
)
1959 db_ro_task_update
["vim_info.refresh_at"] = next_refresh
1960 ro_task
["vim_info"]["refresh_at"] = next_refresh
1964 # Log RO tasks only when loglevel is DEBUG
1965 if self.logger.getEffectiveLevel() == logging.DEBUG:
1966 self._log_ro_task(ro_task, None, None, "TASK_WF", "GET_TASK")
1968 # 0: get task_status_create
1970 task_status_create
= None
1974 for t
in ro_task
["tasks"]
1976 and t
["action"] == "CREATE"
1977 and t
["status"] in ("BUILD", "DONE")
1983 task_status_create
= task_create
["status"]
1985 # 1: look for tasks in status SCHEDULED, or in status CREATE if action is DONE or BUILD
1986 for task_action
in ("DELETE", "CREATE", "EXEC"):
1987 db_vim_update
= None
1990 for task_index
, task
in enumerate(ro_task
["tasks"]):
1992 continue # task deleted
1995 target_update
= None
1999 task_action
in ("DELETE", "EXEC")
2000 and task
["status"] not in ("SCHEDULED", "BUILD")
2002 or task
["action"] != task_action
2004 task_action
== "CREATE"
2005 and task
["status"] in ("FINISHED", "SUPERSEDED")
2010 task_path
= "tasks.{}.status".format(task_index
)
2012 db_vim_info_update
= None
2014 if task
["status"] == "SCHEDULED":
2015 # check if tasks that this depends on have been completed
2016 dependency_not_completed
= False
2018 for dependency_task_id
in task
.get("depends_on") or ():
2021 dependency_task_index
,
2022 ) = self
._get
_dependency
(
2023 dependency_task_id
, target_id
=ro_task
["target_id"]
2025 dependency_task
= dependency_ro_task
["tasks"][
2026 dependency_task_index
2029 if dependency_task
["status"] == "SCHEDULED":
2030 dependency_not_completed
= True
2031 next_check_at
= min(
2032 next_check_at
, dependency_ro_task
["to_check_at"]
2034 # must allow dependent task to be processed first
2035 # to do this set time after last_task_processed
2036 next_check_at
= max(
2037 self
.time_last_task_processed
, next_check_at
2040 elif dependency_task
["status"] == "FAILED":
2041 error_text
= "Cannot {} {} because depends on failed {} {} id={}): {}".format(
2044 dependency_task
["action"],
2045 dependency_task
["item"],
2047 dependency_ro_task
["vim_info"].get(
2052 "task={} {}".format(task
["task_id"], error_text
)
2054 raise NsWorkerException(error_text
)
2056 task_depends
[dependency_task_id
] = dependency_ro_task
[
2060 "TASK-{}".format(dependency_task_id
)
2061 ] = dependency_ro_task
["vim_info"]["vim_id"]
2063 if dependency_not_completed
:
2064 # TODO set at vim_info.vim_details that it is waiting
2067 # before calling VIM-plugin as it can take more than task_locked_time, insert to LockRenew
2068 # the task of renew this locking. It will update database locket_at periodically
2070 lock_object
= LockRenew
.add_lock_object(
2071 "ro_tasks", ro_task
, self
2074 if task
["action"] == "DELETE":
2075 (new_status
, db_vim_info_update
,) = self
._delete
_task
(
2076 ro_task
, task_index
, task_depends
, db_ro_task_update
2079 "FINISHED" if new_status
== "DONE" else new_status
2081 # ^with FINISHED instead of DONE it will not be refreshing
2083 if new_status
in ("FINISHED", "SUPERSEDED"):
2084 target_update
= "DELETE"
2085 elif task
["action"] == "EXEC":
2090 ) = self
.item2class
[task
["item"]].exec(
2091 ro_task
, task_index
, task_depends
2094 "FINISHED" if new_status
== "DONE" else new_status
2096 # ^with FINISHED instead of DONE it will not be refreshing
2099 # load into database the modified db_task_update "retries" and "next_retry"
2100 if db_task_update
.get("retries"):
2102 "tasks.{}.retries".format(task_index
)
2103 ] = db_task_update
["retries"]
2105 next_check_at
= time
.time() + db_task_update
.get(
2108 target_update
= None
2109 elif task
["action"] == "CREATE":
2110 if task
["status"] == "SCHEDULED":
2111 if task_status_create
:
2112 new_status
= task_status_create
2113 target_update
= "COPY_VIM_INFO"
2115 new_status
, db_vim_info_update
= self
.item2class
[
2117 ].new(ro_task
, task_index
, task_depends
)
2118 # self._create_task(ro_task, task_index, task_depends, db_ro_task_update)
2119 _update_refresh(new_status
)
2122 ro_task
["vim_info"]["refresh_at"]
2123 and now
> ro_task
["vim_info"]["refresh_at"]
2125 new_status
, db_vim_info_update
= self
.item2class
[
2128 _update_refresh(new_status
)
2130 # The refresh is updated to avoid set the value of "refresh_at" to
2131 # default value (next_check_at = now + (24 * 60 * 60)) when status is BUILD,
2132 # because it can happen that in this case the task is never processed
2133 _update_refresh(task
["status"])
2135 except Exception as e
:
2136 new_status
= "FAILED"
2137 db_vim_info_update
= {
2138 "vim_status": "VIM_ERROR",
2139 "vim_details": str(e
),
2143 e
, (NsWorkerException
, vimconn
.VimConnException
)
2146 "Unexpected exception at _delete_task task={}: {}".format(
2153 if db_vim_info_update
:
2154 db_vim_update
= db_vim_info_update
.copy()
2155 db_ro_task_update
.update(
2158 for k
, v
in db_vim_info_update
.items()
2161 ro_task
["vim_info"].update(db_vim_info_update
)
2164 if task_action
== "CREATE":
2165 task_status_create
= new_status
2166 db_ro_task_update
[task_path
] = new_status
2168 if target_update
or db_vim_update
:
2169 if target_update
== "DELETE":
2170 self
._update
_target
(task
, None)
2171 elif target_update
== "COPY_VIM_INFO":
2172 self
._update
_target
(task
, ro_task
["vim_info"])
2174 self
._update
_target
(task
, db_vim_update
)
2176 except Exception as e
:
2178 isinstance(e
, DbException
)
2179 and e
.http_code
== HTTPStatus
.NOT_FOUND
2181 # if the vnfrs or nsrs has been removed from database, this task must be removed
2183 "marking to delete task={}".format(task
["task_id"])
2185 self
.tasks_to_delete
.append(task
)
2188 "Unexpected exception at _update_target task={}: {}".format(
2194 locked_at
= ro_task
["locked_at"]
2198 lock_object
["locked_at"],
2199 lock_object
["locked_at"] + self
.task_locked_time
,
2201 # locked_at contains two times to avoid race condition. In case the lock has been renew, it will
2202 # contain exactly locked_at + self.task_locked_time
2203 LockRenew
.remove_lock_object(lock_object
)
2206 "_id": ro_task
["_id"],
2207 "to_check_at": ro_task
["to_check_at"],
2208 "locked_at": locked_at
,
2210 # modify own task. Try filtering by to_next_check. For race condition if to_check_at has been modified,
2211 # outside this task (by ro_nbi) do not update it
2212 db_ro_task_update
["locked_by"] = None
2213 # locked_at converted to int only for debugging. When has not decimals it means it has been unlocked
2214 db_ro_task_update
["locked_at"] = int(now
- self
.task_locked_time
)
2215 db_ro_task_update
["modified_at"] = now
2216 db_ro_task_update
["to_check_at"] = next_check_at
2219 # Log RO tasks only when loglevel is DEBUG
2220 if self.logger.getEffectiveLevel() == logging.DEBUG:
2221 db_ro_task_update_log = db_ro_task_update.copy()
2222 db_ro_task_update_log["_id"] = q_filter["_id"]
2223 self._log_ro_task(None, db_ro_task_update_log, None, "TASK_WF", "SET_TASK")
2226 if not self
.db
.set_one(
2228 update_dict
=db_ro_task_update
,
2230 fail_on_empty
=False,
2232 del db_ro_task_update
["to_check_at"]
2233 del q_filter
["to_check_at"]
2235 # Log RO tasks only when loglevel is DEBUG
2236 if self.logger.getEffectiveLevel() == logging.DEBUG:
2239 db_ro_task_update_log,
2242 "SET_TASK " + str(q_filter),
2248 update_dict
=db_ro_task_update
,
2251 except DbException
as e
:
2253 "ro_task={} Error updating database {}".format(ro_task_id
, e
)
2255 except Exception as e
:
2257 "Error executing ro_task={}: {}".format(ro_task_id
, e
), exc_info
=True
2260 def _update_target(self
, task
, ro_vim_item_update
):
2261 table
, _
, temp
= task
["target_record"].partition(":")
2262 _id
, _
, path_vim_status
= temp
.partition(":")
2263 path_item
= path_vim_status
[: path_vim_status
.rfind(".")]
2264 path_item
= path_item
[: path_item
.rfind(".")]
2265 # path_vim_status: dot separated list targeting vim information, e.g. "vdur.10.vim_info.vim:id"
2266 # path_item: dot separated list targeting record information, e.g. "vdur.10"
2268 if ro_vim_item_update
:
2270 path_vim_status
+ "." + k
: v
2271 for k
, v
in ro_vim_item_update
.items()
2273 in ("vim_id", "vim_details", "vim_name", "vim_status", "interfaces")
2276 if path_vim_status
.startswith("vdur."):
2277 # for backward compatibility, add vdur.name apart from vdur.vim_name
2278 if ro_vim_item_update
.get("vim_name"):
2279 update_dict
[path_item
+ ".name"] = ro_vim_item_update
["vim_name"]
2281 # for backward compatibility, add vdur.vim-id apart from vdur.vim_id
2282 if ro_vim_item_update
.get("vim_id"):
2283 update_dict
[path_item
+ ".vim-id"] = ro_vim_item_update
["vim_id"]
2285 # update general status
2286 if ro_vim_item_update
.get("vim_status"):
2287 update_dict
[path_item
+ ".status"] = ro_vim_item_update
[
2291 if ro_vim_item_update
.get("interfaces"):
2292 path_interfaces
= path_item
+ ".interfaces"
2294 for i
, iface
in enumerate(ro_vim_item_update
.get("interfaces")):
2298 path_interfaces
+ ".{}.".format(i
) + k
: v
2299 for k
, v
in iface
.items()
2300 if k
in ("vlan", "compute_node", "pci")
2304 # put ip_address and mac_address with ip-address and mac-address
2305 if iface
.get("ip_address"):
2307 path_interfaces
+ ".{}.".format(i
) + "ip-address"
2308 ] = iface
["ip_address"]
2310 if iface
.get("mac_address"):
2312 path_interfaces
+ ".{}.".format(i
) + "mac-address"
2313 ] = iface
["mac_address"]
2315 if iface
.get("mgmt_vnf_interface") and iface
.get("ip_address"):
2316 update_dict
["ip-address"] = iface
.get("ip_address").split(
2320 if iface
.get("mgmt_vdu_interface") and iface
.get("ip_address"):
2321 update_dict
[path_item
+ ".ip-address"] = iface
.get(
2325 self
.db
.set_one(table
, q_filter
={"_id": _id
}, update_dict
=update_dict
)
2327 update_dict
= {path_item
+ ".status": "DELETED"}
2330 q_filter
={"_id": _id
},
2331 update_dict
=update_dict
,
2332 unset
={path_vim_status
: None},
2335 def _process_delete_db_tasks(self
):
2337 Delete task from database because vnfrs or nsrs or both have been deleted
2338 :return: None. Uses and modify self.tasks_to_delete
2340 while self
.tasks_to_delete
:
2341 task
= self
.tasks_to_delete
[0]
2342 vnfrs_deleted
= None
2343 nsr_id
= task
["nsr_id"]
2345 if task
["target_record"].startswith("vnfrs:"):
2346 # check if nsrs is present
2347 if self
.db
.get_one("nsrs", {"_id": nsr_id
}, fail_on_empty
=False):
2348 vnfrs_deleted
= task
["target_record"].split(":")[1]
2351 self
.delete_db_tasks(self
.db
, nsr_id
, vnfrs_deleted
)
2352 except Exception as e
:
2354 "Error deleting task={}: {}".format(task
["task_id"], e
)
2356 self
.tasks_to_delete
.pop(0)
2359 def delete_db_tasks(db
, nsr_id
, vnfrs_deleted
):
2361 Static method because it is called from osm_ng_ro.ns
2362 :param db: instance of database to use
2363 :param nsr_id: affected nsrs id
2364 :param vnfrs_deleted: only tasks with this vnfr id. If None, all affected by nsr_id
2365 :return: None, exception is fails
2368 for retry
in range(retries
):
2369 ro_tasks
= db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2373 for ro_task
in ro_tasks
:
2375 to_delete_ro_task
= True
2377 for index
, task
in enumerate(ro_task
["tasks"]):
2380 elif (not vnfrs_deleted
and task
["nsr_id"] == nsr_id
) or (
2382 and task
["target_record"].startswith("vnfrs:" + vnfrs_deleted
)
2384 db_update
["tasks.{}".format(index
)] = None
2386 # used by other nsr, ro_task cannot be deleted
2387 to_delete_ro_task
= False
2389 # delete or update if nobody has changed ro_task meanwhile. Used modified_at for known if changed
2390 if to_delete_ro_task
:
2394 "_id": ro_task
["_id"],
2395 "modified_at": ro_task
["modified_at"],
2397 fail_on_empty
=False,
2401 db_update
["modified_at"] = now
2405 "_id": ro_task
["_id"],
2406 "modified_at": ro_task
["modified_at"],
2408 update_dict
=db_update
,
2409 fail_on_empty
=False,
2415 raise NsWorkerException("Exceeded {} retries".format(retries
))
2419 self
.logger
.info("Starting")
2421 # step 1: get commands from queue
2423 if self
.vim_targets
:
2424 task
= self
.task_queue
.get(block
=False)
2427 self
.logger
.debug("enters in idle state")
2429 task
= self
.task_queue
.get(block
=True)
2432 if task
[0] == "terminate":
2434 elif task
[0] == "load_vim":
2435 self
.logger
.info("order to load vim {}".format(task
[1]))
2436 self
._load
_vim
(task
[1])
2437 elif task
[0] == "unload_vim":
2438 self
.logger
.info("order to unload vim {}".format(task
[1]))
2439 self
._unload
_vim
(task
[1])
2440 elif task
[0] == "reload_vim":
2441 self
._reload
_vim
(task
[1])
2442 elif task
[0] == "check_vim":
2443 self
.logger
.info("order to check vim {}".format(task
[1]))
2444 self
._check
_vim
(task
[1])
2446 except Exception as e
:
2447 if isinstance(e
, queue
.Empty
):
2450 self
.logger
.critical(
2451 "Error processing task: {}".format(e
), exc_info
=True
2454 # step 2: process pending_tasks, delete not needed tasks
2456 if self
.tasks_to_delete
:
2457 self
._process
_delete
_db
_tasks
()
2460 # Log RO tasks only when loglevel is DEBUG
2461 if self.logger.getEffectiveLevel() == logging.DEBUG:
2462 _ = self._get_db_all_tasks()
2464 ro_task
= self
._get
_db
_task
()
2466 self
._process
_pending
_tasks
(ro_task
)
2470 except Exception as e
:
2471 self
.logger
.critical(
2472 "Unexpected exception at run: " + str(e
), exc_info
=True
2475 self
.logger
.info("Finishing")