Passed
Pull Request — master (#99)
by
unknown
10:21 queued 02:59
created

build.main.Main.on_lldp_loop_detected()   A

Complexity

Conditions 1

Size

Total Lines 7
Code Lines 6

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 2
CRAP Score 1.2963

Importance

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