Passed
Pull Request — master (#286)
by Aldo
09:06 queued 03:57
created

build.main   F

Complexity

Total Complexity 296

Size/Duplication

Total Lines 1489
Duplicated Lines 4.3 %

Test Coverage

Coverage 92.75%

Importance

Changes 0
Metric Value
eloc 1114
dl 64
loc 1489
ccs 908
cts 979
cp 0.9275
rs 1.1439
c 0
b 0
f 0
wmc 296

92 Methods

Rating   Name   Duplication   Size   Complexity  
A Main._get_switches_dict() 0 12 3
A Main._get_links_dict() 0 4 1
A Main.shutdown() 0 3 1
A Main.execute() 0 3 1
A Main._get_metadata() 0 7 2
A Main.get_topo_controller() 0 4 1
A Main.handle_link_down() 0 8 3
A Main.load_interfaces_tags_values() 0 20 4
A Main.on_add_links() 0 4 1
B Main.disable_switch() 0 26 6
A Main.enable_switch() 0 15 2
B Main.load_topology() 0 35 5
A Main.add_switch_metadata() 0 14 2
A Main.get_interfaces() 0 10 3
A Main.get_switches() 0 4 1
A Main.setup() 0 25 1
B Main._load_switch() 0 43 4
A Main.get_switch_metadata() 0 9 2
A Main.get_topology() 0 7 1
A Main._load_link() 0 17 3
A Main.delete_switch_metadata() 0 19 3
A Main._get_topology_dict() 0 4 1
C Main.delete_switch() 0 50 11
A Main._get_topology() 0 4 1
A Main.on_notify_port_created() 0 4 1
A Main.notify_port_created() 0 5 1
A Main.get_intf_usage() 0 15 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.disable_link() 0 16 4
C Main.delete_link() 0 35 9
C Main.enable_interface() 0 43 11
A Main.link_status_hook_link_up_timer() 0 13 3
A Main.on_notify_link_up_if_status() 0 6 1
B Main.delete_interface_metadata() 0 31 5
A Main.add_interface_metadata() 0 21 3
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 36 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 36 8
A Main.notify_switch_enabled() 0 5 1
A Main.handle_interface_created() 0 11 2
A Main.link_status_mismatched() 0 5 2
A Main.delete_tag_range() 0 16 3
A Main.get_interface_metadata() 0 17 3
A Main.handle_lldp_status_updated() 0 15 4
A Main.delete_link_metadata() 0 19 4
A Main.on_lldp_status_updated() 0 4 1
A Main.notify_topology_update() 0 9 1
A Main.on_interface_deleted() 0 4 1
A Main.notify_link_up_if_status() 0 19 5
B Main.enable_link() 0 22 6
A Main.on_new_switch() 0 8 1
A Main.handle_new_switch() 0 9 2
A Main.get_flows_by_switch() 0 14 3
A Main.notify_link_enabled_state() 0 7 1
C Main.get_flow_id_by_intf() 0 24 11
A Main.handle_connection_lost() 0 7 2
A Main._notify_interface_link_status() 0 15 4
A Main.set_tag_range() 0 21 4
A Main.handle_on_interface_tags() 0 7 1
B Main.delete_interface() 0 28 5
A Main._delete_interface() 0 7 1
C Main.disable_interface() 0 43 10
A Main.get_links() 0 7 1
A Main.on_interruption() 0 11 3
A Main.notify_interface_status() 0 5 1
A Main.handle_interface_link_down() 0 14 4
A Main.handle_interfaces_created() 0 11 3
B Main.handle_link_liveness_status() 0 15 6
A Main.notify_switch_disabled() 0 5 1
B Main.add_links() 0 23 6
B Main.on_link_liveness() 0 21 6
A Main.handle_interface_deleted() 0 10 2
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 3
A Main.notify_metadata_changes() 0 21 4
A Main.notify_switch_links_status() 0 12 5
A Main.detect_mismatched_link() 0 7 4
A Main.handle_link_liveness_disabled() 0 14 4
A Main.get_link_metadata() 0 9 2
A Main.get_latest_topology() 0 4 2
B Main.handle_interruption_start() 32 32 6
B Main.handle_interruption_end() 32 32 6

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