Test Failed
Pull Request — master (#96)
by Jose
02:05
created

build.main.Main._store_changed_flows()   B

Complexity

Conditions 6

Size

Total Lines 48
Code Lines 27

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 9
CRAP Score 6.2163

Importance

Changes 0
Metric Value
cc 6
eloc 27
nop 4
dl 0
loc 48
ccs 9
cts 11
cp 0.8182
crap 6.2163
rs 8.2986
c 0
b 0
f 0
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
    # def resend_stored_flows(self, event):
119 1
    # Add the new version here
120 1
121
    # pylint: disable=attribute-defined-outside-init
122
    def _load_flows(self):
123 1
        """Load stored flows."""
124 1
        try:
125
            data = self.storehouse.get_data()['flow_persistence']
126 1
            if 'id' in data:
127
                del data['id']
128 1
            self.stored_flows = data
129
130 1
        except KeyError as error:
131
            log.debug(f'There are no flows to load: {error}')
132 1
        else:
133
            log.info('Flows loaded.')
134 1
135 1
    @staticmethod
136
    def _generate_match_fields(flow):
137 1
        """Generate flow match fields."""
138
        match_fields = {}
139
        for field in flow:
140 1
            if 'priority' in field:
141 1
                match_fields['priority'] = flow['priority']
142
            if 'cookie' in field:
143 1
                match_fields['cookie'] = flow['cookie']
144
            if 'match' in field:
145 1
                match_fields.update(flow['match'])
146 1
        return match_fields
147 1
148 1
    def _is_equal_flows(self, flow_1, flow_2):
149 1
        """Check if two flows are equal."""
150 1
        flow_1_match_fields = self._generate_match_fields(flow_1)
151
        flow_2_match_fields = self._generate_match_fields(flow_2)
152
        if flow_1_match_fields == flow_2_match_fields:
153 1
            return True
154
        return False
155 1
156 1
    def _store_changed_flows(self, command, flow, switch):
157 1
        """Store changed flows.
158
159 1
        Args:
160
            command: Flow command to be installed
161
            flow: Flows to be stored
162
            switch: Switch target
163
        """
164
        stored_flows_box = self.stored_flows.copy()
165
        # if the flow has a destination dpid it can be stored.
166 1
        if not switch:
167 1
            log.info('The Flow cannot be stored, the destination switch '
168 1
                     f'have not been specified: {switch}')
169 1
            return
170 1
171
        new_flow = {}
172
        flow_list = []
173
        new_flow['command'] = command
174 1
        new_flow['flow'] = flow
175
176
        if switch.id not in stored_flows_box:
177
            # Switch not stored, add to box.
178
            flow_list.append(new_flow)
179
            stored_flows_box[switch.id] = {'flow_list': flow_list}
180
        else:
181
            stored_flows = stored_flows_box[switch.id].get('flow_list', [])
182
            # Check if flow already stored
183
            for stored_flow in stored_flows:
184
                if self._is_equal_flows(new_flow['flow'], stored_flow['flow']):
185
                    if stored_flow['command'] == new_flow['command']:
186
                        log.debug('Data already stored.')
187
                        return
188
                    # Flow with inconsistency in "command" fields : Remove the
189
                    # old instruction. This happens when there is a stored
190
                    # instruction to install the flow, but the new instruction
191
                    # is to remove it. In this case, the old instruction is
192
                    # removed and the new one is stored.
193
                    stored_flow['command'] = new_flow.get('command')
194
                    stored_flows.remove(stored_flow)
195
                    break
196
197
            stored_flows.append(new_flow)
198
            stored_flows_box[switch.id]['flow_list'] = stored_flows
199
200
        stored_flows_box['id'] = 'flow_persistence'
201
        self.storehouse.save_flow(stored_flows_box)
202
        del stored_flows_box['id']
203
        self.stored_flows = stored_flows_box.copy()
204
205
    @rest('v2/flows')
206
    @rest('v2/flows/<dpid>')
207
    def list(self, dpid=None):
208
        """Retrieve all flows from a switch identified by dpid.
209
210
        If no dpid is specified, return all flows from all switches.
211
        """
212
        if dpid is None:
213
            switches = self.controller.switches.values()
214
        else:
215
            switches = [self.controller.get_switch_by_dpid(dpid)]
216
217
        switch_flows = {}
218
219
        for switch in switches:
220
            flows_dict = [flow.as_dict() for flow in switch.flows]
221
            switch_flows[switch.dpid] = {'flows': flows_dict}
222
223
        return jsonify(switch_flows)
224
225
    @rest('v2/flows', methods=['POST'])
226
    @rest('v2/flows/<dpid>', methods=['POST'])
227
    def add(self, dpid=None):
228
        """Install new flows in the switch identified by dpid.
229
230
        If no dpid is specified, install flows in all switches.
231
        """
232
        return self._send_flow_mods_from_request(dpid, "add")
233
234
    @rest('v2/delete', methods=['POST'])
235
    @rest('v2/delete/<dpid>', methods=['POST'])
236
    @rest('v2/flows', methods=['DELETE'])
237
    @rest('v2/flows/<dpid>', methods=['DELETE'])
238
    def delete(self, dpid=None):
239
        """Delete existing flows in the switch identified by dpid.
240
241
        If no dpid is specified, delete flows from all switches.
242
        """
243
        return self._send_flow_mods_from_request(dpid, "delete")
244
245
    def _get_all_switches_enabled(self):
246
        """Get a list of all switches enabled."""
247
        switches = self.controller.switches.values()
248
        return [switch for switch in switches if switch.is_enabled()]
249
250
    def _send_flow_mods_from_request(self, dpid, command, flows_dict=None):
251
        """Install FlowsMods from request."""
252
        if flows_dict is None:
253
            flows_dict = request.get_json()
254
            if flows_dict is None:
255
                return jsonify({"response": 'flows dict is none.'}), 404
256
257
        if dpid:
258
            switch = self.controller.get_switch_by_dpid(dpid)
259
            if not switch:
260
                return jsonify({"response": 'dpid not found.'}), 404
261
            elif switch.is_enabled() is False:
262
                return jsonify({"response": 'switch is disabled.'}), 404
263
            else:
264
                self._install_flows(command, flows_dict, [switch])
265
        else:
266
            self._install_flows(command, flows_dict,
267
                                self._get_all_switches_enabled())
268
269
        return jsonify({"response": "FlowMod Messages Sent"})
270
271
    def _install_flows(self, command, flows_dict, switches=[]):
272
        """Execute all procedures to install flows in the switches.
273
274
        Args:
275
            command: Flow command to be installed
276
            flows_dict: Dictionary with flows to be installed in the switches.
277
            switches: A list of switches
278
        """
279
        for switch in switches:
280
            serializer = FlowFactory.get_class(switch)
281
            flows = flows_dict.get('flows', [])
282
            for flow_dict in flows:
283
                flow = serializer.from_dict(flow_dict, switch)
284
                if command == "delete":
285
                    flow_mod = flow.as_of_delete_flow_mod()
286
                elif command == "add":
287
                    flow_mod = flow.as_of_add_flow_mod()
288
                else:
289
                    raise InvalidCommandError
290
                self._send_flow_mod(flow.switch, flow_mod)
291
                self._add_flow_mod_sent(flow_mod.header.xid, flow, command)
292
293
                self._send_napp_event(switch, flow, command)
294
                self._store_changed_flows(command, flow_dict, switch)
295
296
    def _add_flow_mod_sent(self, xid, flow, command):
297
        """Add the flow mod to the list of flow mods sent."""
298
        if len(self._flow_mods_sent) >= self._flow_mods_sent_max_size:
299
            self._flow_mods_sent.popitem(last=False)
300
        self._flow_mods_sent[xid] = (flow, command)
301
302
    def _send_flow_mod(self, switch, flow_mod):
303
        event_name = 'kytos/flow_manager.messages.out.ofpt_flow_mod'
304
305
        content = {'destination': switch.connection,
306
                   'message': flow_mod}
307
308
        event = KytosEvent(name=event_name, content=content)
309
        self.controller.buffers.msg_out.put(event)
310
311
    def _send_napp_event(self, switch, flow, command, **kwargs):
312
        """Send an Event to other apps informing about a FlowMod."""
313
        if command == 'add':
314
            name = 'kytos/flow_manager.flow.added'
315
        elif command == 'delete':
316
            name = 'kytos/flow_manager.flow.removed'
317
        elif command == 'error':
318
            name = 'kytos/flow_manager.flow.error'
319
        else:
320
            raise InvalidCommandError
321
        content = {'datapath': switch,
322
                   'flow': flow}
323
        content.update(kwargs)
324
        event_app = KytosEvent(name, content)
325
        self.controller.buffers.app.put(event_app)
326
327
    @listen_to('.*.of_core.*.ofpt_error')
328
    def handle_errors(self, event):
329
        """Receive OpenFlow error and send a event.
330
331
        The event is sent only if the error is related to a request made
332
        by flow_manager.
333
        """
334
        xid = event.content["message"].header.xid.value
335
        error_type = event.content["message"].error_type
336
        error_code = event.content["message"].code
337
        try:
338
            flow, error_command = self._flow_mods_sent[xid]
339
        except KeyError:
340
            pass
341
        else:
342
            self._send_napp_event(flow.switch, flow, 'error',
343
                                  error_command=error_command,
344
                                  error_type=error_type, error_code=error_code)
345