build.models.EVC.install_nni_flows() - Code Metrics - Inspection of "Add tests for discover backup path" - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#72)

by macartur

created 2018-09-04 18:19 UTC

build.models.EVC.install_nni_flows() A

↳ Parent: build.models.EVCBase.as_dict()

Complexity

Conditions

Size

Total Lines	17
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

Metric	Value
cc	2
eloc	12
nop	1
dl	0
loc	17
ccs	0
cts	8
cp	0
crap	6
rs	9.8
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
335			for switch in switches:
336			self.send_flow_mods(switch, flows, 'delete')
337
338			self.deactivate()
339
340			@staticmethod
341			def choose_vlans(path=None):
342			"""Choose the VLANs to be used for the circuit."""
343			for link in path:
344			tag = link.get_next_available_tag()
345			link.use_tag(tag)
346			link.add_metadata('s_vlan', tag)
347
348			@staticmethod
349			def links_zipped(path=None):
350			"""Return an iterator which yields pairs of links in order."""
351			if not path:
352			return []
353			return zip(path[:-1], path[1:])
354
355			def should_deploy(self, path=None):
356			"""Verify if the circuit should be deployed."""
357			if not path:
358			log.debug("Path is empty.")
359			return False
360
361			if not self.is_enabled():
362			log.debug(f'{self} is disabled.')
363			return False
364
365			if not self.is_active():
366			log.debug(f'{self} will be deployed.')
367			return True
368
369			return False
370
371			def deploy(self, path=None):
372			"""Install the flows for this circuit.
373
374			Procedures to deploy:
375
376			0. Remove current flows installed
377			1. Decide if will deploy "path" or discover a new path
378			2. Choose vlan
379			3. Install NNI flows
380			4. Install UNI flows
381			5. Activate
382			6. Update current_path
383			7. Update links caches(primary, current, backup)
384
385			"""
386			self.remove_current_flows()
387
388			if not self.should_deploy(path):
389			return False
390
391			if path is None:
392			path = self.discover_new_path()
393
394			if not path:
395			return False
396
397			self.choose_vlans(path)
398			self.install_nni_flows(path)
399			self.install_uni_flows(path)
400			self.activate()
401			log.info(f"{self} was deployed.")
402			return True
403
404			def install_nni_flows(self, path=None):
405			"""Install NNI flows."""
406			for incoming, outcoming in self.links_zipped(path):
407			in_vlan = incoming.get_metadata('s_vlan').value
408			out_vlan = outcoming.get_metadata('s_vlan').value
409
410			flows = []
411			# Flow for one direction
412			flows.append(self.prepare_nni_flow(incoming.endpoint_b,
413			outcoming.endpoint_a,
414			in_vlan, out_vlan))
415
416			# Flow for the other direction
417			flows.append(self.prepare_nni_flow(outcoming.endpoint_a,
418			incoming.endpoint_b,
419			out_vlan, in_vlan))
420			self.send_flow_mods(incoming.endpoint_b.switch, flows)
421
422			def install_uni_flows(self, path=None):
423			"""Install UNI flows."""
424			if not path:
425			log.info('install uni flows without path.')
426			return
427
428			# Determine VLANs
429			in_vlan_a = self.uni_a.user_tag.value if self.uni_a.user_tag else None
430			out_vlan_a = path[0].get_metadata('s_vlan').value
431
432			in_vlan_z = self.uni_z.user_tag.value if self.uni_z.user_tag else None
433			out_vlan_z = path[-1].get_metadata('s_vlan').value
434
435			# Flows for the first UNI
436			flows_a = []
437
438			# Flow for one direction, pushing the service tag
439			push_flow = self.prepare_push_flow(self.uni_a.interface,
440			path[0].endpoint_a,
441			in_vlan_a, out_vlan_a, in_vlan_z)
442			flows_a.append(push_flow)
443
444			# Flow for the other direction, popping the service tag
445			pop_flow = self.prepare_pop_flow(path[0].endpoint_a,
446			self.uni_a.interface, out_vlan_a)
447			flows_a.append(pop_flow)
448
449			self.send_flow_mods(self.uni_a.interface.switch, flows_a)
450
451			# Flows for the second UNI
452			flows_z = []
453
454			# Flow for one direction, pushing the service tag
455			push_flow = self.prepare_push_flow(self.uni_z.interface,
456			path[-1].endpoint_b,
457			in_vlan_z, out_vlan_z, in_vlan_a)
458			flows_z.append(push_flow)
459
460			# Flow for the other direction, popping the service tag
461			pop_flow = self.prepare_pop_flow(path[-1].endpoint_b,
462			self.uni_z.interface, out_vlan_z)
463			flows_z.append(pop_flow)
464
465			self.send_flow_mods(self.uni_z.interface.switch, flows_z)
466
467			@staticmethod
468			def send_flow_mods(switch, flow_mods, command='flows'):
469			"""Send a flow_mod list to a specific switch.
470
471			Args:
472			switch(Switch): The target of flows.
473			flow_mods(dict): Python dictionary with flow_mods.
474			command(str): By default is 'flows'. To remove a flow is 'remove'.
475
476			"""
477			endpoint = f'{settings.MANAGER_URL}/{command}/{switch.id}'
478
479			data = {"flows": flow_mods}
480			requests.post(endpoint, json=data)
481
482			def get_cookie(self):
483			"""Return the cookie integer from evc id."""
484			value = self.id[len(self.id)//2:]
485			return int(value, 16)
486
487			def prepare_flow_mod(self, in_interface, out_interface):
488			"""Prepare a common flow mod."""
489			default_action = {"action_type": "output",
490			"port": out_interface.port_number}
491
492			flow_mod = {"match": {"in_port": in_interface.port_number},
493			"cookie": self.get_cookie(),
494			"actions": [default_action]}
495
496			return flow_mod
497
498			def prepare_nni_flow(self, in_interface, out_interface, in_vlan, out_vlan):
499			"""Create NNI flows."""
500			flow_mod = self.prepare_flow_mod(in_interface, out_interface)
501			flow_mod['match']['dl_vlan'] = in_vlan
502
503			new_action = {"action_type": "set_vlan",
504			"vlan_id": out_vlan}
505			flow_mod["actions"].insert(0, new_action)
506
507			return flow_mod
508
509			def prepare_push_flow(self, *args):
510			"""Prepare push flow.
511
512			Arguments:
513			in_interface(str): Interface input.
514			out_interface(str): Interface output.
515			in_vlan(str): Vlan input.
516			out_vlan(str): Vlan output.
517			new_in_vlan(str): Interface input.
518
519			Return:
520			dict: An python dictionary representing a FlowMod
521
522			"""
523			# assign all arguments
524			in_interface, out_interface, in_vlan, out_vlan, new_in_vlan = args
525
526			flow_mod = self.prepare_flow_mod(in_interface, out_interface)
527			flow_mod['match']['dl_vlan'] = in_vlan
528
529			new_action = {"action_type": "set_vlan",
530			"vlan_id": out_vlan}
531			flow_mod["actions"].insert(0, new_action)
532
533			new_action = {"action_type": "push_vlan",
534			"tag_type": "s"}
535			flow_mod["actions"].insert(0, new_action)
536
537			new_action = {"action_type": "set_vlan",
538			"vlan_id": new_in_vlan}
539			flow_mod["actions"].insert(0, new_action)
540
541			return flow_mod
542
543			def prepare_pop_flow(self, in_interface, out_interface, in_vlan):
544			"""Prepare pop flow."""
545			flow_mod = self.prepare_flow_mod(in_interface, out_interface)
546			flow_mod['match']['dl_vlan'] = in_vlan
547			new_action = {"action_type": "pop_vlan"}
548			flow_mod["actions"].insert(0, new_action)
549			return flow_mod
550
551
552			class LinkProtection(EVCDeploy):
553			"""Class to handle link protection."""
554
555			def is_affected_by_link(self, link=None):
556			"""Verify if the current path is affected by link down event."""
557			return self.current_path.is_affected_by_link(link)
558
559			def is_using_primary_path(self):
560			"""Verify if the current deployed path is self.primary_path."""
561			return self.current_path == self.primary_path
562
563			def is_using_backup_path(self):
564			"""Verify if the current deployed path is self.backup_path."""
565			return self.current_path == self.backup_path
566
567			def is_using_dynamic_path(self):
568			"""Verify if the current deployed path is dynamic."""
569			if not self.is_using_primary_path() and \
570			not self.is_using_backup_path() and \
571			self.current_path.status is EntityStatus.UP:
572			return True
573			return False
574
575			def deploy_to(self, path_name=None, path=None):
576			"""Create a deploy to path."""
577			if self.current_path == path:
578			log.debug(f'{path_name} is equal to current_path.')
579			return True
580
581			if path.status is EntityStatus.UP:
582			return self.deploy(path)
583
584			return False
585
586			def handle_link_up(self, link):
587			"""Handle circuit when link down.
588
589			Args:
590			link(Link): Link affected by link.down event.
591
592			"""
593			if self.is_using_primary_path():
594			return True
595
596			success = False
597			if self.primary_path.is_affected_by_link(link):
598			success = self.deploy_to('primary_path', self.primary_path)
599
600			if success:
601			return True
602
603			# We tried to deploy(primary_path) without success.
604			# And in this case is up by some how. Nothing to do.
605			if self.is_using_backup_path() or self.is_using_dynamic_path():
606			return True
607
608			# In this case, probably the circuit is not being used and
609			# we can move to backup
610			if self.backup_path.is_affected_by_link(link):
611			success = self.deploy_to('backup_path', self.backup_path)
612
613			if success:
614			return True
615
616			# In this case, the circuit is not being used and we should
617			# try a dynamic path
618			if self.dynamic_backup_path:
619			return self.deploy()
620
621			return True
622
623			def handle_link_down(self):
624			"""Handle circuit when link down.
625
626			Returns:
627			bool: True if the re-deploy was successly otherwise False.
628
629			"""
630			success = False
631			if self.is_using_primary_path():
632			success = self.deploy_to('backup_path', self.backup_path)
633			elif self.is_using_backup_path():
634			success = self.deploy_to('primary_path', self.primary_path)
635
636			if not success and self.dynamic_backup_path:
637			success = self.deploy()
638
639			if success:
640			log.debug(f"{self} deployed after link down.")
641			else:
642			log.debug(f'Failed to re-deploy {self} after link down.')
643
644			return success
645
646
647			class EVC(LinkProtection):
648			"""Class that represents a E-Line Virtual Connection."""
649
650			pass
651

kytos / mef_eline

Pull Request — master (#72)

build.models.EVC.install_nni_flows() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like