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