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