Passed
Pull Request — master (#181)
by
unknown
02:54
created

build.main   F

Complexity

Total Complexity 265

Size/Duplication

Total Lines 1379
Duplicated Lines 9.43 %

Test Coverage

Coverage 93.05%

Importance

Changes 0
Metric Value
eloc 1016
dl 130
loc 1379
ccs 844
cts 907
cp 0.9305
rs 1.536
c 0
b 0
f 0
wmc 265

90 Methods

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