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