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 LcmExceptionNoMgmtIP(LcmException
):
47 class LcmExceptionExit(LcmException
):
52 """utility for compare dot separate versions. Fills with zeros to proper number comparison
53 package version will be something like 4.0.1.post11+gb3f024d.dirty-1. Where 4.0.1 is the git tag, postXX is the
54 number of commits from this tag, and +XXXXXXX is the git commit short id. Total length is 16 with until 999 commits
57 for point
in v
.split("."):
58 point
, _
, _
= point
.partition("+")
59 point
, _
, _
= point
.partition("-")
60 filled
.append(point
.zfill(20))
64 def deep_get(target_dict
, key_list
, default_value
=None):
66 Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
67 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
68 :param target_dict: dictionary to be read
69 :param key_list: list of keys to read from target_dict
70 :param default_value: value to return if key is not present in the nested dictionary
71 :return: The wanted value if exist, None otherwise
74 if not isinstance(target_dict
, dict) or key
not in target_dict
:
76 target_dict
= target_dict
[key
]
80 def get_iterable(in_dict
, in_key
):
82 Similar to <dict>.get(), but if value is None, False, ..., An empty tuple is returned instead
83 :param in_dict: a dictionary
84 :param in_key: the key to look for at in_dict
85 :return: in_dict[in_var] or () if it is None or not present
87 if not in_dict
.get(in_key
):
89 return in_dict
[in_key
]
92 def check_juju_bundle_existence(vnfd
: dict) -> str:
93 """Checks the existence of juju-bundle in the descriptor
96 vnfd: Descriptor as a dictionary
99 Juju bundle if dictionary has juju-bundle else None
105 for kdu
in vnfd
.get("kdu", []):
106 return kdu
.get("juju-bundle", None)
109 def get_charm_artifact_path(base_folder
, charm_name
, charm_type
, revision
=str()) -> str:
110 """Finds the charm artifact paths
113 base_folder: Main folder which will be looked up for charm
114 charm_name: Charm name
115 charm_type: Type of charm native_charm, lxc_proxy_charm or k8s_proxy_charm
116 revision: vnf package revision number if there is
124 extension
= ":" + str(revision
)
126 if base_folder
.get("pkg-dir"):
127 artifact_path
= "{}/{}/{}/{}".format(
128 base_folder
["folder"].split(":")[0] + extension
,
129 base_folder
["pkg-dir"],
131 if charm_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
137 # For SOL004 packages
138 artifact_path
= "{}/Scripts/{}/{}".format(
139 base_folder
["folder"].split(":")[0] + extension
,
141 if charm_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
149 def populate_dict(target_dict
, key_list
, value
):
151 Update target_dict creating nested dictionaries with the key_list. Last key_list item is asigned the value.
152 Example target_dict={K: J}; key_list=[a,b,c]; target_dict will be {K: J, a: {b: {c: value}}}
153 :param target_dict: dictionary to be changed
154 :param key_list: list of keys to insert at target_dict
158 for key
in key_list
[0:-1]:
159 if key
not in target_dict
:
160 target_dict
[key
] = {}
161 target_dict
= target_dict
[key
]
162 target_dict
[key_list
[-1]] = value
165 def get_ee_id_parts(ee_id
):
167 Parses ee_id stored at database that can be either 'version:namespace.helm_id' or only
168 namespace.helm_id for backward compatibility
169 If exists helm version can be helm-v3 or helm (helm-v2 old version)
171 version
, _
, part_id
= ee_id
.rpartition(":")
172 namespace
, _
, helm_id
= part_id
.rpartition(".")
173 return version
, namespace
, helm_id
176 def vld_to_ro_ip_profile(source_data
):
180 if "v4" in source_data
.get("ip-version", "ipv4")
182 "subnet_address": source_data
.get("cidr")
183 or source_data
.get("subnet-address"),
184 "gateway_address": source_data
.get("gateway-ip")
185 or source_data
.get("gateway-address"),
186 "dns_address": ";".join(
187 [v
["address"] for v
in source_data
["dns-server"] if v
.get("address")]
189 if source_data
.get("dns-server")
191 "dhcp_enabled": source_data
.get("dhcp-params", {}).get("enabled", False)
192 or source_data
.get("dhcp-enabled", False),
193 "dhcp_start_address": source_data
["dhcp-params"].get("start-address")
194 if source_data
.get("dhcp-params")
196 "dhcp_count": source_data
["dhcp-params"].get("count")
197 if source_data
.get("dhcp-params")
199 "ipv6_address_mode": source_data
["ipv6-address-mode"]
200 if "ipv6-address-mode" in source_data
206 def __init__(self
, msg
, logger
):
209 :param db: database connection
211 self
.db
= Database().instance
.db
213 self
.fs
= Filesystem().instance
.fs
216 def update_db_2(self
, item
, _id
, _desc
):
218 Updates database with _desc information. If success _desc is cleared
219 :param item: collection
220 :param _id: the _id to use in the query filter
221 :param _desc: dictionary with the content to update. Keys are dot separated keys for
222 :return: None. Exception is raised on error
227 _desc
["_admin.modified"] = now
228 self
.db
.set_one(item
, {"_id": _id
}, _desc
)
230 # except DbException as e:
231 # self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
234 def calculate_charm_hash(zipped_file
):
235 """Calculate the hash of charm files which ends with .charm
238 zipped_file (str): Existing charm package full path
241 hex digest (str): The hash of the charm file
243 filehash
= hashlib
.sha256()
244 with
open(zipped_file
, mode
="rb") as file:
245 contents
= file.read()
246 filehash
.update(contents
)
247 return filehash
.hexdigest()
250 def compare_charm_hash(current_charm
, target_charm
):
251 """Compare the existing charm and the target charm if the charms
252 are given as zip files ends with .charm
255 current_charm (str): Existing charm package full path
256 target_charm (str): Target charm package full path
259 True/False (bool): if charm has changed it returns True
261 return LcmBase
.calculate_charm_hash(
263 ) != LcmBase
.calculate_charm_hash(target_charm
)
266 def compare_charmdir_hash(current_charm_dir
, target_charm_dir
):
267 """Compare the existing charm and the target charm if the charms
268 are given as directories
271 current_charm_dir (str): Existing charm package directory path
272 target_charm_dir (str): Target charm package directory path
275 True/False (bool): if charm has changed it returns True
277 return checksumdir
.dirhash(current_charm_dir
) != checksumdir
.dirhash(
281 def check_charm_hash_changed(
282 self
, current_charm_path
: str, target_charm_path
: str
284 """Find the target charm has changed or not by checking the hash of
285 old and new charm packages
288 current_charm_path (str): Existing charm package artifact path
289 target_charm_path (str): Target charm package artifact path
292 True/False (bool): if charm has changed it returns True
296 # Check if the charm artifacts are available
297 current_charm
= self
.fs
.path
+ current_charm_path
298 target_charm
= self
.fs
.path
+ target_charm_path
300 if os
.path
.exists(current_charm
) and os
.path
.exists(target_charm
):
301 # Compare the hash of .charm files
302 if current_charm
.endswith(".charm"):
303 return LcmBase
.compare_charm_hash(current_charm
, target_charm
)
305 # Compare the hash of charm folders
306 return LcmBase
.compare_charmdir_hash(current_charm
, target_charm
)
310 "Charm artifact {} does not exist in the VNF Package".format(
311 self
.fs
.path
+ target_charm_path
314 except (IOError, OSError, TypeError) as error
:
315 self
.logger
.debug(traceback
.format_exc())
316 self
.logger
.error(f
"{error} occured while checking the charm hashes")
317 raise LcmException(error
)
320 def get_charm_name(charm_metadata_file
: str) -> str:
321 """Get the charm name from metadata file.
324 charm_metadata_file (str): charm metadata file full path
327 charm_name (str): charm name
330 # Read charm metadata.yaml to get the charm name
331 with
open(charm_metadata_file
, "r") as metadata_file
:
332 content
= yaml
.safe_load(metadata_file
)
333 charm_name
= content
["name"]
334 return str(charm_name
)
337 self
, nsd_package_path
: str, nsd_package_name
: str, charm_folder_name
: str
339 """Get the full path of charm folder.
342 nsd_package_path (str): NSD package full path
343 nsd_package_name (str): NSD package name
344 charm_folder_name (str): folder name
347 charm_path (str): charm folder full path
359 def _get_charm_metadata_file(
361 charm_folder_name
: str,
362 nsd_package_path
: str,
363 nsd_package_name
: str,
364 charm_path
: str = None,
366 """Get the path of charm metadata file.
369 charm_folder_name (str): folder name
370 nsd_package_path (str): NSD package full path
371 nsd_package_name (str): NSD package name
372 charm_path (str): Charm full path
375 charm_metadata_file_path (str): charm metadata file full path
378 # Locate the charm metadata.yaml
379 if charm_folder_name
.endswith(".charm"):
386 + charm_folder_name
.replace(".charm", "")
388 # Extract .charm to extract path
389 with
ZipFile(charm_path
, "r") as zipfile
:
390 zipfile
.extractall(extract_path
)
391 return extract_path
+ "/metadata.yaml"
393 return charm_path
+ "/metadata.yaml"
395 def find_charm_name(self
, db_nsr
: dict, charm_folder_name
: str) -> str:
396 """Get the charm name from metadata.yaml of charm package.
399 db_nsr (dict): NS record as a dictionary
400 charm_folder_name (str): charm folder name
403 charm_name (str): charm name
406 if not charm_folder_name
:
407 raise LcmException("charm_folder_name should be provided.")
409 # Find nsd_package details: path, name
410 revision
= db_nsr
.get("revision", "")
412 # Get the NSD package path
414 nsd_package_path
= db_nsr
["nsd-id"] + ":" + str(revision
)
415 db_nsd
= self
.db
.get_one("nsds_revisions", {"_id": nsd_package_path
})
418 nsd_package_path
= db_nsr
["nsd-id"]
420 db_nsd
= self
.db
.get_one("nsds", {"_id": nsd_package_path
})
422 # Get the NSD package name
423 nsd_package_name
= db_nsd
["_admin"]["storage"]["pkg-dir"]
425 # Remove the existing nsd package and sync from FsMongo
426 shutil
.rmtree(self
.fs
.path
+ nsd_package_path
, ignore_errors
=True)
427 self
.fs
.sync(from_path
=nsd_package_path
)
430 charm_path
= self
._get
_charm
_path
(
431 nsd_package_path
, nsd_package_name
, charm_folder_name
434 # Find charm metadata file full path
435 charm_metadata_file
= self
._get
_charm
_metadata
_file
(
436 charm_folder_name
, nsd_package_path
, nsd_package_name
, charm_path
440 return self
.get_charm_name(charm_metadata_file
)
451 self
.logger
.debug(traceback
.format_exc())
452 self
.logger
.error(f
"{error} occured while getting the charm name")
453 raise LcmException(error
)
456 class TaskRegistry(LcmBase
):
458 Implements a registry of task needed for later cancelation, look for related tasks that must be completed before
459 etc. It stores a four level dict
460 First level is the topic, ns, vim_account, sdn
461 Second level is the _id
462 Third level is the operation id
463 Fourth level is a descriptive name, the value is the task class
465 The HA (High-Availability) methods are used when more than one LCM instance is running.
466 To register the current task in the external DB, use LcmBase as base class, to be able
467 to reuse LcmBase.update_db_2()
468 The DB registry uses the following fields to distinguish a task:
469 - op_type: operation type ("nslcmops" or "nsilcmops")
470 - op_id: operation ID
471 - worker: the worker ID for this process
474 # NS/NSI: "services" VIM/WIM/SDN: "accounts"
475 topic_service_list
= ["ns", "nsi"]
476 topic_account_list
= ["vim", "wim", "sdn", "k8scluster", "vca", "k8srepo"]
478 # Map topic to InstanceID
479 topic2instid_dict
= {"ns": "nsInstanceId", "nsi": "netsliceInstanceId"}
481 # Map topic to DB table name
482 topic2dbtable_dict
= {
485 "vim": "vim_accounts",
486 "wim": "wim_accounts",
488 "k8scluster": "k8sclusters",
490 "k8srepo": "k8srepos",
493 def __init__(self
, worker_id
=None, logger
=None):
494 self
.task_registry
= {
504 self
.worker_id
= worker_id
505 self
.db
= Database().instance
.db
508 def register(self
, topic
, _id
, op_id
, task_name
, task
):
511 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
512 :param _id: _id of the related item
513 :param op_id: id of the operation of the related item
514 :param task_name: Task descriptive name, as create, instantiate, terminate. Must be unique in this op_id
515 :param task: Task class
518 if _id
not in self
.task_registry
[topic
]:
519 self
.task_registry
[topic
][_id
] = OrderedDict()
520 if op_id
not in self
.task_registry
[topic
][_id
]:
521 self
.task_registry
[topic
][_id
][op_id
] = {task_name
: task
}
523 self
.task_registry
[topic
][_id
][op_id
][task_name
] = task
524 # print("registering task", topic, _id, op_id, task_name, task)
526 def remove(self
, topic
, _id
, op_id
, task_name
=None):
528 When task is ended, it should be removed. It ignores missing tasks. It also removes tasks done with this _id
529 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
530 :param _id: _id of the related item
531 :param op_id: id of the operation of the related item
532 :param task_name: Task descriptive name. If none it deletes all tasks with same _id and op_id
535 if not self
.task_registry
[topic
].get(_id
):
538 self
.task_registry
[topic
][_id
].pop(op_id
, None)
539 elif self
.task_registry
[topic
][_id
].get(op_id
):
540 self
.task_registry
[topic
][_id
][op_id
].pop(task_name
, None)
543 for op_id_
in list(self
.task_registry
[topic
][_id
]):
544 for name
, task
in self
.task_registry
[topic
][_id
][op_id_
].items():
548 del self
.task_registry
[topic
][_id
][op_id_
]
549 if not self
.task_registry
[topic
][_id
]:
550 del self
.task_registry
[topic
][_id
]
552 def lookfor_related(self
, topic
, _id
, my_op_id
=None):
555 if _id
not in self
.task_registry
[topic
]:
556 return "", task_name_list
557 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
559 if my_op_id
== op_id
:
560 my_op_id
= None # so that the next task is taken
563 for task_name
, task
in self
.task_registry
[topic
][_id
][op_id
].items():
565 task_list
.append(task
)
566 task_name_list
.append(task_name
)
568 return ", ".join(task_name_list
), task_list
570 def cancel(self
, topic
, _id
, target_op_id
=None, target_task_name
=None):
572 Cancel all active tasks of a concrete ns, nsi, vim_account, sdn identified for _id. If op_id is supplied only
573 this is cancelled, and the same with task_name
574 :return: cancelled task to be awaited if needed
576 if not self
.task_registry
[topic
].get(_id
):
578 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
579 if target_op_id
and target_op_id
!= op_id
:
581 for task_name
, task
in list(self
.task_registry
[topic
][_id
][op_id
].items()):
582 if target_task_name
and target_task_name
!= task_name
:
588 # self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, op_id, task_name))
591 def _is_service_type_HA(self
, topic
):
592 return topic
in self
.topic_service_list
594 # Is topic VIM/WIM/SDN?
595 def _is_account_type_HA(self
, topic
):
596 return topic
in self
.topic_account_list
598 # Input: op_id, example: 'abc123def:3' Output: account_id='abc123def', op_index=3
599 def _get_account_and_op_HA(self
, op_id
):
602 account_id
, _
, op_index
= op_id
.rpartition(":")
603 if not account_id
or not op_index
.isdigit():
605 return account_id
, op_index
607 # Get '_id' for any topic and operation
608 def _get_instance_id_HA(self
, topic
, op_type
, op_id
):
610 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
613 # NS/NSI: Use op_id as '_id'
614 elif self
._is
_service
_type
_HA
(topic
):
616 # VIM/SDN/WIM/K8SCLUSTER: Split op_id to get Account ID and Operation Index, use Account ID as '_id'
617 elif self
._is
_account
_type
_HA
(topic
):
618 _id
, _
= self
._get
_account
_and
_op
_HA
(op_id
)
621 # Set DB _filter for querying any related process state
622 def _get_waitfor_filter_HA(self
, db_lcmop
, topic
, op_type
, op_id
):
624 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
625 # In this special case, the timestamp is ignored
627 _filter
= {"operationState": "PROCESSING"}
628 # Otherwise, get 'startTime' timestamp for this operation
631 if self
._is
_service
_type
_HA
(topic
):
633 starttime_this_op
= db_lcmop
.get("startTime")
634 instance_id_label
= self
.topic2instid_dict
.get(topic
)
635 instance_id
= db_lcmop
.get(instance_id_label
)
637 instance_id_label
: instance_id
,
638 "operationState": "PROCESSING",
639 "startTime.lt": starttime_this_op
,
640 "_admin.modified.gt": now
641 - 2 * 3600, # ignore if tow hours of inactivity
643 # VIM/WIM/SDN/K8scluster
644 elif self
._is
_account
_type
_HA
(topic
):
645 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
646 _ops
= db_lcmop
["_admin"]["operations"]
647 _this_op
= _ops
[int(op_index
)]
648 starttime_this_op
= _this_op
.get("startTime", None)
650 "operationState": "PROCESSING",
651 "startTime.lt": starttime_this_op
,
655 # Get DB params for any topic and operation
656 def _get_dbparams_for_lock_HA(self
, topic
, op_type
, op_id
):
660 if self
._is
_service
_type
_HA
(topic
):
661 q_filter
= {"_id": op_id
, "_admin.worker": None}
662 update_dict
= {"_admin.worker": self
.worker_id
}
664 elif self
._is
_account
_type
_HA
(topic
):
665 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
668 if op_type
== "create":
669 # Creating a VIM/WIM/SDN account implies setting '_admin.current_operation' = 0
673 "_admin.operations.{}.worker".format(op_index
): None,
676 "_admin.operations.{}.worker".format(op_index
): self
.worker_id
,
677 "_admin.current_operation": op_index
,
679 return q_filter
, update_dict
681 def lock_HA(self
, topic
, op_type
, op_id
):
683 Lock a task, if possible, to indicate to the HA system that
684 the task will be executed in this LCM instance.
685 :param topic: Can be "ns", "nsi", "vim", "wim", or "sdn"
686 :param op_type: Operation type, can be "nslcmops", "nsilcmops", "create", "edit", "delete"
687 :param op_id: NS, NSI: Operation ID VIM,WIM,SDN: Account ID + ':' + Operation Index
689 True=lock was successful => execute the task (not registered by any other LCM instance)
690 False=lock failed => do NOT execute the task (already registered by another LCM instance)
692 HA tasks and backward compatibility:
693 If topic is "account type" (VIM/WIM/SDN) and op_id is None, 'op_id' was not provided by NBI.
694 This means that the running NBI instance does not support HA.
695 In such a case this method should always return True, to always execute
696 the task in this instance of LCM, without querying the DB.
699 # Backward compatibility for VIM/WIM/SDN/k8scluster without op_id
700 if self
._is
_account
_type
_HA
(topic
) and op_id
is None:
703 # Try to lock this task
704 db_table_name
= self
.topic2dbtable_dict
[topic
]
705 q_filter
, update_dict
= self
._get
_dbparams
_for
_lock
_HA
(topic
, op_type
, op_id
)
706 db_lock_task
= self
.db
.set_one(
709 update_dict
=update_dict
,
712 if db_lock_task
is None:
714 "Task {} operation={} already locked by another worker".format(
720 # Set 'detailed-status' to 'In progress' for VIM/WIM/SDN operations
721 if self
._is
_account
_type
_HA
(topic
):
722 detailed_status
= "In progress"
723 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
724 q_filter
= {"_id": account_id
}
726 "_admin.operations.{}.detailed-status".format(
733 update_dict
=update_dict
,
738 def unlock_HA(self
, topic
, op_type
, op_id
, operationState
, detailed_status
):
740 Register a task, done when finished a VIM/WIM/SDN 'create' operation.
741 :param topic: Can be "vim", "wim", or "sdn"
742 :param op_type: Operation type, can be "create", "edit", "delete"
743 :param op_id: Account ID + ':' + Operation Index
747 # Backward compatibility
748 if not self
._is
_account
_type
_HA
(topic
) or not op_id
:
751 # Get Account ID and Operation Index
752 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
753 db_table_name
= self
.topic2dbtable_dict
[topic
]
755 # If this is a 'delete' operation, the account may have been deleted (SUCCESS) or may still exist (FAILED)
756 # If the account exist, register the HA task.
757 # Update DB for HA tasks
758 q_filter
= {"_id": account_id
}
760 "_admin.operations.{}.operationState".format(op_index
): operationState
,
761 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
762 "_admin.operations.{}.worker".format(op_index
): None,
763 "_admin.current_operation": None,
768 update_dict
=update_dict
,
773 async def waitfor_related_HA(self
, topic
, op_type
, op_id
=None):
775 Wait for any pending related HA tasks
778 # Backward compatibility
780 self
._is
_service
_type
_HA
(topic
) or self
._is
_account
_type
_HA
(topic
)
781 ) and (op_id
is None):
785 db_table_name
= self
.topic2dbtable_dict
.get(topic
)
788 _id
= self
._get
_instance
_id
_HA
(topic
, op_type
, op_id
)
789 _filter
= {"_id": _id
}
790 db_lcmop
= self
.db
.get_one(db_table_name
, _filter
, fail_on_empty
=False)
794 # Set DB _filter for querying any related process state
795 _filter
= self
._get
_waitfor
_filter
_HA
(db_lcmop
, topic
, op_type
, op_id
)
797 # For HA, get list of tasks from DB instead of from dictionary (in-memory) variable.
798 timeout_wait_for_task
= (
799 3600 # Max time (seconds) to wait for a related task to finish
801 # interval_wait_for_task = 30 # A too long polling interval slows things down considerably
802 interval_wait_for_task
= 10 # Interval in seconds for polling related tasks
803 time_left
= timeout_wait_for_task
804 old_num_related_tasks
= 0
806 # Get related tasks (operations within the same instance as this) which are
807 # still running (operationState='PROCESSING') and which were started before this task.
808 # In the case of op_type='ANY', get any related tasks with operationState='PROCESSING', ignore timestamps.
809 db_waitfor_related_task
= self
.db
.get_list(db_table_name
, q_filter
=_filter
)
810 new_num_related_tasks
= len(db_waitfor_related_task
)
811 # If there are no related tasks, there is nothing to wait for, so return.
812 if not new_num_related_tasks
:
814 # If number of pending related tasks have changed,
815 # update the 'detailed-status' field and log the change.
816 # Do NOT update the 'detailed-status' for SDNC-associated-to-VIM operations ('ANY').
817 if (op_type
!= "ANY") and (new_num_related_tasks
!= old_num_related_tasks
):
818 step
= "Waiting for {} related tasks to be completed.".format(
819 new_num_related_tasks
822 q_filter
= {"_id": _id
}
824 if self
._is
_service
_type
_HA
(topic
):
826 "detailed-status": step
,
827 "queuePosition": new_num_related_tasks
,
830 elif self
._is
_account
_type
_HA
(topic
):
831 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
833 "_admin.operations.{}.detailed-status".format(op_index
): step
835 self
.logger
.debug("Task {} operation={} {}".format(topic
, _id
, step
))
839 update_dict
=update_dict
,
842 old_num_related_tasks
= new_num_related_tasks
843 time_left
-= interval_wait_for_task
846 "Timeout ({}) when waiting for related tasks to be completed".format(
847 timeout_wait_for_task
850 await asyncio
.sleep(interval_wait_for_task
)