build.main.Main._store_changed_flows() - Code Metrics - Inspection of "Add consistency" - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#96)

by Jose

created 2020-12-14 20:02 UTC

build.main.Main._store_changed_flows() B

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	52
Code Lines	31

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	24
CRAP Score	6.0493

Importance

Changes

Metric	Value
cc	6
eloc	31
nop	4
dl	0
loc	52
ccs	24
cts	27
cp	0.8889
crap	6.0493
rs	8.2026
c	0
b	0
f	0

How to fix Long Method

"""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, CONSISTENCY_INTERVAL


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

    def consistency_check(self):
        """Check the consistency of flows in each switch."""
        switches = self.controller.switches.values()

        for switch in switches:
            if switch.dpid in self.stored_flows:
                self.consistency_in_switch(switch)
            else:
                continue

    def consistency_in_switch(self, switch):
        """Check consistency for a specific switch."""
        dpid = switch.dpid

        # Flows installed in switch
        switch_flows = [flow.as_dict() for flow in switch.flows]

        # Flows stored in storehouse
        stored_flows = self.stored_flows[dpid]['flow_list']


        for stored_flow in stored_flows:
            command = stored_flow['command']

            # flow to send to switch and resolve the inconsistency
            flow = {'flows': [stored_flow['flow']]}

            if not self._is_flow_in_switch(stored_flow['flow'], switch_flows):
                if command == 'add':
                    msg = f'Flow forwarded to switch {dpid} to be installed.'
                    self._resolve_inconsistency(command, flow, [switch], msg)
            elif command == 'delete':
                msg = f'Flow forwarded to switch {dpid} to be deleted.'
                self._resolve_inconsistency(command, flow, [switch], msg)

        for installed_flow in switch_flows:
            if not self._is_flow_in_storehouse(installed_flow, stored_flows) :
                flow = {'flows': [installed_flow]}
                command = 'delete'
                msg = f'flow nao encontrado no storehouse. Flow forwarded to switch {dpid} to be deleted.'
                self._resolve_inconsistency(command, flow, [switch], msg)

    def _resolve_inconsistency(self, command, flow, switch, msg):
        """Forward flow to be installed resolving the inconsistency."""
        log.info('A problem with consistency were dectected.')
        self._install_flows(command, flow, switch)
        log.info(msg)

    def _is_flow_in_switch(self, flow, switch_flows):
        """Check if the flow is installed in switch."""
        for switch_flow in switch_flows:
            if self._is_equal_flows(flow, switch_flow):
                return True
        return False

    def _is_flow_in_storehouse(self, flow, switch_flows):
        """Check if the flow is stored in storehouse."""
        for switch_flow in switch_flows:
            if self._is_equal_flows(flow, switch_flow['flow']):
                return True
        return False


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

kytos / flow_manager

Pull Request — master (#96)

build.main.Main._store_changed_flows() B

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like