osm_ro/wim/actions.py

# -*- 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
from types import StringTypes

from six.moves import range

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, StringTypes):
        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