Passed
Pull Request — master (#121)
by Vinicius
04:01
created

build.main.Main.on_switch_deleted()   A

Complexity

Conditions 4

Size

Total Lines 15
Code Lines 12

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 11
CRAP Score 4

Importance

Changes 0
Metric Value
cc 4
eloc 12
nop 2
dl 0
loc 15
ccs 11
cts 11
cp 1
crap 4
rs 9.8
c 0
b 0
f 0
1
"""NApp responsible to discover new switches and hosts."""
2 1
import struct
3 1
from threading import Lock
4
5 1
import httpx
6 1
import tenacity
7 1
from napps.kytos.of_core.msg_prios import of_msg_prio
8 1
from napps.kytos.of_lldp import constants, settings
9 1
from napps.kytos.of_lldp.managers import LivenessManager, LoopManager
10 1
from napps.kytos.of_lldp.managers.loop_manager import LoopState
11 1
from napps.kytos.of_lldp.utils import (get_cookie, try_to_gen_intf_mac,
12
                                       update_flow)
13 1
from pyof.foundation.basic_types import DPID, UBInt32
14 1
from pyof.foundation.network_types import LLDP, VLAN, Ethernet, EtherType
15 1
from pyof.v0x04.common.action import ActionOutput as AO13
16 1
from pyof.v0x04.common.port import PortNo as Port13
17 1
from pyof.v0x04.controller2switch.packet_out import PacketOut as PO13
18 1
from tenacity import (retry, retry_if_exception_type, stop_after_attempt,
19
                      wait_combine, wait_fixed, wait_random)
20
21 1
from kytos.core import KytosEvent, KytosNApp, log, rest
22 1
from kytos.core.exceptions import KytosTagError
23 1
from kytos.core.helpers import alisten_to, listen_to
24 1
from kytos.core.link import Link
25 1
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
26
                                 aget_json_or_400, get_json_or_400)
27 1
from kytos.core.retry import before_sleep
28 1
from kytos.core.switch import Switch
29
30 1
from .controllers import LivenessController
31
32
33 1
class Main(KytosNApp):
34
    """Main OF_LLDP NApp Class."""
35
36 1
    def setup(self):
37
        """Make this NApp run in a loop."""
38 1
        self.vlan_id = None
39 1
        self.polling_time = settings.POLLING_TIME
40 1
        if hasattr(settings, "FLOW_VLAN_VID"):
41 1
            self.vlan_id = settings.FLOW_VLAN_VID
42 1
        self.liveness_dead_multipler = settings.LIVENESS_DEAD_MULTIPLIER
43 1
        self.execute_as_loop(self.polling_time)
44 1
        self.loop_manager = LoopManager(self.controller)
45 1
        self.dead_interval = self.polling_time * self.liveness_dead_multipler
46 1
        self.liveness_controller = self.get_liveness_controller()
47 1
        self.liveness_controller.bootstrap_indexes()
48 1
        self.liveness_manager = LivenessManager(self.controller)
49 1
        self._liveness_ops_lock = Lock()
50 1
        Link.register_status_func(f"{self.napp_id}_liveness",
51
                                  LivenessManager.link_status_hook_liveness)
52 1
        status_reason_func = LivenessManager.link_status_reason_hook_liveness
53 1
        Link.register_status_reason_func(f"{self.napp_id}_liveness",
54
                                         status_reason_func)
55 1
        self.table_group = {"base": 0}
56
57 1
    @staticmethod
58 1
    def get_liveness_controller() -> LivenessController:
59
        """Get LivenessController."""
60
        return LivenessController()
61
62 1
    def execute(self):
63
        """Send LLDP Packets every 'POLLING_TIME' seconds to all switches."""
64 1
        switches = list(self.controller.switches.values())
65 1
        for switch in switches:
66 1
            try:
67 1
                of_version = switch.connection.protocol.version
68
            except AttributeError:
69
                of_version = None
70
71 1
            if not switch.is_connected():
72
                continue
73
74 1
            if of_version == 0x04:
75 1
                port_type = UBInt32
76 1
                local_port = Port13.OFPP_LOCAL
77
            else:
78
                # skip the current switch with unsupported OF version
79
                continue
80
81 1
            interfaces = list(switch.interfaces.values())
82 1
            for interface in interfaces:
83
                # Interface marked to receive lldp packet
84
                # Only send LLDP packet to active interface
85 1
                if (not interface.lldp or not interface.is_active()
86
                   or not interface.is_enabled()):
87
                    continue
88
                # Avoid the interface that connects to the controller.
89 1
                if interface.port_number == local_port:
90
                    continue
91
92 1
                lldp = LLDP()
93 1
                lldp.chassis_id.sub_value = DPID(switch.dpid)
94 1
                lldp.port_id.sub_value = port_type(interface.port_number)
95
96 1
                src_addr = try_to_gen_intf_mac(interface.address, switch.id,
97
                                               interface.port_number)
98 1
                ethernet = Ethernet()
99 1
                ethernet.ether_type = EtherType.LLDP
100 1
                ethernet.source = src_addr
101 1
                ethernet.destination = constants.LLDP_MULTICAST_MAC
102 1
                ethernet.data = lldp.pack()
103
                # self.vlan_id == None will result in a packet with no VLAN.
104 1
                ethernet.vlans.append(VLAN(vid=self.vlan_id))
105
106 1
                packet_out = self._build_lldp_packet_out(
107
                                    of_version,
108
                                    interface.port_number, ethernet.pack())
109
110 1
                if packet_out is None:
111
                    continue
112
113 1
                event_out = KytosEvent(
114
                    name='kytos/of_lldp.messages.out.ofpt_packet_out',
115
                    priority=of_msg_prio(packet_out.header.message_type.value),
116
                    content={
117
                            'destination': switch.connection,
118
                            'message': packet_out})
119
120 1
                self.controller.buffers.msg_out.put(event_out)
121 1
                log.debug(
122
                    "Sending a LLDP PacketOut to the switch %s",
123
                    switch.dpid)
124
125 1
                msg = 'Switch: %s (%s)'
126 1
                msg += ' Interface: %s'
127 1
                msg += ' -- LLDP PacketOut --'
128 1
                msg += ' Ethernet: eth_type (%s) | src (%s) | dst (%s) /'
129 1
                msg += ' LLDP: Switch (%s) | portno (%s)'
130
131 1
                log.debug(
132
                    msg,
133
                    switch.connection, switch.dpid,
134
                    interface.id, ethernet.ether_type,
135
                    ethernet.source, ethernet.destination,
136
                    switch.dpid, interface.port_number)
137
138 1
        self.try_to_publish_stopped_loops()
139 1
        self.liveness_manager.reaper(self.dead_interval)
140
141 1
    def load_liveness(self) -> None:
142
        """Load liveness."""
143 1
        enabled_intf_ids = {
144
            intf["id"]
145
            for intf in self.liveness_controller.get_enabled_interfaces()
146
        }
147 1
        intfs_to_enable = [
148
            intf
149
            for intf in self._get_interfaces()
150
            if intf.id in enabled_intf_ids
151
        ]
152 1
        self.liveness_manager.enable(*intfs_to_enable)
153
154 1
    def try_to_publish_stopped_loops(self):
155
        """Try to publish current stopped loops."""
156
        for dpid, port_pairs in self.loop_manager.get_stopped_loops().items():
157
            try:
158
                switch = self.controller.get_switch_by_dpid(dpid)
159
                for port_pair in port_pairs:
160
                    interface_a = switch.interfaces[port_pair[0]]
161
                    interface_b = switch.interfaces[port_pair[1]]
162
                    self.loop_manager.publish_loop_state(
163
                        interface_a, interface_b, LoopState.stopped.value
164
                    )
165
            except (KeyError, AttributeError) as exc:
166
                log.error("try_to_publish_stopped_loops failed with switch:"
167
                          f"{dpid}, port_pair: {port_pair}. {str(exc)}")
168
169 1
    @listen_to('kytos/topology.switch.(enabled|disabled)')
170 1
    def handle_lldp_flows(self, event):
171
        """Install or remove flows in a switch.
172
173
        Install a flow to send LLDP packets to the controller. The proactive
174
        flow is installed whenever a switch is enabled. If the switch is
175
        disabled the flow is removed.
176
177
        Args:
178
            event (:class:`~kytos.core.events.KytosEvent`):
179
                Event with new switch information.
180
181
        """
182
        self._handle_lldp_flows(event)
183
184 1
    @alisten_to("kytos/of_lldp.loop.action.log")
185 1
    async def on_lldp_loop_log_action(self, event: KytosEvent):
186
        """Handle LLDP loop log action."""
187
        interface_a = event.content["interface_a"]
188
        interface_b = event.content["interface_b"]
189
        await self.loop_manager.handle_log_action(interface_a, interface_b)
190
191 1
    @alisten_to("kytos/of_lldp.loop.action.disable")
192 1
    async def on_lldp_loop_disable_action(self, event: KytosEvent):
193
        """Handle LLDP loop disable action."""
194
        interface_a = event.content["interface_a"]
195
        interface_b = event.content["interface_b"]
196
        await self.loop_manager.handle_disable_action(interface_a, interface_b)
197
198 1
    @alisten_to("kytos/of_lldp.loop.stopped")
199 1
    async def on_lldp_loop_stopped(self, event: KytosEvent):
200
        """Handle LLDP loop stopped."""
201
        dpid = event.content["dpid"]
202
        port_pair = event.content["port_numbers"]
203
        try:
204
            switch = self.controller.get_switch_by_dpid(dpid)
205
            interface_a = switch.interfaces[port_pair[0]]
206
            interface_b = switch.interfaces[port_pair[1]]
207
            await self.loop_manager.handle_loop_stopped(interface_a,
208
                                                        interface_b)
209
        except (KeyError, AttributeError) as exc:
210
            log.error("on_lldp_loop_stopped failed with: "
211
                      f"{event.content} {str(exc)}")
212
213 1
    @alisten_to("kytos/topology.topology_loaded")
214 1
    async def on_topology_loaded(self, event):
215
        """Handle on topology loaded."""
216 1
        topology = event.content["topology"]
217 1
        await self.loop_manager.handle_topology_loaded(topology)
218 1
        self.load_liveness()
219
220 1
    @listen_to("kytos/topology.switch.deleted")
221 1
    def handle_switch_deleted(self, event: KytosEvent) -> None:
222
        """Handle on switch deleted."""
223
        with self._liveness_ops_lock:
224
            self.on_switch_deleted(event)
225
226 1
    def on_switch_deleted(self, event: KytosEvent) -> None:
227
        """Handle on switch deleted."""
228 1
        switch = event.content["switch"]
229 1
        found_intfs = []
230 1
        for intf in list(switch.interfaces.values()):
231 1
            if intf.id in self.liveness_manager.interfaces:
232 1
                found_intfs.append(intf)
233 1
        if found_intfs:
234 1
            intf_ids = [intf.id for intf in found_intfs]
235 1
            log.info(
236
                f"Disabling liveness on {switch} with interfaces "
237
                f" deleted {intf_ids} "
238
            )
239 1
            self.liveness_manager.disable(*found_intfs)
240 1
            self.liveness_controller.delete_interfaces(intf_ids)
241
242 1
    @listen_to("kytos/topology.interface.deleted")
243 1
    def handle_interface_deleted(self, event: KytosEvent) -> None:
244
        """Handle on interface deleted."""
245
        with self._liveness_ops_lock:
246
            self.on_interface_deleted(event)
247
248 1
    def on_interface_deleted(self, event: KytosEvent) -> None:
249
        """Handle on interface deleted."""
250 1
        intf = event.content["interface"]
251 1
        if intf.id in self.liveness_manager.interfaces:
252 1
            log.info(f"Disabling liveness on {intf} deleted")
253 1
            self.liveness_manager.disable(intf)
254 1
            self.liveness_controller.delete_interface(intf.id)
255
256 1
    @alisten_to("kytos/topology.switches.metadata.(added|removed)")
257 1
    async def on_switches_metadata_changed(self, event):
258
        """Handle on switches metadata changed."""
259
        switch = event.content["switch"]
260
        await self.loop_manager.handle_switch_metadata_changed(switch)
261
262 1
    def _handle_lldp_flows(self, event):
263
        """Install or remove flows in a switch.
264
265
        Install a flow to send LLDP packets to the controller. The proactive
266
        flow is installed whenever a switch is enabled. If the switch is
267
        disabled the flow is removed.
268
        """
269 1
        try:
270 1
            dpid = event.content['dpid']
271 1
            switch = self.controller.get_switch_by_dpid(dpid)
272 1
            of_version = switch.connection.protocol.version
273
        except AttributeError:
274
            of_version = None
275
276 1
        try:
277 1
            installed_flows = self.get_flows_by_switch(switch.id)
278 1
        except tenacity.RetryError as err:
279 1
            msg = f"Error: {err.last_attempt.exception()} when "\
280
                   "obtaining flows."
281 1
            log.error(msg)
282 1
            return
283 1
        except ValueError as err:
284 1
            log.error(f"Error when getting flows, error: {err}")
285 1
            return
286
287 1
        flow = None
288 1
        if ("switch.enabled" in event.name and not installed_flows or
289
                "switch.disabled" in event.name and installed_flows):
290 1
            flow = self._build_lldp_flow(of_version, get_cookie(switch.dpid))
291
292 1
        if flow:
293 1
            data = {'flows': [flow]}
294 1
            try:
295 1
                self.send_flow(switch, event.name, data=data)
296 1
            except tenacity.RetryError as err:
297 1
                msg = f"Error: {err.last_attempt.exception()} when"\
298
                      f" sending flows to {switch.id}, {data}"
299 1
                log.error(msg)
300
301
    # pylint: disable=unexpected-keyword-arg
302 1
    @retry(
303
        stop=stop_after_attempt(3),
304
        wait=wait_combine(wait_fixed(3), wait_random(min=2, max=7)),
305
        before_sleep=before_sleep,
306
        retry=retry_if_exception_type(httpx.RequestError),
307
        after=update_flow(),
308
    )
309 1
    def send_flow(self, switch, event_name, data=None):
310
        """Send flows to flow_manager to be installed/deleted"""
311 1
        endpoint = f'{settings.FLOW_MANAGER_URL}/flows/{switch.id}'
312 1
        client_error = {424, 404, 400}
313 1
        if event_name == 'kytos/topology.switch.enabled':
314 1
            for flow in data['flows']:
315 1
                flow.pop("cookie_mask", None)
316 1
            res = httpx.post(endpoint, json=data, timeout=10)
317 1
            if res.is_server_error or res.status_code in client_error:
318 1
                raise httpx.RequestError(res.text)
319 1
            self.use_vlan(switch)
320
321 1
        elif event_name == 'kytos/topology.switch.disabled':
322 1
            res = httpx.request("DELETE", endpoint, json=data, timeout=10)
323 1
            if res.is_server_error or res.status_code in client_error:
324
                raise httpx.RequestError(res.text)
325 1
            self.make_vlan_available(switch)
326
327 1
    def use_vlan(self, switch: Switch) -> None:
328
        """Use vlan from interface"""
329 1
        if self.vlan_id is None:
330 1
            return
331 1
        for interface_id in switch.interfaces:
332 1
            interface = switch.interfaces[interface_id]
333 1
            try:
334 1
                interface.use_tags(self.controller, self.vlan_id)
335 1
            except KytosTagError as err:
336 1
                log.error(err)
337
338 1
    def make_vlan_available(self, switch: Switch) -> None:
339
        """Makes vlan from interface available"""
340 1
        if self.vlan_id is None:
341 1
            return
342 1
        for interface_id in switch.interfaces:
343 1
            interface = switch.interfaces[interface_id]
344 1
            try:
345 1
                conflict = interface.make_tags_available(
346
                    self.controller, self.vlan_id
347
                )
348 1
                if conflict:
349 1
                    log.warning(f"Tags {conflict} was already available"
350
                                f" in {switch.id}:{interface_id}")
351 1
            except KytosTagError as err:
352 1
                log.error(err)
353
354 1
    @retry(
355
        stop=stop_after_attempt(3),
356
        wait=wait_combine(wait_fixed(3), wait_random(min=2, max=7)),
357
        before_sleep=before_sleep,
358
        retry=retry_if_exception_type(httpx.RequestError),
359
    )
360 1
    def get_flows_by_switch(self, dpid: str) -> dict:
361
        """Get of_lldp flows by switch"""
362 1
        start = settings.COOKIE_PREFIX << 56
363 1
        end = start | 0x00FFFFFFFFFFFFFF
364 1
        endpoint = f'{settings.FLOW_MANAGER_URL}/stored_flows?state='\
365
                   f'installed&cookie_range={start}&cookie_range={end}'\
366
                   f'&dpid={dpid}'
367 1
        res = httpx.get(endpoint)
368 1
        if res.is_server_error or res.status_code == 404:
369
            raise httpx.RequestError(res.text)
370 1
        if res.status_code == 400:
371 1
            raise ValueError(res.json()["description"])
372
        return res.json()
373
374 1
    @alisten_to('kytos/of_core.v0x04.messages.in.ofpt_packet_in')
375 1
    async def on_ofpt_packet_in(self, event):
376
        """Dispatch two KytosEvents to notify identified NNI interfaces.
377
378
        Args:
379
            event (:class:`~kytos.core.events.KytosEvent`):
380
                Event with an LLDP packet as data.
381
382
        """
383 1
        ethernet = self._unpack_non_empty(Ethernet, event.message.data)
384 1
        if ethernet.ether_type == EtherType.LLDP:
385 1
            try:
386 1
                lldp = self._unpack_non_empty(LLDP, ethernet.data)
387 1
                dpid = self._unpack_non_empty(DPID, lldp.chassis_id.sub_value)
388
            except struct.error:
389
                #: If we have a LLDP packet but we cannot unpack it, or the
390
                #: unpacked packet does not contain the dpid attribute, then
391
                #: we are dealing with a LLDP generated by someone else. Thus
392
                #: this packet is not useful for us and we may just ignore it.
393
                return
394
395 1
            switch_a = event.source.switch
396 1
            port_a = event.message.in_port
397 1
            switch_b = None
398 1
            port_b = None
399
400
            # in_port is currently an Int in v0x04.
401 1
            if isinstance(port_a, int):
402 1
                port_a = UBInt32(port_a)
403
404 1
            try:
405 1
                switch_b = self.controller.get_switch_by_dpid(dpid.value)
406 1
                port_type = UBInt32
407 1
                port_b = self._unpack_non_empty(port_type,
408
                                                lldp.port_id.sub_value)
409
            except AttributeError:
410
                log.debug("Couldn't find datapath %s.", dpid.value)
411
412
            # Return if any of the needed information are not available
413 1
            if not (switch_a and port_a and switch_b and port_b):
414
                return
415
416 1
            interface_a = switch_a.get_interface_by_port_no(port_a.value)
417 1
            interface_b = switch_b.get_interface_by_port_no(port_b.value)
418 1
            if not interface_a or not interface_b:
419 1
                return
420
421 1
            await self.loop_manager.process_if_looped(interface_a, interface_b)
422 1
            await self.liveness_manager.consume_hello_if_enabled(interface_a,
423
                                                                 interface_b)
424 1
            event_out = KytosEvent(name='kytos/of_lldp.interface.is.nni',
425
                                   content={'interface_a': interface_a,
426
                                            'interface_b': interface_b})
427 1
            await self.controller.buffers.app.aput(event_out)
428
429 1
    def notify_lldp_change(self, state, interface_ids):
430
        """Dispatch a KytosEvent to notify changes to the LLDP status."""
431 1
        content = {'attribute': 'LLDP',
432
                   'state': state,
433
                   'interface_ids': interface_ids}
434 1
        event_out = KytosEvent(name='kytos/of_lldp.network_status.updated',
435
                               content=content)
436 1
        self.controller.buffers.app.put(event_out)
437
438 1
    def publish_liveness_status(self, event_suffix, interfaces):
439
        """Dispatch a KytosEvent to publish liveness admin status."""
440 1
        content = {"interfaces": interfaces}
441 1
        name = f"kytos/of_lldp.liveness.{event_suffix}"
442 1
        event_out = KytosEvent(name=name, content=content)
443 1
        self.controller.buffers.app.put(event_out)
444
445 1
    def shutdown(self):
446
        """End of the application."""
447
        log.debug('Shutting down...')
448
449 1
    @staticmethod
450 1
    def _build_lldp_packet_out(version, port_number, data):
451
        """Build a LLDP PacketOut message.
452
453
        Args:
454
            version (int): OpenFlow version
455
            port_number (int): Switch port number where the packet must be
456
                forwarded to.
457
            data (bytes): Binary data to be sent through the port.
458
459
        Returns:
460
            PacketOut message for the specific given OpenFlow version, if it
461
                is supported.
462
            None if the OpenFlow version is not supported.
463
464
        """
465 1
        if version == 0x04:
466 1
            action_output_class = AO13
467 1
            packet_out_class = PO13
468
        else:
469 1
            log.info('Openflow version %s is not yet supported.', version)
470 1
            return None
471
472 1
        output_action = action_output_class()
473 1
        output_action.port = port_number
474
475 1
        packet_out = packet_out_class()
476 1
        packet_out.data = data
477 1
        packet_out.actions.append(output_action)
478
479 1
        return packet_out
480
481 1
    def _build_lldp_flow(self, version, cookie,
482
                         cookie_mask=0xffffffffffffffff):
483
        """Build a Flow message to send LLDP to the controller.
484
485
        Args:
486
            version (int): OpenFlow version.
487
488
        Returns:
489
            Flow dictionary message for the specific given OpenFlow version,
490
            if it is supported.
491
            None if the OpenFlow version is not supported.
492
493
        """
494 1
        flow = {}
495 1
        if version == 0x04:
496 1
            flow['actions'] = [{'action_type': 'output',
497
                                'port': Port13.OFPP_CONTROLLER}]
498
        else:
499 1
            return None
500
501 1
        match = {}
502 1
        self.set_flow_table_group_owner(flow)
503 1
        flow['priority'] = settings.FLOW_PRIORITY
504 1
        flow['cookie'] = cookie
505 1
        flow['cookie_mask'] = cookie_mask
506 1
        match['dl_type'] = EtherType.LLDP
507 1
        if self.vlan_id:
508 1
            match['dl_vlan'] = self.vlan_id
509 1
        flow['match'] = match
510
511 1
        return flow
512
513 1
    @staticmethod
514 1
    def _unpack_non_empty(desired_class, data):
515
        """Unpack data using an instance of desired_class.
516
517
        Args:
518
            desired_class (class): The class to be used to unpack data.
519
            data (bytes): bytes to be unpacked.
520
521
        Return:
522
            An instance of desired_class class with data unpacked into it.
523
524
        Raises:
525
            UnpackException if the unpack could not be performed.
526
527
        """
528 1
        obj = desired_class()
529
530 1
        if hasattr(data, 'value'):
531 1
            data = data.value
532
533 1
        obj.unpack(data)
534
535 1
        return obj
536
537 1
    def _get_data(self, request: Request) -> list:
538
        """Get request data."""
539 1
        data = get_json_or_400(request, self.controller.loop)
540 1
        return data.get('interfaces', [])
541
542 1
    def _get_interfaces(self):
543
        """Get all interfaces."""
544 1
        interfaces = []
545 1
        for switch in list(self.controller.switches.values()):
546 1
            interfaces += list(switch.interfaces.values())
547 1
        return interfaces
548
549 1
    @staticmethod
550 1
    def _get_interfaces_dict(interfaces):
551
        """Return a dict of interfaces."""
552 1
        return {inter.id: inter for inter in interfaces}
553
554 1
    def _get_lldp_interfaces(self):
555
        """Get interfaces enabled to receive LLDP packets."""
556 1
        return [inter.id for inter in self._get_interfaces() if inter.lldp]
557
558 1
    @rest('v1/interfaces', methods=['GET'])
559 1
    async def get_lldp_interfaces(self, _request: Request) -> JSONResponse:
560
        """Return all the interfaces that have LLDP traffic enabled."""
561 1
        return JSONResponse({"interfaces": self._get_lldp_interfaces()})
562
563 1
    @rest('v1/interfaces/disable', methods=['POST'])
564 1
    def disable_lldp(self, request: Request) -> JSONResponse:
565
        """Disables an interface to receive LLDP packets."""
566 1
        interface_ids = self._get_data(request)
567 1
        error_list = []  # List of interfaces that were not activated.
568 1
        changed_interfaces = []
569 1
        interface_ids = filter(None, interface_ids)
570 1
        interfaces = self._get_interfaces()
571 1
        intfs = []
572 1
        if not interfaces:
573
            raise HTTPException(404, detail="No interfaces were found.")
574 1
        interfaces = self._get_interfaces_dict(interfaces)
575 1
        for id_ in interface_ids:
576 1
            interface = interfaces.get(id_)
577 1
            if interface:
578 1
                interface.lldp = False
579 1
                changed_interfaces.append(id_)
580 1
                intfs.append(interface)
581
            else:
582 1
                error_list.append(id_)
583 1
        if changed_interfaces:
584 1
            self.notify_lldp_change('disabled', changed_interfaces)
585 1
            intf_ids = [intf.id for intf in intfs]
586 1
            with self._liveness_ops_lock:
587 1
                self.liveness_controller.disable_interfaces(intf_ids)
588 1
                self.liveness_manager.disable(*intfs)
589 1
            self.publish_liveness_status("disabled", intfs)
590 1
        if not error_list:
591 1
            return JSONResponse(
592
                "All the requested interfaces have been disabled.")
593
594
        # Return a list of interfaces that couldn't be disabled
595 1
        msg_error = "Some interfaces couldn't be found and deactivated: "
596 1
        return JSONResponse({msg_error: error_list}, status_code=400)
597
598 1
    @rest('v1/interfaces/enable', methods=['POST'])
599 1
    def enable_lldp(self, request: Request) -> JSONResponse:
600
        """Enable an interface to receive LLDP packets."""
601 1
        interface_ids = self._get_data(request)
602 1
        error_list = []  # List of interfaces that were not activated.
603 1
        changed_interfaces = []
604 1
        interface_ids = filter(None, interface_ids)
605 1
        interfaces = self._get_interfaces()
606 1
        if not interfaces:
607
            raise HTTPException(404, detail="No interfaces were found.")
608 1
        interfaces = self._get_interfaces_dict(interfaces)
609 1
        for id_ in interface_ids:
610 1
            interface = interfaces.get(id_)
611 1
            if interface:
612 1
                interface.lldp = True
613 1
                changed_interfaces.append(id_)
614
            else:
615 1
                error_list.append(id_)
616 1
        if changed_interfaces:
617 1
            self.notify_lldp_change('enabled', changed_interfaces)
618 1
        if not error_list:
619 1
            return JSONResponse(
620
                "All the requested interfaces have been enabled.")
621
622
        # Return a list of interfaces that couldn't be enabled
623 1
        msg_error = "Some interfaces couldn't be found and activated: "
624 1
        return JSONResponse({msg_error: error_list}, status_code=400)
625
626 1
    @rest("v1/liveness/enable", methods=["POST"])
627 1
    def enable_liveness(self, request: Request) -> JSONResponse:
628
        """Enable liveness link detection on interfaces."""
629 1
        intf_ids = self._get_data(request)
630 1
        if not intf_ids:
631
            raise HTTPException(400, "Interfaces payload is empty")
632 1
        interfaces = {intf.id: intf for intf in self._get_interfaces()}
633 1
        diff = set(intf_ids) - set(interfaces.keys())
634 1
        if diff:
635
            raise HTTPException(404, f"Interface IDs {diff} not found")
636
637 1
        intfs = [interfaces[_id] for _id in intf_ids]
638 1
        non_lldp = [intf.id for intf in intfs if not intf.lldp]
639 1
        if non_lldp:
640
            msg = f"Interface IDs {non_lldp} don't have LLDP enabled"
641
            raise HTTPException(400, msg)
642 1
        with self._liveness_ops_lock:
643 1
            self.liveness_controller.enable_interfaces(intf_ids)
644 1
            self.liveness_manager.enable(*intfs)
645 1
            self.publish_liveness_status("enabled", intfs)
646 1
        return JSONResponse({})
647
648 1
    @rest("v1/liveness/disable", methods=["POST"])
649 1
    def disable_liveness(self, request: Request) -> JSONResponse:
650
        """Disable liveness link detection on interfaces."""
651 1
        intf_ids = self._get_data(request)
652 1
        if not intf_ids:
653
            raise HTTPException(400, "Interfaces payload is empty")
654
655 1
        interfaces = {intf.id: intf for intf in self._get_interfaces()}
656 1
        diff = set(intf_ids) - set(interfaces.keys())
657 1
        if diff:
658
            raise HTTPException(404, f"Interface IDs {diff} not found")
659
660 1
        intfs = [interfaces[_id] for _id in intf_ids if _id in interfaces]
661 1
        with self._liveness_ops_lock:
662 1
            self.liveness_controller.disable_interfaces(intf_ids)
663 1
            self.liveness_manager.disable(*intfs)
664 1
            self.publish_liveness_status("disabled", intfs)
665 1
        return JSONResponse({})
666
667 1
    @rest("v1/liveness/", methods=["GET"])
668 1
    async def get_liveness_interfaces(self, request: Request) -> JSONResponse:
669
        """Get liveness interfaces."""
670 1
        args = request.query_params
671 1
        interface_id = args.get("interface_id")
672 1
        if interface_id:
673
            status, last_hello_at = self.liveness_manager.get_interface_status(
674
                interface_id
675
            )
676
            if not status:
677
                return {"interfaces": []}, 200
678
            body = {
679
                "interfaces": [
680
                    {
681
                        "id": interface_id,
682
                        "status": status,
683
                        "last_hello_at": last_hello_at,
684
                    }
685
                ]
686
            }
687
            return JSONResponse(body)
688 1
        interfaces = []
689 1
        for interface_id in list(self.liveness_manager.interfaces.keys()):
690
            status, last_hello_at = self.liveness_manager.get_interface_status(
691
                interface_id
692
            )
693
            interfaces.append({"id": interface_id, "status": status,
694
                              "last_hello_at": last_hello_at})
695 1
        return JSONResponse({"interfaces": interfaces})
696
697 1
    @rest("v1/liveness/pair", methods=["GET"])
698 1
    async def get_liveness_interface_pairs(self,
699
                                           _request: Request) -> JSONResponse:
700
        """Get liveness interface pairs."""
701 1
        pairs = []
702 1
        for entry in list(self.liveness_manager.liveness.values()):
703
            lsm = entry["lsm"]
704
            pair = {
705
                "interface_a": {
706
                    "id": entry["interface_a"].id,
707
                    "status": lsm.ilsm_a.state,
708
                    "last_hello_at": lsm.ilsm_a.last_hello_at,
709
                },
710
                "interface_b": {
711
                    "id": entry["interface_b"].id,
712
                    "status": lsm.ilsm_b.state,
713
                    "last_hello_at": lsm.ilsm_b.last_hello_at,
714
                },
715
                "status": lsm.state
716
            }
717
            pairs.append(pair)
718 1
        return JSONResponse({"pairs": pairs})
719
720 1
    @rest('v1/polling_time', methods=['GET'])
721 1
    async def get_time(self, _request: Request) -> JSONResponse:
722
        """Get LLDP polling time in seconds."""
723 1
        return JSONResponse({"polling_time": self.polling_time})
724
725 1
    @rest('v1/polling_time', methods=['POST'])
726 1
    async def set_time(self, request: Request) -> JSONResponse:
727
        """Set LLDP polling time."""
728
        # pylint: disable=attribute-defined-outside-init
729 1
        try:
730 1
            payload = await aget_json_or_400(request)
731 1
            polling_time = int(payload['polling_time'])
732 1
            if polling_time <= 0:
733
                msg = f"invalid polling_time {polling_time}, " \
734
                        "must be greater than zero"
735
                raise HTTPException(400, detail=msg)
736 1
            self.polling_time = polling_time
737 1
            self.execute_as_loop(self.polling_time)
738 1
            log.info("Polling time has been updated to %s"
739
                     " second(s), but this change will not be saved"
740
                     " permanently.", self.polling_time)
741 1
            return JSONResponse("Polling time has been updated.")
742 1
        except (ValueError, KeyError) as error:
743 1
            msg = f"This operation is not completed: {error}"
744 1
            raise HTTPException(400, detail=msg) from error
745
746 1
    def set_flow_table_group_owner(self,
747
                                   flow: dict,
748
                                   group: str = "base") -> dict:
749
        """Set owner, table_group and table_id"""
750 1
        flow["table_id"] = self.table_group[group]
751 1
        flow["owner"] = "of_lldp"
752 1
        flow["table_group"] = group
753 1
        return flow
754
755
    # pylint: disable=attribute-defined-outside-init
756 1
    @alisten_to("kytos/of_multi_table.enable_table")
757 1
    async def on_table_enabled(self, event):
758
        """Handle a recently table enabled.
759
        of_lldp only allows "base" as flow group
760
        """
761 1
        table_group = event.content.get("of_lldp", None)
762 1
        if not table_group:
763
            return
764 1
        for group in table_group:
765 1
            if group not in settings.TABLE_GROUP_ALLOWED:
766 1
                log.error(f'The table group "{group}" is not allowed for '
767
                          f'of_lldp. Allowed table groups are '
768
                          f'{settings.TABLE_GROUP_ALLOWED}')
769 1
                return
770 1
        self.table_group.update(table_group)
771 1
        content = {"group_table": self.table_group}
772 1
        event_out = KytosEvent(name="kytos/of_lldp.enable_table",
773
                               content=content)
774
        await self.controller.buffers.app.aput(event_out)
775