Passed
Push — master ( 6282cb...7ef10c )
by
unknown
02:55 queued 15s
created

build.main.Main.get_links()   A

Complexity

Conditions 1

Size

Total Lines 7
Code Lines 3

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 2
CRAP Score 1.037

Importance

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