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

glances.plugins.network.PluginModel.get_key() A
last analyzed 2024-06-18 00:16 UTC

↳ Parent: glances.plugins.network

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

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

"""Network plugin."""

import psutil

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: if True then compute and add *_gauge and *_rate_per_is fields
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'interface_name': {'description': 'Interface name.'},
    'alias': {'description': 'Interface alias name (optional).'},
    'bytes_recv': {
        'description': 'Number of bytes received.',
        'rate': True,
        'unit': 'byte',
    },
    'bytes_sent': {
        'description': 'Number of bytes sent.',
        'rate': True,
        'unit': 'byte',
    },
    'bytes_all': {
        'description': 'Number of bytes received and sent.',
        'rate': True,
        'unit': 'byte',
    },
    'speed': {
        'description': 'Maximum interface speed (in bit per second). Can return 0 on some operating-system.',
        'unit': 'bitpersecond',
    },
    'is_up': {'description': 'Is the interface up ?', 'unit': 'bool'},
}

# SNMP OID
# http://www.net-snmp.org/docs/mibs/interfaces.html
# Dict key = interface_name
snmp_oid = {
    'default': {
        'interface_name': '1.3.6.1.2.1.2.2.1.2',
        'bytes_recv': '1.3.6.1.2.1.2.2.1.10',
        'bytes_sent': '1.3.6.1.2.1.2.2.1.16',
    }
}

# Define the history items list
items_history_list = [
    {'name': 'bytes_recv_rate_per_sec', 'description': 'Download rate per second', 'y_unit': 'B/s'},
    {'name': 'bytes_sent_rate_per_sec', 'description': 'Upload rate per second', 'y_unit': 'B/s'},
]


class PluginModel(GlancesPluginModel):
    """Glances network 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,
            fields_description=fields_description,
            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
        if config is not None:
            self.hide_zero = config.get_bool_value(self.plugin_name, 'hide_zero', default=False)
        else:
            self.hide_zero = False
        self.hide_zero_fields = ['bytes_recv', 'bytes_sent']

        #  Add support for automatically hiding network interfaces that are down
        # or that don't have any IP addresses #2799
        self.hide_no_up = config.get_bool_value(self.plugin_name, 'hide_no_up', default=False)
        self.hide_no_ip = config.get_bool_value(self.plugin_name, 'hide_no_ip', default=False)

        # Force a first update because we need two updates to have the first stat
        self.update()
        self.refresh_timer.set(0)

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

    # @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update network stats using the input method.

        :return: list of stats dict (one dict per interface)
        """
        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._manage_rate
    def update_local(self):
        # Update stats using the standard system lib

        stats = self.get_init_value()

        # Grab network interface stat using the psutil net_io_counter method
        # Example:
        # { 'veth4cbf8f0a': snetio(
        #   bytes_sent=102038421, bytes_recv=1263258,
        #   packets_sent=25046, packets_recv=14114,
        #   errin=0, errout=0, dropin=0, dropout=0), ... }
        try:
            net_io_counters = psutil.net_io_counters(pernic=True)
        except UnicodeDecodeError as e:
            logger.debug(f'Can not get network interface counters ({e})')
            return self.stats

        # Grab interface's status (issue #765)
        # Grab interface's speed (issue #718)
        # Example:
        # { 'veth4cbf8f0a': snicstats(
        #   isup=True, duplex=<NicDuplex.NIC_DUPLEX_FULL: 2>,
        #   speed=10000, mtu=1500, flags='up,broadcast,running,multicast'), ... }
        net_status = {}
        try:
            net_status = psutil.net_if_stats()
            net_addrs = psutil.net_if_addrs()
        except OSError as e:
            # see psutil #797/glances #1106
            logger.debug(f'Can not get network interface status ({e})')

        # Filter interfaces (related to #2799)
        if self.hide_no_up:
            net_status = {k: v for k, v in net_status.items() if v.isup}
        if self.hide_no_ip:
            net_status = {
                k: v
                for k, v in net_status.items()
                if k in net_addrs and any(a.family != psutil.AF_LINK for a in net_addrs[k])

            }

        for interface_name, interface_stat in net_io_counters.items():
            # Do not take hidden interface into account
            # or KeyError: 'eth0' when interface is not connected #1348
            if not self.is_display(interface_name) or interface_name not in net_status:
                continue

            # Filter stats to keep only the fields we want (define in fields_description)
            # It will also convert psutil objects to a standard Python dict
            stat = self.filter_stats(interface_stat)
            stat.update(self.filter_stats(net_status[interface_name]))

            # Add the key
            stat['key'] = self.get_key()

            # Add disk name
            stat['interface_name'] = interface_name

            # Add alias define in the configuration file
            stat['alias'] = self.has_alias(interface_name)

            # Add sent + revc  stats
            stat['bytes_all'] = stat['bytes_sent'] + stat['bytes_recv']

            # Interface speed in Mbps, convert it to bps
            # Can be always 0 on some OSes
            stat['speed'] = stat['speed'] * 1048576

            stats.append(stat)

        return stats

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

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

        # Add specifics information
        # Alert
        for i in self.get_raw():
            # Skip alert if no timespan to measure
            if 'bytes_recv_rate_per_sec' not in i or 'bytes_sent_rate_per_sec' not in i:
                continue

            # Convert rate to bps (to be able to compare to interface speed)
            bps_rx = int(i['bytes_recv_rate_per_sec'] * 8)
            bps_tx = int(i['bytes_sent_rate_per_sec'] * 8)

            # Decorate the bitrate with the configuration file thresholds
            if_real_name = i['interface_name'].split(':')[0]
            alert_rx = self.get_alert(bps_rx, header=if_real_name + '_rx')
            alert_tx = self.get_alert(bps_tx, header=if_real_name + '_tx')

            # If nothing is define in the configuration file...
            # ... then use the interface speed (not available on all systems)
            if alert_rx == 'DEFAULT' and 'speed' in i and i['speed'] != 0:
                alert_rx = self.get_alert(current=bps_rx, maximum=i['speed'], header='rx')
            if alert_tx == 'DEFAULT' and 'speed' in i and i['speed'] != 0:
                alert_tx = self.get_alert(current=bps_tx, maximum=i['speed'], header='tx')

            # then decorates
            self.views[i[self.get_key()]]['bytes_recv']['decoration'] = alert_rx
            self.views[i[self.get_key()]]['bytes_sent']['decoration'] = alert_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_disabled():
            return ret

        # Max size for the interface name
        if max_width:
            name_max_width = max_width - 12
        else:
            # No max_width defined, return an emptu curse message
            logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
            return ret

        # Header
        msg = '{:{width}}'.format('NETWORK', width=name_max_width)
        ret.append(self.curse_add_line(msg, "TITLE"))
        if args.network_cumul:
            # Cumulative stats
            if args.network_sum:
                # Sum stats
                msg = '{:>14}'.format('Rx+Tx')
                ret.append(self.curse_add_line(msg))
            else:
                # Rx/Tx stats
                msg = '{:>7}'.format('Rx')
                ret.append(self.curse_add_line(msg))
                msg = '{:>7}'.format('Tx')
                ret.append(self.curse_add_line(msg))
        else:
            # Bitrate stats
            if args.network_sum:
                # Sum stats
                msg = '{:>14}'.format('Rx+Tx/s')
                ret.append(self.curse_add_line(msg))
            else:
                msg = '{:>7}'.format('Rx/s')
                ret.append(self.curse_add_line(msg))
                msg = '{:>7}'.format('Tx/s')
                ret.append(self.curse_add_line(msg))
        # Interface list (sorted by name)
        for i in self.sorted_stats():
            # Do not display interface in down state (issue #765)
            if ('is_up' in i) and (i['is_up'] is False):
                continue
            # 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
            # Format stats
            # Is there an alias for the interface name ?
            if i['alias'] is None:
                if_name = i['interface_name'].split(':')[0]
            else:
                if_name = i['alias']
            if len(if_name) > name_max_width:
                # Cut interface name if it is too long
                if_name = '_' + if_name[-name_max_width + 1 :]

            if args.byte:
                # Bytes per second (for dummy)
                to_bit = 1
                unit = ''
            else:
                # Bits per second (for real network administrator | Default)
                to_bit = 8
                unit = 'b'

            if args.network_cumul and 'bytes_recv' in i:
                rx = self.auto_unit(int(i['bytes_recv'] * to_bit)) + unit
                tx = self.auto_unit(int(i['bytes_sent'] * to_bit)) + unit
                ax = self.auto_unit(int(i['bytes_all'] * to_bit)) + unit
            elif 'bytes_recv_rate_per_sec' in i:
                rx = self.auto_unit(int(i['bytes_recv_rate_per_sec'] * to_bit)) + unit
                tx = self.auto_unit(int(i['bytes_sent_rate_per_sec'] * to_bit)) + unit
                ax = self.auto_unit(int(i['bytes_all_rate_per_sec'] * to_bit)) + unit
            else:
                # Avoid issue when a new interface is created on the fly
                # Example: start Glances, then start a new container
                continue

            # New line
            ret.append(self.curse_new_line())
            msg = '{:{width}}'.format(if_name, width=name_max_width)
            ret.append(self.curse_add_line(msg))
            if args.network_sum:
                msg = f'{ax:>14}'
                ret.append(self.curse_add_line(msg))
            else:
                msg = f'{rx:>7}'
                ret.append(
                    self.curse_add_line(
                        msg, self.get_views(item=i[self.get_key()], key='bytes_recv', option='decoration')
                    )
                )
                msg = f'{tx:>7}'
                ret.append(
                    self.curse_add_line(
                        msg, self.get_views(item=i[self.get_key()], key='bytes_sent', option='decoration')
                    )
                )

        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			"""Network plugin."""
10
11			import psutil
12
13			from glances.logger import logger
14			from glances.plugins.plugin.model import GlancesPluginModel
15
16			# Fields description
17			# description: human readable description
18			# short_name: shortname to use un UI
19			# unit: unit type
20			# rate: if True then compute and add _gauge and _rate_per_is fields
21			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
22			fields_description = {
23			'interface_name': {'description': 'Interface name.'},
24			'alias': {'description': 'Interface alias name (optional).'},
25			'bytes_recv': {
26			'description': 'Number of bytes received.',
27			'rate': True,
28			'unit': 'byte',
29			},
30			'bytes_sent': {
31			'description': 'Number of bytes sent.',
32			'rate': True,
33			'unit': 'byte',
34			},
35			'bytes_all': {
36			'description': 'Number of bytes received and sent.',
37			'rate': True,
38			'unit': 'byte',
39			},
40			'speed': {
41			'description': 'Maximum interface speed (in bit per second). Can return 0 on some operating-system.',
42			'unit': 'bitpersecond',
43			},
44			'is_up': {'description': 'Is the interface up ?', 'unit': 'bool'},
45			}
46
47			# SNMP OID
48			# http://www.net-snmp.org/docs/mibs/interfaces.html
49			# Dict key = interface_name
50			snmp_oid = {
51			'default': {
52			'interface_name': '1.3.6.1.2.1.2.2.1.2',
53			'bytes_recv': '1.3.6.1.2.1.2.2.1.10',
54			'bytes_sent': '1.3.6.1.2.1.2.2.1.16',
55			}
56			}
57
58			# Define the history items list
59			items_history_list = [
60			{'name': 'bytes_recv_rate_per_sec', 'description': 'Download rate per second', 'y_unit': 'B/s'},
61			{'name': 'bytes_sent_rate_per_sec', 'description': 'Upload rate per second', 'y_unit': 'B/s'},
62			]
63
64
65			class PluginModel(GlancesPluginModel):
66			"""Glances network plugin.
67
68			stats is a list
69			"""
70
71			def __init__(self, args=None, config=None):
72			"""Init the plugin."""
73			super().__init__(
74			args=args,
75			config=config,
76			items_history_list=items_history_list,
77			fields_description=fields_description,
78			stats_init_value=[],
79			)
80
81			# We want to display the stat in the curse interface
82			self.display_curse = True
83
84			# Hide stats if it has never been != 0
85			if config is not None:
86			self.hide_zero = config.get_bool_value(self.plugin_name, 'hide_zero', default=False)
87			else:
88			self.hide_zero = False
89			self.hide_zero_fields = ['bytes_recv', 'bytes_sent']
90
91			# Add support for automatically hiding network interfaces that are down
92			# or that don't have any IP addresses #2799
93			self.hide_no_up = config.get_bool_value(self.plugin_name, 'hide_no_up', default=False)
94			self.hide_no_ip = config.get_bool_value(self.plugin_name, 'hide_no_ip', default=False)
95
96			# Force a first update because we need two updates to have the first stat
97			self.update()
98			self.refresh_timer.set(0)
99
100			def get_key(self):
101			"""Return the key of the list."""
102			return 'interface_name'
103
104			# @GlancesPluginModel._check_decorator
105			@GlancesPluginModel._log_result_decorator
106			def update(self):
107			"""Update network stats using the input method.
108
109			:return: list of stats dict (one dict per interface)
110			"""
111			if self.input_method == 'local':
112			stats = self.update_local()
113			else:
114			stats = self.get_init_value()
115
116			# Update the stats
117			self.stats = stats
118
119			return self.stats
120
121			@GlancesPluginModel._manage_rate
122			def update_local(self):
123			# Update stats using the standard system lib
124
125			stats = self.get_init_value()
126
127			# Grab network interface stat using the psutil net_io_counter method
128			# Example:
129			# { 'veth4cbf8f0a': snetio(
130			# bytes_sent=102038421, bytes_recv=1263258,
131			# packets_sent=25046, packets_recv=14114,
132			# errin=0, errout=0, dropin=0, dropout=0), ... }
133			try:
134			net_io_counters = psutil.net_io_counters(pernic=True)
135			except UnicodeDecodeError as e:
136			logger.debug(f'Can not get network interface counters ({e})')
137			return self.stats
138
139			# Grab interface's status (issue #765)
140			# Grab interface's speed (issue #718)
141			# Example:
142			# { 'veth4cbf8f0a': snicstats(
143			# isup=True, duplex=<NicDuplex.NIC_DUPLEX_FULL: 2>,
144			# speed=10000, mtu=1500, flags='up,broadcast,running,multicast'), ... }
145			net_status = {}
146			try:
147			net_status = psutil.net_if_stats()
148			net_addrs = psutil.net_if_addrs()
149			except OSError as e:
150			# see psutil #797/glances #1106
151			logger.debug(f'Can not get network interface status ({e})')
152
153			# Filter interfaces (related to #2799)
154			if self.hide_no_up:
155			net_status = {k: v for k, v in net_status.items() if v.isup}
156			if self.hide_no_ip:
157			net_status = {
158			k: v
159			for k, v in net_status.items()
160			if k in net_addrs and any(a.family != psutil.AF_LINK for a in net_addrs[k])
			0 ignored issues – show introduced 2024-06-01 09:05 UTC by Report Bug Copy Issue Report The variable `net_addrs` does not seem to be defined for all execution paths. Loading history...
161			}
162
163			for interface_name, interface_stat in net_io_counters.items():
164			# Do not take hidden interface into account
165			# or KeyError: 'eth0' when interface is not connected #1348
166			if not self.is_display(interface_name) or interface_name not in net_status:
167			continue
168
169			# Filter stats to keep only the fields we want (define in fields_description)
170			# It will also convert psutil objects to a standard Python dict
171			stat = self.filter_stats(interface_stat)
172			stat.update(self.filter_stats(net_status[interface_name]))
173
174			# Add the key
175			stat['key'] = self.get_key()
176
177			# Add disk name
178			stat['interface_name'] = interface_name
179
180			# Add alias define in the configuration file
181			stat['alias'] = self.has_alias(interface_name)
182
183			# Add sent + revc stats
184			stat['bytes_all'] = stat['bytes_sent'] + stat['bytes_recv']
185
186			# Interface speed in Mbps, convert it to bps
187			# Can be always 0 on some OSes
188			stat['speed'] = stat['speed'] * 1048576
189
190			stats.append(stat)
191
192			return stats
193
194			def update_views(self):
195			"""Update stats views."""
196			# Call the father's method
197			super().update_views()
198
199			# Check if the stats should be hidden
200			self.update_views_hidden()
201
202			# Add specifics information
203			# Alert
204			for i in self.get_raw():
205			# Skip alert if no timespan to measure
206			if 'bytes_recv_rate_per_sec' not in i or 'bytes_sent_rate_per_sec' not in i:
207			continue
208
209			# Convert rate to bps (to be able to compare to interface speed)
210			bps_rx = int(i['bytes_recv_rate_per_sec'] * 8)
211			bps_tx = int(i['bytes_sent_rate_per_sec'] * 8)
212
213			# Decorate the bitrate with the configuration file thresholds
214			if_real_name = i['interface_name'].split(':')[0]
215			alert_rx = self.get_alert(bps_rx, header=if_real_name + '_rx')
216			alert_tx = self.get_alert(bps_tx, header=if_real_name + '_tx')
217
218			# If nothing is define in the configuration file...
219			# ... then use the interface speed (not available on all systems)
220			if alert_rx == 'DEFAULT' and 'speed' in i and i['speed'] != 0:
221			alert_rx = self.get_alert(current=bps_rx, maximum=i['speed'], header='rx')
222			if alert_tx == 'DEFAULT' and 'speed' in i and i['speed'] != 0:
223			alert_tx = self.get_alert(current=bps_tx, maximum=i['speed'], header='tx')
224
225			# then decorates
226			self.views[i[self.get_key()]]['bytes_recv']['decoration'] = alert_rx
227			self.views[i[self.get_key()]]['bytes_sent']['decoration'] = alert_tx
228
229			def msg_curse(self, args=None, max_width=None):
230			"""Return the dict to display in the curse interface."""
231			# Init the return message
232			ret = []
233
234			# Only process if stats exist and display plugin enable...
235			if not self.stats or self.is_disabled():
236			return ret
237
238			# Max size for the interface name
239			if max_width:
240			name_max_width = max_width - 12
241			else:
242			# No max_width defined, return an emptu curse message
243			logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
244			return ret
245
246			# Header
247			msg = '{:{width}}'.format('NETWORK', width=name_max_width)
248			ret.append(self.curse_add_line(msg, "TITLE"))
249			if args.network_cumul:
250			# Cumulative stats
251			if args.network_sum:
252			# Sum stats
253			msg = '{:>14}'.format('Rx+Tx')
254			ret.append(self.curse_add_line(msg))
255			else:
256			# Rx/Tx stats
257			msg = '{:>7}'.format('Rx')
258			ret.append(self.curse_add_line(msg))
259			msg = '{:>7}'.format('Tx')
260			ret.append(self.curse_add_line(msg))
261			else:
262			# Bitrate stats
263			if args.network_sum:
264			# Sum stats
265			msg = '{:>14}'.format('Rx+Tx/s')
266			ret.append(self.curse_add_line(msg))
267			else:
268			msg = '{:>7}'.format('Rx/s')
269			ret.append(self.curse_add_line(msg))
270			msg = '{:>7}'.format('Tx/s')
271			ret.append(self.curse_add_line(msg))
272			# Interface list (sorted by name)
273			for i in self.sorted_stats():
274			# Do not display interface in down state (issue #765)
275			if ('is_up' in i) and (i['is_up'] is False):
276			continue
277			# Hide stats if never be different from 0 (issue #1787)
278			if all(self.get_views(item=i[self.get_key()], key=f, option='hidden') for f in self.hide_zero_fields):
279			continue
280			# Format stats
281			# Is there an alias for the interface name ?
282			if i['alias'] is None:
283			if_name = i['interface_name'].split(':')[0]
284			else:
285			if_name = i['alias']
286			if len(if_name) > name_max_width:
287			# Cut interface name if it is too long
288			if_name = '_' + if_name[-name_max_width + 1 :]
289
290			if args.byte:
291			# Bytes per second (for dummy)
292			to_bit = 1
293			unit = ''
294			else:
295			# Bits per second (for real network administrator \| Default)
296			to_bit = 8
297			unit = 'b'
298
299			if args.network_cumul and 'bytes_recv' in i:
300			rx = self.auto_unit(int(i['bytes_recv'] * to_bit)) + unit
301			tx = self.auto_unit(int(i['bytes_sent'] * to_bit)) + unit
302			ax = self.auto_unit(int(i['bytes_all'] * to_bit)) + unit
303			elif 'bytes_recv_rate_per_sec' in i:
304			rx = self.auto_unit(int(i['bytes_recv_rate_per_sec'] * to_bit)) + unit
305			tx = self.auto_unit(int(i['bytes_sent_rate_per_sec'] * to_bit)) + unit
306			ax = self.auto_unit(int(i['bytes_all_rate_per_sec'] * to_bit)) + unit
307			else:
308			# Avoid issue when a new interface is created on the fly
309			# Example: start Glances, then start a new container
310			continue
311
312			# New line
313			ret.append(self.curse_new_line())
314			msg = '{:{width}}'.format(if_name, width=name_max_width)
315			ret.append(self.curse_add_line(msg))
316			if args.network_sum:
317			msg = f'{ax:>14}'
318			ret.append(self.curse_add_line(msg))
319			else:
320			msg = f'{rx:>7}'
321			ret.append(
322			self.curse_add_line(
323			msg, self.get_views(item=i[self.get_key()], key='bytes_recv', option='decoration')
324			)
325			)
326			msg = f'{tx:>7}'
327			ret.append(
328			self.curse_add_line(
329			msg, self.get_views(item=i[self.get_key()], key='bytes_sent', option='decoration')
330			)
331			)
332
333			return ret
334

nicolargo / glances

glances.plugins.network.PluginModel.get_key() A last analyzed 2024-06-18 00:16 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

glances.plugins.network.PluginModel.get_key() A
last analyzed 2024-06-18 00:16 UTC