osm_ro/wim/engine.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.
##

"""This module contains the domain logic, and the implementation of the
required steps to perform VNF management and orchestration in a WAN
environment.

It works as an extension/complement to the main functions contained in the
``nfvo.py`` file and avoids interacting directly with the database, by relying
on the `persistence` module.

No http request handling/direct interaction with the database should be present
in this file.
"""
import json
import logging
from contextlib import contextmanager
from itertools import groupby
from operator import itemgetter
from sys import exc_info
from uuid import uuid4

from six import reraise

from ..utils import remove_none_items
from .actions import Action
from .errors import (
    DbBaseException,
    NoWimConnectedToDatacenters,
    UnexpectedDatabaseError,
    WimAccountNotActive
)
from .wim_thread import WimThread


class WimEngine(object):
    """Logic supporting the establishment of WAN links when NS spans across
    different datacenters.
    """
    def __init__(self, persistence, logger=None, ovim=None):
        self.persist = persistence
        self.logger = logger or logging.getLogger('openmano.wim.engine')
        self.threads = {}
        self.connectors = {}
        self.ovim = ovim

    def create_wim(self, properties):
        """Create a new wim record according to the properties

        Please check the wim schema to have more information about
        ``properties``.

        The ``config`` property might contain a ``wim_port_mapping`` dict,
        In this case, the method ``create_wim_port_mappings`` will be
        automatically invoked.

        Returns:
            str: uuid of the newly created WIM record
        """
        port_mapping = ((properties.get('config', {}) or {})
                        .pop('wim_port_mapping', {}))
        uuid = self.persist.create_wim(properties)

        if port_mapping:
            try:
                self.create_wim_port_mappings(uuid, port_mapping)
            except DbBaseException:
                # Rollback
                self.delete_wim(uuid)
                ex = UnexpectedDatabaseError('Failed to create port mappings'
                                             'Rolling back wim creation')
                self.logger.exception(str(ex))
                reraise(ex.__class__, ex, exc_info()[2])

        return uuid

    def get_wim(self, uuid_or_name, tenant_id=None):
        """Retrieve existing WIM record by name or id.

        If ``tenant_id`` is specified, the query will be
        limited to the WIM associated to the given tenant.
        """
        # Since it is a pure DB operation, we can delegate it directly
        return self.persist.get_wim(uuid_or_name, tenant_id)

    def update_wim(self, uuid_or_name, properties):
        """Edit an existing WIM record.

        ``properties`` is a dictionary with the properties being changed,
        if a property is not present, the old value will be preserved

        Similarly to create_wim, the ``config`` property might contain a
        ``wim_port_mapping`` dict, In this case, port mappings will be
        automatically updated.
        """
        port_mapping = ((properties.get('config', {}) or {})
                        .pop('wim_port_mapping', {}))
        orig_props = self.persist.get_by_name_or_uuid('wims', uuid_or_name)
        uuid = orig_props['uuid']

        response = self.persist.update_wim(uuid, properties)

        if port_mapping:
            try:
                # It is very complex to diff and update individually all the
                # port mappings. Therefore a practical approach is just delete
                # and create it again.
                self.persist.delete_wim_port_mappings(uuid)
                # ^  Calling from persistence avoid reloading twice the thread
                self.create_wim_port_mappings(uuid, port_mapping)
            except DbBaseException:
                # Rollback
                self.update_wim(uuid_or_name, orig_props)
                ex = UnexpectedDatabaseError('Failed to update port mappings'
                                             'Rolling back wim updates\n')
                self.logger.exception(str(ex))
                reraise(ex.__class__, ex, exc_info()[2])

        return response

    def delete_wim(self, uuid_or_name):
        """Kill the corresponding wim threads and erase the WIM record"""
        # Theoretically, we can rely on the database to drop the wim_accounts
        # automatically, since we have configures 'ON CASCADE DELETE'.
        # However, use use `delete_wim_accounts` to kill all the running
        # threads.
        self.delete_wim_accounts(uuid_or_name)
        return self.persist.delete_wim(uuid_or_name)

    def create_wim_account(self, wim, tenant, properties):
        """Create an account that associates a tenant to a WIM.

        As a side effect this function will spawn a new thread

        Arguments:
            wim (str): name or uuid of the WIM related to the account being
                created
            tenant (str): name or uuid of the nfvo tenant to which the account
                will be created
            properties (dict): properties of the account
                (eg. username, password, ...)

        Returns:
            dict: Created record
        """
        uuid = self.persist.create_wim_account(wim, tenant, properties)
        account = self.persist.get_wim_account_by(uuid=uuid)
        # ^  We need to use get_wim_account_by here, since this methods returns
        #    all the associations, and we need the wim to create the thread
        self._spawn_thread(account)
        return account

    def _update_single_wim_account(self, account, properties):
        """Update WIM Account, taking care to reload the corresponding thread

        Arguments:
            account (dict): Current account record
            properties (dict): Properties to be updated

        Returns:
            dict: updated record
        """
        account = self.persist.update_wim_account(account['uuid'], properties)
        self.threads[account['uuid']].reload()
        return account

    def update_wim_accounts(self, wim, tenant, properties):
        """Update all the accounts related to a WIM and a tenant,
        thanking care of reloading threads.

        Arguments:
            wim (str): uuid or name of a WIM record
            tenant (str): uuid or name of a NFVO tenant record
            properties (dict): attributes with values to be updated

        Returns
            list: Records that were updated
        """
        accounts = self.persist.get_wim_accounts_by(wim, tenant)
        return [self._update_single_wim_account(account, properties)
                for account in accounts]

    def _delete_single_wim_account(self, account):
        """Delete WIM Account, taking care to remove the corresponding thread
        and delete the internal WIM account, if it was automatically generated.

        Arguments:
            account (dict): Current account record
            properties (dict): Properties to be updated

        Returns:
            dict: current record (same as input)
        """
        self.persist.delete_wim_account(account['uuid'])

        if account['uuid'] not in self.threads:
            raise WimAccountNotActive(
                'Requests send to the WIM Account %s are not currently '
                'being processed.', account['uuid'])
        else:
            self.threads[account['uuid']].exit()
            del self.threads[account['uuid']]

        return account

    def delete_wim_accounts(self, wim, tenant=None, **kwargs):
        """Delete all the accounts related to a WIM (and a tenant),
        thanking care of threads and internal WIM accounts.

        Arguments:
            wim (str): uuid or name of a WIM record
            tenant (str): uuid or name of a NFVO tenant record

        Returns
            list: Records that were deleted
        """
        kwargs.setdefault('error_if_none', False)
        accounts = self.persist.get_wim_accounts_by(wim, tenant, **kwargs)
        return [self._delete_single_wim_account(a) for a in accounts]

    def _reload_wim_threads(self, wim_id):
        for thread in self.threads.values():
            if thread.wim_account['wim_id'] == wim_id:
                thread.reload()

    def create_wim_port_mappings(self, wim, properties, tenant=None):
        """Store information about port mappings from Database"""
        # TODO: Review tenants... WIMs can exist across different tenants,
        #       and the port_mappings are a WIM property, not a wim_account
        #       property, so the concepts are not related
        wim = self.persist.get_by_name_or_uuid('wims', wim)
        result = self.persist.create_wim_port_mappings(wim, properties, tenant)
        self._reload_wim_threads(wim['uuid'])
        return result

    def get_wim_port_mappings(self, wim):
        """Retrive information about port mappings from Database"""
        return self.persist.get_wim_port_mappings(wim)

    def delete_wim_port_mappings(self, wim):
        """Erase the port mapping records associated with the WIM"""
        wim = self.persist.get_by_name_or_uuid('wims', wim)
        message = self.persist.delete_wim_port_mappings(wim['uuid'])
        self._reload_wim_threads(wim['uuid'])
        return message

    def find_common_wims(self, datacenter_ids, tenant):
        """Find WIMs that are common to all datacenters listed"""
        mappings = self.persist.get_wim_port_mappings(
            datacenter=datacenter_ids, tenant=tenant, error_if_none=False)

        wim_id_of = itemgetter('wim_id')
        sorted_mappings = sorted(mappings, key=wim_id_of)  # needed by groupby
        grouped_mappings = groupby(sorted_mappings, key=wim_id_of)
        mapped_datacenters = {
            wim_id: [m['datacenter_id'] for m in mappings]
            for wim_id, mappings in grouped_mappings
        }

        return [
            wim_id
            for wim_id, connected_datacenters in mapped_datacenters.items()
            if set(connected_datacenters) >= set(datacenter_ids)
        ]

    def find_common_wim(self, datacenter_ids, tenant):
        """Find a single WIM that is able to connect all the datacenters
        listed

        Raises:
            NoWimConnectedToDatacenters: if no WIM connected to all datacenters
                at once is found
        """
        suitable_wim_ids = self.find_common_wims(datacenter_ids, tenant)

        if not suitable_wim_ids:
            raise NoWimConnectedToDatacenters(datacenter_ids)

        # TODO: use a criteria to determine which WIM is going to be used,
        #       instead of always using the first one (strategy pattern can be
        #       used here)
        return suitable_wim_ids[0]

    def find_suitable_wim_account(self, datacenter_ids, tenant):
        """Find a WIM account that is able to connect all the datacenters
        listed

        Arguments:
            datacenter_ids (list): List of UUIDs of all the datacenters (vims),
                that need to be connected.
            tenant (str): UUID of the OSM tenant

        Returns:
            object with the WIM account that is able to connect all the
                 datacenters.
        """
        wim_id = self.find_common_wim(datacenter_ids, tenant)
        return self.persist.get_wim_account_by(wim_id, tenant)

    def derive_wan_link(self,
                        wim_usage,
                        instance_scenario_id, sce_net_id,
                        networks, tenant, related=None):
        """Create a instance_wim_nets record for the given information"""
        if sce_net_id in wim_usage:
            account_id = wim_usage[sce_net_id]
            account = self.persist.get_wim_account_by(uuid=account_id)
            wim_id = account['wim_id']
        else:
            datacenters = [n['datacenter_id'] for n in networks]
            wim_id = self.find_common_wim(datacenters, tenant)
            account = self.persist.get_wim_account_by(wim_id, tenant)

        return {
            'uuid': str(uuid4()),
            'instance_scenario_id': instance_scenario_id,
            'sce_net_id': sce_net_id,
            'wim_id': wim_id,
            'wim_account_id': account['uuid'],
            'related': related
        }

    def derive_wan_links(self, wim_usage, networks, tenant=None):
        """Discover and return what are the wan_links that have to be created
        considering a set of networks (VLDs) required for a scenario instance
        (NSR).

        Arguments:
            wim_usage(dict): Mapping between sce_net_id and wim_id. If wim_id is False, means not create wam_links
            networks(list): Dicts containing the information about the networks
                that will be instantiated to materialize a Network Service
                (scenario) instance.
                Corresponding to the ``instance_net`` record.

        Returns:
            list: list of WAN links to be written to the database
        """
        # Group networks by key=(instance_scenario_id, sce_net_id)
        related = None
        if networks:
            related = networks[0].get("related")
        filtered = _filter_multi_vim(networks)
        grouped_networks = _group_networks(filtered)
        datacenters_per_group = _count_datacenters(grouped_networks)
        # For each group count the number of networks. If greater then 1,
        # we have to create a wan link connecting them.
        wan_groups = [key
                      for key, counter in datacenters_per_group
                      if counter > 1]
        # Keys are tuples(instance_scenario_id, sce_net_id)
        return [
            self.derive_wan_link(wim_usage,
                                 key[0], key[1], grouped_networks[key], tenant, related)
            for key in wan_groups if wim_usage.get(key[1]) is not False
        ]

    def create_action(self, wan_link):
        """For a single wan_link create the corresponding create action"""
        return {
            'action': 'CREATE',
            'status': 'SCHEDULED',
            'item': 'instance_wim_nets',
            'item_id': wan_link['uuid'],
            'wim_account_id': wan_link['wim_account_id']
        }

    def create_actions(self, wan_links):
        """For an array of wan_links, create all the corresponding actions"""
        return [self.create_action(l) for l in wan_links]

    def delete_action(self, wan_link):
        """For a single wan_link create the corresponding create action"""
        return {
            'action': 'DELETE',
            'status': 'SCHEDULED',
            'item': 'instance_wim_nets',
            'item_id': wan_link['uuid'],
            'wim_account_id': wan_link['wim_account_id'],
            'extra': json.dumps({'wan_link': wan_link})
            # We serialize and cache the wan_link here, because it can be
            # deleted during the delete process
        }

    def delete_actions(self, wan_links=(), instance_scenario_id=None):
        """Given a Instance Scenario, remove all the WAN Links created in the
        past"""
        if instance_scenario_id:
            wan_links = self.persist.get_wan_links(
                instance_scenario_id=instance_scenario_id)
        return [self.delete_action(l) for l in wan_links]

    def incorporate_actions(self, wim_actions, instance_action):
        """Make the instance action consider new WIM actions and make the WIM
        actions aware of the instance action
        """
        current = instance_action.setdefault('number_tasks', 0)
        for i, action in enumerate(wim_actions):
            action['task_index'] = current + i
            action['instance_action_id'] = instance_action['uuid']
        instance_action['number_tasks'] += len(wim_actions)

        return wim_actions, instance_action

    def dispatch(self, tasks):
        """Enqueue a list of tasks for further processing.

        This function is supposed to be called outside from the WIM Thread.
        """
        for task in tasks:
            if task['wim_account_id'] not in self.threads:
                error_msg = str(WimAccountNotActive(
                    'Requests send to the WIM Account %s are not currently '
                    'being processed.', task['wim_account_id']))
                Action(task, self.logger).fail(self.persist, error_msg)
                self.persist.update_wan_link(task['item_id'],
                                             {'status': 'ERROR',
                                              'error_msg': error_msg})
                self.logger.error('Task %s %s %s not dispatched.\n%s',
                                  task['action'], task['item'],
                                  task['instance_account_id'], error_msg)
            else:
                self.threads[task['wim_account_id']].insert_task(task)
                self.logger.debug('Task %s %s %s dispatched',
                                  task['action'], task['item'],
                                  task['instance_action_id'])

    def _spawn_thread(self, wim_account):
        """Spawn a WIM thread

        Arguments:
            wim_account (dict): WIM information (usually persisted)
                The `wim` field is required to be set with a valid WIM record
                inside the `wim_account` dict

        Return:
            threading.Thread: Thread object
        """
        thread = None
        try:
            thread = WimThread(self.persist, wim_account, ovim=self.ovim)
            self.threads[wim_account['uuid']] = thread
            thread.start()
        except:  # noqa
            self.logger.error('Error when spawning WIM thread for %s',
                              wim_account['uuid'], exc_info=True)

        return thread

    def start_threads(self):
        """Start the threads responsible for processing WIM Actions"""
        accounts = self.persist.get_wim_accounts(error_if_none=False)
        self.threads = remove_none_items(
            {a['uuid']: self._spawn_thread(a) for a in accounts})

    def stop_threads(self):
        """Stop the threads responsible for processing WIM Actions"""
        for uuid, thread in self.threads.items():
            thread.exit()
            del self.threads[uuid]

    @contextmanager
    def threads_running(self):
        """Ensure no thread will be left running"""
        # This method is particularly important for testing :)
        try:
            self.start_threads()
            yield
        finally:
            self.stop_threads()


def _filter_multi_vim(networks):
    """Ignore networks without sce_net_id (all VNFs go to the same VIM)"""
    return [n for n in networks if 'sce_net_id' in n and n['sce_net_id']]


def _group_networks(networks):
    """Group networks that correspond to the same instance_scenario_id and
    sce_net_id (NSR and VLD).

    Arguments:
        networks(list): Dicts containing the information about the networks
            that will be instantiated to materialize a Network Service
            (scenario) instance.
    Returns:
        dict: Keys are tuples (instance_scenario_id, sce_net_id) and values
            are list of networks.
    """
    criteria = itemgetter('instance_scenario_id', 'sce_net_id')

    networks = sorted(networks, key=criteria)
    return {k: list(v) for k, v in groupby(networks, key=criteria)}


def _count_datacenters(grouped_networks):
    """Count the number of datacenters in each group of networks

    Returns:
        list of tuples: the first element is the group key, while the second
            element is the number of datacenters in each group.
    """
    return ((key, len(set(n['datacenter_id'] for n in group)))
            for key, group in grouped_networks.items())