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
20 from collections
import OrderedDict
22 from osm_lcm
.data_utils
.database
.database
import Database
23 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
25 # from osm_common.dbbase import DbException
27 __author__
= "Alfonso Tierno"
30 class LcmException(Exception):
34 class LcmExceptionNoMgmtIP(LcmException
):
38 class LcmExceptionExit(LcmException
):
43 """utility for compare dot separate versions. Fills with zeros to proper number comparison
44 package version will be something like 4.0.1.post11+gb3f024d.dirty-1. Where 4.0.1 is the git tag, postXX is the
45 number of commits from this tag, and +XXXXXXX is the git commit short id. Total length is 16 with until 999 commits
48 for point
in v
.split("."):
49 point
, _
, _
= point
.partition("+")
50 point
, _
, _
= point
.partition("-")
51 filled
.append(point
.zfill(20))
55 def deep_get(target_dict
, key_list
, default_value
=None):
57 Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
58 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
59 :param target_dict: dictionary to be read
60 :param key_list: list of keys to read from target_dict
61 :param default_value: value to return if key is not present in the nested dictionary
62 :return: The wanted value if exist, None otherwise
65 if not isinstance(target_dict
, dict) or key
not in target_dict
:
67 target_dict
= target_dict
[key
]
71 def get_iterable(in_dict
, in_key
):
73 Similar to <dict>.get(), but if value is None, False, ..., An empty tuple is returned instead
74 :param in_dict: a dictionary
75 :param in_key: the key to look for at in_dict
76 :return: in_dict[in_var] or () if it is None or not present
78 if not in_dict
.get(in_key
):
80 return in_dict
[in_key
]
83 def populate_dict(target_dict
, key_list
, value
):
85 Update target_dict creating nested dictionaries with the key_list. Last key_list item is asigned the value.
86 Example target_dict={K: J}; key_list=[a,b,c]; target_dict will be {K: J, a: {b: {c: value}}}
87 :param target_dict: dictionary to be changed
88 :param key_list: list of keys to insert at target_dict
92 for key
in key_list
[0:-1]:
93 if key
not in target_dict
:
95 target_dict
= target_dict
[key
]
96 target_dict
[key_list
[-1]] = value
101 def __init__(self
, msg
, logger
):
104 :param db: database connection
106 self
.db
= Database().instance
.db
108 self
.fs
= Filesystem().instance
.fs
111 def update_db_2(self
, item
, _id
, _desc
):
113 Updates database with _desc information. If success _desc is cleared
116 :param _desc: dictionary with the content to update. Keys are dot separated keys for
117 :return: None. Exception is raised on error
122 _desc
["_admin.modified"] = now
123 self
.db
.set_one(item
, {"_id": _id
}, _desc
)
125 # except DbException as e:
126 # self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
129 class TaskRegistry(LcmBase
):
131 Implements a registry of task needed for later cancelation, look for related tasks that must be completed before
132 etc. It stores a four level dict
133 First level is the topic, ns, vim_account, sdn
134 Second level is the _id
135 Third level is the operation id
136 Fourth level is a descriptive name, the value is the task class
138 The HA (High-Availability) methods are used when more than one LCM instance is running.
139 To register the current task in the external DB, use LcmBase as base class, to be able
140 to reuse LcmBase.update_db_2()
141 The DB registry uses the following fields to distinguish a task:
142 - op_type: operation type ("nslcmops" or "nsilcmops")
143 - op_id: operation ID
144 - worker: the worker ID for this process
147 # NS/NSI: "services" VIM/WIM/SDN: "accounts"
148 topic_service_list
= ['ns', 'nsi']
149 topic_account_list
= ['vim', 'wim', 'sdn', 'k8scluster', 'vca', 'k8srepo']
151 # Map topic to InstanceID
152 topic2instid_dict
= {
153 'ns': 'nsInstanceId',
154 'nsi': 'netsliceInstanceId'}
156 # Map topic to DB table name
157 topic2dbtable_dict
= {
160 'vim': 'vim_accounts',
161 'wim': 'wim_accounts',
163 'k8scluster': 'k8sclusters',
165 'k8srepo': 'k8srepos'}
167 def __init__(self
, worker_id
=None, logger
=None):
168 self
.task_registry
= {
178 self
.worker_id
= worker_id
179 self
.db
= Database().instance
.db
182 def register(self
, topic
, _id
, op_id
, task_name
, task
):
185 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
186 :param _id: _id of the related item
187 :param op_id: id of the operation of the related item
188 :param task_name: Task descriptive name, as create, instantiate, terminate. Must be unique in this op_id
189 :param task: Task class
192 if _id
not in self
.task_registry
[topic
]:
193 self
.task_registry
[topic
][_id
] = OrderedDict()
194 if op_id
not in self
.task_registry
[topic
][_id
]:
195 self
.task_registry
[topic
][_id
][op_id
] = {task_name
: task
}
197 self
.task_registry
[topic
][_id
][op_id
][task_name
] = task
198 # print("registering task", topic, _id, op_id, task_name, task)
200 def remove(self
, topic
, _id
, op_id
, task_name
=None):
202 When task is ended, it should be removed. It ignores missing tasks. It also removes tasks done with this _id
203 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
204 :param _id: _id of the related item
205 :param op_id: id of the operation of the related item
206 :param task_name: Task descriptive name. If none it deletes all tasks with same _id and op_id
209 if not self
.task_registry
[topic
].get(_id
):
212 self
.task_registry
[topic
][_id
].pop(op_id
, None)
213 elif self
.task_registry
[topic
][_id
].get(op_id
):
214 self
.task_registry
[topic
][_id
][op_id
].pop(task_name
, None)
217 for op_id_
in list(self
.task_registry
[topic
][_id
]):
218 for name
, task
in self
.task_registry
[topic
][_id
][op_id_
].items():
222 del self
.task_registry
[topic
][_id
][op_id_
]
223 if not self
.task_registry
[topic
][_id
]:
224 del self
.task_registry
[topic
][_id
]
226 def lookfor_related(self
, topic
, _id
, my_op_id
=None):
229 if _id
not in self
.task_registry
[topic
]:
230 return "", task_name_list
231 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
233 if my_op_id
== op_id
:
234 my_op_id
= None # so that the next task is taken
237 for task_name
, task
in self
.task_registry
[topic
][_id
][op_id
].items():
239 task_list
.append(task
)
240 task_name_list
.append(task_name
)
242 return ", ".join(task_name_list
), task_list
244 def cancel(self
, topic
, _id
, target_op_id
=None, target_task_name
=None):
246 Cancel all active tasks of a concrete ns, nsi, vim_account, sdn identified for _id. If op_id is supplied only
247 this is cancelled, and the same with task_name
249 if not self
.task_registry
[topic
].get(_id
):
251 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
252 if target_op_id
and target_op_id
!= op_id
:
254 for task_name
, task
in self
.task_registry
[topic
][_id
][op_id
].items():
255 if target_task_name
and target_task_name
!= task_name
:
260 # self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, op_id, task_name))
263 def _is_service_type_HA(self
, topic
):
264 return topic
in self
.topic_service_list
266 # Is topic VIM/WIM/SDN?
267 def _is_account_type_HA(self
, topic
):
268 return topic
in self
.topic_account_list
270 # Input: op_id, example: 'abc123def:3' Output: account_id='abc123def', op_index=3
271 def _get_account_and_op_HA(self
, op_id
):
274 account_id
, _
, op_index
= op_id
.rpartition(':')
275 if not account_id
or not op_index
.isdigit():
277 return account_id
, op_index
279 # Get '_id' for any topic and operation
280 def _get_instance_id_HA(self
, topic
, op_type
, op_id
):
282 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
285 # NS/NSI: Use op_id as '_id'
286 elif self
._is
_service
_type
_HA
(topic
):
288 # VIM/SDN/WIM/K8SCLUSTER: Split op_id to get Account ID and Operation Index, use Account ID as '_id'
289 elif self
._is
_account
_type
_HA
(topic
):
290 _id
, _
= self
._get
_account
_and
_op
_HA
(op_id
)
293 # Set DB _filter for querying any related process state
294 def _get_waitfor_filter_HA(self
, db_lcmop
, topic
, op_type
, op_id
):
296 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
297 # In this special case, the timestamp is ignored
299 _filter
= {'operationState': 'PROCESSING'}
300 # Otherwise, get 'startTime' timestamp for this operation
303 if self
._is
_service
_type
_HA
(topic
):
305 starttime_this_op
= db_lcmop
.get("startTime")
306 instance_id_label
= self
.topic2instid_dict
.get(topic
)
307 instance_id
= db_lcmop
.get(instance_id_label
)
308 _filter
= {instance_id_label
: instance_id
,
309 'operationState': 'PROCESSING',
310 'startTime.lt': starttime_this_op
,
311 "_admin.modified.gt": now
- 2*3600, # ignore if tow hours of inactivity
313 # VIM/WIM/SDN/K8scluster
314 elif self
._is
_account
_type
_HA
(topic
):
315 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
316 _ops
= db_lcmop
['_admin']['operations']
317 _this_op
= _ops
[int(op_index
)]
318 starttime_this_op
= _this_op
.get('startTime', None)
319 _filter
= {'operationState': 'PROCESSING',
320 'startTime.lt': starttime_this_op
}
323 # Get DB params for any topic and operation
324 def _get_dbparams_for_lock_HA(self
, topic
, op_type
, op_id
):
328 if self
._is
_service
_type
_HA
(topic
):
329 q_filter
= {'_id': op_id
, '_admin.worker': None}
330 update_dict
= {'_admin.worker': self
.worker_id
}
332 elif self
._is
_account
_type
_HA
(topic
):
333 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
336 if op_type
== 'create':
337 # Creating a VIM/WIM/SDN account implies setting '_admin.current_operation' = 0
339 q_filter
= {'_id': account_id
, "_admin.operations.{}.worker".format(op_index
): None}
340 update_dict
= {'_admin.operations.{}.worker'.format(op_index
): self
.worker_id
,
341 '_admin.current_operation': op_index
}
342 return q_filter
, update_dict
344 def lock_HA(self
, topic
, op_type
, op_id
):
346 Lock a task, if possible, to indicate to the HA system that
347 the task will be executed in this LCM instance.
348 :param topic: Can be "ns", "nsi", "vim", "wim", or "sdn"
349 :param op_type: Operation type, can be "nslcmops", "nsilcmops", "create", "edit", "delete"
350 :param op_id: NS, NSI: Operation ID VIM,WIM,SDN: Account ID + ':' + Operation Index
352 True=lock was successful => execute the task (not registered by any other LCM instance)
353 False=lock failed => do NOT execute the task (already registered by another LCM instance)
355 HA tasks and backward compatibility:
356 If topic is "account type" (VIM/WIM/SDN) and op_id is None, 'op_id' was not provided by NBI.
357 This means that the running NBI instance does not support HA.
358 In such a case this method should always return True, to always execute
359 the task in this instance of LCM, without querying the DB.
362 # Backward compatibility for VIM/WIM/SDN/k8scluster without op_id
363 if self
._is
_account
_type
_HA
(topic
) and op_id
is None:
366 # Try to lock this task
367 db_table_name
= self
.topic2dbtable_dict
[topic
]
368 q_filter
, update_dict
= self
._get
_dbparams
_for
_lock
_HA
(topic
, op_type
, op_id
)
369 db_lock_task
= self
.db
.set_one(db_table_name
,
371 update_dict
=update_dict
,
373 if db_lock_task
is None:
374 self
.logger
.debug("Task {} operation={} already locked by another worker".format(topic
, op_id
))
377 # Set 'detailed-status' to 'In progress' for VIM/WIM/SDN operations
378 if self
._is
_account
_type
_HA
(topic
):
379 detailed_status
= 'In progress'
380 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
381 q_filter
= {'_id': account_id
}
382 update_dict
= {'_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
383 self
.db
.set_one(db_table_name
,
385 update_dict
=update_dict
,
389 def unlock_HA(self
, topic
, op_type
, op_id
, operationState
, detailed_status
):
391 Register a task, done when finished a VIM/WIM/SDN 'create' operation.
392 :param topic: Can be "vim", "wim", or "sdn"
393 :param op_type: Operation type, can be "create", "edit", "delete"
394 :param op_id: Account ID + ':' + Operation Index
398 # Backward compatibility
399 if not self
._is
_account
_type
_HA
(topic
) or not op_id
:
402 # Get Account ID and Operation Index
403 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
404 db_table_name
= self
.topic2dbtable_dict
[topic
]
406 # If this is a 'delete' operation, the account may have been deleted (SUCCESS) or may still exist (FAILED)
407 # If the account exist, register the HA task.
408 # Update DB for HA tasks
409 q_filter
= {'_id': account_id
}
410 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
411 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
,
412 '_admin.operations.{}.worker'.format(op_index
): None,
413 '_admin.current_operation': None}
414 self
.db
.set_one(db_table_name
,
416 update_dict
=update_dict
,
420 async def waitfor_related_HA(self
, topic
, op_type
, op_id
=None):
422 Wait for any pending related HA tasks
425 # Backward compatibility
426 if not (self
._is
_service
_type
_HA
(topic
) or self
._is
_account
_type
_HA
(topic
)) and (op_id
is None):
430 db_table_name
= self
.topic2dbtable_dict
.get(topic
)
433 _id
= self
._get
_instance
_id
_HA
(topic
, op_type
, op_id
)
434 _filter
= {"_id": _id
}
435 db_lcmop
= self
.db
.get_one(db_table_name
,
441 # Set DB _filter for querying any related process state
442 _filter
= self
._get
_waitfor
_filter
_HA
(db_lcmop
, topic
, op_type
, op_id
)
444 # For HA, get list of tasks from DB instead of from dictionary (in-memory) variable.
445 timeout_wait_for_task
= 3600 # Max time (seconds) to wait for a related task to finish
446 # interval_wait_for_task = 30 # A too long polling interval slows things down considerably
447 interval_wait_for_task
= 10 # Interval in seconds for polling related tasks
448 time_left
= timeout_wait_for_task
449 old_num_related_tasks
= 0
451 # Get related tasks (operations within the same instance as this) which are
452 # still running (operationState='PROCESSING') and which were started before this task.
453 # In the case of op_type='ANY', get any related tasks with operationState='PROCESSING', ignore timestamps.
454 db_waitfor_related_task
= self
.db
.get_list(db_table_name
,
456 new_num_related_tasks
= len(db_waitfor_related_task
)
457 # If there are no related tasks, there is nothing to wait for, so return.
458 if not new_num_related_tasks
:
460 # If number of pending related tasks have changed,
461 # update the 'detailed-status' field and log the change.
462 # Do NOT update the 'detailed-status' for SDNC-associated-to-VIM operations ('ANY').
463 if (op_type
!= 'ANY') and (new_num_related_tasks
!= old_num_related_tasks
):
464 step
= "Waiting for {} related tasks to be completed.".format(new_num_related_tasks
)
466 q_filter
= {'_id': _id
}
468 if self
._is
_service
_type
_HA
(topic
):
469 update_dict
= {'detailed-status': step
, 'queuePosition': new_num_related_tasks
}
471 elif self
._is
_account
_type
_HA
(topic
):
472 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
473 update_dict
= {'_admin.operations.{}.detailed-status'.format(op_index
): step
}
474 self
.logger
.debug("Task {} operation={} {}".format(topic
, _id
, step
))
475 self
.db
.set_one(db_table_name
,
477 update_dict
=update_dict
,
479 old_num_related_tasks
= new_num_related_tasks
480 time_left
-= interval_wait_for_task
483 "Timeout ({}) when waiting for related tasks to be completed".format(
484 timeout_wait_for_task
))
485 await asyncio
.sleep(interval_wait_for_task
)