glances.plugins.glances_connections.Plugin.update() - Code Metrics - Inspection of "Merge branch 'master' into develop" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 05aaee...10b5c2 )

by Nicolas

created 2023-05-16 07:38 UTC

Plugin.update() F

↳ Parent: glances.plugins.glances_connections

Complexity

Conditions

Size

Total Lines	63
Code Lines	46

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	14
eloc	46
nop	1
dl	0
loc	63
rs	3.6
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
#

"""Connections plugin."""
from __future__ import unicode_literals

from glances.logger import logger
from glances.plugins.glances_plugin import GlancesPlugin
from glances.compat import nativestr, FileNotFoundError

import psutil

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


class Plugin(GlancesPlugin):
    """Glances connections plugin.

    stats is a dict
    """

    status_list = [psutil.CONN_LISTEN, psutil.CONN_ESTABLISHED]
    initiated_states = [psutil.CONN_SYN_SENT, psutil.CONN_SYN_RECV]
    terminated_states = [
        psutil.CONN_FIN_WAIT1,
        psutil.CONN_FIN_WAIT2,
        psutil.CONN_TIME_WAIT,
        psutil.CONN_CLOSE,
        psutil.CONN_CLOSE_WAIT,
        psutil.CONN_LAST_ACK,
    ]
    conntrack = {
        'nf_conntrack_count': '/proc/sys/net/netfilter/nf_conntrack_count',
        'nf_conntrack_max': '/proc/sys/net/netfilter/nf_conntrack_max',
    }

    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={'net_connections_enabled': True, 'nf_conntrack_enabled': True},
        )

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

    @GlancesPlugin._check_decorator
    @GlancesPlugin._log_result_decorator
    def update(self):
        """Update connections stats using the input method.

        Stats is a dict
        """
        # Init new stats
        stats = self.get_init_value()

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

            # Grab network interface stat using the psutil net_connections method
            if stats['net_connections_enabled']:
                try:
                    net_connections = psutil.net_connections(kind="tcp")
                except Exception as e:
                    logger.warning('Can not get network connections stats ({})'.format(e))
                    logger.info('Disable connections stats')
                    stats['net_connections_enabled'] = False
                    self.stats = stats
                    return self.stats

                for s in self.status_list:
                    stats[s] = len([c for c in net_connections if c.status == s])
                initiated = 0
                for s in self.initiated_states:
                    stats[s] = len([c for c in net_connections if c.status == s])
                    initiated += stats[s]
                stats['initiated'] = initiated
                terminated = 0
                for s in self.initiated_states:
                    stats[s] = len([c for c in net_connections if c.status == s])
                    terminated += stats[s]
                stats['terminated'] = terminated

            if stats['nf_conntrack_enabled']:
                # Grab connections track directly from the /proc file
                for i in self.conntrack:
                    try:
                        with open(self.conntrack[i], 'r') as f:
                            stats[i] = float(f.readline().rstrip("\n"))
                    except (IOError, FileNotFoundError) as e:
                        logger.warning('Can not get network connections track ({})'.format(e))
                        logger.info('Disable connections track')
                        stats['nf_conntrack_enabled'] = False
                        self.stats = stats
                        return self.stats
                if 'nf_conntrack_max' in stats and 'nf_conntrack_count' in stats:
                    stats['nf_conntrack_percent'] = stats['nf_conntrack_count'] * 100 / stats['nf_conntrack_max']
                else:
                    stats['nf_conntrack_enabled'] = False
                    self.stats = stats
                    return self.stats

        elif self.input_method == 'snmp':
            # Update stats using SNMP
            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()

        # Add specific information
        try:
            # Alert and log
            if self.stats['nf_conntrack_enabled']:
                self.views['nf_conntrack_percent']['decoration'] = self.get_alert(header='nf_conntrack_percent')
        except KeyError:
            # try/except mandatory for Windows compatibility (no conntrack stats)
            pass

    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

        # Header
        if self.stats['net_connections_enabled'] or self.stats['nf_conntrack_enabled']:
            msg = '{}'.format('TCP CONNECTIONS')
            ret.append(self.curse_add_line(msg, "TITLE"))
        # Connections status
        if self.stats['net_connections_enabled']:
            for s in [psutil.CONN_LISTEN, 'initiated', psutil.CONN_ESTABLISHED, 'terminated']:
                ret.append(self.curse_new_line())
                msg = '{:{width}}'.format(nativestr(s).capitalize(), width=len(s))
                ret.append(self.curse_add_line(msg))
                msg = '{:>{width}}'.format(self.stats[s], width=max_width - len(s) + 2)
                ret.append(self.curse_add_line(msg))
        # Connections track
        if self.stats['nf_conntrack_enabled']:
            s = 'Tracked'
            ret.append(self.curse_new_line())
            msg = '{:{width}}'.format(nativestr(s).capitalize(), width=len(s))
            ret.append(self.curse_add_line(msg))
            msg = '{:>{width}}'.format(
                '{:0.0f}/{:0.0f}'.format(self.stats['nf_conntrack_count'], self.stats['nf_conntrack_max']),
                width=max_width - len(s) + 2,
            )
            ret.append(self.curse_add_line(msg, self.get_views(key='nf_conntrack_percent', option='decoration')))

        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			"""Connections plugin."""
11			from __future__ import unicode_literals
12
13			from glances.logger import logger
14			from glances.plugins.glances_plugin import GlancesPlugin
15			from glances.compat import nativestr, FileNotFoundError
16
17			import psutil
18
19			# Define the history items list
20			# items_history_list = [{'name': 'rx',
21			# 'description': 'Download rate per second',
22			# 'y_unit': 'bit/s'},
23			# {'name': 'tx',
24			# 'description': 'Upload rate per second',
25			# 'y_unit': 'bit/s'}]
26
27
28			class Plugin(GlancesPlugin):
29			"""Glances connections plugin.
30
31			stats is a dict
32			"""
33
34			status_list = [psutil.CONN_LISTEN, psutil.CONN_ESTABLISHED]
35			initiated_states = [psutil.CONN_SYN_SENT, psutil.CONN_SYN_RECV]
36			terminated_states = [
37			psutil.CONN_FIN_WAIT1,
38			psutil.CONN_FIN_WAIT2,
39			psutil.CONN_TIME_WAIT,
40			psutil.CONN_CLOSE,
41			psutil.CONN_CLOSE_WAIT,
42			psutil.CONN_LAST_ACK,
43			]
44			conntrack = {
45			'nf_conntrack_count': '/proc/sys/net/netfilter/nf_conntrack_count',
46			'nf_conntrack_max': '/proc/sys/net/netfilter/nf_conntrack_max',
47			}
48
49			def __init__(self, args=None, config=None):
50			"""Init the plugin."""
51			super(Plugin, self).__init__(
52			args=args,
53			config=config,
54			# items_history_list=items_history_list,
55			stats_init_value={'net_connections_enabled': True, 'nf_conntrack_enabled': True},
56			)
57
58			# We want to display the stat in the curse interface
59			self.display_curse = True
60
61			@GlancesPlugin._check_decorator
62			@GlancesPlugin._log_result_decorator
63			def update(self):
64			"""Update connections stats using the input method.
65
66			Stats is a dict
67			"""
68			# Init new stats
69			stats = self.get_init_value()
70
71			if self.input_method == 'local':
72			# Update stats using the PSUtils lib
73
74			# Grab network interface stat using the psutil net_connections method
75			if stats['net_connections_enabled']:
76			try:
77			net_connections = psutil.net_connections(kind="tcp")
78			except Exception as e:
79			logger.warning('Can not get network connections stats ({})'.format(e))
80			logger.info('Disable connections stats')
81			stats['net_connections_enabled'] = False
82			self.stats = stats
83			return self.stats
84
85			for s in self.status_list:
86			stats[s] = len([c for c in net_connections if c.status == s])
87			initiated = 0
88			for s in self.initiated_states:
89			stats[s] = len([c for c in net_connections if c.status == s])
90			initiated += stats[s]
91			stats['initiated'] = initiated
92			terminated = 0
93			for s in self.initiated_states:
94			stats[s] = len([c for c in net_connections if c.status == s])
95			terminated += stats[s]
96			stats['terminated'] = terminated
97
98			if stats['nf_conntrack_enabled']:
99			# Grab connections track directly from the /proc file
100			for i in self.conntrack:
101			try:
102			with open(self.conntrack[i], 'r') as f:
103			stats[i] = float(f.readline().rstrip("\n"))
104			except (IOError, FileNotFoundError) as e:
105			logger.warning('Can not get network connections track ({})'.format(e))
106			logger.info('Disable connections track')
107			stats['nf_conntrack_enabled'] = False
108			self.stats = stats
109			return self.stats
110			if 'nf_conntrack_max' in stats and 'nf_conntrack_count' in stats:
111			stats['nf_conntrack_percent'] = stats['nf_conntrack_count'] * 100 / stats['nf_conntrack_max']
112			else:
113			stats['nf_conntrack_enabled'] = False
114			self.stats = stats
115			return self.stats
116
117			elif self.input_method == 'snmp':
118			# Update stats using SNMP
119			pass
120
121			# Update the stats
122			self.stats = stats
123			return self.stats
124
125			def update_views(self):
126			"""Update stats views."""
127			# Call the father's method
128			super(Plugin, self).update_views()
129
130			# Add specific information
131			try:
132			# Alert and log
133			if self.stats['nf_conntrack_enabled']:
134			self.views['nf_conntrack_percent']['decoration'] = self.get_alert(header='nf_conntrack_percent')
135			except KeyError:
136			# try/except mandatory for Windows compatibility (no conntrack stats)
137			pass
138
139			def msg_curse(self, args=None, max_width=None):
140			"""Return the dict to display in the curse interface."""
141			# Init the return message
142			ret = []
143
144			# Only process if stats exist and display plugin enable...
145			if not self.stats or self.is_disabled():
146			return ret
147
148			# Header
149			if self.stats['net_connections_enabled'] or self.stats['nf_conntrack_enabled']:
150			msg = '{}'.format('TCP CONNECTIONS')
151			ret.append(self.curse_add_line(msg, "TITLE"))
152			# Connections status
153			if self.stats['net_connections_enabled']:
154			for s in [psutil.CONN_LISTEN, 'initiated', psutil.CONN_ESTABLISHED, 'terminated']:
155			ret.append(self.curse_new_line())
156			msg = '{:{width}}'.format(nativestr(s).capitalize(), width=len(s))
157			ret.append(self.curse_add_line(msg))
158			msg = '{:>{width}}'.format(self.stats[s], width=max_width - len(s) + 2)
159			ret.append(self.curse_add_line(msg))
160			# Connections track
161			if self.stats['nf_conntrack_enabled']:
162			s = 'Tracked'
163			ret.append(self.curse_new_line())
164			msg = '{:{width}}'.format(nativestr(s).capitalize(), width=len(s))
165			ret.append(self.curse_add_line(msg))
166			msg = '{:>{width}}'.format(
167			'{:0.0f}/{:0.0f}'.format(self.stats['nf_conntrack_count'], self.stats['nf_conntrack_max']),
168			width=max_width - len(s) + 2,
169			)
170			ret.append(self.curse_add_line(msg, self.get_views(key='nf_conntrack_percent', option='decoration')))
171
172			return ret
173

nicolargo / glances

Push — master ( 05aaee...10b5c2 )

Plugin.update() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like