Passed
Pull Request — master (#259)
by Aldo
04:39
created

build.main.Main.delete_link()   C

Complexity

Conditions 9

Size

Total Lines 35
Code Lines 29

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 27
CRAP Score 9

Importance

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