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