Test Failed
Pull Request — master (#129)
by Vinicius
09:59 queued 04:16
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
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 1
                                 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 1
24
from .controllers import TopoController
25 1
from .exceptions import RestoreError
26 1
from .models import Topology
27 1
28
DEFAULT_LINK_UP_TIMER = 10
29 1
30
31
class Main(KytosNApp):  # pylint: disable=too-many-public-methods
32 1
    """Main class of kytos/topology NApp.
33
34
    This class is the entry point for this napp.
35
    """
36
37
    def setup(self):
38 1
        """Initialize the NApp's links list."""
39
        self.links = {}
40 1
        self.intf_available_tags = {}
41 1
        self.link_up_timer = getattr(settings, 'LINK_UP_TIMER',
42 1
                                     DEFAULT_LINK_UP_TIMER)
43
44
        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 1
                                  self.link_status_hook_link_up_timer)
49
        self.topo_controller.bootstrap_indexes()
50 1
        self.load_topology()
51 1
52
    @staticmethod
53 1
    def get_topo_controller() -> TopoController:
54 1
        """Get TopoController."""
55
        return TopoController()
56
57
    def execute(self):
58 1
        """Execute once when the napp is running."""
59
        pass
60
61
    def shutdown(self):
62 1
        """Do nothing."""
63
        log.info('NApp kytos/topology shutting down.')
64
65
    def _get_metadata(self, request: Request) -> dict:
66 1
        """Return a JSON with metadata."""
67 1
        content_type_json_or_415(request)
68
        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 1
73 1
    def _get_link_or_create(self, endpoint_a, endpoint_b):
74 1
        """Get an existing link or create a new one.
75 1
76
        Returns:
77
            Tuple(Link, bool): Link and a boolean whether it has been created.
78 1
        """
79 1
        new_link = Link(endpoint_a, endpoint_b)
80
81
        for link in self.links.values():
82
            if new_link == link:
83 1
                return (link, False)
84 1
85 1
        self.links[new_link.id] = new_link
86
        return (new_link, True)
87 1
88
    def _get_switches_dict(self):
89
        """Return a dictionary with the known switches."""
90
        switches = {'switches': {}}
91
        for idx, switch in enumerate(self.controller.switches.copy().values()):
92
            switch_data = switch.as_dict()
93 1
            if not all(key in switch_data['metadata']
94
                       for key in ('lat', 'lng')):
95 1
                # Switches are initialized somewhere in the ocean
96 1
                switch_data['metadata']['lat'] = str(0.0)
97 1
                switch_data['metadata']['lng'] = str(-30.0+idx*10.0)
98
            switches['switches'][switch.id] = switch_data
99 1
        return switches
100 1
101
    def _get_links_dict(self):
102 1
        """Return a dictionary with the known links."""
103
        return {'links': {link.id: link.as_dict() for link in
104 1
                          self.links.copy().values()}}
105 1
106 1
    def _get_topology_dict(self):
107 1
        """Return a dictionary with the known topology."""
108
        return {'topology': {**self._get_switches_dict(),
109
                             **self._get_links_dict()}}
110
111
    def _get_topology(self):
112 1
        """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
            for link in self.links.values():
119
                if interface in (link.endpoint_a, link.endpoint_b):
120 1
                    return link
121
            return None
122 1
123
    def _load_link(self, link_att):
124
        endpoint_a = link_att['endpoint_a']['id']
125 1
        endpoint_b = link_att['endpoint_b']['id']
126
        link_str = link_att['id']
127 1
        log.info(f"Loading link: {link_str}")
128
        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 1
            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 1
140 1
        if link_att['enabled']:
141 1
            link.enable()
142 1
        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
        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
        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
        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 1
                            port_no=iface_att['port_number'],
169 1
                            name=iface_att['name'],
170
                            address=iface_att.get('mac', None),
171 1
                            speed=iface_att.get('speed', None))
172 1
            if iface_att['enabled']:
173 1
                interface.enable()
174 1
            else:
175 1
                interface.disable()
176 1
            interface.lldp = iface_att['lldp']
177 1
            interface.extend_metadata(iface_att["metadata"])
178
            interface.deactivate()
179 1
            name = 'kytos/topology.port.created'
180 1
            event = KytosEvent(name=name, content={
181 1
                                              'switch': switch_id,
182
                                              'port': interface.port_number,
183
                                              'port_description': {
184
                                                  'alias': interface.name,
185
                                                  'mac': interface.address,
186 1
                                                  'state': interface.state
187 1
                                                  }
188
                                              })
189 1
            self.controller.buffers.app.put(event)
190 1
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
    def load_topology(self):
198
        """Load network topology from DB."""
199
        topology = self.topo_controller.get_topology()
200
        switches = topology["topology"]["switches"]
201
        links = topology["topology"]["links"]
202
203 1
        failed_switches = {}
204
        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 1
            # pylint: disable=broad-except
209
            except Exception as err:
210
                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
            try:
217 1
                self._load_link(link_att)
218 1
            # 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 1
            name=name,
226
            content={
227 1
                'topology': self._get_topology(),
228 1
                'failed_switches': failed_switches,
229 1
                'failed_links': failed_links
230 1
            })
231 1
        self.controller.buffers.app.put(event)
232
233 1
    @rest('v3/')
234 1
    def get_topology(self, _request: Request) -> JSONResponse:
235 1
        """Return the latest known topology.
236
237 1
        This topology is updated when there are network events.
238 1
        """
239
        return JSONResponse(self._get_topology_dict())
240
241
    # Switch related methods
242
    @rest('v3/switches')
243
    def get_switches(self, _request: Request) -> JSONResponse:
244
        """Return a json with all the switches in the topology."""
245 1
        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
        dpid = request.path_params["dpid"]
251
        try:
252
            switch = self.controller.switches[dpid]
253 1
            self.topo_controller.enable_switch(dpid)
254
            switch.enable()
255
        except KeyError:
256 1
            raise HTTPException(404, detail="Switch not found")
257 1
258
        self.notify_switch_enabled(dpid)
259
        self.notify_topology_update()
260
        self.notify_switch_links_status(switch, "link enabled")
261 1
        return JSONResponse("Operation successful", status_code=201)
262 1
263
    @rest('v3/switches/{dpid}/disable', methods=['POST'])
264 1
    def disable_switch(self, request: Request) -> JSONResponse:
265 1
        """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
            switch.disable()
271 1
        except KeyError:
272 1
            raise HTTPException(404, detail="Switch not found")
273 1
274 1
        self.notify_switch_disabled(dpid)
275
        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 1
        """Get metadata from a switch."""
282 1
        dpid = request.path_params["dpid"]
283 1
        try:
284 1
            metadata = self.controller.switches[dpid].metadata
285
            return JSONResponse({"metadata": metadata})
286 1
        except KeyError:
287 1
            raise HTTPException(404, detail="Switch not found")
288 1
289 1
    @rest('v3/switches/{dpid}/metadata', methods=['POST'])
290
    def add_switch_metadata(self, request: Request) -> JSONResponse:
291 1
        """Add metadata to a switch."""
292 1
        dpid = request.path_params["dpid"]
293
        metadata = self._get_metadata(request)
294 1
        try:
295 1
            switch = self.controller.switches[dpid]
296
        except KeyError:
297 1
            raise HTTPException(404, detail="Switch not found")
298 1
299
        self.topo_controller.add_switch_metadata(dpid, metadata)
300 1
        switch.extend_metadata(metadata)
301 1
        self.notify_metadata_changes(switch, 'added')
302
        return JSONResponse("Operation successful", status_code=201)
303 1
304
    @rest('v3/switches/{dpid}/metadata/{key}', methods=['DELETE'])
305 1
    def delete_switch_metadata(self, request: Request) -> JSONResponse:
306 1
        """Delete metadata from a switch."""
307 1
        dpid = request.path_params["dpid"]
308 1
        key = request.path_params["key"]
309
        try:
310 1
            switch = self.controller.switches[dpid]
311 1
        except KeyError:
312 1
            raise HTTPException(404, detail="Switch not found")
313 1
314
        try:
315 1
            _ = switch.metadata[key]
316 1
        except KeyError:
317
            raise HTTPException(404, "Metadata not found")
318 1
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
        return JSONResponse("Operation successful")
323 1
324 1
    # Interface related methods
325
    @rest('v3/interfaces')
326
    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
                interfaces[interface_id] = interface
333
334 1
        return JSONResponse({'interfaces': interfaces})
335 1
336 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 1
        """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
        if dpid is None:
343 1
            dpid = ":".join(interface_enable_id.split(":")[:-1])
344
        try:
345 1
            switch = self.controller.switches[dpid]
346 1
        except KeyError:
347 1
            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 1
352 1
            try:
353 1
                interface = switch.interfaces[interface_number]
354 1
                self.topo_controller.enable_interface(interface.id)
355
                interface.enable()
356 1
                self.notify_interface_link_status(interface, "link enabled")
357 1
            except KeyError:
358
                msg = f"Switch {dpid} interface {interface_number} not found"
359 1
                raise HTTPException(404, detail=msg)
360 1
        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 1
        """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
        if dpid is None:
375 1
            dpid = ":".join(interface_disable_id.split(":")[:-1])
376 1
        try:
377 1
            switch = self.controller.switches[dpid]
378
        except KeyError:
379 1
            raise HTTPException(404, detail="Switch not found")
380 1
381 1
        if interface_disable_id:
382 1
            interface_number = int(interface_disable_id.split(":")[-1])
383 1
384 1
            try:
385
                interface = switch.interfaces[interface_number]
386 1
                self.topo_controller.disable_interface(interface.id)
387 1
                interface.disable()
388
                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 1
        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
        self.notify_topology_update()
398 1
        return JSONResponse("Operation successful")
399 1
400 1
    @rest('v3/interfaces/{interface_id}/metadata')
401 1
    def get_interface_metadata(self, request: Request) -> JSONResponse:
402 1
        """Get metadata from an interface."""
403 1
        interface_id = request.path_params["interface_id"]
404
        switch_id = ":".join(interface_id.split(":")[:-1])
405 1
        interface_number = int(interface_id.split(":")[-1])
406 1
        try:
407
            switch = self.controller.switches[switch_id]
408 1
        except KeyError:
409 1
            raise HTTPException(404, detail="Switch not found")
410 1
411 1
        try:
412 1
            interface = switch.interfaces[interface_number]
413 1
        except KeyError:
414
            raise HTTPException(404, detail="Interface not found")
415 1
416 1
        return JSONResponse({"metadata": interface.metadata})
417 1
418 1
    @rest('v3/interfaces/{interface_id}/metadata', methods=['POST'])
419
    def add_interface_metadata(self, request: Request) -> JSONResponse:
420 1
        """Add metadata to an interface."""
421
        interface_id = request.path_params["interface_id"]
422 1
        metadata = self._get_metadata(request)
423 1
        switch_id = ":".join(interface_id.split(":")[:-1])
424
        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 1
430 1
        try:
431 1
            interface = switch.interfaces[interface_number]
432
            self.topo_controller.add_interface_metadata(interface.id, metadata)
433 1
        except KeyError:
434 1
            raise HTTPException(404, detail="Interface not found")
435 1
436 1
        interface.extend_metadata(metadata)
437 1
        self.notify_metadata_changes(interface, 'added')
438
        return JSONResponse("Operation successful", status_code=201)
439 1
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
        switch_id = ":".join(interface_id.split(":")[:-1])
446 1
        try:
447 1
            interface_number = int(interface_id.split(":")[-1])
448
        except ValueError:
449 1
            detail = f"Invalid interface_id {interface_id}"
450 1
            raise HTTPException(400, detail=detail)
451 1
452 1
        try:
453
            switch = self.controller.switches[switch_id]
454 1
        except KeyError:
455 1
            raise HTTPException(404, detail="Switch not found")
456 1
457 1
        try:
458
            interface = switch.interfaces[interface_number]
459 1
        except KeyError:
460 1
            raise HTTPException(404, detail="Interface not found")
461 1
462 1
        try:
463
            _ = interface.metadata[key]
464 1
        except KeyError:
465 1
            raise HTTPException(404, detail="Metadata not found")
466 1
467 1
        self.topo_controller.delete_interface_metadata_key(interface.id, key)
468
        interface.remove_metadata(key)
469
        self.notify_metadata_changes(interface, 'removed')
470 1
        return JSONResponse("Operation successful")
471 1
472
    # Link related methods
473
    @rest('v3/links')
474
    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 1
        """
479 1
        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 1
        """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
                link.enable()
490
        except KeyError:
491
            raise HTTPException(404, detail="Link not found")
492 1
        self.notify_link_status_change(
493 1
            self.links[link_id],
494
            reason='link enabled'
495 1
        )
496 1
        self.notify_topology_update()
497
        return JSONResponse("Operation successful", status_code=201)
498 1
499 1
    @rest('v3/links/{link_id}/disable', methods=['POST'])
500 1
    def disable_link(self, request: Request) -> JSONResponse:
501 1
        """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
                self.topo_controller.disable_link(link_id)
507
                link.disable()
508
        except KeyError:
509 1
            raise HTTPException(404, detail="Link not found")
510 1
        self.notify_link_status_change(
511
            self.links[link_id],
512 1
            reason='link disabled'
513 1
        )
514
        self.notify_topology_update()
515 1
        return JSONResponse("Operation successful", status_code=201)
516 1
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
            return JSONResponse({"metadata": self.links[link_id].metadata})
523 1
        except KeyError:
524 1
            raise HTTPException(404, detail="Link not found")
525 1
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
        except KeyError:
534 1
            raise HTTPException(404, detail="Link not found")
535 1
536
        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 1
541
    @rest('v3/links/{link_id}/metadata/{key}', methods=['DELETE'])
542 1
    def delete_link_metadata(self, request: Request) -> JSONResponse:
543 1
        """Delete metadata from a link."""
544 1
        link_id = request.path_params["link_id"]
545 1
        key = request.path_params["key"]
546
        try:
547 1
            link = self.links[link_id]
548 1
        except KeyError:
549 1
            raise HTTPException(404, detail="Link not found")
550 1
551
        try:
552 1
            _ = link.metadata[key]
553
        except KeyError:
554 1
            raise HTTPException(404, detail="Metadata not found")
555 1
556
        self.topo_controller.delete_link_metadata_key(link.id, key)
557 1
        link.remove_metadata(key)
558
        self.notify_metadata_changes(link, 'removed')
559 1
        return JSONResponse("Operation successful")
560 1
561
    def notify_current_topology(self) -> None:
562
        """Notify current topology."""
563
        name = "kytos/topology.current"
564 1
        event = KytosEvent(name=name, content={"topology":
565 1
                                               self._get_topology()})
566
        self.controller.buffers.app.put(event)
567
568
    @listen_to("kytos/topology.get")
569
    def on_get_topology(self, _event) -> None:
570
        """Handle kytos/topology.get."""
571
        self.notify_current_topology()
572
573
    @listen_to("kytos/.*.liveness.(up|down)")
574
    def on_link_liveness_status(self, event) -> None:
575
        """Handle link liveness up|down status event."""
576 1
        link = Link(event.content["interface_a"], event.content["interface_b"])
577
        try:
578 1
            link = self.links[link.id]
579 1
        except KeyError:
580 1
            log.error(f"Link id {link.id} not found, {link}")
581 1
            return
582 1
        liveness_status = event.name.split(".")[-1]
583 1
        self.handle_link_liveness_status(self.links[link.id], liveness_status)
584 1
585 1
    def handle_link_liveness_status(self, link, liveness_status) -> None:
586 1
        """Handle link liveness."""
587
        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
        link.extend_metadata(metadata)
591
        self.notify_topology_update()
592
        if link.status == EntityStatus.UP and liveness_status == "up":
593
            self.notify_link_status_change(link, reason="liveness_up")
594 1
        if link.status == EntityStatus.DOWN and liveness_status == "down":
595
            self.notify_link_status_change(link, reason="liveness_down")
596 1
597 1
    @listen_to("kytos/.*.liveness.disabled")
598 1
    def on_link_liveness_disabled(self, event) -> None:
599 1
        """Handle link liveness disabled event."""
600 1
        interfaces = event.content["interfaces"]
601
        self.handle_link_liveness_disabled(interfaces)
602
603
    def get_links_from_interfaces(self, interfaces) -> dict:
604 1
        """Get links from interfaces."""
605 1
        links_found = {}
606
        with self._links_lock:
607 1
            for interface in interfaces:
608
                for link in self.links.values():
609 1
                    if any((
610
                        interface.id == link.endpoint_a.id,
611 1
                        interface.id == link.endpoint_b.id,
612 1
                    )):
613 1
                        links_found[link.id] = link
614 1
        return links_found
615 1
616 1
    def handle_link_liveness_disabled(self, interfaces) -> None:
617 1
        """Handle link liveness disabled."""
618 1
        log.info(f"Link liveness disabled interfaces: {interfaces}")
619 1
620
        key = "liveness_status"
621 1
        links = self.get_links_from_interfaces(interfaces)
622 1
        for link in links.values():
623
            link.remove_metadata(key)
624
        link_ids = list(links.keys())
625
        self.topo_controller.bulk_delete_link_metadata_key(link_ids, key)
626
        self.notify_topology_update()
627 1
        for link in links.values():
628
            self.notify_link_status_change(link, reason="liveness_disabled")
629 1
630
    @listen_to("kytos/.*.link_available_tags")
631 1
    def on_link_available_tags(self, event):
632 1
        """Handle on_link_available_tags."""
633 1
        with self._links_lock:
634 1
            self.handle_on_link_available_tags(event.content.get("link"))
635 1
636
    def handle_on_link_available_tags(self, link):
637
        """Handle on_link_available_tags."""
638
        if link.id not in self.links:
639
            return
640
        endpoint_a = self.links[link.id].endpoint_a
641 1
        endpoint_b = self.links[link.id].endpoint_b
642
        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
        self.topo_controller.bulk_upsert_interface_details(ids_details)
651
652 1
    @listen_to('.*.switch.(new|reconnected)')
653
    def on_new_switch(self, event):
654 1
        """Create a new Device on the Topology.
655 1
656 1
        Handle the event of a new created switch and update the topology with
657 1
        this new device. Also notify if the switch is enabled.
658 1
        """
659 1
        self.handle_new_switch(event)
660 1
661
    def handle_new_switch(self, event):
662 1
        """Create a new Device on the Topology."""
663 1
        switch = event.content['switch']
664
        switch.activate()
665
        self.topo_controller.upsert_switch(switch.id, switch.as_dict())
666
        log.debug('Switch %s added to the Topology.', switch.id)
667
        self.notify_topology_update()
668
        if switch.is_enabled():
669
            self.notify_switch_enabled(switch.id)
670
671 1
    @listen_to('.*.connection.lost')
672
    def on_connection_lost(self, event):
673 1
        """Remove a Device from the topology.
674 1
675 1
        Remove the disconnected Device and every link that has one of its
676 1
        interfaces.
677 1
        """
678 1
        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
            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 1
689 1
    def handle_interfaces_created(self, event):
690 1
        """Update the topology based on the interfaces created."""
691
        interfaces = event.content["interfaces"]
692 1
        if not interfaces:
693
            return
694
        switch = interfaces[0].switch
695
        self.topo_controller.upsert_switch(switch.id, switch.as_dict())
696
        name = "kytos/topology.switch.interface.created"
697
        for interface in interfaces:
698 1
            event = KytosEvent(name=name, content={'interface': interface})
699 1
            self.controller.buffers.app.put(event)
700 1
701 1
    def handle_interface_created(self, event):
702
        """Update the topology based on an interface created event.
703 1
704 1
        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
        interface = event.content['interface']
708
        if not interface.is_active():
709
            return
710
        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 1
        created event it can belong to an existign link.
718
        """
719
        self.handle_interface_created(event)
720
721
    @listen_to('.*.switch.interfaces.created')
722 1
    def on_interfaces_created(self, event):
723 1
        """Update the topology based on a list of created interfaces."""
724 1
        self.handle_interfaces_created(event)
725 1
726
    def handle_interface_down(self, event):
727 1
        """Update the topology based on a Port Modify event.
728 1
729
        The event notifies that an interface was changed to 'down'.
730
        """
731
        interface = event.content['interface']
732 1
        interface.deactivate()
733
        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
    def handle_interface_deleted(self, event):
742
        """Update the topology based on a Port Delete event."""
743
        self.handle_interface_down(event)
744
745 1
    @listen_to('.*.switch.interface.link_up')
746
    def on_interface_link_up(self, event):
747 1
        """Update the topology based on a Port Modify event.
748
749 1
        The event notifies that an interface's link was changed to 'up'.
750 1
        """
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 1
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 1
763 1
    def handle_switch_maintenance_end(self, event):
764 1
        """Handle the end of the maintenance of a switch."""
765 1
        switches = event.content['switches']
766 1
        for switch_id in switches:
767
            try:
768 1
                switch = self.controller.switches[switch_id]
769
            except KeyError:
770 1
                continue
771 1
            switch.enable()
772
            switch.activate()
773
            for interface in switch.interfaces.values():
774
                interface.enable()
775
                self.handle_link_up(interface)
776
777 1
    def link_status_hook_link_up_timer(self, link) -> Optional[EntityStatus]:
778 1
        """Link status hook link up timer."""
779
        tnow = time.time()
780 1
        if (
781
            link.is_active()
782
            and link.is_enabled()
783
            and "last_status_change" in link.metadata
784 1
            and tnow - link.metadata['last_status_change'] < self.link_up_timer
785 1
        ):
786
            return EntityStatus.DOWN
787 1
        return None
788 1
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
        time.sleep(self.link_up_timer)
794 1
        if link.status != EntityStatus.UP:
795 1
            return
796 1
        with self._links_notify_lock[link.id]:
797 1
            notified_at = link.get_metadata("notified_up_at")
798 1
            if (
799 1
                notified_at
800
                and (now() - notified_at.replace(tzinfo=timezone.utc)).seconds
801 1
                < 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
            self.notify_link_status_change(link, reason)
809 1
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
        if not link:
817
            return
818
        if link.endpoint_a == interface:
819 1
            other_interface = link.endpoint_b
820 1
        else:
821 1
            other_interface = link.endpoint_a
822 1
        if other_interface.is_active() is False:
823
            return
824 1
        metadata = {
825 1
            'last_status_change': time.time(),
826
            'last_status_is_active': True
827
        }
828
        link.extend_metadata(metadata)
829
        link.activate()
830
        self.topo_controller.activate_link(link.id, **metadata)
831
        self.notify_link_up_if_status(link, "link up")
832
833 1
    @listen_to('.*.switch.interface.link_down')
834
    def on_interface_link_down(self, event):
835 1
        """Update the topology based on a Port Modify event.
836
837 1
        The event notifies that an interface's link was changed to 'down'.
838 1
        """
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 1
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 1
851 1
    def handle_switch_maintenance_start(self, event):
852 1
        """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
                switch = self.controller.switches[switch_id]
857 1
            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
    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
            link.deactivate()
871 1
            last_status_change = time.time()
872 1
            last_status_is_active = False
873 1
            metadata = {
874 1
                "last_status_change": last_status_change,
875 1
                "last_status_is_active": last_status_is_active,
876 1
            }
877 1
            link.extend_metadata(metadata)
878
            self.topo_controller.deactivate_link(link.id, last_status_change,
879
                                                 last_status_is_active)
880
            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
                last_status_change = time.time()
885
                last_status_is_active = False
886 1
                metadata = {
887 1
                    "last_status_change": last_status_change,
888 1
                    "last_status_is_active": last_status_is_active,
889 1
                }
890
                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
                    self.notify_link_status_change(link, reason='link down')
896 1
        interface.deactivate()
897
        self.topo_controller.deactivate_interface(interface.id)
898 1
        self.notify_topology_update()
899 1
900
    @listen_to('.*.interface.is.nni')
901 1
    def on_add_links(self, event):
902 1
        """Update the topology with links related to the NNI interfaces."""
903 1
        self.add_links(event)
904
905 1
    def add_links(self, event):
906 1
        """Update the topology with links related to the NNI interfaces."""
907
        interface_a = event.content['interface_a']
908 1
        interface_b = event.content['interface_b']
909 1
910
        try:
911 1
            with self._links_lock:
912 1
                link, created = self._get_link_or_create(interface_a,
913
                                                         interface_b)
914
                interface_a.update_link(link)
915
                interface_b.update_link(link)
916
917
                link.endpoint_a = interface_a
918 1
                link.endpoint_b = interface_b
919
920
                interface_a.nni = True
921 1
                interface_b.nni = True
922 1
923
        except KytosLinkCreationError as err:
924
            log.error(f'Error creating link: {err}.')
925 1
            return
926
927
        if not created:
928
            return
929 1
930 1
        self.notify_topology_update()
931 1
        if not link.is_active():
932
            return
933 1
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
        self.notify_link_up_if_status(link, "link up")
941
942
    @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 1
        self.handle_lldp_status_updated(event)
946
947 1
    @listen_to(".*.topo_controller.upsert_switch")
948
    def on_topo_controller_upsert_switch(self, event) -> None:
949 1
        """Listen to topo_controller_upsert_switch."""
950 1
        self.handle_topo_controller_upsert_switch(event.content["switch"])
951 1
952 1
    def handle_topo_controller_upsert_switch(self, switch) -> Optional[dict]:
953 1
        """Handle topo_controller_upsert_switch."""
954 1
        return self.topo_controller.upsert_switch(switch.id, switch.as_dict())
955 1
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
            dpid = ":".join(interface_id.split(":")[:-1])
963 1
            switch = self.controller.get_switch_by_dpid(dpid)
964
            if switch:
965 1
                switches.add(switch)
966 1
967 1
        name = "kytos/topology.topo_controller.upsert_switch"
968
        for switch in switches:
969 1
            event = KytosEvent(name=name, content={"switch": switch})
970
            self.controller.buffers.app.put(event)
971 1
972 1
    def notify_switch_enabled(self, dpid):
973 1
        """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 1
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
            for link in self.links.values():
982 1
                if switch in (link.endpoint_a.switch, link.endpoint_b.switch):
983
                    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
                        self.controller.buffers.app.put(event)
988 1
                    else:
989
                        self.notify_link_status_change(link, reason)
990 1
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
        event = KytosEvent(name=name, content={'dpid': dpid})
995 1
        self.controller.buffers.app.put(event)
996
997
    def notify_topology_update(self):
998 1
        """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 1
                                               self._get_topology()})
1002 1
        self.controller.buffers.app.put(event)
1003 1
1004 1
    def notify_interface_link_status(self, interface, reason):
1005
        """Send an event to notify the status of a link from
1006 1
        an interface."""
1007
        link = self._get_link_from_interface(interface)
1008 1
        if link:
1009
            if reason == "link enabled":
1010 1
                name = 'kytos/topology.notify_link_up_if_status'
1011 1
                content = {'reason': reason, "link": link}
1012
                event = KytosEvent(name=name, content=content)
1013
                self.controller.buffers.app.put(event)
1014 1
            else:
1015 1
                self.notify_link_status_change(link, reason)
1016
1017
    def notify_link_status_change(self, link, reason='not given'):
1018
        """Send an event to notify about a status change on a link."""
1019
        name = 'kytos/topology.'
1020
        if link.status == EntityStatus.UP:
1021 1
            status = 'link_up'
1022
        else:
1023 1
            status = 'link_down'
1024
        event = KytosEvent(
1025 1
            name=name+status,
1026 1
            content={
1027 1
                'link': link,
1028 1
                'reason': reason
1029 1
            })
1030 1
        self.controller.buffers.app.put(event)
1031 1
1032 1
    def notify_metadata_changes(self, obj, action):
1033 1
        """Send an event to notify about metadata changes."""
1034
        if isinstance(obj, Switch):
1035 1
            entity = 'switch'
1036
            entities = 'switches'
1037
        elif isinstance(obj, Interface):
1038
            entity = 'interface'
1039 1
            entities = 'interfaces'
1040 1
        elif isinstance(obj, Link):
1041 1
            entity = 'link'
1042 1
            entities = 'links'
1043 1
        else:
1044
            raise ValueError(
1045 1
                'Invalid object, supported: Switch, Interface, Link'
1046 1
            )
1047
1048
        name = f'kytos/topology.{entities}.metadata.{action}'
1049
        content = {entity: obj, 'metadata': obj.metadata.copy()}
1050
        event = KytosEvent(name=name, content=content)
1051
        self.controller.buffers.app.put(event)
1052 1
        log.debug(f'Metadata from {obj.id} was {action}.')
1053 1
1054
    @listen_to('kytos/topology.notify_link_up_if_status')
1055
    def on_notify_link_up_if_status(self, event):
1056
        """Tries to notify link up and topology changes"""
1057 1
        link = event.content["link"]
1058
        reason = event.content["reason"]
1059 1
        self.notify_link_up_if_status(link, reason)
1060 1
1061 1
    @listen_to('.*.switch.port.created')
1062
    def on_notify_port_created(self, event):
1063 1
        """Notify when a port is created."""
1064 1
        self.notify_port_created(event)
1065
1066
    def notify_port_created(self, event):
1067 1
        """Notify when a port is created."""
1068
        name = 'kytos/topology.port.created'
1069 1
        event = KytosEvent(name=name, content=event.content)
1070 1
        self.controller.buffers.app.put(event)
1071 1
1072
    @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 1
            return
1078 1
        for interface_details in interfaces_details:
1079
            available_vlans = interface_details["available_vlans"]
1080 1
            if not available_vlans:
1081 1
                continue
1082
            log.debug(f"Interface id {interface_details['id']} loading "
1083
                      f"{len(interface_details['available_vlans'])} "
1084
                      "available tags")
1085
            port_number = int(interface_details["id"].split(":")[-1])
1086 1
            interface = switch.interfaces[port_number]
1087
            interface.set_available_tags(interface_details['available_vlans'])
1088 1
1089 1
    @listen_to('kytos/maintenance.start_link')
1090 1
    def on_link_maintenance_start(self, event):
1091 1
        """Deals with the start of links maintenance."""
1092 1
        with self._links_lock:
1093 1
            self.handle_link_maintenance_start(event)
1094 1
1095 1
    def handle_link_maintenance_start(self, event):
1096 1
        """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
            notify_links.append(link)
1105 1
        for link in notify_links:
1106 1
            link.disable()
1107
            link.deactivate()
1108
            link.endpoint_a.deactivate()
1109
            link.endpoint_b.deactivate()
1110
            link.endpoint_a.disable()
1111 1
            link.endpoint_b.disable()
1112
            self.notify_link_status_change(link, reason='maintenance')
1113 1
1114 1
    @listen_to('kytos/maintenance.end_link')
1115 1
    def on_link_maintenance_end(self, event):
1116 1
        """Deals with the end of links maintenance."""
1117 1
        with self._links_lock:
1118 1
            self.handle_link_maintenance_end(event)
1119 1
1120 1
    def handle_link_maintenance_end(self, event):
1121 1
        """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
            notify_links.append(link)
1130
        for link in notify_links:
1131
            link.enable()
1132
            link.activate()
1133
            link.endpoint_a.activate()
1134
            link.endpoint_b.activate()
1135
            link.endpoint_a.enable()
1136
            link.endpoint_b.enable()
1137
            self.notify_link_status_change(link, reason='maintenance')
1138