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

Test Failed

Pull Request — master (#90)

unknown

created 2020-10-06 15:17 UTC

build.main.Main._load_flows() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	13
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	8
CRAP Score	2

Importance

Changes

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

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

        # object to save and load flows
        self.storehouse = StoreHouseClient(self.controller)
        self.stored_flows = {}

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

    # pylint: disable=attribute-defined-outside-init
    def _load_flows(self):
        """Load stored flows."""
        try:
            data = self.storehouse.get_data()['flow_persistence']
            del data['id']
            self.stored_flows = data

        except (KeyError) as error:
            log.info(f'Error:{error}')
            log.info('Not has persisted flows to load.')
            # self.flows_stored = {}
            return
        log.info('Flow box loaded.')

    def _flows_change(self, command, flows, switches):
        """Flows change, store Flows. Persist all flows."""
        switch_flows = {}
        store_box_updated = self.stored_flows.copy()
        for switch in switches:
            flow_box = {}
            flow_list = []
            # match fields to an stored flow
            match_storage_fields = {}
            match_storage_fields['command'] = command
            for i in flows['flows']:
                if 'priority' in i:
                    match_storage_fields['priority'] = i['priority']
                if 'cookie' in i:
                    match_storage_fields['cookie'] = i['cookie']
                if 'match' in i:
                    # dict comprehension
                    # match_storage_fields = {key:value for key,value in
                    #                         i['match'].items()}
                    for key, value in i['match'].items():
                        match_storage_fields[key] = value
            flow_box['match_storage'] = match_storage_fields

            flow_box['data'] = flows
            flow_list.append(flow_box)
            switch_flows[switch.id] = {"flow_list": flow_list}

        # TODO: simplify, too-many-branches
        for key, value in switch_flows.items():
            flag_match_stored_flow = 0

            # Switch have stored flow, append new flow to flow_list.
            if key in self.stored_flows:
                stored_flows = self.stored_flows[key]['flow_list']
                new_flows = value['flow_list']

                # Check if flow already stored
                for new in new_flows:
                    for old in stored_flows:
                        # if already stored, set flag and break
                        if new['match_storage'] == old['match_storage']:
                            flag_match_stored_flow = 1
                            break

                # if flow not stored append to flow_list
                if flag_match_stored_flow == 0:
                    temp_list = store_box_updated[key]['flow_list']
                    temp_list.append(new_flows)
                    store_box_updated[key]['flow_list'] = temp_list

            else:
                # switch not stored, add to box.
                store_box_updated[key] = value

        # TODO: need check if has saved if sucess
        self.flows_stored = store_box_updated.copy()
        store_box_updated['id'] = 'flow_persistence'

        # print(f"Box To storage .. {store_box_updated}")
        # print(f"Len BOX {len(store_box_updated)}")
        self.storehouse.save_flow(store_box_updated)

    @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}

        # self._flows_change()
        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)
        self._flows_change(command, flows_dict, switches)

    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.
        """
        xid = event.content["message"].header.xid.value
        error_type = event.content["message"].error_type
        error_code = event.content["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 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 StoreHouseClient
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		# object to save and load flows
29		self.storehouse = StoreHouseClient(self.controller)
30		self.stored_flows = {}
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	self._load_flows()
39	1
40	1	def shutdown(self):
41		"""Shutdown routine of the NApp."""
42		log.debug("flow-manager stopping")
43
44		# pylint: disable=attribute-defined-outside-init
45	1	def _load_flows(self):
46	1	"""Load stored flows."""
47		try:
48	1	data = self.storehouse.get_data()['flow_persistence']
49		del data['id']
50	1	self.stored_flows = data
51
52	1	except (KeyError) as error:
53	1	log.info(f'Error:{error}')
54	1	log.info('Not has persisted flows to load.')
55		# self.flows_stored = {}
56	1	return
57		log.info('Flow box loaded.')
58	1
59	1	def _flows_change(self, command, flows, switches):
60	1	"""Flows change, store Flows. Persist all flows."""
61		switch_flows = {}
62		store_box_updated = self.stored_flows.copy()
63		for switch in switches:
64		flow_box = {}
65	1	flow_list = []
66		# match fields to an stored flow
67	1	match_storage_fields = {}
68	1	match_storage_fields['command'] = command
69	1	for i in flows['flows']:
70	1	if 'priority' in i:
71	1	match_storage_fields['priority'] = i['priority']
72		if 'cookie' in i:
73		match_storage_fields['cookie'] = i['cookie']
74		if 'match' in i:
75		# dict comprehension
76	1	# match_storage_fields = {key:value for key,value in
77		# i['match'].items()}
78	1	for key, value in i['match'].items():
79		match_storage_fields[key] = value
80	1	flow_box['match_storage'] = match_storage_fields
81	1
82		flow_box['data'] = flows
83	1	flow_list.append(flow_box)
84		switch_flows[switch.id] = {"flow_list": flow_list}
85	1
86		# TODO: simplify, too-many-branches
87	1	for key, value in switch_flows.items():
88	1	flag_match_stored_flow = 0
89
90	1	# Switch have stored flow, append new flow to flow_list.
91	1	if key in self.stored_flows:
92	1	stored_flows = self.stored_flows[key]['flow_list']
93	1	new_flows = value['flow_list']
94	1
95	1	# Check if flow already stored
96		for new in new_flows:
97	1	for old in stored_flows:
98		# if already stored, set flag and break
99	1	if new['match_storage'] == old['match_storage']:
100		flag_match_stored_flow = 1
101		break
102	1
103		# if flow not stored append to flow_list
104	1	if flag_match_stored_flow == 0:
105		temp_list = store_box_updated[key]['flow_list']
106		temp_list.append(new_flows)
107		store_box_updated[key]['flow_list'] = temp_list
108
109		else:
110		# switch not stored, add to box.
111		store_box_updated[key] = value
112	1
113	1	# TODO: need check if has saved if sucess
114	1	self.flows_stored = store_box_updated.copy()
115	1	store_box_updated['id'] = 'flow_persistence'
116	1
117	1	# print(f"Box To storage .. {store_box_updated}")
118		# print(f"Len BOX {len(store_box_updated)}")
119	1	self.storehouse.save_flow(store_box_updated)
120	1
121		@rest('v2/flows')
122		@rest('v2/flows/<dpid>')
123	1	def list(self, dpid=None):
124	1	"""Retrieve all flows from a switch identified by dpid.
125
126	1	If no dpid is specified, return all flows from all switches.
127		"""
128	1	if dpid is None:
129		switches = self.controller.switches.values()
130	1	else:
131		switches = [self.controller.get_switch_by_dpid(dpid)]
132	1
133		switch_flows = {}
134	1
135	1	for switch in switches:
136		flows_dict = [flow.as_dict() for flow in switch.flows]
137	1	switch_flows[switch.dpid] = {'flows': flows_dict}
138
139		# self._flows_change()
140	1	return jsonify(switch_flows)
141	1
142		@rest('v2/flows', methods=['POST'])
143	1	@rest('v2/flows/<dpid>', methods=['POST'])
144		def add(self, dpid=None):
145	1	"""Install new flows in the switch identified by dpid.
146	1
147	1	If no dpid is specified, install flows in all switches.
148	1	"""
149	1	return self._send_flow_mods_from_request(dpid, "add")
150	1
151		@rest('v2/delete', methods=['POST'])
152		@rest('v2/delete/<dpid>', methods=['POST'])
153	1	@rest('v2/flows', methods=['DELETE'])
154		@rest('v2/flows/<dpid>', methods=['DELETE'])
155	1	def delete(self, dpid=None):
156	1	"""Delete existing flows in the switch identified by dpid.
157	1
158		If no dpid is specified, delete flows from all switches.
159	1	"""
160		return self._send_flow_mods_from_request(dpid, "delete")
161
162		def _get_all_switches_enabled(self):
163		"""Get a list of all switches enabled."""
164		switches = self.controller.switches.values()
165		return [switch for switch in switches if switch.is_enabled()]
166	1
167	1	def _send_flow_mods_from_request(self, dpid, command):
168	1	"""Install FlowsMods from request."""
169	1	flows_dict = request.get_json()
170	1
171		if flows_dict is None:
172		return jsonify({"response": 'flows dict is none.'}), 404
173
174	1	if dpid:
175		switch = self.controller.get_switch_by_dpid(dpid)
176		if not switch:
177		return jsonify({"response": 'dpid not found.'}), 404
178		elif switch.is_enabled() is False:
179		return jsonify({"response": 'switch is disabled.'}), 404
180		else:
181		self._install_flows(command, flows_dict, [switch])
182		else:
183		self._install_flows(command, flows_dict,
184		self._get_all_switches_enabled())
185
186		return jsonify({"response": "FlowMod Messages Sent"})
187
188		def _install_flows(self, command, flows_dict, switches=[]):
189		"""Execute all procedures to install flows in the switches.
190
191		Args:
192		command: Flow command to be installed
193		flows_dict: Dictionary with flows to be installed in the switches.
194		switches: A list of switches
195		"""
196		for switch in switches:
197		serializer = FlowFactory.get_class(switch)
198		flows = flows_dict.get('flows', [])
199		for flow_dict in flows:
200		flow = serializer.from_dict(flow_dict, switch)
201		if command == "delete":
202		flow_mod = flow.as_of_delete_flow_mod()
203		elif command == "add":
204		flow_mod = flow.as_of_add_flow_mod()
205		else:
206		raise InvalidCommandError
207		self._send_flow_mod(flow.switch, flow_mod)
208		self._add_flow_mod_sent(flow_mod.header.xid, flow)
209
210		self._send_napp_event(switch, flow, command)
211		self._flows_change(command, flows_dict, switches)
212
213		def _add_flow_mod_sent(self, xid, flow):
214		"""Add the flow mod to the list of flow mods sent."""
215		if len(self._flow_mods_sent) >= self._flow_mods_sent_max_size:
216		self._flow_mods_sent.popitem(last=False)
217		self._flow_mods_sent[xid] = flow
218
219		def _send_flow_mod(self, switch, flow_mod):
220		event_name = 'kytos/flow_manager.messages.out.ofpt_flow_mod'
221
222		content = {'destination': switch.connection,
223		'message': flow_mod}
224
225		event = KytosEvent(name=event_name, content=content)
226		self.controller.buffers.msg_out.put(event)
227
228		def _send_napp_event(self, switch, flow, command, **kwargs):
229		"""Send an Event to other apps informing about a FlowMod."""
230		if command == 'add':
231		name = 'kytos/flow_manager.flow.added'
232		elif command == 'delete':
233		name = 'kytos/flow_manager.flow.removed'
234		elif command == 'error':
235		name = 'kytos/flow_manager.flow.error'
236		else:
237		raise InvalidCommandError
238		content = {'datapath': switch,
239		'flow': flow}
240		content.update(kwargs)
241		event_app = KytosEvent(name, content)
242		self.controller.buffers.app.put(event_app)
243
244		@listen_to('..of_core..ofpt_error')
245		def handle_errors(self, event):
246		"""Receive OpenFlow error and send a event.
247
248		The event is sent only if the error is related to a request made
249		by flow_manager.
250		"""
251		xid = event.content["message"].header.xid.value
252		error_type = event.content["message"].error_type
253		error_code = event.content["message"].code
254		try:
255		flow = self._flow_mods_sent[xid]
256		except KeyError:
257		pass
258		else:
259		self._send_napp_event(flow.switch, flow, 'error',
260		error_type=error_type, error_code=error_code)
261

kytos / flow_manager

Pull Request — master (#90)

build.main.Main._load_flows() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like