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