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_common.dbbase import DbException
24 __author__
= "Alfonso Tierno"
27 class LcmException(Exception):
31 class LcmExceptionNoMgmtIP(LcmException
):
35 class LcmExceptionExit(LcmException
):
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
45 for point
in v
.split("."):
46 point
, _
, _
= point
.partition("+")
47 point
, _
, _
= point
.partition("-")
48 filled
.append(point
.zfill(20))
52 def deep_get(target_dict
, key_list
):
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 :return: The wanted value if exist, None otherwise
61 if not isinstance(target_dict
, dict) or key
not in target_dict
:
63 target_dict
= target_dict
[key
]
67 # LcmBase must be listed before TaskRegistry, as it is a dependency.
70 def __init__(self
, db
, msg
, fs
, logger
):
73 :param db: database connection
80 def update_db_2(self
, item
, _id
, _desc
):
82 Updates database with _desc information. If success _desc is cleared
85 :param _desc: dictionary with the content to update. Keys are dot separated keys for
86 :return: None. Exception is raised on error
91 _desc
["_admin.modified"] = now
92 self
.db
.set_one(item
, {"_id": _id
}, _desc
)
94 # except DbException as e:
95 # self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
98 class TaskRegistry(LcmBase
):
100 Implements a registry of task needed for later cancelation, look for related tasks that must be completed before
101 etc. It stores a four level dict
102 First level is the topic, ns, vim_account, sdn
103 Second level is the _id
104 Third level is the operation id
105 Fourth level is a descriptive name, the value is the task class
107 The HA (High-Availability) methods are used when more than one LCM instance is running.
108 To register the current task in the external DB, use LcmBase as base class, to be able
109 to reuse LcmBase.update_db_2()
110 The DB registry uses the following fields to distinguish a task:
111 - op_type: operation type ("nslcmops" or "nsilcmops")
112 - op_id: operation ID
113 - worker: the worker ID for this process
116 # NS/NSI: "services" VIM/WIM/SDN: "accounts"
117 topic_service_list
= ['ns', 'nsi']
118 topic_account_list
= ['vim', 'wim', 'sdn', 'k8scluster', 'k8srepo']
120 # Map topic to InstanceID
121 topic2instid_dict
= {
122 'ns': 'nsInstanceId',
123 'nsi': 'netsliceInstanceId'}
125 # Map topic to DB table name
126 topic2dbtable_dict
= {
129 'vim': 'vim_accounts',
130 'wim': 'wim_accounts',
132 'k8scluster': 'k8sclusters',
133 'k8srepo': 'k8srepos'}
135 def __init__(self
, worker_id
=None, db
=None, logger
=None):
136 self
.task_registry
= {
145 self
.worker_id
= worker_id
149 def register(self
, topic
, _id
, op_id
, task_name
, task
):
152 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
153 :param _id: _id of the related item
154 :param op_id: id of the operation of the related item
155 :param task_name: Task descriptive name, as create, instantiate, terminate. Must be unique in this op_id
156 :param task: Task class
159 if _id
not in self
.task_registry
[topic
]:
160 self
.task_registry
[topic
][_id
] = OrderedDict()
161 if op_id
not in self
.task_registry
[topic
][_id
]:
162 self
.task_registry
[topic
][_id
][op_id
] = {task_name
: task
}
164 self
.task_registry
[topic
][_id
][op_id
][task_name
] = task
165 # print("registering task", topic, _id, op_id, task_name, task)
167 def remove(self
, topic
, _id
, op_id
, task_name
=None):
169 When task is ended, it should be removed. It ignores missing tasks. It also removes tasks done with this _id
170 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
171 :param _id: _id of the related item
172 :param op_id: id of the operation of the related item
173 :param task_name: Task descriptive name. If none it deletes all tasks with same _id and op_id
176 if not self
.task_registry
[topic
].get(_id
):
179 self
.task_registry
[topic
][_id
].pop(op_id
, None)
180 elif self
.task_registry
[topic
][_id
].get(op_id
):
181 self
.task_registry
[topic
][_id
][op_id
].pop(task_name
, None)
184 for op_id_
in list(self
.task_registry
[topic
][_id
]):
185 for name
, task
in self
.task_registry
[topic
][_id
][op_id_
].items():
189 del self
.task_registry
[topic
][_id
][op_id_
]
190 if not self
.task_registry
[topic
][_id
]:
191 del self
.task_registry
[topic
][_id
]
193 def lookfor_related(self
, topic
, _id
, my_op_id
=None):
196 if _id
not in self
.task_registry
[topic
]:
197 return "", task_name_list
198 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
200 if my_op_id
== op_id
:
201 my_op_id
= None # so that the next task is taken
204 for task_name
, task
in self
.task_registry
[topic
][_id
][op_id
].items():
206 task_list
.append(task
)
207 task_name_list
.append(task_name
)
209 return ", ".join(task_name_list
), task_list
211 def cancel(self
, topic
, _id
, target_op_id
=None, target_task_name
=None):
213 Cancel all active tasks of a concrete ns, nsi, vim_account, sdn identified for _id. If op_id is supplied only
214 this is cancelled, and the same with task_name
216 if not self
.task_registry
[topic
].get(_id
):
218 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
219 if target_op_id
and target_op_id
!= op_id
:
221 for task_name
, task
in self
.task_registry
[topic
][_id
][op_id
].items():
222 if target_task_name
and target_task_name
!= task_name
:
227 # self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, op_id, task_name))
230 def _is_service_type_HA(self
, topic
):
231 return topic
in self
.topic_service_list
233 # Is topic VIM/WIM/SDN?
234 def _is_account_type_HA(self
, topic
):
235 return topic
in self
.topic_account_list
237 # Input: op_id, example: 'abc123def:3' Output: account_id='abc123def', op_index=3
238 def _get_account_and_op_HA(self
, op_id
):
241 account_id
, _
, op_index
= op_id
.rpartition(':')
244 if not op_index
.isdigit():
246 return account_id
, op_index
248 # Get '_id' for any topic and operation
249 def _get_instance_id_HA(self
, topic
, op_type
, op_id
):
251 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
254 # NS/NSI: Use op_id as '_id'
255 elif self
._is
_service
_type
_HA
(topic
):
257 # VIM/SDN/WIM/K8SCLUSTER: Split op_id to get Account ID and Operation Index, use Account ID as '_id'
258 elif self
._is
_account
_type
_HA
(topic
):
259 _id
, _
= self
._get
_account
_and
_op
_HA
(op_id
)
262 # Set DB _filter for querying any related process state
263 def _get_waitfor_filter_HA(self
, db_lcmop
, topic
, op_type
, op_id
):
265 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
266 # In this special case, the timestamp is ignored
268 _filter
= {'operationState': 'PROCESSING'}
269 # Otherwise, get 'startTime' timestamp for this operation
272 if self
._is
_service
_type
_HA
(topic
):
274 starttime_this_op
= db_lcmop
.get("startTime")
275 instance_id_label
= self
.topic2instid_dict
.get(topic
)
276 instance_id
= db_lcmop
.get(instance_id_label
)
277 _filter
= {instance_id_label
: instance_id
,
278 'operationState': 'PROCESSING',
279 'startTime.lt': starttime_this_op
,
280 "_admin.modified.gt": now
- 2*3600, # ignore if tow hours of inactivity
282 # VIM/WIM/SDN/K8scluster
283 elif self
._is
_account
_type
_HA
(topic
):
284 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
285 _ops
= db_lcmop
['_admin']['operations']
286 _this_op
= _ops
[int(op_index
)]
287 starttime_this_op
= _this_op
.get('startTime', None)
288 _filter
= {'operationState': 'PROCESSING',
289 'startTime.lt': starttime_this_op
}
292 # Get DB params for any topic and operation
293 def _get_dbparams_for_lock_HA(self
, topic
, op_type
, op_id
):
297 if self
._is
_service
_type
_HA
(topic
):
298 q_filter
= {'_id': op_id
, '_admin.worker': None}
299 update_dict
= {'_admin.worker': self
.worker_id
}
301 elif self
._is
_account
_type
_HA
(topic
):
302 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
305 if op_type
== 'create':
306 # Creating a VIM/WIM/SDN account implies setting '_admin.current_operation' = 0
308 q_filter
= {'_id': account_id
, "_admin.operations.{}.worker".format(op_index
): None}
309 update_dict
= {'_admin.operations.{}.worker'.format(op_index
): self
.worker_id
,
310 '_admin.current_operation': op_index
}
311 return q_filter
, update_dict
313 def lock_HA(self
, topic
, op_type
, op_id
):
315 Lock a task, if possible, to indicate to the HA system that
316 the task will be executed in this LCM instance.
317 :param topic: Can be "ns", "nsi", "vim", "wim", or "sdn"
318 :param op_type: Operation type, can be "nslcmops", "nsilcmops", "create", "edit", "delete"
319 :param op_id: NS, NSI: Operation ID VIM,WIM,SDN: Account ID + ':' + Operation Index
321 True=lock was successful => execute the task (not registered by any other LCM instance)
322 False=lock failed => do NOT execute the task (already registered by another LCM instance)
324 HA tasks and backward compatibility:
325 If topic is "account type" (VIM/WIM/SDN) and op_id is None, 'op_id' was not provided by NBI.
326 This means that the running NBI instance does not support HA.
327 In such a case this method should always return True, to always execute
328 the task in this instance of LCM, without querying the DB.
331 # Backward compatibility for VIM/WIM/SDN/k8scluster without op_id
332 if self
._is
_account
_type
_HA
(topic
) and op_id
is None:
335 # Try to lock this task
336 db_table_name
= self
.topic2dbtable_dict
.get(topic
)
337 q_filter
, update_dict
= self
._get
_dbparams
_for
_lock
_HA
(topic
, op_type
, op_id
)
338 db_lock_task
= self
.db
.set_one(db_table_name
,
340 update_dict
=update_dict
,
342 if db_lock_task
is None:
343 self
.logger
.debug("Task {} operation={} already locked by another worker".format(topic
, op_id
))
346 # Set 'detailed-status' to 'In progress' for VIM/WIM/SDN operations
347 if self
._is
_account
_type
_HA
(topic
):
348 detailed_status
= 'In progress'
349 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
350 q_filter
= {'_id': account_id
}
351 update_dict
= {'_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
352 self
.db
.set_one(db_table_name
,
354 update_dict
=update_dict
,
358 def register_HA(self
, topic
, op_type
, op_id
, operationState
, detailed_status
):
360 Register a task, done when finished a VIM/WIM/SDN 'create' operation.
361 :param topic: Can be "vim", "wim", or "sdn"
362 :param op_type: Operation type, can be "create", "edit", "delete"
363 :param op_id: Account ID + ':' + Operation Index
367 # Backward compatibility
368 if not self
._is
_account
_type
_HA
(topic
) or (self
._is
_account
_type
_HA
(topic
) and op_id
is None):
371 # Get Account ID and Operation Index
372 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
373 db_table_name
= self
.topic2dbtable_dict
.get(topic
)
375 # If this is a 'delete' operation, the account may have been deleted (SUCCESS) or may still exist (FAILED)
376 # If the account exist, register the HA task.
377 # Update DB for HA tasks
378 q_filter
= {'_id': account_id
}
379 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
380 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
381 self
.db
.set_one(db_table_name
,
383 update_dict
=update_dict
,
387 async def waitfor_related_HA(self
, topic
, op_type
, op_id
=None):
389 Wait for any pending related HA tasks
392 # Backward compatibility
393 if not (self
._is
_service
_type
_HA
(topic
) or self
._is
_account
_type
_HA
(topic
)) and (op_id
is None):
397 db_table_name
= self
.topic2dbtable_dict
.get(topic
)
400 _id
= self
._get
_instance
_id
_HA
(topic
, op_type
, op_id
)
401 _filter
= {"_id": _id
}
402 db_lcmop
= self
.db
.get_one(db_table_name
,
408 # Set DB _filter for querying any related process state
409 _filter
= self
._get
_waitfor
_filter
_HA
(db_lcmop
, topic
, op_type
, op_id
)
411 # For HA, get list of tasks from DB instead of from dictionary (in-memory) variable.
412 timeout_wait_for_task
= 3600 # Max time (seconds) to wait for a related task to finish
413 # interval_wait_for_task = 30 # A too long polling interval slows things down considerably
414 interval_wait_for_task
= 10 # Interval in seconds for polling related tasks
415 time_left
= timeout_wait_for_task
416 old_num_related_tasks
= 0
418 # Get related tasks (operations within the same instance as this) which are
419 # still running (operationState='PROCESSING') and which were started before this task.
420 # In the case of op_type='ANY', get any related tasks with operationState='PROCESSING', ignore timestamps.
421 db_waitfor_related_task
= self
.db
.get_list(db_table_name
,
423 new_num_related_tasks
= len(db_waitfor_related_task
)
424 # If there are no related tasks, there is nothing to wait for, so return.
425 if not new_num_related_tasks
:
427 # If number of pending related tasks have changed,
428 # update the 'detailed-status' field and log the change.
429 # Do NOT update the 'detailed-status' for SDNC-associated-to-VIM operations ('ANY').
430 if (op_type
!= 'ANY') and (new_num_related_tasks
!= old_num_related_tasks
):
431 step
= "Waiting for {} related tasks to be completed.".format(new_num_related_tasks
)
433 q_filter
= {'_id': _id
}
435 if self
._is
_service
_type
_HA
(topic
):
436 update_dict
= {'detailed-status': step
}
438 elif self
._is
_account
_type
_HA
(topic
):
439 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
440 update_dict
= {'_admin.operations.{}.detailed-status'.format(op_index
): step
}
441 self
.logger
.debug("Task {} operation={} {}".format(topic
, _id
, step
))
442 self
.db
.set_one(db_table_name
,
444 update_dict
=update_dict
,
446 old_num_related_tasks
= new_num_related_tasks
447 time_left
-= interval_wait_for_task
450 "Timeout ({}) when waiting for related tasks to be completed".format(
451 timeout_wait_for_task
))
452 await asyncio
.sleep(interval_wait_for_task
)