build.main._validate_range() - Code Metrics - Inspection of "Add support to list of exceptions in consistency m..." - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#112)

by Carlos

created 2021-05-25 19:54 UTC

build.main._validate_range() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	12
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	23.7834

Importance

Changes

Metric	Value
cc	5
eloc	11
nop	1
dl	0
loc	12
ccs	1
cts	11
cp	0.0909
crap	23.7834
rs	9.3333
c	0
b	0
f	0

"""kytos/flow_manager NApp installs, lists and deletes switch flows."""
from collections import OrderedDict
from copy import deepcopy

from flask import jsonify, request
from pyof.foundation.base import UBIntBase
from pyof.v0x01.asynchronous.error_msg import BadActionCode
from pyof.v0x01.common.phy_port import PortConfig
from werkzeug.exceptions import NotFound

from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to
from napps.kytos.flow_manager.match import match_flow
from napps.kytos.flow_manager.storehouse import StoreHouse
from napps.kytos.of_core.flow import FlowFactory

from .exceptions import InvalidCommandError
from .settings import (CONSISTENCY_COOKIE_IGNORED_RANGE, CONSISTENCY_INTERVAL,
                       CONSISTENCY_TABLE_ID_IGNORED_RANGE, FLOWS_DICT_MAX_SIZE)


def cast_fields(flow_dict):
    """Make casting the match fields from UBInt() to native int ."""
    match = flow_dict['match']
    for field, value in match.items():
        if isinstance(value, UBIntBase):
            match[field] = int(value)
    flow_dict['match'] = match
    return flow_dict


def _validate_range(values):
    """Check that the range of flows ignored by the consistency is valid."""
    if len(values) != 2:
        msg = f'The tuple must have 2 items, not {len(values)}'
        raise ValueError(msg)
    first, second = values
    if second < first:
        msg = f'The first value is bigger than the second: {values}'
        raise ValueError(msg)
    if not isinstance(first, int) or not isinstance(second, int):
        msg = f'Expected a tuple of integers, received {values}'
        raise TypeError(msg)


def _valid_consistency_ignored(consistency_ignored_list):
    """Check the format of the list of ignored consistency flows.

    Check that the list of ignored flows in the consistency check
    is well formatted. Returns True, if the list is well
    formatted, otherwise return False.
    """
    msg = ('The list of ignored flows in the consistency check'
           'is not well formatted, it will be ignored: %s')
    for consistency_ignored in consistency_ignored_list:
        if isinstance(consistency_ignored, tuple):
            try:
                _validate_range(consistency_ignored)
            except (TypeError, ValueError) as error:
                log.warn(msg, error)
                return False
        elif not isinstance(consistency_ignored, (int, tuple)):
            error_msg = ('The elements must be of class int or tuple'
                         f' but they are: {type(consistency_ignored)}')
            log.warn(msg, error_msg)
            return False
    return True


class Main(KytosNApp):
    """Main class to be used by Kytos controller."""

    def setup(self):
        """Replace the 'init' method for the KytosApp subclass.

        The setup method is automatically called by the run method.
        Users shouldn't call this method directly.
        """
        log.debug("flow-manager starting")
        self._flow_mods_sent = OrderedDict()
        self._flow_mods_sent_max_size = FLOWS_DICT_MAX_SIZE
        self.cookie_ignored_range = []
        self.tab_id_ignored_range = []
        if _valid_consistency_ignored(CONSISTENCY_COOKIE_IGNORED_RANGE):
            self.cookie_ignored_range = CONSISTENCY_COOKIE_IGNORED_RANGE
        if _valid_consistency_ignored(CONSISTENCY_TABLE_ID_IGNORED_RANGE):
            self.tab_id_ignored_range = CONSISTENCY_TABLE_ID_IGNORED_RANGE

        # Storehouse client to save and restore flow data:
        self.storehouse = StoreHouse(self.controller)

        # Format of stored flow data:
        # {'flow_persistence': {'dpid_str': {'flow_list': [
        #                                     {'command': '<add|delete>',
        #                                      'flow': {flow_dict}}]}}}
        self.stored_flows = {}
        self.resent_flows = set()
        if CONSISTENCY_INTERVAL > 0:
            self.execute_as_loop(CONSISTENCY_INTERVAL)

    def execute(self):
        """Run once on NApp 'start' or in a loop.

        The execute method is called by the run method of KytosNApp class.
        Users shouldn't call this method directly.
        """
        self._load_flows()

        if CONSISTENCY_INTERVAL > 0:
            self.consistency_check()

    def shutdown(self):
        """Shutdown routine of the NApp."""
        log.debug("flow-manager stopping")

    @listen_to('kytos/of_core.handshake.completed')
    def resend_stored_flows(self, event):
        """Resend stored Flows."""
        # if consistency check is enabled, it should take care of this
        if CONSISTENCY_INTERVAL >= 0:
            return
        switch = event.content['switch']
        dpid = str(switch.dpid)
        # This can be a problem because this code is running a thread
        if dpid in self.resent_flows:
            log.debug(f'Flow already resent to the switch {dpid}')
            return
        if dpid in self.stored_flows:
            flow_list = self.stored_flows[dpid]['flow_list']
            for flow in flow_list:
                command = flow['command']
                flows_dict = {"flows": [flow['flow']]}
                self._install_flows(command, flows_dict, [switch])
            self.resent_flows.add(dpid)
            log.info(f'Flows resent to Switch {dpid}')

    @staticmethod
    def is_ignored(field, ignored_range):
        """Check that the flow field is in the range of ignored flows.

        Returns True, if the field is in the range of ignored flows,
        otherwise it returns False.
        """
        for i in ignored_range:
            if isinstance(i, tuple):
                start_range, end_range = i
                if start_range <= field <= end_range:
                    return True
            if isinstance(i, int):
                if field == i:
                    return True
        return False

    def consistency_ignored_check(self, flow):
        """Check if the flow is in the list of flows ignored by consistency.

        Check by `cookie` range and `table_id` range.
        Return True if the flow is in the ignored range, otherwise return
        False.
        """
        # Check by cookie
        if self.is_ignored(flow.cookie, self.cookie_ignored_range):
            return True

        # Check by `table_id`
        if self.is_ignored(flow.table_id, self.tab_id_ignored_range):
            return True
        return False

    def consistency_check(self):
        """Check the consistency of flows in each switch."""
        switches = self.controller.switches.values()

        for switch in switches:
            # Check if a dpid is a key in 'stored_flows' dictionary
            if switch.is_enabled():
                self.check_storehouse_consistency(switch)

                if switch.dpid in self.stored_flows:
                    self.check_switch_consistency(switch)

    def check_switch_consistency(self, switch):
        """Check consistency of installed flows for a specific switch."""
        dpid = switch.dpid

        # Flows stored in storehouse
        stored_flows = self.stored_flows[dpid]['flow_list']

        serializer = FlowFactory.get_class(switch)

        for stored_flow in stored_flows:
            command = stored_flow['command']
            stored_flow_obj = serializer.from_dict(stored_flow['flow'], switch)

            flow = {'flows': [stored_flow['flow']]}

            if stored_flow_obj not in switch.flows:
                if command == 'add':
                    log.info('A consistency problem was detected in '
                             f'switch {dpid}.')
                    self._install_flows(command, flow, [switch])
                    log.info(f'Flow forwarded to switch {dpid} to be '
                             'installed.')
            elif command == 'delete':
                log.info('A consistency problem was detected in '
                         f'switch {dpid}.')
                command = 'delete_strict'
                self._install_flows(command, flow, [switch])
                log.info(f'Flow forwarded to switch {dpid} to be deleted.')

    def check_storehouse_consistency(self, switch):
        """Check consistency of installed flows for a specific switch."""
        dpid = switch.dpid

        for installed_flow in switch.flows:

            # Check if the flow are in the ignored flow list
            if self.consistency_ignored_check(installed_flow):
                continue

            if dpid not in self.stored_flows:
                log.info('A consistency problem was detected in '
                         f'switch {dpid}.')
                flow = {'flows': [installed_flow.as_dict()]}
                command = 'delete_strict'
                self._install_flows(command, flow, [switch])
                log.info(f'Flow forwarded to switch {dpid} to be deleted.')
            else:
                serializer = FlowFactory.get_class(switch)
                stored_flows = self.stored_flows[dpid]['flow_list']
                stored_flows_list = [serializer.from_dict(stored_flow['flow'],
                                                          switch)
                                     for stored_flow in stored_flows]

                if installed_flow not in stored_flows_list:
                    log.info('A consistency problem was detected in '
                             f'switch {dpid}.')
                    flow = {'flows': [installed_flow.as_dict()]}
                    command = 'delete_strict'
                    self._install_flows(command, flow, [switch])
                    log.info(f'Flow forwarded to switch {dpid} to be deleted.')

    # pylint: disable=attribute-defined-outside-init
    def _load_flows(self):
        """Load stored flows."""
        try:
            data = self.storehouse.get_data()['flow_persistence']
            if 'id' in data:
                del data['id']
            self.stored_flows = data
        except (KeyError, FileNotFoundError) as error:
            log.debug(f'There are no flows to load: {error}')
        else:
            log.info('Flows loaded.')

    def _store_changed_flows(self, command, flow, switch):
        """Store changed flows.

        Args:
            command: Flow command to be installed
            flow: Flows to be stored
            switch: Switch target
        """
        stored_flows_box = deepcopy(self.stored_flows)
        # if the flow has a destination dpid it can be stored.
        if not switch:
            log.info('The Flow cannot be stored, the destination switch '
                     f'have not been specified: {switch}')
            return
        installed_flow = {}
        flow_list = []
        installed_flow['command'] = command
        installed_flow['flow'] = flow
        deleted_flows = []

        serializer = FlowFactory.get_class(switch)
        installed_flow_obj = serializer.from_dict(flow, switch)

        if switch.id not in stored_flows_box:
            # Switch not stored, add to box.
            flow_list.append(installed_flow)
            stored_flows_box[switch.id] = {'flow_list': flow_list}
        else:
            stored_flows = stored_flows_box[switch.id].get('flow_list', [])
            # Check if flow already stored
            for stored_flow in stored_flows:
                stored_flow_obj = serializer.from_dict(stored_flow['flow'],
                                                       switch)

                version = switch.connection.protocol.version

                if installed_flow['command'] == 'delete':
                    # No strict match
                    if match_flow(flow, version, stored_flow['flow']):
                        deleted_flows.append(stored_flow)

                elif installed_flow_obj == stored_flow_obj:
                    if stored_flow['command'] == installed_flow['command']:
                        log.debug('Data already stored.')
                        return
                    # Flow with inconsistency in "command" fields : Remove the
                    # old instruction. This happens when there is a stored
                    # instruction to install the flow, but the new instruction
                    # is to remove it. In this case, the old instruction is
                    # removed and the new one is stored.
                    stored_flow['command'] = installed_flow.get('command')
                    deleted_flows.append(stored_flow)
                    break

            # if installed_flow['command'] != 'delete':
            stored_flows.append(installed_flow)
            for i in deleted_flows:
                stored_flows.remove(i)
            stored_flows_box[switch.id]['flow_list'] = stored_flows

        stored_flows_box['id'] = 'flow_persistence'
        self.storehouse.save_flow(stored_flows_box)
        del stored_flows_box['id']
        self.stored_flows = deepcopy(stored_flows_box)

    @rest('v2/flows')
    @rest('v2/flows/<dpid>')
    def list(self, dpid=None):
        """Retrieve all flows from a switch identified by dpid.

        If no dpid is specified, return all flows from all switches.
        """
        if dpid is None:
            switches = self.controller.switches.values()
        else:
            switches = [self.controller.get_switch_by_dpid(dpid)]

            if not any(switches):
                raise NotFound("Switch not found")

        switch_flows = {}

        for switch in switches:
            flows_dict = [cast_fields(flow.as_dict())
                          for flow in switch.flows]
            switch_flows[switch.dpid] = {'flows': flows_dict}

        return jsonify(switch_flows)

    @rest('v2/flows', methods=['POST'])
    @rest('v2/flows/<dpid>', methods=['POST'])
    def add(self, dpid=None):
        """Install new flows in the switch identified by dpid.

        If no dpid is specified, install flows in all switches.
        """
        return self._send_flow_mods_from_request(dpid, "add")

    @rest('v2/delete', methods=['POST'])
    @rest('v2/delete/<dpid>', methods=['POST'])
    @rest('v2/flows', methods=['DELETE'])
    @rest('v2/flows/<dpid>', methods=['DELETE'])
    def delete(self, dpid=None):
        """Delete existing flows in the switch identified by dpid.

        If no dpid is specified, delete flows from all switches.
        """
        return self._send_flow_mods_from_request(dpid, "delete")

    def _get_all_switches_enabled(self):
        """Get a list of all switches enabled."""
        switches = self.controller.switches.values()
        return [switch for switch in switches if switch.is_enabled()]

    def _send_flow_mods_from_request(self, dpid, command, flows_dict=None):
        """Install FlowsMods from request."""
        if flows_dict is None:
            flows_dict = request.get_json()
            if flows_dict is None:
                return jsonify({"response": 'flows dict is none.'}), 404

        if dpid:
            switch = self.controller.get_switch_by_dpid(dpid)
            if not switch:
                return jsonify({"response": 'dpid not found.'}), 404
            elif switch.is_enabled() is False:
                if command == "delete":
                    self._install_flows(command, flows_dict, [switch])
                else:
                    return jsonify({"response": 'switch is disabled.'}), 404
            else:
                self._install_flows(command, flows_dict, [switch])
        else:
            self._install_flows(command, flows_dict,
                                self._get_all_switches_enabled())

        return jsonify({"response": "FlowMod Messages Sent"})

    def _install_flows(self, command, flows_dict, switches=[]):
        """Execute all procedures to install flows in the switches.

        Args:
            command: Flow command to be installed
            flows_dict: Dictionary with flows to be installed in the switches.
            switches: A list of switches
        """
        for switch in switches:
            serializer = FlowFactory.get_class(switch)
            flows = flows_dict.get('flows', [])
            for flow_dict in flows:
                flow = serializer.from_dict(flow_dict, switch)
                if command == "delete":
                    flow_mod = flow.as_of_delete_flow_mod()
                elif command == "delete_strict":
                    flow_mod = flow.as_of_strict_delete_flow_mod()
                elif command == "add":
                    flow_mod = flow.as_of_add_flow_mod()
                else:
                    raise InvalidCommandError
                self._send_flow_mod(flow.switch, flow_mod)
                self._add_flow_mod_sent(flow_mod.header.xid, flow, command)

                self._send_napp_event(switch, flow, command)
                self._store_changed_flows(command, flow_dict, switch)

    def _add_flow_mod_sent(self, xid, flow, command):
        """Add the flow mod to the list of flow mods sent."""
        if len(self._flow_mods_sent) >= self._flow_mods_sent_max_size:
            self._flow_mods_sent.popitem(last=False)
        self._flow_mods_sent[xid] = (flow, command)

    def _send_flow_mod(self, switch, flow_mod):
        event_name = 'kytos/flow_manager.messages.out.ofpt_flow_mod'

        content = {'destination': switch.connection,
                   'message': flow_mod}

        event = KytosEvent(name=event_name, content=content)
        self.controller.buffers.msg_out.put(event)

    def _send_napp_event(self, switch, flow, command, **kwargs):
        """Send an Event to other apps informing about a FlowMod."""
        if command == 'add':
            name = 'kytos/flow_manager.flow.added'
        elif command in ('delete', 'delete_strict'):
            name = 'kytos/flow_manager.flow.removed'
        elif command == 'error':
            name = 'kytos/flow_manager.flow.error'
        else:
            raise InvalidCommandError
        content = {'datapath': switch,
                   'flow': flow}
        content.update(kwargs)
        event_app = KytosEvent(name, content)
        self.controller.buffers.app.put(event_app)

    @listen_to('.*.of_core.*.ofpt_error')
    def handle_errors(self, event):
        """Receive OpenFlow error and send a event.

        The event is sent only if the error is related to a request made
        by flow_manager.
        """
        message = event.content["message"]

        connection = event.source
        switch = connection.switch

        xid = message.header.xid.value
        error_type = message.error_type
        error_code = message.code
        error_data = message.data.pack()

        # Get the packet responsible for the error
        error_packet = connection.protocol.unpack(error_data)

        if message.code == BadActionCode.OFPBAC_BAD_OUT_PORT:
            actions = []
            if hasattr(error_packet, 'actions'):
                # Get actions from the flow mod (OF 1.0)
                actions = error_packet.actions
            else:
                # Get actions from the list of flow mod instructions (OF 1.3)
                for instruction in error_packet.instructions:
                    actions.extend(instruction.actions)

            for action in actions:
                iface = switch.get_interface_by_port_no(action.port)

                # Set interface to drop packets forwarded to it
                if iface:
                    iface.config = PortConfig.OFPPC_NO_FWD

        try:
            flow, error_command = self._flow_mods_sent[xid]
        except KeyError:
            pass
        else:
            self._send_napp_event(flow.switch, flow, 'error',
                                  error_command=error_command,
                                  error_type=error_type, error_code=error_code)


1		"""kytos/flow_manager NApp installs, lists and deletes switch flows."""
2	1	from collections import OrderedDict
3	1	from copy import deepcopy
4
5	1	from flask import jsonify, request
6	1	from pyof.foundation.base import UBIntBase
7	1	from pyof.v0x01.asynchronous.error_msg import BadActionCode
8	1	from pyof.v0x01.common.phy_port import PortConfig
9	1	from werkzeug.exceptions import NotFound
10
11	1	from kytos.core import KytosEvent, KytosNApp, log, rest
12	1	from kytos.core.helpers import listen_to
13	1	from napps.kytos.flow_manager.match import match_flow
14	1	from napps.kytos.flow_manager.storehouse import StoreHouse
15	1	from napps.kytos.of_core.flow import FlowFactory
16
17	1	from .exceptions import InvalidCommandError
18	1	from .settings import (CONSISTENCY_COOKIE_IGNORED_RANGE, CONSISTENCY_INTERVAL,
19		CONSISTENCY_TABLE_ID_IGNORED_RANGE, FLOWS_DICT_MAX_SIZE)
20
21
22	1	def cast_fields(flow_dict):
23		"""Make casting the match fields from UBInt() to native int ."""
24	1	match = flow_dict['match']
25	1	for field, value in match.items():
26	1	if isinstance(value, UBIntBase):
27		match[field] = int(value)
28	1	flow_dict['match'] = match
29	1	return flow_dict
30
31
32	1	def _validate_range(values):
33		"""Check that the range of flows ignored by the consistency is valid."""
34		if len(values) != 2:
35		msg = f'The tuple must have 2 items, not {len(values)}'
36		raise ValueError(msg)
37		first, second = values
38		if second < first:
39		msg = f'The first value is bigger than the second: {values}'
40		raise ValueError(msg)
41		if not isinstance(first, int) or not isinstance(second, int):
42		msg = f'Expected a tuple of integers, received {values}'
43		raise TypeError(msg)
44
45
46	1	def _valid_consistency_ignored(consistency_ignored_list):
47		"""Check the format of the list of ignored consistency flows.
48
49		Check that the list of ignored flows in the consistency check
50		is well formatted. Returns True, if the list is well
51		formatted, otherwise return False.
52		"""
53	1	msg = ('The list of ignored flows in the consistency check'
54		'is not well formatted, it will be ignored: %s')
55	1	for consistency_ignored in consistency_ignored_list:
56		if isinstance(consistency_ignored, tuple):
57		try:
58		_validate_range(consistency_ignored)
59		except (TypeError, ValueError) as error:
60		log.warn(msg, error)
61		return False
62		elif not isinstance(consistency_ignored, (int, tuple)):
63		error_msg = ('The elements must be of class int or tuple'
64		f' but they are: {type(consistency_ignored)}')
65		log.warn(msg, error_msg)
66		return False
67	1	return True
68
69
70	1	class Main(KytosNApp):
71		"""Main class to be used by Kytos controller."""
72
73	1	def setup(self):
74		"""Replace the 'init' method for the KytosApp subclass.
75
76		The setup method is automatically called by the run method.
77		Users shouldn't call this method directly.
78		"""
79	1	log.debug("flow-manager starting")
80	1	self._flow_mods_sent = OrderedDict()
81	1	self._flow_mods_sent_max_size = FLOWS_DICT_MAX_SIZE
82	1	self.cookie_ignored_range = []
83	1	self.tab_id_ignored_range = []
84	1	if _valid_consistency_ignored(CONSISTENCY_COOKIE_IGNORED_RANGE):
85	1	self.cookie_ignored_range = CONSISTENCY_COOKIE_IGNORED_RANGE
86	1	if _valid_consistency_ignored(CONSISTENCY_TABLE_ID_IGNORED_RANGE):
87	1	self.tab_id_ignored_range = CONSISTENCY_TABLE_ID_IGNORED_RANGE
88
89		# Storehouse client to save and restore flow data:
90	1	self.storehouse = StoreHouse(self.controller)
91
92		# Format of stored flow data:
93		# {'flow_persistence': {'dpid_str': {'flow_list': [
94		# {'command': '<add\|delete>',
95		# 'flow': {flow_dict}}]}}}
96	1	self.stored_flows = {}
97	1	self.resent_flows = set()
98	1	if CONSISTENCY_INTERVAL > 0:
99	1	self.execute_as_loop(CONSISTENCY_INTERVAL)
100
101	1	def execute(self):
102		"""Run once on NApp 'start' or in a loop.
103
104		The execute method is called by the run method of KytosNApp class.
105		Users shouldn't call this method directly.
106		"""
107		self._load_flows()
108
109		if CONSISTENCY_INTERVAL > 0:
110		self.consistency_check()
111
112	1	def shutdown(self):
113		"""Shutdown routine of the NApp."""
114		log.debug("flow-manager stopping")
115
116	1	@listen_to('kytos/of_core.handshake.completed')
117		def resend_stored_flows(self, event):
118		"""Resend stored Flows."""
119		# if consistency check is enabled, it should take care of this
120	1	if CONSISTENCY_INTERVAL >= 0:
121		return
122	1	switch = event.content['switch']
123	1	dpid = str(switch.dpid)
124		# This can be a problem because this code is running a thread
125	1	if dpid in self.resent_flows:
126		log.debug(f'Flow already resent to the switch {dpid}')
127		return
128	1	if dpid in self.stored_flows:
129	1	flow_list = self.stored_flows[dpid]['flow_list']
130	1	for flow in flow_list:
131	1	command = flow['command']
132	1	flows_dict = {"flows": [flow['flow']]}
133	1	self._install_flows(command, flows_dict, [switch])
134	1	self.resent_flows.add(dpid)
135	1	log.info(f'Flows resent to Switch {dpid}')
136
137	1	@staticmethod
138		def is_ignored(field, ignored_range):
139		"""Check that the flow field is in the range of ignored flows.
140
141		Returns True, if the field is in the range of ignored flows,
142		otherwise it returns False.
143		"""
144	1	for i in ignored_range:
145	1	if isinstance(i, tuple):
146	1	start_range, end_range = i
147	1	if start_range <= field <= end_range:
148	1	return True
149	1	if isinstance(i, int):
150	1	if field == i:
151	1	return True
152	1	return False
153
154	1	def consistency_ignored_check(self, flow):
155		"""Check if the flow is in the list of flows ignored by consistency.
156
157		Check by `cookie` range and `table_id` range.
158		Return True if the flow is in the ignored range, otherwise return
159		False.
160		"""
161		# Check by cookie
162	1	if self.is_ignored(flow.cookie, self.cookie_ignored_range):
163	1	return True
164
165		# Check by `table_id`
166	1	if self.is_ignored(flow.table_id, self.tab_id_ignored_range):
167	1	return True
168	1	return False
169
170	1	def consistency_check(self):
171		"""Check the consistency of flows in each switch."""
172		switches = self.controller.switches.values()
173
174		for switch in switches:
175		# Check if a dpid is a key in 'stored_flows' dictionary
176		if switch.is_enabled():
177		self.check_storehouse_consistency(switch)
178
179		if switch.dpid in self.stored_flows:
180		self.check_switch_consistency(switch)
181
182	1	def check_switch_consistency(self, switch):
183		"""Check consistency of installed flows for a specific switch."""
184	1	dpid = switch.dpid
185
186		# Flows stored in storehouse
187	1	stored_flows = self.stored_flows[dpid]['flow_list']
188
189	1	serializer = FlowFactory.get_class(switch)
190
191	1	for stored_flow in stored_flows:
192	1	command = stored_flow['command']
193	1	stored_flow_obj = serializer.from_dict(stored_flow['flow'], switch)
194
195	1	flow = {'flows': [stored_flow['flow']]}
196
197	1	if stored_flow_obj not in switch.flows:
198	1	if command == 'add':
199	1	log.info('A consistency problem was detected in '
200		f'switch {dpid}.')
201	1	self._install_flows(command, flow, [switch])
202	1	log.info(f'Flow forwarded to switch {dpid} to be '
203		'installed.')
204	1	elif command == 'delete':
205	1	log.info('A consistency problem was detected in '
206		f'switch {dpid}.')
207	1	command = 'delete_strict'
208	1	self._install_flows(command, flow, [switch])
209	1	log.info(f'Flow forwarded to switch {dpid} to be deleted.')
210
211	1	def check_storehouse_consistency(self, switch):
212		"""Check consistency of installed flows for a specific switch."""
213	1	dpid = switch.dpid
214
215	1	for installed_flow in switch.flows:
216
217		# Check if the flow are in the ignored flow list
218	1	if self.consistency_ignored_check(installed_flow):
219	1	continue
220
221	1	if dpid not in self.stored_flows:
222		log.info('A consistency problem was detected in '
223		f'switch {dpid}.')
224		flow = {'flows': [installed_flow.as_dict()]}
225		command = 'delete_strict'
226		self._install_flows(command, flow, [switch])
227		log.info(f'Flow forwarded to switch {dpid} to be deleted.')
228		else:
229	1	serializer = FlowFactory.get_class(switch)
230	1	stored_flows = self.stored_flows[dpid]['flow_list']
231	1	stored_flows_list = [serializer.from_dict(stored_flow['flow'],
232		switch)
233		for stored_flow in stored_flows]
234
235	1	if installed_flow not in stored_flows_list:
236	1	log.info('A consistency problem was detected in '
237		f'switch {dpid}.')
238	1	flow = {'flows': [installed_flow.as_dict()]}
239	1	command = 'delete_strict'
240	1	self._install_flows(command, flow, [switch])
241	1	log.info(f'Flow forwarded to switch {dpid} to be deleted.')
242
243		# pylint: disable=attribute-defined-outside-init
244	1	def _load_flows(self):
245		"""Load stored flows."""
246	1	try:
247	1	data = self.storehouse.get_data()['flow_persistence']
248	1	if 'id' in data:
249		del data['id']
250	1	self.stored_flows = data
251		except (KeyError, FileNotFoundError) as error:
252		log.debug(f'There are no flows to load: {error}')
253		else:
254	1	log.info('Flows loaded.')
255
256	1	def _store_changed_flows(self, command, flow, switch):
257		"""Store changed flows.
258
259		Args:
260		command: Flow command to be installed
261		flow: Flows to be stored
262		switch: Switch target
263		"""
264	1	stored_flows_box = deepcopy(self.stored_flows)
265		# if the flow has a destination dpid it can be stored.
266	1	if not switch:
267		log.info('The Flow cannot be stored, the destination switch '
268		f'have not been specified: {switch}')
269		return
270	1	installed_flow = {}
271	1	flow_list = []
272	1	installed_flow['command'] = command
273	1	installed_flow['flow'] = flow
274	1	deleted_flows = []
275
276	1	serializer = FlowFactory.get_class(switch)
277	1	installed_flow_obj = serializer.from_dict(flow, switch)
278
279	1	if switch.id not in stored_flows_box:
280		# Switch not stored, add to box.
281	1	flow_list.append(installed_flow)
282	1	stored_flows_box[switch.id] = {'flow_list': flow_list}
283		else:
284	1	stored_flows = stored_flows_box[switch.id].get('flow_list', [])
285		# Check if flow already stored
286	1	for stored_flow in stored_flows:
287	1	stored_flow_obj = serializer.from_dict(stored_flow['flow'],
288		switch)
289
290	1	version = switch.connection.protocol.version
291
292	1	if installed_flow['command'] == 'delete':
293		# No strict match
294	1	if match_flow(flow, version, stored_flow['flow']):
295	1	deleted_flows.append(stored_flow)
296
297	1	elif installed_flow_obj == stored_flow_obj:
298	1	if stored_flow['command'] == installed_flow['command']:
299		log.debug('Data already stored.')
300		return
301		# Flow with inconsistency in "command" fields : Remove the
302		# old instruction. This happens when there is a stored
303		# instruction to install the flow, but the new instruction
304		# is to remove it. In this case, the old instruction is
305		# removed and the new one is stored.
306	1	stored_flow['command'] = installed_flow.get('command')
307	1	deleted_flows.append(stored_flow)
308	1	break
309
310		# if installed_flow['command'] != 'delete':
311	1	stored_flows.append(installed_flow)
312	1	for i in deleted_flows:
313	1	stored_flows.remove(i)
314	1	stored_flows_box[switch.id]['flow_list'] = stored_flows
315
316	1	stored_flows_box['id'] = 'flow_persistence'
317	1	self.storehouse.save_flow(stored_flows_box)
318	1	del stored_flows_box['id']
319	1	self.stored_flows = deepcopy(stored_flows_box)
320
321	1	@rest('v2/flows')
322	1	@rest('v2/flows/<dpid>')
323	1	def list(self, dpid=None):
324		"""Retrieve all flows from a switch identified by dpid.
325
326		If no dpid is specified, return all flows from all switches.
327		"""
328	1	if dpid is None:
329	1	switches = self.controller.switches.values()
330		else:
331	1	switches = [self.controller.get_switch_by_dpid(dpid)]
332
333	1	if not any(switches):
334	1	raise NotFound("Switch not found")
335
336	1	switch_flows = {}
337
338	1	for switch in switches:
339	1	flows_dict = [cast_fields(flow.as_dict())
340		for flow in switch.flows]
341	1	switch_flows[switch.dpid] = {'flows': flows_dict}
342
343	1	return jsonify(switch_flows)
344
345	1	@rest('v2/flows', methods=['POST'])
346	1	@rest('v2/flows/<dpid>', methods=['POST'])
347	1	def add(self, dpid=None):
348		"""Install new flows in the switch identified by dpid.
349
350		If no dpid is specified, install flows in all switches.
351		"""
352	1	return self._send_flow_mods_from_request(dpid, "add")
353
354	1	@rest('v2/delete', methods=['POST'])
355	1	@rest('v2/delete/<dpid>', methods=['POST'])
356	1	@rest('v2/flows', methods=['DELETE'])
357	1	@rest('v2/flows/<dpid>', methods=['DELETE'])
358	1	def delete(self, dpid=None):
359		"""Delete existing flows in the switch identified by dpid.
360
361		If no dpid is specified, delete flows from all switches.
362		"""
363	1	return self._send_flow_mods_from_request(dpid, "delete")
364
365	1	def _get_all_switches_enabled(self):
366		"""Get a list of all switches enabled."""
367	1	switches = self.controller.switches.values()
368	1	return [switch for switch in switches if switch.is_enabled()]
369
370	1	def _send_flow_mods_from_request(self, dpid, command, flows_dict=None):
371		"""Install FlowsMods from request."""
372	1	if flows_dict is None:
373	1	flows_dict = request.get_json()
374	1	if flows_dict is None:
375	1	return jsonify({"response": 'flows dict is none.'}), 404
376
377	1	if dpid:
378	1	switch = self.controller.get_switch_by_dpid(dpid)
379	1	if not switch:
380	1	return jsonify({"response": 'dpid not found.'}), 404
381	1	elif switch.is_enabled() is False:
382	1	if command == "delete":
383	1	self._install_flows(command, flows_dict, [switch])
384		else:
385	1	return jsonify({"response": 'switch is disabled.'}), 404
386		else:
387	1	self._install_flows(command, flows_dict, [switch])
388		else:
389	1	self._install_flows(command, flows_dict,
390		self._get_all_switches_enabled())
391
392	1	return jsonify({"response": "FlowMod Messages Sent"})
393
394	1	def _install_flows(self, command, flows_dict, switches=[]):
395		"""Execute all procedures to install flows in the switches.
396
397		Args:
398		command: Flow command to be installed
399		flows_dict: Dictionary with flows to be installed in the switches.
400		switches: A list of switches
401		"""
402	1	for switch in switches:
403	1	serializer = FlowFactory.get_class(switch)
404	1	flows = flows_dict.get('flows', [])
405	1	for flow_dict in flows:
406	1	flow = serializer.from_dict(flow_dict, switch)
407	1	if command == "delete":
408		flow_mod = flow.as_of_delete_flow_mod()
409	1	elif command == "delete_strict":
410	1	flow_mod = flow.as_of_strict_delete_flow_mod()
411	1	elif command == "add":
412	1	flow_mod = flow.as_of_add_flow_mod()
413		else:
414		raise InvalidCommandError
415	1	self._send_flow_mod(flow.switch, flow_mod)
416	1	self._add_flow_mod_sent(flow_mod.header.xid, flow, command)
417
418	1	self._send_napp_event(switch, flow, command)
419	1	self._store_changed_flows(command, flow_dict, switch)
420
421	1	def _add_flow_mod_sent(self, xid, flow, command):
422		"""Add the flow mod to the list of flow mods sent."""
423	1	if len(self._flow_mods_sent) >= self._flow_mods_sent_max_size:
424		self._flow_mods_sent.popitem(last=False)
425	1	self._flow_mods_sent[xid] = (flow, command)
426
427	1	def _send_flow_mod(self, switch, flow_mod):
428	1	event_name = 'kytos/flow_manager.messages.out.ofpt_flow_mod'
429
430	1	content = {'destination': switch.connection,
431		'message': flow_mod}
432
433	1	event = KytosEvent(name=event_name, content=content)
434	1	self.controller.buffers.msg_out.put(event)
435
436	1	def _send_napp_event(self, switch, flow, command, **kwargs):
437		"""Send an Event to other apps informing about a FlowMod."""
438	1	if command == 'add':
439	1	name = 'kytos/flow_manager.flow.added'
440	1	elif command in ('delete', 'delete_strict'):
441	1	name = 'kytos/flow_manager.flow.removed'
442	1	elif command == 'error':
443	1	name = 'kytos/flow_manager.flow.error'
444		else:
445		raise InvalidCommandError
446	1	content = {'datapath': switch,
447		'flow': flow}
448	1	content.update(kwargs)
449	1	event_app = KytosEvent(name, content)
450	1	self.controller.buffers.app.put(event_app)
451
452	1	@listen_to('..of_core..ofpt_error')
453		def handle_errors(self, event):
454		"""Receive OpenFlow error and send a event.
455
456		The event is sent only if the error is related to a request made
457		by flow_manager.
458		"""
459	1	message = event.content["message"]
460
461	1	connection = event.source
462	1	switch = connection.switch
463
464	1	xid = message.header.xid.value
465	1	error_type = message.error_type
466	1	error_code = message.code
467	1	error_data = message.data.pack()
468
469		# Get the packet responsible for the error
470	1	error_packet = connection.protocol.unpack(error_data)
471
472	1	if message.code == BadActionCode.OFPBAC_BAD_OUT_PORT:
473		actions = []
474		if hasattr(error_packet, 'actions'):
475		# Get actions from the flow mod (OF 1.0)
476		actions = error_packet.actions
477		else:
478		# Get actions from the list of flow mod instructions (OF 1.3)
479		for instruction in error_packet.instructions:
480		actions.extend(instruction.actions)
481
482		for action in actions:
483		iface = switch.get_interface_by_port_no(action.port)
484
485		# Set interface to drop packets forwarded to it
486		if iface:
487		iface.config = PortConfig.OFPPC_NO_FWD
488
489	1	try:
490	1	flow, error_command = self._flow_mods_sent[xid]
491		except KeyError:
492		pass
493		else:
494	1	self._send_napp_event(flow.switch, flow, 'error',
495		error_command=error_command,
496		error_type=error_type, error_code=error_code)
497

kytos / flow_manager

Pull Request — master (#112)

build.main._validate_range() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like