build.main.Main.execute() - Code Metrics - Inspection of "Listen to OpenFlow errors sent by of_core." - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#74)

by Antonio

created 2019-11-07 15:44 UTC

build.main.Main.execute() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	7
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.125

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	7
ccs	1
cts	2
cp	0.5
rs	10
c	0
b	0
f	0
cc	1
nop	1
crap	1.125

"""kytos/flow_manager NApp installs, lists and deletes switch flows."""
from collections import OrderedDict
from flask import jsonify, request
from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to

from napps.kytos.of_core.v0x01.flow import Flow as Flow10
from napps.kytos.of_core.v0x04.flow import Flow as Flow13

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

    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'])
    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 = self._get_flow_serializer(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()
                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):
        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):
        """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'
        content = {'datapath': switch,
                   'flow': flow}
        event_app = KytosEvent(name, content)

        self.controller.buffers.app.put(event_app)

    @staticmethod
    def _get_flow_serializer(switch):
        """Return the serializer with for the switch OF protocol version."""
        version = switch.connection.protocol.version
        return Flow10 if version == 0x01 else Flow13

    @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.
        """
        xid = event.content["message"].header.xid.value
        try:
            flow = self.flow_mods_sent[xid]
            self._send_napp_event(flow.switch, flow, 'error')
        except KeyError:
            pass


1		"""kytos/flow_manager NApp installs, lists and deletes switch flows."""
2	1	from collections import OrderedDict
3	1	from flask import jsonify, request
4	1	from kytos.core import KytosEvent, KytosNApp, log, rest
5	1	from kytos.core.helpers import listen_to
6
7	1	from napps.kytos.of_core.v0x01.flow import Flow as Flow10
8	1	from napps.kytos.of_core.v0x04.flow import Flow as Flow13
9
10	1	from .settings import FLOWS_DICT_MAX_SIZE
11
12
13	1	class Main(KytosNApp):
14		"""Main class to be used by Kytos controller."""
15
16	1	def setup(self):
17		"""Replace the 'init' method for the KytosApp subclass.
18
19		The setup method is automatically called by the run method.
20		Users shouldn't call this method directly.
21		"""
22	1	log.debug("flow-manager starting")
23	1	self._flow_mods_sent = OrderedDict()
24	1	self._flow_mods_sent_max_size = FLOWS_DICT_MAX_SIZE
25
26	1	def execute(self):
27		"""Run once on NApp 'start' or in a loop.
28
29		The execute method is called by the run method of KytosNApp class.
30		Users shouldn't call this method directly.
31		"""
32		pass
33
34	1	def shutdown(self):
35		"""Shutdown routine of the NApp."""
36		log.debug("flow-manager stopping")
37
38	1	@rest('v2/flows')
39	1	@rest('v2/flows/<dpid>')
40	1	def list(self, dpid=None):
41		"""Retrieve all flows from a switch identified by dpid.
42
43		If no dpid is specified, return all flows from all switches.
44		"""
45		if dpid is None:
46		switches = self.controller.switches.values()
47		else:
48		switches = [self.controller.get_switch_by_dpid(dpid)]
49
50		switch_flows = {}
51
52		for switch in switches:
53		flows_dict = [flow.as_dict() for flow in switch.flows]
54		switch_flows[switch.dpid] = {'flows': flows_dict}
55
56		return jsonify(switch_flows)
57
58	1	@rest('v2/flows', methods=['POST'])
59	1	@rest('v2/flows/<dpid>', methods=['POST'])
60	1	def add(self, dpid=None):
61		"""Install new flows in the switch identified by dpid.
62
63		If no dpid is specified, install flows in all switches.
64		"""
65		return self._send_flow_mods_from_request(dpid, "add")
66
67	1	@rest('v2/delete', methods=['POST'])
68	1	@rest('v2/delete/<dpid>', methods=['POST'])
69	1	def delete(self, dpid=None):
70		"""Delete existing flows in the switch identified by dpid.
71
72		If no dpid is specified, delete flows from all switches.
73		"""
74		return self._send_flow_mods_from_request(dpid, "delete")
75
76	1	def _get_all_switches_enabled(self):
77		"""Get a list of all switches enabled."""
78		switches = self.controller.switches.values()
79		return [switch for switch in switches if switch.is_enabled()]
80
81	1	def _send_flow_mods_from_request(self, dpid, command):
82		"""Install FlowsMods from request."""
83		flows_dict = request.get_json()
84
85		if flows_dict is None:
86		return jsonify({"response": 'flows dict is none.'}), 404
87
88		if dpid:
89		switch = self.controller.get_switch_by_dpid(dpid)
90		if not switch:
91		return jsonify({"response": 'dpid not found.'}), 404
92		elif switch.is_enabled() is False:
93		return jsonify({"response": 'switch is disabled.'}), 404
94		else:
95		self._install_flows(command, flows_dict, [switch])
96		else:
97		self._install_flows(command, flows_dict,
98		self._get_all_switches_enabled())
99
100		return jsonify({"response": "FlowMod Messages Sent"})
101
102	1	def _install_flows(self, command, flows_dict, switches=[]):
103		"""Execute all procedures to install flows in the switches.
104
105		Args:
106		command: Flow command to be installed
107		flows_dict: Dictionary with flows to be installed in the switches.
108		switches: A list of switches
109		"""
110		for switch in switches:
111		serializer = self._get_flow_serializer(switch)
112		flows = flows_dict.get('flows', [])
113		for flow_dict in flows:
114		flow = serializer.from_dict(flow_dict, switch)
115		if command == "delete":
116		flow_mod = flow.as_of_delete_flow_mod()
117		elif command == "add":
118		flow_mod = flow.as_of_add_flow_mod()
119		self._send_flow_mod(flow.switch, flow_mod)
		0 ignored issues – show introduced 2019-03-14 01:39 UTC by Report Bug Copy Issue Report The variable `flow_mod` does not seem to be defined for all execution paths. Loading history...
120		self._add_flow_mod_sent(flow_mod.header.xid, flow)
121
122		self._send_napp_event(switch, flow, command)
123
124	1	def _add_flow_mod_sent(self, xid, flow):
125	1	if len(self._flow_mods_sent) >= self._flow_mods_sent_max_size:
126	1	self._flow_mods_sent.popitem(last=False)
127	1	self._flow_mods_sent[xid] = flow
128
129	1	def _send_flow_mod(self, switch, flow_mod):
130		event_name = 'kytos/flow_manager.messages.out.ofpt_flow_mod'
131
132		content = {'destination': switch.connection,
133		'message': flow_mod}
134
135		event = KytosEvent(name=event_name, content=content)
136		self.controller.buffers.msg_out.put(event)
137
138	1	def _send_napp_event(self, switch, flow, command):
139		"""Send an Event to other apps informing about a FlowMod."""
140		if command == 'add':
141		name = 'kytos/flow_manager.flow.added'
142		elif command == 'delete':
143		name = 'kytos/flow_manager.flow.removed'
144		elif command == 'error':
145		name = 'kytos/flow_manager.flow.error'
146		content = {'datapath': switch,
147		'flow': flow}
148		event_app = KytosEvent(name, content)
		0 ignored issues – show introduced 2019-03-14 01:39 UTC by Report Bug Copy Issue Report The variable `name` does not seem to be defined for all execution paths. Loading history...
149		self.controller.buffers.app.put(event_app)
150
151	1	@staticmethod
152		def _get_flow_serializer(switch):
153		"""Return the serializer with for the switch OF protocol version."""
154		version = switch.connection.protocol.version
155		return Flow10 if version == 0x01 else Flow13
156
157	1	@listen_to('..of_core..ofpt_error')
158		def handle_errors(self, event):
159		"""Receive OpenFlow error and send a event.
160
161		The event is sent only if the error is related to a request made
162		by flow_manager.
163		"""
164		xid = event.content["message"].header.xid.value
165		try:
166		flow = self.flow_mods_sent[xid]
167		self._send_napp_event(flow.switch, flow, 'error')
168		except KeyError:
169		pass
170

kytos / flow_manager

Pull Request — master (#74)

build.main.Main.execute() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like