Test Failed
Pull Request — master (#160)
by
unknown
02:49
created

main.py (2 issues)

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