build.main.Main.add() - 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 (#88)

by Gleyberson

created 2020-09-17 18:14 UTC

build.main.Main.add() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	8
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	1

Importance

Changes

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

"""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 as BadActionCode01
from pyof.v0x01.common.phy_port import PortConfig as PortConfig01
from pyof.v0x04.asynchronous.error_msg import BadActionCode as BadActionCode04
from pyof.v0x04.common.port import PortConfig as PortConfig04

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)

                self._send_napp_event(switch, flow, command)

    def _add_flow_mod_sent(self, xid, flow):
        """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

    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

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

        if (message.code == BadActionCode01.OFPBAC_BAD_OUT_PORT or
           message.code == BadActionCode04.OFPBAC_BAD_OUT_PORT):

            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:
                    if connection.protocol.version == 0x01:
                        iface.config = PortConfig01.OFPPC_NO_FWD
                    elif connection.protocol.version == 0x04:
                        iface.config = PortConfig04.OFPPC_NO_FWD

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

kytos / flow_manager

Pull Request — master (#88)

build.main.Main.add() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like