--- /dev/null
+# -*- coding: utf-8 -*-
+##
+# Copyright 2018 University of Bristol - High Performance Networks Research
+# Group
+# All Rights Reserved.
+#
+# Contributors: Anderson Bravalheri, Dimitrios Gkounis, Abubakar Siddique
+# Muqaddas, Navdeep Uniyal, Reza Nejabati and Dimitra Simeonidou
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact with: <highperformance-networks@bristol.ac.uk>
+#
+# Neither the name of the University of Bristol nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# This work has been performed in the context of DCMS UK 5G Testbeds
+# & Trials Programme and in the framework of the Metro-Haul project -
+# funded by the European Commission under Grant number 761727 through the
+# Horizon 2020 and 5G-PPP programmes.
+##
+# pylint: disable=E1101,E0203,W0201
+
+"""Common logic for task management"""
+import logging
+from time import time
+
+import yaml
+
+from ..utils import (
+ filter_dict_keys,
+ filter_out_dict_keys,
+ merge_dicts,
+ remove_none_items,
+ truncate
+)
+
+PENDING, REFRESH, IGNORE = range(3)
+
+TIMEOUT = 1 * 60 * 60 # 1 hour
+MIN_ATTEMPTS = 10
+
+
+class Action(object):
+ """Create a basic object representing the action record.
+
+ Arguments:
+ record (dict): record as returned by the database
+ **kwargs: extra keyword arguments to overwrite the fields in record
+ """
+
+ PROPERTIES = [
+ 'task_index', # MD - Index number of the task.
+ # This together with the instance_action_id
+ # forms a unique key identifier
+ 'action', # MD - CREATE, DELETE, FIND
+ 'item', # MD - table name, eg. instance_wim_nets
+ 'item_id', # MD - uuid of the referenced entry in the
+ # previous table
+ 'instance_action_id', # MD - reference to a cohesive group of actions
+ # related to the same instance-scenario
+ 'wim_account_id', # MD - reference to the WIM account used
+ # by the thread/connector
+ 'wim_internal_id', # MD - internal ID used by the WIM to refer to
+ # the item
+ 'datacenter_vim_id', # MD - reference to the VIM account used
+ # by the thread/connector
+ 'vim_id', # MD - internal ID used by the VIM to refer to
+ # the item
+ 'status', # MD - SCHEDULED,BUILD,DONE,FAILED,SUPERSEDED
+ 'extra', # MD - text with yaml format at database,
+ # dict at memory with:
+ # `- params: list with the params to be sent to the VIM for CREATE
+ # or FIND. For DELETE the vim_id is taken from other
+ # related tasks
+ # `- find: (only for CREATE tasks) if present it should FIND
+ # before creating and use if existing.
+ # Contains the FIND params
+ # `- depends_on: list with the 'task_index'es of tasks that must be
+ # completed before. e.g. a vm creation depends on a net
+ # creation
+ # `- sdn_net_id: used for net.
+ # `- tries
+ # `- created_items:
+ # dictionary with extra elements created that need
+ # to be deleted. e.g. ports,
+ # `- volumes,...
+ # `- created: False if the VIM element is not created by
+ # other actions, and it should not be deleted
+ # `- wim_status: WIM status of the element. Stored also at database
+ # in the item table
+ 'params', # M - similar to extra[params]
+ 'depends_on', # M - similar to extra[depends_on]
+ 'depends', # M - dict with task_index(from depends_on) to
+ # task class
+ 'error_msg', # MD - descriptive text upon an error
+ 'created_at', # MD - task DB creation time
+ 'modified_at', # MD - last DB update time
+ 'process_at', # M - unix epoch when to process the task
+ ]
+
+ __slots__ = PROPERTIES + [
+ 'logger',
+ ]
+
+ def __init__(self, record, logger=None, **kwargs):
+ self.logger = logger or logging.getLogger('openmano.wim.action')
+ attrs = merge_dicts(dict.fromkeys(self.PROPERTIES), record, kwargs)
+ self.update(_expand_extra(attrs))
+
+ def __repr__(self):
+ return super(Action, self).__repr__() + repr(self.as_dict())
+
+ def as_dict(self, *fields):
+ """Representation of the object as a dict"""
+ attrs = (set(self.PROPERTIES) & set(fields)
+ if fields else self.PROPERTIES)
+ return {k: getattr(self, k) for k in attrs}
+
+ def as_record(self):
+ """Returns a dict that can be send to the persistence layer"""
+ special = ['params', 'depends_on', 'depends']
+ record = self.as_dict()
+ record['extra'].update(self.as_dict(*special))
+ non_fields = special + ['process_at']
+
+ return remove_none_items(filter_out_dict_keys(record, non_fields))
+
+ def update(self, values=None, **kwargs):
+ """Update the in-memory representation of the task (works similarly to
+ dict.update). The update is NOT automatically persisted.
+ """
+ # "white-listed mass assignment"
+ updates = merge_dicts(values, kwargs)
+ for attr in set(self.PROPERTIES) & set(updates.keys()):
+ setattr(self, attr, updates[attr])
+
+ def save(self, persistence, **kwargs):
+ """Persist current state of the object to the database.
+
+ Arguments:
+ persistence: object encapsulating the database
+ **kwargs: extra properties to be updated before saving
+
+ Note:
+ If any key word argument is passed, the object itself will be
+ changed as an extra side-effect.
+ """
+ action_id = self.instance_action_id
+ index = self.task_index
+ if kwargs:
+ self.update(kwargs)
+ properties = self.as_record()
+
+ return persistence.update_action(action_id, index, properties)
+
+ def fail(self, persistence, reason, status='FAILED'):
+ """Mark action as FAILED, updating tables accordingly"""
+ persistence.update_instance_action_counters(
+ self.instance_action_id,
+ failed=1,
+ done=(-1 if self.status == 'DONE' else 0))
+
+ self.status = status
+ self.error_msg = truncate(reason)
+ self.logger.error('%s %s: %s', self.id, status, reason)
+ return self.save(persistence)
+
+ def succeed(self, persistence, status='DONE'):
+ """Mark action as DONE, updating tables accordingly"""
+ persistence.update_instance_action_counters(
+ self.instance_action_id, done=1)
+ self.status = status
+ self.logger.debug('%s %s', self.id, status)
+ return self.save(persistence)
+
+ def defer(self, persistence, reason,
+ timeout=TIMEOUT, min_attempts=MIN_ATTEMPTS):
+ """Postpone the task processing, taking care to not timeout.
+
+ Arguments:
+ persistence: object encapsulating the database
+ reason (str): explanation for the delay
+ timeout (int): maximum delay tolerated since the first attempt.
+ Note that this number is a time delta, in seconds
+ min_attempts (int): Number of attempts to try before giving up.
+ """
+ now = time()
+ last_attempt = self.extra.get('last_attempted_at') or time()
+ attempts = self.extra.get('attempts') or 0
+
+ if last_attempt - now > timeout and attempts > min_attempts:
+ self.fail(persistence,
+ 'Timeout reached. {} attempts in the last {:d} min'
+ .format(attempts, last_attempt / 60))
+
+ self.extra['last_attempted_at'] = time()
+ self.extra['attempts'] = attempts + 1
+ self.logger.info('%s DEFERRED: %s', self.id, reason)
+ return self.save(persistence)
+
+ @property
+ def group_key(self):
+ """Key defining the group to which this tasks belongs"""
+ return (self.item, self.item_id)
+
+ @property
+ def processing(self):
+ """Processing status for the task (PENDING, REFRESH, IGNORE)"""
+ if self.status == 'SCHEDULED':
+ return PENDING
+
+ return IGNORE
+
+ @property
+ def id(self):
+ """Unique identifier of this particular action"""
+ return '{}[{}]'.format(self.instance_action_id, self.task_index)
+
+ @property
+ def is_scheduled(self):
+ return self.status == 'SCHEDULED'
+
+ @property
+ def is_build(self):
+ return self.status == 'BUILD'
+
+ @property
+ def is_done(self):
+ return self.status == 'DONE'
+
+ @property
+ def is_failed(self):
+ return self.status == 'FAILED'
+
+ @property
+ def is_superseded(self):
+ return self.status == 'SUPERSEDED'
+
+ def refresh(self, connector, persistence):
+ """Use the connector/persistence to refresh the status of the item.
+
+ After the item status is refreshed any change in the task should be
+ persisted to the database.
+
+ Arguments:
+ connector: object containing the classes to access the WIM or VIM
+ persistence: object containing the methods necessary to query the
+ database and to persist the updates
+ """
+ self.logger.debug(
+ 'Action `%s` has no refresh to be done',
+ self.__class__.__name__)
+
+ def expand_dependency_links(self, task_group):
+ """Expand task indexes into actual IDs"""
+ if not self.depends_on or (
+ isinstance(self.depends, dict) and self.depends):
+ return
+
+ num_tasks = len(task_group)
+ references = {
+ "TASK-{}".format(i): task_group[i]
+ for i in self.depends_on
+ if i < num_tasks and task_group[i].task_index == i and
+ task_group[i].instance_action_id == self.instance_action_id
+ }
+ self.depends = references
+
+ def become_superseded(self, superseding):
+ """When another action tries to supersede this one,
+ we need to change both of them, so the surviving actions will be
+ logic consistent.
+
+ This method should do the required internal changes, and also
+ suggest changes for the other, superseding, action.
+
+ Arguments:
+ superseding: other task superseding this one
+
+ Returns:
+ dict: changes suggested to the action superseding this one.
+ A special key ``superseding_needed`` is used to
+ suggest if the superseding is actually required or not.
+ If not present, ``superseding_needed`` is assumed to
+ be False.
+ """
+ self.status = 'SUPERSEDED'
+ self.logger.debug(
+ 'Action `%s` was superseded by `%s`',
+ self.__class__.__name__, superseding.__class__.__name__)
+ return {}
+
+ def supersede(self, others):
+ """Supersede other tasks, if necessary
+
+ Arguments:
+ others (list): action objects being superseded
+
+ When the task decide to supersede others, this method should call
+ ``become_superseded`` on the other actions, collect the suggested
+ updates and perform the necessary changes
+ """
+ # By default actions don't supersede others
+ self.logger.debug(
+ 'Action `%s` does not supersede other actions',
+ self.__class__.__name__)
+
+ def process(self, connector, persistence, ovim):
+ """Abstract method, that needs to be implemented.
+ Process the current task.
+
+ Arguments:
+ connector: object with API for accessing the WAN
+ Infrastructure Manager system
+ persistence: abstraction layer for the database
+ ovim: instance of openvim, abstraction layer that enable
+ SDN-related operations
+ """
+ raise NotImplementedError
+
+
+class FindAction(Action):
+ """Abstract class that should be inherited for FIND actions, depending on
+ the item type.
+ """
+ @property
+ def processing(self):
+ if self.status in ('DONE', 'BUILD'):
+ return REFRESH
+
+ return super(FindAction, self).processing
+
+ def become_superseded(self, superseding):
+ super(FindAction, self).become_superseded(superseding)
+ info = ('vim_id', 'wim_internal_id')
+ return remove_none_items({f: getattr(self, f) for f in info})
+
+
+class CreateAction(Action):
+ """Abstract class that should be inherited for CREATE actions, depending on
+ the item type.
+ """
+ @property
+ def processing(self):
+ if self.status in ('DONE', 'BUILD'):
+ return REFRESH
+
+ return super(CreateAction, self).processing
+
+ def become_superseded(self, superseding):
+ super(CreateAction, self).become_superseded(superseding)
+
+ created = self.extra.get('created', True)
+ sdn_net_id = self.extra.get('sdn_net_id')
+ pending_info = self.wim_internal_id or self.vim_id or sdn_net_id
+ if not(created and pending_info):
+ return {}
+
+ extra_fields = ('sdn_net_id', 'interfaces', 'created_items')
+ extra_info = filter_dict_keys(self.extra or {}, extra_fields)
+
+ return {'superseding_needed': True,
+ 'wim_internal_id': self.wim_internal_id,
+ 'vim_id': self.vim_id,
+ 'extra': remove_none_items(extra_info)}
+
+
+class DeleteAction(Action):
+ """Abstract class that should be inherited for DELETE actions, depending on
+ the item type.
+ """
+ def supersede(self, others):
+ self.logger.debug('%s %s %s %s might supersede other actions',
+ self.id, self.action, self.item, self.item_id)
+ # First collect all the changes from the superseded tasks
+ changes = [other.become_superseded(self) for other in others]
+ needed = any(change.pop('superseding_needed', False)
+ for change in changes)
+
+ # Deal with the nested ones first
+ extras = [change.pop('extra', None) or {} for change in changes]
+ items = [extra.pop('created_items', None) or {} for extra in extras]
+ items = merge_dicts(self.extra.get('created_items', {}), *items)
+ self.extra = merge_dicts(self.extra, {'created_items': items}, *extras)
+
+ # Accept the other ones
+ change = ((key, value) for key, value in merge_dicts(*changes).items()
+ if key in self.PROPERTIES)
+ for attr, value in change:
+ setattr(self, attr, value)
+
+ # Reevaluate if the action itself is needed
+ if not needed:
+ self.status = 'SUPERSEDED'
+
+
+def _expand_extra(record):
+ extra = record.pop('extra', None) or {}
+ if isinstance(extra, str):
+ extra = yaml.safe_load(extra)
+
+ record['params'] = extra.get('params')
+ record['depends_on'] = extra.get('depends_on', [])
+ record['depends'] = extra.get('depends', None)
+ record['extra'] = extra
+
+ return record
projects / osm / RO.git / blobdiff