Passed
Push — master ( 930f03...9c9752 )
by Italo Valcy
02:06 queued 17s
created

main.py (4 issues)

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