glances.plugins.fs.FsPlugin.get_disk_partitions() - Code Metrics - Inspection of "Merge pull request #3291 from nicolargo/issue3290-..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 420cf2...653997 )

by Nicolas

created 2025-09-28 12:17 UTC

glances.plugins.fs.FsPlugin.get_disk_partitions() A

↳ Parent: glances.plugins.fs

Complexity

Conditions

Size

Total Lines	11
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	7
nop	3
dl	0
loc	11
rs	10
c	0
b	0
f	0

#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""File system plugin."""

import operator

import psutil

from glances.globals import PermissionError, exit_after, nativestr, u
from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel

# Fields description
# description: human readable description
# short_name: shortname to use un UI
# unit: unit type
# rate: is it a rate ? If yes, // by time_since_update when displayed,
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'device_name': {'description': 'Device name.'},
    'fs_type': {'description': 'File system type.'},
    'mnt_point': {'description': 'Mount point.'},
    'options': {'description': 'Mount options.'},
    'size': {
        'description': 'Total size.',
        'unit': 'byte',
    },
    'used': {
        'description': 'Used size.',
        'unit': 'byte',
    },
    'free': {
        'description': 'Free size.',
        'unit': 'byte',
    },
    'percent': {
        'description': 'File system usage in percent.',
        'unit': 'percent',
    },
}

# SNMP OID
# The snmpd.conf needs to be edited.
# Add the following to enable it on all disk
# ...
# includeAllDisks 10%
# ...
# The OIDs are as follows (for the first disk)
# Path where the disk is mounted: .1.3.6.1.4.1.2021.9.1.2.1
# Path of the device for the partition: .1.3.6.1.4.1.2021.9.1.3.1
# Total size of the disk/partition (kBytes): .1.3.6.1.4.1.2021.9.1.6.1
# Available space on the disk: .1.3.6.1.4.1.2021.9.1.7.1
# Used space on the disk: .1.3.6.1.4.1.2021.9.1.8.1
# Percentage of space used on disk: .1.3.6.1.4.1.2021.9.1.9.1
# Percentage of inodes used on disk: .1.3.6.1.4.1.2021.9.1.10.1
snmp_oid = {
    'default': {
        'mnt_point': '1.3.6.1.4.1.2021.9.1.2',
        'device_name': '1.3.6.1.4.1.2021.9.1.3',
        'size': '1.3.6.1.4.1.2021.9.1.6',
        'used': '1.3.6.1.4.1.2021.9.1.8',
        'percent': '1.3.6.1.4.1.2021.9.1.9',
    },
    'windows': {
        'mnt_point': '1.3.6.1.2.1.25.2.3.1.3',
        'alloc_unit': '1.3.6.1.2.1.25.2.3.1.4',
        'size': '1.3.6.1.2.1.25.2.3.1.5',
        'used': '1.3.6.1.2.1.25.2.3.1.6',
    },
    'netapp': {
        'mnt_point': '1.3.6.1.4.1.789.1.5.4.1.2',
        'device_name': '1.3.6.1.4.1.789.1.5.4.1.10',
        'size': '1.3.6.1.4.1.789.1.5.4.1.3',
        'used': '1.3.6.1.4.1.789.1.5.4.1.4',
        'percent': '1.3.6.1.4.1.789.1.5.4.1.6',
    },
}
snmp_oid['esxi'] = snmp_oid['windows']

# Define the history items list
# All items in this list will be historised if the --enable-history tag is set
items_history_list = [{'name': 'percent', 'description': 'File system usage in percent', 'y_unit': '%'}]


@exit_after(1, default=None)
def get_disk_usage(fs):
    """Return all partitions."""
    try:
        return psutil.disk_usage(fs.mountpoint)
    except OSError:
        # Disk is ejected during the command
        logger.debug("Plugin - fs: PsUtil fetch failed")
        return None


class FsPlugin(GlancesPluginModel):
    """Glances file system plugin.

    stats is a list
    """

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super().__init__(
            args=args,
            config=config,
            items_history_list=items_history_list,
            stats_init_value=[],
            fields_description=fields_description,
        )

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

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

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update the FS stats using the input method."""
        # Update the stats
        if self.input_method == 'local':
            stats = self.update_local()
        else:
            stats = self.get_init_value()

        # Update the stats
        self.stats = stats

        return self.stats

    @GlancesPluginModel._exit_after(3)
    def get_all_stats_partitions(self):
        """Return all partitions."""
        try:
            return psutil.disk_partitions(all=True)
        except (UnicodeDecodeError, PermissionError):
            logger.debug("Plugin - fs: PsUtil fetch failed")
            return []

    @GlancesPluginModel._exit_after(3)
    def get_disk_partitions(self, *, fetch_all: bool = False):
        """Return all partitions."""
        try:
            # Grab the stats using the psutil disk_partitions
            # If fetch_all is False, then returns physical devices only (e.g. hard disks, cd-rom drives, USB keys)
            # and ignore all others (e.g. memory partitions such as /dev/shm)
            return psutil.disk_partitions(all=fetch_all)
        except (UnicodeDecodeError, PermissionError):
            logger.debug("Plugin - fs: PsUtil fetch failed")
            return []

    def update_local(self):
        """Update the FS stats using the input method."""
        # Init new stats
        stats = self.get_init_value()

        # Update stats using the standard system lib
        fs_stat = self.get_disk_partitions()

        # Optional hack to allow logical mounts points (issue #448)
        allowed_fs_types = self.get_conf_value('allow')
        if allowed_fs_types:
            # Avoid Psutil call unless mounts need to be allowed
            all_mounted_fs = self.get_disk_partitions(fetch_all=True)
            # Discard duplicates (#2299) and add entries matching allowed fs types
            tracked_mnt_points = {f.mountpoint for f in fs_stat}
            for f in all_mounted_fs:
                if (
                    any(f.fstype.find(fs_type) >= 0 for fs_type in allowed_fs_types)
                    and f.mountpoint not in tracked_mnt_points
                ):
                    fs_stat.append(f)

        # Loop over fs
        for fs in fs_stat:
            # Hide the stats if the mount point is in the exclude list
            # It avoids unnecessary call to PsUtil disk_usage
            if not self.is_display_any(fs.mountpoint, fs.device):
                continue

            # Grab the disk usage
            fs_usage = get_disk_usage(fs)
            if fs_usage is None:
                continue
            fs_current = {
                'device_name': fs.device,
                'fs_type': fs.fstype,
                # Manage non breaking space (see issue #1065)
                'mnt_point': u(fs.mountpoint).replace('\u00a0', ' '),
                'options': fs.opts,
                'size': fs_usage.total,
                'used': fs_usage.used,
                'free': fs_usage.free,
                'percent': fs_usage.percent,
                'key': self.get_key(),
            }

            # Add alias if exist (define in the configuration file)
            if self.has_alias(fs_current['mnt_point']) is not None:
                fs_current['alias'] = self.has_alias(fs_current['mnt_point'])

            stats.append(fs_current)

        return stats

    def update_snmp(self):
        """Update the FS stats using the input method."""
        # Init new stats
        stats = self.get_init_value()

        # Update stats using SNMP

        # SNMP bulk command to get all file system in one shot
        try:
            fs_stat = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name], bulk=True)
        except KeyError:
            fs_stat = self.get_stats_snmp(snmp_oid=snmp_oid['default'], bulk=True)

        # Loop over fs
        if self.short_system_name in ('windows', 'esxi'):
            # Windows or ESXi tips
            for fs, fs_value in fs_stat.item():
                # Do not take hidden file system into account
                if not self.is_display(fs):
                    continue

                # Memory stats are grabbed in the same OID table (ignore it)
                if fs == 'Virtual Memory' or fs == 'Physical Memory' or fs == 'Real Memory':
                    continue
                size = int(fs_value['size']) * int(fs_value['alloc_unit'])
                used = int(fs_value['used']) * int(fs_value['alloc_unit'])
                percent = float(used * 100 / size)
                fs_current = {
                    'device_name': '',
                    'mnt_point': fs.partition(' ')[0],
                    'options': '',
                    'size': size,
                    'used': used,
                    'percent': percent,
                    'key': self.get_key(),
                }
                stats.append(fs_current)
        else:
            # Default behavior
            for fs, fs_value in fs_stat.item():
                # Do not take hidden file system into account
                if not self.is_display_any(fs, fs_value['device_name']):
                    continue

                fs_current = {
                    'device_name': fs_value['device_name'],
                    'mnt_point': fs,
                    'options': '',
                    'size': int(fs_value['size']) * 1024,
                    'used': int(fs_value['used']) * 1024,
                    'percent': float(fs_value['percent']),
                    'key': self.get_key(),
                }
                stats.append(fs_current)

        return stats

    def update_views(self):
        """Update stats views."""
        # Call the father's method
        super().update_views()

        # Add specifics information
        # Alert
        # Do not display threshold for volume mounted in 'ro' (read-only) #3143
        for i in [d for d in self.stats if 'ro' not in d.get('options', '').split(',')]:
            self.views[i[self.get_key()]]['used']['decoration'] = self.get_alert(
                current=i['size'] - i['free'], maximum=i['size'], header=i['mnt_point']
            )

    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 and display plugin enable...
        if not self.stats or self.is_disabled():
            return ret

        # Max size for the interface name
        if max_width:
            name_max_width = max_width - 13
        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

        # Build the string message
        # Header
        msg = '{:{width}}'.format('FILE SYS', width=name_max_width)
        ret.append(self.curse_add_line(msg, "TITLE"))
        if args.fs_free_space:
            msg = '{:>8}'.format('Free')
        else:
            msg = '{:>8}'.format('Used')
        ret.append(self.curse_add_line(msg))
        msg = '{:>7}'.format('Total')
        ret.append(self.curse_add_line(msg))

        # Filesystem list (sorted by name)
        for i in sorted(self.stats, key=operator.itemgetter(self.get_key())):
            # New line
            ret.append(self.curse_new_line())
            mnt_point = i['alias'] if 'alias' in i else i['mnt_point']
            if len(mnt_point) + len(i['device_name'].split('/')[-1]) <= name_max_width - 3:
                # If possible concatenate mode info... Glances touch inside :)
                mnt_point += ' (' + i['device_name'].split('/')[-1] + ')'
            elif len(mnt_point) > name_max_width:
                mnt_point = mnt_point[:name_max_width] + '_'
            msg = '{:{width}}'.format(nativestr(mnt_point), width=name_max_width + 1)
            ret.append(self.curse_add_line(msg))
            if args.fs_free_space:
                msg = '{:>7}'.format(self.auto_unit(i['free']))
            else:
                msg = '{:>7}'.format(self.auto_unit(i['used']))
            ret.append(
                self.curse_add_line(msg, self.get_views(item=i[self.get_key()], key='used', option='decoration'))
            )
            msg = '{:>7}'.format(self.auto_unit(i['size']))
            ret.append(self.curse_add_line(msg))

        return ret


1			#
2			# This file is part of Glances.
3			#
4			# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
5			#
6			# SPDX-License-Identifier: LGPL-3.0-only
7			#
8
9			"""File system plugin."""
10
11			import operator
12
13			import psutil
14
15			from glances.globals import PermissionError, exit_after, nativestr, u
16			from glances.logger import logger
17			from glances.plugins.plugin.model import GlancesPluginModel
18
19			# Fields description
20			# description: human readable description
21			# short_name: shortname to use un UI
22			# unit: unit type
23			# rate: is it a rate ? If yes, // by time_since_update when displayed,
24			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
25			fields_description = {
26			'device_name': {'description': 'Device name.'},
27			'fs_type': {'description': 'File system type.'},
28			'mnt_point': {'description': 'Mount point.'},
29			'options': {'description': 'Mount options.'},
30			'size': {
31			'description': 'Total size.',
32			'unit': 'byte',
33			},
34			'used': {
35			'description': 'Used size.',
36			'unit': 'byte',
37			},
38			'free': {
39			'description': 'Free size.',
40			'unit': 'byte',
41			},
42			'percent': {
43			'description': 'File system usage in percent.',
44			'unit': 'percent',
45			},
46			}
47
48			# SNMP OID
49			# The snmpd.conf needs to be edited.
50			# Add the following to enable it on all disk
51			# ...
52			# includeAllDisks 10%
53			# ...
54			# The OIDs are as follows (for the first disk)
55			# Path where the disk is mounted: .1.3.6.1.4.1.2021.9.1.2.1
56			# Path of the device for the partition: .1.3.6.1.4.1.2021.9.1.3.1
57			# Total size of the disk/partition (kBytes): .1.3.6.1.4.1.2021.9.1.6.1
58			# Available space on the disk: .1.3.6.1.4.1.2021.9.1.7.1
59			# Used space on the disk: .1.3.6.1.4.1.2021.9.1.8.1
60			# Percentage of space used on disk: .1.3.6.1.4.1.2021.9.1.9.1
61			# Percentage of inodes used on disk: .1.3.6.1.4.1.2021.9.1.10.1
62			snmp_oid = {
63			'default': {
64			'mnt_point': '1.3.6.1.4.1.2021.9.1.2',
65			'device_name': '1.3.6.1.4.1.2021.9.1.3',
66			'size': '1.3.6.1.4.1.2021.9.1.6',
67			'used': '1.3.6.1.4.1.2021.9.1.8',
68			'percent': '1.3.6.1.4.1.2021.9.1.9',
69			},
70			'windows': {
71			'mnt_point': '1.3.6.1.2.1.25.2.3.1.3',
72			'alloc_unit': '1.3.6.1.2.1.25.2.3.1.4',
73			'size': '1.3.6.1.2.1.25.2.3.1.5',
74			'used': '1.3.6.1.2.1.25.2.3.1.6',
75			},
76			'netapp': {
77			'mnt_point': '1.3.6.1.4.1.789.1.5.4.1.2',
78			'device_name': '1.3.6.1.4.1.789.1.5.4.1.10',
79			'size': '1.3.6.1.4.1.789.1.5.4.1.3',
80			'used': '1.3.6.1.4.1.789.1.5.4.1.4',
81			'percent': '1.3.6.1.4.1.789.1.5.4.1.6',
82			},
83			}
84			snmp_oid['esxi'] = snmp_oid['windows']
85
86			# Define the history items list
87			# All items in this list will be historised if the --enable-history tag is set
88			items_history_list = [{'name': 'percent', 'description': 'File system usage in percent', 'y_unit': '%'}]
89
90
91			@exit_after(1, default=None)
92			def get_disk_usage(fs):
93			"""Return all partitions."""
94			try:
95			return psutil.disk_usage(fs.mountpoint)
96			except OSError:
97			# Disk is ejected during the command
98			logger.debug("Plugin - fs: PsUtil fetch failed")
99			return None
100
101
102			class FsPlugin(GlancesPluginModel):
103			"""Glances file system plugin.
104
105			stats is a list
106			"""
107
108			def __init__(self, args=None, config=None):
109			"""Init the plugin."""
110			super().__init__(
111			args=args,
112			config=config,
113			items_history_list=items_history_list,
114			stats_init_value=[],
115			fields_description=fields_description,
116			)
117
118			# We want to display the stat in the curse interface
119			self.display_curse = True
120
121			def get_key(self):
122			"""Return the key of the list."""
123			return 'mnt_point'
124
125			@GlancesPluginModel._check_decorator
126			@GlancesPluginModel._log_result_decorator
127			def update(self):
128			"""Update the FS stats using the input method."""
129			# Update the stats
130			if self.input_method == 'local':
131			stats = self.update_local()
132			else:
133			stats = self.get_init_value()
134
135			# Update the stats
136			self.stats = stats
137
138			return self.stats
139
140			@GlancesPluginModel._exit_after(3)
141			def get_all_stats_partitions(self):
142			"""Return all partitions."""
143			try:
144			return psutil.disk_partitions(all=True)
145			except (UnicodeDecodeError, PermissionError):
146			logger.debug("Plugin - fs: PsUtil fetch failed")
147			return []
148
149			@GlancesPluginModel._exit_after(3)
150			def get_disk_partitions(self, *, fetch_all: bool = False):
151			"""Return all partitions."""
152			try:
153			# Grab the stats using the psutil disk_partitions
154			# If fetch_all is False, then returns physical devices only (e.g. hard disks, cd-rom drives, USB keys)
155			# and ignore all others (e.g. memory partitions such as /dev/shm)
156			return psutil.disk_partitions(all=fetch_all)
157			except (UnicodeDecodeError, PermissionError):
158			logger.debug("Plugin - fs: PsUtil fetch failed")
159			return []
160
161			def update_local(self):
162			"""Update the FS stats using the input method."""
163			# Init new stats
164			stats = self.get_init_value()
165
166			# Update stats using the standard system lib
167			fs_stat = self.get_disk_partitions()
168
169			# Optional hack to allow logical mounts points (issue #448)
170			allowed_fs_types = self.get_conf_value('allow')
171			if allowed_fs_types:
172			# Avoid Psutil call unless mounts need to be allowed
173			all_mounted_fs = self.get_disk_partitions(fetch_all=True)
174			# Discard duplicates (#2299) and add entries matching allowed fs types
175			tracked_mnt_points = {f.mountpoint for f in fs_stat}
176			for f in all_mounted_fs:
177			if (
178			any(f.fstype.find(fs_type) >= 0 for fs_type in allowed_fs_types)
179			and f.mountpoint not in tracked_mnt_points
180			):
181			fs_stat.append(f)
182
183			# Loop over fs
184			for fs in fs_stat:
185			# Hide the stats if the mount point is in the exclude list
186			# It avoids unnecessary call to PsUtil disk_usage
187			if not self.is_display_any(fs.mountpoint, fs.device):
188			continue
189
190			# Grab the disk usage
191			fs_usage = get_disk_usage(fs)
192			if fs_usage is None:
193			continue
194			fs_current = {
195			'device_name': fs.device,
196			'fs_type': fs.fstype,
197			# Manage non breaking space (see issue #1065)
198			'mnt_point': u(fs.mountpoint).replace('\u00a0', ' '),
199			'options': fs.opts,
200			'size': fs_usage.total,
201			'used': fs_usage.used,
202			'free': fs_usage.free,
203			'percent': fs_usage.percent,
204			'key': self.get_key(),
205			}
206
207			# Add alias if exist (define in the configuration file)
208			if self.has_alias(fs_current['mnt_point']) is not None:
209			fs_current['alias'] = self.has_alias(fs_current['mnt_point'])
210
211			stats.append(fs_current)
212
213			return stats
214
215			def update_snmp(self):
216			"""Update the FS stats using the input method."""
217			# Init new stats
218			stats = self.get_init_value()
219
220			# Update stats using SNMP
221
222			# SNMP bulk command to get all file system in one shot
223			try:
224			fs_stat = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name], bulk=True)
225			except KeyError:
226			fs_stat = self.get_stats_snmp(snmp_oid=snmp_oid['default'], bulk=True)
227
228			# Loop over fs
229			if self.short_system_name in ('windows', 'esxi'):
230			# Windows or ESXi tips
231			for fs, fs_value in fs_stat.item():
232			# Do not take hidden file system into account
233			if not self.is_display(fs):
234			continue
235
236			# Memory stats are grabbed in the same OID table (ignore it)
237			if fs == 'Virtual Memory' or fs == 'Physical Memory' or fs == 'Real Memory':
238			continue
239			size = int(fs_value['size']) * int(fs_value['alloc_unit'])
240			used = int(fs_value['used']) * int(fs_value['alloc_unit'])
241			percent = float(used * 100 / size)
242			fs_current = {
243			'device_name': '',
244			'mnt_point': fs.partition(' ')[0],
245			'options': '',
246			'size': size,
247			'used': used,
248			'percent': percent,
249			'key': self.get_key(),
250			}
251			stats.append(fs_current)
252			else:
253			# Default behavior
254			for fs, fs_value in fs_stat.item():
255			# Do not take hidden file system into account
256			if not self.is_display_any(fs, fs_value['device_name']):
257			continue
258
259			fs_current = {
260			'device_name': fs_value['device_name'],
261			'mnt_point': fs,
262			'options': '',
263			'size': int(fs_value['size']) * 1024,
264			'used': int(fs_value['used']) * 1024,
265			'percent': float(fs_value['percent']),
266			'key': self.get_key(),
267			}
268			stats.append(fs_current)
269
270			return stats
271
272			def update_views(self):
273			"""Update stats views."""
274			# Call the father's method
275			super().update_views()
276
277			# Add specifics information
278			# Alert
279			# Do not display threshold for volume mounted in 'ro' (read-only) #3143
280			for i in [d for d in self.stats if 'ro' not in d.get('options', '').split(',')]:
281			self.views[i[self.get_key()]]['used']['decoration'] = self.get_alert(
282			current=i['size'] - i['free'], maximum=i['size'], header=i['mnt_point']
283			)
284
285			def msg_curse(self, args=None, max_width=None):
286			"""Return the dict to display in the curse interface."""
287			# Init the return message
288			ret = []
289
290			# Only process if stats exist and display plugin enable...
291			if not self.stats or self.is_disabled():
292			return ret
293
294			# Max size for the interface name
295			if max_width:
296			name_max_width = max_width - 13
297			else:
298			# No max_width defined, return an empty curse message
299			logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
300			return ret
301
302			# Build the string message
303			# Header
304			msg = '{:{width}}'.format('FILE SYS', width=name_max_width)
305			ret.append(self.curse_add_line(msg, "TITLE"))
306			if args.fs_free_space:
307			msg = '{:>8}'.format('Free')
308			else:
309			msg = '{:>8}'.format('Used')
310			ret.append(self.curse_add_line(msg))
311			msg = '{:>7}'.format('Total')
312			ret.append(self.curse_add_line(msg))
313
314			# Filesystem list (sorted by name)
315			for i in sorted(self.stats, key=operator.itemgetter(self.get_key())):
316			# New line
317			ret.append(self.curse_new_line())
318			mnt_point = i['alias'] if 'alias' in i else i['mnt_point']
319			if len(mnt_point) + len(i['device_name'].split('/')[-1]) <= name_max_width - 3:
320			# If possible concatenate mode info... Glances touch inside :)
321			mnt_point += ' (' + i['device_name'].split('/')[-1] + ')'
322			elif len(mnt_point) > name_max_width:
323			mnt_point = mnt_point[:name_max_width] + '_'
324			msg = '{:{width}}'.format(nativestr(mnt_point), width=name_max_width + 1)
325			ret.append(self.curse_add_line(msg))
326			if args.fs_free_space:
327			msg = '{:>7}'.format(self.auto_unit(i['free']))
328			else:
329			msg = '{:>7}'.format(self.auto_unit(i['used']))
330			ret.append(
331			self.curse_add_line(msg, self.get_views(item=i[self.get_key()], key='used', option='decoration'))
332			)
333			msg = '{:>7}'.format(self.auto_unit(i['size']))
334			ret.append(self.curse_add_line(msg))
335
336			return ret
337

nicolargo / glances

Push — develop ( 420cf2...653997 )

glances.plugins.fs.FsPlugin.get_disk_partitions() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like