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