glances.plugins.smart

Complexity

Total Complexity

37

Size/Duplication

Total Lines	297
Duplicated Lines	0 %

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	SmartPlugin.get_key()	0	3	1
A	SmartPlugin.hide_attributes()	0	4	1
A	SmartPlugin.update()	0	20	4
D	SmartPlugin.msg_curse()	0	50	12
A	SmartPlugin.__init__()	0	21	4

#
# This file is part of Glances.
#
# Copyright (C) 2018 Tim Nibert <docz2a@gmail.com>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""
Hard disk SMART attributes plugin.
Depends on pySMART and smartmontools
Must execute as root
"usermod -a -G disk USERNAME" is not sufficient unfortunately
SmartCTL (/usr/sbin/smartctl) must be in system path for python2.

Regular PySMART is a python2 library.
We are using the pySMART.smartx updated library to support both python 2 and 3.

If we only have disk group access (no root):
$ smartctl -i /dev/sda
smartctl 6.6 2016-05-31 r4324 [x86_64-linux-4.15.0-30-generic] (local build)
Copyright (C) 2002-16, Bruce Allen, Christian Franke, www.smartmontools.org


Probable ATA device behind a SAT layer
Try an additional '-d ata' or '-d sat' argument.

This is not very hopeful: https://medium.com/opsops/why-smartctl-could-not-be-run-without-root-7ea0583b1323

So, here is what we are going to do:
Check for admin access.  If no admin access, disable SMART plugin.

If smartmontools is not installed, we should catch the error upstream in plugin initialization.
"""

from glances.globals import is_admin
from glances.logger import logger
from glances.main import disable
from glances.plugins.plugin.model import GlancesPluginModel

# Import plugin specific dependency
try:
    from pySMART import DeviceList
    from pySMART.interface.nvme import NvmeAttributes
except ImportError as e:
    import_error_tag = True
    logger.warning(f"Missing Python Lib ({e}), HDD Smart plugin is disabled")
else:
    import_error_tag = False


def convert_attribute_to_dict(attr):
    return {
        'name': attr.name,
        'key': attr.name,
        'num': attr.num,
        'flags': attr.flags,
        'raw': attr.raw,
        'value': attr.value,
        'worst': attr.worst,
        'threshold': attr.thresh,
        'type': attr.type,
        'updated': attr.updated,
        'when_failed': attr.when_failed,
    }


NVME_ATTRIBUTE_LABELS = {
    "criticalWarning": "Number of critical warnings",
    "_temperature": "Temperature (°C)",
    "availableSpare": "Available spare (%)",
    "availableSpareThreshold": "Available spare threshold (%)",
    "percentageUsed": "Percentage used (%)",
    "dataUnitsRead": "Data units read",
    "bytesRead": "Bytes read",
    "dataUnitsWritten": "Data units written",
    "bytesWritten": "Bytes written",
    "hostReadCommands": "Host read commands",
    "hostWriteCommands": "Host write commands",
    "controllerBusyTime": "Controller busy time (min)",
    "powerCycles": "Power cycles",
    "powerOnHours": "Power-on hours",
    "unsafeShutdowns": "Unsafe shutdowns",
    "integrityErrors": "Integrity errors",
    "errorEntries": "Error log entries",
    "warningTemperatureTime": "Warning temperature time (min)",
    "criticalTemperatureTime": "Critical temperature time (min)",
    "_logical_sector_size": "Logical sector size",
    "_physical_sector_size": "Physical sector size",
    "errors": "Errors",
    "tests": "Self-tests",
}

def convert_nvme_attribute_to_dict(key,value):
    label = NVME_ATTRIBUTE_LABELS.get(key, key)

    return {
        'name': label,
        'key': key,
        'value': value,
        'flags': None,
        'raw': value,
        'worst': None,
        'threshold': None,
        'type': None,
        'updated': None,
        'when_failed': None
    }

def get_smart_data(hide_attributes):
    """
    Get SMART attribute data
    :return: list of multi leveled dictionaries
             each dict has a key "DeviceName" with the identification of the device in smartctl
             also has keys of the SMART attribute id, with value of another dict of the attributes
             [
                {
                    "DeviceName": "/dev/sda blahblah",
                    "1":
                    {
                        "flags": "..",
                        "raw": "..",
                        etc,
                    }
                    ...
                }
             ]
    """
    stats = []
    # get all devices
    try:
        devlist = DeviceList()
    except TypeError as e:
        # Catch error  (see #1806)
        logger.debug(f'Smart plugin error - Can not grab device list ({e})')
        global import_error_tag
        import_error_tag = True
        return stats

    for dev in devlist.devices:
        stats.append(
            {
                'DeviceName': f'{dev.name} {dev.model}',
            }
        )
        for attribute in dev.attributes:
            if attribute is None:
                pass
            elif attribute.name in hide_attributes:
                pass
            else:
                attrib_dict = convert_attribute_to_dict(attribute)

                # we will use the attribute number as the key
                num = attrib_dict.pop('num', None)
                try:
                    assert num is not None
                except Exception as e:
                    # we should never get here, but if we do, continue to next iteration and skip this attribute
                    logger.debug(f'Smart plugin error - Skip the attribute {attribute} ({e})')
                    continue

                stats[-1][num] = attrib_dict

        if isinstance(dev.if_attributes, NvmeAttributes):
            idx = 0
            for attr in dev.if_attributes.__dict__.keys():
                idx +=1

                attrib_dict = convert_nvme_attribute_to_dict(attr,  dev.if_attributes.__dict__[attr])
                if attrib_dict['name'] in hide_attributes:
                    pass
                else:
                    try:
                        if dev.if_attributes.__dict__[attr] is not None: # make sure the value is serializable to prevent errors in rendering
                            serialized = str(dev.if_attributes.__dict__[attr])
                    except Exception as e:
                        logger.debug(f'Unable to serialize attribute {attr} from NVME')
                        attrib_dict['value'] = None
                        attrib_dict['raw'] = None
                    finally:
                        stats[-1][idx] = attrib_dict

    return stats


class SmartPlugin(GlancesPluginModel):
    """Glances' HDD SMART plugin."""

    def __init__(self, args=None, config=None, stats_init_value=[]):
        """Init the plugin."""
        # check if user is admin
        if not is_admin() and args:
            disable(args, "smart")
            logger.debug("Current user is not admin, HDD SMART plugin disabled.")

        super().__init__(args=args, config=config)

        # We want to display the stat in the curse interface
        self.display_curse = True

        if 'hide_attributes' in config.as_dict()['smart']:
            logger.info(
                'Followings SMART attributes wil not be displayed: {}'.format(
                    config.as_dict()['smart']['hide_attributes']
                )
            )
            self.hide_attributes = config.as_dict()['smart']['hide_attributes'].split(',')
        else:
            self.hide_attributes = []

    @property
    def hide_attributes(self):
        """Set hide_attributes list"""
        return self._hide_attributes

    @hide_attributes.setter
    def hide_attributes(self, attr_list):
        """Set hide_attributes list"""
        self._hide_attributes = [i for i in attr_list]

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update SMART stats using the input method."""
        # Init new stats
        stats = self.get_init_value()

        if import_error_tag:
            return self.stats

        if self.input_method == 'local':
            # Update stats and hide some sensors(#2996)
            stats = [s for s in get_smart_data(self.hide_attributes) if self.is_display(s[self.get_key()])]
        elif self.input_method == 'snmp':
            pass

        # Update the stats
        self.stats = stats

        return self.stats

    def get_key(self):
        """Return the key of the list."""
        return 'DeviceName'

    def msg_curse(self, args=None, max_width=None):
        """Return the dict to display in the curse interface."""
        # Init the return message
        ret = []

        # Only process if stats exist...
        if import_error_tag or not self.stats or self.is_disabled():
            return ret

        # Max size for the interface name
        if max_width:
            name_max_width = max_width - 6
        else:
            # No max_width defined, return an empty curse message
            logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
            return ret

        # Header
        msg = '{:{width}}'.format('SMART disks', width=name_max_width)
        ret.append(self.curse_add_line(msg, "TITLE"))
        # Data
        for device_stat in self.stats:
            # New line
            ret.append(self.curse_new_line())
            msg = '{:{width}}'.format(device_stat['DeviceName'][:max_width], width=max_width)
            ret.append(self.curse_add_line(msg))
            try:
                device_stat_sorted = sorted([i for i in device_stat if i != 'DeviceName'], key=int)
            except ValueError:
                # Catch ValueError, see #2904
                device_stat_sorted = [i for i in device_stat if i != 'DeviceName']
            for smart_stat in device_stat_sorted:
                ret.append(self.curse_new_line())
                msg = ' {:{width}}'.format(
                    device_stat[smart_stat]['name'][: name_max_width - 1].replace('_', ' '), width=name_max_width - 1
                )
                ret.append(self.curse_add_line(msg))
                try:
                    raw = device_stat[smart_stat]['raw']
                    if device_stat[smart_stat]['key'] in ["bytesWritten", "bytesRead", "dataUnitsRead", "dataUnitsWritten", "hostReadCommands", "hostWriteCommands" ]:
                        msg = '{:>8}'.format("" if raw is None else self.auto_unit(raw))
                    else:    
                        msg = '{:>8}'.format("" if raw is None else str(raw))
                    ret.append(self.curse_add_line(msg))
                except Exception as e:
                    logger.debug(f"Failed to serialize {smart_stat}")
                    meg = ""
                    ret.append(self.curse_add_line(msg))

        return ret