sonic-buildimage/platform/mellanox/mlnx-sfpd/scripts/mlnx-sfpd

#!/usr/bin/env python
'''
This code is for a mlnx platform specific daemon, mlnx-sfpd.
Which listen to the SDK for the SFP change event and post the event to DB.
'''

from __future__ import print_function
import sys, errno
import os
import time
import syslog
import signal
import select
import json
import threading
from python_sdk_api.sx_api import *
from swsssdk import SonicV2Connector

VERSION = '1.0'

SYSLOG_IDENTIFIER = "mlnx-sfpd"

REDIS_HOSTIP = "127.0.0.1"

SDK_SFP_STATE_IN = 0x1
SDK_SFP_STATE_OUT = 0x2
STATUS_PLUGIN = '1'
STATUS_PLUGOUT = '0'
STATUS_UNKNOWN = '2'

SFPD_LIVENESS_EXPIRE_SECS = 30

SDK_DAEMON_READY_FILE = '/tmp/sdk_ready'

sfp_value_status_dict = {
        SDK_SFP_STATE_IN:  STATUS_PLUGIN,
        SDK_SFP_STATE_OUT: STATUS_PLUGOUT,
}

# ========================== Syslog wrappers ==========================
def log_info(msg, also_print_to_console=False):
    syslog.openlog(SYSLOG_IDENTIFIER)
    syslog.syslog(syslog.LOG_INFO, msg)
    syslog.closelog()

    if also_print_to_console:
        print(msg)

def log_warning(msg, also_print_to_console=False):
    syslog.openlog(SYSLOG_IDENTIFIER)
    syslog.syslog(syslog.LOG_WARNING, msg)
    syslog.closelog()

    if also_print_to_console:
        print(msg)

def log_error(msg, also_print_to_console=False):
    syslog.openlog(SYSLOG_IDENTIFIER)
    syslog.syslog(syslog.LOG_ERR, msg)
    syslog.closelog()

    if also_print_to_console:
        print(msg)

# ========================== MlnxSfpd class ==========================
class MlnxSfpd:
    ''' Listen to plugin/plugout cable events '''

    SX_OPEN_RETRIES = 30
    SX_OPEN_TIMEOUT = 5
    SELECT_TIMEOUT = 1

    def __init__(self):
        self.swid = 0
        self.running = False
        self.handle = None

        # Allocate SDK fd and user channel structures
        self.rx_fd_p = new_sx_fd_t_p()
        self.user_channel_p = new_sx_user_channel_t_p()
        self.state_db = SonicV2Connector(host=REDIS_HOSTIP)

        # Register our signal handlers
        signal.signal(signal.SIGHUP, self.signal_handler)
        signal.signal(signal.SIGINT, self.signal_handler)
        signal.signal(signal.SIGTERM, self.signal_handler)

    def signal_handler(self, signum, frame):
        if signum == signal.SIGHUP:
            log_info("Caught SIGHUP - ignoring...")
        elif signum == signal.SIGINT:
            log_info("Caught SIGINT - exiting...")
            self.running = False
        elif signum == signal.SIGTERM:
            log_info("Caught SIGINT - exiting...")
            self.running = False
        else:
            log_warning("Caught unhandled signal '{}'".format(signum))

    def initialize(self):
        self.state_db.connect("STATE_DB")

        swid_cnt_p = None

        try:
            # Wait for SDK daemon to be started with detect the sdk_ready file
            retry = 0
            while not os.path.exists(SDK_DAEMON_READY_FILE):  
                if retry >= self.SX_OPEN_RETRIES:
                    raise RuntimeError("SDK daemon failed to start after {} retries and {} seconds waiting, exiting..."
                        .format(retry, self.SX_OPEN_TIMEOUT * self.SX_OPEN_RETRIES))
                else:
                    log_info("SDK daemon not started yet, retry {} times".format(retry))
                    retry = retry + 1
                    time.sleep(self.SX_OPEN_TIMEOUT)

            # to make sure SDK daemon has started
            time.sleep(self.SX_OPEN_TIMEOUT)

            # After SDK daemon started, sx_api_open and sx_api_host_ifc_open is ready for call
            rc, self.handle = sx_api_open(None)
            if rc != SX_STATUS_SUCCESS:
                raise RuntimeError("failed to call sx_api_open with rc {}, exiting...".format(rc))

            rc = sx_api_host_ifc_open(self.handle, self.rx_fd_p)
            if rc != SX_STATUS_SUCCESS:
                raise RuntimeError("failed to call sx_api_host_ifc_open with rc {}, exiting...".format(rc))

            self.user_channel_p.type = SX_USER_CHANNEL_TYPE_FD
            self.user_channel_p.channel.fd = self.rx_fd_p

            # Wait for switch to be created and initialized inside SDK
            retry = 0
            swid_cnt_p = new_uint32_t_p()
            uint32_t_p_assign(swid_cnt_p, 0)
            swid_cnt = 0
            while True:
                if retry >= self.SX_OPEN_RETRIES:
                    raise RuntimeError("switch not created after {} retries and {} seconds waiting, exiting..."
                        .format(retry, self.SX_OPEN_RETRIES * self.SX_OPEN_TIMEOUT))
                else:
                    rc = sx_api_port_swid_list_get(self.handle, None, swid_cnt_p)
                    if rc == SX_STATUS_SUCCESS:
                        swid_cnt = uint32_t_p_value(swid_cnt_p)
                        if swid_cnt > 0:
                            delete_uint32_t_p(swid_cnt_p)
                            swid_cnt_p = None
                            break
                        else:
                            log_info("switch not created yet, swid_cnt {}, retry {} times and wait for {} seconds"
                                .format(swid_cnt, retry, self.SX_OPEN_TIMEOUT * retry))
                    else:
                        raise RuntimeError("sx_api_port_swid_list_get fail with rc {}, retry {} times and wait for {} seconds".
                            format(rc, retry, self.SX_OPEN_TIMEOUT * retry))

                    retry = retry + 1
                    time.sleep(self.SX_OPEN_TIMEOUT)

            # After switch was created inside SDK, sx_api_host_ifc_trap_id_register_set is ready to call
            rc = sx_api_host_ifc_trap_id_register_set(self.handle,
                                                    SX_ACCESS_CMD_REGISTER,
                                                    self.swid,
                                                    SX_TRAP_ID_PMPE,
                                                    self.user_channel_p)

            if rc != SX_STATUS_SUCCESS:
                raise RuntimeError("sx_api_host_ifc_trap_id_register_set failed with rc {}, exiting...".format(rc))

            self.running = True
        except Exception as e:
            log_error("mlnx-sfpd initialization failed due to {}, exiting...".format(repr(e)))
            if swid_cnt_p is not None:
                delete_uint32_t_p(swid_cnt_p)
            self.deinitialize()

    def deinitialize(self):
        # remove mlnx-sfpd liveness key in DB if not expired yet
        if self.state_db.exists("STATE_DB", "MLNX_SFPD_TASK|LIVENESS"):
            self.state_db.delete("STATE_DB", "MLNX_SFPD_TASK|LIVENESS")

        if self.handle is None:
            return

        # unregister trap id
        rc = sx_api_host_ifc_trap_id_register_set(self.handle,
                                                  SX_ACCESS_CMD_DEREGISTER,
                                                  self.swid,
                                                  SX_TRAP_ID_PMPE,
                                                  self.user_channel_p)
        if rc != SX_STATUS_SUCCESS:
            log_error("sx_api_host_ifc_trap_id_register_set exited with error, rc {}".format(rc))

        rc = sx_api_host_ifc_close(self.handle, self.rx_fd_p)
        if rc != SX_STATUS_SUCCESS:
            log_error("sx_api_host_ifc_close exited with error, rc {}".format(rc))

        rc = sx_api_close(self.handle)
        if rc != SX_STATUS_SUCCESS:
            log_error("sx_api_close exited with error, rc {}".format(rc))

    def run(self):

        while self.running:
            try:
                read, _, _ = select.select([self.rx_fd_p.fd], [], [], self.SELECT_TIMEOUT)
            except select.error as err:
                rc, msg = err
                if rc == errno.EAGAIN or rc == errno.EINTR:
                    continue
                else:
                    raise

            for fd in read:
                if fd == self.rx_fd_p.fd:
                    success, port_list, module_state = self.on_pmpe(self.rx_fd_p)
                    if not success:
                        raise RuntimeError("failed to read from {}".format(fd))

                    sfp_state = sfp_value_status_dict.get(module_state, STATUS_UNKNOWN)
                    if sfp_state == STATUS_UNKNOWN:
                        log_info("Got module state {} on port(s) {}, ignored".format(module_state, port_list))
                        continue

                    for port in port_list:
                        log_info("SFP on port {} state {}".format(port, sfp_state))
                        self.send_sfp_notification(str(port), sfp_state)

            self.update_sfpd_liveness_key(SFPD_LIVENESS_EXPIRE_SECS)

    def send_sfp_notification(self, port, state):
        sfp_notify = [port, state]
        msg = json.dumps(sfp_notify, separators=(',', ':'))
        self.state_db.publish('STATE_DB', 'TRANSCEIVER_NOTIFY', msg)

    def update_sfpd_liveness_key(self, timeout_secs):
        if not self.state_db.exists('STATE_DB', 'MLNX_SFPD_TASK|LIVENESS'):
            self.state_db.set('STATE_DB', 'MLNX_SFPD_TASK|LIVENESS', 'value', 'ok')
        self.state_db.expire('STATE_DB', 'MLNX_SFPD_TASK|LIVENESS', timeout_secs)

    def on_pmpe(self, fd_p):
        ''' on port module plug event handler '''

        # recv parameters
        pkt_size = 2000
        pkt_size_p = new_uint32_t_p()
        uint32_t_p_assign(pkt_size_p, pkt_size)
        pkt = new_uint8_t_arr(pkt_size)
        recv_info_p = new_sx_receive_info_t_p()
        pmpe_t = sx_event_pmpe_t()
        port_attributes_list = new_sx_port_attributes_t_arr(64)
        port_cnt_p = new_uint32_t_p()
        uint32_t_p_assign(port_cnt_p,64)
        label_port_list = []
        module_state = 0

        rc = sx_lib_host_ifc_recv(fd_p, pkt, pkt_size_p, recv_info_p)
        if rc != 0:
            log_error("sx_lib_host_ifc_recv exited with error, rc %d" % rc)
            status = False
        else:
            status = True
            pmpe_t = recv_info_p.event_info.pmpe
            port_list_size = pmpe_t.list_size
            logical_port_list = pmpe_t.log_port_list
            module_state = pmpe_t.module_state

            for i in xrange(port_list_size):
                logical_port = sx_port_log_id_t_arr_getitem(logical_port_list, i)
                rc = sx_api_port_device_get(self.handle, 1 , 0, port_attributes_list,  port_cnt_p)
                port_cnt = uint32_t_p_value(port_cnt_p)

                for i in xrange(port_cnt):
                    port_attributes = sx_port_attributes_t_arr_getitem(port_attributes_list,i)
                    if port_attributes.log_port == logical_port:
                        lable_port = port_attributes.port_mapping.module_port
                        break
                label_port_list.append(lable_port)

        delete_uint32_t_p(pkt_size_p)
        delete_uint8_t_arr(pkt)
        delete_sx_receive_info_t_p(recv_info_p)
        delete_sx_port_attributes_t_arr(port_attributes_list)
        delete_uint32_t_p(port_cnt_p)

        return status, label_port_list, module_state,


# main start
def main():
    log_info("mlnx-sfpd daemon started")

    sfpd = MlnxSfpd()
    try:
        sfpd.initialize()
        sfpd.run()
    except (RuntimeError, select.error) as err:
        log_error("error: {}".format(err))
    finally:
        sfpd.deinitialize()

    log_info("mlnx-sfpd daemon exited")


if __name__ == '__main__':
    main()