Add n2vc relations after install configuration sw
[osm/LCM.git] / osm_lcm / lcm_utils.py
1 # -*- coding: utf-8 -*-
2
3 ##
4 # Copyright 2018 Telefonica S.A.
5 #
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
9 #
10 # http://www.apache.org/licenses/LICENSE-2.0
11 #
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
16 # under the License.
17 ##
18
19 import asyncio
20 from collections import OrderedDict
21 from time import time
22 # from osm_common.dbbase import DbException
23
24 __author__ = "Alfonso Tierno"
25
26
27 class LcmException(Exception):
28 pass
29
30
31 class LcmExceptionNoMgmtIP(LcmException):
32 pass
33
34
35 class LcmExceptionExit(LcmException):
36 pass
37
38
39 def versiontuple(v):
40 """utility for compare dot separate versions. Fills with zeros to proper number comparison
41 package version will be something like 4.0.1.post11+gb3f024d.dirty-1. Where 4.0.1 is the git tag, postXX is the
42 number of commits from this tag, and +XXXXXXX is the git commit short id. Total length is 16 with until 999 commits
43 """
44 filled = []
45 for point in v.split("."):
46 point, _, _ = point.partition("+")
47 point, _, _ = point.partition("-")
48 filled.append(point.zfill(20))
49 return tuple(filled)
50
51
52 def deep_get(target_dict, key_list, default_value=None):
53 """
54 Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
55 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
56 :param target_dict: dictionary to be read
57 :param key_list: list of keys to read from target_dict
58 :param default_value: value to return if key is not present in the nested dictionary
59 :return: The wanted value if exist, None otherwise
60 """
61 for key in key_list:
62 if not isinstance(target_dict, dict) or key not in target_dict:
63 return default_value
64 target_dict = target_dict[key]
65 return target_dict
66
67
68 def get_iterable(in_dict, in_key):
69 """
70 Similar to <dict>.get(), but if value is None, False, ..., An empty tuple is returned instead
71 :param in_dict: a dictionary
72 :param in_key: the key to look for at in_dict
73 :return: in_dict[in_var] or () if it is None or not present
74 """
75 if not in_dict.get(in_key):
76 return ()
77 return in_dict[in_key]
78
79
80 def populate_dict(target_dict, key_list, value):
81 """
82 Update target_dict creating nested dictionaries with the key_list. Last key_list item is asigned the value.
83 Example target_dict={K: J}; key_list=[a,b,c]; target_dict will be {K: J, a: {b: {c: value}}}
84 :param target_dict: dictionary to be changed
85 :param key_list: list of keys to insert at target_dict
86 :param value:
87 :return: None
88 """
89 for key in key_list[0:-1]:
90 if key not in target_dict:
91 target_dict[key] = {}
92 target_dict = target_dict[key]
93 target_dict[key_list[-1]] = value
94
95
96 class LcmBase:
97
98 def __init__(self, db, msg, fs, logger):
99 """
100
101 :param db: database connection
102 """
103 self.db = db
104 self.msg = msg
105 self.fs = fs
106 self.logger = logger
107
108 def update_db_2(self, item, _id, _desc):
109 """
110 Updates database with _desc information. If success _desc is cleared
111 :param item:
112 :param _id:
113 :param _desc: dictionary with the content to update. Keys are dot separated keys for
114 :return: None. Exception is raised on error
115 """
116 if not _desc:
117 return
118 now = time()
119 _desc["_admin.modified"] = now
120 self.db.set_one(item, {"_id": _id}, _desc)
121 _desc.clear()
122 # except DbException as e:
123 # self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
124
125
126 class TaskRegistry(LcmBase):
127 """
128 Implements a registry of task needed for later cancelation, look for related tasks that must be completed before
129 etc. It stores a four level dict
130 First level is the topic, ns, vim_account, sdn
131 Second level is the _id
132 Third level is the operation id
133 Fourth level is a descriptive name, the value is the task class
134
135 The HA (High-Availability) methods are used when more than one LCM instance is running.
136 To register the current task in the external DB, use LcmBase as base class, to be able
137 to reuse LcmBase.update_db_2()
138 The DB registry uses the following fields to distinguish a task:
139 - op_type: operation type ("nslcmops" or "nsilcmops")
140 - op_id: operation ID
141 - worker: the worker ID for this process
142 """
143
144 # NS/NSI: "services" VIM/WIM/SDN: "accounts"
145 topic_service_list = ['ns', 'nsi']
146 topic_account_list = ['vim', 'wim', 'sdn', 'k8scluster', 'k8srepo']
147
148 # Map topic to InstanceID
149 topic2instid_dict = {
150 'ns': 'nsInstanceId',
151 'nsi': 'netsliceInstanceId'}
152
153 # Map topic to DB table name
154 topic2dbtable_dict = {
155 'ns': 'nslcmops',
156 'nsi': 'nsilcmops',
157 'vim': 'vim_accounts',
158 'wim': 'wim_accounts',
159 'sdn': 'sdns',
160 'k8scluster': 'k8sclusters',
161 'k8srepo': 'k8srepos'}
162
163 def __init__(self, worker_id=None, db=None, logger=None):
164 self.task_registry = {
165 "ns": {},
166 "nsi": {},
167 "vim_account": {},
168 "wim_account": {},
169 "sdn": {},
170 "k8scluster": {},
171 "k8srepo": {},
172 }
173 self.worker_id = worker_id
174 self.db = db
175 self.logger = logger
176
177 def register(self, topic, _id, op_id, task_name, task):
178 """
179 Register a new task
180 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
181 :param _id: _id of the related item
182 :param op_id: id of the operation of the related item
183 :param task_name: Task descriptive name, as create, instantiate, terminate. Must be unique in this op_id
184 :param task: Task class
185 :return: none
186 """
187 if _id not in self.task_registry[topic]:
188 self.task_registry[topic][_id] = OrderedDict()
189 if op_id not in self.task_registry[topic][_id]:
190 self.task_registry[topic][_id][op_id] = {task_name: task}
191 else:
192 self.task_registry[topic][_id][op_id][task_name] = task
193 # print("registering task", topic, _id, op_id, task_name, task)
194
195 def remove(self, topic, _id, op_id, task_name=None):
196 """
197 When task is ended, it should be removed. It ignores missing tasks. It also removes tasks done with this _id
198 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
199 :param _id: _id of the related item
200 :param op_id: id of the operation of the related item
201 :param task_name: Task descriptive name. If none it deletes all tasks with same _id and op_id
202 :return: None
203 """
204 if not self.task_registry[topic].get(_id):
205 return
206 if not task_name:
207 self.task_registry[topic][_id].pop(op_id, None)
208 elif self.task_registry[topic][_id].get(op_id):
209 self.task_registry[topic][_id][op_id].pop(task_name, None)
210
211 # delete done tasks
212 for op_id_ in list(self.task_registry[topic][_id]):
213 for name, task in self.task_registry[topic][_id][op_id_].items():
214 if not task.done():
215 break
216 else:
217 del self.task_registry[topic][_id][op_id_]
218 if not self.task_registry[topic][_id]:
219 del self.task_registry[topic][_id]
220
221 def lookfor_related(self, topic, _id, my_op_id=None):
222 task_list = []
223 task_name_list = []
224 if _id not in self.task_registry[topic]:
225 return "", task_name_list
226 for op_id in reversed(self.task_registry[topic][_id]):
227 if my_op_id:
228 if my_op_id == op_id:
229 my_op_id = None # so that the next task is taken
230 continue
231
232 for task_name, task in self.task_registry[topic][_id][op_id].items():
233 if not task.done():
234 task_list.append(task)
235 task_name_list.append(task_name)
236 break
237 return ", ".join(task_name_list), task_list
238
239 def cancel(self, topic, _id, target_op_id=None, target_task_name=None):
240 """
241 Cancel all active tasks of a concrete ns, nsi, vim_account, sdn identified for _id. If op_id is supplied only
242 this is cancelled, and the same with task_name
243 """
244 if not self.task_registry[topic].get(_id):
245 return
246 for op_id in reversed(self.task_registry[topic][_id]):
247 if target_op_id and target_op_id != op_id:
248 continue
249 for task_name, task in self.task_registry[topic][_id][op_id].items():
250 if target_task_name and target_task_name != task_name:
251 continue
252 # result =
253 task.cancel()
254 # if result:
255 # self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, op_id, task_name))
256
257 # Is topic NS/NSI?
258 def _is_service_type_HA(self, topic):
259 return topic in self.topic_service_list
260
261 # Is topic VIM/WIM/SDN?
262 def _is_account_type_HA(self, topic):
263 return topic in self.topic_account_list
264
265 # Input: op_id, example: 'abc123def:3' Output: account_id='abc123def', op_index=3
266 def _get_account_and_op_HA(self, op_id):
267 if not op_id:
268 return (None, None)
269 account_id, _, op_index = op_id.rpartition(':')
270 if not account_id:
271 return (None, None)
272 if not op_index.isdigit():
273 return (None, None)
274 return account_id, op_index
275
276 # Get '_id' for any topic and operation
277 def _get_instance_id_HA(self, topic, op_type, op_id):
278 _id = None
279 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
280 if op_type == 'ANY':
281 _id = op_id
282 # NS/NSI: Use op_id as '_id'
283 elif self._is_service_type_HA(topic):
284 _id = op_id
285 # VIM/SDN/WIM/K8SCLUSTER: Split op_id to get Account ID and Operation Index, use Account ID as '_id'
286 elif self._is_account_type_HA(topic):
287 _id, _ = self._get_account_and_op_HA(op_id)
288 return _id
289
290 # Set DB _filter for querying any related process state
291 def _get_waitfor_filter_HA(self, db_lcmop, topic, op_type, op_id):
292 _filter = {}
293 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
294 # In this special case, the timestamp is ignored
295 if op_type == 'ANY':
296 _filter = {'operationState': 'PROCESSING'}
297 # Otherwise, get 'startTime' timestamp for this operation
298 else:
299 # NS/NSI
300 if self._is_service_type_HA(topic):
301 now = time()
302 starttime_this_op = db_lcmop.get("startTime")
303 instance_id_label = self.topic2instid_dict.get(topic)
304 instance_id = db_lcmop.get(instance_id_label)
305 _filter = {instance_id_label: instance_id,
306 'operationState': 'PROCESSING',
307 'startTime.lt': starttime_this_op,
308 "_admin.modified.gt": now - 2*3600, # ignore if tow hours of inactivity
309 }
310 # VIM/WIM/SDN/K8scluster
311 elif self._is_account_type_HA(topic):
312 _, op_index = self._get_account_and_op_HA(op_id)
313 _ops = db_lcmop['_admin']['operations']
314 _this_op = _ops[int(op_index)]
315 starttime_this_op = _this_op.get('startTime', None)
316 _filter = {'operationState': 'PROCESSING',
317 'startTime.lt': starttime_this_op}
318 return _filter
319
320 # Get DB params for any topic and operation
321 def _get_dbparams_for_lock_HA(self, topic, op_type, op_id):
322 q_filter = {}
323 update_dict = {}
324 # NS/NSI
325 if self._is_service_type_HA(topic):
326 q_filter = {'_id': op_id, '_admin.worker': None}
327 update_dict = {'_admin.worker': self.worker_id}
328 # VIM/WIM/SDN
329 elif self._is_account_type_HA(topic):
330 account_id, op_index = self._get_account_and_op_HA(op_id)
331 if not account_id:
332 return None, None
333 if op_type == 'create':
334 # Creating a VIM/WIM/SDN account implies setting '_admin.current_operation' = 0
335 op_index = 0
336 q_filter = {'_id': account_id, "_admin.operations.{}.worker".format(op_index): None}
337 update_dict = {'_admin.operations.{}.worker'.format(op_index): self.worker_id,
338 '_admin.current_operation': op_index}
339 return q_filter, update_dict
340
341 def lock_HA(self, topic, op_type, op_id):
342 """
343 Lock a task, if possible, to indicate to the HA system that
344 the task will be executed in this LCM instance.
345 :param topic: Can be "ns", "nsi", "vim", "wim", or "sdn"
346 :param op_type: Operation type, can be "nslcmops", "nsilcmops", "create", "edit", "delete"
347 :param op_id: NS, NSI: Operation ID VIM,WIM,SDN: Account ID + ':' + Operation Index
348 :return:
349 True=lock was successful => execute the task (not registered by any other LCM instance)
350 False=lock failed => do NOT execute the task (already registered by another LCM instance)
351
352 HA tasks and backward compatibility:
353 If topic is "account type" (VIM/WIM/SDN) and op_id is None, 'op_id' was not provided by NBI.
354 This means that the running NBI instance does not support HA.
355 In such a case this method should always return True, to always execute
356 the task in this instance of LCM, without querying the DB.
357 """
358
359 # Backward compatibility for VIM/WIM/SDN/k8scluster without op_id
360 if self._is_account_type_HA(topic) and op_id is None:
361 return True
362
363 # Try to lock this task
364 db_table_name = self.topic2dbtable_dict.get(topic)
365 q_filter, update_dict = self._get_dbparams_for_lock_HA(topic, op_type, op_id)
366 db_lock_task = self.db.set_one(db_table_name,
367 q_filter=q_filter,
368 update_dict=update_dict,
369 fail_on_empty=False)
370 if db_lock_task is None:
371 self.logger.debug("Task {} operation={} already locked by another worker".format(topic, op_id))
372 return False
373 else:
374 # Set 'detailed-status' to 'In progress' for VIM/WIM/SDN operations
375 if self._is_account_type_HA(topic):
376 detailed_status = 'In progress'
377 account_id, op_index = self._get_account_and_op_HA(op_id)
378 q_filter = {'_id': account_id}
379 update_dict = {'_admin.operations.{}.detailed-status'.format(op_index): detailed_status}
380 self.db.set_one(db_table_name,
381 q_filter=q_filter,
382 update_dict=update_dict,
383 fail_on_empty=False)
384 return True
385
386 def register_HA(self, topic, op_type, op_id, operationState, detailed_status):
387 """
388 Register a task, done when finished a VIM/WIM/SDN 'create' operation.
389 :param topic: Can be "vim", "wim", or "sdn"
390 :param op_type: Operation type, can be "create", "edit", "delete"
391 :param op_id: Account ID + ':' + Operation Index
392 :return: nothing
393 """
394
395 # Backward compatibility
396 if not self._is_account_type_HA(topic) or (self._is_account_type_HA(topic) and op_id is None):
397 return
398
399 # Get Account ID and Operation Index
400 account_id, op_index = self._get_account_and_op_HA(op_id)
401 db_table_name = self.topic2dbtable_dict.get(topic)
402
403 # If this is a 'delete' operation, the account may have been deleted (SUCCESS) or may still exist (FAILED)
404 # If the account exist, register the HA task.
405 # Update DB for HA tasks
406 q_filter = {'_id': account_id}
407 update_dict = {'_admin.operations.{}.operationState'.format(op_index): operationState,
408 '_admin.operations.{}.detailed-status'.format(op_index): detailed_status}
409 self.db.set_one(db_table_name,
410 q_filter=q_filter,
411 update_dict=update_dict,
412 fail_on_empty=False)
413 return
414
415 async def waitfor_related_HA(self, topic, op_type, op_id=None):
416 """
417 Wait for any pending related HA tasks
418 """
419
420 # Backward compatibility
421 if not (self._is_service_type_HA(topic) or self._is_account_type_HA(topic)) and (op_id is None):
422 return
423
424 # Get DB table name
425 db_table_name = self.topic2dbtable_dict.get(topic)
426
427 # Get instance ID
428 _id = self._get_instance_id_HA(topic, op_type, op_id)
429 _filter = {"_id": _id}
430 db_lcmop = self.db.get_one(db_table_name,
431 _filter,
432 fail_on_empty=False)
433 if not db_lcmop:
434 return
435
436 # Set DB _filter for querying any related process state
437 _filter = self._get_waitfor_filter_HA(db_lcmop, topic, op_type, op_id)
438
439 # For HA, get list of tasks from DB instead of from dictionary (in-memory) variable.
440 timeout_wait_for_task = 3600 # Max time (seconds) to wait for a related task to finish
441 # interval_wait_for_task = 30 # A too long polling interval slows things down considerably
442 interval_wait_for_task = 10 # Interval in seconds for polling related tasks
443 time_left = timeout_wait_for_task
444 old_num_related_tasks = 0
445 while True:
446 # Get related tasks (operations within the same instance as this) which are
447 # still running (operationState='PROCESSING') and which were started before this task.
448 # In the case of op_type='ANY', get any related tasks with operationState='PROCESSING', ignore timestamps.
449 db_waitfor_related_task = self.db.get_list(db_table_name,
450 q_filter=_filter)
451 new_num_related_tasks = len(db_waitfor_related_task)
452 # If there are no related tasks, there is nothing to wait for, so return.
453 if not new_num_related_tasks:
454 return
455 # If number of pending related tasks have changed,
456 # update the 'detailed-status' field and log the change.
457 # Do NOT update the 'detailed-status' for SDNC-associated-to-VIM operations ('ANY').
458 if (op_type != 'ANY') and (new_num_related_tasks != old_num_related_tasks):
459 step = "Waiting for {} related tasks to be completed.".format(new_num_related_tasks)
460 update_dict = {}
461 q_filter = {'_id': _id}
462 # NS/NSI
463 if self._is_service_type_HA(topic):
464 update_dict = {'detailed-status': step, 'queuePosition': new_num_related_tasks}
465 # VIM/WIM/SDN
466 elif self._is_account_type_HA(topic):
467 _, op_index = self._get_account_and_op_HA(op_id)
468 update_dict = {'_admin.operations.{}.detailed-status'.format(op_index): step}
469 self.logger.debug("Task {} operation={} {}".format(topic, _id, step))
470 self.db.set_one(db_table_name,
471 q_filter=q_filter,
472 update_dict=update_dict,
473 fail_on_empty=False)
474 old_num_related_tasks = new_num_related_tasks
475 time_left -= interval_wait_for_task
476 if time_left < 0:
477 raise LcmException(
478 "Timeout ({}) when waiting for related tasks to be completed".format(
479 timeout_wait_for_task))
480 await asyncio.sleep(interval_wait_for_task)
481
482 return