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
27 from osm_common
.fsbase
import FsException
28 from osm_lcm
.data_utils
.database
.database
import Database
29 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
31 from zipfile
import ZipFile
, BadZipfile
33 # from osm_common.dbbase import DbException
35 __author__
= "Alfonso Tierno"
38 class LcmException(Exception):
42 class LcmExceptionNoMgmtIP(LcmException
):
46 class LcmExceptionExit(LcmException
):
51 """utility for compare dot separate versions. Fills with zeros to proper number comparison
52 package version will be something like 4.0.1.post11+gb3f024d.dirty-1. Where 4.0.1 is the git tag, postXX is the
53 number of commits from this tag, and +XXXXXXX is the git commit short id. Total length is 16 with until 999 commits
56 for point
in v
.split("."):
57 point
, _
, _
= point
.partition("+")
58 point
, _
, _
= point
.partition("-")
59 filled
.append(point
.zfill(20))
63 def deep_get(target_dict
, key_list
, default_value
=None):
65 Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
66 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
67 :param target_dict: dictionary to be read
68 :param key_list: list of keys to read from target_dict
69 :param default_value: value to return if key is not present in the nested dictionary
70 :return: The wanted value if exist, None otherwise
73 if not isinstance(target_dict
, dict) or key
not in target_dict
:
75 target_dict
= target_dict
[key
]
79 def get_iterable(in_dict
, in_key
):
81 Similar to <dict>.get(), but if value is None, False, ..., An empty tuple is returned instead
82 :param in_dict: a dictionary
83 :param in_key: the key to look for at in_dict
84 :return: in_dict[in_var] or () if it is None or not present
86 if not in_dict
.get(in_key
):
88 return in_dict
[in_key
]
91 def check_juju_bundle_existence(vnfd
: dict) -> str:
92 """Checks the existence of juju-bundle in the descriptor
95 vnfd: Descriptor as a dictionary
98 Juju bundle if dictionary has juju-bundle else None
104 for kdu
in vnfd
.get("kdu", []):
105 return kdu
.get("juju-bundle", None)
108 def get_charm_artifact_path(base_folder
, charm_name
, charm_type
, revision
=str()) -> str:
109 """Finds the charm artifact paths
112 base_folder: Main folder which will be looked up for charm
113 charm_name: Charm name
114 charm_type: Type of charm native_charm, lxc_proxy_charm or k8s_proxy_charm
115 revision: vnf package revision number if there is
123 extension
= ":" + str(revision
)
125 if base_folder
.get("pkg-dir"):
126 artifact_path
= "{}/{}/{}/{}".format(
127 base_folder
["folder"].split(":")[0] + extension
,
128 base_folder
["pkg-dir"],
130 if charm_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
136 # For SOL004 packages
137 artifact_path
= "{}/Scripts/{}/{}".format(
138 base_folder
["folder"].split(":")[0] + extension
,
140 if charm_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
148 def populate_dict(target_dict
, key_list
, value
):
150 Update target_dict creating nested dictionaries with the key_list. Last key_list item is asigned the value.
151 Example target_dict={K: J}; key_list=[a,b,c]; target_dict will be {K: J, a: {b: {c: value}}}
152 :param target_dict: dictionary to be changed
153 :param key_list: list of keys to insert at target_dict
157 for key
in key_list
[0:-1]:
158 if key
not in target_dict
:
159 target_dict
[key
] = {}
160 target_dict
= target_dict
[key
]
161 target_dict
[key_list
[-1]] = value
164 def get_ee_id_parts(ee_id
):
166 Parses ee_id stored at database that can be either 'version:namespace.helm_id' or only
167 namespace.helm_id for backward compatibility
168 If exists helm version can be helm-v3 or helm (helm-v2 old version)
170 version
, _
, part_id
= ee_id
.rpartition(":")
171 namespace
, _
, helm_id
= part_id
.rpartition(".")
172 return version
, namespace
, helm_id
176 def __init__(self
, msg
, logger
):
179 :param db: database connection
181 self
.db
= Database().instance
.db
183 self
.fs
= Filesystem().instance
.fs
186 def update_db_2(self
, item
, _id
, _desc
):
188 Updates database with _desc information. If success _desc is cleared
189 :param item: collection
190 :param _id: the _id to use in the query filter
191 :param _desc: dictionary with the content to update. Keys are dot separated keys for
192 :return: None. Exception is raised on error
197 _desc
["_admin.modified"] = now
198 self
.db
.set_one(item
, {"_id": _id
}, _desc
)
200 # except DbException as e:
201 # self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
203 def check_charm_hash_changed(
204 self
, current_charm_path
: str, target_charm_path
: str
206 """Find the target charm has changed or not by checking the hash of
207 old and new charm packages
210 current_charm_path (str): Existing charm package artifact path
211 target_charm_path (str): Target charm package artifact path
214 True/False (bool): if charm has changed it returns True
217 # Check if the charm artifacts are available
218 if os
.path
.exists(self
.fs
.path
+ current_charm_path
) and os
.path
.exists(
219 self
.fs
.path
+ target_charm_path
221 # Compare the hash of charm folders
222 if checksumdir
.dirhash(
223 self
.fs
.path
+ current_charm_path
224 ) != checksumdir
.dirhash(self
.fs
.path
+ target_charm_path
):
232 "Charm artifact {} does not exist in the VNF Package".format(
233 self
.fs
.path
+ target_charm_path
238 def get_charm_name(charm_metadata_file
: str) -> str:
239 """Get the charm name from metadata file.
242 charm_metadata_file (str): charm metadata file full path
245 charm_name (str): charm name
248 # Read charm metadata.yaml to get the charm name
249 with
open(charm_metadata_file
, "r") as metadata_file
:
250 content
= yaml
.safe_load(metadata_file
)
251 charm_name
= content
["name"]
252 return str(charm_name
)
255 self
, nsd_package_path
: str, nsd_package_name
: str, charm_folder_name
: str
257 """Get the full path of charm folder.
260 nsd_package_path (str): NSD package full path
261 nsd_package_name (str): NSD package name
262 charm_folder_name (str): folder name
265 charm_path (str): charm folder full path
277 def _get_charm_metadata_file(
279 charm_folder_name
: str,
280 nsd_package_path
: str,
281 nsd_package_name
: str,
282 charm_path
: str = None,
284 """Get the path of charm metadata file.
287 charm_folder_name (str): folder name
288 nsd_package_path (str): NSD package full path
289 nsd_package_name (str): NSD package name
290 charm_path (str): Charm full path
293 charm_metadata_file_path (str): charm metadata file full path
296 # Locate the charm metadata.yaml
297 if charm_folder_name
.endswith(".charm"):
304 + charm_folder_name
.replace(".charm", "")
306 # Extract .charm to extract path
307 with
ZipFile(charm_path
, "r") as zipfile
:
308 zipfile
.extractall(extract_path
)
309 return extract_path
+ "/metadata.yaml"
311 return charm_path
+ "/metadata.yaml"
313 def find_charm_name(self
, db_nsr
: dict, charm_folder_name
: str) -> str:
314 """Get the charm name from metadata.yaml of charm package.
317 db_nsr (dict): NS record as a dictionary
318 charm_folder_name (str): charm folder name
321 charm_name (str): charm name
324 if not charm_folder_name
:
325 raise LcmException("charm_folder_name should be provided.")
327 # Find nsd_package details: path, name
328 revision
= db_nsr
.get("revision", "")
330 # Get the NSD package path
333 nsd_package_path
= db_nsr
["nsd-id"] + ":" + str(revision
)
334 db_nsd
= self
.db
.get_one("nsds_revisions", {"_id": nsd_package_path
})
337 nsd_package_path
= db_nsr
["nsd-id"]
339 db_nsd
= self
.db
.get_one("nsds", {"_id": nsd_package_path
})
341 # Get the NSD package name
342 nsd_package_name
= db_nsd
["_admin"]["storage"]["pkg-dir"]
344 # Remove the existing nsd package and sync from FsMongo
345 shutil
.rmtree(self
.fs
.path
+ nsd_package_path
, ignore_errors
=True)
346 self
.fs
.sync(from_path
=nsd_package_path
)
349 charm_path
= self
._get
_charm
_path
(
350 nsd_package_path
, nsd_package_name
, charm_folder_name
353 # Find charm metadata file full path
354 charm_metadata_file
= self
._get
_charm
_metadata
_file
(
355 charm_folder_name
, nsd_package_path
, nsd_package_name
, charm_path
359 return self
.get_charm_name(charm_metadata_file
)
370 self
.logger
.debug(traceback
.format_exc())
371 self
.logger
.error(f
"{error} occured while getting the charm name")
372 raise LcmException(error
)
375 class TaskRegistry(LcmBase
):
377 Implements a registry of task needed for later cancelation, look for related tasks that must be completed before
378 etc. It stores a four level dict
379 First level is the topic, ns, vim_account, sdn
380 Second level is the _id
381 Third level is the operation id
382 Fourth level is a descriptive name, the value is the task class
384 The HA (High-Availability) methods are used when more than one LCM instance is running.
385 To register the current task in the external DB, use LcmBase as base class, to be able
386 to reuse LcmBase.update_db_2()
387 The DB registry uses the following fields to distinguish a task:
388 - op_type: operation type ("nslcmops" or "nsilcmops")
389 - op_id: operation ID
390 - worker: the worker ID for this process
393 # NS/NSI: "services" VIM/WIM/SDN: "accounts"
394 topic_service_list
= ["ns", "nsi"]
395 topic_account_list
= ["vim", "wim", "sdn", "k8scluster", "vca", "k8srepo"]
397 # Map topic to InstanceID
398 topic2instid_dict
= {"ns": "nsInstanceId", "nsi": "netsliceInstanceId"}
400 # Map topic to DB table name
401 topic2dbtable_dict
= {
404 "vim": "vim_accounts",
405 "wim": "wim_accounts",
407 "k8scluster": "k8sclusters",
409 "k8srepo": "k8srepos",
412 def __init__(self
, worker_id
=None, logger
=None):
413 self
.task_registry
= {
423 self
.worker_id
= worker_id
424 self
.db
= Database().instance
.db
427 def register(self
, topic
, _id
, op_id
, task_name
, task
):
430 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
431 :param _id: _id of the related item
432 :param op_id: id of the operation of the related item
433 :param task_name: Task descriptive name, as create, instantiate, terminate. Must be unique in this op_id
434 :param task: Task class
437 if _id
not in self
.task_registry
[topic
]:
438 self
.task_registry
[topic
][_id
] = OrderedDict()
439 if op_id
not in self
.task_registry
[topic
][_id
]:
440 self
.task_registry
[topic
][_id
][op_id
] = {task_name
: task
}
442 self
.task_registry
[topic
][_id
][op_id
][task_name
] = task
443 # print("registering task", topic, _id, op_id, task_name, task)
445 def remove(self
, topic
, _id
, op_id
, task_name
=None):
447 When task is ended, it should be removed. It ignores missing tasks. It also removes tasks done with this _id
448 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
449 :param _id: _id of the related item
450 :param op_id: id of the operation of the related item
451 :param task_name: Task descriptive name. If none it deletes all tasks with same _id and op_id
454 if not self
.task_registry
[topic
].get(_id
):
457 self
.task_registry
[topic
][_id
].pop(op_id
, None)
458 elif self
.task_registry
[topic
][_id
].get(op_id
):
459 self
.task_registry
[topic
][_id
][op_id
].pop(task_name
, None)
462 for op_id_
in list(self
.task_registry
[topic
][_id
]):
463 for name
, task
in self
.task_registry
[topic
][_id
][op_id_
].items():
467 del self
.task_registry
[topic
][_id
][op_id_
]
468 if not self
.task_registry
[topic
][_id
]:
469 del self
.task_registry
[topic
][_id
]
471 def lookfor_related(self
, topic
, _id
, my_op_id
=None):
474 if _id
not in self
.task_registry
[topic
]:
475 return "", task_name_list
476 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
478 if my_op_id
== op_id
:
479 my_op_id
= None # so that the next task is taken
482 for task_name
, task
in self
.task_registry
[topic
][_id
][op_id
].items():
484 task_list
.append(task
)
485 task_name_list
.append(task_name
)
487 return ", ".join(task_name_list
), task_list
489 def cancel(self
, topic
, _id
, target_op_id
=None, target_task_name
=None):
491 Cancel all active tasks of a concrete ns, nsi, vim_account, sdn identified for _id. If op_id is supplied only
492 this is cancelled, and the same with task_name
494 if not self
.task_registry
[topic
].get(_id
):
496 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
497 if target_op_id
and target_op_id
!= op_id
:
499 for task_name
, task
in self
.task_registry
[topic
][_id
][op_id
].items():
500 if target_task_name
and target_task_name
!= task_name
:
505 # self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, op_id, task_name))
508 def _is_service_type_HA(self
, topic
):
509 return topic
in self
.topic_service_list
511 # Is topic VIM/WIM/SDN?
512 def _is_account_type_HA(self
, topic
):
513 return topic
in self
.topic_account_list
515 # Input: op_id, example: 'abc123def:3' Output: account_id='abc123def', op_index=3
516 def _get_account_and_op_HA(self
, op_id
):
519 account_id
, _
, op_index
= op_id
.rpartition(":")
520 if not account_id
or not op_index
.isdigit():
522 return account_id
, op_index
524 # Get '_id' for any topic and operation
525 def _get_instance_id_HA(self
, topic
, op_type
, op_id
):
527 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
530 # NS/NSI: Use op_id as '_id'
531 elif self
._is
_service
_type
_HA
(topic
):
533 # VIM/SDN/WIM/K8SCLUSTER: Split op_id to get Account ID and Operation Index, use Account ID as '_id'
534 elif self
._is
_account
_type
_HA
(topic
):
535 _id
, _
= self
._get
_account
_and
_op
_HA
(op_id
)
538 # Set DB _filter for querying any related process state
539 def _get_waitfor_filter_HA(self
, db_lcmop
, topic
, op_type
, op_id
):
541 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
542 # In this special case, the timestamp is ignored
544 _filter
= {"operationState": "PROCESSING"}
545 # Otherwise, get 'startTime' timestamp for this operation
548 if self
._is
_service
_type
_HA
(topic
):
550 starttime_this_op
= db_lcmop
.get("startTime")
551 instance_id_label
= self
.topic2instid_dict
.get(topic
)
552 instance_id
= db_lcmop
.get(instance_id_label
)
554 instance_id_label
: instance_id
,
555 "operationState": "PROCESSING",
556 "startTime.lt": starttime_this_op
,
557 "_admin.modified.gt": now
558 - 2 * 3600, # ignore if tow hours of inactivity
560 # VIM/WIM/SDN/K8scluster
561 elif self
._is
_account
_type
_HA
(topic
):
562 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
563 _ops
= db_lcmop
["_admin"]["operations"]
564 _this_op
= _ops
[int(op_index
)]
565 starttime_this_op
= _this_op
.get("startTime", None)
567 "operationState": "PROCESSING",
568 "startTime.lt": starttime_this_op
,
572 # Get DB params for any topic and operation
573 def _get_dbparams_for_lock_HA(self
, topic
, op_type
, op_id
):
577 if self
._is
_service
_type
_HA
(topic
):
578 q_filter
= {"_id": op_id
, "_admin.worker": None}
579 update_dict
= {"_admin.worker": self
.worker_id
}
581 elif self
._is
_account
_type
_HA
(topic
):
582 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
585 if op_type
== "create":
586 # Creating a VIM/WIM/SDN account implies setting '_admin.current_operation' = 0
590 "_admin.operations.{}.worker".format(op_index
): None,
593 "_admin.operations.{}.worker".format(op_index
): self
.worker_id
,
594 "_admin.current_operation": op_index
,
596 return q_filter
, update_dict
598 def lock_HA(self
, topic
, op_type
, op_id
):
600 Lock a task, if possible, to indicate to the HA system that
601 the task will be executed in this LCM instance.
602 :param topic: Can be "ns", "nsi", "vim", "wim", or "sdn"
603 :param op_type: Operation type, can be "nslcmops", "nsilcmops", "create", "edit", "delete"
604 :param op_id: NS, NSI: Operation ID VIM,WIM,SDN: Account ID + ':' + Operation Index
606 True=lock was successful => execute the task (not registered by any other LCM instance)
607 False=lock failed => do NOT execute the task (already registered by another LCM instance)
609 HA tasks and backward compatibility:
610 If topic is "account type" (VIM/WIM/SDN) and op_id is None, 'op_id' was not provided by NBI.
611 This means that the running NBI instance does not support HA.
612 In such a case this method should always return True, to always execute
613 the task in this instance of LCM, without querying the DB.
616 # Backward compatibility for VIM/WIM/SDN/k8scluster without op_id
617 if self
._is
_account
_type
_HA
(topic
) and op_id
is None:
620 # Try to lock this task
621 db_table_name
= self
.topic2dbtable_dict
[topic
]
622 q_filter
, update_dict
= self
._get
_dbparams
_for
_lock
_HA
(topic
, op_type
, op_id
)
623 db_lock_task
= self
.db
.set_one(
626 update_dict
=update_dict
,
629 if db_lock_task
is None:
631 "Task {} operation={} already locked by another worker".format(
637 # Set 'detailed-status' to 'In progress' for VIM/WIM/SDN operations
638 if self
._is
_account
_type
_HA
(topic
):
639 detailed_status
= "In progress"
640 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
641 q_filter
= {"_id": account_id
}
643 "_admin.operations.{}.detailed-status".format(
650 update_dict
=update_dict
,
655 def unlock_HA(self
, topic
, op_type
, op_id
, operationState
, detailed_status
):
657 Register a task, done when finished a VIM/WIM/SDN 'create' operation.
658 :param topic: Can be "vim", "wim", or "sdn"
659 :param op_type: Operation type, can be "create", "edit", "delete"
660 :param op_id: Account ID + ':' + Operation Index
664 # Backward compatibility
665 if not self
._is
_account
_type
_HA
(topic
) or not op_id
:
668 # Get Account ID and Operation Index
669 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
670 db_table_name
= self
.topic2dbtable_dict
[topic
]
672 # If this is a 'delete' operation, the account may have been deleted (SUCCESS) or may still exist (FAILED)
673 # If the account exist, register the HA task.
674 # Update DB for HA tasks
675 q_filter
= {"_id": account_id
}
677 "_admin.operations.{}.operationState".format(op_index
): operationState
,
678 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
679 "_admin.operations.{}.worker".format(op_index
): None,
680 "_admin.current_operation": None,
685 update_dict
=update_dict
,
690 async def waitfor_related_HA(self
, topic
, op_type
, op_id
=None):
692 Wait for any pending related HA tasks
695 # Backward compatibility
697 self
._is
_service
_type
_HA
(topic
) or self
._is
_account
_type
_HA
(topic
)
698 ) and (op_id
is None):
702 db_table_name
= self
.topic2dbtable_dict
.get(topic
)
705 _id
= self
._get
_instance
_id
_HA
(topic
, op_type
, op_id
)
706 _filter
= {"_id": _id
}
707 db_lcmop
= self
.db
.get_one(db_table_name
, _filter
, fail_on_empty
=False)
711 # Set DB _filter for querying any related process state
712 _filter
= self
._get
_waitfor
_filter
_HA
(db_lcmop
, topic
, op_type
, op_id
)
714 # For HA, get list of tasks from DB instead of from dictionary (in-memory) variable.
715 timeout_wait_for_task
= (
716 3600 # Max time (seconds) to wait for a related task to finish
718 # interval_wait_for_task = 30 # A too long polling interval slows things down considerably
719 interval_wait_for_task
= 10 # Interval in seconds for polling related tasks
720 time_left
= timeout_wait_for_task
721 old_num_related_tasks
= 0
723 # Get related tasks (operations within the same instance as this) which are
724 # still running (operationState='PROCESSING') and which were started before this task.
725 # In the case of op_type='ANY', get any related tasks with operationState='PROCESSING', ignore timestamps.
726 db_waitfor_related_task
= self
.db
.get_list(db_table_name
, q_filter
=_filter
)
727 new_num_related_tasks
= len(db_waitfor_related_task
)
728 # If there are no related tasks, there is nothing to wait for, so return.
729 if not new_num_related_tasks
:
731 # If number of pending related tasks have changed,
732 # update the 'detailed-status' field and log the change.
733 # Do NOT update the 'detailed-status' for SDNC-associated-to-VIM operations ('ANY').
734 if (op_type
!= "ANY") and (new_num_related_tasks
!= old_num_related_tasks
):
735 step
= "Waiting for {} related tasks to be completed.".format(
736 new_num_related_tasks
739 q_filter
= {"_id": _id
}
741 if self
._is
_service
_type
_HA
(topic
):
743 "detailed-status": step
,
744 "queuePosition": new_num_related_tasks
,
747 elif self
._is
_account
_type
_HA
(topic
):
748 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
750 "_admin.operations.{}.detailed-status".format(op_index
): step
752 self
.logger
.debug("Task {} operation={} {}".format(topic
, _id
, step
))
756 update_dict
=update_dict
,
759 old_num_related_tasks
= new_num_related_tasks
760 time_left
-= interval_wait_for_task
763 "Timeout ({}) when waiting for related tasks to be completed".format(
764 timeout_wait_for_task
767 await asyncio
.sleep(interval_wait_for_task
)