Passed
Push — master ( 3e32a3...5ad286 )
by Humberto
02:39
created

main.py (6 issues)

1
"""Main module of kytos/topology Kytos Network Application.
2
3
Manage the network topology
4
"""
5 1
import time
6
7 1
from flask import jsonify, request
8
9 1
from kytos.core import KytosEvent, KytosNApp, log, rest
10 1
from kytos.core.exceptions import KytosLinkCreationError
11 1
from kytos.core.helpers import listen_to
12 1
from kytos.core.interface import Interface
13 1
from kytos.core.link import Link
14 1
from kytos.core.switch import Switch
15 1
from napps.kytos.topology import settings
16 1
from napps.kytos.topology.exceptions import RestoreError
17 1
from napps.kytos.topology.models import Topology
18 1
from napps.kytos.topology.storehouse import StoreHouse
19
20 1
DEFAULT_LINK_UP_TIMER = 10
21 1
DEFAULT_INTERFACE_RESTORE_TIMER = 2
22 1
RESTORE_INTERFACE_ATTEMPTS = 20
23
24
25 1
class Main(KytosNApp):  # pylint: disable=too-many-public-methods
26
    """Main class of kytos/topology NApp.
27
28
    This class is the entry point for this napp.
29
    """
30
31 1
    def setup(self):
32
        """Initialize the NApp's links list."""
33 1
        self.links = {}
34 1
        self.store_items = {}
35 1
        self.switches_state = {}
36 1
        self.interfaces_state = {}
37 1
        self.links_state = {}
38 1
        self._verified_links = []
39 1
        self.link_up_timer = getattr(settings, 'LINK_UP_TIMER',
40
                                     DEFAULT_LINK_UP_TIMER)
41 1
        self.interface_restore = getattr(settings, 'INTERFACE_RESTORE_TIMER',
42
                                         DEFAULT_INTERFACE_RESTORE_TIMER)
43
44 1
        self.verify_storehouse('switches')
45 1
        self.verify_storehouse('interfaces')
46 1
        self.verify_storehouse('links')
47
48 1
        self.storehouse = StoreHouse(self.controller)
49
50 1
    def execute(self):
51
        """Execute once when the napp is running."""
52
        self._load_network_status()
53
54 1
    def shutdown(self):
55
        """Do nothing."""
56
        log.info('NApp kytos/topology shutting down.')
57
58 1
    def _get_link_or_create(self, endpoint_a, endpoint_b):
59 1
        new_link = Link(endpoint_a, endpoint_b)
60
61 1
        for link in self.links.values():
62
            if new_link == link:
63
                return link
64
65 1
        self.links[new_link.id] = new_link
66 1
        return new_link
67
68 1
    def _get_switches_dict(self):
69
        """Return a dictionary with the known switches."""
70 1
        switches = {'switches': {}}
71 1
        for idx, switch in enumerate(self.controller.switches.values()):
72 1
            switch_data = switch.as_dict()
73 1
            if not all(key in switch_data['metadata']
74
                       for key in ('lat', 'lng')):
75
                # Switches are initialized somewhere in the ocean
76
                switch_data['metadata']['lat'] = str(0.0)
77
                switch_data['metadata']['lng'] = str(-30.0+idx*10.0)
78 1
            switches['switches'][switch.id] = switch_data
79 1
        return switches
80
81 1
    def _get_links_dict(self):
82
        """Return a dictionary with the known links."""
83 1
        return {'links': {l.id: l.as_dict() for l in
84
                          self.links.values()}}
85
86 1
    def _get_topology_dict(self):
87
        """Return a dictionary with the known topology."""
88 1
        return {'topology': {**self._get_switches_dict(),
89
                             **self._get_links_dict()}}
90
91 1
    def _get_topology(self):
92
        """Return an object representing the topology."""
93 1
        return Topology(self.controller.switches, self.links)
94
95 1
    def _get_link_from_interface(self, interface):
96
        """Return the link of the interface, or None if it does not exist."""
97 1
        for link in self.links.values():
98 1
            if interface in (link.endpoint_a, link.endpoint_b):
99 1
                return link
100 1
        return None
101
102 1
    def _restore_link(self, link_id):
103
        """Restore link's administrative state from storehouse."""
104 1
        try:
105 1
            state = self.links_state[link_id]
106 1
        except KeyError:
107 1
            error = (f'The link {link_id} has no stored '
108
                     'administrative state to be restored.')
109 1
            raise RestoreError(error)
110
111 1
        try:
112 1
            link = self.links[link_id]
113 1
            if state['enabled']:
114 1
                link.enable()
115
            else:
116 1
                link.disable()
117 1
        except KeyError:
118 1
            error = ('Error restoring link status.'
119
                     f'The link {link_id} does not exist.')
120 1
            raise RestoreError(error)
121 1
        log.info(f'The state of link {link.id} has been restored.')
122 1
        self.notify_topology_update()
123 1
        self.update_instance_metadata(link)
124 1
        self.notify_link_status_change(link)
125
126 1
    def _restore_switch(self, switch_id):
127
        """Restore switch's administrative state from storehouse."""
128 1
        try:
129 1
            state = self.switches_state[switch_id]
130 1
        except KeyError:
131 1
            error = (f'The switch {switch_id} has no stored'
132
                     ' administrative state to be restored.')
133 1
            raise RestoreError(error)
134
135 1
        try:
136 1
            switch = self.controller.switches[switch_id]
137 1
        except KeyError:
138
            # Maybe we should remove the switch from switches_state here
139 1
            error = ('Error while restoring switches status. The '
140
                     f'switch {switch_id} does not exist.')
141 1
            raise RestoreError(error)
142
143 1
        if state:
144 1
            switch.enable()
145 1
            self.notify_switch_enabled(switch_id)
146
        else:
147 1
            switch.disable()
148 1
            self.notify_switch_disabled(switch_id)
149
150 1
        log.debug('Waiting to restore administrative state of switch '
151
                  f'{switch_id} interfaces.')
152 1
        i = 0
153
        # wait to restore interfaces
154 1
        while not switch.interfaces and i < RESTORE_INTERFACE_ATTEMPTS:
155
            time.sleep(self.interface_restore)
156
            i += 1
157 1
        if not switch.interfaces:
158
            error = ('Error restoring administrative state of switch '
159
                     f'{switch_id} interfaces.')
160
            raise RestoreError(error)
161
162
        # restore interfaces
163 1
        for interface_id in switch.interfaces:
164 1
            iface_id = ":".join([switch_id, str(interface_id)])
165
            # restore only the administrative state of saved interfaces
166 1
            if iface_id not in self.interfaces_state:
167
                error = ("The stored topology is different from the current "
168
                         f"topology. The interface {iface_id} hasn't been "
169
                         "stored.")
170
                log.info(error)
171
                continue
172 1
            state = self.interfaces_state[iface_id]
173 1
            iface_number = int(interface_id)
174 1
            iface_status, lldp_status = state
175 1
            try:
176 1
                interface = switch.interfaces[iface_number]
177
            except KeyError:
178
                log.error('Error restoring interface status: '
179
                          '%s does not exist.', iface_id)
180
                continue
181
182 1
            if iface_status:
183 1
                interface.enable()
184
            else:
185 1
                interface.disable()
186 1
            interface.lldp = lldp_status
187 1
            self.update_instance_metadata(interface)
188
189 1
        log.info(f'The state of switch {switch_id} has been restored.')
190
191
    # pylint: disable=attribute-defined-outside-init
192 1
    def _load_network_status(self):
193
        """Load network status saved in storehouse."""
194 1
        try:
195 1
            status = self.storehouse.get_data()
196
        except FileNotFoundError as error:
197
            log.info(error)
198
            return
199 1
        if status:
200 1
            switches = status['network_status']['switches']
201 1
            self.links_state = status['network_status']['links']
202
203 1
            for switch_id, switch_att in switches.items():
204
                # get switches status
205 1
                self.switches_state[switch_id] = switch_att['enabled']
206 1
                iface = switch_att['interfaces']
207
                # get interface status
208 1
                for iface_id, iface_att in iface.items():
209 1
                    enabled_value = iface_att['enabled']
210 1
                    lldp_value = iface_att['lldp']
211 1
                    self.interfaces_state[iface_id] = (enabled_value,
212
                                                       lldp_value)
213
214
        else:
215
            error = 'There is no status saved to restore.'
216
            log.info(error)
217
218 1
    @rest('v3/')
219
    def get_topology(self):
220
        """Return the latest known topology.
221
222
        This topology is updated when there are network events.
223
        """
224 1
        return jsonify(self._get_topology_dict())
225
226 1
    def restore_network_status(self, obj):
227
        """Restore the network administrative status saved in storehouse."""
228 1
        try:
229 1
            if isinstance(obj, Switch):
230 1
                self._restore_switch(obj.id)
231 1
            elif isinstance(obj, Link):
232 1
                if obj.id not in self._verified_links:
233 1
                    self._verified_links.append(obj.id)
234 1
                    self._restore_link(obj.id)
235
        except RestoreError as exc:
236
            log.debug(exc)
237
238
    # Switch related methods
239 1
    @rest('v3/switches')
240
    def get_switches(self):
241
        """Return a json with all the switches in the topology."""
242
        return jsonify(self._get_switches_dict())
243
244 1
    @rest('v3/switches/<dpid>/enable', methods=['POST'])
245
    def enable_switch(self, dpid):
246
        """Administratively enable a switch in the topology."""
247 1
        try:
248 1
            self.controller.switches[dpid].enable()
249 1
        except KeyError:
250 1
            return jsonify("Switch not found"), 404
251
252 1
        log.info(f"Storing administrative state from switch {dpid}"
253
                 " to enabled.")
254 1
        self.save_status_on_storehouse()
255 1
        self.notify_switch_enabled(dpid)
256 1
        return jsonify("Operation successful"), 201
257
258 1
    @rest('v3/switches/<dpid>/disable', methods=['POST'])
259
    def disable_switch(self, dpid):
260
        """Administratively disable a switch in the topology."""
261 1
        try:
262 1
            self.controller.switches[dpid].disable()
263 1
        except KeyError:
264 1
            return jsonify("Switch not found"), 404
265
266 1
        log.info(f"Storing administrative state from switch {dpid}"
267
                 " to disabled.")
268 1
        self.save_status_on_storehouse()
269 1
        self.notify_switch_disabled(dpid)
270 1
        return jsonify("Operation successful"), 201
271
272 1
    @rest('v3/switches/<dpid>/metadata')
273
    def get_switch_metadata(self, dpid):
274
        """Get metadata from a switch."""
275 1
        try:
276 1
            return jsonify({"metadata":
277
                            self.controller.switches[dpid].metadata}), 200
278 1
        except KeyError:
279 1
            return jsonify("Switch not found"), 404
280
281 1
    @rest('v3/switches/<dpid>/metadata', methods=['POST'])
282
    def add_switch_metadata(self, dpid):
283
        """Add metadata to a switch."""
284 1
        metadata = request.get_json()
285 1
        try:
286 1
            switch = self.controller.switches[dpid]
287 1
        except KeyError:
288 1
            return jsonify("Switch not found"), 404
289
290 1
        switch.extend_metadata(metadata)
291 1
        self.notify_metadata_changes(switch, 'added')
292 1
        return jsonify("Operation successful"), 201
293
294 1
    @rest('v3/switches/<dpid>/metadata/<key>', methods=['DELETE'])
295
    def delete_switch_metadata(self, dpid, key):
296
        """Delete metadata from a switch."""
297 1
        try:
298 1
            switch = self.controller.switches[dpid]
299 1
        except KeyError:
300 1
            return jsonify("Switch not found"), 404
301
302 1
        switch.remove_metadata(key)
303 1
        self.notify_metadata_changes(switch, 'removed')
304 1
        return jsonify("Operation successful"), 200
305
306
    # Interface related methods
307 1
    @rest('v3/interfaces')
308
    def get_interfaces(self):
309
        """Return a json with all the interfaces in the topology."""
310
        interfaces = {}
311
        switches = self._get_switches_dict()
312
        for switch in switches['switches'].values():
313
            for interface_id, interface in switch['interfaces'].items():
314
                interfaces[interface_id] = interface
315
316
        return jsonify({'interfaces': interfaces})
317
318 1 View Code Duplication
    @rest('v3/interfaces/switch/<dpid>/enable', methods=['POST'])
0 ignored issues
show
This code seems to be duplicated in your project.
Loading history...
319 1
    @rest('v3/interfaces/<interface_enable_id>/enable', methods=['POST'])
320 1
    def enable_interface(self, interface_enable_id=None, dpid=None):
321
        """Administratively enable interfaces in the topology."""
322 1
        error_list = []  # List of interfaces that were not activated.
323 1
        msg_error = "Some interfaces couldn't be found and activated: "
324 1
        if dpid is None:
325 1
            dpid = ":".join(interface_enable_id.split(":")[:-1])
326 1
        try:
327 1
            switch = self.controller.switches[dpid]
328 1
        except KeyError as exc:
329 1
            return jsonify(f"Switch not found: {exc}"), 404
330
331 1
        if interface_enable_id:
332 1
            interface_number = int(interface_enable_id.split(":")[-1])
333
334 1
            try:
335 1
                switch.interfaces[interface_number].enable()
336 1
            except KeyError as exc:
337 1
                error_list.append(f"Switch {dpid} Interface {exc}")
338
        else:
339 1
            for interface in switch.interfaces.values():
340 1
                interface.enable()
341 1
        if not error_list:
342 1
            log.info(f"Storing administrative state for enabled interfaces.")
343 1
            self.save_status_on_storehouse()
344 1
            return jsonify("Operation successful"), 200
345 1
        return jsonify({msg_error:
346
                        error_list}), 409
347
348 1 View Code Duplication
    @rest('v3/interfaces/switch/<dpid>/disable', methods=['POST'])
0 ignored issues
show
This code seems to be duplicated in your project.
Loading history...
349 1
    @rest('v3/interfaces/<interface_disable_id>/disable', methods=['POST'])
350 1
    def disable_interface(self, interface_disable_id=None, dpid=None):
351
        """Administratively disable interfaces in the topology."""
352 1
        error_list = []  # List of interfaces that were not deactivated.
353 1
        msg_error = "Some interfaces couldn't be found and deactivated: "
354 1
        if dpid is None:
355 1
            dpid = ":".join(interface_disable_id.split(":")[:-1])
356 1
        try:
357 1
            switch = self.controller.switches[dpid]
358 1
        except KeyError as exc:
359 1
            return jsonify(f"Switch not found: {exc}"), 404
360
361 1
        if interface_disable_id:
362 1
            interface_number = int(interface_disable_id.split(":")[-1])
363
364 1
            try:
365 1
                switch.interfaces[interface_number].disable()
366 1
            except KeyError as exc:
367 1
                error_list.append(f"Switch {dpid} Interface {exc}")
368
        else:
369 1
            for interface in switch.interfaces.values():
370 1
                interface.disable()
371 1
        if not error_list:
372 1
            log.info(f"Storing administrative state for disabled interfaces.")
373 1
            self.save_status_on_storehouse()
374 1
            return jsonify("Operation successful"), 200
375 1
        return jsonify({msg_error:
376
                        error_list}), 409
377
378 1
    @rest('v3/interfaces/<interface_id>/metadata')
379
    def get_interface_metadata(self, interface_id):
380
        """Get metadata from an interface."""
381 1
        switch_id = ":".join(interface_id.split(":")[:-1])
382 1
        interface_number = int(interface_id.split(":")[-1])
383 1
        try:
384 1
            switch = self.controller.switches[switch_id]
385 1
        except KeyError:
386 1
            return jsonify("Switch not found"), 404
387
388 1
        try:
389 1
            interface = switch.interfaces[interface_number]
390 1
        except KeyError:
391 1
            return jsonify("Interface not found"), 404
392
393 1
        return jsonify({"metadata": interface.metadata}), 200
394
395 1
    @rest('v3/interfaces/<interface_id>/metadata', methods=['POST'])
396
    def add_interface_metadata(self, interface_id):
397
        """Add metadata to an interface."""
398 1
        metadata = request.get_json()
399
400 1
        switch_id = ":".join(interface_id.split(":")[:-1])
401 1
        interface_number = int(interface_id.split(":")[-1])
402 1
        try:
403 1
            switch = self.controller.switches[switch_id]
404 1
        except KeyError:
405 1
            return jsonify("Switch not found"), 404
406
407 1
        try:
408 1
            interface = switch.interfaces[interface_number]
409 1
        except KeyError:
410 1
            return jsonify("Interface not found"), 404
411
412 1
        interface.extend_metadata(metadata)
413 1
        self.notify_metadata_changes(interface, 'added')
414 1
        return jsonify("Operation successful"), 201
415
416 1
    @rest('v3/interfaces/<interface_id>/metadata/<key>', methods=['DELETE'])
417
    def delete_interface_metadata(self, interface_id, key):
418
        """Delete metadata from an interface."""
419 1
        switch_id = ":".join(interface_id.split(":")[:-1])
420 1
        interface_number = int(interface_id.split(":")[-1])
421
422 1
        try:
423 1
            switch = self.controller.switches[switch_id]
424 1
        except KeyError:
425 1
            return jsonify("Switch not found"), 404
426
427 1
        try:
428 1
            interface = switch.interfaces[interface_number]
429 1
        except KeyError:
430 1
            return jsonify("Interface not found"), 404
431
432 1
        if interface.remove_metadata(key) is False:
433 1
            return jsonify("Metadata not found"), 404
434
435 1
        self.notify_metadata_changes(interface, 'removed')
436 1
        return jsonify("Operation successful"), 200
437
438
    # Link related methods
439 1
    @rest('v3/links')
440
    def get_links(self):
441
        """Return a json with all the links in the topology.
442
443
        Links are connections between interfaces.
444
        """
445
        return jsonify(self._get_links_dict()), 200
446
447 1
    @rest('v3/links/<link_id>/enable', methods=['POST'])
448
    def enable_link(self, link_id):
449
        """Administratively enable a link in the topology."""
450 1
        try:
451 1
            self.links[link_id].enable()
452 1
        except KeyError:
453 1
            return jsonify("Link not found"), 404
454 1
        self.save_status_on_storehouse()
455 1
        self.notify_link_status_change(self.links[link_id])
456 1
        return jsonify("Operation successful"), 201
457
458 1
    @rest('v3/links/<link_id>/disable', methods=['POST'])
459
    def disable_link(self, link_id):
460
        """Administratively disable a link in the topology."""
461 1
        try:
462 1
            self.links[link_id].disable()
463 1
        except KeyError:
464 1
            return jsonify("Link not found"), 404
465 1
        self.save_status_on_storehouse()
466 1
        self.notify_link_status_change(self.links[link_id])
467 1
        return jsonify("Operation successful"), 201
468
469 1
    @rest('v3/links/<link_id>/metadata')
470
    def get_link_metadata(self, link_id):
471
        """Get metadata from a link."""
472 1
        try:
473 1
            return jsonify({"metadata": self.links[link_id].metadata}), 200
474 1
        except KeyError:
475 1
            return jsonify("Link not found"), 404
476
477 1
    @rest('v3/links/<link_id>/metadata', methods=['POST'])
478
    def add_link_metadata(self, link_id):
479
        """Add metadata to a link."""
480 1
        metadata = request.get_json()
481 1
        try:
482 1
            link = self.links[link_id]
483 1
        except KeyError:
484 1
            return jsonify("Link not found"), 404
485
486 1
        link.extend_metadata(metadata)
487 1
        self.notify_metadata_changes(link, 'added')
488 1
        return jsonify("Operation successful"), 201
489
490 1
    @rest('v3/links/<link_id>/metadata/<key>', methods=['DELETE'])
491
    def delete_link_metadata(self, link_id, key):
492
        """Delete metadata from a link."""
493 1
        try:
494 1
            link = self.links[link_id]
495 1
        except KeyError:
496 1
            return jsonify("Link not found"), 404
497
498 1
        if link.remove_metadata(key) is False:
499 1
            return jsonify("Metadata not found"), 404
500
501 1
        self.notify_metadata_changes(link, 'removed')
502 1
        return jsonify("Operation successful"), 200
503
504 1
    @listen_to('.*.switch.(new|reconnected)')
505
    def handle_new_switch(self, event):
506
        """Create a new Device on the Topology.
507
508
        Handle the event of a new created switch and update the topology with
509
        this new device. Also notify if the switch is enabled.
510
        """
511 1
        switch = event.content['switch']
512 1
        switch.activate()
513 1
        log.debug('Switch %s added to the Topology.', switch.id)
514 1
        self.notify_topology_update()
515 1
        self.update_instance_metadata(switch)
516 1
        self.restore_network_status(switch)
517 1
        if switch.is_enabled():
518 1
            self.notify_switch_enabled(switch.id)
519
520 1
    @listen_to('.*.connection.lost')
521
    def handle_connection_lost(self, event):
522
        """Remove a Device from the topology.
523
524
        Remove the disconnected Device and every link that has one of its
525
        interfaces.
526
        """
527 1
        switch = event.content['source'].switch
528 1
        if switch:
529 1
            switch.deactivate()
530 1
            log.debug('Switch %s removed from the Topology.', switch.id)
531 1
            self.notify_topology_update()
532
533 1
    def handle_interface_up(self, event):
534
        """Update the topology based on a Port Modify event.
535
536
        The event notifies that an interface was changed to 'up'.
537
        """
538 1
        interface = event.content['interface']
539 1
        interface.activate()
540 1
        self.notify_topology_update()
541 1
        self.update_instance_metadata(interface)
542
543 1
    @listen_to('.*.switch.interface.created')
544
    def handle_interface_created(self, event):
545
        """Update the topology based on a Port Create event."""
546 1
        self.handle_interface_up(event)
547
548 1
    def handle_interface_down(self, event):
549
        """Update the topology based on a Port Modify event.
550
551
        The event notifies that an interface was changed to 'down'.
552
        """
553 1
        interface = event.content['interface']
554 1
        interface.deactivate()
555 1
        self.handle_interface_link_down(event)
556 1
        self.notify_topology_update()
557
558 1
    @listen_to('.*.switch.interface.deleted')
559
    def handle_interface_deleted(self, event):
560
        """Update the topology based on a Port Delete event."""
561 1
        self.handle_interface_down(event)
562
563 1
    @listen_to('.*.switch.interface.link_up')
564
    def handle_interface_link_up(self, event):
565
        """Update the topology based on a Port Modify event.
566
567
        The event notifies that an interface's link was changed to 'up'.
568
        """
569 1
        interface = event.content['interface']
570 1
        self.handle_link_up(interface)
571
572 1
    @listen_to('kytos/maintenance.end_switch')
573
    def handle_switch_maintenance_end(self, event):
574
        """Handle the end of the maintenance of a switch."""
575 1
        switches = event.content['switches']
576 1
        for switch in switches:
577 1
            switch.enable()
578 1
            switch.activate()
579 1
            for interface in switch.interfaces.values():
580 1
                interface.enable()
581 1
                self.handle_link_up(interface)
582
583 1
    def handle_link_up(self, interface):
584
        """Notify a link is up."""
585 1
        link = self._get_link_from_interface(interface)
586 1
        if not link:
587
            return
588 1
        if link.endpoint_a == interface:
589 1
            other_interface = link.endpoint_b
590
        else:
591
            other_interface = link.endpoint_a
592 1
        interface.activate()
593 1
        if other_interface.is_active() is False:
594
            return
595 1
        if link.is_active() is False:
596 1
            link.update_metadata('last_status_change', time.time())
597 1
            link.activate()
598
599
            # As each run of this method uses a different thread,
600
            # there is no risk this sleep will lock the NApp.
601 1
            time.sleep(self.link_up_timer)
602
603 1
            last_status_change = link.get_metadata('last_status_change')
604 1
            now = time.time()
605 1
            if link.is_active() and \
606
                    now - last_status_change >= self.link_up_timer:
607 1
                self.notify_topology_update()
608 1
                self.update_instance_metadata(link)
609 1
                self.notify_link_status_change(link)
610
611 1
    @listen_to('.*.switch.interface.link_down')
612
    def handle_interface_link_down(self, event):
613
        """Update the topology based on a Port Modify event.
614
615
        The event notifies that an interface's link was changed to 'down'.
616
        """
617 1
        interface = event.content['interface']
618 1
        self.handle_link_down(interface)
619
620 1
    @listen_to('kytos/maintenance.start_switch')
621
    def handle_switch_maintenance_start(self, event):
622
        """Handle the start of the maintenance of a switch."""
623 1
        switches = event.content['switches']
624 1
        for switch in switches:
625 1
            switch.disable()
626 1
            switch.deactivate()
627 1
            for interface in switch.interfaces.values():
628 1
                interface.disable()
629 1
                if interface.is_active():
630 1
                    self.handle_link_down(interface)
631
632 1
    def handle_link_down(self, interface):
633
        """Notify a link is down."""
634 1
        link = self._get_link_from_interface(interface)
635 1
        if link and link.is_active():
636 1
            link.deactivate()
637 1
            link.update_metadata('last_status_change', time.time())
638 1
            self.notify_topology_update()
639 1
            self.notify_link_status_change(link)
640
641 1
    @listen_to('.*.interface.is.nni')
642
    def add_links(self, event):
643
        """Update the topology with links related to the NNI interfaces."""
644 1
        interface_a = event.content['interface_a']
645 1
        interface_b = event.content['interface_b']
646
647 1
        try:
648 1
            link = self._get_link_or_create(interface_a, interface_b)
649
        except KytosLinkCreationError as err:
650
            log.error(f'Error creating link: {err}.')
651
            return
652
653 1
        interface_a.update_link(link)
654 1
        interface_b.update_link(link)
655
656 1
        interface_a.nni = True
657 1
        interface_b.nni = True
658
659 1
        self.notify_topology_update()
660 1
        self.restore_network_status(link)
661
662
    # def add_host(self, event):
663
    #    """Update the topology with a new Host."""
664
665
    #    interface = event.content['port']
666
    #    mac = event.content['reachable_mac']
667
668
    #    host = Host(mac)
669
    #    link = self.topology.get_link(interface.id)
670
    #    if link is not None:
671
    #        return
672
673
    #    self.topology.add_link(interface.id, host.id)
674
    #    self.topology.add_device(host)
675
676
    #    if settings.DISPLAY_FULL_DUPLEX_LINKS:
677
    #        self.topology.add_link(host.id, interface.id)
678
679
    # pylint: disable=unused-argument
680 1
    @listen_to('.*.network_status.updated')
681 1
    def save_status_on_storehouse(self, event=None):
682
        """Save the network administrative status using storehouse."""
683 1
        status = self._get_switches_dict()
684 1
        status['id'] = 'network_status'
685 1
        if event:
686
            content = event.content
687
            log.info(f"Storing the administrative state of the"
688
                     f" {content['attribute']} attribute to"
689
                     f" {content['state']} in the interfaces"
690
                     f" {content['interface_ids']}")
691 1
        status.update(self._get_links_dict())
692 1
        self.storehouse.save_status(status)
693
694 1
    def notify_switch_enabled(self, dpid):
695
        """Send an event to notify that a switch is enabled."""
696 1
        name = 'kytos/topology.switch.enabled'
697 1
        event = KytosEvent(name=name, content={'dpid': dpid})
698 1
        self.controller.buffers.app.put(event)
699
700 1
    def notify_switch_disabled(self, dpid):
701
        """Send an event to notify that a switch is disabled."""
702 1
        name = 'kytos/topology.switch.disabled'
703 1
        event = KytosEvent(name=name, content={'dpid': dpid})
704 1
        self.controller.buffers.app.put(event)
705
706 1
    def notify_topology_update(self):
707
        """Send an event to notify about updates on the topology."""
708 1
        name = 'kytos/topology.updated'
709 1
        event = KytosEvent(name=name, content={'topology':
710
                                               self._get_topology()})
711 1
        self.controller.buffers.app.put(event)
712
713 1
    def notify_link_status_change(self, link):
714
        """Send an event to notify about a status change on a link."""
715 1
        name = 'kytos/topology.'
716 1
        if link.is_active() and link.is_enabled():
717 1
            status = 'link_up'
718
        else:
719
            status = 'link_down'
720 1
        event = KytosEvent(name=name+status, content={'link': link})
721 1
        self.controller.buffers.app.put(event)
722
723 1
    def notify_metadata_changes(self, obj, action):
724
        """Send an event to notify about metadata changes."""
725 1
        if isinstance(obj, Switch):
726 1
            entity = 'switch'
727 1
            entities = 'switches'
728 1
        elif isinstance(obj, Interface):
729 1
            entity = 'interface'
730 1
            entities = 'interfaces'
731
        elif isinstance(obj, Link):
732
            entity = 'link'
733
            entities = 'links'
734
735 1
        name = f'kytos/topology.{entities}.metadata.{action}'
736 1
        event = KytosEvent(name=name, content={entity: obj,
0 ignored issues
show
The variable entity does not seem to be defined for all execution paths.
Loading history...
737
                                               'metadata': obj.metadata})
738 1
        self.controller.buffers.app.put(event)
739 1
        log.debug(f'Metadata from {obj.id} was {action}.')
740
741 1
    @listen_to('.*.switch.port.created')
742
    def notify_port_created(self, original_event):
743
        """Notify when a port is created."""
744 1
        name = 'kytos/topology.port.created'
745 1
        event = KytosEvent(name=name, content=original_event.content)
746 1
        self.controller.buffers.app.put(event)
747
748 1
    @listen_to('kytos/topology.*.metadata.*')
749
    def save_metadata_on_store(self, event):
750
        """Send to storehouse the data updated."""
751 1
        name = 'kytos.storehouse.update'
752 1
        if 'switch' in event.content:
753 1
            store = self.store_items.get('switches')
754 1
            obj = event.content.get('switch')
755 1
            namespace = 'kytos.topology.switches.metadata'
756 1
        elif 'interface' in event.content:
757 1
            store = self.store_items.get('interfaces')
758 1
            obj = event.content.get('interface')
759 1
            namespace = 'kytos.topology.interfaces.metadata'
760 1
        elif 'link' in event.content:
761 1
            store = self.store_items.get('links')
762 1
            obj = event.content.get('link')
763 1
            namespace = 'kytos.topology.links.metadata'
764
765 1
        store.data[obj.id] = obj.metadata
0 ignored issues
show
The variable store does not seem to be defined for all execution paths.
Loading history...
The variable obj does not seem to be defined for all execution paths.
Loading history...
766 1
        content = {'namespace': namespace,
0 ignored issues
show
The variable namespace does not seem to be defined for all execution paths.
Loading history...
767
                   'box_id': store.box_id,
768
                   'data': store.data,
769
                   'callback': self.update_instance}
770
771 1
        event = KytosEvent(name=name, content=content)
772 1
        self.controller.buffers.app.put(event)
773
774 1
    @staticmethod
775
    def update_instance(event, _data, error):
776
        """Display in Kytos console if the data was updated."""
777
        entities = event.content.get('namespace', '').split('.')[-2]
778
        if error:
779
            log.error(f'Error trying to update storehouse {entities}.')
780
        else:
781
            log.debug(f'Storehouse update to entities: {entities}.')
782
783 1
    def verify_storehouse(self, entities):
784
        """Request a list of box saved by specific entity."""
785 1
        name = 'kytos.storehouse.list'
786 1
        content = {'namespace': f'kytos.topology.{entities}.metadata',
787
                   'callback': self.request_retrieve_entities}
788 1
        event = KytosEvent(name=name, content=content)
789 1
        self.controller.buffers.app.put(event)
790 1
        log.info(f'verify data in storehouse for {entities}.')
791
792 1
    def request_retrieve_entities(self, event, data, _error):
793
        """Create a box or retrieve an existent box from storehouse."""
794 1
        msg = ''
795 1
        content = {'namespace': event.content.get('namespace'),
796
                   'callback': self.load_from_store,
797
                   'data': {}}
798
799 1
        if not data:
800 1
            name = 'kytos.storehouse.create'
801 1
            msg = 'Create new box in storehouse'
802
        else:
803 1
            name = 'kytos.storehouse.retrieve'
804 1
            content['box_id'] = data[0]
805 1
            msg = 'Retrieve data from storehouse.'
806
807 1
        event = KytosEvent(name=name, content=content)
808 1
        self.controller.buffers.app.put(event)
809 1
        log.debug(msg)
810
811 1
    def load_from_store(self, event, box, error):
812
        """Save the data retrived from storehouse."""
813
        entities = event.content.get('namespace', '').split('.')[-2]
814
        if error:
815
            log.error('Error while get a box from storehouse.')
816
        else:
817
            self.store_items[entities] = box
818
            log.debug('Data updated')
819
820 1
    def update_instance_metadata(self, obj):
821
        """Update object instance with saved metadata."""
822 1
        metadata = None
823 1
        if isinstance(obj, Interface):
824 1
            all_metadata = self.store_items.get('interfaces', None)
825 1
            if all_metadata:
826
                metadata = all_metadata.data.get(obj.id)
827 1
        elif isinstance(obj, Switch):
828
            all_metadata = self.store_items.get('switches', None)
829
            if all_metadata:
830
                metadata = all_metadata.data.get(obj.id)
831 1
        elif isinstance(obj, Link):
832 1
            all_metadata = self.store_items.get('links', None)
833 1
            if all_metadata:
834
                metadata = all_metadata.data.get(obj.id)
835
836 1
        if metadata:
837
            obj.extend_metadata(metadata)
838
            log.debug(f'Metadata to {obj.id} was updated')
839
840 1
    @listen_to('kytos/maintenance.start_link')
841
    def handle_link_maintenance_start(self, event):
842
        """Deals with the start of links maintenance."""
843 1
        notify_links = []
844 1
        maintenance_links = event.content['links']
845 1
        for maintenance_link in maintenance_links:
846 1
            try:
847 1
                link = self.links[maintenance_link.id]
848 1
            except KeyError:
849 1
                continue
850 1
            notify_links.append(link)
851 1
        for link in notify_links:
852 1
            link.disable()
853 1
            link.deactivate()
854 1
            link.endpoint_a.deactivate()
855 1
            link.endpoint_b.deactivate()
856 1
            link.endpoint_a.disable()
857 1
            link.endpoint_b.disable()
858 1
            self.notify_link_status_change(link)
859
860 1
    @listen_to('kytos/maintenance.end_link')
861
    def handle_link_maintenance_end(self, event):
862
        """Deals with the end of links maintenance."""
863 1
        notify_links = []
864 1
        maintenance_links = event.content['links']
865 1
        for maintenance_link in maintenance_links:
866 1
            try:
867 1
                link = self.links[maintenance_link.id]
868 1
            except KeyError:
869 1
                continue
870 1
            notify_links.append(link)
871 1
        for link in notify_links:
872 1
            link.enable()
873 1
            link.activate()
874 1
            link.endpoint_a.activate()
875 1
            link.endpoint_b.activate()
876 1
            link.endpoint_a.enable()
877 1
            link.endpoint_b.enable()
878
            self.notify_link_status_change(link)
879