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 # LcmBase must be listed before TaskRegistry, as it is a dependency.
55 def __init__(self
, db
, msg
, fs
, logger
):
58 :param db: database connection
65 def update_db_2(self
, item
, _id
, _desc
):
67 Updates database with _desc information. If success _desc is cleared
70 :param _desc: dictionary with the content to update. Keys are dot separated keys for
71 :return: None. Exception is raised on error
76 _desc
["_admin.modified"] = now
77 self
.db
.set_one(item
, {"_id": _id
}, _desc
)
79 # except DbException as e:
80 # self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
83 class TaskRegistry(LcmBase
):
85 Implements a registry of task needed for later cancelation, look for related tasks that must be completed before
86 etc. It stores a four level dict
87 First level is the topic, ns, vim_account, sdn
88 Second level is the _id
89 Third level is the operation id
90 Fourth level is a descriptive name, the value is the task class
92 The HA (High-Availability) methods are used when more than one LCM instance is running.
93 To register the current task in the external DB, use LcmBase as base class, to be able
94 to reuse LcmBase.update_db_2()
95 The DB registry uses the following fields to distinguish a task:
96 - op_type: operation type ("nslcmops" or "nsilcmops")
98 - worker: the worker ID for this process
101 # NS/NSI: "services" VIM/WIM/SDN: "accounts"
102 topic_service_list
= ['ns', 'nsi']
103 topic_account_list
= ['vim', 'wim', 'sdn', 'k8scluster', 'k8srepo']
105 # Map topic to InstanceID
106 topic2instid_dict
= {
107 'ns': 'nsInstanceId',
108 'nsi': 'netsliceInstanceId'}
110 # Map topic to DB table name
111 topic2dbtable_dict
= {
114 'vim': 'vim_accounts',
115 'wim': 'wim_accounts',
117 'k8scluster': 'k8sclusters',
118 'k8srepo': 'k8srepos'}
120 def __init__(self
, worker_id
=None, db
=None, logger
=None):
121 self
.task_registry
= {
130 self
.worker_id
= worker_id
134 def register(self
, topic
, _id
, op_id
, task_name
, task
):
137 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
138 :param _id: _id of the related item
139 :param op_id: id of the operation of the related item
140 :param task_name: Task descriptive name, as create, instantiate, terminate. Must be unique in this op_id
141 :param task: Task class
144 if _id
not in self
.task_registry
[topic
]:
145 self
.task_registry
[topic
][_id
] = OrderedDict()
146 if op_id
not in self
.task_registry
[topic
][_id
]:
147 self
.task_registry
[topic
][_id
][op_id
] = {task_name
: task
}
149 self
.task_registry
[topic
][_id
][op_id
][task_name
] = task
150 # print("registering task", topic, _id, op_id, task_name, task)
152 def remove(self
, topic
, _id
, op_id
, task_name
=None):
154 When task is ended, it should be removed. It ignores missing tasks. It also removes tasks done with this _id
155 :param topic: Can be "ns", "nsi", "vim_account", "sdn"
156 :param _id: _id of the related item
157 :param op_id: id of the operation of the related item
158 :param task_name: Task descriptive name. If none it deletes all tasks with same _id and op_id
161 if not self
.task_registry
[topic
].get(_id
):
164 self
.task_registry
[topic
][_id
].pop(op_id
, None)
165 elif self
.task_registry
[topic
][_id
].get(op_id
):
166 self
.task_registry
[topic
][_id
][op_id
].pop(task_name
, None)
169 for op_id_
in list(self
.task_registry
[topic
][_id
]):
170 for name
, task
in self
.task_registry
[topic
][_id
][op_id_
].items():
174 del self
.task_registry
[topic
][_id
][op_id_
]
175 if not self
.task_registry
[topic
][_id
]:
176 del self
.task_registry
[topic
][_id
]
178 def lookfor_related(self
, topic
, _id
, my_op_id
=None):
181 if _id
not in self
.task_registry
[topic
]:
182 return "", task_name_list
183 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
185 if my_op_id
== op_id
:
186 my_op_id
= None # so that the next task is taken
189 for task_name
, task
in self
.task_registry
[topic
][_id
][op_id
].items():
191 task_list
.append(task
)
192 task_name_list
.append(task_name
)
194 return ", ".join(task_name_list
), task_list
196 def cancel(self
, topic
, _id
, target_op_id
=None, target_task_name
=None):
198 Cancel all active tasks of a concrete ns, nsi, vim_account, sdn identified for _id. If op_id is supplied only
199 this is cancelled, and the same with task_name
201 if not self
.task_registry
[topic
].get(_id
):
203 for op_id
in reversed(self
.task_registry
[topic
][_id
]):
204 if target_op_id
and target_op_id
!= op_id
:
206 for task_name
, task
in self
.task_registry
[topic
][_id
][op_id
].items():
207 if target_task_name
and target_task_name
!= task_name
:
212 # self.logger.debug("{} _id={} order_id={} task={} cancelled".format(topic, _id, op_id, task_name))
215 def _is_service_type_HA(self
, topic
):
216 return topic
in self
.topic_service_list
218 # Is topic VIM/WIM/SDN?
219 def _is_account_type_HA(self
, topic
):
220 return topic
in self
.topic_account_list
222 # Input: op_id, example: 'abc123def:3' Output: account_id='abc123def', op_index=3
223 def _get_account_and_op_HA(self
, op_id
):
226 account_id
, _
, op_index
= op_id
.rpartition(':')
229 if not op_index
.isdigit():
231 return account_id
, op_index
233 # Get '_id' for any topic and operation
234 def _get_instance_id_HA(self
, topic
, op_type
, op_id
):
236 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
239 # NS/NSI: Use op_id as '_id'
240 elif self
._is
_service
_type
_HA
(topic
):
242 # VIM/SDN/WIM/K8SCLUSTER: Split op_id to get Account ID and Operation Index, use Account ID as '_id'
243 elif self
._is
_account
_type
_HA
(topic
):
244 _id
, _
= self
._get
_account
_and
_op
_HA
(op_id
)
247 # Set DB _filter for querying any related process state
248 def _get_waitfor_filter_HA(self
, db_lcmop
, topic
, op_type
, op_id
):
250 # Special operation 'ANY', for SDN account associated to a VIM account: op_id as '_id'
251 # In this special case, the timestamp is ignored
253 _filter
= {'operationState': 'PROCESSING'}
254 # Otherwise, get 'startTime' timestamp for this operation
257 if self
._is
_service
_type
_HA
(topic
):
259 starttime_this_op
= db_lcmop
.get("startTime")
260 instance_id_label
= self
.topic2instid_dict
.get(topic
)
261 instance_id
= db_lcmop
.get(instance_id_label
)
262 _filter
= {instance_id_label
: instance_id
,
263 'operationState': 'PROCESSING',
264 'startTime.lt': starttime_this_op
,
265 "_admin.modified.gt": now
- 2*3600, # ignore if tow hours of inactivity
267 # VIM/WIM/SDN/K8scluster
268 elif self
._is
_account
_type
_HA
(topic
):
269 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
270 _ops
= db_lcmop
['_admin']['operations']
271 _this_op
= _ops
[int(op_index
)]
272 starttime_this_op
= _this_op
.get('startTime', None)
273 _filter
= {'operationState': 'PROCESSING',
274 'startTime.lt': starttime_this_op
}
277 # Get DB params for any topic and operation
278 def _get_dbparams_for_lock_HA(self
, topic
, op_type
, op_id
):
282 if self
._is
_service
_type
_HA
(topic
):
283 q_filter
= {'_id': op_id
, '_admin.worker': None}
284 update_dict
= {'_admin.worker': self
.worker_id
}
286 elif self
._is
_account
_type
_HA
(topic
):
287 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
290 if op_type
== 'create':
291 # Creating a VIM/WIM/SDN account implies setting '_admin.current_operation' = 0
293 q_filter
= {'_id': account_id
, "_admin.operations.{}.worker".format(op_index
): None}
294 update_dict
= {'_admin.operations.{}.worker'.format(op_index
): self
.worker_id
,
295 '_admin.current_operation': op_index
}
296 return q_filter
, update_dict
298 def lock_HA(self
, topic
, op_type
, op_id
):
300 Lock a task, if possible, to indicate to the HA system that
301 the task will be executed in this LCM instance.
302 :param topic: Can be "ns", "nsi", "vim", "wim", or "sdn"
303 :param op_type: Operation type, can be "nslcmops", "nsilcmops", "create", "edit", "delete"
304 :param op_id: NS, NSI: Operation ID VIM,WIM,SDN: Account ID + ':' + Operation Index
306 True=lock was successful => execute the task (not registered by any other LCM instance)
307 False=lock failed => do NOT execute the task (already registered by another LCM instance)
309 HA tasks and backward compatibility:
310 If topic is "account type" (VIM/WIM/SDN) and op_id is None, 'op_id' was not provided by NBI.
311 This means that the running NBI instance does not support HA.
312 In such a case this method should always return True, to always execute
313 the task in this instance of LCM, without querying the DB.
316 # Backward compatibility for VIM/WIM/SDN/k8scluster without op_id
317 if self
._is
_account
_type
_HA
(topic
) and op_id
is None:
320 # Try to lock this task
321 db_table_name
= self
.topic2dbtable_dict
.get(topic
)
322 q_filter
, update_dict
= self
._get
_dbparams
_for
_lock
_HA
(topic
, op_type
, op_id
)
323 db_lock_task
= self
.db
.set_one(db_table_name
,
325 update_dict
=update_dict
,
327 if db_lock_task
is None:
328 self
.logger
.debug("Task {} operation={} already locked by another worker".format(topic
, op_id
))
331 # Set 'detailed-status' to 'In progress' for VIM/WIM/SDN operations
332 if self
._is
_account
_type
_HA
(topic
):
333 detailed_status
= 'In progress'
334 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
335 q_filter
= {'_id': account_id
}
336 update_dict
= {'_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
337 self
.db
.set_one(db_table_name
,
339 update_dict
=update_dict
,
343 def register_HA(self
, topic
, op_type
, op_id
, operationState
, detailed_status
):
345 Register a task, done when finished a VIM/WIM/SDN 'create' operation.
346 :param topic: Can be "vim", "wim", or "sdn"
347 :param op_type: Operation type, can be "create", "edit", "delete"
348 :param op_id: Account ID + ':' + Operation Index
352 # Backward compatibility
353 if not self
._is
_account
_type
_HA
(topic
) or (self
._is
_account
_type
_HA
(topic
) and op_id
is None):
356 # Get Account ID and Operation Index
357 account_id
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
358 db_table_name
= self
.topic2dbtable_dict
.get(topic
)
360 # If this is a 'delete' operation, the account may have been deleted (SUCCESS) or may still exist (FAILED)
361 # If the account exist, register the HA task.
362 # Update DB for HA tasks
363 q_filter
= {'_id': account_id
}
364 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
365 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
366 self
.db
.set_one(db_table_name
,
368 update_dict
=update_dict
,
372 async def waitfor_related_HA(self
, topic
, op_type
, op_id
=None):
374 Wait for any pending related HA tasks
377 # Backward compatibility
378 if not (self
._is
_service
_type
_HA
(topic
) or self
._is
_account
_type
_HA
(topic
)) and (op_id
is None):
382 db_table_name
= self
.topic2dbtable_dict
.get(topic
)
385 _id
= self
._get
_instance
_id
_HA
(topic
, op_type
, op_id
)
386 _filter
= {"_id": _id
}
387 db_lcmop
= self
.db
.get_one(db_table_name
,
393 # Set DB _filter for querying any related process state
394 _filter
= self
._get
_waitfor
_filter
_HA
(db_lcmop
, topic
, op_type
, op_id
)
396 # For HA, get list of tasks from DB instead of from dictionary (in-memory) variable.
397 timeout_wait_for_task
= 3600 # Max time (seconds) to wait for a related task to finish
398 # interval_wait_for_task = 30 # A too long polling interval slows things down considerably
399 interval_wait_for_task
= 10 # Interval in seconds for polling related tasks
400 time_left
= timeout_wait_for_task
401 old_num_related_tasks
= 0
403 # Get related tasks (operations within the same instance as this) which are
404 # still running (operationState='PROCESSING') and which were started before this task.
405 # In the case of op_type='ANY', get any related tasks with operationState='PROCESSING', ignore timestamps.
406 db_waitfor_related_task
= self
.db
.get_list(db_table_name
,
408 new_num_related_tasks
= len(db_waitfor_related_task
)
409 # If there are no related tasks, there is nothing to wait for, so return.
410 if not new_num_related_tasks
:
412 # If number of pending related tasks have changed,
413 # update the 'detailed-status' field and log the change.
414 # Do NOT update the 'detailed-status' for SDNC-associated-to-VIM operations ('ANY').
415 if (op_type
!= 'ANY') and (new_num_related_tasks
!= old_num_related_tasks
):
416 step
= "Waiting for {} related tasks to be completed.".format(new_num_related_tasks
)
418 q_filter
= {'_id': _id
}
420 if self
._is
_service
_type
_HA
(topic
):
421 update_dict
= {'detailed-status': step
}
423 elif self
._is
_account
_type
_HA
(topic
):
424 _
, op_index
= self
._get
_account
_and
_op
_HA
(op_id
)
425 update_dict
= {'_admin.operations.{}.detailed-status'.format(op_index
): step
}
426 self
.logger
.debug("Task {} operation={} {}".format(topic
, _id
, step
))
427 self
.db
.set_one(db_table_name
,
429 update_dict
=update_dict
,
431 old_num_related_tasks
= new_num_related_tasks
432 time_left
-= interval_wait_for_task
435 "Timeout ({}) when waiting for related tasks to be completed".format(
436 timeout_wait_for_task
))
437 await asyncio
.sleep(interval_wait_for_task
)