build.main.Main.execute() - 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-03-03 14:33 UTC

build.main.Main.execute() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	10
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	3.6875

Importance

Changes

Metric	Value
cc	2
eloc	4
nop	1
dl	0
loc	10
ccs	1
cts	4
cp	0.25
crap	3.6875
rs	10
c	0
b	0
f	0

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

from flask import jsonify, request
from pyof.v0x01.asynchronous.error_msg import BadActionCode
from pyof.v0x01.common.phy_port import PortConfig

from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to
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_EXCEPTION_RANGE,
                       CONSISTENCY_INTERVAL,
                       CONSISTENCY_TABLE_ID_EXCEPTION_RANGE,
                       FLOWS_DICT_MAX_SIZE)


def _validate_range(values):
    """Check if the consistency exception range is valid."""
    instance = int
    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, instance) or not isinstance(second, instance):
        msg = 'The elements of the range must be of the class {instance}'
        raise TypeError(msg)


def _valid_consistency_exceptions(consistency_exception_list):
    """Check that the list of consistency exceptions is well formatted.

    Returns True if the consistency exception list is well formatted or False
    otherwise.
    """
    msg = ('The list of consistency exception is not well formatted'
           ', it will be ignored: %s')
    for consistency_exception in consistency_exception_list:
        if isinstance(consistency_exception, tuple):
            try:
                _validate_range(consistency_exception)
            except (TypeError, ValueError) as error:
                log.warn(msg, error)
                return False
        elif not isinstance(consistency_exception, (int, tuple)):
            error_msg = ('The elements must be of class int or tuple'
                         f' but they are: {type(consistency_exception)}')
            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_exception_range = []
        self.tab_id_exception_range = []
        if _valid_consistency_exceptions(CONSISTENCY_COOKIE_EXCEPTION_RANGE):
            self.cookie_exception_range = CONSISTENCY_COOKIE_EXCEPTION_RANGE
        if _valid_consistency_exceptions(CONSISTENCY_TABLE_ID_EXCEPTION_RANGE):
            self.tab_id_exception_range = CONSISTENCY_TABLE_ID_EXCEPTION_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."""
        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_exception(field, exceptions):
        """Check if the field is in the list of exceptions.

        Returns True if the field is in the list of exceptions.
        """
        for i in exceptions:
            if isinstance(i, tuple):
                begin_interval, end_interval = i
                if begin_interval <= field <= end_interval:
                    return True
            if isinstance(i, int):
                if field == i:
                    return True
        return False

    def consistency_exception_check(self, flow):
        """Verify if the flow is in the exception range.

        Check by `cookie` range and `table_id` range.
        Returns True if the flow is in the exception list.
        """
        # Check by cookie
        if self.is_exception(flow.cookie, self.cookie_exception_range):
            return True

        # Check by `table_id`
        if self.is_exception(flow.table_id, self.tab_id_exception_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}.')
                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 excluded flow list
            if self.consistency_exception_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'
                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'
                    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 = self.stored_flows.copy()
        # 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

        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)
                if 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')
                    stored_flows.remove(stored_flow)
                    break

            stored_flows.append(installed_flow)
            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 = stored_flows_box.copy()

    @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)]

        switch_flows = {}

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

kytos / flow_manager

Pull Request — master (#112)

build.main.Main.execute() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like