glances.plugins.fs.PluginModel.update() - Code Metrics - Inspection of "Refactor plugin folder in the same way than export..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 30cc9f...48251c )

by Nicolas

created 2023-10-15 16:05 UTC

glances.plugins.fs.PluginModel.update() F

↳ Parent: glances.plugins.fs

Complexity

Conditions

Size

Total Lines	122
Code Lines	76

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	24
eloc	76
nop	1
dl	0
loc	122
rs	0
c	0
b	0
f	0

How to fix Long Method Complexity

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

"""File system plugin."""
from __future__ import unicode_literals

import operator

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

import psutil

# 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': '%'}]


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

    stats is a list
    """

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

        # 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."""
        # Init new stats
        stats = self.get_init_value()

        if self.input_method == 'local':
            # Update stats using the standard system lib

            # Grab the stats using the psutil disk_partitions
            # If 'all'=False return physical devices only (e.g. hard disks, cd-rom drives, USB keys)
            # and ignore all others (e.g. memory partitions such as /dev/shm)
            try:
                fs_stat = psutil.disk_partitions(all=False)
            except (UnicodeDecodeError, PermissionError):
                logger.debug("Plugin - fs: PsUtil fetch failed")
                return self.stats

            # 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
                try:
                    all_mounted_fs = psutil.disk_partitions(all=True)
                except (UnicodeDecodeError, PermissionError):
                    logger.debug("Plugin - fs: PsUtil extended fetch failed")
                else:
                    # Discard duplicates (#2299) and add entries matching allowed fs types
                    tracked_mnt_points = set(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
                if not self.is_display(fs.mountpoint):
                    continue

                # Grab the disk usage
                try:
                    fs_usage = psutil.disk_usage(fs.mountpoint)
                except OSError:
                    # Correct issue #346
                    # Disk is ejected during the command
                    continue
                fs_current = {
                    'device_name': fs.device,
                    'fs_type': fs.fstype,
                    # Manage non breaking space (see issue #1065)
                    'mnt_point': u(fs.mountpoint).replace(u'\u00A0', ' '),
                    'size': fs_usage.total,
                    'used': fs_usage.used,
                    'free': fs_usage.free,
                    'percent': fs_usage.percent,
                    'key': self.get_key(),
                }

                # Hide the stats if the device name is in the exclude list
                # Correct issue: glances.conf FS hide not applying #1666
                if not self.is_display(fs_current['device_name']):
                    continue

                stats.append(fs_current)

        elif self.input_method == 'snmp':
            # 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 in fs_stat:
                    # 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_stat[fs]['size']) * int(fs_stat[fs]['alloc_unit'])
                    used = int(fs_stat[fs]['used']) * int(fs_stat[fs]['alloc_unit'])
                    percent = float(used * 100 / size)
                    fs_current = {
                        'device_name': '',
                        'mnt_point': fs.partition(' ')[0],
                        'size': size,
                        'used': used,
                        'percent': percent,
                        'key': self.get_key(),
                    }
                    # Do not take hidden file system into account
                    if self.is_hide(fs_current['mnt_point']):
                        continue
                    else:
                        stats.append(fs_current)
            else:
                # Default behavior
                for fs in fs_stat:
                    fs_current = {
                        'device_name': fs_stat[fs]['device_name'],
                        'mnt_point': fs,
                        'size': int(fs_stat[fs]['size']) * 1024,
                        'used': int(fs_stat[fs]['used']) * 1024,
                        'percent': float(fs_stat[fs]['percent']),
                        'key': self.get_key(),
                    }
                    # Do not take hidden file system into account
                    if self.is_hide(fs_current['mnt_point']) or self.is_hide(fs_current['device_name']):
                        continue
                    else:
                        stats.append(fs_current)

        # Update the stats
        self.stats = stats

        return self.stats

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

        # Add specifics information
        # Alert
        for i in self.stats:
            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
        name_max_width = max_width - 12

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

nicolargo / glances

Push — develop ( 30cc9f...48251c )

glances.plugins.fs.PluginModel.update() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like