build.models.DynamicPathManager.get_best_path() - Code Metrics - Inspection of "Implement dynamic path" - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#71)

by macartur

created 2018-09-05 17:19 UTC

build.models.DynamicPathManager.get_best_path() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	2
dl	0
loc	7
ccs	0
cts	3
cp	0
crap	6
rs	10
c	0
b	0
f	0

"""Classes used in the main application."""
from datetime import datetime
from uuid import uuid4

import requests

from kytos.core import log
from kytos.core.common import EntityStatus, GenericEntity
from kytos.core.helpers import get_time, now
from kytos.core.interface import UNI
from kytos.core.link import Link
from napps.kytos.mef_eline import settings


class Path(list, GenericEntity):
    """Class to represent a Path."""

    def __init__(self, *args, **kwargs):
        """Create a path instance using links."""
        super().__init__(*args, **kwargs)
        self.links_cache = set(self)

    def __eq__(self, other=None):
        """Compare paths."""
        if not other or not isinstance(other, Path):
            return False
        return super().__eq__(other)

    def is_affected_by_link(self, link=None):
        """Verify if the current path is affected by link."""
        if not link:
            return False
        return link in self.links_cache

    @property
    def status(self):
        """Check for the  status of a path.

        If any link in this path is down, the path is considered down.
        """
        if not self:
            return EntityStatus.DISABLED

        for link in self:
            if link.status is not EntityStatus.UP:
                return link.status
        return EntityStatus.UP

    def as_dict(self):
        """Return list comprehension of links as_dict."""
        return [link.as_dict() for link in self if link]


class DynamicPathManager:
    """Class to handle and create paths."""

    controller = None

    @classmethod
    def set_controller(cls, controller=None):
        """Set the controller to discovery news paths."""
        cls.controller = controller

    @staticmethod
    def get_paths(circuit):
        """Get a valid path for the circuit from the Pathfinder."""
        endpoint = settings.PATHFINDER_URL
        request_data = {"source": circuit.uni_a.interface.id,
                        "destination": circuit.uni_z.interface.id}
        api_reply = requests.post(endpoint, json=request_data)

        if api_reply.status_code != getattr(requests.codes, 'ok'):
            log.error("Failed to get paths at %s. Returned %s",
                      endpoint, api_reply.status_code)
            return None
        reply_data = api_reply.json()
        return reply_data.get('paths')

    @staticmethod
    def _clear_path(path):
        """Remove switches from a path, returning only interfaeces."""
        return [endpoint for endpoint in path if len(endpoint) > 23]

    @classmethod
    def get_best_path(cls, circuit):
        """Return the best path available for a circuit, if exists."""
        paths = cls.get_paths(circuit)
        if paths:
            return cls.create_path(cls.get_paths(circuit)[0]['hops'])
        return None

    @classmethod
    def create_path(cls, path):
        """Return the path containing only the interfaces."""
        new_path = Path()
        clean_path = cls._clear_path(path)

        if len(clean_path) % 2:
            return None

        for link in zip(clean_path[1:-1:2], clean_path[2::2]):
            interface_a = cls.controller.get_interface_by_id(link[0])
            interface_b = cls.controller.get_interface_by_id(link[1])
            if interface_a is None or interface_b is None:
                return None
            new_path.append(Link(interface_a, interface_b))

        return new_path


class EVCBase(GenericEntity):
    """"Class to represent a circuit."""

    unique_attributes = ['name', 'uni_a', 'uni_z']

    def __init__(self, **kwargs):
        """Create an EVC instance with the provided parameters.

        Args:
            id(str): EVC identifier. Whether it's None an ID will be genereted.
            name: represents an EVC name.(Required)
            uni_a (UNI): Endpoint A for User Network Interface.(Required)
            uni_z (UNI): Endpoint Z for User Network Interface.(Required)
            start_date(datetime|str): Date when the EVC was registred.
                                      Default is now().
            end_date(datetime|str): Final date that the EVC will be fineshed.
                                    Default is None.
            bandwidth(int): Bandwidth used by EVC instance. Default is 0.
            primary_links(list): Primary links used by evc. Default is []
            backup_links(list): Backups links used by evc. Default is []
            current_path(list): Circuit being used at the moment if this is an
                                active circuit. Default is [].
            primary_path(list): primary circuit offered to user IF one or more
                                links were provided. Default is [].
            backup_path(list): backup circuit offered to the user IF one or
                               more links were provided. Default is [].
            dynamic_backup_path(bool): Enable computer backup path dynamically.
                                       Dafault is False.
            creation_time(datetime|str): datetime when the circuit should be
                                         activated. default is now().
            enabled(Boolean): attribute to indicate the operational state.
                              default is False.
            active(Boolean): attribute to Administrative state;
                             default is False.
            owner(str): The EVC owner. Default is None.
            priority(int): Service level provided in the request. Default is 0.

        Raises:
            ValueError: raised when object attributes are invalid.

        """
        self._validate(**kwargs)
        super().__init__()

        # required attributes
        self._id = kwargs.get('id', uuid4().hex)
        self.uni_a = kwargs.get('uni_a')
        self.uni_z = kwargs.get('uni_z')
        self.name = kwargs.get('name')

        # optional attributes
        self.start_date = get_time(kwargs.get('start_date')) or now()
        self.end_date = get_time(kwargs.get('end_date')) or None

        self.bandwidth = kwargs.get('bandwidth', 0)
        self.primary_links = Path(kwargs.get('primary_links', []))
        self.backup_links = Path(kwargs.get('backup_links', []))
        self.current_path = Path(kwargs.get('current_path', []))
        self.primary_path = Path(kwargs.get('primary_path', []))
        self.backup_path = Path(kwargs.get('backup_path', []))
        self.dynamic_backup_path = kwargs.get('dynamic_backup_path', False)
        self.creation_time = get_time(kwargs.get('creation_time')) or now()
        self.owner = kwargs.get('owner', None)
        self.priority = kwargs.get('priority', 0)
        self.circuit_scheduler = kwargs.get('circuit_scheduler', [])

        if kwargs.get('active', False):
            self.activate()
        else:
            self.deactivate()

        if kwargs.get('enabled', False):
            self.enable()
        else:
            self.disable()

        # datetime of user request for a EVC (or datetime when object was
        # created)
        self.request_time = kwargs.get('request_time', now())
        # dict with the user original request (input)
        self._requested = kwargs

    def update(self, **kwargs):
        """Update evc attributes.

        This method will raises an error trying to change the following
        attributes: [name, uni_a and uni_z]

        Raises:
            ValueError: message with error detail.

        """
        for attribute, value in kwargs.items():
            if attribute in self.unique_attributes:
                raise ValueError(f'{attribute} can\'t be be updated.')
            if hasattr(self, attribute):
                setattr(self, attribute, value)
            else:
                raise ValueError(f'The attribute "{attribute}" is invalid.')

    def __repr__(self):
        """Repr method."""
        return f"EVC({self._id}, {self.name})"

    def _validate(self, **kwargs):
        """Do Basic validations.

        Verify required attributes: name, uni_a, uni_z
        Verify if the attributes uni_a and uni_z are valid.

        Raises:
            ValueError: message with error detail.

        """
        for attribute in self.unique_attributes:

            if attribute not in kwargs:
                raise ValueError(f'{attribute} is required.')

            if 'uni' in attribute:
                uni = kwargs.get(attribute)
                if not isinstance(uni, UNI):
                    raise ValueError(f'{attribute} is an invalid UNI.')

                elif not uni.is_valid():
                    tag = uni.user_tag.value
                    message = f'VLAN tag {tag} is not available in {attribute}'
                    raise ValueError(message)

    def __eq__(self, other):
        """Override the default implementation."""
        if not isinstance(other, EVC):
            return False

        attrs_to_compare = ['name', 'uni_a', 'uni_z', 'owner', 'bandwidth']
        for attribute in attrs_to_compare:
            if getattr(other, attribute) != getattr(self, attribute):
                return False
        return True

    def as_dict(self):
        """Return a dictionary representing an EVC object."""
        evc_dict = {"id": self.id, "name": self.name,
                    "uni_a": self.uni_a.as_dict(),
                    "uni_z": self.uni_z.as_dict()}

        time_fmt = "%Y-%m-%dT%H:%M:%S"

        evc_dict["start_date"] = self.start_date
        if isinstance(self.start_date, datetime):
            evc_dict["start_date"] = self.start_date.strftime(time_fmt)

        evc_dict["end_date"] = self.end_date
        if isinstance(self.end_date, datetime):
            evc_dict["end_date"] = self.end_date.strftime(time_fmt)

        evc_dict['bandwidth'] = self.bandwidth
        evc_dict['primary_links'] = self.primary_links.as_dict()
        evc_dict['backup_links'] = self.backup_links.as_dict()
        evc_dict['current_path'] = self.current_path.as_dict()
        evc_dict['primary_path'] = self.primary_path.as_dict()
        evc_dict['backup_path'] = self.backup_path.as_dict()
        evc_dict['dynamic_backup_path'] = self.dynamic_backup_path

        if self._requested:
            request_dict = self._requested.copy()
            request_dict['uni_a'] = request_dict['uni_a'].as_dict()
            request_dict['uni_z'] = request_dict['uni_z'].as_dict()
            evc_dict['_requested'] = request_dict

        evc_dict["request_time"] = self.request_time
        if isinstance(self.request_time, datetime):
            evc_dict["request_time"] = self.request_time.strftime(time_fmt)

        time = self.creation_time.strftime(time_fmt)
        evc_dict['creation_time'] = time

        evc_dict['owner'] = self.owner
        evc_dict['circuit_scheduler'] = self.circuit_scheduler
        evc_dict['active'] = self.is_active()
        evc_dict['enabled'] = self.is_enabled()
        evc_dict['priority'] = self.priority

        return evc_dict

    @property
    def id(self):  # pylint: disable=invalid-name
        """Return this EVC's ID."""
        return self._id


class EVCDeploy(EVCBase):
    """Class to handle the deploy procedures."""

    def create(self):
        """Create a EVC."""
        pass

    def discover_new_path(self):
        """Discover a new path to satisfy this circuit and deploy."""
        return DynamicPathManager.get_best_path(self)

    def change_path(self):
        """Change EVC path."""
        pass

    def reprovision(self):
        """Force the EVC (re-)provisioning."""
        pass

    def remove(self):
        """Remove EVC path."""
        pass

    def remove_current_flows(self):
        """Remove all flows from current path."""
        switches = set()

        for link in self.current_path:
            switches.add(link.endpoint_a.switch)
            switches.add(link.endpoint_b.switch)

        flows = [{'cookie': self.get_cookie()}]
        for switch in switches:
            self.send_flow_mods(switch, flows, 'delete')
        self.deactivate()

    @staticmethod
    def choose_vlans(path=None):
        """Choose the VLANs to be used for the circuit."""
        for link in path:
            tag = link.get_next_available_tag()
            link.use_tag(tag)
            link.add_metadata('s_vlan', tag)

    @staticmethod
    def links_zipped(path=None):
        """Return an iterator which yields pairs of links in order."""
        if not path:
            return []
        return zip(path[:-1], path[1:])

    def should_deploy(self, path=None):
        """Verify if the circuit should be deployed."""
        if not path:
            log.debug("Path is empty.")
            return False

        if not self.is_enabled():
            log.debug(f'{self} is disabled.')
            return False

        if not self.is_active():
            log.debug(f'{self} will be deployed.')
            return True

        return False

    def deploy(self, path=None):
        """Install the flows for this circuit.

        Procedures to deploy:

        0. Remove current flows installed
        1. Decide if will deploy "path" or discover a new path
        2. Choose vlan
        3. Install NNI flows
        4. Install UNI flows
        5. Activate
        6. Update current_path
        7. Update links caches(primary, current, backup)

        """
        self.remove_current_flows()

        if not self.should_deploy(path):
            return False

        if path is None:
            path = self.discover_new_path()

            if not path:
                return False

        self.choose_vlans(path)
        self.install_nni_flows(path)
        self.install_uni_flows(path)
        self.activate()
        log.info(f"{self} was deployed.")
        return True

    def install_nni_flows(self, path=None):
        """Install NNI flows."""
        for incoming, outcoming in self.links_zipped(path):
            in_vlan = incoming.get_metadata('s_vlan').value
            out_vlan = outcoming.get_metadata('s_vlan').value

            flows = []
            # Flow for one direction
            flows.append(self.prepare_nni_flow(incoming.endpoint_b,
                                               outcoming.endpoint_a,
                                               in_vlan, out_vlan))

            # Flow for the other direction
            flows.append(self.prepare_nni_flow(outcoming.endpoint_a,
                                               incoming.endpoint_b,
                                               out_vlan, in_vlan))
            self.send_flow_mods(incoming.endpoint_b.switch, flows)

    def install_uni_flows(self, path=None):
        """Install UNI flows."""
        if not path:
            log.info('install uni flows without path.')
            return

        # Determine VLANs
        in_vlan_a = self.uni_a.user_tag.value if self.uni_a.user_tag else None
        out_vlan_a = path[0].get_metadata('s_vlan').value

        in_vlan_z = self.uni_z.user_tag.value if self.uni_z.user_tag else None
        out_vlan_z = path[-1].get_metadata('s_vlan').value

        # Flows for the first UNI
        flows_a = []

        # Flow for one direction, pushing the service tag
        push_flow = self.prepare_push_flow(self.uni_a.interface,
                                           path[0].endpoint_a,
                                           in_vlan_a, out_vlan_a, in_vlan_z)
        flows_a.append(push_flow)

        # Flow for the other direction, popping the service tag
        pop_flow = self.prepare_pop_flow(path[0].endpoint_a,
                                         self.uni_a.interface, out_vlan_a)
        flows_a.append(pop_flow)

        self.send_flow_mods(self.uni_a.interface.switch, flows_a)

        # Flows for the second UNI
        flows_z = []

        # Flow for one direction, pushing the service tag
        push_flow = self.prepare_push_flow(self.uni_z.interface,
                                           path[-1].endpoint_b,
                                           in_vlan_z, out_vlan_z, in_vlan_a)
        flows_z.append(push_flow)

        # Flow for the other direction, popping the service tag
        pop_flow = self.prepare_pop_flow(path[-1].endpoint_b,
                                         self.uni_z.interface, out_vlan_z)
        flows_z.append(pop_flow)

        self.send_flow_mods(self.uni_z.interface.switch, flows_z)

    @staticmethod
    def send_flow_mods(switch, flow_mods, command='flows'):
        """Send a flow_mod list to a specific switch.

        Args:
            switch(Switch): The target of flows.
            flow_mods(dict): Python dictionary with flow_mods.
            command(str): By default is 'flows'. To remove a flow is 'remove'.

        """
        endpoint = f'{settings.MANAGER_URL}/{command}/{switch.id}'

        data = {"flows": flow_mods}
        requests.post(endpoint, json=data)

    def get_cookie(self):
        """Return the cookie integer from evc id."""
        value = self.id[len(self.id)//2:]
        return int(value, 16)

    def prepare_flow_mod(self, in_interface, out_interface):
        """Prepare a common flow mod."""
        default_action = {"action_type": "output",
                          "port": out_interface.port_number}

        flow_mod = {"match": {"in_port": in_interface.port_number},
                    "cookie": self.get_cookie(),
                    "actions": [default_action]}

        return flow_mod

    def prepare_nni_flow(self, in_interface, out_interface, in_vlan, out_vlan):
        """Create NNI flows."""
        flow_mod = self.prepare_flow_mod(in_interface, out_interface)
        flow_mod['match']['dl_vlan'] = in_vlan

        new_action = {"action_type": "set_vlan",
                      "vlan_id": out_vlan}
        flow_mod["actions"].insert(0, new_action)

        return flow_mod

    def prepare_push_flow(self, *args):
        """Prepare push flow.

        Arguments:
            in_interface(str): Interface input.
            out_interface(str): Interface output.
            in_vlan(str): Vlan input.
            out_vlan(str): Vlan output.
            new_in_vlan(str): Interface input.

        Return:
            dict: An python dictionary representing a FlowMod

        """
        # assign all arguments
        in_interface, out_interface, in_vlan, out_vlan, new_in_vlan = args

        flow_mod = self.prepare_flow_mod(in_interface, out_interface)
        flow_mod['match']['dl_vlan'] = in_vlan

        new_action = {"action_type": "set_vlan",
                      "vlan_id": out_vlan}
        flow_mod["actions"].insert(0, new_action)

        new_action = {"action_type": "push_vlan",
                      "tag_type": "s"}
        flow_mod["actions"].insert(0, new_action)

        new_action = {"action_type": "set_vlan",
                      "vlan_id": new_in_vlan}
        flow_mod["actions"].insert(0, new_action)

        return flow_mod

    def prepare_pop_flow(self, in_interface, out_interface, in_vlan):
        """Prepare pop flow."""
        flow_mod = self.prepare_flow_mod(in_interface, out_interface)
        flow_mod['match']['dl_vlan'] = in_vlan
        new_action = {"action_type": "pop_vlan"}
        flow_mod["actions"].insert(0, new_action)
        return flow_mod


class LinkProtection(EVCDeploy):
    """Class to handle link protection."""

    def is_affected_by_link(self, link=None):
        """Verify if the current path is affected by link down event."""
        return self.current_path.is_affected_by_link(link)

    def is_using_primary_path(self):
        """Verify if the current deployed path is self.primary_path."""
        return self.current_path == self.primary_path

    def is_using_backup_path(self):
        """Verify if the current deployed path is self.backup_path."""
        return self.current_path == self.backup_path

    def is_using_dynamic_path(self):
        """Verify if the current deployed path is dynamic."""
        if not self.is_using_primary_path() and \
           not self.is_using_backup_path() and \
           self.current_path.status is EntityStatus.UP:
            return True
        return False

    def deploy_to(self, path_name=None, path=None):
        """Create a deploy to path."""
        if self.current_path == path:
            log.debug(f'{path_name} is equal to current_path.')
            return True

        if path.status is EntityStatus.UP:
            return self.deploy(path)

        return False

    def handle_link_up(self, link):
        """Handle circuit when link down.

        Args:
            link(Link): Link affected by link.down event.

        """
        if self.is_using_primary_path():
            return True

        success = False
        if self.primary_path.is_affected_by_link(link):
            success = self.deploy_to('primary_path', self.primary_path)

        if success:
            return True

        # We tried to deploy(primary_path) without success.
        # And in this case is up by some how. Nothing to do.
        if self.is_using_backup_path() or self.is_using_dynamic_path():
            return True

        # In this case, probably the circuit is not being used and
        # we can move to backup
        if self.backup_path.is_affected_by_link(link):
            success = self.deploy_to('backup_path', self.backup_path)

        if success:
            return True

        # In this case, the circuit is not being used and we should
        # try a dynamic path
        if self.dynamic_backup_path:
            return self.deploy()

        return True

    def handle_link_down(self):
        """Handle circuit when link down.

        Returns:
            bool: True if the re-deploy was successly otherwise False.

        """
        success = False
        if self.is_using_primary_path():
            success = self.deploy_to('backup_path', self.backup_path)
        elif self.is_using_backup_path():
            success = self.deploy_to('primary_path', self.primary_path)

        if not success and self.dynamic_backup_path:
            success = self.deploy()

        if success:
            log.debug(f"{self} deployed after link down.")
        else:
            log.debug(f'Failed to re-deploy {self} after link down.')

        return success


class EVC(LinkProtection):
    """Class that represents a E-Line Virtual Connection."""

    pass


1			"""Classes used in the main application."""
2			from datetime import datetime
3			from uuid import uuid4
4
5			import requests
6
7			from kytos.core import log
8			from kytos.core.common import EntityStatus, GenericEntity
9			from kytos.core.helpers import get_time, now
10			from kytos.core.interface import UNI
11			from kytos.core.link import Link
12			from napps.kytos.mef_eline import settings
13
14
15			class Path(list, GenericEntity):
16			"""Class to represent a Path."""
17
18			def __init__(self, args, *kwargs):
19			"""Create a path instance using links."""
20			super().__init__(args, *kwargs)
21			self.links_cache = set(self)
22
23			def __eq__(self, other=None):
24			"""Compare paths."""
25			if not other or not isinstance(other, Path):
26			return False
27			return super().__eq__(other)
28
29			def is_affected_by_link(self, link=None):
30			"""Verify if the current path is affected by link."""
31			if not link:
32			return False
33			return link in self.links_cache
34
35			@property
36			def status(self):
37			"""Check for the status of a path.
38
39			If any link in this path is down, the path is considered down.
40			"""
41			if not self:
42			return EntityStatus.DISABLED
43
44			for link in self:
45			if link.status is not EntityStatus.UP:
46			return link.status
47			return EntityStatus.UP
48
49			def as_dict(self):
50			"""Return list comprehension of links as_dict."""
51			return [link.as_dict() for link in self if link]
52
53
54			class DynamicPathManager:
55			"""Class to handle and create paths."""
56
57			controller = None
58
59			@classmethod
60			def set_controller(cls, controller=None):
61			"""Set the controller to discovery news paths."""
62			cls.controller = controller
63
64			@staticmethod
65			def get_paths(circuit):
66			"""Get a valid path for the circuit from the Pathfinder."""
67			endpoint = settings.PATHFINDER_URL
68			request_data = {"source": circuit.uni_a.interface.id,
69			"destination": circuit.uni_z.interface.id}
70			api_reply = requests.post(endpoint, json=request_data)
71
72			if api_reply.status_code != getattr(requests.codes, 'ok'):
73			log.error("Failed to get paths at %s. Returned %s",
74			endpoint, api_reply.status_code)
75			return None
76			reply_data = api_reply.json()
77			return reply_data.get('paths')
78
79			@staticmethod
80			def _clear_path(path):
81			"""Remove switches from a path, returning only interfaeces."""
82			return [endpoint for endpoint in path if len(endpoint) > 23]
83
84			@classmethod
85			def get_best_path(cls, circuit):
86			"""Return the best path available for a circuit, if exists."""
87			paths = cls.get_paths(circuit)
88			if paths:
89			return cls.create_path(cls.get_paths(circuit)[0]['hops'])
90			return None
91
92			@classmethod
93			def create_path(cls, path):
94			"""Return the path containing only the interfaces."""
95			new_path = Path()
96			clean_path = cls._clear_path(path)
97
98			if len(clean_path) % 2:
99			return None
100
101			for link in zip(clean_path[1:-1:2], clean_path[2::2]):
102			interface_a = cls.controller.get_interface_by_id(link[0])
103			interface_b = cls.controller.get_interface_by_id(link[1])
104			if interface_a is None or interface_b is None:
105			return None
106			new_path.append(Link(interface_a, interface_b))
107
108			return new_path
109
110
111			class EVCBase(GenericEntity):
112			""""Class to represent a circuit."""
113
114			unique_attributes = ['name', 'uni_a', 'uni_z']
115
116			def __init__(self, **kwargs):
117			"""Create an EVC instance with the provided parameters.
118
119			Args:
120			id(str): EVC identifier. Whether it's None an ID will be genereted.
121			name: represents an EVC name.(Required)
122			uni_a (UNI): Endpoint A for User Network Interface.(Required)
123			uni_z (UNI): Endpoint Z for User Network Interface.(Required)
124			start_date(datetime\|str): Date when the EVC was registred.
125			Default is now().
126			end_date(datetime\|str): Final date that the EVC will be fineshed.
127			Default is None.
128			bandwidth(int): Bandwidth used by EVC instance. Default is 0.
129			primary_links(list): Primary links used by evc. Default is []
130			backup_links(list): Backups links used by evc. Default is []
131			current_path(list): Circuit being used at the moment if this is an
132			active circuit. Default is [].
133			primary_path(list): primary circuit offered to user IF one or more
134			links were provided. Default is [].
135			backup_path(list): backup circuit offered to the user IF one or
136			more links were provided. Default is [].
137			dynamic_backup_path(bool): Enable computer backup path dynamically.
138			Dafault is False.
139			creation_time(datetime\|str): datetime when the circuit should be
140			activated. default is now().
141			enabled(Boolean): attribute to indicate the operational state.
142			default is False.
143			active(Boolean): attribute to Administrative state;
144			default is False.
145			owner(str): The EVC owner. Default is None.
146			priority(int): Service level provided in the request. Default is 0.
147
148			Raises:
149			ValueError: raised when object attributes are invalid.
150
151			"""
152			self._validate(**kwargs)
153			super().__init__()
154
155			# required attributes
156			self._id = kwargs.get('id', uuid4().hex)
157			self.uni_a = kwargs.get('uni_a')
158			self.uni_z = kwargs.get('uni_z')
159			self.name = kwargs.get('name')
160
161			# optional attributes
162			self.start_date = get_time(kwargs.get('start_date')) or now()
163			self.end_date = get_time(kwargs.get('end_date')) or None
164
165			self.bandwidth = kwargs.get('bandwidth', 0)
166			self.primary_links = Path(kwargs.get('primary_links', []))
167			self.backup_links = Path(kwargs.get('backup_links', []))
168			self.current_path = Path(kwargs.get('current_path', []))
169			self.primary_path = Path(kwargs.get('primary_path', []))
170			self.backup_path = Path(kwargs.get('backup_path', []))
171			self.dynamic_backup_path = kwargs.get('dynamic_backup_path', False)
172			self.creation_time = get_time(kwargs.get('creation_time')) or now()
173			self.owner = kwargs.get('owner', None)
174			self.priority = kwargs.get('priority', 0)
175			self.circuit_scheduler = kwargs.get('circuit_scheduler', [])
176
177			if kwargs.get('active', False):
178			self.activate()
179			else:
180			self.deactivate()
181
182			if kwargs.get('enabled', False):
183			self.enable()
184			else:
185			self.disable()
186
187			# datetime of user request for a EVC (or datetime when object was
188			# created)
189			self.request_time = kwargs.get('request_time', now())
190			# dict with the user original request (input)
191			self._requested = kwargs
192
193			def update(self, **kwargs):
194			"""Update evc attributes.
195
196			This method will raises an error trying to change the following
197			attributes: [name, uni_a and uni_z]
198
199			Raises:
200			ValueError: message with error detail.
201
202			"""
203			for attribute, value in kwargs.items():
204			if attribute in self.unique_attributes:
205			raise ValueError(f'{attribute} can\'t be be updated.')
206			if hasattr(self, attribute):
207			setattr(self, attribute, value)
208			else:
209			raise ValueError(f'The attribute "{attribute}" is invalid.')
210
211			def __repr__(self):
212			"""Repr method."""
213			return f"EVC({self._id}, {self.name})"
214
215			def _validate(self, **kwargs):
216			"""Do Basic validations.
217
218			Verify required attributes: name, uni_a, uni_z
219			Verify if the attributes uni_a and uni_z are valid.
220
221			Raises:
222			ValueError: message with error detail.
223
224			"""
225			for attribute in self.unique_attributes:
226
227			if attribute not in kwargs:
228			raise ValueError(f'{attribute} is required.')
229
230			if 'uni' in attribute:
231			uni = kwargs.get(attribute)
232			if not isinstance(uni, UNI):
233			raise ValueError(f'{attribute} is an invalid UNI.')
234
235			elif not uni.is_valid():
236			tag = uni.user_tag.value
237			message = f'VLAN tag {tag} is not available in {attribute}'
238			raise ValueError(message)
239
240			def __eq__(self, other):
241			"""Override the default implementation."""
242			if not isinstance(other, EVC):
243			return False
244
245			attrs_to_compare = ['name', 'uni_a', 'uni_z', 'owner', 'bandwidth']
246			for attribute in attrs_to_compare:
247			if getattr(other, attribute) != getattr(self, attribute):
248			return False
249			return True
250
251			def as_dict(self):
252			"""Return a dictionary representing an EVC object."""
253			evc_dict = {"id": self.id, "name": self.name,
254			"uni_a": self.uni_a.as_dict(),
255			"uni_z": self.uni_z.as_dict()}
256
257			time_fmt = "%Y-%m-%dT%H:%M:%S"
258
259			evc_dict["start_date"] = self.start_date
260			if isinstance(self.start_date, datetime):
261			evc_dict["start_date"] = self.start_date.strftime(time_fmt)
262
263			evc_dict["end_date"] = self.end_date
264			if isinstance(self.end_date, datetime):
265			evc_dict["end_date"] = self.end_date.strftime(time_fmt)
266
267			evc_dict['bandwidth'] = self.bandwidth
268			evc_dict['primary_links'] = self.primary_links.as_dict()
269			evc_dict['backup_links'] = self.backup_links.as_dict()
270			evc_dict['current_path'] = self.current_path.as_dict()
271			evc_dict['primary_path'] = self.primary_path.as_dict()
272			evc_dict['backup_path'] = self.backup_path.as_dict()
273			evc_dict['dynamic_backup_path'] = self.dynamic_backup_path
274
275			if self._requested:
276			request_dict = self._requested.copy()
277			request_dict['uni_a'] = request_dict['uni_a'].as_dict()
278			request_dict['uni_z'] = request_dict['uni_z'].as_dict()
279			evc_dict['_requested'] = request_dict
280
281			evc_dict["request_time"] = self.request_time
282			if isinstance(self.request_time, datetime):
283			evc_dict["request_time"] = self.request_time.strftime(time_fmt)
284
285			time = self.creation_time.strftime(time_fmt)
286			evc_dict['creation_time'] = time
287
288			evc_dict['owner'] = self.owner
289			evc_dict['circuit_scheduler'] = self.circuit_scheduler
290			evc_dict['active'] = self.is_active()
291			evc_dict['enabled'] = self.is_enabled()
292			evc_dict['priority'] = self.priority
293
294			return evc_dict
295
296			@property
297			def id(self): # pylint: disable=invalid-name
298			"""Return this EVC's ID."""
299			return self._id
300
301
302			class EVCDeploy(EVCBase):
303			"""Class to handle the deploy procedures."""
304
305			def create(self):
306			"""Create a EVC."""
307			pass
308
309			def discover_new_path(self):
310			"""Discover a new path to satisfy this circuit and deploy."""
311			return DynamicPathManager.get_best_path(self)
312
313			def change_path(self):
314			"""Change EVC path."""
315			pass
316
317			def reprovision(self):
318			"""Force the EVC (re-)provisioning."""
319			pass
320
321			def remove(self):
322			"""Remove EVC path."""
323			pass
324
325			def remove_current_flows(self):
326			"""Remove all flows from current path."""
327			switches = set()
328
329			for link in self.current_path:
330			switches.add(link.endpoint_a.switch)
331			switches.add(link.endpoint_b.switch)
332
333			flows = [{'cookie': self.get_cookie()}]
334			for switch in switches:
335			self.send_flow_mods(switch, flows, 'delete')
336			self.deactivate()
337
338			@staticmethod
339			def choose_vlans(path=None):
340			"""Choose the VLANs to be used for the circuit."""
341			for link in path:
342			tag = link.get_next_available_tag()
343			link.use_tag(tag)
344			link.add_metadata('s_vlan', tag)
345
346			@staticmethod
347			def links_zipped(path=None):
348			"""Return an iterator which yields pairs of links in order."""
349			if not path:
350			return []
351			return zip(path[:-1], path[1:])
352
353			def should_deploy(self, path=None):
354			"""Verify if the circuit should be deployed."""
355			if not path:
356			log.debug("Path is empty.")
357			return False
358
359			if not self.is_enabled():
360			log.debug(f'{self} is disabled.')
361			return False
362
363			if not self.is_active():
364			log.debug(f'{self} will be deployed.')
365			return True
366
367			return False
368
369			def deploy(self, path=None):
370			"""Install the flows for this circuit.
371
372			Procedures to deploy:
373
374			0. Remove current flows installed
375			1. Decide if will deploy "path" or discover a new path
376			2. Choose vlan
377			3. Install NNI flows
378			4. Install UNI flows
379			5. Activate
380			6. Update current_path
381			7. Update links caches(primary, current, backup)
382
383			"""
384			self.remove_current_flows()
385
386			if not self.should_deploy(path):
387			return False
388
389			if path is None:
390			path = self.discover_new_path()
391
392			if not path:
393			return False
394
395			self.choose_vlans(path)
396			self.install_nni_flows(path)
397			self.install_uni_flows(path)
398			self.activate()
399			log.info(f"{self} was deployed.")
400			return True
401
402			def install_nni_flows(self, path=None):
403			"""Install NNI flows."""
404			for incoming, outcoming in self.links_zipped(path):
405			in_vlan = incoming.get_metadata('s_vlan').value
406			out_vlan = outcoming.get_metadata('s_vlan').value
407
408			flows = []
409			# Flow for one direction
410			flows.append(self.prepare_nni_flow(incoming.endpoint_b,
411			outcoming.endpoint_a,
412			in_vlan, out_vlan))
413
414			# Flow for the other direction
415			flows.append(self.prepare_nni_flow(outcoming.endpoint_a,
416			incoming.endpoint_b,
417			out_vlan, in_vlan))
418			self.send_flow_mods(incoming.endpoint_b.switch, flows)
419
420			def install_uni_flows(self, path=None):
421			"""Install UNI flows."""
422			if not path:
423			log.info('install uni flows without path.')
424			return
425
426			# Determine VLANs
427			in_vlan_a = self.uni_a.user_tag.value if self.uni_a.user_tag else None
428			out_vlan_a = path[0].get_metadata('s_vlan').value
429
430			in_vlan_z = self.uni_z.user_tag.value if self.uni_z.user_tag else None
431			out_vlan_z = path[-1].get_metadata('s_vlan').value
432
433			# Flows for the first UNI
434			flows_a = []
435
436			# Flow for one direction, pushing the service tag
437			push_flow = self.prepare_push_flow(self.uni_a.interface,
438			path[0].endpoint_a,
439			in_vlan_a, out_vlan_a, in_vlan_z)
440			flows_a.append(push_flow)
441
442			# Flow for the other direction, popping the service tag
443			pop_flow = self.prepare_pop_flow(path[0].endpoint_a,
444			self.uni_a.interface, out_vlan_a)
445			flows_a.append(pop_flow)
446
447			self.send_flow_mods(self.uni_a.interface.switch, flows_a)
448
449			# Flows for the second UNI
450			flows_z = []
451
452			# Flow for one direction, pushing the service tag
453			push_flow = self.prepare_push_flow(self.uni_z.interface,
454			path[-1].endpoint_b,
455			in_vlan_z, out_vlan_z, in_vlan_a)
456			flows_z.append(push_flow)
457
458			# Flow for the other direction, popping the service tag
459			pop_flow = self.prepare_pop_flow(path[-1].endpoint_b,
460			self.uni_z.interface, out_vlan_z)
461			flows_z.append(pop_flow)
462
463			self.send_flow_mods(self.uni_z.interface.switch, flows_z)
464
465			@staticmethod
466			def send_flow_mods(switch, flow_mods, command='flows'):
467			"""Send a flow_mod list to a specific switch.
468
469			Args:
470			switch(Switch): The target of flows.
471			flow_mods(dict): Python dictionary with flow_mods.
472			command(str): By default is 'flows'. To remove a flow is 'remove'.
473
474			"""
475			endpoint = f'{settings.MANAGER_URL}/{command}/{switch.id}'
476
477			data = {"flows": flow_mods}
478			requests.post(endpoint, json=data)
479
480			def get_cookie(self):
481			"""Return the cookie integer from evc id."""
482			value = self.id[len(self.id)//2:]
483			return int(value, 16)
484
485			def prepare_flow_mod(self, in_interface, out_interface):
486			"""Prepare a common flow mod."""
487			default_action = {"action_type": "output",
488			"port": out_interface.port_number}
489
490			flow_mod = {"match": {"in_port": in_interface.port_number},
491			"cookie": self.get_cookie(),
492			"actions": [default_action]}
493
494			return flow_mod
495
496			def prepare_nni_flow(self, in_interface, out_interface, in_vlan, out_vlan):
497			"""Create NNI flows."""
498			flow_mod = self.prepare_flow_mod(in_interface, out_interface)
499			flow_mod['match']['dl_vlan'] = in_vlan
500
501			new_action = {"action_type": "set_vlan",
502			"vlan_id": out_vlan}
503			flow_mod["actions"].insert(0, new_action)
504
505			return flow_mod
506
507			def prepare_push_flow(self, *args):
508			"""Prepare push flow.
509
510			Arguments:
511			in_interface(str): Interface input.
512			out_interface(str): Interface output.
513			in_vlan(str): Vlan input.
514			out_vlan(str): Vlan output.
515			new_in_vlan(str): Interface input.
516
517			Return:
518			dict: An python dictionary representing a FlowMod
519
520			"""
521			# assign all arguments
522			in_interface, out_interface, in_vlan, out_vlan, new_in_vlan = args
523
524			flow_mod = self.prepare_flow_mod(in_interface, out_interface)
525			flow_mod['match']['dl_vlan'] = in_vlan
526
527			new_action = {"action_type": "set_vlan",
528			"vlan_id": out_vlan}
529			flow_mod["actions"].insert(0, new_action)
530
531			new_action = {"action_type": "push_vlan",
532			"tag_type": "s"}
533			flow_mod["actions"].insert(0, new_action)
534
535			new_action = {"action_type": "set_vlan",
536			"vlan_id": new_in_vlan}
537			flow_mod["actions"].insert(0, new_action)
538
539			return flow_mod
540
541			def prepare_pop_flow(self, in_interface, out_interface, in_vlan):
542			"""Prepare pop flow."""
543			flow_mod = self.prepare_flow_mod(in_interface, out_interface)
544			flow_mod['match']['dl_vlan'] = in_vlan
545			new_action = {"action_type": "pop_vlan"}
546			flow_mod["actions"].insert(0, new_action)
547			return flow_mod
548
549
550			class LinkProtection(EVCDeploy):
551			"""Class to handle link protection."""
552
553			def is_affected_by_link(self, link=None):
554			"""Verify if the current path is affected by link down event."""
555			return self.current_path.is_affected_by_link(link)
556
557			def is_using_primary_path(self):
558			"""Verify if the current deployed path is self.primary_path."""
559			return self.current_path == self.primary_path
560
561			def is_using_backup_path(self):
562			"""Verify if the current deployed path is self.backup_path."""
563			return self.current_path == self.backup_path
564
565			def is_using_dynamic_path(self):
566			"""Verify if the current deployed path is dynamic."""
567			if not self.is_using_primary_path() and \
568			not self.is_using_backup_path() and \
569			self.current_path.status is EntityStatus.UP:
570			return True
571			return False
572
573			def deploy_to(self, path_name=None, path=None):
574			"""Create a deploy to path."""
575			if self.current_path == path:
576			log.debug(f'{path_name} is equal to current_path.')
577			return True
578
579			if path.status is EntityStatus.UP:
580			return self.deploy(path)
581
582			return False
583
584			def handle_link_up(self, link):
585			"""Handle circuit when link down.
586
587			Args:
588			link(Link): Link affected by link.down event.
589
590			"""
591			if self.is_using_primary_path():
592			return True
593
594			success = False
595			if self.primary_path.is_affected_by_link(link):
596			success = self.deploy_to('primary_path', self.primary_path)
597
598			if success:
599			return True
600
601			# We tried to deploy(primary_path) without success.
602			# And in this case is up by some how. Nothing to do.
603			if self.is_using_backup_path() or self.is_using_dynamic_path():
604			return True
605
606			# In this case, probably the circuit is not being used and
607			# we can move to backup
608			if self.backup_path.is_affected_by_link(link):
609			success = self.deploy_to('backup_path', self.backup_path)
610
611			if success:
612			return True
613
614			# In this case, the circuit is not being used and we should
615			# try a dynamic path
616			if self.dynamic_backup_path:
617			return self.deploy()
618
619			return True
620
621			def handle_link_down(self):
622			"""Handle circuit when link down.
623
624			Returns:
625			bool: True if the re-deploy was successly otherwise False.
626
627			"""
628			success = False
629			if self.is_using_primary_path():
630			success = self.deploy_to('backup_path', self.backup_path)
631			elif self.is_using_backup_path():
632			success = self.deploy_to('primary_path', self.primary_path)
633
634			if not success and self.dynamic_backup_path:
635			success = self.deploy()
636
637			if success:
638			log.debug(f"{self} deployed after link down.")
639			else:
640			log.debug(f'Failed to re-deploy {self} after link down.')
641
642			return success
643
644
645			class EVC(LinkProtection):
646			"""Class that represents a E-Line Virtual Connection."""
647
648			pass
649

kytos / mef_eline

Pull Request — master (#71)

build.models.DynamicPathManager.get_best_path() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like