Test Failed
Pull Request — master (#179)
by
unknown
03:33
created

build.main   F

Complexity

Total Complexity 232

Size/Duplication

Total Lines 1275
Duplicated Lines 9.41 %

Test Coverage

Coverage 91.71%

Importance

Changes 0
Metric Value
eloc 938
dl 120
loc 1275
ccs 741
cts 808
cp 0.9171
rs 1.662
c 0
b 0
f 0
wmc 232

86 Methods

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