Passed
Pull Request — master (#255)
by Aldo
04:25
created
1
"""Main module of kytos/topology Kytos Network Application.
2
3
Manage the network topology
4
"""
5
# pylint: disable=wrong-import-order
6 1
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
import httpx
14 1
import tenacity
15 1
from tenacity import (retry_if_exception_type, stop_after_attempt,
16
                      wait_combine, wait_fixed, wait_random)
17
18 1
from kytos.core import KytosEvent, KytosNApp, log, rest
19 1
from kytos.core.common import EntityStatus, GenericEntity
20 1
from kytos.core.exceptions import (KytosInvalidTagRanges,
21
                                   KytosLinkCreationError, KytosTagError)
22 1
from kytos.core.helpers import listen_to, load_spec, now, validate_openapi
23 1
from kytos.core.interface import Interface
24 1
from kytos.core.link import Link
25 1
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
26
                                 content_type_json_or_415, get_json_or_400)
27 1
from kytos.core.retry import before_sleep
28 1
from kytos.core.switch import Switch
29 1
from kytos.core.tag_ranges import get_tag_ranges
30 1
from napps.kytos.topology import settings
31
32 1
from .controllers import TopoController
33 1
from .exceptions import RestoreError
34 1
from .models import Topology
35
36 1
DEFAULT_LINK_UP_TIMER = 10
37
38
39 1
class Main(KytosNApp):  # pylint: disable=too-many-public-methods
40
    """Main class of kytos/topology NApp.
41
42
    This class is the entry point for this napp.
43
    """
44
45 1
    spec = load_spec(pathlib.Path(__file__).parent / "openapi.yml")
46
47 1
    def setup(self):
48
        """Initialize the NApp's links list."""
49 1
        self.links: dict[str, Link] = {}
50 1
        self.intf_available_tags = {}
51 1
        self.link_up_timer = getattr(settings, 'LINK_UP_TIMER',
52
                                     DEFAULT_LINK_UP_TIMER)
53
54 1
        self._links_lock = Lock()
55
        # to keep track of potential unorded scheduled interface events
56 1
        self._intfs_lock = defaultdict(Lock)
57 1
        self._intfs_updated_at = {}
58 1
        self._intfs_tags_updated_at = {}
59 1
        self.link_up = set()
60 1
        self.link_status_lock = Lock()
61 1
        self._switch_lock = defaultdict(Lock)
62 1
        self.topo_controller = self.get_topo_controller()
63
64
        # Track when we last received a link up, that resulted in
65
        # activating a deactivated link.
66 1
        self.link_status_change = defaultdict[str, dict](dict)
67 1
        Link.register_status_func(f"{self.napp_id}_link_up_timer",
68
                                  self.link_status_hook_link_up_timer)
69 1
        Link.register_status_reason_func(f"{self.napp_id}_mismatched_reason",
70
                                         self.detect_mismatched_link)
71 1
        Link.register_status_func(f"{self.napp_id}_mismatched_status",
72
                                  self.link_status_mismatched)
73 1
        self.topo_controller.bootstrap_indexes()
74 1
        self.load_topology()
75
76 1
    @staticmethod
77 1
    def get_topo_controller() -> TopoController:
78
        """Get TopoController."""
79
        return TopoController()
80
81 1
    def execute(self):
82
        """Execute once when the napp is running."""
83
        pass
84
85 1
    def shutdown(self):
86
        """Do nothing."""
87
        log.info('NApp kytos/topology shutting down.')
88
89 1
    def _get_metadata(self, request: Request) -> dict:
90
        """Return a JSON with metadata."""
91 1
        content_type_json_or_415(request)
92 1
        metadata = get_json_or_400(request, self.controller.loop)
93 1
        if not isinstance(metadata, dict):
94 1
            raise HTTPException(400, "Invalid metadata value: {metadata}")
95 1
        return metadata
96
97 1
    def _get_link_or_create(
98
        self,
99
        endpoint_a: Interface,
100
        endpoint_b: Interface
101
    ):
102
        """Get an existing link or create a new one.
103
104
        Returns:
105
            Tuple(Link, bool): Link and a boolean whether it has been created.
106
        """
107 1
        new_link = Link(endpoint_a, endpoint_b)
108
109
        # If link is an old link but mismatched, then treat it as a new link
110 1
        if (new_link.id in self.links
111
                and not self.detect_mismatched_link(new_link)):
112 1
            return (self.links[new_link.id], False)
113
114
        # Check if any interface already has a link
115
        # This old_link is a leftover link that needs to be removed
116
        # The other endpoint of the link is the leftover interface
117 1
        if endpoint_a.link and endpoint_a.link != new_link:
118 1
            old_link = endpoint_a.link
119 1
            leftover_interface = (old_link.endpoint_a
120
                                  if old_link.endpoint_a != endpoint_a
121
                                  else old_link.endpoint_b)
122 1
            log.warning(f"Leftover mismatched link {endpoint_a.link} "
123
                        f"in interface {leftover_interface}")
124
125 1
        if endpoint_b.link and endpoint_b.link != new_link:
126 1
            old_link = endpoint_b.link
127 1
            leftover_interface = (old_link.endpoint_b
128
                                  if old_link.endpoint_b != endpoint_b
129
                                  else old_link.endpoint_a)
130 1
            log.warning(f"Leftover mismatched link {endpoint_b.link} "
131
                        f"in interface {leftover_interface}")
132
133 1
        if new_link.id not in self.links:
134 1
            self.links[new_link.id] = new_link
135 1
        return (self.links[new_link.id], True)
136
137 1
    def _get_switches_dict(self):
138
        """Return a dictionary with the known switches."""
139 1
        switches = {'switches': {}}
140 1
        for idx, switch in enumerate(self.controller.switches.copy().values()):
141 1
            switch_data = switch.as_dict()
142 1
            if not all(key in switch_data['metadata']
143
                       for key in ('lat', 'lng')):
144
                # Switches are initialized somewhere in the ocean
145
                switch_data['metadata']['lat'] = str(0.0)
146
                switch_data['metadata']['lng'] = str(-30.0+idx*10.0)
147 1
            switches['switches'][switch.id] = switch_data
148 1
        return switches
149
150 1
    def _get_links_dict(self):
151
        """Return a dictionary with the known links."""
152 1
        return {'links': {link.id: link.as_dict() for link in
153
                          self.links.copy().values()}}
154
155 1
    def _get_topology_dict(self):
156
        """Return a dictionary with the known topology."""
157 1
        return {'topology': {**self._get_switches_dict(),
158
                             **self._get_links_dict()}}
159
160 1
    def _get_topology(self):
161
        """Return an object representing the topology."""
162 1
        return Topology(self.controller.switches.copy(), self.links.copy())
163
164 1
    def _get_link_from_interface(self, interface: Interface):
165
        """Return the link of the interface, or None if it does not exist."""
166 1
        for link in list(self.links.values()):
167 1
            if interface in (link.endpoint_a, link.endpoint_b):
168 1
                return link
169 1
        return None
170
171 1
    def _load_link(self, link_att):
172 1
        endpoint_a = link_att['endpoint_a']['id']
173 1
        endpoint_b = link_att['endpoint_b']['id']
174 1
        link_str = link_att['id']
175 1
        log.info(f"Loading link: {link_str}")
176 1
        interface_a = self.controller.get_interface_by_id(endpoint_a)
177 1
        interface_b = self.controller.get_interface_by_id(endpoint_b)
178
179 1
        error = f"Fail to load endpoints for link {link_str}. "
180 1
        if not interface_a:
181 1
            raise RestoreError(f"{error}, endpoint_a {endpoint_a} not found")
182 1
        if not interface_b:
183
            raise RestoreError(f"{error}, endpoint_b {endpoint_b} not found")
184
185 1
        with self._links_lock:
186 1
            link, _ = self._get_link_or_create(interface_a, interface_b)
187
188 1
            interface_a.update_link(link)
189 1
            interface_b.update_link(link)
190 1
            interface_a.nni = True
191 1
            interface_b.nni = True
192
193 1
            if link_att['enabled']:
194 1
                link.enable()
195
            else:
196 1
                link.disable()
197
198 1
        link.extend_metadata(link_att["metadata"])
199
200 1
    def _load_switch(self, switch_id, switch_att):
201 1
        log.info(f'Loading switch dpid: {switch_id}')
202 1
        switch = self.controller.get_switch_or_create(switch_id)
203 1
        if switch_att['enabled']:
204 1
            switch.enable()
205
        else:
206 1
            switch.disable()
207 1
        switch.description['manufacturer'] = switch_att.get('manufacturer', '')
208 1
        switch.description['hardware'] = switch_att.get('hardware', '')
209 1
        switch.description['software'] = switch_att.get('software')
210 1
        switch.description['serial'] = switch_att.get('serial', '')
211 1
        switch.description['data_path'] = switch_att.get('data_path', '')
212 1
        switch.extend_metadata(switch_att["metadata"])
213
214 1
        for iface_id, iface_att in switch_att.get('interfaces', {}).items():
215 1
            log.info(f'Loading interface iface_id={iface_id}')
216 1
            interface = switch.update_or_create_interface(
217
                            port_no=iface_att['port_number'],
218
                            name=iface_att['name'],
219
                            address=iface_att.get('mac', None),
220
                            speed=iface_att.get('speed', None))
221 1
            if iface_att['enabled']:
222 1
                interface.enable()
223
            else:
224 1
                interface.disable()
225 1
            interface.lldp = iface_att['lldp']
226 1
            interface.extend_metadata(iface_att["metadata"])
227 1
            interface.deactivate()
228 1
            name = 'kytos/topology.port.created'
229 1
            event = KytosEvent(name=name, content={
230
                                              'switch': switch_id,
231
                                              'port': interface.port_number,
232
                                              'port_description': {
233
                                                  'alias': interface.name,
234
                                                  'mac': interface.address,
235
                                                  'state': interface.state
236
                                                  }
237
                                              })
238 1
            self.controller.buffers.app.put(event, timeout=1)
239
240 1
        intf_ids = [v["id"] for v in switch_att.get("interfaces", {}).values()]
241 1
        intf_details = self.topo_controller.get_interfaces_details(intf_ids)
242 1
        with self._links_lock:
243 1
            self.load_interfaces_tags_values(switch, intf_details)
244
245
    # pylint: disable=attribute-defined-outside-init
246 1
    def load_topology(self):
247
        """Load network topology from DB."""
248 1
        topology = self.topo_controller.get_topology()
249 1
        switches = topology["topology"]["switches"]
250 1
        links = topology["topology"]["links"]
251
252 1
        failed_switches = {}
253 1
        log.debug(f"_load_network_status switches={switches}")
254 1
        for switch_id, switch_att in switches.items():
255 1
            try:
256 1
                self._load_switch(switch_id, switch_att)
257 1
            except (KeyError, AttributeError, TypeError) as err:
258 1
                failed_switches[switch_id] = err
259 1
                log.error(f'Error loading switch: {err}')
260
261 1
        failed_links = {}
262 1
        log.debug(f"_load_network_status links={links}")
263 1
        for link_id, link_att in links.items():
264 1
            try:
265 1
                self._load_link(link_att)
266 1
            except (KeyError, AttributeError, TypeError) as err:
267 1
                failed_links[link_id] = err
268 1
                log.error(f'Error loading link {link_id}: {err}')
269
270 1
        name = 'kytos/topology.topology_loaded'
271 1
        event = KytosEvent(
272
            name=name,
273
            content={
274
                'topology': self._get_topology(),
275
                'failed_switches': failed_switches,
276
                'failed_links': failed_links
277
            })
278 1
        self.controller.buffers.app.put(event, timeout=1)
279
280 1
    @rest('v3/')
281 1
    def get_topology(self, _request: Request) -> JSONResponse:
282
        """Return the latest known topology.
283
284
        This topology is updated when there are network events.
285
        """
286 1
        return JSONResponse(self._get_topology_dict())
287
288
    # Switch related methods
289 1
    @rest('v3/switches')
290 1
    def get_switches(self, _request: Request) -> JSONResponse:
291
        """Return a json with all the switches in the topology."""
292
        return JSONResponse(self._get_switches_dict())
293
294 1
    @rest('v3/switches/{dpid}/enable', methods=['POST'])
295 1
    def enable_switch(self, request: Request) -> JSONResponse:
296
        """Administratively enable a switch in the topology."""
297 1
        dpid = request.path_params["dpid"]
298 1
        try:
299 1
            switch = self.controller.switches[dpid]
300 1
            self.topo_controller.enable_switch(dpid)
301 1
            switch.enable()
302 1
        except KeyError:
303 1
            raise HTTPException(404, detail="Switch not found")
304
305 1
        self.notify_topology_update()
306 1
        self.notify_switch_enabled(dpid)
307 1
        self.notify_switch_links_status(switch, "link enabled")
308 1
        return JSONResponse("Operation successful", status_code=201)
309
310 1
    @rest('v3/switches/{dpid}/disable', methods=['POST'])
311 1
    def disable_switch(self, request: Request) -> JSONResponse:
312
        """Administratively disable a switch in the topology."""
313 1
        dpid = request.path_params["dpid"]
314 1
        try:
315 1
            switch = self.controller.switches[dpid]
316 1
            link_ids = set()
317 1
            for _, interface in switch.interfaces.copy().items():
318 1
                if (interface.link and interface.link.is_enabled()):
319 1
                    link_ids.add(interface.link.id)
320 1
                    interface.link.disable()
321 1
                    self.notify_link_enabled_state(interface.link, "disabled")
322 1
            self.topo_controller.bulk_disable_links(link_ids)
323 1
            self.topo_controller.disable_switch(dpid)
324 1
            switch.disable()
325 1
        except KeyError:
326 1
            raise HTTPException(404, detail="Switch not found")
327
328 1
        self.notify_topology_update()
329 1
        self.notify_switch_disabled(dpid)
330 1
        self.notify_switch_links_status(switch, "link disabled")
331 1
        return JSONResponse("Operation successful", status_code=201)
332
333 1
    @rest('v3/switches/{dpid}', methods=['DELETE'])
334 1
    def delete_switch(self, request: Request) -> JSONResponse:
335
        """Delete a switch.
336
337
        Requirements:
338
            - There should not be installed flows related to switch.
339
            - The switch should be disabled.
340
            - All tags from switch interfaces should be available.
341
            - The switch should not have links.
342
        """
343 1
        dpid = request.path_params["dpid"]
344 1
        try:
345 1
            switch: Switch = self.controller.switches[dpid]
346 1
            with self._switch_lock[dpid]:
347 1
                if switch.status != EntityStatus.DISABLED:
348 1
                    raise HTTPException(
349
                        409, detail="Switch should be disabled."
350
                    )
351 1
                for intf_id, interface in switch.interfaces.copy().items():
352 1
                    if not interface.all_tags_available():
353 1
                        detail = f"Interface {intf_id} vlans are being used."\
354
                                 " Delete any service using vlans."
355 1
                        raise HTTPException(409, detail=detail)
356 1
                with self._links_lock:
357 1
                    for link_id, link in self.links.items():
358 1
                        if (dpid in
359
                                (link.endpoint_a.switch.dpid,
360
                                 link.endpoint_b.switch.dpid)):
361 1
                            raise HTTPException(
362
                                409, detail=f"Switch should not have links. "
363
                                            f"Link found {link_id}."
364
                            )
365 1
                try:
366 1
                    flows = self.get_flows_by_switch(dpid)
367
                except tenacity.RetryError as err:
368
                    detail = "Error while getting flows: "\
369
                             f"{err.last_attempt.exception()}."
370
                    raise HTTPException(409, detail=detail)
371 1
                if flows:
372
                    raise HTTPException(409, detail="Switch has flows. Verify"
373
                                                    " if a switch is used.")
374 1
                switch = self.controller.switches.pop(dpid)
375 1
                self.topo_controller.delete_switch_data(dpid)
376 1
        except KeyError:
377 1
            raise HTTPException(404, detail="Switch not found.")
378 1
        name = 'kytos/topology.switch.deleted'
379 1
        event = KytosEvent(name=name, content={'switch': switch})
380 1
        self.controller.buffers.app.put(event)
381 1
        self.notify_topology_update()
382 1
        return JSONResponse("Operation successful")
383
384 1
    @rest('v3/switches/{dpid}/metadata')
385 1
    def get_switch_metadata(self, request: Request) -> JSONResponse:
386
        """Get metadata from a switch."""
387 1
        dpid = request.path_params["dpid"]
388 1
        try:
389 1
            metadata = self.controller.switches[dpid].metadata
390 1
            return JSONResponse({"metadata": metadata})
391 1
        except KeyError:
392 1
            raise HTTPException(404, detail="Switch not found")
393
394 1
    @rest('v3/switches/{dpid}/metadata', methods=['POST'])
395 1
    def add_switch_metadata(self, request: Request) -> JSONResponse:
396
        """Add metadata to a switch."""
397 1
        dpid = request.path_params["dpid"]
398 1
        metadata = self._get_metadata(request)
399 1
        try:
400 1
            switch = self.controller.switches[dpid]
401 1
        except KeyError:
402 1
            raise HTTPException(404, detail="Switch not found")
403
404 1
        self.topo_controller.add_switch_metadata(dpid, metadata)
405 1
        switch.extend_metadata(metadata)
406 1
        self.notify_metadata_changes(switch, 'added')
407 1
        return JSONResponse("Operation successful", status_code=201)
408
409 1
    @rest('v3/switches/{dpid}/metadata/{key}', methods=['DELETE'])
410 1
    def delete_switch_metadata(self, request: Request) -> JSONResponse:
411
        """Delete metadata from a switch."""
412 1
        dpid = request.path_params["dpid"]
413 1
        key = request.path_params["key"]
414 1
        try:
415 1
            switch = self.controller.switches[dpid]
416 1
        except KeyError:
417 1
            raise HTTPException(404, detail="Switch not found")
418
419 1
        try:
420 1
            _ = switch.metadata[key]
421 1
        except KeyError:
422 1
            raise HTTPException(404, "Metadata not found")
423
424 1
        self.topo_controller.delete_switch_metadata_key(dpid, key)
425 1
        switch.remove_metadata(key)
426 1
        self.notify_metadata_changes(switch, 'removed')
427 1
        return JSONResponse("Operation successful")
428
429
    # Interface related methods
430 1
    @rest('v3/interfaces')
431 1
    def get_interfaces(self, _request: Request) -> JSONResponse:
432
        """Return a json with all the interfaces in the topology."""
433 1
        interfaces = {}
434 1
        switches = self._get_switches_dict()
435 1
        for switch in switches['switches'].values():
436 1
            for interface_id, interface in switch['interfaces'].items():
437 1
                interfaces[interface_id] = interface
438
439 1
        return JSONResponse({'interfaces': interfaces})
440
441 1
    @rest('v3/interfaces/switch/{dpid}/enable', methods=['POST'])
442 1
    @rest('v3/interfaces/{interface_enable_id}/enable', methods=['POST'])
443 1
    def enable_interface(self, request: Request) -> JSONResponse:
444
        """Administratively enable interfaces in the topology."""
445 1
        interface_enable_id = request.path_params.get("interface_enable_id")
446 1
        dpid = request.path_params.get("dpid")
447 1
        if dpid is None:
448 1
            dpid = ":".join(interface_enable_id.split(":")[:-1])
449 1
        try:
450 1
            switch = self.controller.switches[dpid]
451 1
            if not switch.is_enabled():
452 1
                raise HTTPException(409, detail="Enable Switch first")
453 1
        except KeyError:
454 1
            raise HTTPException(404, detail="Switch not found")
455
456 1
        if interface_enable_id:
457 1
            interface_number = int(interface_enable_id.split(":")[-1])
458
459 1
            try:
460 1
                interface = switch.interfaces[interface_number]
461 1
                self.topo_controller.enable_interface(interface.id)
462 1
                interface.enable()
463 1
                self.notify_interface_link_status(interface, "link enabled")
464 1
            except KeyError:
465 1
                msg = f"Switch {dpid} interface {interface_number} not found"
466 1
                raise HTTPException(404, detail=msg)
467
        else:
468 1
            for interface in switch.interfaces.copy().values():
469 1
                interface.enable()
470 1
                self.notify_interface_link_status(interface, "link enabled")
471 1
            self.topo_controller.upsert_switch(switch.id, switch.as_dict())
472 1
        self.notify_topology_update()
473 1
        return JSONResponse("Operation successful")
474
475 1
    @rest('v3/interfaces/switch/{dpid}/disable', methods=['POST'])
476 1
    @rest('v3/interfaces/{interface_disable_id}/disable', methods=['POST'])
477 1
    def disable_interface(self, request: Request) -> JSONResponse:
478
        """Administratively disable interfaces in the topology."""
479 1
        interface_disable_id = request.path_params.get("interface_disable_id")
480 1
        dpid = request.path_params.get("dpid")
481 1
        if dpid is None:
482 1
            dpid = ":".join(interface_disable_id.split(":")[:-1])
483 1
        try:
484 1
            switch = self.controller.switches[dpid]
485 1
        except KeyError:
486 1
            raise HTTPException(404, detail="Switch not found")
487
488 1
        if interface_disable_id:
489 1
            interface_number = int(interface_disable_id.split(":")[-1])
490
491 1
            try:
492 1
                interface = switch.interfaces[interface_number]
493 1
                self.topo_controller.disable_interface(interface.id)
494 1
                if interface.link and interface.link.is_enabled():
495 1
                    self.topo_controller.disable_link(interface.link.id)
496 1
                    interface.link.disable()
497 1
                    self.notify_link_enabled_state(interface.link, "disabled")
498 1
                interface.disable()
499 1
                self.notify_interface_link_status(interface, "link disabled")
500 1
            except KeyError:
501 1
                msg = f"Switch {dpid} interface {interface_number} not found"
502 1
                raise HTTPException(404, detail=msg)
503
        else:
504 1
            link_ids = set()
505 1
            for interface in switch.interfaces.copy().values():
506 1
                if interface.link and interface.link.is_enabled():
507 1
                    link_ids.add(interface.link.id)
508 1
                    interface.link.disable()
509 1
                    self.notify_link_enabled_state(interface.link, "disabled")
510 1
                interface.disable()
511 1
                self.notify_interface_link_status(interface, "link disabled")
512 1
            self.topo_controller.bulk_disable_links(link_ids)
513 1
            self.topo_controller.upsert_switch(switch.id, switch.as_dict())
514 1
        self.notify_topology_update()
515 1
        return JSONResponse("Operation successful")
516
517 1
    @rest('v3/interfaces/{interface_id}/metadata')
518 1
    def get_interface_metadata(self, request: Request) -> JSONResponse:
519
        """Get metadata from an interface."""
520 1
        interface_id = request.path_params["interface_id"]
521 1
        switch_id = ":".join(interface_id.split(":")[:-1])
522 1
        interface_number = int(interface_id.split(":")[-1])
523 1
        try:
524 1
            switch = self.controller.switches[switch_id]
525 1
        except KeyError:
526 1
            raise HTTPException(404, detail="Switch not found")
527
528 1
        try:
529 1
            interface = switch.interfaces[interface_number]
530 1
        except KeyError:
531 1
            raise HTTPException(404, detail="Interface not found")
532
533 1
        return JSONResponse({"metadata": interface.metadata})
534
535 1
    @rest('v3/interfaces/{interface_id}/metadata', methods=['POST'])
536 1
    def add_interface_metadata(self, request: Request) -> JSONResponse:
537
        """Add metadata to an interface."""
538 1
        interface_id = request.path_params["interface_id"]
539 1
        metadata = self._get_metadata(request)
540 1
        switch_id = ":".join(interface_id.split(":")[:-1])
541 1
        interface_number = int(interface_id.split(":")[-1])
542 1
        try:
543 1
            switch = self.controller.switches[switch_id]
544 1
        except KeyError:
545 1
            raise HTTPException(404, detail="Switch not found")
546
547 1
        try:
548 1
            interface = switch.interfaces[interface_number]
549 1
            self.topo_controller.add_interface_metadata(interface.id, metadata)
550 1
        except KeyError:
551 1
            raise HTTPException(404, detail="Interface not found")
552
553 1
        interface.extend_metadata(metadata)
554 1
        self.notify_metadata_changes(interface, 'added')
555 1
        return JSONResponse("Operation successful", status_code=201)
556
557 1
    @rest('v3/interfaces/{interface_id}/metadata/{key}', methods=['DELETE'])
558 1
    def delete_interface_metadata(self, request: Request) -> JSONResponse:
559
        """Delete metadata from an interface."""
560 1
        interface_id = request.path_params["interface_id"]
561 1
        key = request.path_params["key"]
562 1
        switch_id = ":".join(interface_id.split(":")[:-1])
563 1
        try:
564 1
            interface_number = int(interface_id.split(":")[-1])
565
        except ValueError:
566
            detail = f"Invalid interface_id {interface_id}"
567
            raise HTTPException(400, detail=detail)
568
569 1
        try:
570 1
            switch = self.controller.switches[switch_id]
571 1
        except KeyError:
572 1
            raise HTTPException(404, detail="Switch not found")
573
574 1
        try:
575 1
            interface = switch.interfaces[interface_number]
576 1
        except KeyError:
577 1
            raise HTTPException(404, detail="Interface not found")
578
579 1
        try:
580 1
            _ = interface.metadata[key]
581 1
        except KeyError:
582 1
            raise HTTPException(404, detail="Metadata not found")
583
584 1
        self.topo_controller.delete_interface_metadata_key(interface.id, key)
585 1
        interface.remove_metadata(key)
586 1
        self.notify_metadata_changes(interface, 'removed')
587 1
        return JSONResponse("Operation successful")
588
589 1
    @rest('v3/interfaces/{interface_id}/tag_ranges', methods=['POST'])
590 1
    @validate_openapi(spec)
591 1
    def set_tag_range(self, request: Request) -> JSONResponse:
592
        """Set tag range"""
593 1
        content_type_json_or_415(request)
594 1
        content = get_json_or_400(request, self.controller.loop)
595 1
        tag_type = content.get("tag_type")
596 1
        try:
597 1
            ranges = get_tag_ranges(content["tag_ranges"])
598
        except KytosInvalidTagRanges as err:
599
            raise HTTPException(400, detail=str(err))
600 1
        interface_id = request.path_params["interface_id"]
601 1
        interface = self.controller.get_interface_by_id(interface_id)
602 1
        if not interface:
603 1
            raise HTTPException(404, detail="Interface not found")
604 1
        try:
605 1
            interface.set_tag_ranges(ranges, tag_type)
606 1
            self.handle_on_interface_tags(interface)
607 1
        except KytosTagError as err:
608 1
            raise HTTPException(400, detail=str(err))
609 1
        return JSONResponse("Operation Successful", status_code=200)
610
611 1
    @rest('v3/interfaces/{interface_id}/tag_ranges', methods=['DELETE'])
612 1
    @validate_openapi(spec)
613 1
    def delete_tag_range(self, request: Request) -> JSONResponse:
614
        """Set tag_range from tag_type to default value [1, 4095]"""
615 1
        interface_id = request.path_params["interface_id"]
616 1
        params = request.query_params
617 1
        tag_type = params.get("tag_type", 'vlan')
618 1
        interface = self.controller.get_interface_by_id(interface_id)
619 1
        if not interface:
620 1
            raise HTTPException(404, detail="Interface not found")
621 1
        try:
622 1
            interface.remove_tag_ranges(tag_type)
623 1
            self.handle_on_interface_tags(interface)
624 1
        except KytosTagError as err:
625 1
            raise HTTPException(400, detail=str(err))
626 1
        return JSONResponse("Operation Successful", status_code=200)
627
628 1
    @rest('v3/interfaces/{interface_id}/special_tags', methods=['POST'])
629 1
    @validate_openapi(spec)
630 1
    def set_special_tags(self, request: Request) -> JSONResponse:
631
        """Set special_tags"""
632 1
        content_type_json_or_415(request)
633 1
        content = get_json_or_400(request, self.controller.loop)
634 1
        tag_type = content.get("tag_type")
635 1
        special_tags = content["special_tags"]
636 1
        interface_id = request.path_params["interface_id"]
637 1
        interface = self.controller.get_interface_by_id(interface_id)
638 1
        if not interface:
639 1
            raise HTTPException(404, detail="Interface not found")
640 1
        try:
641 1
            interface.set_special_tags(tag_type, special_tags)
642 1
            self.handle_on_interface_tags(interface)
643 1
        except KytosTagError as err:
644 1
            raise HTTPException(400, detail=str(err))
645 1
        return JSONResponse("Operation Successful", status_code=200)
646
647 1
    @rest('v3/interfaces/tag_ranges', methods=['GET'])
648 1
    @validate_openapi(spec)
649 1
    def get_all_tag_ranges(self, _: Request) -> JSONResponse:
650
        """Get all tag_ranges, available_tags, special_tags
651
         and special_available_tags from interfaces"""
652 1
        result = {}
653 1
        for switch in self.controller.switches.copy().values():
654 1
            for interface in switch.interfaces.copy().values():
655 1
                result[interface.id] = {
656
                    "available_tags": interface.available_tags,
657
                    "tag_ranges": interface.tag_ranges,
658
                    "special_tags": interface.special_tags,
659
                    "special_available_tags": interface.special_available_tags
660
                }
661 1
        return JSONResponse(result, status_code=200)
662
663 1
    @rest('v3/interfaces/{interface_id}/tag_ranges', methods=['GET'])
664 1
    @validate_openapi(spec)
665 1
    def get_tag_ranges_by_intf(self, request: Request) -> JSONResponse:
666
        """Get tag_ranges, available_tags, special_tags
667
         and special_available_tags from an interface"""
668 1
        interface_id = request.path_params["interface_id"]
669 1
        interface = self.controller.get_interface_by_id(interface_id)
670 1
        if not interface:
671 1
            raise HTTPException(404, detail="Interface not found")
672 1
        result = {
673
            interface_id: {
674
                "available_tags": interface.available_tags,
675
                "tag_ranges": interface.tag_ranges,
676
                "special_tags": interface.special_tags,
677
                "special_available_tags": interface.special_available_tags
678
            }
679
        }
680 1
        return JSONResponse(result, status_code=200)
681
682
    # Link related methods
683 1
    @rest('v3/links')
684 1
    def get_links(self, _request: Request) -> JSONResponse:
685
        """Return a json with all the links in the topology.
686
687
        Links are connections between interfaces.
688
        """
689
        return JSONResponse(self._get_links_dict())
690
691 1
    @rest('v3/links/{link_id}/enable', methods=['POST'])
692 1
    def enable_link(self, request: Request) -> JSONResponse:
693
        """Administratively enable a link in the topology."""
694 1
        link_id = request.path_params["link_id"]
695 1
        try:
696 1
            with self._links_lock:
697 1
                link = self.links[link_id]
698 1
                if not link.endpoint_a.is_enabled():
699 1
                    detail = f"{link.endpoint_a.id} needs enabling."
700 1
                    raise HTTPException(409, detail=detail)
701 1
                if not link.endpoint_b.is_enabled():
702 1
                    detail = f"{link.endpoint_b.id} needs enabling."
703 1
                    raise HTTPException(409, detail=detail)
704 1
                if not link.is_enabled():
705 1
                    self.topo_controller.enable_link(link.id)
706 1
                    link.enable()
707 1
                    self.notify_link_enabled_state(link, "enabled")
708 1
        except KeyError:
709 1
            raise HTTPException(404, detail="Link not found")
710 1
        self.notify_link_status_change(
711
            self.links[link_id],
712
            reason='link enabled'
713
        )
714 1
        self.notify_topology_update()
715 1
        return JSONResponse("Operation successful", status_code=201)
716
717 1
    @rest('v3/links/{link_id}/disable', methods=['POST'])
718 1
    def disable_link(self, request: Request) -> JSONResponse:
719
        """Administratively disable a link in the topology."""
720 1
        link_id = request.path_params["link_id"]
721 1
        try:
722 1
            with self._links_lock:
723 1
                link = self.links[link_id]
724 1
                if link.is_enabled():
725 1
                    self.topo_controller.disable_link(link.id)
726 1
                    link.disable()
727 1
                    self.notify_link_enabled_state(link, "disabled")
728 1
        except KeyError:
729 1
            raise HTTPException(404, detail="Link not found")
730 1
        self.notify_link_status_change(
731
            self.links[link_id],
732
            reason='link disabled'
733
        )
734 1
        self.notify_topology_update()
735 1
        return JSONResponse("Operation successful", status_code=201)
736
737 1
    def notify_link_enabled_state(self, link: Link, action: str):
738
        """Send a KytosEvent whether a link status (enabled/disabled)
739
         has changed its status."""
740 1
        name = f'kytos/topology.link.{action}'
741 1
        content = {'link': link}
742 1
        event = KytosEvent(name=name, content=content)
743 1
        self.controller.buffers.app.put(event)
744
745 1
    @rest('v3/links/{link_id}/metadata')
746 1
    def get_link_metadata(self, request: Request) -> JSONResponse:
747
        """Get metadata from a link."""
748 1
        link_id = request.path_params["link_id"]
749 1
        try:
750 1
            return JSONResponse({"metadata": self.links[link_id].metadata})
751 1
        except KeyError:
752 1
            raise HTTPException(404, detail="Link not found")
753
754 1
    @rest('v3/links/{link_id}/metadata', methods=['POST'])
755 1
    def add_link_metadata(self, request: Request) -> JSONResponse:
756
        """Add metadata to a link."""
757 1
        link_id = request.path_params["link_id"]
758 1
        metadata = self._get_metadata(request)
759 1
        try:
760 1
            link = self.links[link_id]
761 1
        except KeyError:
762 1
            raise HTTPException(404, detail="Link not found")
763
764 1
        self.topo_controller.add_link_metadata(link_id, metadata)
765 1
        link.extend_metadata(metadata)
766 1
        self.notify_metadata_changes(link, 'added')
767 1
        self.notify_topology_update()
768 1
        return JSONResponse("Operation successful", status_code=201)
769
770 1
    @rest('v3/links/{link_id}/metadata/{key}', methods=['DELETE'])
771 1
    def delete_link_metadata(self, request: Request) -> JSONResponse:
772
        """Delete metadata from a link."""
773 1
        link_id = request.path_params["link_id"]
774 1
        key = request.path_params["key"]
775 1
        try:
776 1
            link = self.links[link_id]
777 1
        except KeyError:
778 1
            raise HTTPException(404, detail="Link not found")
779
780 1
        try:
781 1
            _ = link.metadata[key]
782 1
        except KeyError:
783 1
            raise HTTPException(404, detail="Metadata not found")
784
785 1
        self.topo_controller.delete_link_metadata_key(link.id, key)
786 1
        link.remove_metadata(key)
787 1
        self.notify_metadata_changes(link, 'removed')
788 1
        self.notify_topology_update()
789 1
        return JSONResponse("Operation successful")
790
791 1
    @rest('v3/links/{link_id}', methods=['DELETE'])
792 1
    def delete_link(self, request: Request) -> JSONResponse:
793
        """Delete a disabled link from topology.
794
         It won't work for link with other statuses.
795
        """
796 1
        link_id = request.path_params["link_id"]
797 1
        try:
798 1
            with self._links_lock:
799 1
                link = self.links[link_id]
800 1
                if link.status != EntityStatus.DISABLED:
801 1
                    raise HTTPException(409, detail="Link is not disabled.")
802 1
                if link.endpoint_a.link and link == link.endpoint_a.link:
803 1
                    switch = link.endpoint_a.switch
804 1
                    link.endpoint_a.link = None
805 1
                    link.endpoint_a.nni = False
806 1
                    self.topo_controller.upsert_switch(
807
                        switch.id, switch.as_dict()
808
                    )
809 1
                if link.endpoint_b.link and link == link.endpoint_b.link:
810 1
                    switch = link.endpoint_b.switch
811 1
                    link.endpoint_b.link = None
812 1
                    link.endpoint_b.nni = False
813 1
                    self.topo_controller.upsert_switch(
814
                        switch.id, switch.as_dict()
815
                    )
816 1
                self.topo_controller.delete_link(link_id)
817 1
                link = self.links.pop(link_id)
818 1
        except KeyError:
819 1
            raise HTTPException(404, detail="Link not found.")
820 1
        self.notify_topology_update()
821 1
        name = 'kytos/topology.link.deleted'
822 1
        event = KytosEvent(name=name, content={'link': link})
823 1
        self.controller.buffers.app.put(event)
824 1
        return JSONResponse("Operation successful")
825
826 1
    @rest('v3/interfaces/{intf_id}', methods=['DELETE'])
827 1
    def delete_interface(self, request: Request) -> JSONResponse:
828
        """Delete an interface only if it is not used."""
829 1
        intf_id = request.path_params.get("intf_id")
830 1
        intf_split = intf_id.split(":")
831 1
        switch_id = ":".join(intf_split[:-1])
832 1
        try:
833 1
            intf_port = int(intf_split[-1])
834 1
        except ValueError:
835 1
            raise HTTPException(400, detail="Invalid interface id.")
836 1
        try:
837 1
            switch = self.controller.switches[switch_id]
838 1
        except KeyError:
839 1
            raise HTTPException(404, detail="Switch not found.")
840 1
        try:
841 1
            interface = switch.interfaces[intf_port]
842 1
        except KeyError:
843 1
            raise HTTPException(404, detail="Interface not found.")
844
845 1
        usage = self.get_intf_usage(interface)
846 1
        if usage:
847 1
            raise HTTPException(409, detail=f"Interface could not be "
848
                                            f"deleted. Reason: {usage}")
849 1
        self._delete_interface(interface)
850 1
        return JSONResponse("Operation Successful", status_code=200)
851
852 1
    @listen_to(
853
        "kytos/.*.liveness.(up|down|disabled)",
854
        pool="dynamic_single"
855
    )
856 1
    def on_link_liveness(self, event) -> None:
857
        """Handle link liveness up|down|disabled event."""
858
        with self._links_lock:
859
            liveness_status = event.name.split(".")[-1]
860
            if liveness_status == "disabled":
861
                interfaces = event.content["interfaces"]
862
                self.handle_link_liveness_disabled(interfaces)
863
            elif liveness_status in ("up", "down"):
864
                link = Link(event.content["interface_a"],
865
                            event.content["interface_b"])
866
                try:
867
                    link = self.links[link.id]
868
                except KeyError:
869
                    log.error(f"Link id {link.id} not found, {link}")
870
                    return
871
                self.handle_link_liveness_status(self.links[link.id],
872
                                                 liveness_status)
873
874 1
    def handle_link_liveness_status(self, link, liveness_status) -> None:
875
        """Handle link liveness."""
876 1
        metadata = {"liveness_status": liveness_status}
877 1
        log.info(f"Link liveness {liveness_status}: {link}")
878 1
        link.extend_metadata(metadata)
879 1
        self.notify_topology_update()
880 1
        if link.status == EntityStatus.UP and liveness_status == "up":
881 1
            self.notify_link_status_change(link, reason="liveness_up")
882 1
        if link.status == EntityStatus.DOWN and liveness_status == "down":
883 1
            self.notify_link_status_change(link, reason="liveness_down")
884
885 1
    def get_links_from_interfaces(self, interfaces) -> dict:
886
        """Get links from interfaces."""
887 1
        links_found = {}
888 1
        for interface in interfaces:
889 1
            for link in list(self.links.values()):
890 1
                if any((
891
                    interface.id == link.endpoint_a.id,
892
                    interface.id == link.endpoint_b.id,
893
                )):
894 1
                    links_found[link.id] = link
895 1
        return links_found
896
897 1
    def handle_link_liveness_disabled(self, interfaces) -> None:
898
        """Handle link liveness disabled."""
899 1
        log.info(f"Link liveness disabled interfaces: {interfaces}")
900
901 1
        key = "liveness_status"
902 1
        links = self.get_links_from_interfaces(interfaces)
903 1
        for link in links.values():
904 1
            link.remove_metadata(key)
905 1
        self.notify_topology_update()
906 1
        for link in links.values():
907 1
            self.notify_link_status_change(link, reason="liveness_disabled")
908
909 1
    @listen_to("kytos/core.interface_tags")
910 1
    def on_interface_tags(self, event):
911
        """Handle on_interface_tags."""
912
        interface = event.content['interface']
913
        with self._intfs_lock[interface.id]:
914
            if (
915
                interface.id in self._intfs_tags_updated_at
916
                and self._intfs_tags_updated_at[interface.id] > event.timestamp
917
            ):
918
                return
919
            self._intfs_tags_updated_at[interface.id] = event.timestamp
920
            self.handle_on_interface_tags(interface)
921
922 1
    def handle_on_interface_tags(self, interface):
923
        """Update interface details"""
924 1
        intf_id = interface.id
925 1
        self.topo_controller.upsert_interface_details(
926
            intf_id, interface.available_tags, interface.tag_ranges,
927
            interface.special_available_tags,
928
            interface.special_tags
929
        )
930
931 1
    @listen_to('.*.switch.(new|reconnected)')
932 1
    def on_new_switch(self, event):
933
        """Create a new Device on the Topology.
934
935
        Handle the event of a new created switch and update the topology with
936
        this new device. Also notify if the switch is enabled.
937
        """
938
        self.handle_new_switch(event)
939
940 1
    def handle_new_switch(self, event):
941
        """Create a new Device on the Topology."""
942 1
        switch = event.content['switch']
943 1
        switch.activate()
944 1
        self.topo_controller.upsert_switch(switch.id, switch.as_dict())
945 1
        log.debug('Switch %s added to the Topology.', switch.id)
946 1
        self.notify_topology_update()
947 1
        if switch.is_enabled():
948 1
            self.notify_switch_enabled(switch.id)
949
950 1
    @listen_to('.*.connection.lost')
951 1
    def on_connection_lost(self, event):
952
        """Remove a Device from the topology.
953
954
        Remove the disconnected Device and every link that has one of its
955
        interfaces.
956
        """
957
        self.handle_connection_lost(event)
958
959 1
    def handle_connection_lost(self, event):
960
        """Remove a Device from the topology."""
961 1
        switch = event.content['source'].switch
962 1
        if switch:
963 1
            switch.deactivate()
964 1
            log.debug('Switch %s removed from the Topology.', switch.id)
965 1
            self.notify_topology_update()
966
967 1
    def handle_interfaces_created(self, event):
968
        """Update the topology based on the interfaces created."""
969 1
        interfaces = event.content["interfaces"]
970 1
        if not interfaces:
971
            return
972 1
        switch = interfaces[0].switch
973 1
        self.topo_controller.upsert_switch(switch.id, switch.as_dict())
974 1
        name = "kytos/topology.switch.interface.created"
975 1
        for interface in interfaces:
976 1
            event = KytosEvent(name=name, content={'interface': interface})
977 1
            self.controller.buffers.app.put(event)
978
979 1
    def handle_interface_created(self, event):
980
        """Update the topology based on an interface created event.
981
982
        It's handled as a link_up in case a switch send a
983
        created event again and it can be belong to a link.
984
        """
985 1
        interface = event.content['interface']
986 1
        if not interface.is_active():
987 1
            self.handle_interface_link_down(interface, event)
988
        else:
989 1
            self.handle_interface_link_up(interface, event)
990
991 1
    @listen_to('.*.topology.switch.interface.created')
992 1
    def on_interface_created(self, event):
993
        """Handle individual interface create event.
994
995
        It's handled as a link_up in case a switch send a
996
        created event it can belong to an existign link.
997
        """
998
        self.handle_interface_created(event)
999
1000 1
    @listen_to('.*.switch.interfaces.created')
1001 1
    def on_interfaces_created(self, event):
1002
        """Update the topology based on a list of created interfaces."""
1003
        self.handle_interfaces_created(event)
1004
1005 1
    def handle_interface_down(self, event):
1006
        """Update the topology based on a Port Modify event.
1007
1008
        The event notifies that an interface was changed to 'down'.
1009
        """
1010 1
        interface = event.content['interface']
1011 1
        with self._intfs_lock[interface.id]:
1012 1
            if (
1013
                interface.id in self._intfs_updated_at
1014
                and self._intfs_updated_at[interface.id] > event.timestamp
1015
            ):
1016
                return
1017 1
            self._intfs_updated_at[interface.id] = event.timestamp
1018 1
            interface.deactivate()
1019 1
        self.handle_interface_link_down(interface, event)
1020
1021 1
    @listen_to('.*.switch.interface.deleted')
1022 1
    def on_interface_deleted(self, event):
1023
        """Update the topology based on a Port Delete event."""
1024
        self.handle_interface_deleted(event)
1025
1026 1
    def handle_interface_deleted(self, event):
1027
        """Update the topology based on a Port Delete event."""
1028 1
        self.handle_interface_down(event)
1029 1
        interface = event.content['interface']
1030 1
        usage = self.get_intf_usage(interface)
1031 1
        if usage:
1032 1
            log.info(f"Interface {interface.id} could not be safely removed."
1033
                     f" Reason: {usage}")
1034
        else:
1035 1
            self._delete_interface(interface)
1036
1037 1
    def get_intf_usage(self, interface: Interface) -> Optional[str]:
1038
        """Determines how an interface is used explained in a string,
1039
        returns None if unused."""
1040 1
        if interface.is_enabled() or interface.is_active():
1041 1
            return "It is enabled or active."
1042
1043 1
        with self._links_lock:
1044 1
            link = interface.link
1045 1
            if link:
1046 1
                return f"It has a link, {link.id}."
1047
1048 1
        flow_id = self.get_flow_id_by_intf(interface)
1049 1
        if flow_id:
1050 1
            return f"There is a flow installed, {flow_id}."
1051
1052 1
        return None
1053
1054 1
    def get_flow_id_by_intf(self, interface: Interface) -> str:
1055
        """Return flow_id from first found flow used by interface."""
1056 1
        flows = self.get_flows_by_switch(interface.switch.id)
1057 1
        port_n = int(interface.id.split(":")[-1])
1058 1
        for flow in flows:
1059 1
            in_port = flow["flow"].get("match", {}).get("in_port")
1060 1
            if in_port == port_n:
1061 1
                return flow["flow_id"]
1062
1063 1
            instructions = flow["flow"].get("instructions", [])
1064 1
            for instruction in instructions:
1065 1
                if instruction["instruction_type"] == "apply_actions":
1066 1
                    actions = instruction["actions"]
1067 1
                    for action in actions:
1068 1
                        if (action["action_type"] == "output"
1069
                                and action.get("port") == port_n):
1070 1
                            return flow["flow_id"]
1071
1072 1
            actions = flow["flow"].get("actions", [])
1073 1
            for action in actions:
1074 1
                if (action["action_type"] == "output"
1075
                        and action.get("port") == port_n):
1076 1
                    return flow["flow_id"]
1077 1
        return None
1078
1079 1
    def _delete_interface(self, interface: Interface):
1080
        """Delete any trace of an interface. Only use this method when
1081
         it was confirmed that the interface is not used."""
1082 1
        switch: Switch = interface.switch
1083 1
        switch.remove_interface(interface)
1084 1
        self.topo_controller.upsert_switch(switch.id, switch.as_dict())
1085 1
        self.topo_controller.delete_interface_from_details(interface.id)
1086
1087 1
    @listen_to('.*.switch.interface.link_up')
1088 1
    def on_interface_link_up(self, event):
1089
        """Update the topology based on a Port Modify event.
1090
1091
        The event notifies that an interface's link was changed to 'up'.
1092
        """
1093
        interface = event.content['interface']
1094
        self.handle_interface_link_up(interface, event)
1095
1096 1
    def handle_interface_link_up(self, interface, event):
1097
        """Update the topology based on a Port Modify event."""
1098 1
        with self._intfs_lock[interface.id]:
1099 1
            if (
1100
                interface.id in self._intfs_updated_at
1101
                and self._intfs_updated_at[interface.id] > event.timestamp
1102
            ):
1103 1
                return
1104 1
            self._intfs_updated_at[interface.id] = event.timestamp
1105 1
            self.handle_link_up(interface)
1106
1107 1
    @tenacity.retry(
1108
        stop=stop_after_attempt(3),
1109
        wait=wait_combine(wait_fixed(3), wait_random(min=2, max=7)),
1110
        before_sleep=before_sleep,
1111
        retry=retry_if_exception_type(httpx.RequestError),
1112
    )
1113 1
    def get_flows_by_switch(self, dpid: str) -> list:
1114
        """Get installed flows by switch from flow_manager."""
1115 1
        endpoint = settings.FLOW_MANAGER_URL +\
1116
            f'/stored_flows?state=installed&dpid={dpid}'
1117 1
        res = httpx.get(endpoint)
1118 1
        if res.is_server_error or res.status_code in (404, 400):
1119 1
            raise httpx.RequestError(res.text)
1120 1
        return res.json().get(dpid, [])
1121
1122 1
    def link_status_hook_link_up_timer(
1123
        self,
1124
        link: Link
1125
    ) -> Optional[EntityStatus]:
1126
        """Link status hook link up timer."""
1127 1
        tnow = time.time()
1128 1
        if link.id not in self.link_status_change:
1129
            return None
1130 1
        link_status_info = self.link_status_change[link.id]
1131 1
        tdelta = tnow - link_status_info['last_status_change']
1132 1
        if tdelta < self.link_up_timer:
1133 1
            return EntityStatus.DOWN
1134 1
        return None
1135
1136 1
    @staticmethod
1137 1
    def detect_mismatched_link(link: Link) -> frozenset[str]:
1138
        """Check if a link is mismatched."""
1139 1
        if (link.endpoint_a.link and link.endpoint_b
1140
                and link.endpoint_a.link == link.endpoint_b.link):
1141 1
            return frozenset()
1142 1
        return frozenset(["mismatched_link"])
1143
1144 1
    def link_status_mismatched(self, link: Link) -> Optional[EntityStatus]:
1145
        """Check if a link is mismatched and return a status."""
1146 1
        if self.detect_mismatched_link(link):
1147 1
            return EntityStatus.DOWN
1148 1
        return None
1149
1150 1
    def notify_link_up_if_status(self, link: Link, reason="link up") -> None:
1151
        """Tries to notify link up and topology changes based on its status
1152
1153
        Currently, it needs to wait up to a timer."""
1154 1
        time.sleep(self.link_up_timer)
1155 1
        if link.status != EntityStatus.UP:
1156
            return
1157 1
        with self._links_lock:
1158 1
            status_change_info = self.link_status_change[link.id]
1159 1
            notified_at = status_change_info.get("notified_up_at")
1160 1
            if (
1161
                notified_at
1162
                and (now() - notified_at.replace(tzinfo=timezone.utc)).seconds
1163
                < self.link_up_timer
1164
            ):
1165 1
                return
1166 1
            status_change_info["notified_up_at"] = now()
1167 1
            self.notify_topology_update()
1168 1
            self.notify_link_status_change(link, reason)
1169
1170 1
    def handle_link_up(self, interface: Interface):
1171
        """Handle link up for an interface."""
1172 1
        with self._links_lock:
1173 1
            link = interface.link
1174 1
            if not link:
1175
                self.notify_topology_update()
1176
                return
1177 1
            other_interface = (
1178
                link.endpoint_b if link.endpoint_a == interface
1179
                else link.endpoint_a
1180
            )
1181 1
            if (
1182
                link.id not in self.link_status_change or
1183
                not link.is_active()
1184
            ):
1185 1
                status_change_info = self.link_status_change[link.id]
1186 1
                status_change_info['last_status_change'] = time.time()
1187 1
                link.activate()
1188 1
            self.notify_topology_update()
1189 1
            link_dependencies: list[GenericEntity] = [
1190
                other_interface.switch,
1191
                interface.switch,
1192
                other_interface,
1193
                interface,
1194
            ]
1195 1
            for dependency in link_dependencies:
1196 1
                if not dependency.is_active():
1197 1
                    log.info(
1198
                        f"{link} dependency {dependency} was not active yet."
1199
                    )
1200 1
                    return
1201 1
            event = KytosEvent(
1202
                name="kytos/topology.notify_link_up_if_status",
1203
                content={"reason": "link up", "link": link}
1204
            )
1205 1
            self.controller.buffers.app.put(event)
1206
1207 1
    @listen_to('.*.switch.interface.link_down')
1208 1
    def on_interface_link_down(self, event):
1209
        """Update the topology based on a Port Modify event.
1210
1211
        The event notifies that an interface's link was changed to 'down'.
1212
        """
1213
        interface = event.content['interface']
1214
        self.handle_interface_link_down(interface, event)
1215
1216 1
    def handle_interface_link_down(self, interface, event):
1217
        """Update the topology based on an interface."""
1218 1
        with self._intfs_lock[interface.id]:
1219 1
            if (
1220
                interface.id in self._intfs_updated_at
1221
                and self._intfs_updated_at[interface.id] > event.timestamp
1222
            ):
1223 1
                return
1224 1
            self._intfs_updated_at[interface.id] = event.timestamp
1225 1
            self.handle_link_down(interface)
1226
1227 1
    def handle_link_down(self, interface):
1228
        """Notify a link is down."""
1229 1
        with self._links_lock:
1230 1
            link = interface.link
1231 1
            if link:
1232 1
                link.deactivate()
1233 1
                self.notify_link_status_change(link, reason="link down")
1234 1
            self.notify_topology_update()
1235
1236 1
    @listen_to('.*.interface.is.nni')
1237 1
    def on_add_links(self, event):
1238
        """Update the topology with links related to the NNI interfaces."""
1239
        self.add_links(event)
1240
1241 1
    def add_links(self, event):
1242
        """Update the topology with links related to the NNI interfaces."""
1243 1
        interface_a = event.content['interface_a']
1244 1
        interface_b = event.content['interface_b']
1245
1246 1
        try:
1247 1
            with self._links_lock:
1248 1
                link, created = self._get_link_or_create(interface_a,
1249
                                                         interface_b)
1250 1
                interface_a.update_link(link)
1251 1
                interface_b.update_link(link)
1252
1253 1
                link.endpoint_a = interface_a
1254 1
                link.endpoint_b = interface_b
1255
1256 1
                interface_a.nni = True
1257 1
                interface_b.nni = True
1258
1259
        except KytosLinkCreationError as err:
1260
            log.error(f'Error creating link: {err}.')
1261
            return
1262
1263 1
        if not created:
1264
            return
1265
1266 1
        self.notify_topology_update()
1267 1
        if not link.is_active():
1268
            return
1269 1
        status_change_info = self.link_status_change[link.id]
1270 1
        status_change_info['last_status_change'] = time.time()
1271
1272 1
        self.topo_controller.upsert_link(link.id, link.as_dict())
1273 1
        self.notify_link_up_if_status(link, "link up")
1274
1275 1
    @listen_to('.*.of_lldp.network_status.updated')
1276 1
    def on_lldp_status_updated(self, event):
1277
        """Handle of_lldp.network_status.updated from of_lldp."""
1278
        self.handle_lldp_status_updated(event)
1279
1280 1
    @listen_to(".*.topo_controller.upsert_switch")
1281 1
    def on_topo_controller_upsert_switch(self, event) -> None:
1282
        """Listen to topo_controller_upsert_switch."""
1283
        self.handle_topo_controller_upsert_switch(event.content["switch"])
1284
1285 1
    def handle_topo_controller_upsert_switch(self, switch) -> Optional[dict]:
1286
        """Handle topo_controller_upsert_switch."""
1287 1
        return self.topo_controller.upsert_switch(switch.id, switch.as_dict())
1288
1289 1
    def handle_lldp_status_updated(self, event) -> None:
1290
        """Handle .*.network_status.updated events from of_lldp."""
1291 1
        content = event.content
1292 1
        interface_ids = content["interface_ids"]
1293 1
        switches = set()
1294 1
        for interface_id in interface_ids:
1295 1
            dpid = ":".join(interface_id.split(":")[:-1])
1296 1
            switch = self.controller.get_switch_by_dpid(dpid)
1297 1
            if switch:
1298 1
                switches.add(switch)
1299
1300 1
        name = "kytos/topology.topo_controller.upsert_switch"
1301 1
        for switch in switches:
1302 1
            event = KytosEvent(name=name, content={"switch": switch})
1303 1
            self.controller.buffers.app.put(event)
1304
1305 1
    def notify_switch_enabled(self, dpid):
1306
        """Send an event to notify that a switch is enabled."""
1307 1
        name = 'kytos/topology.switch.enabled'
1308 1
        event = KytosEvent(name=name, content={'dpid': dpid})
1309 1
        self.controller.buffers.app.put(event)
1310
1311 1
    def notify_switch_links_status(self, switch, reason):
1312
        """Send an event to notify the status of a link in a switch"""
1313 1
        with self._links_lock:
1314 1
            for link in self.links.values():
1315 1
                if switch in (link.endpoint_a.switch, link.endpoint_b.switch):
1316 1
                    if reason == "link enabled":
1317 1
                        name = 'kytos/topology.notify_link_up_if_status'
1318 1
                        content = {'reason': reason, "link": link}
1319 1
                        event = KytosEvent(name=name, content=content)
1320 1
                        self.controller.buffers.app.put(event)
1321
                    else:
1322 1
                        self.notify_link_status_change(link, reason)
1323
1324 1
    def notify_switch_disabled(self, dpid):
1325
        """Send an event to notify that a switch is disabled."""
1326 1
        name = 'kytos/topology.switch.disabled'
1327 1
        event = KytosEvent(name=name, content={'dpid': dpid})
1328 1
        self.controller.buffers.app.put(event)
1329
1330 1
    def notify_topology_update(self):
1331
        """Send an event to notify about updates on the topology."""
1332 1
        name = 'kytos/topology.updated'
1333 1
        event = KytosEvent(name=name, content={'topology':
1334
                                               self._get_topology()})
1335 1
        self.controller.buffers.app.put(event)
1336
1337 1
    def notify_interface_link_status(self, interface, reason):
1338
        """Send an event to notify the status of a link from
1339
        an interface."""
1340 1
        with self._links_lock:
1341 1
            link = interface.link
1342 1
            if link:
1343 1
                if reason == "link enabled":
1344 1
                    name = 'kytos/topology.notify_link_up_if_status'
1345 1
                    content = {'reason': reason, "link": link}
1346 1
                    event = KytosEvent(name=name, content=content)
1347 1
                    self.controller.buffers.app.put(event)
1348
                else:
1349 1
                    self.notify_link_status_change(link, reason)
1350
1351 1
    def notify_link_status_change(self, link: Link, reason='not given'):
1352
        """Send an event to notify (up/down) from a status change on
1353
         a link."""
1354 1
        link_id = link.id
1355 1
        with self.link_status_lock:
1356 1
            if (
1357
                (not link.status_reason and link.status == EntityStatus.UP)
1358
                and link_id not in self.link_up
1359
            ):
1360 1
                log.info(f"{link} changed status {link.status}, "
1361
                         f"reason: {reason}")
1362 1
                self.link_up.add(link_id)
1363 1
                event = KytosEvent(
1364
                    name='kytos/topology.link_up',
1365
                    content={
1366
                        'link': link,
1367
                        'reason': reason
1368
                    },
1369
                )
1370 1
            elif (
1371
                (link.status_reason or link.status != EntityStatus.UP)
1372
                and link_id in self.link_up
1373
            ):
1374 1
                log.info(f"{link} changed status {link.status}, "
1375
                         f"reason: {reason}")
1376 1
                self.link_up.remove(link_id)
1377 1
                event = KytosEvent(
1378
                    name='kytos/topology.link_down',
1379
                    content={
1380
                        'link': link,
1381
                        'reason': reason
1382
                    },
1383
                )
1384
            else:
1385 1
                return
1386 1
        self.controller.buffers.app.put(event)
1387
1388 1
    def notify_metadata_changes(self, obj, action):
1389
        """Send an event to notify about metadata changes."""
1390 1
        if isinstance(obj, Switch):
1391 1
            entity = 'switch'
1392 1
            entities = 'switches'
1393 1
        elif isinstance(obj, Interface):
1394 1
            entity = 'interface'
1395 1
            entities = 'interfaces'
1396 1
        elif isinstance(obj, Link):
1397 1
            entity = 'link'
1398 1
            entities = 'links'
1399
        else:
1400 1
            raise ValueError(
1401
                'Invalid object, supported: Switch, Interface, Link'
1402
            )
1403
1404 1
        name = f'kytos/topology.{entities}.metadata.{action}'
1405 1
        content = {entity: obj, 'metadata': obj.metadata.copy()}
1406 1
        event = KytosEvent(name=name, content=content)
1407 1
        self.controller.buffers.app.put(event)
1408 1
        log.debug(f'Metadata from {obj.id} was {action}.')
1409
1410 1
    @listen_to('kytos/topology.notify_link_up_if_status')
1411 1
    def on_notify_link_up_if_status(self, event):
1412
        """Tries to notify link up and topology changes"""
1413
        link = event.content["link"]
1414
        reason = event.content["reason"]
1415
        self.notify_link_up_if_status(link, reason)
1416
1417 1
    @listen_to('.*.switch.port.created')
1418 1
    def on_notify_port_created(self, event):
1419
        """Notify when a port is created."""
1420
        self.notify_port_created(event)
1421
1422 1
    def notify_port_created(self, event):
1423
        """Notify when a port is created."""
1424 1
        name = 'kytos/topology.port.created'
1425 1
        event = KytosEvent(name=name, content=event.content)
1426 1
        self.controller.buffers.app.put(event)
1427
1428 1
    @staticmethod
1429 1
    def load_interfaces_tags_values(switch: Switch,
1430
                                    interfaces_details: List[dict]) -> None:
1431
        """Load interfaces available tags (vlans)."""
1432 1
        if not interfaces_details:
1433
            return
1434 1
        for interface_details in interfaces_details:
1435 1
            available_tags = interface_details['available_tags']
1436 1
            if not available_tags:
1437
                continue
1438 1
            log.debug(f"Interface id {interface_details['id']} loading "
1439
                      f"{len(available_tags)} "
1440
                      "available tags")
1441 1
            port_number = int(interface_details["id"].split(":")[-1])
1442 1
            interface = switch.interfaces[port_number]
1443 1
            interface.set_available_tags_tag_ranges(
1444
                available_tags,
1445
                interface_details['tag_ranges'],
1446
                interface_details['special_available_tags'],
1447
                interface_details['special_tags'],
1448
            )
1449
1450 1
    @listen_to(
1451
        'topology.interruption.(start|end)',
1452
        pool="dynamic_single"
1453
    )
1454 1
    def on_interruption(self, event: KytosEvent):
1455
        """Deals with service interruptions."""
1456
        with self._links_lock:
1457
            _, _, interrupt_type = event.name.rpartition(".")
1458
            if interrupt_type == "start":
1459
                self.handle_interruption_start(event)
1460
            elif interrupt_type == "end":
1461
                self.handle_interruption_end(event)
1462
1463 1 View Code Duplication
    def handle_interruption_start(self, event: KytosEvent):
0 ignored issues
show
This code seems to be duplicated in your project.
Loading history...
1464
        """Deals with the start of service interruption."""
1465 1
        interrupt_type = event.content['type']
1466 1
        switches = event.content.get('switches', [])
1467 1
        interfaces = event.content.get('interfaces', [])
1468 1
        links = event.content.get('links', [])
1469 1
        log.info(
1470
            'Received interruption start of type \'%s\' '
1471
            'affecting switches %s, interfaces %s, links %s',
1472
            interrupt_type,
1473
            switches,
1474
            interfaces,
1475
            links
1476
        )
1477
        # for switch_id in switches:
1478
        #     pass
1479
        # for interface_id in interfaces:
1480
        #     pass
1481 1
        for link_id in links:
1482 1
            link = self.links.get(link_id)
1483 1
            if link is None:
1484
                log.error(
1485
                    "Invalid link id '%s' for interruption of type '%s;",
1486
                    link_id,
1487
                    interrupt_type
1488
                )
1489
            else:
1490 1
                self.notify_link_status_change(link, interrupt_type)
1491 1
        self.notify_topology_update()
1492
1493 1 View Code Duplication
    def handle_interruption_end(self, event: KytosEvent):
0 ignored issues
show
This code seems to be duplicated in your project.
Loading history...
1494
        """Deals with the end of service interruption."""
1495 1
        interrupt_type = event.content['type']
1496 1
        switches = event.content.get('switches', [])
1497 1
        interfaces = event.content.get('interfaces', [])
1498 1
        links = event.content.get('links', [])
1499 1
        log.info(
1500
            'Received interruption end of type \'%s\' '
1501
            'affecting switches %s, interfaces %s, links %s',
1502
            interrupt_type,
1503
            switches,
1504
            interfaces,
1505
            links
1506
        )
1507
        # for switch_id in switches:
1508
        #     pass
1509
        # for interface_id in interfaces:
1510
        #     pass
1511 1
        for link_id in links:
1512 1
            link = self.links.get(link_id)
1513 1
            if link is None:
1514
                log.error(
1515
                    "Invalid link id '%s' for interruption of type '%s;",
1516
                    link_id,
1517
                    interrupt_type
1518
                )
1519
            else:
1520 1
                self.notify_link_status_change(link, interrupt_type)
1521
        self.notify_topology_update()
1522