path: root/ironic_python_agent/hardware_managers/nvidia/nvidia_fw_update.py

# Copyright 2022 Nvidia
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import re
import shutil
import tempfile
from urllib import error as urlError
from urllib.parse import urlparse
from urllib import request

from ironic_lib.common.i18n import _
from ironic_lib.exception import IronicException
from oslo_concurrency import processutils
from oslo_log import log
from oslo_utils import fileutils

from ironic_python_agent import utils

FW_VERSION_REGEX = r'FW Version:\s*\t*(?P<fw_ver>\d+\.\d+\.\d+)'
RUNNING_FW_VERSION_REGEX = \
    r'FW Version\(Running\):\s*\t*(?P<fw_ver>\d+\.\d+\.\d+)'
ARRAY_PARAM_REGEX = r'(?P<param_name>\w+)\[((?P<index>\d+)|' \
                    r'((?P<first_index>\d+)\.\.(?P<last_index>\d+)))\]'
ARRAY_PARAM_VALUE_REGEX = r'Array\[(?P<first_index>\d+)' \
                          r'\.\.(?P<last_index>\d+)\]'
PSID_REGEX = r'PSID:\s*\t*(?P<psid>\w+)'
NETWORK_DEVICE_REGEX = r'02\d\d'
LOG = log.getLogger()

"""
Example of Nvidia NIC Firmware images list:
[
  {
    "url": "file:///firmware_images/fw1.bin",
    "checksum": "a94e683ea16d9ae44768f0a65942234d",
    "checksumType": "md5",
    "componentFlavor": "MT_0000000540",
    "version": "24.34.1002"
  },
  {
    "url": "http://10.10.10.10/firmware_images/fw2.bin",
    "checksum": "a94e683ea16d9ae44768f0a65942234c",
    "checksumType": "sha512",
    "componentFlavor": "MT_0000000652",
    "version": "24.34.1002"
  }
]

Example of Nvidia NIC Firmware settings list:
[
  {
    "deviceID": "1017",
    "globalConfig": {
      "NUM_OF_VFS": 127,
      "SRIOV_EN": True
    },
    "function0Config": {
      "PF_TOTAL_SF": 500
    },
    "function1Config": {
      "PF_TOTAL_SF": 600
    }
  },
  {
    "deviceID": "101B",
    "globalConfig": {
      "NUM_OF_VFS": 127,
      "SRIOV_EN": True
    },
    "function0Config": {
      "PF_TOTAL_SF": 500
    },
    "function1Config": {
      "PF_TOTAL_SF": 600
    }
  }
]
"""


def check_prereq():
    """Check that all needed tools are available in the system.

    :returns:   None
    :raises:    processutils.ProcessExecutionError
    """
    try:
        # check for mstflint
        utils.execute('mstflint', '-v')
        # check for mstconfig
        utils.execute('mstconfig', '-v')
        # check for mstfwreset
        utils.execute('mstfwreset', '-v')
        # check for lspci
        utils.execute('lspci', '--version')
    except processutils.ProcessExecutionError as e:
        LOG.error('Failed Prerequisite check. %s', e)
        raise e


class InvalidFirmwareImageConfig(IronicException):
    _msg_fmt = _('Invalid firmware image config: %(error_msg)s')


class InvalidFirmwareSettingsConfig(IronicException):
    _msg_fmt = _('Invalid firmware settings config: %(error_msg)s')


class MismatchChecksumError(IronicException):
    _msg_fmt = _('Mismatch Checksum for the firmware image: %(error_msg)s')


class MismatchComponentFlavor(IronicException):
    _msg_fmt = _('Mismatch Component Flavor: %(error_msg)s')


class MismatchFWVersion(IronicException):
    _msg_fmt = _('Mismatch Firmware version: %(error_msg)s')


class DuplicateComponentFlavor(IronicException):
    _msg_fmt = _('Duplicate Component Flavor for the firmware image: '
                 '%(error_msg)s')


class DuplicateDeviceID(IronicException):
    _msg_fmt = _('Duplicate Device ID for firmware settings: '
                 '%(error_msg)s')


class UnSupportedConfigByMstflintPackage(IronicException):
    _msg_fmt = _('Unsupported config by mstflint package: %(error_msg)s')


class UnSupportedConfigByFW(IronicException):
    _msg_fmt = _('Unsupported config by Firmware: %(error_msg)s')


class InvalidURLScheme(IronicException):
    _msg_fmt = _('Invalid URL Scheme: %(error_msg)s')


class NvidiaNicFirmwareOps(object):
    """Perform various Firmware related operations on nic device"""

    def __init__(self, dev):
        self.dev = dev
        self.dev_info = {}

    def parse_mstflint_query_output(out):
        """Parse Mstflint query output

        For now just extract 'FW Version' and 'PSID'
        :param out: string, mstflint query output
        :returns:   dict of query attributes
        """
        query_info = {}
        for line in out.split('\n'):
            line = line.strip()
            fw_ver = re.match(FW_VERSION_REGEX, line)
            running_fw_ver = re.match(RUNNING_FW_VERSION_REGEX, line)
            psid = re.match(PSID_REGEX, line)
            if fw_ver is not None:
                query_info['fw_ver'] = fw_ver.group('fw_ver')
            if running_fw_ver is not None:
                query_info['running_fw_ver'] = running_fw_ver.group('fw_ver')
            if psid is not None:
                query_info['psid'] = psid.group('psid')
        return query_info

    def _query_device(self, force=False):
        """Get firmware information from nvidia nic device

        :param force:   bool, force device query, even if query was executed in
                        previous calls.
        :returns:       dict of firmware image attributes
        :raises:        processutils.ProcessExecutionError
        """
        if not force and self.dev_info.get('device', '') == self.dev:
            return self.dev_info
        try:
            cmd = ('mstflint', '-d', self.dev, '-qq', 'query')
            out, _r = utils.execute(*cmd)
        except processutils.ProcessExecutionError as e:
            LOG.error('Failed to query firmware of device %s: %s',
                      self.dev, e)
            raise e
        self.dev_info = NvidiaNicFirmwareOps.parse_mstflint_query_output(out)
        self.dev_info['device'] = self.dev
        return self.dev_info

    def get_nic_psid(self):
        """Get the psid of nvidia nic device

        :returns:   string, the psid of the nic device
        """
        return self._query_device().get('psid')

    def is_image_changed(self):
        """Check if image changed and nic device requires firmware reset

        before applying any configurations on the device.
        Currently the reset happens if image was changed
        :returns:    bool, True if image changed
        """
        self._query_device(force=True)
        is_image_changed = 'running_fw_ver' in self.dev_info and \
                           self.dev_info['running_fw_ver'] != \
                           self.dev_info['fw_ver']
        return is_image_changed

    def _need_update(self, fw_version):
        """Check if nic device requires firmware update

        :param fw_version:  string, the firmware version of image
        :returns:           bool, True if update is needed
        """
        self._query_device(force=True)
        LOG.info('Device firmware version: %s , Image firmware version: %s',
                 self.dev_info['fw_ver'], fw_version)
        return self.dev_info['fw_ver'] != fw_version

    def _burn_firmware(self, image_path):
        """Burn firmware on device

        :param image_path:  string, firmware binary file path
        :returns:           None
        :raises:            processutils.ProcessExecutionError
        """
        LOG.info('Updating firmware image (%s) for device: %s',
                 image_path, self.dev)
        try:
            cmd = ('mstflint', '-d', self.dev, '-i', image_path,
                   '-y', 'burn')
            utils.execute(*cmd)
        except processutils.ProcessExecutionError as e:
            LOG.error('Failed to update firmware image for device %s, %s',
                      self.dev, e)
            raise e
        LOG.info('Device %s: firmware image successfully updated.', self.dev)

    def reset_device(self, raise_exception=False):
        """Reset nvidia nic to load the new firmware image

        :returns:   None
        :raises:    processutils.ProcessExecutionError
        """
        LOG.info('Device %s: Performing firmware reset.', self.dev)
        cmd = ('mstfwreset', '-d', self.dev, '-y', '--sync', '1', 'reset')
        try:
            utils.execute(*cmd)
            LOG.info('Device %s: Firmware successfully reset.', self.dev)
        except processutils.ProcessExecutionError as e:
            LOG.error('Failed to reset device %s %s', self.dev, e)
            if raise_exception:
                raise e

    def fw_update_if_needed(self, version, image_path):
        """Update firmware if the current version not equal image version

        :param version:     string, the firmware version of image
        :param image_path:  string, the firmware image path
        :returns:           None
        """
        if self._need_update(version):
            if 'running_fw_ver' in self.dev_info:
                self.reset_device(raise_exception=True)
            self._burn_firmware(image_path)
        else:
            LOG.info('Firmware update is not required for Device.')


class NvidiaNic(object):
    """A class of nvidia nic contains pci, device ID,  device PSID and

    an instance of NvidiaNicFirmwareOps
    """

    def __init__(self, dev_pci, dev_id, dev_psid, dev_ops):
        self.dev_pci = dev_pci
        self.dev_id = dev_id
        self.dev_psid = dev_psid
        self.dev_ops = dev_ops


class NvidiaNics(object):
    """Discover and retrieve Nvidia Nics on the system.

    Can be used as an iterator once discover has been called.
    """

    def __init__(self):
        self._devs = []
        self._devs_psids = []
        self._dev_ids = []

    def discover(self):
        """Discover Nvidia Nics in the system.

        :returns:   None
        :raises:    processutils.ProcessExecutionError
        """
        if len(self._devs) > 0:
            return self._devs
        devs = []

        cmd = ('lspci', '-Dn', '-d', '15b3:')
        try:
            out, _r = utils.execute(*cmd)
        except processutils.ProcessExecutionError as e:
            LOG.error('Exception occurred while discovering Nvidia Nics %s',
                      e)
            raise e
        for line in out.strip().split('\n'):
            if not line:
                continue
            dev_class = line.split()[1].split(':')[0]
            if not re.match(NETWORK_DEVICE_REGEX, dev_class):
                continue
            dev_pci = line.split()[0]
            dev_id = line.split('15b3:')[1].split()[0]
            dev_ops = NvidiaNicFirmwareOps(dev_pci)
            dev_psid = dev_ops.get_nic_psid()
            self._dev_ids.append(dev_id)
            self._devs_psids.append(dev_psid)
            devs.append(NvidiaNic(dev_pci, dev_id, dev_psid, dev_ops))
        self._devs = devs

    def get_psids_list(self):
        """Get a list of PSIDs of Nvidia Nics in the system.

        :returns:   list of PSIDs of Nvidia Nics in the system
        """
        return set(self._devs_psids)

    def get_ids_list(self):
        """Get a list of IDs of Nvidia Nics in the system.

        :returns:   list of IDs of Nvidia Nics in the system
        """
        return set(self._dev_ids)

    def __iter__(self):
        return self._devs.__iter__()


class NvidiaNicFirmwareBinary(object):
    """A class of nvidia nic firmware binary which manages the binary

    firmware image, downloads it, validates it and provides its path on the
    system
    """

    def __init__(self, url, checksum, checksum_type,
                 component_flavor, version):
        self.url = url
        self.checksum = checksum
        self.checksum_type = checksum_type
        self.psid = component_flavor
        self.version = version
        self.image_info = {}
        self._process_url()
        self._validate_image_psid()
        self._validate_image_firmware_version()
        self._validate_image_checksum()

    def __del__(self):
        self._cleanup_file()

    def _cleanup_file(self):
        """Delete the temporary downloaded firmware image if exist in cleanup

        :returns:   None
        """
        if os.path.exists(os.path.dirname(self.dest_file_path)):
            try:
                shutil.rmtree(os.path.dirname(self.dest_file_path))
            except Exception as e:
                LOG.error('Failed to remove temporary directory for FW '
                          'binary: %s', e)

    def _download_file_based_fw(self):
        """Download the firmware image file from the provided file url (move)

        :returns:    None
        :raises:    Exception
        """
        src_file = self.parsed_url.path
        try:
            LOG.info('Moving file: %s to %s', self.url,
                     self.dest_file_path)
            shutil.move(src_file, self.dest_file_path)
        except Exception as e:
            LOG.error('Failed to move file: %s, %s', src_file, e)
            raise e

    def _download_http_based_fw(self):
        """Download the firmware image file from the provided url

        :returns:   None
        :raises:    urlError.HTTPError
        """
        try:
            LOG.info('Downloading file: %s to %s', self.url,
                     self.dest_file_path)
            url_data = request.urlopen(self.url)
        except urlError.URLError as url_error:
            LOG.error('Failed to open URL data: %s', url_error)
            raise url_error
        except urlError.HTTPError as http_error:
            LOG.error('Failed to download data: %s', http_error)
            raise http_error
        with open(self.dest_file_path, 'wb') as f:
            f.write(url_data.read())

    def _process_url(self):
        """Process the firmware url and download the image to a temporary

        destination in the system.
        The supported firmware URL schemes are (file://, http://)
        :returns:   None
        :raises:    InvalidURLScheme, for unsupported firmware url
        """
        parsed_url = urlparse(self.url)
        self.parsed_url = parsed_url
        file_name = os.path.basename(str(parsed_url.path))
        self.dest_file_path = os.path.join(tempfile.mkdtemp(
            prefix='nvidia_firmware'), file_name)
        url_scheme = parsed_url.scheme
        if url_scheme == 'file':
            self._download_file_based_fw()
        elif url_scheme == 'http':
            self._download_http_based_fw()
        else:
            err = 'Firmware URL scheme %s is not supported.' \
                  'The supported firmware URL schemes are' \
                  '(http://, file://)' % url_scheme
            raise InvalidURLScheme(error_msg=_(err))

    def _get_info(self):
        """Get firmware information from firmware binary image

        Caller should wrap this call under try catch to skip non compliant
        firmware binaries.
        :returns:   dict of firmware image attributes
        :raises:    processutils.ProcessExecutionError
        """
        if self.image_info:
            return self.image_info
        try:
            cmd = ('mstflint', '-i', self.dest_file_path, 'query')
            out, _r = utils.execute(*cmd)
        except processutils.ProcessExecutionError as e:
            LOG.error('Failed to query firmware image %s, %s',
                      self.dest_file_path, e)
            raise e
        self.image_info = NvidiaNicFirmwareOps.parse_mstflint_query_output(
            out)
        return self.image_info

    def _validate_image_psid(self):
        """Validate that the provided PSID same as the PSID in provided

        firmware image
        :raises:    MismatchComponentFlavor if they are not equal
        """

        image_psid = self._get_info().get('psid')
        if image_psid != self.psid:
            err = 'The provided psid %s does not match the image psid %s' % \
                  (self.psid, image_psid)
            LOG.error(err)
            raise MismatchComponentFlavor(error_msg=_(err))

    def _validate_image_firmware_version(self):
        """Validate that the provided firmware version same as the version

        in provided firmware image
        :raises:    MismatchFWVersion if they are not equal
        """

        image_version = self._get_info().get('fw_ver')
        if image_version != self.version:
            err = 'The provided firmware version %s does not match ' \
                  'image firmware version %s' % (self.version, image_version)
            LOG.error(err)
            raise MismatchFWVersion(error_msg=_(err))

    def _validate_image_checksum(self):
        """Validate the provided checksum with the calculated one of the

        provided firmware image
        :raises:    MismatchChecksumError if they are not equal
        """
        calculated_checksum = fileutils.compute_file_checksum(
            self.dest_file_path, algorithm=self.checksum_type)
        if self.checksum != calculated_checksum:
            err = 'Mismatch provided checksum %s for image %s' % (
                self.checksum, self.url)
            LOG.error(err)
            raise MismatchChecksumError(error_msg=_(err))


class NvidiaFirmwareImages(object):
    """A class of nvidia firmware images which manages the user provided

    firmware images list
    """

    def __init__(self, firmware_images):
        self.firmware_images = firmware_images
        self.filtered_images_psid_dict = {}

    def validate_images_schema(self):
        """Validate the provided firmware images list schema

        :raises:    InvalidFirmwareImageConfig if any param is missing
        """
        for image in self.firmware_images:
            if not (image.get('url')
                    and image.get('checksum')
                    and image.get('checksumType')
                    and image.get('componentFlavor')
                    and image.get('version')):
                err = 'Invalid parameters for image %s,' \
                      'please provide the following parameters ' \
                      'url, checksum, checksumType, componentFlavor, ' \
                      'version' % image
                LOG.error(err)
                raise InvalidFirmwareImageConfig(error_msg=_(err))

    def filter_images(self, psids_list):
        """Filter firmware images according to the system nics PSIDs,

        and create a map of PSIDs on the system and user provided images.
        Duplicate PSID is not allowed

        :param psids_list:  list of psids of machines nics
        :returns:           None
        :raises:            DuplicateComponentFlavor
        """
        for image in self.firmware_images:
            if image.get('componentFlavor') in psids_list:
                if self.filtered_images_psid_dict.get(
                        image.get('componentFlavor')):
                    err = 'Duplicate componentFlavor %s' % \
                          image['componentFlavor']
                    LOG.error(err)
                    raise DuplicateComponentFlavor(error_msg=_(err))
                else:
                    self.filtered_images_psid_dict[
                        image.get('componentFlavor')] = image
            else:
                LOG.debug('Image with component Flavor %s does not match '
                          'any nic in the system',
                          image.get('componentFlavor'))

    def apply_net_firmware_update(self, nvidia_nics):
        """Apply nic firmware update for all nvidia nics on the system

        which have mappings to the user provided firmware images
        :param nvidia_nics: an object of NvidiaNics
        """
        seen_nics = set()
        for nic in nvidia_nics:
            if self.filtered_images_psid_dict.get(nic.dev_psid):
                # pci_prefix is the pci address without the function number
                # we use it to check if we saw the nic before or not
                pci_prefix = nic.dev_pci[:-1]
                is_seen_nic = pci_prefix in seen_nics
                if not is_seen_nic:
                    seen_nics.add(pci_prefix)
                    fw_bin = NvidiaNicFirmwareBinary(
                        self.filtered_images_psid_dict[nic.dev_psid]['url'],
                        self.filtered_images_psid_dict[nic.dev_psid][
                            'checksum'],
                        self.filtered_images_psid_dict[nic.dev_psid][
                            'checksumType'],
                        self.filtered_images_psid_dict[nic.dev_psid][
                            'componentFlavor'],
                        self.filtered_images_psid_dict[nic.dev_psid][
                            'version'])
                    nic.dev_ops.fw_update_if_needed(
                        self.filtered_images_psid_dict[nic.dev_psid][
                            'version'],
                        fw_bin.dest_file_path)


class NvidiaNicConfig(object):
    """Get/Set Nvidia nics configurations"""

    def __init__(self, nvidia_dev, params):
        self.nvidia_dev = nvidia_dev
        self.params = params
        self._tool_confs = None
        self.device_conf_dict = {}

    def _mstconfig_parse_data(self, data):
        """Parsing the mstconfig out to json

        :param data:    mstconfig query output
        :returns:       dict of nic configuration
        """
        data = list(filter(None, data.split('\n')))
        data_dict = {}
        lines_counter = 0
        for line in data:
            lines_counter += 1
            if 'Configurations:' in line:
                break
        for i in range(lines_counter, len(data)):
            line_data = list(filter(None, data[i].strip().split()))
            data_dict[line_data[0]] = line_data[1]

        return data_dict

    def _get_device_conf_dict(self):
        """Get device Configurations

        :returns:   dict {"PARAM_NAME": "Param value", ....}
        :raises:    processutils.ProcessExecutionError
        """
        LOG.info('Getting configurations for device: %s',
                 self.nvidia_dev.dev_pci)
        if not self.device_conf_dict:
            try:
                cmd = ['mstconfig', '-d', self.nvidia_dev.dev_pci, 'q']
                out, _r = utils.execute(*cmd)
            except processutils.ProcessExecutionError as e:
                LOG.error('Failed to query firmware of device %s: %s',
                          self.nvidia_dev.dev_pci, e)
                raise e
            self.device_conf_dict = self._mstconfig_parse_data(out)
        return self.device_conf_dict

    def _param_supp_by_config_tool(self, param_name):
        """Check if configuration tool supports the provided configuration

        parameter.
        :param param_name:  string, configuration name
        :returns:           bool
        :raises:            processutils.ProcessExecutionError
        """
        if self._tool_confs is None:
            try:
                self._tool_confs, _r = utils.execute(
                    'mstconfig', '-d', self.nvidia_dev.dev_pci, 'i')
            except processutils.ProcessExecutionError as e:
                LOG.error('Failed to query tool configuration of device'
                          ' %s: %s', self.nvidia_dev.dev_pci, e)
                raise e
        # trim any array index if present
        indexed_param = re.match(ARRAY_PARAM_REGEX, param_name)
        if indexed_param:
            param_name = indexed_param.group('param_name')
        return param_name in self._tool_confs

    def _param_supp_by_fw(self, param_name):
        """Check if fw image supports the provided configuration

        parameter.
        :param param_name:  string, configuration name
        :returns:           bool
        :raises:            processutils.ProcessExecutionError
        """
        current_mlx_config = self._get_device_conf_dict()
        indexed_param = re.match(ARRAY_PARAM_REGEX, param_name)
        if indexed_param:
            param_name = indexed_param.group('param_name')
            if param_name not in current_mlx_config:
                return False
            indexed_value = re.match(ARRAY_PARAM_VALUE_REGEX,
                                     current_mlx_config[param_name])
            if not (indexed_value):
                return False
            value_first_index = int(indexed_value.group('first_index'))
            value_last_index = int(indexed_value.group('last_index'))
            param_index = indexed_param.group('index')
            if param_index:
                if int(param_index) in range(value_first_index,
                                             value_last_index):
                    return True
            else:
                param_first_index = int(indexed_param.group('first_index'))
                param_last_index = int(indexed_param.group('last_index'))
                if param_first_index in range(
                        value_first_index, value_last_index) \
                        and param_last_index in range(value_first_index,
                                                      value_last_index) \
                        and param_first_index < param_last_index:
                    return True
            return False
        else:
            return param_name in current_mlx_config

    def validate_config(self):
        """Validate that the firmware settings is supported by mstflint

        package and with current firmware image
        :returns: None
        :raises:    UnSupportedConfigByMstflintPackage
        :raises:    UnSupportedConfigByFW
        """
        LOG.info('Validating config for device %s',
                 self.nvidia_dev.dev_pci)
        for key, value in self.params.items():
            if not self._param_supp_by_config_tool(key):
                err = 'Configuraiton: %s is not supported by mstconfig, ' \
                      'please update to the latest mstflint package.' % key

                LOG.error(err)
                raise UnSupportedConfigByMstflintPackage(error_msg=_(err))

            if not self._param_supp_by_fw(key):
                err = 'Configuraiton %s for device %s is not supported with ' \
                      'current fw' % (key, self.nvidia_dev.dev_pci)
                LOG.error(err)
                raise UnSupportedConfigByFW(error_msg=_(err))

    def set_config(self):
        """Set device configurations

        :param conf_dict:   a dict of:
                            {'PARAM_NAME': 'Param value to set', ...}
        :returns:           None
        :raises:            processutils.ProcessExecutionError
        """
        LOG.info('Setting config for device %s', self.nvidia_dev.dev_pci)
        current_mlx_config = self._get_device_conf_dict()
        params_to_set = []
        for key, value in self.params.items():
            if re.match(ARRAY_PARAM_REGEX, key):
                params_to_set.append('%s=%s' % (key, value))
            else:
                try:
                    # Handle integer values
                    if int(value) != int(current_mlx_config.get(key)):
                        # Aggregate all configurations required to be modified
                        params_to_set.append('%s=%s' % (key, value))
                    else:
                        LOG.info('value of %s for device %s is already '
                                 'configured as %s no need to update it',
                                 key, self.nvidia_dev.dev_pci, value)
                except ValueError:
                    # Handle other values
                    # E.G:
                    # SRIOV_EN                            False(0)
                    # LINK_TYPE_P1                        ETH(2)
                    if str(value).lower() not in \
                            str(current_mlx_config.get(key)).lower():
                        # Aggregate all configurations required to be modified
                        params_to_set.append('%s=%s' % (key, value))
                    else:
                        LOG.info('value of %s for device %s  is already '
                                 'configured as %s, no need to update it',
                                 key, self.nvidia_dev.dev_pci, value)
        if len(params_to_set) > 0:
            try:
                cmd = ['mstconfig', '-d', self.nvidia_dev.dev_pci, '-y', 'set']
                cmd.extend(params_to_set)
                LOG.info('Setting configurations for device: %s',
                         )
                utils.execute(*cmd)
                LOG.info('Set device configurations: Setting %s '
                         'done successfully',
                         ' '.join(params_to_set))
            except processutils.ProcessExecutionError as e:
                LOG.error('Failed to set configuration of device %s, '
                          ' %s: %s', self.nvidia_dev.dev_pci,
                          params_to_set, e)
                raise e

        else:
            LOG.info('Set device configurations: No operation required')


class NvidiaNicsConfig(object):
    """A class of nvidia nics config which manages the user provided

     nics firmware settings
     """

    def __init__(self, nvidia_nics, settings):
        self.settings = settings
        self.nvidia_nics = nvidia_nics
        self.settings_map = {}
        self._nvidia_nics_to_be_reset_list = []
        self._nvidia_nics_config_list = []

    def create_settings_map(self):
        """Filter the user provided nics firmware settings according

        to the system nics IDs, and create a map of IDs on the system and
        user provided nics firmware settings.
        Duplicate IDs  and settings without IDs are not allowed
        :returns:   None
        :raises:    DuplicateDeviceID
        :raises:    InvalidFirmwareSettingsConfig
        """
        ids_list = self.nvidia_nics.get_ids_list()
        for setting in self.settings:
            if (setting.get('deviceID')
                    and setting.get('deviceID') in ids_list
                    and not self.settings_map.get(setting.get('deviceID'))):
                self.settings_map[setting.get('deviceID')] = setting
            elif setting.get('deviceID') and setting.get('deviceID') in \
                    ids_list:
                err = 'duplicate settings for device ID %s ' % \
                      setting.get('deviceID')
                LOG.error(err)
                raise DuplicateDeviceID(error_msg=_(err))
            elif setting.get('deviceID'):
                LOG.debug('There are no devices with ID %s on the system',
                          setting.get('deviceID'))
            else:
                err = 'There is no deviceID provided for this settings'
                LOG.error(err)
                raise InvalidFirmwareSettingsConfig(error_msg=_(err))

    def prepare_nvidia_nic_config(self):
        """Expand the settings map per devices PCI and create a list

        of all NvidiaNicConfig per PCI of nvidia nics on the system.
        Also create a list of all devices that require firmware reset
        :returns:   None
        """
        seen_nics = set()
        for nic in self.nvidia_nics:
            if self.settings_map.get(nic.dev_id):
                params = {}
                prefix = nic.dev_pci[:-1]
                is_seen_nic = prefix in seen_nics
                if not is_seen_nic:
                    seen_nics.add(prefix)
                    if self.settings_map[nic.dev_id].get('globalConfig'):
                        params.update(self.settings_map[nic.dev_id].get(
                            'globalConfig'))
                    if nic.dev_ops.is_image_changed():
                        self._nvidia_nics_to_be_reset_list.append(nic)
                is_first_device = nic.dev_pci[-1] == '0'
                if is_first_device and self.settings_map[nic.dev_id].get(
                        'function0Config'):
                    params.update(self.settings_map[nic.dev_id].get(
                        'function0Config'))
                elif not is_first_device and self.settings_map[nic.dev_id].get(
                        'function1Config'):
                    params.update(self.settings_map[nic.dev_id].get(
                        'function1Config'))
                if params:
                    device_config = NvidiaNicConfig(nic, params)
                    self._nvidia_nics_config_list.append(device_config)

    def reset_nvidia_nics(self):
        """Reset firmware image for all nics in _nvidia_nics_to_be_reset_list

        :returns:   None
        """
        for nvidia_nic in self._nvidia_nics_to_be_reset_list:
            nvidia_nic.dev_ops.reset_device()

    def validate_settings_config(self):
        """Validate firmware settings for all nics in _nvidia_nics_config_list

        :returns:   None
        """
        for nvidia_nic_config in self._nvidia_nics_config_list:
            nvidia_nic_config.validate_config()

    def set_settings_config(self):
        """Set firmware settings for all nics in _nvidia_nics_config_list

        :returns:    None
        """
        for nvidia_nic_config in self._nvidia_nics_config_list:
            nvidia_nic_config.set_config()

    def is_not_empty_reset_list(self):
        """Check if _nvidia_nics_to_be_reset_list is empty or not

        :returns:   bool, True if the list is not empty
        """
        return bool(len(self._nvidia_nics_to_be_reset_list))


def update_nvidia_nic_firmware_image(images):
    """Update nvidia nic firmware image from user provided list images

    :param images:     list of images
    :raises:    InvalidFirmwareImageConfig
    """
    if not type(images) is list:
        err = 'The images must be a list of images, %s' % images
        raise InvalidFirmwareImageConfig(error_msg=_(err))
    check_prereq()
    nvidia_fw_images = NvidiaFirmwareImages(images)
    nvidia_fw_images.validate_images_schema()
    nvidia_nics = NvidiaNics()
    nvidia_nics.discover()
    nvidia_fw_images.filter_images(nvidia_nics.get_psids_list())
    nvidia_fw_images.apply_net_firmware_update(nvidia_nics)


def update_nvidia_nic_firmware_settings(settings):
    """Update nvidia nic firmware settings from user provided list of settings

    :param settings:     list of settings
    :raises:    InvalidFirmwareSettingsConfig
    """
    if not type(settings) is list:
        err = 'The settings must be  list of settings, %s' % settings
        raise InvalidFirmwareSettingsConfig(error_msg=_(err))
    check_prereq()
    nvidia_nics = NvidiaNics()
    nvidia_nics.discover()
    nvidia_nics_config = NvidiaNicsConfig(nvidia_nics, settings)
    nvidia_nics_config.create_settings_map()
    nvidia_nics_config.prepare_nvidia_nic_config()
    if nvidia_nics_config.is_not_empty_reset_list():
        nvidia_nics_config.reset_nvidia_nics()
    nvidia_nics_config.validate_settings_config()
    nvidia_nics_config.set_settings_config()