Test Failed
Pull Request — master (#181)
by
unknown
07:22
created

build.main   F

Complexity

Total Complexity 239

Size/Duplication

Total Lines 1303
Duplicated Lines 9.21 %

Test Coverage

Coverage 91.98%

Importance

Changes 0
Metric Value
eloc 961
dl 120
loc 1303
ccs 768
cts 835
cp 0.9198
rs 1.639
c 0
b 0
f 0
wmc 239

87 Methods

Rating   Name   Duplication   Size   Complexity  
A Main.get_link_metadata() 0 8 2
A Main.on_notify_port_created() 0 4 1
A Main.get_links_from_interfaces() 0 12 5
A Main.notify_port_created() 0 5 1
A Main.disable_switch() 0 15 2
A Main.on_connection_lost() 0 8 1
A Main.get_tag_ranges_by_intf() 0 18 2
A Main.on_interface_link_down() 0 8 1
A Main.on_interruption_end() 0 5 2
A Main.disable_link() 0 17 3
A Main.enable_switch() 0 15 2
B Main.delete_link() 0 34 8
B Main.enable_interface() 31 31 6
A Main.link_status_hook_link_up_timer() 0 11 5
B Main.load_topology() 0 35 5
A Main.on_notify_link_up_if_status() 0 6 1
A Main.add_switch_metadata() 0 14 2
A Main._get_switches_dict() 0 12 3
A Main.get_interfaces() 0 10 3
B Main.delete_interface_metadata() 0 31 5
A Main.add_interface_metadata() 0 21 3
A Main.on_link_liveness_status() 0 11 2
A Main.get_all_tag_ranges() 0 15 3
A Main.on_interfaces_created() 0 4 1
A Main.get_switches() 0 4 1
A Main.handle_topo_controller_upsert_switch() 0 3 1
B Main.notify_link_status_change() 0 31 8
A Main.handle_interface_down() 0 15 4
A Main.on_interface_created() 0 8 1
A Main.handle_interface_link_up() 0 10 4
A Main.on_interface_tags() 0 12 4
A Main.on_interface_link_up() 0 8 1
B Main.handle_link_up() 0 22 5
A Main.notify_switch_enabled() 0 5 1
A Main.handle_interface_created() 0 10 2
A Main.delete_tag_range() 0 16 3
A Main.get_interface_metadata() 0 17 3
A Main.on_interruption_start() 0 5 2
A Main.handle_lldp_status_updated() 0 15 4
A Main.delete_link_metadata() 0 20 3
A Main.on_lldp_status_updated() 0 4 1
A Main.shutdown() 0 3 1
A Main.handle_interruption_start() 29 29 3
A Main.notify_topology_update() 0 6 1
A Main.on_interface_deleted() 0 4 1
A Main.notify_link_up_if_status() 0 19 5
A Main.enable_link() 0 17 3
A Main.handle_link_down() 0 15 4
A Main.setup() 0 21 1
A Main.notify_interface_link_status() 0 12 3
A Main.on_new_switch() 0 8 1
A Main.handle_new_switch() 0 9 2
A Main.execute() 0 3 1
A Main.load_interfaces_tags_values() 0 20 4
B Main._load_switch() 0 44 5
A Main.get_switch_metadata() 0 9 2
A Main._get_metadata() 0 7 2
A Main.handle_connection_lost() 0 7 2
A Main.set_tag_range() 0 21 4
A Main.handle_on_interface_tags() 0 7 1
B Main.disable_interface() 31 31 6
A Main.get_links() 0 7 1
A Main.get_topology() 0 7 1
A Main.handle_interface_link_down() 0 10 4
A Main._get_links_dict() 0 4 1
A Main.handle_interfaces_created() 0 11 3
A Main.handle_link_liveness_status() 0 11 5
A Main.notify_switch_disabled() 0 5 1
B Main.add_links() 0 36 5
A Main.on_link_liveness_disabled() 0 5 1
A Main.handle_interface_deleted() 0 3 1
A Main.set_special_tags() 0 18 3
A Main._get_link_or_create() 0 14 3
A Main.on_topo_controller_upsert_switch() 0 4 1
A Main._get_link_from_interface() 0 6 3
A Main.add_link_metadata() 0 15 2
A Main.notify_metadata_changes() 0 21 4
A Main.notify_switch_links_status() 0 12 5
B Main._load_link() 0 35 6
A Main.delete_switch_metadata() 0 19 3
A Main.get_topo_controller() 0 4 1
A Main._get_topology_dict() 0 4 1
B Main.delete_switch() 0 23 7
A Main.handle_link_liveness_disabled() 0 13 3
A Main.on_add_links() 0 4 1
A Main._get_topology() 0 3 1
A Main.handle_interruption_end() 29 29 3

How to fix   Duplicated Code    Complexity   

Duplicated Code

Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.

Common duplication problems, and corresponding solutions are:

Complexity

 Tip:   Before tackling complexity, make sure that you eliminate any duplication first. This often can reduce the size of classes significantly.

Complex classes like build.main often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

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