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