build.main.Main.handle_errors() - 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 2020-11-12 18:08 UTC

build.main.Main.handle_errors() B

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	40
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	7.7656

Importance

Changes

Metric	Value
eloc	27
dl	0
loc	40
rs	7.8319
c	0
b	0
f	0
ccs	6
cts	8
cp	0.75
cc	7
nop	2
crap	7.7656

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

from flask import jsonify, request
from pyof import v0x01

from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to
from napps.kytos.of_core.flow import FlowFactory

from .exceptions import InvalidCommandError
from .settings import 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

    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.
        """

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

    @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):
        """Install FlowsMods from request."""
        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:
                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)

    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)

        error = v0x01.asynchronous.error_msg.BadActionCode.OFPBAC_BAD_OUT_PORT
        if message.code == error:
            for action in error_packet.actions:
                try:
                    iface = switch.get_interface_by_port_no(action.port.value)
                except AttributeError:
                    iface = switch.get_interface_by_port_no(action.port)

                # Check interface to drop packets forwarded to it
                if iface:
                    port_config = v0x01.common.phy_port.PortConfig
                    iface.config = port_config.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 import v0x01
6	1
7	1	from kytos.core import KytosEvent, KytosNApp, log, rest
8	1	from kytos.core.helpers import listen_to
9		from napps.kytos.of_core.flow import FlowFactory
10	1
11	1	from .exceptions import InvalidCommandError
12		from .settings import FLOWS_DICT_MAX_SIZE
13
14	1
15		class Main(KytosNApp):
16		"""Main class to be used by Kytos controller."""
17	1
18		def setup(self):
19		"""Replace the 'init' method for the KytosApp subclass.
20
21		The setup method is automatically called by the run method.
22		Users shouldn't call this method directly.
23	1	"""
24	1	log.debug("flow-manager starting")
25	1	self._flow_mods_sent = OrderedDict()
26		self._flow_mods_sent_max_size = FLOWS_DICT_MAX_SIZE
27	1
28		def execute(self):
29		"""Run once on NApp 'start' or in a loop.
30
31		The execute method is called by the run method of KytosNApp class.
32		Users shouldn't call this method directly.
33		"""
34	1
35		def shutdown(self):
36		"""Shutdown routine of the NApp."""
37		log.debug("flow-manager stopping")
38	1
39	1	@rest('v2/flows')
40	1	@rest('v2/flows/<dpid>')
41		def list(self, dpid=None):
42		"""Retrieve all flows from a switch identified by dpid.
43
44		If no dpid is specified, return all flows from all switches.
45	1	"""
46	1	if dpid is None:
47		switches = self.controller.switches.values()
48	1	else:
49		switches = [self.controller.get_switch_by_dpid(dpid)]
50	1
51		switch_flows = {}
52	1
53	1	for switch in switches:
54	1	flows_dict = [flow.as_dict() for flow in switch.flows]
55		switch_flows[switch.dpid] = {'flows': flows_dict}
56	1
57		return jsonify(switch_flows)
58	1
59	1	@rest('v2/flows', methods=['POST'])
60	1	@rest('v2/flows/<dpid>', methods=['POST'])
61		def add(self, dpid=None):
62		"""Install new flows in the switch identified by dpid.
63
64		If no dpid is specified, install flows in all switches.
65	1	"""
66		return self._send_flow_mods_from_request(dpid, "add")
67	1
68	1	@rest('v2/delete', methods=['POST'])
69	1	@rest('v2/delete/<dpid>', methods=['POST'])
70	1	@rest('v2/flows', methods=['DELETE'])
71	1	@rest('v2/flows/<dpid>', methods=['DELETE'])
72		def delete(self, dpid=None):
73		"""Delete existing flows in the switch identified by dpid.
74
75		If no dpid is specified, delete flows from all switches.
76	1	"""
77		return self._send_flow_mods_from_request(dpid, "delete")
78	1
79		def _get_all_switches_enabled(self):
80	1	"""Get a list of all switches enabled."""
81	1	switches = self.controller.switches.values()
82		return [switch for switch in switches if switch.is_enabled()]
83	1
84		def _send_flow_mods_from_request(self, dpid, command):
85	1	"""Install FlowsMods from request."""
86		flows_dict = request.get_json()
87	1
88	1	if flows_dict is None:
89		return jsonify({"response": 'flows dict is none.'}), 404
90	1
91	1	if dpid:
92	1	switch = self.controller.get_switch_by_dpid(dpid)
93	1	if not switch:
94	1	return jsonify({"response": 'dpid not found.'}), 404
95	1	elif switch.is_enabled() is False:
96		return jsonify({"response": 'switch is disabled.'}), 404
97	1	else:
98		self._install_flows(command, flows_dict, [switch])
99	1	else:
100		self._install_flows(command, flows_dict,
101		self._get_all_switches_enabled())
102	1
103		return jsonify({"response": "FlowMod Messages Sent"})
104	1
105		def _install_flows(self, command, flows_dict, switches=[]):
106		"""Execute all procedures to install flows in the switches.
107
108		Args:
109		command: Flow command to be installed
110		flows_dict: Dictionary with flows to be installed in the switches.
111		switches: A list of switches
112	1	"""
113	1	for switch in switches:
114	1	serializer = FlowFactory.get_class(switch)
115	1	flows = flows_dict.get('flows', [])
116	1	for flow_dict in flows:
117	1	flow = serializer.from_dict(flow_dict, switch)
118		if command == "delete":
119	1	flow_mod = flow.as_of_delete_flow_mod()
120	1	elif command == "add":
121		flow_mod = flow.as_of_add_flow_mod()
122		else:
123	1	raise InvalidCommandError
124	1	self._send_flow_mod(flow.switch, flow_mod)
125		self._add_flow_mod_sent(flow_mod.header.xid, flow, command)
126	1
127		self._send_napp_event(switch, flow, command)
128	1
129		def _add_flow_mod_sent(self, xid, flow, command):
130	1	"""Add the flow mod to the list of flow mods sent."""
131		if len(self._flow_mods_sent) >= self._flow_mods_sent_max_size:
132	1	self._flow_mods_sent.popitem(last=False)
133		self._flow_mods_sent[xid] = (flow, command)
134	1
135	1	def _send_flow_mod(self, switch, flow_mod):
136		event_name = 'kytos/flow_manager.messages.out.ofpt_flow_mod'
137	1
138		content = {'destination': switch.connection,
139		'message': flow_mod}
140	1
141	1	event = KytosEvent(name=event_name, content=content)
142		self.controller.buffers.msg_out.put(event)
143	1
144		def _send_napp_event(self, switch, flow, command, **kwargs):
145	1	"""Send an Event to other apps informing about a FlowMod."""
146	1	if command == 'add':
147	1	name = 'kytos/flow_manager.flow.added'
148	1	elif command == 'delete':
149	1	name = 'kytos/flow_manager.flow.removed'
150	1	elif command == 'error':
151		name = 'kytos/flow_manager.flow.error'
152		else:
153	1	raise InvalidCommandError
154		content = {'datapath': switch,
155	1	'flow': flow}
156	1	content.update(kwargs)
157	1	event_app = KytosEvent(name, content)
158		self.controller.buffers.app.put(event_app)
159	1
160		@listen_to('..of_core..ofpt_error')
161		def handle_errors(self, event):
162		"""Receive OpenFlow error and send a event.
163
164		The event is sent only if the error is related to a request made
165		by flow_manager.
166	1	"""
167	1	message = event.content["message"]
168	1
169	1	connection = event.source
170	1	switch = connection.switch
171
172		xid = message.header.xid.value
173		error_type = message.error_type
174	1	error_code = message.code
175		error_data = message.data.pack()
176
177		# Get the packet responsible for the error
178		error_packet = connection.protocol.unpack(error_data)
179
180		error = v0x01.asynchronous.error_msg.BadActionCode.OFPBAC_BAD_OUT_PORT
181		if message.code == error:
182		for action in error_packet.actions:
183		try:
184		iface = switch.get_interface_by_port_no(action.port.value)
185		except AttributeError:
186		iface = switch.get_interface_by_port_no(action.port)
187
188		# Check interface to drop packets forwarded to it
189		if iface:
190		port_config = v0x01.common.phy_port.PortConfig
191		iface.config = port_config.OFPPC_NO_FWD
192		try:
193		flow, error_command = self._flow_mods_sent[xid]
194		except KeyError:
195		pass
196		else:
197		self._send_napp_event(flow.switch, flow, 'error',
198		error_command=error_command,
199		error_type=error_type, error_code=error_code)
200

kytos / flow_manager

Pull Request — master (#92)

build.main.Main.handle_errors() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like