build.main.Main._send_flow_mods_from_request() - Code Metrics - Inspection of "Added flow persistence" - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#90)

by Carlos

created 2020-11-12 15:17 UTC

build.main.Main._send_flow_mods_from_request() B

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	20
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	14
CRAP Score	6

Importance

Changes

Metric	Value
cc	6
eloc	15
nop	4
dl	0
loc	20
ccs	14
cts	14
cp	1
crap	6
rs	8.6666
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 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 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

        # Format of stored data
        # {"flow_persistence":{
        #     "dpid_str":{
        #         "flow_list":[
        #             {"command":<add|delete>}
        #             "flow":{flow_dict},
        #         ]
        #     }
        # }}
        # client to load and save data in storehouse
        self.storehouse = StoreHouse(self.controller)
        self.stored_flows = {}
        self.resent_flows = set()

    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()

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

    @listen_to('kytos/topology.port.created')
    def resend_stored_flows(self, event):
        """Resend stored Flows."""
        dpid = str(event.content['switch'])
        switch = self.controller.get_switch_by_dpid(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}')

    # 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.')

    @staticmethod
    def _generate_match_fields(flows):
        """Generate flow match fields."""
        match_fields = {}
        flow = flows['flow']
        for field in flow:
            if 'priority' in field:
                match_fields['priority'] = flow['priority']
            if 'cookie' in field:
                match_fields['cookie'] = flow['cookie']
            if 'match' in field:
                for m_field, value in flow['match'].items():
                    match_fields[m_field] = value
        return match_fields

    def _is_equal_flows(self, flow_1, flow_2):
        """Check if two flows are equal."""
        flow_1_match_fields = self._generate_match_fields(flow_1)
        flow_2_match_fields = self._generate_match_fields(flow_2)
        if flow_1_match_fields == flow_2_match_fields:
            return True
        return False

    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

        new_flow = {}
        flow_list = []
        new_flow['command'] = command
        new_flow['flow'] = flow

        if switch.id not in stored_flows_box:
            # Switch not stored, add to box.
            flow_list.append(new_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:
                if self._is_equal_flows(new_flow, stored_flow):
                    if stored_flow['command'] == new_flow['command']:
                        log.debug('Data already stored.')
                        return
                    # Command conflict. Remove the old instruction.
                    # There is a stored instruction to install the flow,
                    # but the new instruction is to remove it.
                    # In this case, remove the old instruction
                    # and store the new.
                    stored_flow['command'] = new_flow.get('command')
                    stored_flows.remove(stored_flow)
                    break

            stored_flows.append(new_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:
                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.
        """
        xid = event.content["message"].header.xid.value
        error_type = event.content["message"].error_type
        error_code = event.content["message"].code
        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
6	1	from kytos.core import KytosEvent, KytosNApp, log, rest
7	1	from kytos.core.helpers import listen_to
8	1	from napps.kytos.flow_manager.storehouse import StoreHouse
9	1	from napps.kytos.of_core.flow import FlowFactory
10
11	1	from .exceptions import InvalidCommandError
12	1	from .settings import FLOWS_DICT_MAX_SIZE
13
14
15	1	class Main(KytosNApp):
16		"""Main class to be used by Kytos controller."""
17
18	1	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		"""
24	1	log.debug("flow-manager starting")
25	1	self._flow_mods_sent = OrderedDict()
26	1	self._flow_mods_sent_max_size = FLOWS_DICT_MAX_SIZE
27
28		# Format of stored data
29		# {"flow_persistence":{
30		# "dpid_str":{
31		# "flow_list":[
32		# {"command":<add\|delete>}
33		# "flow":{flow_dict},
34		# ]
35		# }
36		# }}
37		# client to load and save data in storehouse
38	1	self.storehouse = StoreHouse(self.controller)
39	1	self.stored_flows = {}
40	1	self.resent_flows = set()
41
42	1	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	1	def shutdown(self):
51		"""Shutdown routine of the NApp."""
52		log.debug("flow-manager stopping")
53
54	1	@listen_to('kytos/topology.port.created')
55		def resend_stored_flows(self, event):
56		"""Resend stored Flows."""
57	1	dpid = str(event.content['switch'])
58	1	switch = self.controller.get_switch_by_dpid(dpid)
59		# This can be a problem because this code is running a thread
60	1	if dpid in self.resent_flows:
61		log.info(f'Flow already resended to Switch {dpid}')
62		return
63	1	if dpid in self.stored_flows:
64	1	flow_list = self.stored_flows[dpid]['flow_list']
65	1	for flow in flow_list:
66	1	command = flow['command']
67	1	flows_dict = {"flows": [flow['flow']]}
68	1	self._install_flows(command, flows_dict, [switch])
69	1	self.resent_flows.add(dpid)
70	1	log.info(f'Flows resent to Switch {dpid}')
71
72		# pylint: disable=attribute-defined-outside-init
73	1	def _load_flows(self):
74		"""Load stored flows."""
75	1	try:
76	1	data = self.storehouse.get_data()['flow_persistence']
77	1	if 'id' in data:
78		del data['id']
79	1	self.stored_flows = data
80
81		except KeyError as error:
82		log.debug(f'There are no flows to load: {error}')
83		else:
84	1	log.info('Flows loaded.')
85
86	1	@staticmethod
87		def _generate_match_fields(flows):
88		"""Generate flow match fields."""
89	1	match_fields = {}
90	1	flow = flows['flow']
91	1	for field in flow:
92	1	if 'priority' in field:
93	1	match_fields['priority'] = flow['priority']
94	1	if 'cookie' in field:
95	1	match_fields['cookie'] = flow['cookie']
96	1	if 'match' in field:
97	1	for m_field, value in flow['match'].items():
98	1	match_fields[m_field] = value
99	1	return match_fields
100
101	1	def _is_equal_flows(self, flow_1, flow_2):
102		"""Check if two flows are equal."""
103	1	flow_1_match_fields = self._generate_match_fields(flow_1)
104	1	flow_2_match_fields = self._generate_match_fields(flow_2)
105	1	if flow_1_match_fields == flow_2_match_fields:
106		return True
107	1	return False
108
109	1	def _store_changed_flows(self, command, flow, switch):
110		"""Store changed flows.
111
112		Args:
113		command: Flow command to be installed
114		flow: Flows to be stored
115		switch: Switch target
116		"""
117	1	stored_flows_box = self.stored_flows.copy()
118		# if the flow has a destination dpid it can be stored.
119	1	if not switch:
120		log.info('The Flow cannot be stored, the destination switch '
121		f'have not been specified: {switch}')
122		return
123
124	1	new_flow = {}
125	1	flow_list = []
126	1	new_flow['command'] = command
127	1	new_flow['flow'] = flow
128
129	1	if switch.id not in stored_flows_box:
130		# Switch not stored, add to box.
131	1	flow_list.append(new_flow)
132	1	stored_flows_box[switch.id] = {"flow_list": flow_list}
133		else:
134	1	stored_flows = stored_flows_box[switch.id].get('flow_list', [])
135		# Check if flow already stored
136	1	for stored_flow in stored_flows:
137	1	if self._is_equal_flows(new_flow, stored_flow):
138		if stored_flow['command'] == new_flow['command']:
139		log.debug('Data already stored.')
140		return
141		# Command conflict. Remove the old instruction.
142		# There is a stored instruction to install the flow,
143		# but the new instruction is to remove it.
144		# In this case, remove the old instruction
145		# and store the new.
146		stored_flow['command'] = new_flow.get('command')
147		stored_flows.remove(stored_flow)
148		break
149
150	1	stored_flows.append(new_flow)
151	1	stored_flows_box[switch.id]['flow_list'] = stored_flows
152
153	1	stored_flows_box['id'] = 'flow_persistence'
154	1	self.storehouse.save_flow(stored_flows_box)
155	1	del stored_flows_box['id']
156	1	self.stored_flows = stored_flows_box.copy()
157
158	1	@rest('v2/flows')
159	1	@rest('v2/flows/<dpid>')
160	1	def list(self, dpid=None):
161		"""Retrieve all flows from a switch identified by dpid.
162
163		If no dpid is specified, return all flows from all switches.
164		"""
165	1	if dpid is None:
166	1	switches = self.controller.switches.values()
167		else:
168	1	switches = [self.controller.get_switch_by_dpid(dpid)]
169
170	1	switch_flows = {}
171
172	1	for switch in switches:
173	1	flows_dict = [flow.as_dict() for flow in switch.flows]
174	1	switch_flows[switch.dpid] = {'flows': flows_dict}
175
176	1	return jsonify(switch_flows)
177
178	1	@rest('v2/flows', methods=['POST'])
179	1	@rest('v2/flows/<dpid>', methods=['POST'])
180	1	def add(self, dpid=None):
181		"""Install new flows in the switch identified by dpid.
182
183		If no dpid is specified, install flows in all switches.
184		"""
185	1	return self._send_flow_mods_from_request(dpid, "add")
186
187	1	@rest('v2/delete', methods=['POST'])
188	1	@rest('v2/delete/<dpid>', methods=['POST'])
189	1	@rest('v2/flows', methods=['DELETE'])
190	1	@rest('v2/flows/<dpid>', methods=['DELETE'])
191	1	def delete(self, dpid=None):
192		"""Delete existing flows in the switch identified by dpid.
193
194		If no dpid is specified, delete flows from all switches.
195		"""
196	1	return self._send_flow_mods_from_request(dpid, "delete")
197
198	1	def _get_all_switches_enabled(self):
199		"""Get a list of all switches enabled."""
200	1	switches = self.controller.switches.values()
201	1	return [switch for switch in switches if switch.is_enabled()]
202
203	1	def _send_flow_mods_from_request(self, dpid, command, flows_dict=None):
204		"""Install FlowsMods from request."""
205	1	if flows_dict is None:
206	1	flows_dict = request.get_json()
207	1	if flows_dict is None:
208	1	return jsonify({"response": 'flows dict is none.'}), 404
209
210	1	if dpid:
211	1	switch = self.controller.get_switch_by_dpid(dpid)
212	1	if not switch:
213	1	return jsonify({"response": 'dpid not found.'}), 404
214	1	elif switch.is_enabled() is False:
215	1	return jsonify({"response": 'switch is disabled.'}), 404
216		else:
217	1	self._install_flows(command, flows_dict, [switch])
218		else:
219	1	self._install_flows(command, flows_dict,
220		self._get_all_switches_enabled())
221
222	1	return jsonify({"response": "FlowMod Messages Sent"})
223
224	1	def _install_flows(self, command, flows_dict, switches=[]):
225		"""Execute all procedures to install flows in the switches.
226
227		Args:
228		command: Flow command to be installed
229		flows_dict: Dictionary with flows to be installed in the switches.
230		switches: A list of switches
231		"""
232	1	for switch in switches:
233	1	serializer = FlowFactory.get_class(switch)
234	1	flows = flows_dict.get('flows', [])
235	1	for flow_dict in flows:
236	1	flow = serializer.from_dict(flow_dict, switch)
237	1	if command == "delete":
238		flow_mod = flow.as_of_delete_flow_mod()
239	1	elif command == "add":
240	1	flow_mod = flow.as_of_add_flow_mod()
241		else:
242		raise InvalidCommandError
243	1	self._send_flow_mod(flow.switch, flow_mod)
244	1	self._add_flow_mod_sent(flow_mod.header.xid, flow, command)
245
246	1	self._send_napp_event(switch, flow, command)
247	1	self._store_changed_flows(command, flow_dict, switch)
248
249	1	def _add_flow_mod_sent(self, xid, flow, command):
250		"""Add the flow mod to the list of flow mods sent."""
251	1	if len(self._flow_mods_sent) >= self._flow_mods_sent_max_size:
252		self._flow_mods_sent.popitem(last=False)
253	1	self._flow_mods_sent[xid] = (flow, command)
254
255	1	def _send_flow_mod(self, switch, flow_mod):
256	1	event_name = 'kytos/flow_manager.messages.out.ofpt_flow_mod'
257
258	1	content = {'destination': switch.connection,
259		'message': flow_mod}
260
261	1	event = KytosEvent(name=event_name, content=content)
262	1	self.controller.buffers.msg_out.put(event)
263
264	1	def _send_napp_event(self, switch, flow, command, **kwargs):
265		"""Send an Event to other apps informing about a FlowMod."""
266	1	if command == 'add':
267	1	name = 'kytos/flow_manager.flow.added'
268	1	elif command == 'delete':
269	1	name = 'kytos/flow_manager.flow.removed'
270	1	elif command == 'error':
271	1	name = 'kytos/flow_manager.flow.error'
272		else:
273		raise InvalidCommandError
274	1	content = {'datapath': switch,
275		'flow': flow}
276	1	content.update(kwargs)
277	1	event_app = KytosEvent(name, content)
278	1	self.controller.buffers.app.put(event_app)
279
280	1	@listen_to('..of_core..ofpt_error')
281		def handle_errors(self, event):
282		"""Receive OpenFlow error and send a event.
283
284		The event is sent only if the error is related to a request made
285		by flow_manager.
286		"""
287	1	xid = event.content["message"].header.xid.value
288	1	error_type = event.content["message"].error_type
289	1	error_code = event.content["message"].code
290	1	try:
291	1	flow, error_command = self._flow_mods_sent[xid]
292		except KeyError:
293		pass
294		else:
295	1	self._send_napp_event(flow.switch, flow, 'error',
296		error_command=error_command,
297		error_type=error_type, error_code=error_code)
298

kytos / flow_manager

Pull Request — master (#90)

build.main.Main._send_flow_mods_from_request() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like