glances.plugins.memswap.PluginModel.update() - Code Metrics - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

glances.plugins.memswap.PluginModel.update() F
last analyzed 2024-06-08 08:31 UTC

↳ Parent: glances.plugins.memswap

Complexity

Conditions

Size

Total Lines	74
Code Lines	37

Duplication

Lines	35
Ratio	47.3 %

Importance

Changes

Metric	Value
cc	15
eloc	37
nop	1
dl	35
loc	74
rs	2.9998
c	0
b	0
f	0

How to fix Long Method Complexity

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

"""Swap memory plugin."""

import psutil

from glances.globals import iterkeys
from glances.plugins.plugin.model import GlancesPluginModel
from glances.timer import getTimeSinceLastUpdate

# Fields description
fields_description = {
    'total': {'description': 'Total swap memory.', 'unit': 'bytes', 'min_symbol': 'K'},
    'used': {'description': 'Used swap memory.', 'unit': 'bytes', 'min_symbol': 'K'},
    'free': {'description': 'Free swap memory.', 'unit': 'bytes', 'min_symbol': 'K'},
    'percent': {'description': 'Used swap memory in percentage.', 'unit': 'percent'},
    'sin': {
        'description': 'The number of bytes the system has swapped in from disk (cumulative).',
        'unit': 'bytes',
        'min_symbol': 'K',
    },
    'sout': {
        'description': 'The number of bytes the system has swapped out from disk (cumulative).',
        'unit': 'bytes',
        'min_symbol': 'K',
    },
    'time_since_update': {'description': 'Number of seconds since last update.', 'unit': 'seconds'},
}

# SNMP OID
# Total Swap Size: .1.3.6.1.4.1.2021.4.3.0
# Available Swap Space: .1.3.6.1.4.1.2021.4.4.0
snmp_oid = {
    'default': {'total': '1.3.6.1.4.1.2021.4.3.0', 'free': '1.3.6.1.4.1.2021.4.4.0'},
    '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',
    },
}

# 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': 'Swap memory usage', 'y_unit': '%'}]


class PluginModel(GlancesPluginModel):
    """Glances swap memory plugin.

    stats is a dict
    """

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

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

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update swap memory 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 SWAP using the psutil swap_memory method
            try:
                sm_stats = psutil.swap_memory()
            except (OSError, RuntimeError):
                # Crash on startup on Illumos when no swap is configured #1767
                # OpenBSD crash on start without a swap file/partition #2719
                pass
            else:
                # Get all the swap stats (copy/paste of the psutil documentation)
                # total: total swap memory in bytes
                # used: used swap memory in bytes
                # free: free swap memory in bytes
                # percent: the percentage usage
                # sin: the number of bytes the system has swapped in from disk (cumulative)
                # sout: the number of bytes the system has swapped out from disk (cumulative)
                for swap in ['total', 'used', 'free', 'percent', 'sin', 'sout']:
                    if hasattr(sm_stats, swap):
                        stats[swap] = getattr(sm_stats, swap)

                # By storing time data we enable sin/s and sout/s calculations in the
                # XML/RPC API, which would otherwise be overly difficult work
                # for users of the API
                stats['time_since_update'] = getTimeSinceLastUpdate('memswap')
        elif self.input_method == 'snmp':
            # Update stats using SNMP
            if self.short_system_name == 'windows':

                # Mem stats for Windows OS are stored in the FS table
                try:
                    fs_stat = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name], bulk=True)
                except KeyError:
                    self.reset()
                else:
                    for fs in fs_stat:
                        # The virtual memory concept is used by the operating
                        # system to extend (virtually) the physical memory and
                        # thus to run more programs by swapping unused memory
                        # zone (page) to a disk file.
                        if fs == 'Virtual Memory':
                            stats['total'] = int(fs_stat[fs]['size']) * int(fs_stat[fs]['alloc_unit'])
                            stats['used'] = int(fs_stat[fs]['used']) * int(fs_stat[fs]['alloc_unit'])
                            stats['percent'] = float(stats['used'] * 100 / stats['total'])
                            stats['free'] = stats['total'] - stats['used']
                            break
            else:
                stats = self.get_stats_snmp(snmp_oid=snmp_oid['default'])

                if stats['total'] == '':
                    self.reset()
                    return stats

                for key in iterkeys(stats):
                    if stats[key] != '':
                        stats[key] = float(stats[key]) * 1024

                # used=total-free
                stats['used'] = stats['total'] - stats['free']

                # percent: the percentage usage calculated as (total -
                # available) / total * 100.
                stats['percent'] = float((stats['total'] - stats['free']) / stats['total'] * 100)

        # Update the stats
        self.stats = stats

        return self.stats

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

        # Add specifics information
        # Alert and log
        if 'used' in self.stats and 'total' in self.stats and 'percent' in self.stats:
            self.views['percent']['decoration'] = self.get_alert_log(self.stats['used'], maximum=self.stats['total'])

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

        # First line
        # total%
        msg = '{:4}'.format('SWAP')
        ret.append(self.curse_add_line(msg, "TITLE"))
        msg = ' {:2}'.format(self.trend_msg(self.get_trend('percent')))
        ret.append(self.curse_add_line(msg))
        # Percent memory usage
        msg = '{:>6.1%}'.format(self.stats['percent'] / 100)
        ret.append(self.curse_add_line(msg, self.get_views(key='percent', option='decoration')))

        # Second line
        # total
        ret.append(self.curse_new_line())
        # Total memory usage
        ret.extend(self.curse_add_stat('total', width=15))

        # Third line
        # used
        ret.append(self.curse_new_line())
        # Used memory usage
        ret.extend(self.curse_add_stat('used', width=15))

        # Fourth line
        # free
        ret.append(self.curse_new_line())
        # Free memory usage
        ret.extend(self.curse_add_stat('free', width=15))

        return ret


1		#
2		# This file is part of Glances.
3		#
4		# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
5		#
6		# SPDX-License-Identifier: LGPL-3.0-only
7		#
8
9		"""Swap memory plugin."""
10
11		import psutil
12
13		from glances.globals import iterkeys
14		from glances.plugins.plugin.model import GlancesPluginModel
15		from glances.timer import getTimeSinceLastUpdate
16
17		# Fields description
18		fields_description = {
19		'total': {'description': 'Total swap memory.', 'unit': 'bytes', 'min_symbol': 'K'},
20		'used': {'description': 'Used swap memory.', 'unit': 'bytes', 'min_symbol': 'K'},
21		'free': {'description': 'Free swap memory.', 'unit': 'bytes', 'min_symbol': 'K'},
22		'percent': {'description': 'Used swap memory in percentage.', 'unit': 'percent'},
23		'sin': {
24		'description': 'The number of bytes the system has swapped in from disk (cumulative).',
25		'unit': 'bytes',
26		'min_symbol': 'K',
27		},
28		'sout': {
29		'description': 'The number of bytes the system has swapped out from disk (cumulative).',
30		'unit': 'bytes',
31		'min_symbol': 'K',
32		},
33		'time_since_update': {'description': 'Number of seconds since last update.', 'unit': 'seconds'},
34		}
35
36		# SNMP OID
37		# Total Swap Size: .1.3.6.1.4.1.2021.4.3.0
38		# Available Swap Space: .1.3.6.1.4.1.2021.4.4.0
39		snmp_oid = {
40		'default': {'total': '1.3.6.1.4.1.2021.4.3.0', 'free': '1.3.6.1.4.1.2021.4.4.0'},
41		'windows': {
42		'mnt_point': '1.3.6.1.2.1.25.2.3.1.3',
43		'alloc_unit': '1.3.6.1.2.1.25.2.3.1.4',
44		'size': '1.3.6.1.2.1.25.2.3.1.5',
45		'used': '1.3.6.1.2.1.25.2.3.1.6',
46		},
47		}
48
49		# Define the history items list
50		# All items in this list will be historised if the --enable-history tag is set
51		items_history_list = [{'name': 'percent', 'description': 'Swap memory usage', 'y_unit': '%'}]
52
53
54		class PluginModel(GlancesPluginModel):
55		"""Glances swap memory plugin.
56
57		stats is a dict
58		"""
59
60		def __init__(self, args=None, config=None):
61		"""Init the plugin."""
62		super().__init__(
63		args=args, config=config, items_history_list=items_history_list, fields_description=fields_description
64		)
65
66		# We want to display the stat in the curse interface
67		self.display_curse = True
68
69		@GlancesPluginModel._check_decorator
70		@GlancesPluginModel._log_result_decorator
71		def update(self):
72		"""Update swap memory stats using the input method."""
73		# Init new stats
74		stats = self.get_init_value()
75
76		if self.input_method == 'local':
77		# Update stats using the standard system lib
78		# Grab SWAP using the psutil swap_memory method
79		try:
80		sm_stats = psutil.swap_memory()
81		except (OSError, RuntimeError):
82		# Crash on startup on Illumos when no swap is configured #1767
83		# OpenBSD crash on start without a swap file/partition #2719
84		pass
85		else:
86		# Get all the swap stats (copy/paste of the psutil documentation)
87		# total: total swap memory in bytes
88		# used: used swap memory in bytes
89		# free: free swap memory in bytes
90		# percent: the percentage usage
91		# sin: the number of bytes the system has swapped in from disk (cumulative)
92		# sout: the number of bytes the system has swapped out from disk (cumulative)
93		for swap in ['total', 'used', 'free', 'percent', 'sin', 'sout']:
94		if hasattr(sm_stats, swap):
95		stats[swap] = getattr(sm_stats, swap)
96
97		# By storing time data we enable sin/s and sout/s calculations in the
98		# XML/RPC API, which would otherwise be overly difficult work
99		# for users of the API
100		stats['time_since_update'] = getTimeSinceLastUpdate('memswap')
101		elif self.input_method == 'snmp':
102		# Update stats using SNMP
103	View Code Duplication	if self.short_system_name == 'windows':
		0 ignored issues – show Duplication introduced 2024-05-01 11:48 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
104		# Mem stats for Windows OS are stored in the FS table
105		try:
106		fs_stat = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name], bulk=True)
107		except KeyError:
108		self.reset()
109		else:
110		for fs in fs_stat:
111		# The virtual memory concept is used by the operating
112		# system to extend (virtually) the physical memory and
113		# thus to run more programs by swapping unused memory
114		# zone (page) to a disk file.
115		if fs == 'Virtual Memory':
116		stats['total'] = int(fs_stat[fs]['size']) * int(fs_stat[fs]['alloc_unit'])
117		stats['used'] = int(fs_stat[fs]['used']) * int(fs_stat[fs]['alloc_unit'])
118		stats['percent'] = float(stats['used'] * 100 / stats['total'])
119		stats['free'] = stats['total'] - stats['used']
120		break
121		else:
122		stats = self.get_stats_snmp(snmp_oid=snmp_oid['default'])
123
124		if stats['total'] == '':
125		self.reset()
126		return stats
127
128		for key in iterkeys(stats):
129		if stats[key] != '':
130		stats[key] = float(stats[key]) * 1024
131
132		# used=total-free
133		stats['used'] = stats['total'] - stats['free']
134
135		# percent: the percentage usage calculated as (total -
136		# available) / total * 100.
137		stats['percent'] = float((stats['total'] - stats['free']) / stats['total'] * 100)
138
139		# Update the stats
140		self.stats = stats
141
142		return self.stats
143
144		def update_views(self):
145		"""Update stats views."""
146		# Call the father's method
147		super().update_views()
148
149		# Add specifics information
150		# Alert and log
151		if 'used' in self.stats and 'total' in self.stats and 'percent' in self.stats:
152		self.views['percent']['decoration'] = self.get_alert_log(self.stats['used'], maximum=self.stats['total'])
153
154		def msg_curse(self, args=None, max_width=None):
155		"""Return the dict to display in the curse interface."""
156		# Init the return message
157		ret = []
158
159		# Only process if stats exist and plugin not disabled
160		if not self.stats or self.is_disabled():
161		return ret
162
163		# First line
164		# total%
165		msg = '{:4}'.format('SWAP')
166		ret.append(self.curse_add_line(msg, "TITLE"))
167		msg = ' {:2}'.format(self.trend_msg(self.get_trend('percent')))
168		ret.append(self.curse_add_line(msg))
169		# Percent memory usage
170		msg = '{:>6.1%}'.format(self.stats['percent'] / 100)
171		ret.append(self.curse_add_line(msg, self.get_views(key='percent', option='decoration')))
172
173		# Second line
174		# total
175		ret.append(self.curse_new_line())
176		# Total memory usage
177		ret.extend(self.curse_add_stat('total', width=15))
178
179		# Third line
180		# used
181		ret.append(self.curse_new_line())
182		# Used memory usage
183		ret.extend(self.curse_add_stat('used', width=15))
184
185		# Fourth line
186		# free
187		ret.append(self.curse_new_line())
188		# Free memory usage
189		ret.extend(self.curse_add_stat('free', width=15))
190
191		return ret
192

nicolargo / glances

glances.plugins.memswap.PluginModel.update() F last analyzed 2024-06-08 08:31 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

glances.plugins.memswap.PluginModel.update() F
last analyzed 2024-06-08 08:31 UTC