build.main.Main._load_flows() - Code Metrics - Inspection of "Change behavior when handling an OFPBAC_BAD_OUT_PO..." - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#92)

by Jose

created 2021-01-12 17:12 UTC

build.main.Main._load_flows() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	12
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	4.5923

Importance

Changes

Metric	Value
cc	4
eloc	9
nop	1
dl	0
loc	12
ccs	6
cts	9
cp	0.6667
crap	4.5923
rs	9.95
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_INTERVAL, FLOWS_DICT_MAX_SIZE


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

        # 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.info(f'Flow already resended to 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}')

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

kytos / flow_manager

Pull Request — master (#92)

build.main.Main._load_flows() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like