1 # -*- coding: utf-8 -*-
4 # Copyright 2018 Telefonica S.A.
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 from collections
import OrderedDict
28 from osm_common
.fsbase
import FsException
29 from osm_lcm
.data_utils
.database
.database
import Database
30 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
32 from zipfile
import ZipFile
, BadZipfile
34 # from osm_common.dbbase import DbException
36 __author__
= "Alfonso Tierno"
39 class LcmException(Exception):
43 class LcmExceptionExit(LcmException
):
48 """utility for compare dot separate versions. Fills with zeros to proper number comparison
49 package version will be something like 4.0.1.post11+gb3f024d.dirty-1. Where 4.0.1 is the git tag, postXX is the
50 number of commits from this tag, and +XXXXXXX is the git commit short id. Total length is 16 with until 999 commits
53 for point
in v
.split("."):
54 point
, _
, _
= point
.partition("+")
55 point
, _
, _
= point
.partition("-")
56 filled
.append(point
.zfill(20))
60 def deep_get(target_dict
, key_list
, default_value
=None):
62 Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
63 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
64 :param target_dict: dictionary to be read
65 :param key_list: list of keys to read from target_dict
66 :param default_value: value to return if key is not present in the nested dictionary
67 :return: The wanted value if exist, None otherwise
70 if not isinstance(target_dict
, dict) or key
not in target_dict
:
72 target_dict
= target_dict
[key
]
76 def get_iterable(in_dict
, in_key
):
78 Similar to <dict>.get(), but if value is None, False, ..., An empty tuple is returned instead
79 :param in_dict: a dictionary
80 :param in_key: the key to look for at in_dict
81 :return: in_dict[in_var] or () if it is None or not present
83 if not in_dict
.get(in_key
):
85 return in_dict
[in_key
]
88 def check_juju_bundle_existence(vnfd
: dict) -> str:
89 """Checks the existence of juju-bundle in the descriptor
92 vnfd: Descriptor as a dictionary
95 Juju bundle if dictionary has juju-bundle else None
101 for kdu
in vnfd
.get("kdu", []):
102 return kdu
.get("juju-bundle", None)
105 def get_charm_artifact_path(base_folder
, charm_name
, charm_type
, revision
=str()) -> str:
106 """Finds the charm artifact paths
109 base_folder: Main folder which will be looked up for charm
110 charm_name: Charm name
111 charm_type: Type of charm native_charm, lxc_proxy_charm or k8s_proxy_charm
112 revision: vnf package revision number if there is
120 extension
= ":" + str(revision
)
122 if base_folder
.get("pkg-dir"):
123 artifact_path
= "{}/{}/{}/{}".format(
124 base_folder
["folder"].split(":")[0] + extension
,
125 base_folder
["pkg-dir"],
127 if charm_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
133 # For SOL004 packages
134 artifact_path
= "{}/Scripts/{}/{}".format(
135 base_folder
["folder"].split(":")[0] + extension
,
137 if charm_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
145 def populate_dict(target_dict
, key_list
, value
):
147 Update target_dict creating nested dictionaries with the key_list. Last key_list item is asigned the value.
148 Example target_dict={K: J}; key_list=[a,b,c]; target_dict will be {K: J, a: {b: {c: value}}}
149 :param target_dict: dictionary to be changed
150 :param key_list: list of keys to insert at target_dict
154 for key
in key_list
[0:-1]:
155 if key
not in target_dict
:
156 target_dict
[key
] = {}
157 target_dict
= target_dict
[key
]
158 target_dict
[key_list
[-1]] = value
161 def get_ee_id_parts(ee_id
):
163 Parses ee_id stored at database that can be either 'version:namespace.helm_id' or only
164 namespace.helm_id for backward compatibility
165 If exists helm version can be helm-v3 or helm (helm-v2 old version)
167 version
, _
, part_id
= ee_id
.rpartition(":")
168 namespace
, _
, helm_id
= part_id
.rpartition(".")
169 return version
, namespace
, helm_id
172 def vld_to_ro_ip_profile(source_data
):
176 if "v4" in source_data
.get("ip-version", "ipv4")
178 "subnet_address": source_data
.get("cidr")
179 or source_data
.get("subnet-address"),
180 "gateway_address": source_data
.get("gateway-ip")
181 or source_data
.get("gateway-address"),
182 "dns_address": ";".join(
183 [v
["address"] for v
in source_data
["dns-server"] if v
.get("address")]
185 if source_data
.get("dns-server")
187 "dhcp_enabled": source_data
.get("dhcp-params", {}).get("enabled", False)
188 or source_data
.get("dhcp-enabled", False),
189 "dhcp_start_address": source_data
["dhcp-params"].get("start-address")
190 if source_data
.get("dhcp-params")
192 "dhcp_count": source_data
["dhcp-params"].get("count")
193 if source_data
.get("dhcp-params")
195 "ipv6_address_mode": source_data
["ipv6-address-mode"]
196 if "ipv6-address-mode" in source_data
202 def __init__(self
, msg
, logger
):
205 :param db: database connection
207 self
.db
= Database().instance
.db
209 self
.fs
= Filesystem().instance
.fs
212 def update_db_2(self
, item
, _id
, _desc
):
214 Updates database with _desc information. If success _desc is cleared
215 :param item: collection
216 :param _id: the _id to use in the query filter
217 :param _desc: dictionary with the content to update. Keys are dot separated keys for
218 :return: None. Exception is raised on error
223 _desc
["_admin.modified"] = now
224 self
.db
.set_one(item
, {"_id": _id
}, _desc
)
226 # except DbException as e:
227 # self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
230 def calculate_charm_hash(zipped_file
):
231 """Calculate the hash of charm files which ends with .charm
234 zipped_file (str): Existing charm package full path
237 hex digest (str): The hash of the charm file
239 filehash
= hashlib
.sha256()
240 with
open(zipped_file
, mode
="rb") as file:
241 contents
= file.read()
242 filehash
.update(contents
)
243 return filehash
.hexdigest()
246 def compare_charm_hash(current_charm
, target_charm
):
247 """Compare the existing charm and the target charm if the charms
248 are given as zip files ends with .charm
251 current_charm (str): Existing charm package full path
252 target_charm (str): Target charm package full path
255 True/False (bool): if charm has changed it returns True
257 return LcmBase
.calculate_charm_hash(
259 ) != LcmBase
.calculate_charm_hash(target_charm
)
262 def compare_charmdir_hash(current_charm_dir
, target_charm_dir
):
263 """Compare the existing charm and the target charm if the charms
264 are given as directories
267 current_charm_dir (str): Existing charm package directory path
268 target_charm_dir (str): Target charm package directory path
271 True/False (bool): if charm has changed it returns True
273 return checksumdir
.dirhash(current_charm_dir
) != checksumdir
.dirhash(
277 def check_charm_hash_changed(
278 self
, current_charm_path
: str, target_charm_path
: str
280 """Find the target charm has changed or not by checking the hash of
281 old and new charm packages
284 current_charm_path (str): Existing charm package artifact path
285 target_charm_path (str): Target charm package artifact path
288 True/False (bool): if charm has changed it returns True
292 # Check if the charm artifacts are available
293 current_charm
= self
.fs
.path
+ current_charm_path
294 target_charm
= self
.fs
.path
+ target_charm_path
296 if os
.path
.exists(current_charm
) and os
.path
.exists(target_charm
):
297 # Compare the hash of .charm files
298 if current_charm
.endswith(".charm"):
299 return LcmBase
.compare_charm_hash(current_charm
, target_charm
)
301 # Compare the hash of charm folders
302 return LcmBase
.compare_charmdir_hash(current_charm
, target_charm
)
306 "Charm artifact {} does not exist in the VNF Package".format(
307 self
.fs
.path
+ target_charm_path
310 except (IOError, OSError, TypeError) as error
:
311 self
.logger
.debug(traceback
.format_exc())
312 self
.logger
.error(f
"{error} occured while checking the charm hashes")
313 raise LcmException(error
)
316 def get_charm_name(charm_metadata_file
: str) -> str:
317 """Get the charm name from metadata file.
320 charm_metadata_file (str): charm metadata file full path
323 charm_name (str): charm name
326 # Read charm metadata.yaml to get the charm name
327 with
open(charm_metadata_file
, "r") as metadata_file
:
328 content
= yaml
.safe_load(metadata_file
)
329 charm_name
= content
["name"]
330 return str(charm_name
)
333 self
, nsd_package_path
: str, nsd_package_name
: str, charm_folder_name
: str
335 """Get the full path of charm folder.
338 nsd_package_path (str): NSD package full path
339 nsd_package_name (str): NSD package name
340 charm_folder_name (str): folder name
343 charm_path (str): charm folder full path
355 def _get_charm_metadata_file(
357 charm_folder_name
: str,
358 nsd_package_path
: str,
359 nsd_package_name
: str,
360 charm_path
: str = None,
362 """Get the path of charm metadata file.
365 charm_folder_name (str): folder name
366 nsd_package_path (str): NSD package full path
367 nsd_package_name (str): NSD package name
368 charm_path (str): Charm full path
371 charm_metadata_file_path (str): charm metadata file full path
374 # Locate the charm metadata.yaml
375 if charm_folder_name
.endswith(".charm"):
382 + charm_folder_name
.replace(".charm", "")
384 # Extract .charm to extract path
385 with
ZipFile(charm_path
, "r") as zipfile
:
386 zipfile
.extractall(extract_path
)
387 return extract_path
+ "/metadata.yaml"
389 return charm_path
+ "/metadata.yaml"
391 def find_charm_name(self
, db_nsr
: dict, charm_folder_name
: str) -> str:
392 """Get the charm name from metadata.yaml of charm package.
395 db_nsr (dict): NS record as a dictionary
396 charm_folder_name (str): charm folder name
399 charm_name (str): charm name
402 if not charm_folder_name
:
403 raise LcmException("charm_folder_name should be provided.")
405 # Find nsd_package details: path, name
406 revision
= db_nsr
.get("revision", "")
408 # Get the NSD package path
410 nsd_package_path
= db_nsr
["nsd-id"] + ":" + str(revision
)
411 db_nsd
= self
.db
.get_one("nsds_revisions", {"_id": nsd_package_path
})
414 nsd_package_path
= db_nsr
["nsd-id"]
416 db_nsd
= self
.db
.get_one("nsds", {"_id": nsd_package_path
})
418 # Get the NSD package name
419 nsd_package_name
= db_nsd
["_admin"]["storage"]["pkg-dir"]
421 # Remove the existing nsd package and sync from FsMongo
422 shutil
.rmtree(self
.fs
.path
+ nsd_package_path
, ignore_errors
=True)
423 self
.fs
.sync(from_path
=nsd_package_path
)
426 charm_path
= self
._get
_charm
_path
(
427 nsd_package_path
, nsd_package_name
, charm_folder_name
430 # Find charm metadata file full path
431 charm_metadata_file
= self
._get
_charm
_metadata
_file
(
432 charm_folder_name
, nsd_package_path
, nsd_package_name
, charm_path
436 return self
.get_charm_name(charm_metadata_file
)
447 self
.logger
.debug(traceback
.format_exc())
448 self
.logger
.error(f
"{error} occured while getting the charm name")
449 raise LcmException(error
)
451 def get_vca_info(self
, ee_item
, db_nsr
, get_charm_name
: bool):
452 vca_name
= charm_name
= vca_type
= None
453 if ee_item
.get("juju"):
454 vca_name
= ee_item
["juju"].get("charm")
456 charm_name
= self
.find_charm_name(db_nsr
, str(vca_name
))
459 if ee_item
["juju"].get("charm") is not None
462 if ee_item
["juju"].get("cloud") == "k8s":
463 vca_type
= "k8s_proxy_charm"
464 elif ee_item
["juju"].get("proxy") is False:
465 vca_type
= "native_charm"
466 elif ee_item
.get("helm-chart"):
467 vca_name
= ee_item
["helm-chart"]
469 return vca_name
, charm_name
, vca_type
472 class TaskRegistry(LcmBase
):
474 Implements a registry of task needed for later cancelation, look for related tasks that must be completed before
475 etc. It stores a four level dict
476 First level is the topic, ns, vim_account, sdn
477 Second level is the _id
478 Third level is the operation id
479 Fourth level is a descriptive name, the value is the task class
481 The HA (High-Availability) methods are used when more than one LCM instance is running.
482 To register the current task in the external DB, use LcmBase as base class, to be able
483 to reuse LcmBase.update_db_2()
484 The DB registry uses the following fields to distinguish a task:
485 - op_type: operation type ("nslcmops" or "nsilcmops")
486 - op_id: operation ID
487 - worker: the worker ID for this process
490 # NS/NSI: "services" VIM/WIM/SDN: "accounts"
491 topic_service_list
= ["ns", "nsi"]
492 topic_account_list
= ["vim", "wim", "sdn", "k8scluster", "vca", "k8srepo"]
494 # Map topic to InstanceID
495 topic2instid_dict
= {"ns": "nsInstanceId", "nsi": "netsliceInstanceId"}
497 # Map topic to DB table name
498 topic2dbtable_dict
= {
501 "vim": "vim_accounts",
502 "wim": "wim_accounts",
504 "k8scluster": "k8sclusters",
506 "k8srepo": "k8srepos",
509 def __init__(self
, worker_id
=None, logger
=None):
510 self
.task_registry
= {
520 self
.worker_id
= worker_id
521 self
.db
= Database().instance
.db
524 def register(self
, topic
, _id
, op_id
, task_name
, task
):
527 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
528 :param _id: _id of the related item
529 :param op_id: id of the operation of the related item
530 :param task_name: Task descriptive name, as create, instantiate, terminate. Must be unique in this op_id
531 :param task: Task class
534 if _id
not in self
.task_registry
[topic
]:
535 self
.task_registry
[topic
][_id
] = OrderedDict()
536 if op_id
not in self
.task_registry
[topic
][_id
]:
537 self
.task_registry
[topic
][_id
][op_id
] = {task_name
: task
}
539 self
.task_registry
[topic
][_id
][op_id
][task_name
] = task
540 # print("registering task", topic, _id, op_id, task_name, task)
542 def remove(self
, topic
, _id
, op_id
, task_name
=None):
544 When task is ended, it should be removed. It ignores missing tasks. It also removes tasks done with this _id
545 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
546 :param _id: _id of the related item
547 :param op_id: id of the operation of the related item
548 :param task_name: Task descriptive name. If none it deletes all tasks with same _id and op_id
551 if not self
.task_registry
[topic
].get(_id
):
554 self
.task_registry
[topic
][_id
].pop(op_id
, None)
555 elif self
.task_registry
[topic
][_id
].get(op_id
):
556 self
.task_registry
[topic
][_id
][op_id
].pop(task_name
, None)
559 for op_id_
in list(self
.task_registry
[topic
][_id
]):
560 for name
, task
in self
.task_registry
[topic
][_id
][op_id_
].items():
564 del self
.task_registry
[topic
][_id
][op_id_
]
565 if not self
.task_registry
[topic
][_id
]:
566 del self
.task_registry
[topic
][_id
]
568 def lookfor_related(self
, topic
, _id
, my_op_id
=None):
571 if _id
not in self
.task_registry
[topic
]:
572 return "", task_name_list
573 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
575 if my_op_id
== op_id
:
576 my_op_id
= None # so that the next task is taken
579 for task_name
, task
in self
.task_registry
[topic
][_id
][op_id
].items():
581 task_list
.append(task
)
582 task_name_list
.append(task_name
)
584 return ", ".join(task_name_list
), task_list
586 def cancel(self
, topic
, _id
, target_op_id
=None, target_task_name
=None):
588 Cancel all active tasks of a concrete ns, nsi, vim_account, sdn identified for _id. If op_id is supplied only
589 this is cancelled, and the same with task_name
590 :return: cancelled task to be awaited if needed
592 if not self
.task_registry
[topic
].get(_id
):
594 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
595 if target_op_id
and target_op_id
!= op_id
:
597 for task_name
, task
in list(self
.task_registry
[topic
][_id
][op_id
].items()):
598 if target_task_name
and target_task_name
!= task_name
:
604 # self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, op_id, task_name))
607 def _is_service_type_HA(self
, topic
):
608 return topic
in self
.topic_service_list
610 # Is topic VIM/WIM/SDN?
611 def _is_account_type_HA(self
, topic
):
612 return topic
in self
.topic_account_list
614 # Input: op_id, example: 'abc123def:3' Output: account_id='abc123def', op_index=3
615 def _get_account_and_op_HA(self
, op_id
):
618 account_id
, _
, op_index
= op_id
.rpartition(":")
619 if not account_id
or not op_index
.isdigit():
621 return account_id
, op_index
623 # Get '_id' for any topic and operation
624 def _get_instance_id_HA(self
, topic
, op_type
, op_id
):
626 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
629 # NS/NSI: Use op_id as '_id'
630 elif self
._is
_service
_type
_HA
(topic
):
632 # VIM/SDN/WIM/K8SCLUSTER: Split op_id to get Account ID and Operation Index, use Account ID as '_id'
633 elif self
._is
_account
_type
_HA
(topic
):
634 _id
, _
= self
._get
_account
_and
_op
_HA
(op_id
)
637 # Set DB _filter for querying any related process state
638 def _get_waitfor_filter_HA(self
, db_lcmop
, topic
, op_type
, op_id
):
640 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
641 # In this special case, the timestamp is ignored
643 _filter
= {"operationState": "PROCESSING"}
644 # Otherwise, get 'startTime' timestamp for this operation
647 if self
._is
_service
_type
_HA
(topic
):
649 starttime_this_op
= db_lcmop
.get("startTime")
650 instance_id_label
= self
.topic2instid_dict
.get(topic
)
651 instance_id
= db_lcmop
.get(instance_id_label
)
653 instance_id_label
: instance_id
,
654 "operationState": "PROCESSING",
655 "startTime.lt": starttime_this_op
,
656 "_admin.modified.gt": now
657 - 2 * 3600, # ignore if tow hours of inactivity
659 # VIM/WIM/SDN/K8scluster
660 elif self
._is
_account
_type
_HA
(topic
):
661 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
662 _ops
= db_lcmop
["_admin"]["operations"]
663 _this_op
= _ops
[int(op_index
)]
664 starttime_this_op
= _this_op
.get("startTime", None)
666 "operationState": "PROCESSING",
667 "startTime.lt": starttime_this_op
,
671 # Get DB params for any topic and operation
672 def _get_dbparams_for_lock_HA(self
, topic
, op_type
, op_id
):
676 if self
._is
_service
_type
_HA
(topic
):
677 q_filter
= {"_id": op_id
, "_admin.worker": None}
678 update_dict
= {"_admin.worker": self
.worker_id
}
680 elif self
._is
_account
_type
_HA
(topic
):
681 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
684 if op_type
== "create":
685 # Creating a VIM/WIM/SDN account implies setting '_admin.current_operation' = 0
689 "_admin.operations.{}.worker".format(op_index
): None,
692 "_admin.operations.{}.worker".format(op_index
): self
.worker_id
,
693 "_admin.current_operation": op_index
,
695 return q_filter
, update_dict
697 def lock_HA(self
, topic
, op_type
, op_id
):
699 Lock a task, if possible, to indicate to the HA system that
700 the task will be executed in this LCM instance.
701 :param topic: Can be "ns", "nsi", "vim", "wim", or "sdn"
702 :param op_type: Operation type, can be "nslcmops", "nsilcmops", "create", "edit", "delete"
703 :param op_id: NS, NSI: Operation ID VIM,WIM,SDN: Account ID + ':' + Operation Index
705 True=lock was successful => execute the task (not registered by any other LCM instance)
706 False=lock failed => do NOT execute the task (already registered by another LCM instance)
708 HA tasks and backward compatibility:
709 If topic is "account type" (VIM/WIM/SDN) and op_id is None, 'op_id' was not provided by NBI.
710 This means that the running NBI instance does not support HA.
711 In such a case this method should always return True, to always execute
712 the task in this instance of LCM, without querying the DB.
715 # Backward compatibility for VIM/WIM/SDN/k8scluster without op_id
716 if self
._is
_account
_type
_HA
(topic
) and op_id
is None:
719 # Try to lock this task
720 db_table_name
= self
.topic2dbtable_dict
[topic
]
721 q_filter
, update_dict
= self
._get
_dbparams
_for
_lock
_HA
(topic
, op_type
, op_id
)
722 db_lock_task
= self
.db
.set_one(
725 update_dict
=update_dict
,
728 if db_lock_task
is None:
730 "Task {} operation={} already locked by another worker".format(
736 # Set 'detailed-status' to 'In progress' for VIM/WIM/SDN operations
737 if self
._is
_account
_type
_HA
(topic
):
738 detailed_status
= "In progress"
739 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
740 q_filter
= {"_id": account_id
}
742 "_admin.operations.{}.detailed-status".format(
749 update_dict
=update_dict
,
754 def unlock_HA(self
, topic
, op_type
, op_id
, operationState
, detailed_status
):
756 Register a task, done when finished a VIM/WIM/SDN 'create' operation.
757 :param topic: Can be "vim", "wim", or "sdn"
758 :param op_type: Operation type, can be "create", "edit", "delete"
759 :param op_id: Account ID + ':' + Operation Index
763 # Backward compatibility
764 if not self
._is
_account
_type
_HA
(topic
) or not op_id
:
767 # Get Account ID and Operation Index
768 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
769 db_table_name
= self
.topic2dbtable_dict
[topic
]
771 # If this is a 'delete' operation, the account may have been deleted (SUCCESS) or may still exist (FAILED)
772 # If the account exist, register the HA task.
773 # Update DB for HA tasks
774 q_filter
= {"_id": account_id
}
776 "_admin.operations.{}.operationState".format(op_index
): operationState
,
777 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
778 "_admin.operations.{}.worker".format(op_index
): None,
779 "_admin.current_operation": None,
784 update_dict
=update_dict
,
789 async def waitfor_related_HA(self
, topic
, op_type
, op_id
=None):
791 Wait for any pending related HA tasks
794 # Backward compatibility
796 self
._is
_service
_type
_HA
(topic
) or self
._is
_account
_type
_HA
(topic
)
797 ) and (op_id
is None):
801 db_table_name
= self
.topic2dbtable_dict
.get(topic
)
804 _id
= self
._get
_instance
_id
_HA
(topic
, op_type
, op_id
)
805 _filter
= {"_id": _id
}
806 db_lcmop
= self
.db
.get_one(db_table_name
, _filter
, fail_on_empty
=False)
810 # Set DB _filter for querying any related process state
811 _filter
= self
._get
_waitfor
_filter
_HA
(db_lcmop
, topic
, op_type
, op_id
)
813 # For HA, get list of tasks from DB instead of from dictionary (in-memory) variable.
814 timeout_wait_for_task
= (
815 3600 # Max time (seconds) to wait for a related task to finish
817 # interval_wait_for_task = 30 # A too long polling interval slows things down considerably
818 interval_wait_for_task
= 10 # Interval in seconds for polling related tasks
819 time_left
= timeout_wait_for_task
820 old_num_related_tasks
= 0
822 # Get related tasks (operations within the same instance as this) which are
823 # still running (operationState='PROCESSING') and which were started before this task.
824 # In the case of op_type='ANY', get any related tasks with operationState='PROCESSING', ignore timestamps.
825 db_waitfor_related_task
= self
.db
.get_list(db_table_name
, q_filter
=_filter
)
826 new_num_related_tasks
= len(db_waitfor_related_task
)
827 # If there are no related tasks, there is nothing to wait for, so return.
828 if not new_num_related_tasks
:
830 # If number of pending related tasks have changed,
831 # update the 'detailed-status' field and log the change.
832 # Do NOT update the 'detailed-status' for SDNC-associated-to-VIM operations ('ANY').
833 if (op_type
!= "ANY") and (new_num_related_tasks
!= old_num_related_tasks
):
834 step
= "Waiting for {} related tasks to be completed.".format(
835 new_num_related_tasks
838 q_filter
= {"_id": _id
}
840 if self
._is
_service
_type
_HA
(topic
):
842 "detailed-status": step
,
843 "queuePosition": new_num_related_tasks
,
846 elif self
._is
_account
_type
_HA
(topic
):
847 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
849 "_admin.operations.{}.detailed-status".format(op_index
): step
851 self
.logger
.debug("Task {} operation={} {}".format(topic
, _id
, step
))
855 update_dict
=update_dict
,
858 old_num_related_tasks
= new_num_related_tasks
859 time_left
-= interval_wait_for_task
862 "Timeout ({}) when waiting for related tasks to be completed".format(
863 timeout_wait_for_task
866 await asyncio
.sleep(interval_wait_for_task
)