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