glances.plugins.connections.PluginModel.update_with_nf_conntrack_method() - Code Metrics - Inspection of "plugins: connections: refactor" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — develop (#3161)

unknown

created 2025-04-26 17:03 UTC

PluginModel.update_with_nf_conntrack_method() B

↳ Parent: glances.plugins.connections

Complexity

Conditions

Size

Total Lines	20
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	17
nop	2
dl	0
loc	20
rs	8.6166
c	0
b	0
f	0

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

"""Connections plugin."""

import psutil

from glances.globals import nativestr
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: is it a rate ? If yes, // by time_since_update when displayed,
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'LISTEN': {
        'description': 'Number of TCP connections in LISTEN state',
        'unit': 'number',
    },
    'ESTABLISHED': {
        'description': 'Number of TCP connections in ESTABLISHED state',
        'unit': 'number',
    },
    'SYN_SENT': {
        'description': 'Number of TCP connections in SYN_SENT state',
        'unit': 'number',
    },
    'SYN_RECV': {
        'description': 'Number of TCP connections in SYN_RECV state',
        'unit': 'number',
    },
    'initiated': {
        'description': 'Number of TCP connections initiated',
        'unit': 'number',
    },
    'terminated': {
        'description': 'Number of TCP connections terminated',
        'unit': 'number',
    },
    'nf_conntrack_count': {
        'description': 'Number of tracked connections',
        'unit': 'number',
    },
    'nf_conntrack_max': {
        'description': 'Maximum number of tracked connections',
        'unit': 'number',
    },
    'nf_conntrack_percent': {
        'description': 'Percentage of tracked connections',
        'unit': 'percent',
    },
}

# 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 PluginModel(GlancesPluginModel):
    """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().__init__(
            args=args,
            config=config,
            # items_history_list=items_history_list,
            stats_init_value={'net_connections_enabled': True, 'nf_conntrack_enabled': True},
            fields_description=fields_description,
        )

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

    def update_with_net_connections_method(self, stats):
        # Grab network interface stat using the psutil net_connections method
        try:
            net_connections = psutil.net_connections(kind="tcp")
        except Exception as e:
            logger.warning(f'Can not get network connections stats ({e})')
            logger.info('Disable connections stats')
            stats['net_connections_enabled'] = False
            self.stats = stats
            return 'previous-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

        return stats

    def update_with_nf_conntrack_method(self, stats):
        # Grab connections track directly from the /proc file
        for i in self.conntrack:
            try:
                with open(self.conntrack[i]) as f:
                    stats[i] = float(f.readline().rstrip("\n"))
            except (OSError, FileNotFoundError) as e:
                logger.warning(f'Can not get network connections track ({e})')
                logger.info('Disable connections track')
                stats['nf_conntrack_enabled'] = False
                self.stats = stats
                return 'previous-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 'previous-stats'

        return stats

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

        Stats is a dict
        """
        init = self.get_init_value()

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

            if init['net_connections_enabled']:
                stats = self.update_with_net_connections_method(init)

            if init['nf_conntrack_enabled']:
                stats = self.update_with_nf_conntrack_method(init)
        elif self.input_method == 'snmp':
            # Update stats using SNMP
            pass

        if stats == 'previous-stats':

            return self.stats

        # 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 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() or not max_width:
            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']:
                if s not in self.stats:
                    continue
                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']
            and 'nf_conntrack_count' in self.stats
            and 'nf_conntrack_max' in self.stats
        ):
            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			#
2			# This file is part of Glances.
3			#
4			# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
5			#
6			# SPDX-License-Identifier: LGPL-3.0-only
7			#
8
9			"""Connections plugin."""
10
11			import psutil
12
13			from glances.globals import nativestr
14			from glances.logger import logger
15			from glances.plugins.plugin.model import GlancesPluginModel
16
17			# Fields description
18			# description: human readable description
19			# short_name: shortname to use un UI
20			# unit: unit type
21			# rate: is it a rate ? If yes, // by time_since_update when displayed,
22			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
23			fields_description = {
24			'LISTEN': {
25			'description': 'Number of TCP connections in LISTEN state',
26			'unit': 'number',
27			},
28			'ESTABLISHED': {
29			'description': 'Number of TCP connections in ESTABLISHED state',
30			'unit': 'number',
31			},
32			'SYN_SENT': {
33			'description': 'Number of TCP connections in SYN_SENT state',
34			'unit': 'number',
35			},
36			'SYN_RECV': {
37			'description': 'Number of TCP connections in SYN_RECV state',
38			'unit': 'number',
39			},
40			'initiated': {
41			'description': 'Number of TCP connections initiated',
42			'unit': 'number',
43			},
44			'terminated': {
45			'description': 'Number of TCP connections terminated',
46			'unit': 'number',
47			},
48			'nf_conntrack_count': {
49			'description': 'Number of tracked connections',
50			'unit': 'number',
51			},
52			'nf_conntrack_max': {
53			'description': 'Maximum number of tracked connections',
54			'unit': 'number',
55			},
56			'nf_conntrack_percent': {
57			'description': 'Percentage of tracked connections',
58			'unit': 'percent',
59			},
60			}
61
62			# Define the history items list
63			# items_history_list = [{'name': 'rx',
64			# 'description': 'Download rate per second',
65			# 'y_unit': 'bit/s'},
66			# {'name': 'tx',
67			# 'description': 'Upload rate per second',
68			# 'y_unit': 'bit/s'}]
69
70
71			class PluginModel(GlancesPluginModel):
72			"""Glances connections plugin.
73
74			stats is a dict
75			"""
76
77			status_list = [psutil.CONN_LISTEN, psutil.CONN_ESTABLISHED]
78			initiated_states = [psutil.CONN_SYN_SENT, psutil.CONN_SYN_RECV]
79			terminated_states = [
80			psutil.CONN_FIN_WAIT1,
81			psutil.CONN_FIN_WAIT2,
82			psutil.CONN_TIME_WAIT,
83			psutil.CONN_CLOSE,
84			psutil.CONN_CLOSE_WAIT,
85			psutil.CONN_LAST_ACK,
86			]
87			conntrack = {
88			'nf_conntrack_count': '/proc/sys/net/netfilter/nf_conntrack_count',
89			'nf_conntrack_max': '/proc/sys/net/netfilter/nf_conntrack_max',
90			}
91
92			def __init__(self, args=None, config=None):
93			"""Init the plugin."""
94			super().__init__(
95			args=args,
96			config=config,
97			# items_history_list=items_history_list,
98			stats_init_value={'net_connections_enabled': True, 'nf_conntrack_enabled': True},
99			fields_description=fields_description,
100			)
101
102			# We want to display the stat in the curse interface
103			self.display_curse = True
104
105			def update_with_net_connections_method(self, stats):
106			# Grab network interface stat using the psutil net_connections method
107			try:
108			net_connections = psutil.net_connections(kind="tcp")
109			except Exception as e:
110			logger.warning(f'Can not get network connections stats ({e})')
111			logger.info('Disable connections stats')
112			stats['net_connections_enabled'] = False
113			self.stats = stats
114			return 'previous-stats'
115
116			for s in self.status_list:
117			stats[s] = len([c for c in net_connections if c.status == s])
118			initiated = 0
119			for s in self.initiated_states:
120			stats[s] = len([c for c in net_connections if c.status == s])
121			initiated += stats[s]
122			stats['initiated'] = initiated
123			terminated = 0
124			for s in self.initiated_states:
125			stats[s] = len([c for c in net_connections if c.status == s])
126			terminated += stats[s]
127			stats['terminated'] = terminated
128
129			return stats
130
131			def update_with_nf_conntrack_method(self, stats):
132			# Grab connections track directly from the /proc file
133			for i in self.conntrack:
134			try:
135			with open(self.conntrack[i]) as f:
136			stats[i] = float(f.readline().rstrip("\n"))
137			except (OSError, FileNotFoundError) as e:
138			logger.warning(f'Can not get network connections track ({e})')
139			logger.info('Disable connections track')
140			stats['nf_conntrack_enabled'] = False
141			self.stats = stats
142			return 'previous-stats'
143			if 'nf_conntrack_max' in stats and 'nf_conntrack_count' in stats:
144			stats['nf_conntrack_percent'] = stats['nf_conntrack_count'] * 100 / stats['nf_conntrack_max']
145			else:
146			stats['nf_conntrack_enabled'] = False
147			self.stats = stats
148			return 'previous-stats'
149
150			return stats
151
152			@GlancesPluginModel._check_decorator
153			@GlancesPluginModel._log_result_decorator
154			def update(self):
155			"""Update connections stats using the input method.
156
157			Stats is a dict
158			"""
159			init = self.get_init_value()
160
161			if self.input_method == 'local':
162			# Update stats using the PSUtils lib
163
164			if init['net_connections_enabled']:
165			stats = self.update_with_net_connections_method(init)
166
167			if init['nf_conntrack_enabled']:
168			stats = self.update_with_nf_conntrack_method(init)
169			elif self.input_method == 'snmp':
170			# Update stats using SNMP
171			pass
172
173			if stats == 'previous-stats':
			0 ignored issues – show introduced 2025-04-26 17:06 UTC by Report Bug Copy Issue Report The variable `stats` does not seem to be defined for all execution paths. Loading history...
174			return self.stats
175
176			# Update the stats
177			self.stats = stats
178			return self.stats
179
180			def update_views(self):
181			"""Update stats views."""
182			# Call the father's method
183			super().update_views()
184
185			# Add specific information
186			try:
187			# Alert and log
188			if self.stats['nf_conntrack_enabled']:
189			self.views['nf_conntrack_percent']['decoration'] = self.get_alert(header='nf_conntrack_percent')
190			except KeyError:
191			# try/except mandatory for Windows compatibility (no conntrack stats)
192			pass
193
194			def msg_curse(self, args=None, max_width=None):
195			"""Return the dict to display in the curse interface."""
196			# Init the return message
197			ret = []
198
199			# Only process if stats exist and display plugin enable...
200			if not self.stats or self.is_disabled() or not max_width:
201			return ret
202
203			# Header
204			if self.stats['net_connections_enabled'] or self.stats['nf_conntrack_enabled']:
205			msg = '{}'.format('TCP CONNECTIONS')
206			ret.append(self.curse_add_line(msg, "TITLE"))
207			# Connections status
208			if self.stats['net_connections_enabled']:
209			for s in [psutil.CONN_LISTEN, 'initiated', psutil.CONN_ESTABLISHED, 'terminated']:
210			if s not in self.stats:
211			continue
212			ret.append(self.curse_new_line())
213			msg = '{:{width}}'.format(nativestr(s).capitalize(), width=len(s))
214			ret.append(self.curse_add_line(msg))
215			msg = '{:>{width}}'.format(self.stats[s], width=max_width - len(s) + 2)
216			ret.append(self.curse_add_line(msg))
217			# Connections track
218			if (
219			self.stats['nf_conntrack_enabled']
220			and 'nf_conntrack_count' in self.stats
221			and 'nf_conntrack_max' in self.stats
222			):
223			s = 'Tracked'
224			ret.append(self.curse_new_line())
225			msg = '{:{width}}'.format(nativestr(s).capitalize(), width=len(s))
226			ret.append(self.curse_add_line(msg))
227			msg = '{:>{width}}'.format(
228			'{:0.0f}/{:0.0f}'.format(self.stats['nf_conntrack_count'], self.stats['nf_conntrack_max']),
229			width=max_width - len(s) + 2,
230			)
231			ret.append(self.curse_add_line(msg, self.get_views(key='nf_conntrack_percent', option='decoration')))
232
233			return ret
234

nicolargo / glances

Pull Request — develop (#3161)

PluginModel.update_with_nf_conntrack_method() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like