build.main.Main.event_flows_install_delete() - Code Metrics - Inspection of "Add support to install or remove flows in the swit..." - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#129)

by Carlos

created 2021-05-25 17:55 UTC

build.main.Main.event_flows_install_delete() B

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	27
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	11
CRAP Score	6.4704

Importance

Changes

Metric	Value
cc	5
eloc	19
nop	2
dl	0
loc	27
rs	8.9833
c	0
b	0
f	0
ccs	11
cts	18
cp	0.6111
crap	6.4704

"""kytos/flow_manager NApp installs, lists and deletes switch flows."""
from collections import OrderedDict
from copy import deepcopy

from flask import jsonify, request
from pyof.foundation.base import UBIntBase
from pyof.v0x01.asynchronous.error_msg import BadActionCode
from pyof.v0x01.common.phy_port import PortConfig
from werkzeug.exceptions import NotFound

from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to
from napps.kytos.flow_manager.match import match_flow
from napps.kytos.flow_manager.storehouse import StoreHouse
from napps.kytos.of_core.flow import FlowFactory

from .exceptions import InvalidCommandError
from .settings import CONSISTENCY_INTERVAL, FLOWS_DICT_MAX_SIZE


def cast_fields(flow_dict):
    """Make casting the match fields from UBInt() to native int ."""
    match = flow_dict['match']
    for field, value in match.items():
        if isinstance(value, UBIntBase):
            match[field] = int(value)
    flow_dict['match'] = match
    return flow_dict


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

    @listen_to('kytos/of_core.handshake.completed')
    def resend_stored_flows(self, event):
        """Resend stored Flows."""
        # if consistency check is enabled, it should take care of this
        if CONSISTENCY_INTERVAL >= 0:
            return
        switch = event.content['switch']
        dpid = str(switch.dpid)
        # This can be a problem because this code is running a thread
        if dpid in self.resent_flows:
            log.debug(f'Flow already resent to the 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}')

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

        for switch in switches:
            # Check if a dpid is a key in 'stored_flows' dictionary
            if switch.is_enabled():
                self.check_storehouse_consistency(switch)

                if switch.dpid in self.stored_flows:
                    self.check_switch_consistency(switch)

    def check_switch_consistency(self, switch):
        """Check consistency of installed flows for a specific switch."""
        dpid = switch.dpid

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

        serializer = FlowFactory.get_class(switch)

        for stored_flow in stored_flows:
            command = stored_flow['command']
            stored_flow_obj = serializer.from_dict(stored_flow['flow'], switch)

            flow = {'flows': [stored_flow['flow']]}

            if stored_flow_obj not in switch.flows:
                if command == 'add':
                    log.info('A consistency problem was detected in '
                             f'switch {dpid}.')
                    self._install_flows(command, flow, [switch])
                    log.info(f'Flow forwarded to switch {dpid} to be '
                             'installed.')
            elif command == 'delete':
                log.info('A consistency problem was detected in '
                         f'switch {dpid}.')
                command = 'delete_strict'
                self._install_flows(command, flow, [switch])
                log.info(f'Flow forwarded to switch {dpid} to be deleted.')

    def check_storehouse_consistency(self, switch):
        """Check consistency of installed flows for a specific switch."""
        dpid = switch.dpid

        for installed_flow in switch.flows:
            if dpid not in self.stored_flows:
                log.info('A consistency problem was detected in '
                         f'switch {dpid}.')
                flow = {'flows': [installed_flow.as_dict()]}
                command = 'delete_strict'
                self._install_flows(command, flow, [switch])
                log.info(f'Flow forwarded to switch {dpid} to be deleted.')
            else:
                serializer = FlowFactory.get_class(switch)
                stored_flows = self.stored_flows[dpid]['flow_list']
                stored_flows_list = [serializer.from_dict(stored_flow['flow'],
                                                          switch)
                                     for stored_flow in stored_flows]

                if installed_flow not in stored_flows_list:
                    log.info('A consistency problem was detected in '
                             f'switch {dpid}.')
                    flow = {'flows': [installed_flow.as_dict()]}
                    command = 'delete_strict'
                    self._install_flows(command, flow, [switch])
                    log.info(f'Flow forwarded to switch {dpid} to be deleted.')

    # 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, FileNotFoundError) 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 = deepcopy(self.stored_flows)
        # 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
        deleted_flows = []

        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)

                version = switch.connection.protocol.version

                if installed_flow['command'] == 'delete':
                    # No strict match
                    if match_flow(flow, version, stored_flow['flow']):
                        deleted_flows.append(stored_flow)

                elif 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')
                    deleted_flows.append(stored_flow)
                    break

            # if installed_flow['command'] != 'delete':
            stored_flows.append(installed_flow)
            for i in deleted_flows:
                stored_flows.remove(i)
            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 = deepcopy(stored_flows_box)

    @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)]

            if not any(switches):
                raise NotFound("Switch not found")

        switch_flows = {}

        for switch in switches:
            flows_dict = [cast_fields(flow.as_dict())
                          for flow in switch.flows]
            switch_flows[switch.dpid] = {'flows': flows_dict}

        return jsonify(switch_flows)

    @listen_to('kytos.flow_manager.flows.(install|delete)')
    def event_flows_install_delete(self, event):
        """Install or delete flows in the switches through events.

        Install or delete Flow of switches identified by dpid.
        """
        try:
            dpid = event.content['dpid']
            flow_dict = event.content['flow_dict']
        except KeyError as error:
            log.error("Error getting fields to install or remove "
                      f"Flows: {error}")
            return

        if event.name == 'kytos.flow_manager.flows.install':
            command = 'add'
        elif event.name == 'kytos.flow_manager.flows.delete':
            command = 'delete'
        else:
            msg = f'Invalid event "{event.name}", should be install|delete'
            raise ValueError(msg)

        switch = self.controller.get_switch_by_dpid(dpid)
        try:
            self._install_flows(command, flow_dict, [switch])
        except InvalidCommandError as error:
            log.error("Error installing or deleting Flow through"
                      f" Kytos Event: {error}")

    @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:
                if command == "delete":
                    self._install_flows(command, flows_dict, [switch])
                else:
                    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 == "delete_strict":
                    flow_mod = flow.as_of_strict_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 in ('delete', 'delete_strict'):
            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

        xid = message.header.xid.value
        error_type = message.error_type
        error_code = message.code
        error_data = message.data.pack()

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

        if message.code == BadActionCode.OFPBAC_BAD_OUT_PORT:
            actions = []
            if hasattr(error_packet, 'actions'):
                # Get actions from the flow mod (OF 1.0)
                actions = error_packet.actions
            else:
                # Get actions from the list of flow mod instructions (OF 1.3)
                for instruction in error_packet.instructions:
                    actions.extend(instruction.actions)

            for action in actions:
                iface = switch.get_interface_by_port_no(action.port)

                # Set interface to drop packets forwarded to it
                if iface:
                    iface.config = PortConfig.OFPPC_NO_FWD

        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	1	from copy import deepcopy
4
5	1	from flask import jsonify, request
6	1	from pyof.foundation.base import UBIntBase
7	1	from pyof.v0x01.asynchronous.error_msg import BadActionCode
8	1	from pyof.v0x01.common.phy_port import PortConfig
9	1	from werkzeug.exceptions import NotFound
10
11	1	from kytos.core import KytosEvent, KytosNApp, log, rest
12	1	from kytos.core.helpers import listen_to
13	1	from napps.kytos.flow_manager.match import match_flow
14	1	from napps.kytos.flow_manager.storehouse import StoreHouse
15	1	from napps.kytos.of_core.flow import FlowFactory
16
17	1	from .exceptions import InvalidCommandError
18	1	from .settings import CONSISTENCY_INTERVAL, FLOWS_DICT_MAX_SIZE
19
20
21	1	def cast_fields(flow_dict):
22		"""Make casting the match fields from UBInt() to native int ."""
23	1	match = flow_dict['match']
24	1	for field, value in match.items():
25	1	if isinstance(value, UBIntBase):
26		match[field] = int(value)
27	1	flow_dict['match'] = match
28	1	return flow_dict
29
30
31	1	class Main(KytosNApp):
32		"""Main class to be used by Kytos controller."""
33
34	1	def setup(self):
35		"""Replace the 'init' method for the KytosApp subclass.
36
37		The setup method is automatically called by the run method.
38		Users shouldn't call this method directly.
39		"""
40	1	log.debug("flow-manager starting")
41	1	self._flow_mods_sent = OrderedDict()
42	1	self._flow_mods_sent_max_size = FLOWS_DICT_MAX_SIZE
43
44		# Storehouse client to save and restore flow data:
45	1	self.storehouse = StoreHouse(self.controller)
46
47		# Format of stored flow data:
48		# {'flow_persistence': {'dpid_str': {'flow_list': [
49		# {'command': '<add\|delete>',
50		# 'flow': {flow_dict}}]}}}
51	1	self.stored_flows = {}
52	1	self.resent_flows = set()
53	1	if CONSISTENCY_INTERVAL > 0:
54	1	self.execute_as_loop(CONSISTENCY_INTERVAL)
55
56	1	def execute(self):
57		"""Run once on NApp 'start' or in a loop.
58
59		The execute method is called by the run method of KytosNApp class.
60		Users shouldn't call this method directly.
61		"""
62		self._load_flows()
63
64		if CONSISTENCY_INTERVAL > 0:
65		self.consistency_check()
66
67	1	def shutdown(self):
68		"""Shutdown routine of the NApp."""
69		log.debug("flow-manager stopping")
70
71	1	@listen_to('kytos/of_core.handshake.completed')
72		def resend_stored_flows(self, event):
73		"""Resend stored Flows."""
74		# if consistency check is enabled, it should take care of this
75	1	if CONSISTENCY_INTERVAL >= 0:
76		return
77	1	switch = event.content['switch']
78	1	dpid = str(switch.dpid)
79		# This can be a problem because this code is running a thread
80	1	if dpid in self.resent_flows:
81		log.debug(f'Flow already resent to the switch {dpid}')
82		return
83	1	if dpid in self.stored_flows:
84	1	flow_list = self.stored_flows[dpid]['flow_list']
85	1	for flow in flow_list:
86	1	command = flow['command']
87	1	flows_dict = {"flows": [flow['flow']]}
88	1	self._install_flows(command, flows_dict, [switch])
89	1	self.resent_flows.add(dpid)
90	1	log.info(f'Flows resent to Switch {dpid}')
91
92	1	def consistency_check(self):
93		"""Check the consistency of flows in each switch."""
94		switches = self.controller.switches.values()
95
96		for switch in switches:
97		# Check if a dpid is a key in 'stored_flows' dictionary
98		if switch.is_enabled():
99		self.check_storehouse_consistency(switch)
100
101		if switch.dpid in self.stored_flows:
102		self.check_switch_consistency(switch)
103
104	1	def check_switch_consistency(self, switch):
105		"""Check consistency of installed flows for a specific switch."""
106	1	dpid = switch.dpid
107
108		# Flows stored in storehouse
109	1	stored_flows = self.stored_flows[dpid]['flow_list']
110
111	1	serializer = FlowFactory.get_class(switch)
112
113	1	for stored_flow in stored_flows:
114	1	command = stored_flow['command']
115	1	stored_flow_obj = serializer.from_dict(stored_flow['flow'], switch)
116
117	1	flow = {'flows': [stored_flow['flow']]}
118
119	1	if stored_flow_obj not in switch.flows:
120	1	if command == 'add':
121	1	log.info('A consistency problem was detected in '
122		f'switch {dpid}.')
123	1	self._install_flows(command, flow, [switch])
124	1	log.info(f'Flow forwarded to switch {dpid} to be '
125		'installed.')
126	1	elif command == 'delete':
127	1	log.info('A consistency problem was detected in '
128		f'switch {dpid}.')
129	1	command = 'delete_strict'
130	1	self._install_flows(command, flow, [switch])
131	1	log.info(f'Flow forwarded to switch {dpid} to be deleted.')
132
133	1	def check_storehouse_consistency(self, switch):
134		"""Check consistency of installed flows for a specific switch."""
135	1	dpid = switch.dpid
136
137	1	for installed_flow in switch.flows:
138	1	if dpid not in self.stored_flows:
139		log.info('A consistency problem was detected in '
140		f'switch {dpid}.')
141		flow = {'flows': [installed_flow.as_dict()]}
142		command = 'delete_strict'
143		self._install_flows(command, flow, [switch])
144		log.info(f'Flow forwarded to switch {dpid} to be deleted.')
145		else:
146	1	serializer = FlowFactory.get_class(switch)
147	1	stored_flows = self.stored_flows[dpid]['flow_list']
148	1	stored_flows_list = [serializer.from_dict(stored_flow['flow'],
149		switch)
150		for stored_flow in stored_flows]
151
152	1	if installed_flow not in stored_flows_list:
153	1	log.info('A consistency problem was detected in '
154		f'switch {dpid}.')
155	1	flow = {'flows': [installed_flow.as_dict()]}
156	1	command = 'delete_strict'
157	1	self._install_flows(command, flow, [switch])
158	1	log.info(f'Flow forwarded to switch {dpid} to be deleted.')
159
160		# pylint: disable=attribute-defined-outside-init
161	1	def _load_flows(self):
162		"""Load stored flows."""
163	1	try:
164	1	data = self.storehouse.get_data()['flow_persistence']
165	1	if 'id' in data:
166		del data['id']
167	1	self.stored_flows = data
168		except (KeyError, FileNotFoundError) as error:
169		log.debug(f'There are no flows to load: {error}')
170		else:
171	1	log.info('Flows loaded.')
172
173	1	def _store_changed_flows(self, command, flow, switch):
174		"""Store changed flows.
175
176		Args:
177		command: Flow command to be installed
178		flow: Flows to be stored
179		switch: Switch target
180		"""
181	1	stored_flows_box = deepcopy(self.stored_flows)
182		# if the flow has a destination dpid it can be stored.
183	1	if not switch:
184		log.info('The Flow cannot be stored, the destination switch '
185		f'have not been specified: {switch}')
186		return
187	1	installed_flow = {}
188	1	flow_list = []
189	1	installed_flow['command'] = command
190	1	installed_flow['flow'] = flow
191	1	deleted_flows = []
192
193	1	serializer = FlowFactory.get_class(switch)
194	1	installed_flow_obj = serializer.from_dict(flow, switch)
195
196	1	if switch.id not in stored_flows_box:
197		# Switch not stored, add to box.
198	1	flow_list.append(installed_flow)
199	1	stored_flows_box[switch.id] = {'flow_list': flow_list}
200		else:
201	1	stored_flows = stored_flows_box[switch.id].get('flow_list', [])
202		# Check if flow already stored
203	1	for stored_flow in stored_flows:
204	1	stored_flow_obj = serializer.from_dict(stored_flow['flow'],
205		switch)
206
207	1	version = switch.connection.protocol.version
208
209	1	if installed_flow['command'] == 'delete':
210		# No strict match
211	1	if match_flow(flow, version, stored_flow['flow']):
212	1	deleted_flows.append(stored_flow)
213
214	1	elif installed_flow_obj == stored_flow_obj:
215	1	if stored_flow['command'] == installed_flow['command']:
216		log.debug('Data already stored.')
217		return
218		# Flow with inconsistency in "command" fields : Remove the
219		# old instruction. This happens when there is a stored
220		# instruction to install the flow, but the new instruction
221		# is to remove it. In this case, the old instruction is
222		# removed and the new one is stored.
223	1	stored_flow['command'] = installed_flow.get('command')
224	1	deleted_flows.append(stored_flow)
225	1	break
226
227		# if installed_flow['command'] != 'delete':
228	1	stored_flows.append(installed_flow)
229	1	for i in deleted_flows:
230	1	stored_flows.remove(i)
231	1	stored_flows_box[switch.id]['flow_list'] = stored_flows
232
233	1	stored_flows_box['id'] = 'flow_persistence'
234	1	self.storehouse.save_flow(stored_flows_box)
235	1	del stored_flows_box['id']
236	1	self.stored_flows = deepcopy(stored_flows_box)
237
238	1	@rest('v2/flows')
239	1	@rest('v2/flows/<dpid>')
240	1	def list(self, dpid=None):
241		"""Retrieve all flows from a switch identified by dpid.
242
243		If no dpid is specified, return all flows from all switches.
244		"""
245	1	if dpid is None:
246	1	switches = self.controller.switches.values()
247		else:
248	1	switches = [self.controller.get_switch_by_dpid(dpid)]
249
250	1	if not any(switches):
251	1	raise NotFound("Switch not found")
252
253	1	switch_flows = {}
254
255	1	for switch in switches:
256	1	flows_dict = [cast_fields(flow.as_dict())
257		for flow in switch.flows]
258	1	switch_flows[switch.dpid] = {'flows': flows_dict}
259
260	1	return jsonify(switch_flows)
261
262	1	@listen_to('kytos.flow_manager.flows.(install\|delete)')
263		def event_flows_install_delete(self, event):
264		"""Install or delete flows in the switches through events.
265
266		Install or delete Flow of switches identified by dpid.
267		"""
268	1	try:
269	1	dpid = event.content['dpid']
270	1	flow_dict = event.content['flow_dict']
271		except KeyError as error:
272		log.error("Error getting fields to install or remove "
273		f"Flows: {error}")
274		return
275
276	1	if event.name == 'kytos.flow_manager.flows.install':
277	1	command = 'add'
278	1	elif event.name == 'kytos.flow_manager.flows.delete':
279	1	command = 'delete'
280		else:
281		msg = f'Invalid event "{event.name}", should be install\|delete'
282		raise ValueError(msg)
283
284	1	switch = self.controller.get_switch_by_dpid(dpid)
285	1	try:
286	1	self._install_flows(command, flow_dict, [switch])
287		except InvalidCommandError as error:
288		log.error("Error installing or deleting Flow through"
289		f" Kytos Event: {error}")
290
291	1	@rest('v2/flows', methods=['POST'])
292	1	@rest('v2/flows/<dpid>', methods=['POST'])
293	1	def add(self, dpid=None):
294		"""Install new flows in the switch identified by dpid.
295
296		If no dpid is specified, install flows in all switches.
297		"""
298	1	return self._send_flow_mods_from_request(dpid, "add")
299
300	1	@rest('v2/delete', methods=['POST'])
301	1	@rest('v2/delete/<dpid>', methods=['POST'])
302	1	@rest('v2/flows', methods=['DELETE'])
303	1	@rest('v2/flows/<dpid>', methods=['DELETE'])
304	1	def delete(self, dpid=None):
305		"""Delete existing flows in the switch identified by dpid.
306
307		If no dpid is specified, delete flows from all switches.
308		"""
309	1	return self._send_flow_mods_from_request(dpid, "delete")
310
311	1	def _get_all_switches_enabled(self):
312		"""Get a list of all switches enabled."""
313	1	switches = self.controller.switches.values()
314	1	return [switch for switch in switches if switch.is_enabled()]
315
316	1	def _send_flow_mods_from_request(self, dpid, command, flows_dict=None):
317		"""Install FlowsMods from request."""
318	1	if flows_dict is None:
319	1	flows_dict = request.get_json()
320	1	if flows_dict is None:
321	1	return jsonify({"response": 'flows dict is none.'}), 404
322
323	1	if dpid:
324	1	switch = self.controller.get_switch_by_dpid(dpid)
325	1	if not switch:
326	1	return jsonify({"response": 'dpid not found.'}), 404
327	1	elif switch.is_enabled() is False:
328	1	if command == "delete":
329	1	self._install_flows(command, flows_dict, [switch])
330		else:
331	1	return jsonify({"response": 'switch is disabled.'}), 404
332		else:
333	1	self._install_flows(command, flows_dict, [switch])
334		else:
335	1	self._install_flows(command, flows_dict,
336		self._get_all_switches_enabled())
337
338	1	return jsonify({"response": "FlowMod Messages Sent"})
339
340	1	def _install_flows(self, command, flows_dict, switches=[]):
341		"""Execute all procedures to install flows in the switches.
342
343		Args:
344		command: Flow command to be installed
345		flows_dict: Dictionary with flows to be installed in the switches.
346		switches: A list of switches
347		"""
348	1	for switch in switches:
349	1	serializer = FlowFactory.get_class(switch)
350	1	flows = flows_dict.get('flows', [])
351	1	for flow_dict in flows:
352	1	flow = serializer.from_dict(flow_dict, switch)
353	1	if command == "delete":
354		flow_mod = flow.as_of_delete_flow_mod()
355	1	elif command == "delete_strict":
356	1	flow_mod = flow.as_of_strict_delete_flow_mod()
357	1	elif command == "add":
358	1	flow_mod = flow.as_of_add_flow_mod()
359		else:
360		raise InvalidCommandError
361	1	self._send_flow_mod(flow.switch, flow_mod)
362	1	self._add_flow_mod_sent(flow_mod.header.xid, flow, command)
363
364	1	self._send_napp_event(switch, flow, command)
365	1	self._store_changed_flows(command, flow_dict, switch)
366
367	1	def _add_flow_mod_sent(self, xid, flow, command):
368		"""Add the flow mod to the list of flow mods sent."""
369	1	if len(self._flow_mods_sent) >= self._flow_mods_sent_max_size:
370		self._flow_mods_sent.popitem(last=False)
371	1	self._flow_mods_sent[xid] = (flow, command)
372
373	1	def _send_flow_mod(self, switch, flow_mod):
374	1	event_name = 'kytos/flow_manager.messages.out.ofpt_flow_mod'
375
376	1	content = {'destination': switch.connection,
377		'message': flow_mod}
378
379	1	event = KytosEvent(name=event_name, content=content)
380	1	self.controller.buffers.msg_out.put(event)
381
382	1	def _send_napp_event(self, switch, flow, command, **kwargs):
383		"""Send an Event to other apps informing about a FlowMod."""
384	1	if command == 'add':
385	1	name = 'kytos/flow_manager.flow.added'
386	1	elif command in ('delete', 'delete_strict'):
387	1	name = 'kytos/flow_manager.flow.removed'
388	1	elif command == 'error':
389	1	name = 'kytos/flow_manager.flow.error'
390		else:
391		raise InvalidCommandError
392	1	content = {'datapath': switch,
393		'flow': flow}
394	1	content.update(kwargs)
395	1	event_app = KytosEvent(name, content)
396	1	self.controller.buffers.app.put(event_app)
397
398	1	@listen_to('..of_core..ofpt_error')
399		def handle_errors(self, event):
400		"""Receive OpenFlow error and send a event.
401
402		The event is sent only if the error is related to a request made
403		by flow_manager.
404		"""
405	1	message = event.content["message"]
406
407	1	connection = event.source
408	1	switch = connection.switch
409
410	1	xid = message.header.xid.value
411	1	error_type = message.error_type
412	1	error_code = message.code
413	1	error_data = message.data.pack()
414
415		# Get the packet responsible for the error
416	1	error_packet = connection.protocol.unpack(error_data)
417
418	1	if message.code == BadActionCode.OFPBAC_BAD_OUT_PORT:
419		actions = []
420		if hasattr(error_packet, 'actions'):
421		# Get actions from the flow mod (OF 1.0)
422		actions = error_packet.actions
423		else:
424		# Get actions from the list of flow mod instructions (OF 1.3)
425		for instruction in error_packet.instructions:
426		actions.extend(instruction.actions)
427
428		for action in actions:
429		iface = switch.get_interface_by_port_no(action.port)
430
431		# Set interface to drop packets forwarded to it
432		if iface:
433		iface.config = PortConfig.OFPPC_NO_FWD
434
435	1	try:
436	1	flow, error_command = self._flow_mods_sent[xid]
437		except KeyError:
438		pass
439		else:
440	1	self._send_napp_event(flow.switch, flow, 'error',
441		error_command=error_command,
442		error_type=error_type, error_code=error_code)
443

kytos / flow_manager

Pull Request — master (#129)

build.main.Main.event_flows_install_delete() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like