Test Failed
Pull Request — develop (#3355)
by
unknown
02:34
created

convert_nvme_attribute_to_dict()   A

Complexity

Conditions 1

Size

Total Lines 13
Code Lines 12

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 1
eloc 12
nop 2
dl 0
loc 13
rs 9.8
c 0
b 0
f 0
1
#
2
# This file is part of Glances.
3
#
4
# Copyright (C) 2018 Tim Nibert <[email protected]>
5
#
6
# SPDX-License-Identifier: LGPL-3.0-only
7
#
8
9
"""
10
Hard disk SMART attributes plugin.
11
Depends on pySMART and smartmontools
12
Must execute as root
13
"usermod -a -G disk USERNAME" is not sufficient unfortunately
14
SmartCTL (/usr/sbin/smartctl) must be in system path for python2.
15
16
Regular PySMART is a python2 library.
17
We are using the pySMART.smartx updated library to support both python 2 and 3.
18
19
If we only have disk group access (no root):
20
$ smartctl -i /dev/sda
21
smartctl 6.6 2016-05-31 r4324 [x86_64-linux-4.15.0-30-generic] (local build)
22
Copyright (C) 2002-16, Bruce Allen, Christian Franke, www.smartmontools.org
23
24
25
Probable ATA device behind a SAT layer
26
Try an additional '-d ata' or '-d sat' argument.
27
28
This is not very hopeful: https://medium.com/opsops/why-smartctl-could-not-be-run-without-root-7ea0583b1323
29
30
So, here is what we are going to do:
31
Check for admin access.  If no admin access, disable SMART plugin.
32
33
If smartmontools is not installed, we should catch the error upstream in plugin initialization.
34
"""
35
36
from glances.globals import is_admin
37
from glances.logger import logger
38
from glances.main import disable
39
from glances.plugins.plugin.model import GlancesPluginModel
40
41
# Import plugin specific dependency
42
try:
43
    from pySMART import DeviceList
44
    from pySMART.interface.nvme import NvmeAttributes
45
except ImportError as e:
46
    import_error_tag = True
47
    logger.warning(f"Missing Python Lib ({e}), HDD Smart plugin is disabled")
48
else:
49
    import_error_tag = False
50
51
52
def convert_attribute_to_dict(attr):
53
    return {
54
        'name': attr.name,
55
        'num': attr.num,
56
        'flags': attr.flags,
57
        'raw': attr.raw,
58
        'value': attr.value,
59
        'worst': attr.worst,
60
        'threshold': attr.thresh,
61
        'type': attr.type,
62
        'updated': attr.updated,
63
        'when_failed': attr.when_failed,
64
    }
65
66
67
NVME_ATTRIBUTE_LABELS = {
68
    "criticalWarning": "Number of critical warnings",
69
    "_temperature": "Temperature (°C)",
70
    "availableSpare": "Available spare (%)",
71
    "availableSpareThreshold": "Available spare threshold (%)",
72
    "percentageUsed": "Percentage used (%)",
73
    "dataUnitsRead": "Data units read",
74
    "bytesRead": "Bytes read",
75
    "dataUnitsWritten": "Data units written",
76
    "bytesWritten": "Bytes written",
77
    "hostReadCommands": "Host read commands",
78
    "hostWriteCommands": "Host write commands",
79
    "controllerBusyTime": "Controller busy time (min)",
80
    "powerCycles": "Power cycles",
81
    "powerOnHours": "Power-on hours",
82
    "unsafeShutdowns": "Unsafe shutdowns",
83
    "integrityErrors": "Integrity errors",
84
    "errorEntries": "Error log entries",
85
    "warningTemperatureTime": "Warning temperature time (min)",
86
    "criticalTemperatureTime": "Critical temperature time (min)",
87
    "_logical_sector_size": "Logical sector size",
88
    "_physical_sector_size": "Physical sector size",
89
    "errors": "Errors",
90
    "tests": "Self-tests",
91
}
92
93
def convert_nvme_attribute_to_dict(key,value):
94
    label = NVME_ATTRIBUTE_LABELS.get(key, key)
95
96
    return {
97
        'name': label,
98
        'value': value,
99
        'flags': None,
100
        'raw': value,
101
        'worst': None,
102
        'threshold': None,
103
        'type': None,
104
        'updated': None,
105
        'when_failed': None
106
    }
107
108
def get_smart_data():
109
    """
110
    Get SMART attribute data
111
    :return: list of multi leveled dictionaries
112
             each dict has a key "DeviceName" with the identification of the device in smartctl
113
             also has keys of the SMART attribute id, with value of another dict of the attributes
114
             [
115
                {
116
                    "DeviceName": "/dev/sda blahblah",
117
                    "1":
118
                    {
119
                        "flags": "..",
120
                        "raw": "..",
121
                        etc,
122
                    }
123
                    ...
124
                }
125
             ]
126
    """
127
    stats = []
128
    # get all devices
129
    try:
130
        devlist = DeviceList()
131
    except TypeError as e:
132
        # Catch error  (see #1806)
133
        logger.debug(f'Smart plugin error - Can not grab device list ({e})')
134
        global import_error_tag
135
        import_error_tag = True
136
        return stats
137
138
    for dev in devlist.devices:
139
        stats.append(
140
            {
141
                'DeviceName': f'{dev.name} {dev.model}',
142
            }
143
        )
144
        for attribute in dev.attributes:
145
            if attribute is None:
146
                pass
147
            else:
148
                attrib_dict = convert_attribute_to_dict(attribute)
149
150
                # we will use the attribute number as the key
151
                num = attrib_dict.pop('num', None)
152
                try:
153
                    assert num is not None
154
                except Exception as e:
155
                    # we should never get here, but if we do, continue to next iteration and skip this attribute
156
                    logger.debug(f'Smart plugin error - Skip the attribute {attribute} ({e})')
157
                    continue
158
159
                stats[-1][num] = attrib_dict
160
161
        if isinstance(dev.if_attributes, NvmeAttributes):
162
            idx = 0
163
            for attr in dev.if_attributes.__dict__.keys():
164
                idx +=1
165
                attrib_dict = convert_nvme_attribute_to_dict(attr,  dev.if_attributes.__dict__[attr])
166
                try:
167
                    if dev.if_attributes.__dict__[attr] is not None:
168
                        serialized = str(dev.if_attributes.__dict__[attr])
169
                except Exception as e:
170
                    logger.debug(f'Unable to serialize attribute {attr} from NVME')
171
                    attrib_dict['value'] = None
172
                    attrib_dict['raw'] = None
173
                finally:
174
                    stats[-1][idx] = attrib_dict
175
176
    return stats
177
178
179
class SmartPlugin(GlancesPluginModel):
180
    """Glances' HDD SMART plugin."""
181
182
    def __init__(self, args=None, config=None, stats_init_value=[]):
183
        """Init the plugin."""
184
        # check if user is admin
185
        if not is_admin() and args:
186
            disable(args, "smart")
187
            logger.debug("Current user is not admin, HDD SMART plugin disabled.")
188
189
        super().__init__(args=args, config=config)
190
191
        # We want to display the stat in the curse interface
192
        self.display_curse = True
193
194
    @GlancesPluginModel._check_decorator
195
    @GlancesPluginModel._log_result_decorator
196
    def update(self):
197
        """Update SMART stats using the input method."""
198
        # Init new stats
199
        stats = self.get_init_value()
200
201
        if import_error_tag:
202
            return self.stats
203
204
        if self.input_method == 'local':
205
            # Update stats and hide some sensors(#2996)
206
            stats = [s for s in get_smart_data() if self.is_display(s[self.get_key()])]
207
        elif self.input_method == 'snmp':
208
            pass
209
210
        # Update the stats
211
        self.stats = stats
212
213
        return self.stats
214
215
    def get_key(self):
216
        """Return the key of the list."""
217
        return 'DeviceName'
218
219
    def msg_curse(self, args=None, max_width=None):
220
        """Return the dict to display in the curse interface."""
221
        # Init the return message
222
        ret = []
223
224
        # Only process if stats exist...
225
        if import_error_tag or not self.stats or self.is_disabled():
226
            return ret
227
228
        # Max size for the interface name
229
        if max_width:
230
            name_max_width = max_width - 6
231
        else:
232
            # No max_width defined, return an empty curse message
233
            logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
234
            return ret
235
236
        # Header
237
        msg = '{:{width}}'.format('SMART disks', width=name_max_width)
238
        ret.append(self.curse_add_line(msg, "TITLE"))
239
        # Data
240
        for device_stat in self.stats:
241
            # New line
242
            ret.append(self.curse_new_line())
243
            msg = '{:{width}}'.format(device_stat['DeviceName'][:max_width], width=max_width)
244
            ret.append(self.curse_add_line(msg))
245
            try:
246
                device_stat_sorted = sorted([i for i in device_stat if i != 'DeviceName'], key=int)
247
            except ValueError:
248
                # Catch ValueError, see #2904
249
                device_stat_sorted = [i for i in device_stat if i != 'DeviceName']
250
            for smart_stat in device_stat_sorted:
251
                ret.append(self.curse_new_line())
252
                msg = ' {:{width}}'.format(
253
                    device_stat[smart_stat]['name'][: name_max_width - 1].replace('_', ' '), width=name_max_width - 1
254
                )
255
                ret.append(self.curse_add_line(msg))
256
                try:
257
                    raw = device_stat[smart_stat]['raw']
258
                    msg = '{:>8}'.format("" if raw is None else str(raw))
259
                    ret.append(self.curse_add_line(msg))
260
                except Exception as e:
261
                    logger.debug(f"Failed to serialize {smart_stat}")
262
                    meg = ""
263
                    ret.append(self.curse_add_line(msg))
264
265
        return ret
266