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 pkg_resources
import iter_entry_points
33 # from osm_common import dbmongo, dbmemory, fslocal, fsmongo, msglocal, msgkafka, version as common_version
34 from osm_common
.dbbase
import DbException
35 # from osm_common.fsbase import FsException
36 # from osm_common.msgbase import MsgException
37 from osm_ro_plugin
.vim_dummy
import VimDummyConnector
38 from osm_ro_plugin
.sdn_dummy
import SdnDummyConnector
39 from osm_ro_plugin
import vimconn
, sdnconn
40 from copy
import deepcopy
41 from unittest
.mock
import Mock
42 from http
import HTTPStatus
44 from shutil
import rmtree
46 __author__
= "Alfonso Tierno"
47 __date__
= "$28-Sep-2017 12:07:15$"
50 def deep_get(target_dict
, *args
, **kwargs
):
52 Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
53 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
54 :param target_dict: dictionary to be read
55 :param args: list of keys to read from target_dict
56 :param kwargs: only can contain default=value to return if key is not present in the nested dictionary
57 :return: The wanted value if exist, None or default otherwise
60 if not isinstance(target_dict
, dict) or key
not in target_dict
:
61 return kwargs
.get("default")
62 target_dict
= target_dict
[key
]
66 class NsWorkerException(Exception):
70 class FailingConnector
:
71 def __init__(self
, error_msg
):
72 self
.error_msg
= error_msg
73 for method
in dir(vimconn
.VimConnector
):
75 setattr(self
, method
, Mock(side_effect
=vimconn
.VimConnException(error_msg
)))
76 for method
in dir(sdnconn
.SdnConnectorBase
):
78 setattr(self
, method
, Mock(side_effect
=sdnconn
.SdnConnectorError(error_msg
)))
81 class NsWorkerExceptionNotFound(NsWorkerException
):
85 class VimInteractionBase
:
86 """ Base class to call VIM/SDN for creating, deleting and refresh networks, VMs, flavors, ...
87 It implements methods that does nothing and return ok"""
88 def __init__(self
, db
, my_vims
, db_vims
, logger
):
91 self
.my_vims
= my_vims
92 self
.db_vims
= db_vims
94 def new(self
, ro_task
, task_index
, task_depends
):
97 def refresh(self
, ro_task
):
98 """skip calling VIM to get image, flavor status. Assumes ok"""
99 if ro_task
["vim_info"]["vim_status"] == "VIM_ERROR":
103 def delete(self
, ro_task
, task_index
):
104 """skip calling VIM to delete image. Assumes ok"""
107 def exec(self
, ro_task
, task_index
, task_depends
):
108 return "DONE", None, None
111 class VimInteractionNet(VimInteractionBase
):
113 def new(self
, ro_task
, task_index
, task_depends
):
115 task
= ro_task
["tasks"][task_index
]
116 task_id
= task
["task_id"]
119 target_vim
= self
.my_vims
[ro_task
["target_id"]]
122 if task
.get("find_params"):
123 # if management, get configuration of VIM
124 if task
["find_params"].get("filter_dict"):
125 vim_filter
= task
["find_params"]["filter_dict"]
126 elif task
["find_params"].get("mgmt"): # mamagement network
127 if deep_get(self
.db_vims
[ro_task
["target_id"]], "config", "management_network_id"):
128 vim_filter
= {"id": self
.db_vims
[ro_task
["target_id"]]["config"]["management_network_id"]}
129 elif deep_get(self
.db_vims
[ro_task
["target_id"]], "config", "management_network_name"):
130 vim_filter
= {"name": self
.db_vims
[ro_task
["target_id"]]["config"]["management_network_name"]}
132 vim_filter
= {"name": task
["find_params"]["name"]}
134 raise NsWorkerExceptionNotFound("Invalid find_params for new_net {}".format(task
["find_params"]))
136 vim_nets
= target_vim
.get_network_list(vim_filter
)
137 if not vim_nets
and not task
.get("params"):
138 raise NsWorkerExceptionNotFound("Network not found with this criteria: '{}'".format(
139 task
.get("find_params")))
140 elif len(vim_nets
) > 1:
141 raise NsWorkerException(
142 "More than one network found with this criteria: '{}'".format(task
["find_params"]))
144 vim_net_id
= vim_nets
[0]["id"]
147 params
= task
["params"]
148 vim_net_id
, created_items
= target_vim
.new_network(**params
)
151 ro_vim_item_update
= {"vim_id": vim_net_id
,
152 "vim_status": "BUILD",
154 "created_items": created_items
,
157 "task={} {} new-net={} created={}".format(task_id
, ro_task
["target_id"], vim_net_id
, created
))
158 return "BUILD", ro_vim_item_update
159 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
160 self
.logger
.error("task={} vim={} new-net: {}".format(task_id
, ro_task
["target_id"], e
))
161 ro_vim_item_update
= {"vim_status": "VIM_ERROR",
163 "vim_details": str(e
)}
164 return "FAILED", ro_vim_item_update
166 def refresh(self
, ro_task
):
167 """Call VIM to get network status"""
168 ro_task_id
= ro_task
["_id"]
169 target_vim
= self
.my_vims
[ro_task
["target_id"]]
171 vim_id
= ro_task
["vim_info"]["vim_id"]
172 net_to_refresh_list
= [vim_id
]
174 vim_dict
= target_vim
.refresh_nets_status(net_to_refresh_list
)
175 vim_info
= vim_dict
[vim_id
]
176 if vim_info
["status"] == "ACTIVE":
178 elif vim_info
["status"] == "BUILD":
179 task_status
= "BUILD"
181 task_status
= "FAILED"
182 except vimconn
.VimConnException
as e
:
183 # Mark all tasks at VIM_ERROR status
184 self
.logger
.error("ro_task={} vim={} get-net={}: {}".format(ro_task_id
, ro_task
["target_id"], vim_id
, e
))
185 vim_info
= {"status": "VIM_ERROR", "error_msg": str(e
)}
186 task_status
= "FAILED"
188 ro_vim_item_update
= {}
189 if ro_task
["vim_info"]["vim_status"] != vim_info
["status"]:
190 ro_vim_item_update
["vim_status"] = vim_info
["status"]
191 if ro_task
["vim_info"]["vim_name"] != vim_info
.get("name"):
192 ro_vim_item_update
["vim_name"] = vim_info
.get("name")
193 if vim_info
["status"] in ("ERROR", "VIM_ERROR"):
194 if ro_task
["vim_info"]["vim_details"] != vim_info
.get("error_msg"):
195 ro_vim_item_update
["vim_details"] = vim_info
.get("error_msg")
196 elif vim_info
["status"] == "DELETED":
197 ro_vim_item_update
["vim_id"] = None
198 ro_vim_item_update
["vim_details"] = "Deleted externally"
200 if ro_task
["vim_info"]["vim_details"] != vim_info
["vim_info"]:
201 ro_vim_item_update
["vim_details"] = vim_info
["vim_info"]
202 if ro_vim_item_update
:
203 self
.logger
.debug("ro_task={} {} get-net={}: status={} {}".format(
204 ro_task_id
, ro_task
["target_id"], vim_id
, ro_vim_item_update
.get("vim_status"),
205 ro_vim_item_update
.get("vim_details") if ro_vim_item_update
.get("vim_status") != "ACTIVE" else ''))
206 return task_status
, ro_vim_item_update
208 def delete(self
, ro_task
, task_index
):
209 task
= ro_task
["tasks"][task_index
]
210 task_id
= task
["task_id"]
211 net_vim_id
= ro_task
["vim_info"]["vim_id"]
212 ro_vim_item_update_ok
= {"vim_status": "DELETED",
214 "vim_details": "DELETED",
217 if net_vim_id
or ro_task
["vim_info"]["created_items"]:
218 target_vim
= self
.my_vims
[ro_task
["target_id"]]
219 target_vim
.delete_network(net_vim_id
, ro_task
["vim_info"]["created_items"])
221 except vimconn
.VimConnNotFoundException
:
222 ro_vim_item_update_ok
["vim_details"] = "already deleted"
224 except vimconn
.VimConnException
as e
:
225 self
.logger
.error("ro_task={} vim={} del-net={}: {}".format(ro_task
["_id"], ro_task
["target_id"],
227 ro_vim_item_update
= {"vim_status": "VIM_ERROR",
228 "vim_details": "Error while deleting: {}".format(e
)}
229 return "FAILED", ro_vim_item_update
231 self
.logger
.debug("task={} {} del-net={} {}".format(task_id
, ro_task
["target_id"], net_vim_id
,
232 ro_vim_item_update_ok
.get("vim_details", "")))
233 return "DONE", ro_vim_item_update_ok
236 class VimInteractionVdu(VimInteractionBase
):
237 max_retries_inject_ssh_key
= 20 # 20 times
238 time_retries_inject_ssh_key
= 30 # wevery 30 seconds
240 def new(self
, ro_task
, task_index
, task_depends
):
241 task
= ro_task
["tasks"][task_index
]
242 task_id
= task
["task_id"]
245 target_vim
= self
.my_vims
[ro_task
["target_id"]]
248 params
= task
["params"]
249 params_copy
= deepcopy(params
)
250 net_list
= params_copy
["net_list"]
252 if "net_id" in net
and net
["net_id"].startswith("TASK-"): # change task_id into network_id
253 network_id
= task_depends
[net
["net_id"]]
255 raise NsWorkerException("Cannot create VM because depends on a network not created or found "
256 "for {}".format(net
["net_id"]))
257 net
["net_id"] = network_id
258 if params_copy
["image_id"].startswith("TASK-"):
259 params_copy
["image_id"] = task_depends
[params_copy
["image_id"]]
260 if params_copy
["flavor_id"].startswith("TASK-"):
261 params_copy
["flavor_id"] = task_depends
[params_copy
["flavor_id"]]
263 vim_vm_id
, created_items
= target_vim
.new_vminstance(**params_copy
)
264 interfaces
= [iface
["vim_id"] for iface
in params_copy
["net_list"]]
266 ro_vim_item_update
= {"vim_id": vim_vm_id
,
267 "vim_status": "BUILD",
269 "created_items": created_items
,
271 "interfaces_vim_ids": interfaces
,
275 "task={} {} new-vm={} created={}".format(task_id
, ro_task
["target_id"], vim_vm_id
, created
))
276 return "BUILD", ro_vim_item_update
277 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
278 self
.logger
.error("task={} {} new-vm: {}".format(task_id
, ro_task
["target_id"], e
))
279 ro_vim_item_update
= {"vim_status": "VIM_ERROR",
281 "vim_details": str(e
)}
282 return "FAILED", ro_vim_item_update
284 def delete(self
, ro_task
, task_index
):
285 task
= ro_task
["tasks"][task_index
]
286 task_id
= task
["task_id"]
287 vm_vim_id
= ro_task
["vim_info"]["vim_id"]
288 ro_vim_item_update_ok
= {"vim_status": "DELETED",
290 "vim_details": "DELETED",
293 if vm_vim_id
or ro_task
["vim_info"]["created_items"]:
294 target_vim
= self
.my_vims
[ro_task
["target_id"]]
295 target_vim
.delete_vminstance(vm_vim_id
, ro_task
["vim_info"]["created_items"])
297 except vimconn
.VimConnNotFoundException
:
298 ro_vim_item_update_ok
["vim_details"] = "already deleted"
300 except vimconn
.VimConnException
as e
:
301 self
.logger
.error("ro_task={} vim={} del-vm={}: {}".format(ro_task
["_id"], ro_task
["target_id"],
303 ro_vim_item_update
= {"vim_status": "VIM_ERROR",
304 "vim_details": "Error while deleting: {}".format(e
)}
305 return "FAILED", ro_vim_item_update
307 self
.logger
.debug("task={} {} del-vm={} {}".format(task_id
, ro_task
["target_id"], vm_vim_id
,
308 ro_vim_item_update_ok
.get("vim_details", "")))
309 return "DONE", ro_vim_item_update_ok
311 def refresh(self
, ro_task
):
312 """Call VIM to get vm status"""
313 ro_task_id
= ro_task
["_id"]
314 target_vim
= self
.my_vims
[ro_task
["target_id"]]
316 vim_id
= ro_task
["vim_info"]["vim_id"]
319 vm_to_refresh_list
= [vim_id
]
321 vim_dict
= target_vim
.refresh_vms_status(vm_to_refresh_list
)
322 vim_info
= vim_dict
[vim_id
]
323 if vim_info
["status"] == "ACTIVE":
325 elif vim_info
["status"] == "BUILD":
326 task_status
= "BUILD"
328 task_status
= "FAILED"
329 # try to load and parse vim_information
331 vim_info_info
= yaml
.safe_load(vim_info
["vim_info"])
332 if vim_info_info
.get("name"):
333 vim_info
["name"] = vim_info_info
["name"]
336 except vimconn
.VimConnException
as e
:
337 # Mark all tasks at VIM_ERROR status
338 self
.logger
.error("ro_task={} vim={} get-vm={}: {}".format(ro_task_id
, ro_task
["target_id"], vim_id
, e
))
339 vim_info
= {"status": "VIM_ERROR", "error_msg": str(e
)}
340 task_status
= "FAILED"
342 ro_vim_item_update
= {}
343 # Interfaces cannot be present if e.g. VM is not present, that is status=DELETED
345 if vim_info
.get("interfaces"):
346 for vim_iface_id
in ro_task
["vim_info"]["interfaces_vim_ids"]:
347 iface
= next((iface
for iface
in vim_info
["interfaces"] if vim_iface_id
== iface
["vim_interface_id"]),
350 # iface.pop("vim_info", None)
351 vim_interfaces
.append(iface
)
353 task_create
= next(t
for t
in ro_task
["tasks"] if t
and t
["action"] == "CREATE" and t
["status"] != "FINISHED")
354 if vim_interfaces
and task_create
.get("mgmt_vnf_interface") is not None:
355 vim_interfaces
[task_create
["mgmt_vnf_interface"]]["mgmt_vnf_interface"] = True
356 mgmt_vdu_iface
= task_create
.get("mgmt_vdu_interface", task_create
.get("mgmt_vnf_interface", 0))
358 vim_interfaces
[mgmt_vdu_iface
]["mgmt_vdu_interface"] = True
360 if ro_task
["vim_info"]["interfaces"] != vim_interfaces
:
361 ro_vim_item_update
["interfaces"] = vim_interfaces
362 if ro_task
["vim_info"]["vim_status"] != vim_info
["status"]:
363 ro_vim_item_update
["vim_status"] = vim_info
["status"]
364 if ro_task
["vim_info"]["vim_name"] != vim_info
.get("name"):
365 ro_vim_item_update
["vim_name"] = vim_info
.get("name")
366 if vim_info
["status"] in ("ERROR", "VIM_ERROR"):
367 if ro_task
["vim_info"]["vim_details"] != vim_info
.get("error_msg"):
368 ro_vim_item_update
["vim_details"] = vim_info
.get("error_msg")
369 elif vim_info
["status"] == "DELETED":
370 ro_vim_item_update
["vim_id"] = None
371 ro_vim_item_update
["vim_details"] = "Deleted externally"
373 if ro_task
["vim_info"]["vim_details"] != vim_info
["vim_info"]:
374 ro_vim_item_update
["vim_details"] = vim_info
["vim_info"]
375 if ro_vim_item_update
:
376 self
.logger
.debug("ro_task={} {} get-vm={}: status={} {}".format(
377 ro_task_id
, ro_task
["target_id"], vim_id
, ro_vim_item_update
.get("vim_status"),
378 ro_vim_item_update
.get("vim_details") if ro_vim_item_update
.get("vim_status") != "ACTIVE" else ''))
379 return task_status
, ro_vim_item_update
381 def exec(self
, ro_task
, task_index
, task_depends
):
382 task
= ro_task
["tasks"][task_index
]
383 task_id
= task
["task_id"]
384 target_vim
= self
.my_vims
[ro_task
["target_id"]]
385 db_task_update
= {"retries": 0}
386 retries
= task
.get("retries", 0)
388 params
= task
["params"]
389 params_copy
= deepcopy(params
)
390 params_copy
["ro_key"] = self
.db
.decrypt(params_copy
.pop("private_key"),
391 params_copy
.pop("schema_version"), params_copy
.pop("salt"))
392 params_copy
["ip_addr"] = params_copy
.pop("ip_address")
393 target_vim
.inject_user_key(**params_copy
)
395 "task={} {} action-vm=inject_key".format(task_id
, ro_task
["target_id"]))
396 return "DONE", None, db_task_update
, # params_copy["key"]
397 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
399 if retries
< self
.max_retries_inject_ssh_key
:
400 return "BUILD", None, {"retries": retries
, "next_retry": self
.time_retries_inject_ssh_key
}
401 self
.logger
.error("task={} {} inject-ssh-key: {}".format(task_id
, ro_task
["target_id"], e
))
402 ro_vim_item_update
= {"vim_details": str(e
)}
403 return "FAILED", ro_vim_item_update
, db_task_update
406 class VimInteractionImage(VimInteractionBase
):
408 def new(self
, ro_task
, task_index
, task_depends
):
409 task
= ro_task
["tasks"][task_index
]
410 task_id
= task
["task_id"]
413 target_vim
= self
.my_vims
[ro_task
["target_id"]]
416 if task
.get("find_params"):
417 vim_images
= target_vim
.get_image_list(**task
["find_params"])
419 raise NsWorkerExceptionNotFound("Image not found with this criteria: '{}'".format(
420 task
["find_params"]))
421 elif len(vim_images
) > 1:
422 raise NsWorkerException(
423 "More than one network found with this criteria: '{}'".format(task
["find_params"]))
425 vim_image_id
= vim_images
[0]["id"]
427 ro_vim_item_update
= {"vim_id": vim_image_id
,
428 "vim_status": "DONE",
430 "created_items": created_items
,
433 "task={} {} new-image={} created={}".format(task_id
, ro_task
["target_id"], vim_image_id
, created
))
434 return "DONE", ro_vim_item_update
435 except (NsWorkerException
, vimconn
.VimConnException
) as e
:
436 self
.logger
.error("task={} {} new-image: {}".format(task_id
, ro_task
["target_id"], e
))
437 ro_vim_item_update
= {"vim_status": "VIM_ERROR",
439 "vim_details": str(e
)}
440 return "FAILED", ro_vim_item_update
443 class VimInteractionFlavor(VimInteractionBase
):
445 def delete(self
, ro_task
, task_index
):
446 task
= ro_task
["tasks"][task_index
]
447 task_id
= task
["task_id"]
448 flavor_vim_id
= ro_task
["vim_info"]["vim_id"]
449 ro_vim_item_update_ok
= {"vim_status": "DELETED",
451 "vim_details": "DELETED",
455 target_vim
= self
.my_vims
[ro_task
["target_id"]]
456 target_vim
.delete_flavor(flavor_vim_id
)
458 except vimconn
.VimConnNotFoundException
:
459 ro_vim_item_update_ok
["vim_details"] = "already deleted"
461 except vimconn
.VimConnException
as e
:
462 self
.logger
.error("ro_task={} vim={} del-flavor={}: {}".format(
463 ro_task
["_id"], ro_task
["target_id"], flavor_vim_id
, e
))
464 ro_vim_item_update
= {"vim_status": "VIM_ERROR",
465 "vim_details": "Error while deleting: {}".format(e
)}
466 return "FAILED", ro_vim_item_update
468 self
.logger
.debug("task={} {} del-flavor={} {}".format(
469 task_id
, ro_task
["target_id"], flavor_vim_id
, ro_vim_item_update_ok
.get("vim_details", "")))
470 return "DONE", ro_vim_item_update_ok
472 def new(self
, ro_task
, task_index
, task_depends
):
473 task
= ro_task
["tasks"][task_index
]
474 task_id
= task
["task_id"]
477 target_vim
= self
.my_vims
[ro_task
["target_id"]]
481 if task
.get("find_params"):
483 flavor_data
= task
["find_params"]["flavor_data"]
484 vim_flavor_id
= target_vim
.get_flavor_id_from_data(flavor_data
)
485 except vimconn
.VimConnNotFoundException
:
488 if not vim_flavor_id
and task
.get("params"):
490 flavor_data
= task
["params"]["flavor_data"]
491 vim_flavor_id
= target_vim
.new_flavor(flavor_data
)
494 ro_vim_item_update
= {"vim_id": vim_flavor_id
,
495 "vim_status": "DONE",
497 "created_items": created_items
,
500 "task={} {} new-flavor={} created={}".format(task_id
, ro_task
["target_id"], vim_flavor_id
, created
))
501 return "DONE", ro_vim_item_update
502 except (vimconn
.VimConnException
, NsWorkerException
) as e
:
503 self
.logger
.error("task={} vim={} new-flavor: {}".format(task_id
, ro_task
["target_id"], e
))
504 ro_vim_item_update
= {"vim_status": "VIM_ERROR",
506 "vim_details": str(e
)}
507 return "FAILED", ro_vim_item_update
510 class VimInteractionSdnNet(VimInteractionBase
):
513 def _match_pci(port_pci
, mapping
):
515 Check if port_pci matches with mapping
516 mapping can have brackets to indicate that several chars are accepted. e.g
517 pci '0000:af:10.1' matches with '0000:af:1[01].[1357]'
518 :param port_pci: text
519 :param mapping: text, can contain brackets to indicate several chars are available
520 :return: True if matches, False otherwise
522 if not port_pci
or not mapping
:
524 if port_pci
== mapping
:
530 bracket_start
= mapping
.find("[", mapping_index
)
531 if bracket_start
== -1:
533 bracket_end
= mapping
.find("]", bracket_start
)
534 if bracket_end
== -1:
536 length
= bracket_start
- mapping_index
537 if length
and port_pci
[pci_index
:pci_index
+ length
] != mapping
[mapping_index
:bracket_start
]:
539 if port_pci
[pci_index
+ length
] not in mapping
[bracket_start
+1:bracket_end
]:
541 pci_index
+= length
+ 1
542 mapping_index
= bracket_end
+ 1
544 if port_pci
[pci_index
:] != mapping
[mapping_index
:]:
548 def _get_interfaces(self
, vlds_to_connect
, vim_account_id
):
550 :param vlds_to_connect: list with format vnfrs:<id>:vld.<vld_id> or nsrs:<id>:vld.<vld_id>
551 :param vim_account_id:
555 for vld
in vlds_to_connect
:
556 table
, _
, db_id
= vld
.partition(":")
557 db_id
, _
, vld
= db_id
.partition(":")
558 _
, _
, vld_id
= vld
.partition(".")
560 q_filter
= {"vim-account-id": vim_account_id
, "_id": db_id
}
561 iface_key
= "vnf-vld-id"
562 else: # table == "nsrs"
563 q_filter
= {"vim-account-id": vim_account_id
, "nsr-id-ref": db_id
}
564 iface_key
= "ns-vld-id"
565 db_vnfrs
= self
.db
.get_list("vnfrs", q_filter
=q_filter
)
566 for db_vnfr
in db_vnfrs
:
567 for vdu_index
, vdur
in enumerate(db_vnfr
.get("vdur", ())):
568 for iface_index
, interface
in enumerate(vdur
["interfaces"]):
569 if interface
.get(iface_key
) == vld_id
and \
570 interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
572 interface_
= interface
.copy()
573 interface_
["id"] = "vnfrs:{}:vdu.{}.interfaces.{}".format(db_vnfr
["_id"], vdu_index
,
575 if vdur
.get("status") == "ERROR":
576 interface_
["status"] = "ERROR"
577 interfaces
.append(interface_
)
580 def refresh(self
, ro_task
):
581 # look for task create
582 task_create_index
, _
= next(i_t
for i_t
in enumerate(ro_task
["tasks"])
583 if i_t
[1] and i_t
[1]["action"] == "CREATE" and i_t
[1]["status"] != "FINISHED")
585 return self
.new(ro_task
, task_create_index
, None)
587 def new(self
, ro_task
, task_index
, task_depends
):
589 task
= ro_task
["tasks"][task_index
]
590 task_id
= task
["task_id"]
591 target_vim
= self
.my_vims
[ro_task
["target_id"]]
593 sdn_net_id
= ro_task
["vim_info"]["vim_id"]
595 created_items
= ro_task
["vim_info"].get("created_items")
596 connected_ports
= ro_task
["vim_info"].get("connected_ports", [])
597 new_connected_ports
= []
598 last_update
= ro_task
["vim_info"].get("last_update", 0)
599 sdn_status
= ro_task
["vim_info"].get("vim_status", "BUILD") or "BUILD"
601 created
= ro_task
["vim_info"].get("created", False)
606 params
= task
["params"]
607 vlds_to_connect
= params
["vlds"]
608 associated_vim
= params
["target_vim"]
609 additional_ports
= params
.get("sdn-ports") or () # external additional ports
610 _
, _
, vim_account_id
= associated_vim
.partition(":")
613 if associated_vim
not in self
.db_vims
:
614 self
.db_vims
[associated_vim
] = self
.db
.get_one("vim_accounts", {"_id": vim_account_id
})
615 db_vim
= self
.db_vims
[associated_vim
]
617 # look for ports to connect
618 ports
= self
._get
_interfaces
(vlds_to_connect
, vim_account_id
)
622 pending_ports
= error_ports
= 0
624 sdn_need_update
= False
626 vlan_used
= port
.get("vlan") or vlan_used
627 # TODO. Do not connect if already done
628 if not port
.get("compute_node") or not port
.get("pci"):
629 if port
.get("status") == "ERROR":
635 compute_node_mappings
= next((c
for c
in db_vim
["config"].get("sdn-port-mapping", ())
636 if c
and c
["compute_node"] == port
["compute_node"]), None)
637 if compute_node_mappings
:
638 # process port_mapping pci of type 0000:af:1[01].[1357]
639 pmap
= next((p
for p
in compute_node_mappings
["ports"]
640 if self
._match
_pci
(port
["pci"], p
.get("pci"))), None)
642 if not db_vim
["config"].get("mapping_not_needed"):
643 error_list
.append("Port mapping not found for compute_node={} pci={}".format(
644 port
["compute_node"], port
["pci"]))
648 service_endpoint_id
= "{}:{}".format(port
["compute_node"], port
["pci"])
650 "service_endpoint_id": pmap
.get("service_endpoint_id") or service_endpoint_id
,
651 "service_endpoint_encapsulation_type": "dot1q" if port
["type"] == "SR-IOV" else None,
652 "service_endpoint_encapsulation_info": {
653 "vlan": port
.get("vlan"),
654 "mac": port
.get("mac_address"),
655 "device_id": pmap
.get("device_id") or port
["compute_node"], # device_id
656 "device_interface_id": pmap
.get("device_interface_id") or port
["pci"],
657 "switch_dpid": pmap
.get("switch_id") or pmap
.get("switch_dpid"),
658 "switch_port": pmap
.get("switch_port"),
659 "service_mapping_info": pmap
.get("service_mapping_info"),
664 # if port["modified_at"] > last_update:
665 # sdn_need_update = True
666 new_connected_ports
.append(port
["id"]) # TODO
667 sdn_ports
.append(new_port
)
670 error_list
.append("{} interfaces have not been created as VDU is on ERROR status".format(error_ports
))
672 # connect external ports
673 for index
, additional_port
in enumerate(additional_ports
):
674 additional_port_id
= additional_port
.get("service_endpoint_id") or "external-{}".format(index
)
676 "service_endpoint_id": additional_port_id
,
677 "service_endpoint_encapsulation_type": additional_port
.get("service_endpoint_encapsulation_type",
679 "service_endpoint_encapsulation_info": {
680 "vlan": additional_port
.get("vlan") or vlan_used
,
681 "mac": additional_port
.get("mac_address"),
682 "device_id": additional_port
.get("device_id"),
683 "device_interface_id": additional_port
.get("device_interface_id"),
684 "switch_dpid": additional_port
.get("switch_dpid") or additional_port
.get("switch_id"),
685 "switch_port": additional_port
.get("switch_port"),
686 "service_mapping_info": additional_port
.get("service_mapping_info"),
688 new_connected_ports
.append(additional_port_id
)
690 # if there are more ports to connect or they have been modified, call create/update
693 sdn_info
= "; ".join(error_list
)
694 elif set(connected_ports
) != set(new_connected_ports
) or sdn_need_update
:
695 last_update
= time
.time()
697 if len(sdn_ports
) < 2:
698 sdn_status
= "ACTIVE"
699 if not pending_ports
:
700 self
.logger
.debug("task={} {} new-sdn-net done, less than 2 ports".
701 format(task_id
, ro_task
["target_id"]))
703 net_type
= params
.get("type") or "ELAN"
704 sdn_net_id
, created_items
= target_vim
.create_connectivity_service(
707 self
.logger
.debug("task={} {} new-sdn-net={} created={}".
708 format(task_id
, ro_task
["target_id"], sdn_net_id
, created
))
710 created_items
= target_vim
.edit_connectivity_service(
711 sdn_net_id
, conn_info
=created_items
, connection_points
=sdn_ports
)
713 self
.logger
.debug("task={} {} update-sdn-net={} created={}".
714 format(task_id
, ro_task
["target_id"], sdn_net_id
, created
))
715 connected_ports
= new_connected_ports
717 wim_status_dict
= target_vim
.get_connectivity_service_status(sdn_net_id
, conn_info
=created_items
)
718 sdn_status
= wim_status_dict
["sdn_status"]
719 if wim_status_dict
.get("sdn_info"):
720 sdn_info
= str(wim_status_dict
.get("sdn_info")) or ""
721 if wim_status_dict
.get("error_msg"):
722 sdn_info
= wim_status_dict
.get("error_msg") or ""
725 if sdn_status
!= "ERROR":
726 sdn_info
= "Waiting for getting interfaces location from VIM. Obtained '{}' of {}".format(
727 len(ports
)-pending_ports
, len(ports
))
728 if sdn_status
== "ACTIVE":
731 ro_vim_item_update
= {"vim_id": sdn_net_id
,
732 "vim_status": sdn_status
,
734 "created_items": created_items
,
735 "connected_ports": connected_ports
,
736 "vim_details": sdn_info
,
737 "last_update": last_update
}
738 return sdn_status
, ro_vim_item_update
739 except Exception as e
:
740 self
.logger
.error("task={} vim={} new-net: {}".format(task_id
, ro_task
["target_id"], e
),
741 exc_info
=not isinstance(e
, (sdnconn
.SdnConnectorError
, vimconn
.VimConnException
)))
742 ro_vim_item_update
= {"vim_status": "VIM_ERROR",
744 "vim_details": str(e
)}
745 return "FAILED", ro_vim_item_update
747 def delete(self
, ro_task
, task_index
):
748 task
= ro_task
["tasks"][task_index
]
749 task_id
= task
["task_id"]
750 sdn_vim_id
= ro_task
["vim_info"].get("vim_id")
751 ro_vim_item_update_ok
= {"vim_status": "DELETED",
753 "vim_details": "DELETED",
757 target_vim
= self
.my_vims
[ro_task
["target_id"]]
758 target_vim
.delete_connectivity_service(sdn_vim_id
, ro_task
["vim_info"].get("created_items"))
760 except Exception as e
:
761 if isinstance(e
, sdnconn
.SdnConnectorError
) and e
.http_code
== HTTPStatus
.NOT_FOUND
.value
:
762 ro_vim_item_update_ok
["vim_details"] = "already deleted"
764 self
.logger
.error("ro_task={} vim={} del-sdn-net={}: {}".format(ro_task
["_id"], ro_task
["target_id"],
766 exc_info
=not isinstance(e
, (sdnconn
.SdnConnectorError
, vimconn
.VimConnException
)))
767 ro_vim_item_update
= {"vim_status": "VIM_ERROR",
768 "vim_details": "Error while deleting: {}".format(e
)}
769 return "FAILED", ro_vim_item_update
771 self
.logger
.debug("task={} {} del-sdn-net={} {}".format(task_id
, ro_task
["target_id"], sdn_vim_id
,
772 ro_vim_item_update_ok
.get("vim_details", "")))
773 return "DONE", ro_vim_item_update_ok
776 class NsWorker(threading
.Thread
):
777 REFRESH_BUILD
= 5 # 5 seconds
778 REFRESH_ACTIVE
= 60 # 1 minute
780 REFRESH_IMAGE
= 3600 * 10
781 REFRESH_DELETE
= 3600 * 10
783 # TODO delete assigment_lock = Lock()
785 # TODO delete assignment = {}
786 MAX_TIME_LOCKED
= 3600
787 MAX_TIME_VIM_LOCKED
= 120
789 def __init__(self
, worker_index
, config
, plugins
, db
):
792 :param worker_index: thread index
793 :param config: general configuration of RO, among others the process_id with the docker id where it runs
794 :param plugins: global shared dict with the loaded plugins
795 :param db: database class instance to use
797 threading
.Thread
.__init
__(self
)
799 self
.plugins
= plugins
800 self
.plugin_name
= "unknown"
801 self
.logger
= logging
.getLogger('ro.worker{}'.format(worker_index
))
802 self
.worker_index
= worker_index
803 self
.task_queue
= queue
.Queue(self
.QUEUE_SIZE
)
804 self
.my_vims
= {} # targetvim: vimplugin class
805 self
.db_vims
= {} # targetvim: vim information from database
806 self
.vim_targets
= [] # targetvim list
807 self
.my_id
= config
["process_id"] + ":" + str(worker_index
)
810 "net": VimInteractionNet(self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
),
811 "vdu": VimInteractionVdu(self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
),
812 "image": VimInteractionImage(self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
),
813 "flavor": VimInteractionFlavor(self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
),
814 "sdn_net": VimInteractionSdnNet(self
.db
, self
.my_vims
, self
.db_vims
, self
.logger
),
816 self
.time_last_task_processed
= None
817 self
.tasks_to_delete
= [] # lists of tasks to delete because nsrs or vnfrs has been deleted from db
819 def insert_task(self
, task
):
821 self
.task_queue
.put(task
, False)
824 raise NsWorkerException("timeout inserting a task")
827 self
.insert_task("exit")
829 def del_task(self
, task
):
831 if task
["status"] == "SCHEDULED":
832 task
["status"] = "SUPERSEDED"
834 else: # task["status"] == "processing"
835 self
.task_lock
.release()
838 def _process_vim_config(self
, target_id
, db_vim
):
840 Process vim config, creating vim configuration files as ca_cert
841 :param target_id: vim/sdn/wim + id
842 :param db_vim: Vim dictionary obtained from database
843 :return: None. Modifies vim. Creates a folder target_id:worker_index and several files
845 if not db_vim
.get("config"):
849 if db_vim
["config"].get("ca_cert_content"):
850 file_name
= "{}:{}".format(target_id
, self
.worker_index
)
853 except FileExistsError
:
855 file_name
= file_name
+ "/ca_cert"
856 with
open(file_name
, "w") as f
:
857 f
.write(db_vim
["config"]["ca_cert_content"])
858 del db_vim
["config"]["ca_cert_content"]
859 db_vim
["config"]["ca_cert"] = file_name
860 except Exception as e
:
861 raise NsWorkerException("Error writing to file '{}': {}".format(file_name
, e
))
863 def _load_plugin(self
, name
, type="vim"):
864 # type can be vim or sdn
865 if "rovim_dummy" not in self
.plugins
:
866 self
.plugins
["rovim_dummy"] = VimDummyConnector
867 if "rosdn_dummy" not in self
.plugins
:
868 self
.plugins
["rosdn_dummy"] = SdnDummyConnector
869 if name
in self
.plugins
:
870 return self
.plugins
[name
]
872 for v
in iter_entry_points('osm_ro{}.plugins'.format(type), name
):
873 self
.plugins
[name
] = v
.load()
874 except Exception as e
:
875 raise NsWorkerException("Cannot load plugin osm_{}: {}".format(name
, e
))
876 if name
and name
not in self
.plugins
:
877 raise NsWorkerException("Plugin 'osm_{n}' has not been installed".format(n
=name
))
878 return self
.plugins
[name
]
880 def _unload_vim(self
, target_id
):
882 Unload a vim_account. Removes it from self db_vims dictionary, my_vims dictionary and vim_targets list
883 :param target_id: Contains type:_id; where type can be 'vim', ...
887 target
, _
, _id
= target_id
.partition(":")
888 self
.db_vims
.pop(target_id
, None)
889 self
.my_vims
.pop(target_id
, None)
890 self
.vim_targets
.remove(target_id
)
891 rmtree("{}:{}".format(target_id
, self
.worker_index
))
892 except FileNotFoundError
:
893 pass # this is raised by rmtree if folder does not exist
894 except Exception as e
:
895 self
.logger
.error("Cannot unload {}: {}".format(target_id
, e
))
897 def _check_vim(self
, target_id
):
899 Load a VIM/SDN/WIM (if not loaded) and check connectivity, updating database with ENABLE or ERROR
900 :param target_id: Contains type:_id; type can be 'vim', 'sdn' or 'wim'
903 target
, _
, _id
= target_id
.partition(":")
909 loaded
= target_id
in self
.my_vims
910 target_database
= "vim_accounts" if target
== "vim" else "wim_accounts" if target
== "wim" else "sdns"
912 step
= "Getting {} from db".format(target_id
)
913 db_vim
= self
.db
.get_one(target_database
, {"_id": _id
})
914 for op_index
, operation
in enumerate(db_vim
["_admin"].get("operations", ())):
915 if operation
["operationState"] != "PROCESSING":
917 locked_at
= operation
.get("locked_at")
918 if locked_at
is not None and locked_at
>= now
- self
.MAX_TIME_VIM_LOCKED
:
919 # some other thread is doing this operation
922 op_text
= "_admin.operations.{}.".format(op_index
)
923 if not self
.db
.set_one(target_database
,
924 q_filter
={"_id": _id
,
925 op_text
+ "operationState": "PROCESSING",
926 op_text
+ "locked_at": locked_at
928 update_dict
={op_text
+ "locked_at": now
,
929 "admin.current_operation": op_index
},
930 fail_on_empty
=False):
932 unset_dict
[op_text
+ "locked_at"] = None
933 unset_dict
["current_operation"] = None
934 step
= "Loading " + target_id
935 error_text
= self
._load
_vim
(target_id
)
937 step
= "Checking connectivity"
939 self
.my_vims
[target_id
].check_vim_connectivity()
941 self
.my_vims
[target_id
].check_credentials()
942 update_dict
["_admin.operationalState"] = "ENABLED"
943 update_dict
["_admin.detailed-status"] = ""
944 unset_dict
[op_text
+ "detailed-status"] = None
945 update_dict
[op_text
+ "operationState"] = "COMPLETED"
948 except Exception as e
:
949 error_text
= "{}: {}".format(step
, e
)
950 self
.logger
.error("{} for {}: {}".format(step
, target_id
, e
))
953 if update_dict
or unset_dict
:
955 update_dict
[op_text
+ "operationState"] = "FAILED"
956 update_dict
[op_text
+ "detailed-status"] = error_text
957 unset_dict
.pop(op_text
+ "detailed-status", None)
958 update_dict
["_admin.operationalState"] = "ERROR"
959 update_dict
["_admin.detailed-status"] = error_text
961 update_dict
[op_text
+ "statusEnteredTime"] = now
962 self
.db
.set_one(target_database
, q_filter
={"_id": _id
}, update_dict
=update_dict
, unset
=unset_dict
,
965 self
._unload
_vim
(target_id
)
967 def _reload_vim(self
, target_id
):
968 if target_id
in self
.vim_targets
:
969 self
._load
_vim
(target_id
)
971 # if the vim is not loaded, but database information of VIM is cached at self.db_vims,
972 # just remove it to force load again next time it is needed
973 self
.db_vims
.pop(target_id
, None)
975 def _load_vim(self
, target_id
):
977 Load or reload a vim_account, sdn_controller or wim_account.
978 Read content from database, load the plugin if not loaded.
979 In case of error loading the plugin, it load a failing VIM_connector
980 It fills self db_vims dictionary, my_vims dictionary and vim_targets list
981 :param target_id: Contains type:_id; where type can be 'vim', ...
982 :return: None if ok, descriptive text if error
984 target
, _
, _id
= target_id
.partition(":")
985 target_database
= "vim_accounts" if target
== "vim" else "wim_accounts" if target
== "wim" else "sdns"
989 step
= "Getting {}={} from db".format(target
, _id
)
990 # TODO process for wim, sdnc, ...
991 vim
= self
.db
.get_one(target_database
, {"_id": _id
})
993 # if deep_get(vim, "config", "sdn-controller"):
994 # step = "Getting sdn-controller-id='{}' from db".format(vim["config"]["sdn-controller"])
995 # db_sdn = self.db.get_one("sdns", {"_id": vim["config"]["sdn-controller"]})
997 step
= "Decrypting password"
998 schema_version
= vim
.get("schema_version")
999 self
.db
.encrypt_decrypt_fields(vim
, "decrypt", fields
=('password', 'secret'),
1000 schema_version
=schema_version
, salt
=_id
)
1001 self
._process
_vim
_config
(target_id
, vim
)
1003 plugin_name
= "rovim_" + vim
["vim_type"]
1004 step
= "Loading plugin '{}'".format(plugin_name
)
1005 vim_module_conn
= self
._load
_plugin
(plugin_name
)
1006 step
= "Loading {}'".format(target_id
)
1007 self
.my_vims
[target_id
] = vim_module_conn(
1008 uuid
=vim
['_id'], name
=vim
['name'],
1009 tenant_id
=vim
.get('vim_tenant_id'), tenant_name
=vim
.get('vim_tenant_name'),
1010 url
=vim
['vim_url'], url_admin
=None,
1011 user
=vim
['vim_user'], passwd
=vim
['vim_password'],
1012 config
=vim
.get('config') or {}, persistent_info
={}
1015 plugin_name
= "rosdn_" + vim
["type"]
1016 step
= "Loading plugin '{}'".format(plugin_name
)
1017 vim_module_conn
= self
._load
_plugin
(plugin_name
, "sdn")
1018 step
= "Loading {}'".format(target_id
)
1020 wim_config
= wim
.pop("config", {}) or {}
1021 wim
["uuid"] = wim
["_id"]
1022 wim
["wim_url"] = wim
["url"]
1024 wim_config
["dpid"] = wim
.pop("dpid")
1025 if wim
.get("switch_id"):
1026 wim_config
["switch_id"] = wim
.pop("switch_id")
1027 self
.my_vims
[target_id
] = vim_module_conn(wim
, wim
, wim_config
) # wim, wim_account, config
1028 self
.db_vims
[target_id
] = vim
1029 self
.error_status
= None
1030 self
.logger
.info("Connector loaded for {}, plugin={}".format(target_id
, plugin_name
))
1031 except Exception as e
:
1032 self
.logger
.error("Cannot load {} plugin={}: {} {}".format(
1033 target_id
, plugin_name
, step
, e
))
1034 self
.db_vims
[target_id
] = vim
or {}
1035 self
.db_vims
[target_id
] = FailingConnector(str(e
))
1036 error_status
= "{} Error: {}".format(step
, e
)
1039 if target_id
not in self
.vim_targets
:
1040 self
.vim_targets
.append(target_id
)
1042 def _get_db_task(self
):
1044 Read actions from database and reload them at memory. Fill self.refresh_list, pending_list, vim_actions
1048 if not self
.time_last_task_processed
:
1049 self
.time_last_task_processed
= now
1052 locked
= self
.db
.set_one(
1054 q_filter
={"target_id": self
.vim_targets
,
1055 "tasks.status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
1056 "locked_at.lt": now
- self
.MAX_TIME_LOCKED
,
1057 "to_check_at.lt": self
.time_last_task_processed
},
1058 update_dict
={"locked_by": self
.my_id
, "locked_at": now
},
1059 fail_on_empty
=False)
1062 ro_task
= self
.db
.get_one(
1064 q_filter
={"target_id": self
.vim_targets
,
1065 "tasks.status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
1068 if self
.time_last_task_processed
== now
:
1069 self
.time_last_task_processed
= None
1072 self
.time_last_task_processed
= now
1073 # self.time_last_task_processed = min(self.time_last_task_processed + 1000, now)
1075 except DbException
as e
:
1076 self
.logger
.error("Database exception at _get_db_task: {}".format(e
))
1077 except Exception as e
:
1078 self
.logger
.critical("Unexpected exception at _get_db_task: {}".format(e
), exc_info
=True)
1081 def _delete_task(self
, ro_task
, task_index
, task_depends
, db_update
):
1083 Determine if this task need to be done or superseded
1086 my_task
= ro_task
["tasks"][task_index
]
1087 task_id
= my_task
["task_id"]
1088 needed_delete
= ro_task
["vim_info"]["created"] or ro_task
["vim_info"].get("created_items", False)
1089 if my_task
["status"] == "FAILED":
1090 return None, None # TODO need to be retry??
1092 for index
, task
in enumerate(ro_task
["tasks"]):
1093 if index
== task_index
or not task
:
1095 if my_task
["target_record"] == task
["target_record"] and task
["action"] == "CREATE":
1097 db_update
["tasks.{}.status".format(index
)] = task
["status"] = "FINISHED"
1098 elif task
["action"] == "CREATE" and task
["status"] not in ("FINISHED", "SUPERSEDED"):
1099 needed_delete
= False
1101 return self
.item2class
[my_task
["item"]].delete(ro_task
, task_index
)
1103 return "SUPERSEDED", None
1104 except Exception as e
:
1105 if not isinstance(e
, NsWorkerException
):
1106 self
.logger
.critical("Unexpected exception at _delete_task task={}: {}".format(task_id
, e
),
1108 return "FAILED", {"vim_status": "VIM_ERROR", "vim_details": str(e
)}
1110 def _create_task(self
, ro_task
, task_index
, task_depends
, db_update
):
1112 Determine if this task need to create something at VIM
1115 my_task
= ro_task
["tasks"][task_index
]
1116 task_id
= my_task
["task_id"]
1118 if my_task
["status"] == "FAILED":
1119 return None, None # TODO need to be retry??
1120 elif my_task
["status"] == "SCHEDULED":
1121 # check if already created by another task
1122 for index
, task
in enumerate(ro_task
["tasks"]):
1123 if index
== task_index
or not task
:
1125 if task
["action"] == "CREATE" and task
["status"] not in ("SCHEDULED", "FINISHED", "SUPERSEDED"):
1126 return task
["status"], "COPY_VIM_INFO"
1129 task_status
, ro_vim_item_update
= self
.item2class
[my_task
["item"]].new(
1130 ro_task
, task_index
, task_depends
)
1131 # TODO update other CREATE tasks
1132 except Exception as e
:
1133 if not isinstance(e
, NsWorkerException
):
1134 self
.logger
.error("Error executing task={}: {}".format(task_id
, e
), exc_info
=True)
1135 task_status
= "FAILED"
1136 ro_vim_item_update
= {"vim_status": "VIM_ERROR", "vim_details": str(e
)}
1137 # TODO update ro_vim_item_update
1138 return task_status
, ro_vim_item_update
1142 def _get_dependency(self
, task_id
, ro_task
=None, target_id
=None):
1144 Look for dependency task
1145 :param task_id: Can be one of
1146 1. target_vim+blank+task.target_record_id: "(vim|sdn|wim):<id> (vnfrs|nsrs):(vld|vdu|flavor|image).<id>"
1147 2. task.target_record_id: "(vnfrs|nsrs):(vld|vdu|flavor|image).<id>"
1148 3. task.task_id: "<action_id>:number"
1151 :return: database ro_task plus index of task
1153 if task_id
.startswith("vim:") or task_id
.startswith("sdn:") or task_id
.startswith("wim:"):
1154 target_id
, _
, task_id
= task_id
.partition(" ")
1156 if task_id
.startswith("nsrs:") or task_id
.startswith("vnfrs:"):
1157 ro_task_dependency
= self
.db
.get_one(
1159 q_filter
={"target_id": target_id
,
1160 "tasks.target_record_id": task_id
1162 fail_on_empty
=False)
1163 if ro_task_dependency
:
1164 for task_index
, task
in enumerate(ro_task_dependency
["tasks"]):
1165 if task
["target_record_id"] == task_id
:
1166 return ro_task_dependency
, task_index
1170 for task_index
, task
in enumerate(ro_task
["tasks"]):
1171 if task
and task
["task_id"] == task_id
:
1172 return ro_task
, task_index
1173 ro_task_dependency
= self
.db
.get_one(
1175 q_filter
={"tasks.ANYINDEX.task_id": task_id
,
1176 "tasks.ANYINDEX.target_record.ne": None
1178 fail_on_empty
=False)
1179 if ro_task_dependency
:
1180 for task_index
, task
in ro_task_dependency
["tasks"]:
1181 if task
["task_id"] == task_id
:
1182 return ro_task_dependency
, task_index
1183 raise NsWorkerException("Cannot get depending task {}".format(task_id
))
1185 def _process_pending_tasks(self
, ro_task
):
1186 ro_task_id
= ro_task
["_id"]
1188 next_check_at
= now
+ (24*60*60) # one day
1189 db_ro_task_update
= {}
1191 def _update_refresh(new_status
):
1192 # compute next_refresh
1194 nonlocal next_check_at
1195 nonlocal db_ro_task_update
1198 next_refresh
= time
.time()
1199 if task
["item"] in ("image", "flavor"):
1200 next_refresh
+= self
.REFRESH_IMAGE
1201 elif new_status
== "BUILD":
1202 next_refresh
+= self
.REFRESH_BUILD
1203 elif new_status
== "DONE":
1204 next_refresh
+= self
.REFRESH_ACTIVE
1206 next_refresh
+= self
.REFRESH_ERROR
1207 next_check_at
= min(next_check_at
, next_refresh
)
1208 db_ro_task_update
["vim_info.refresh_at"] = next_refresh
1209 ro_task
["vim_info"]["refresh_at"] = next_refresh
1212 # 0 get task_status_create
1213 task_status_create
= None
1214 task_create
= next((t
for t
in ro_task
["tasks"] if t
and t
["action"] == "CREATE" and
1215 t
["status"] in ("BUILD", "DONE")), None)
1217 task_status_create
= task_create
["status"]
1218 # 1. look for tasks in status SCHEDULED, or in status CREATE if action is DONE or BUILD
1219 for task_action
in ("DELETE", "CREATE", "EXEC"):
1220 db_vim_update
= None
1222 for task_index
, task
in enumerate(ro_task
["tasks"]):
1224 continue # task deleted
1225 target_update
= None
1226 if (task_action
in ("DELETE", "EXEC") and task
["status"] not in ("SCHEDULED", "BUILD")) or \
1227 task
["action"] != task_action
or \
1228 (task_action
== "CREATE" and task
["status"] in ("FINISHED", "SUPERSEDED")):
1230 task_path
= "tasks.{}.status".format(task_index
)
1232 db_vim_info_update
= None
1233 if task
["status"] == "SCHEDULED":
1235 # check if tasks that this depends on have been completed
1236 dependency_not_completed
= False
1237 for dependency_task_id
in (task
.get("depends_on") or ()):
1238 dependency_ro_task
, dependency_task_index
= \
1239 self
._get
_dependency
(dependency_task_id
, target_id
=ro_task
["target_id"])
1240 dependency_task
= dependency_ro_task
["tasks"][dependency_task_index
]
1241 if dependency_task
["status"] == "SCHEDULED":
1242 dependency_not_completed
= True
1243 next_check_at
= min(next_check_at
, dependency_ro_task
["to_check_at"])
1245 elif dependency_task
["status"] == "FAILED":
1246 error_text
= "Cannot {} {} because depends on failed {} {} id={}): {}".format(
1247 task
["action"], task
["item"], dependency_task
["action"],
1248 dependency_task
["item"], dependency_task_id
,
1249 dependency_ro_task
["vim_info"].get("vim_details"))
1250 self
.logger
.error("task={} {}".format(task
["task_id"], error_text
))
1251 raise NsWorkerException(error_text
)
1253 task_depends
[dependency_task_id
] = dependency_ro_task
["vim_info"]["vim_id"]
1254 task_depends
["TASK-{}".format(dependency_task_id
)] = \
1255 dependency_ro_task
["vim_info"]["vim_id"]
1256 if dependency_not_completed
:
1257 # TODO set at vim_info.vim_details that it is waiting
1260 if task
["action"] == "DELETE":
1261 new_status
, db_vim_info_update
= self
._delete
_task
(ro_task
, task_index
,
1262 task_depends
, db_ro_task_update
)
1263 new_status
= "FINISHED" if new_status
== "DONE" else new_status
1264 # ^with FINISHED instead of DONE it will not be refreshing
1265 if new_status
in ("FINISHED", "SUPERSEDED"):
1266 target_update
= "DELETE"
1267 elif task
["action"] == "EXEC":
1268 new_status
, db_vim_info_update
, db_task_update
= self
.item2class
[task
["item"]].exec(
1269 ro_task
, task_index
, task_depends
)
1270 new_status
= "FINISHED" if new_status
== "DONE" else new_status
1271 # ^with FINISHED instead of DONE it will not be refreshing
1273 # load into database the modified db_task_update "retries" and "next_retry"
1274 if db_task_update
.get("retries"):
1275 db_ro_task_update
["tasks.{}.retries".format(task_index
)] = db_task_update
["retries"]
1276 next_check_at
= time
.time() + db_task_update
.get("next_retry", 60)
1277 target_update
= None
1278 elif task
["action"] == "CREATE":
1279 if task
["status"] == "SCHEDULED":
1280 if task_status_create
:
1281 new_status
= task_status_create
1282 target_update
= "COPY_VIM_INFO"
1284 new_status
, db_vim_info_update
= \
1285 self
.item2class
[task
["item"]].new(ro_task
, task_index
, task_depends
)
1286 # self._create_task(ro_task, task_index, task_depends, db_ro_task_update)
1287 _update_refresh(new_status
)
1289 if ro_task
["vim_info"]["refresh_at"] and now
> ro_task
["vim_info"]["refresh_at"]:
1290 new_status
, db_vim_info_update
= self
.item2class
[task
["item"]].refresh(ro_task
)
1291 _update_refresh(new_status
)
1292 except Exception as e
:
1293 new_status
= "FAILED"
1294 db_vim_info_update
= {"vim_status": "VIM_ERROR", "vim_details": str(e
)}
1295 if not isinstance(e
, (NsWorkerException
, vimconn
.VimConnException
)):
1296 self
.logger
.error("Unexpected exception at _delete_task task={}: {}".
1297 format(task
["task_id"], e
), exc_info
=True)
1300 if db_vim_info_update
:
1301 db_vim_update
= db_vim_info_update
.copy()
1302 db_ro_task_update
.update({"vim_info." + k
: v
for k
, v
in db_vim_info_update
.items()})
1303 ro_task
["vim_info"].update(db_vim_info_update
)
1306 if task_action
== "CREATE":
1307 task_status_create
= new_status
1308 db_ro_task_update
[task_path
] = new_status
1309 if target_update
or db_vim_update
:
1311 if target_update
== "DELETE":
1312 self
._update
_target
(task
, None)
1313 elif target_update
== "COPY_VIM_INFO":
1314 self
._update
_target
(task
, ro_task
["vim_info"])
1316 self
._update
_target
(task
, db_vim_update
)
1318 except Exception as e
:
1319 if isinstance(e
, DbException
) and e
.http_code
== HTTPStatus
.NOT_FOUND
:
1320 # if the vnfrs or nsrs has been removed from database, this task must be removed
1321 self
.logger
.debug("marking to delete task={}".format(task
["task_id"]))
1322 self
.tasks_to_delete
.append(task
)
1324 self
.logger
.error("Unexpected exception at _update_target task={}: {}".
1325 format(task
["task_id"], e
), exc_info
=True)
1327 q_filter
= {"_id": ro_task
["_id"], "to_check_at": ro_task
["to_check_at"], "locked_at": ro_task
["locked_at"]}
1328 # modify own task. Try filtering by to_next_check. For race condition if to_check_at has been modified,
1329 # outside this task (by ro_nbi) do not update it
1330 db_ro_task_update
["locked_by"] = None
1331 # locked_at converted to int only for debugging. When has not decimals it means it has been unlocked
1332 db_ro_task_update
["locked_at"] = int(now
- self
.MAX_TIME_LOCKED
)
1333 db_ro_task_update
["to_check_at"] = next_check_at
1334 if not self
.db
.set_one("ro_tasks",
1335 update_dict
=db_ro_task_update
,
1337 fail_on_empty
=False):
1338 del db_ro_task_update
["to_check_at"]
1339 del q_filter
["to_check_at"]
1340 self
.db
.set_one("ro_tasks",
1342 update_dict
=db_ro_task_update
,
1344 except DbException
as e
:
1345 self
.logger
.error("ro_task={} Error updating database {}".format(ro_task_id
, e
))
1346 except Exception as e
:
1347 self
.logger
.error("Error executing ro_task={}: {}".format(ro_task_id
, e
), exc_info
=True)
1349 def _update_target(self
, task
, ro_vim_item_update
):
1350 table
, _
, temp
= task
["target_record"].partition(":")
1351 _id
, _
, path_vim_status
= temp
.partition(":")
1352 path_item
= path_vim_status
[:path_vim_status
.rfind(".")]
1353 path_item
= path_item
[:path_item
.rfind(".")]
1354 # path_vim_status: dot separated list targeting vim information, e.g. "vdur.10.vim_info.vim:id"
1355 # path_item: dot separated list targeting record information, e.g. "vdur.10"
1356 if ro_vim_item_update
:
1357 update_dict
= {path_vim_status
+ "." + k
: v
for k
, v
in ro_vim_item_update
.items() if k
in
1358 ('vim_id', 'vim_details', 'vim_name', 'vim_status', 'interfaces')}
1359 if path_vim_status
.startswith("vdur."):
1360 # for backward compatibility, add vdur.name apart from vdur.vim_name
1361 if ro_vim_item_update
.get("vim_name"):
1362 update_dict
[path_item
+ ".name"] = ro_vim_item_update
["vim_name"]
1363 # for backward compatibility, add vdur.vim-id apart from vdur.vim_id
1364 if ro_vim_item_update
.get("vim_id"):
1365 update_dict
[path_item
+ ".vim-id"] = ro_vim_item_update
["vim_id"]
1366 # update general status
1367 if ro_vim_item_update
.get("vim_status"):
1368 update_dict
[path_item
+ ".status"] = ro_vim_item_update
["vim_status"]
1369 if ro_vim_item_update
.get("interfaces"):
1370 path_interfaces
= path_item
+ ".interfaces"
1371 for i
, iface
in enumerate(ro_vim_item_update
.get("interfaces")):
1373 update_dict
.update({path_interfaces
+ ".{}.".format(i
) + k
: v
for k
, v
in iface
.items() if
1374 k
in ('vlan', 'compute_node', 'pci')})
1375 # put ip_address and mac_address with ip-address and mac-address
1376 if iface
.get('ip_address'):
1377 update_dict
[path_interfaces
+ ".{}.".format(i
) + "ip-address"] = iface
['ip_address']
1378 if iface
.get('mac_address'):
1379 update_dict
[path_interfaces
+ ".{}.".format(i
) + "mac-address"] = iface
['mac_address']
1380 if iface
.get("mgmt_vnf_interface") and iface
.get("ip_address"):
1381 update_dict
["ip-address"] = iface
.get("ip_address").split(";")[0]
1382 if iface
.get("mgmt_vdu_interface") and iface
.get("ip_address"):
1383 update_dict
[path_item
+ ".ip-address"] = iface
.get("ip_address").split(";")[0]
1385 self
.db
.set_one(table
, q_filter
={"_id": _id
}, update_dict
=update_dict
)
1387 update_dict
= {path_item
+ ".status": "DELETED"}
1388 self
.db
.set_one(table
, q_filter
={"_id": _id
}, update_dict
=update_dict
, unset
={path_vim_status
: None})
1390 def _process_delete_db_tasks(self
):
1392 Delete task from database because vnfrs or nsrs or both have been deleted
1393 :return: None. Uses and modify self.tasks_to_delete
1395 while self
.tasks_to_delete
:
1396 task
= self
.tasks_to_delete
[0]
1397 vnfrs_deleted
= None
1398 nsr_id
= task
["nsr_id"]
1399 if task
["target_record"].startswith("vnfrs:"):
1400 # check if nsrs is present
1401 if self
.db
.get_one("nsrs", {"_id": nsr_id
}, fail_on_empty
=False):
1402 vnfrs_deleted
= task
["target_record"].split(":")[1]
1404 self
.delete_db_tasks(self
.db
, nsr_id
, vnfrs_deleted
)
1405 except Exception as e
:
1406 self
.logger
.error("Error deleting task={}: {}".format(task
["task_id"], e
))
1407 self
.tasks_to_delete
.pop(0)
1410 def delete_db_tasks(db
, nsr_id
, vnfrs_deleted
):
1412 Static method because it is called from osm_ng_ro.ns
1413 :param db: instance of database to use
1414 :param nsr_id: affected nsrs id
1415 :param vnfrs_deleted: only tasks with this vnfr id. If None, all affected by nsr_id
1416 :return: None, exception is fails
1419 for retry
in range(retries
):
1420 ro_tasks
= db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
1423 for ro_task
in ro_tasks
:
1425 to_delete_ro_task
= True
1426 for index
, task
in enumerate(ro_task
["tasks"]):
1429 elif (not vnfrs_deleted
and task
["nsr_id"] == nsr_id
) or \
1430 (vnfrs_deleted
and task
["target_record"].startswith("vnfrs:"+vnfrs_deleted
)):
1431 db_update
["tasks.{}".format(index
)] = None
1433 to_delete_ro_task
= False # used by other nsr, ro_task cannot be deleted
1434 # delete or update if nobody has changed ro_task meanwhile. Used modified_at for known if changed
1435 if to_delete_ro_task
:
1436 if not db
.del_one("ro_tasks",
1437 q_filter
={"_id": ro_task
["_id"], "modified_at": ro_task
["modified_at"]},
1438 fail_on_empty
=False):
1441 db_update
["modified_at"] = now
1442 if not db
.set_one("ro_tasks",
1443 q_filter
={"_id": ro_task
["_id"], "modified_at": ro_task
["modified_at"]},
1444 update_dict
=db_update
,
1445 fail_on_empty
=False):
1450 raise NsWorkerException("Exceeded {} retries".format(retries
))
1454 self
.logger
.debug("Starting")
1456 # step 1: get commands from queue
1458 task
= self
.task_queue
.get(block
=False if self
.my_vims
else True)
1459 if task
[0] == "terminate":
1461 elif task
[0] == "load_vim":
1462 self
._load
_vim
(task
[1])
1463 elif task
[0] == "unload_vim":
1464 self
._unload
_vim
(task
[1])
1465 elif task
[0] == "reload_vim":
1466 self
._reload
_vim
(task
[1])
1467 elif task
[0] == "check_vim":
1468 self
._check
_vim
(task
[1])
1470 except Exception as e
:
1471 if isinstance(e
, queue
.Empty
):
1474 self
.logger
.critical("Error processing task: {}".format(e
), exc_info
=True)
1476 # step 2: process pending_tasks, delete not needed tasks
1478 if self
.tasks_to_delete
:
1479 self
._process
_delete
_db
_tasks
()
1481 ro_task
= self
._get
_db
_task
()
1483 self
._process
_pending
_tasks
(ro_task
)
1487 except Exception as e
:
1488 self
.logger
.critical("Unexpected exception at run: " + str(e
), exc_info
=True)
1490 self
.logger
.debug("Finishing")