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

#!/usr/bin/env python
'''
This code is for a mlnx platform specific tool, issu.
This tool provides an CLI interface to interact with SDK ISSU module
'''

from __future__ import print_function

import sys
import os
import re
import errno
import syslog
import argparse

from functools import wraps

from python_sdk_api import sx_api

# ========================== Constants ===============================
SDK_VERSION_PATTERN = r'(\d+)\.(\d+)\.(.*)'
SYSLOG_IDENTIFIER = "ISSU"


# Flag that indicates whether to print logs to stdout
verbose = False


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

    if verbose:
        print(msg)

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

    if verbose:
        print(msg)

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

    print(msg, file=sys.stderr)


# ========================== Global functions =========================
def with_sdk_handle(func):
    """ A decorator for @func that use sx api
    that gets a SDK handler, calls func(handler, *args, **kwargs)
    and then closes the handler regardless of func failure"""

    @wraps(func)
    def wrapped(*args, **kwargs):
        log_info("opening sdk")
        rc, handle = sx_api.sx_api_open(None)
        log_info("sx_api_open handle: 0x%x , rc %d " % ( handle, rc) )

        if rc != sx_api.SX_STATUS_SUCCESS:
            log_error("failed to open api handle. Please check that SDK is running")
            sys.exit(errno.EACCES)

        try:
            res = func(handle, *args, **kwargs)
        finally:
            log_info("closing sdk handle")
            rc = sx_api.sx_api_close(handle)

            if rc != sx_api.SX_STATUS_SUCCESS:
                log_error("failed to close api handle")

        return res

    return wrapped


def check_sdk_version_pattern(sdk_version):
    """Checker for @sdk_version"""

    if not re.match(SDK_VERSION_PATTERN, sdk_version):
        raise argparse.ArgumentTypeError("{} is an invalid SDK version string".format(sdk_version))

    return sdk_version


@with_sdk_handle
def check_issu_upgrade_to_sdk_version(handle, new_sdk):
    """This function checks whether ISSU upgrade to @new_sdk version is posible"""

    version = sx_api.new_sx_api_sx_sdk_versions_t_p()
    rc = sx_api.sx_api_sx_sdk_version_get(handle, version)

    if rc != sx_api.SX_STATUS_SUCCESS:
        log_error("failed to get current SDK version")
        sys.exit(errno.EACCES)

    current_sdk = version.sx_sdk

    succeed = True

    log_info('check ISSU upgrade: current SDK: {}, new SDK: {}, check succeed: {}'.format(current_sdk, new_sdk, succeed))

    return succeed


@with_sdk_handle
def issu_start(handle):
    """This function calls ISSU start API"""

    log_info("call ISSU start")
    rc = sx_api.sx_api_issu_start_set(handle)

    if rc != sx_api.SX_STATUS_SUCCESS:
        log_error("failed to execute ISSU start API")
        sys.exit(errno.EACCES)

@with_sdk_handle
def issu_end(handle):
    """This function calls ISSU end API"""

    log_info("call ISSU end")
    rc = sx_api.sx_api_issu_end_set(handle)

    if rc != sx_api.SX_STATUS_SUCCESS:
        log_error("failed to execute ISSU end API")
        sys.exit(errno.EACCES)


def get_parser():
    """This function creates an argument parser"""

    parser = argparse.ArgumentParser()
    parser.add_argument('-c', '--check', nargs=1, action='store', default=None, type=check_sdk_version_pattern,
        help='Check if ISSU upgrade is supported to new SDK version')
    parser.add_argument('-s', '--start', action='store_true', help='Call ISSU start API')
    parser.add_argument('-e', '--end', action='store_true', help='Call ISSU end API')
    parser.add_argument('-v', '--verbose', action='store_true', default=False)

    return parser


def main():

    global verbose

    parser = get_parser()
    args = parser.parse_args()

    verbose = args.verbose

    if args.check is not None:
        new_sdk = args.check[0]
        is_supported = check_issu_upgrade_to_sdk_version(new_sdk)

        if verbose:
            print('SDK upgrade is{}supported'.format(' ' if is_supported else ' not '))

        if not is_supported:
            sys.exit(1)
    elif args.start:
        issu_start()
    elif args.end:
        issu_end()
    else:
        parser.parse_args(['-h'])


if __name__ == '__main__':
    main()