Passed
Pull Request — master (#196)
by Aldo
03:28
created

build.main.Main._delete_interface()   A

Complexity

Conditions 1

Size

Total Lines 7
Code Lines 5

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 5
CRAP Score 1

Importance

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