Issues in glances_diskio.py - New Issues - Inspection of "Glances version 3.1.6.2" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 4c6c3d...040528 )

by Nicolas

created 2021-02-07 14:38 UTC

glances/plugins/glances_diskio.py (1 issue)

Labels

Duplication 1

Severity

Informational 1

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2019 Nicolargo <[email protected]>
#
# Glances is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Glances is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

"""Disk I/O plugin."""
from __future__ import unicode_literals

from glances.compat import nativestr, n
from glances.timer import getTimeSinceLastUpdate
from glances.plugins.glances_plugin import GlancesPlugin
from glances.logger import logger

import psutil


# Define the history items list
items_history_list = [{'name': 'read_bytes',
                       'description': 'Bytes read per second',
                       'y_unit': 'B/s'},
                      {'name': 'write_bytes',
                       'description': 'Bytes write per second',
                       'y_unit': 'B/s'}]


class Plugin(GlancesPlugin):
    """Glances disks I/O plugin.

    stats is a list
    """

    def __init__(self, args=None, config=None):

        """Init the plugin."""
        super(Plugin, 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
        # Hide stats if it has never been != 0
        self.hide_zero = config.get_bool_value(
            self.plugin_name, 'hide_zero', default=False)
        self.hide_zero_fields = ['read_bytes', 'write_bytes']

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

    @GlancesPlugin._check_decorator
    @GlancesPlugin._log_result_decorator
    def update(self):
        """Update disk I/O 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 stat using the psutil disk_io_counters method
            # read_count: number of reads
            # write_count: number of writes
            # read_bytes: number of bytes read
            # write_bytes: number of bytes written
            # read_time: time spent reading from disk (in milliseconds)
            # write_time: time spent writing to disk (in milliseconds)
            try:
                diskiocounters = psutil.disk_io_counters(perdisk=True)
            except Exception:
                return stats

            # Previous disk IO stats are stored in the diskio_old variable
            if not hasattr(self, 'diskio_old'):
                # First call, we init the diskio_old var
                try:
                    self.diskio_old = diskiocounters
                except (IOError, UnboundLocalError):
                    pass
            else:
                # By storing time data we enable Rx/s and Tx/s calculations in the
                # XML/RPC API, which would otherwise be overly difficult work
                # for users of the API
                time_since_update = getTimeSinceLastUpdate('disk')

                diskio_new = diskiocounters
                for disk in diskio_new:
                    # By default, RamFS is not displayed (issue #714)
                    if self.args is not None and not self.args.diskio_show_ramfs and disk.startswith('ram'):
                        continue

                    # Do not take hide disk into account
                    if self.is_hide(disk):
                        continue

                    # Compute count and bit rate
                    try:
                        read_count = (diskio_new[disk].read_count -
                                      self.diskio_old[disk].read_count)
                        write_count = (diskio_new[disk].write_count -
                                       self.diskio_old[disk].write_count)
                        read_bytes = (diskio_new[disk].read_bytes -
                                      self.diskio_old[disk].read_bytes)
                        write_bytes = (diskio_new[disk].write_bytes -
                                       self.diskio_old[disk].write_bytes)
                        diskstat = {
                            'time_since_update': time_since_update,
                            'disk_name': n(disk),
                            'read_count': read_count,
                            'write_count': write_count,
                            'read_bytes': read_bytes,
                            'write_bytes': write_bytes}
                        # Add alias if exist (define in the configuration file)
                        if self.has_alias(disk) is not None:
                            diskstat['alias'] = self.has_alias(disk)
                    except KeyError:
                        continue
                    else:
                        diskstat['key'] = self.get_key()
                        stats.append(diskstat)

                # Save stats to compute next bitrate
                self.diskio_old = diskio_new
        elif self.input_method == 'snmp':
            # Update stats using SNMP
            # No standard way for the moment...
            pass

        # Update the stats
        self.stats = stats

        return self.stats

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

        # Check if the stats should be hidden
        self.update_views_hidden()

        # Add specifics informations
        # Alert
        for i in self.get_raw():
            disk_real_name = i['disk_name']
            self.views[i[self.get_key()]]['read_bytes']['decoration'] = self.get_alert(int(i['read_bytes'] // i['time_since_update']),
                                                                                       header=disk_real_name + '_rx')
            self.views[i[self.get_key()]]['write_bytes']['decoration'] = self.get_alert(int(i['write_bytes'] // i['time_since_update']),
                                                                                        header=disk_real_name + '_tx')

    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_disable():
            return ret

        # Max size for the interface name
        name_max_width = max_width - 13

        # Header
        msg = '{:{width}}'.format('DISK I/O', width=name_max_width)
        ret.append(self.curse_add_line(msg, "TITLE"))
        if args.diskio_iops:
            msg = '{:>7}'.format('IOR/s')
            ret.append(self.curse_add_line(msg))
            msg = '{:>7}'.format('IOW/s')
            ret.append(self.curse_add_line(msg))
        else:
            msg = '{:>7}'.format('R/s')
            ret.append(self.curse_add_line(msg))
            msg = '{:>7}'.format('W/s')
            ret.append(self.curse_add_line(msg))
        # Disk list (sorted by name)
        for i in self.sorted_stats():
            # Hide stats if never be different from 0 (issue #1787)
            if all([self.get_views(item=i[self.get_key()], key=f, option='hidden') for f in self.hide_zero_fields]):
                continue
            # Is there an alias for the disk name ?
            disk_real_name = i['disk_name']
            disk_name = self.has_alias(i['disk_name'])
            if disk_name is None:
                disk_name = disk_real_name
            # New line
            ret.append(self.curse_new_line())
            if len(disk_name) > name_max_width:
                # Cut disk name if it is too long
                disk_name = '_' + disk_name[-name_max_width+1:]
            msg = '{:{width}}'.format(nativestr(disk_name),
                                      width=name_max_width+1)
            ret.append(self.curse_add_line(msg))
            if args.diskio_iops:
                # count
                txps = self.auto_unit(
                    int(i['read_count'] // i['time_since_update']))
                rxps = self.auto_unit(
                    int(i['write_count'] // i['time_since_update']))
                msg = '{:>7}'.format(txps)
                ret.append(self.curse_add_line(msg,
                                               self.get_views(item=i[self.get_key()],
                                                              key='read_count',
                                                              option='decoration')))
                msg = '{:>7}'.format(rxps)
                ret.append(self.curse_add_line(msg,
                                               self.get_views(item=i[self.get_key()],
                                                              key='write_count',
                                                              option='decoration')))
            else:
                # Bitrate
                txps = self.auto_unit(
                    int(i['read_bytes'] // i['time_since_update']))
                rxps = self.auto_unit(
                    int(i['write_bytes'] // i['time_since_update']))
                msg = '{:>7}'.format(txps)
                ret.append(self.curse_add_line(msg,
                                               self.get_views(item=i[self.get_key()],
                                                              key='read_bytes',
                                                              option='decoration')))
                msg = '{:>7}'.format(rxps)
                ret.append(self.curse_add_line(msg,
                                               self.get_views(item=i[self.get_key()],
                                                              key='write_bytes',
                                                              option='decoration')))

        return ret


1		# -- coding: utf-8 --
2		#
3		# This file is part of Glances.
4		#
5		# Copyright (C) 2019 Nicolargo <[email protected]>
6		#
7		# Glances is free software; you can redistribute it and/or modify
8		# it under the terms of the GNU Lesser General Public License as published by
9		# the Free Software Foundation, either version 3 of the License, or
10		# (at your option) any later version.
11		#
12		# Glances is distributed in the hope that it will be useful,
13		# but WITHOUT ANY WARRANTY; without even the implied warranty of
14		# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15		# GNU Lesser General Public License for more details.
16		#
17		# You should have received a copy of the GNU Lesser General Public License
18		# along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20		"""Disk I/O plugin."""
21		from __future__ import unicode_literals
22
23		from glances.compat import nativestr, n
24		from glances.timer import getTimeSinceLastUpdate
25		from glances.plugins.glances_plugin import GlancesPlugin
26		from glances.logger import logger
27
28		import psutil
29
30
31		# Define the history items list
32		items_history_list = [{'name': 'read_bytes',
33		'description': 'Bytes read per second',
34		'y_unit': 'B/s'},
35		{'name': 'write_bytes',
36		'description': 'Bytes write per second',
37		'y_unit': 'B/s'}]
38
39
40		class Plugin(GlancesPlugin):
41		"""Glances disks I/O plugin.
42
43		stats is a list
44		"""
45
46	View Code Duplication	def __init__(self, args=None, config=None):
		0 ignored issues – show Duplication introduced 2019-08-17 09:24 UTC by Report Bug Copy Issue Report Show Similar Issues like this This code seems to be duplicated in your project. Loading history...
47		"""Init the plugin."""
48		super(Plugin, self).__init__(args=args,
49		config=config,
50		items_history_list=items_history_list,
51		stats_init_value=[])
52
53		# We want to display the stat in the curse interface
54		self.display_curse = True
55		# Hide stats if it has never been != 0
56		self.hide_zero = config.get_bool_value(
57		self.plugin_name, 'hide_zero', default=False)
58		self.hide_zero_fields = ['read_bytes', 'write_bytes']
59
60		def get_key(self):
61		"""Return the key of the list."""
62		return 'disk_name'
63
64		@GlancesPlugin._check_decorator
65		@GlancesPlugin._log_result_decorator
66		def update(self):
67		"""Update disk I/O stats using the input method."""
68		# Init new stats
69		stats = self.get_init_value()
70
71		if self.input_method == 'local':
72		# Update stats using the standard system lib
73		# Grab the stat using the psutil disk_io_counters method
74		# read_count: number of reads
75		# write_count: number of writes
76		# read_bytes: number of bytes read
77		# write_bytes: number of bytes written
78		# read_time: time spent reading from disk (in milliseconds)
79		# write_time: time spent writing to disk (in milliseconds)
80		try:
81		diskiocounters = psutil.disk_io_counters(perdisk=True)
82		except Exception:
83		return stats
84
85		# Previous disk IO stats are stored in the diskio_old variable
86		if not hasattr(self, 'diskio_old'):
87		# First call, we init the diskio_old var
88		try:
89		self.diskio_old = diskiocounters
90		except (IOError, UnboundLocalError):
91		pass
92		else:
93		# By storing time data we enable Rx/s and Tx/s calculations in the
94		# XML/RPC API, which would otherwise be overly difficult work
95		# for users of the API
96		time_since_update = getTimeSinceLastUpdate('disk')
97
98		diskio_new = diskiocounters
99		for disk in diskio_new:
100		# By default, RamFS is not displayed (issue #714)
101		if self.args is not None and not self.args.diskio_show_ramfs and disk.startswith('ram'):
102		continue
103
104		# Do not take hide disk into account
105		if self.is_hide(disk):
106		continue
107
108		# Compute count and bit rate
109		try:
110		read_count = (diskio_new[disk].read_count -
111		self.diskio_old[disk].read_count)
112		write_count = (diskio_new[disk].write_count -
113		self.diskio_old[disk].write_count)
114		read_bytes = (diskio_new[disk].read_bytes -
115		self.diskio_old[disk].read_bytes)
116		write_bytes = (diskio_new[disk].write_bytes -
117		self.diskio_old[disk].write_bytes)
118		diskstat = {
119		'time_since_update': time_since_update,
120		'disk_name': n(disk),
121		'read_count': read_count,
122		'write_count': write_count,
123		'read_bytes': read_bytes,
124		'write_bytes': write_bytes}
125		# Add alias if exist (define in the configuration file)
126		if self.has_alias(disk) is not None:
127		diskstat['alias'] = self.has_alias(disk)
128		except KeyError:
129		continue
130		else:
131		diskstat['key'] = self.get_key()
132		stats.append(diskstat)
133
134		# Save stats to compute next bitrate
135		self.diskio_old = diskio_new
136		elif self.input_method == 'snmp':
137		# Update stats using SNMP
138		# No standard way for the moment...
139		pass
140
141		# Update the stats
142		self.stats = stats
143
144		return self.stats
145
146		def update_views(self):
147		"""Update stats views."""
148		# Call the father's method
149		super(Plugin, self).update_views()
150
151		# Check if the stats should be hidden
152		self.update_views_hidden()
153
154		# Add specifics informations
155		# Alert
156		for i in self.get_raw():
157		disk_real_name = i['disk_name']
158		self.views[i[self.get_key()]]['read_bytes']['decoration'] = self.get_alert(int(i['read_bytes'] // i['time_since_update']),
159		header=disk_real_name + '_rx')
160		self.views[i[self.get_key()]]['write_bytes']['decoration'] = self.get_alert(int(i['write_bytes'] // i['time_since_update']),
161		header=disk_real_name + '_tx')
162
163		def msg_curse(self, args=None, max_width=None):
164		"""Return the dict to display in the curse interface."""
165		# Init the return message
166		ret = []
167
168		# Only process if stats exist and display plugin enable...
169		if not self.stats or self.is_disable():
170		return ret
171
172		# Max size for the interface name
173		name_max_width = max_width - 13
174
175		# Header
176		msg = '{:{width}}'.format('DISK I/O', width=name_max_width)
177		ret.append(self.curse_add_line(msg, "TITLE"))
178		if args.diskio_iops:
179		msg = '{:>7}'.format('IOR/s')
180		ret.append(self.curse_add_line(msg))
181		msg = '{:>7}'.format('IOW/s')
182		ret.append(self.curse_add_line(msg))
183		else:
184		msg = '{:>7}'.format('R/s')
185		ret.append(self.curse_add_line(msg))
186		msg = '{:>7}'.format('W/s')
187		ret.append(self.curse_add_line(msg))
188		# Disk list (sorted by name)
189		for i in self.sorted_stats():
190		# Hide stats if never be different from 0 (issue #1787)
191		if all([self.get_views(item=i[self.get_key()], key=f, option='hidden') for f in self.hide_zero_fields]):
192		continue
193		# Is there an alias for the disk name ?
194		disk_real_name = i['disk_name']
195		disk_name = self.has_alias(i['disk_name'])
196		if disk_name is None:
197		disk_name = disk_real_name
198		# New line
199		ret.append(self.curse_new_line())
200		if len(disk_name) > name_max_width:
201		# Cut disk name if it is too long
202		disk_name = '_' + disk_name[-name_max_width+1:]
203		msg = '{:{width}}'.format(nativestr(disk_name),
204		width=name_max_width+1)
205		ret.append(self.curse_add_line(msg))
206		if args.diskio_iops:
207		# count
208		txps = self.auto_unit(
209		int(i['read_count'] // i['time_since_update']))
210		rxps = self.auto_unit(
211		int(i['write_count'] // i['time_since_update']))
212		msg = '{:>7}'.format(txps)
213		ret.append(self.curse_add_line(msg,
214		self.get_views(item=i[self.get_key()],
215		key='read_count',
216		option='decoration')))
217		msg = '{:>7}'.format(rxps)
218		ret.append(self.curse_add_line(msg,
219		self.get_views(item=i[self.get_key()],
220		key='write_count',
221		option='decoration')))
222		else:
223		# Bitrate
224		txps = self.auto_unit(
225		int(i['read_bytes'] // i['time_since_update']))
226		rxps = self.auto_unit(
227		int(i['write_bytes'] // i['time_since_update']))
228		msg = '{:>7}'.format(txps)
229		ret.append(self.curse_add_line(msg,
230		self.get_views(item=i[self.get_key()],
231		key='read_bytes',
232		option='decoration')))
233		msg = '{:>7}'.format(rxps)
234		ret.append(self.curse_add_line(msg,
235		self.get_views(item=i[self.get_key()],
236		key='write_bytes',
237		option='decoration')))
238
239		return ret
240

nicolargo / glances

Push — master ( 4c6c3d...040528 )

glances/plugins/glances_diskio.py (1 issue)

Labels

Severity

Introduced By

Duplication Side-by-Side

Filter issues like