glances.client_browser.GlancesClientBrowser.__update_stats() - Code Metrics - Inspection of "version 4.2.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( cc9054...a8608f )

by Nicolas

created 2024-10-20 16:02 UTC

GlancesClientBrowser.__update_stats() C

↳ Parent: glances.client_browser

Complexity

Conditions

Size

Total Lines	51
Code Lines	35

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	11
eloc	35
nop	2
dl	0
loc	51
rs	5.4
c	0
b	0
f	0

How to fix Long Method Complexity

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

If many parameters/temporary variables are present:
- Replace temporary variables with Query
- Introduce parameter object; often combined with preserve whole object
- If the above two are insufficient: Replace method with method object
If you have long conditionals: Decompose Conditional
Otherwise: Extract method

Complexity

Complex classes like glances.client_browser.GlancesClientBrowser.__update_stats() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

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

"""Manage the Glances client browser (list of Glances server)."""

import threading

from defusedxml import xmlrpc

from glances.autodiscover import GlancesAutoDiscoverServer
from glances.client import GlancesClient, GlancesClientTransport
from glances.globals import json_loads
from glances.logger import LOG_FILENAME, logger
from glances.outputs.glances_curses_browser import GlancesCursesBrowser
from glances.password_list import GlancesPasswordList as GlancesPassword
from glances.static_list import GlancesStaticServer

# Correct issue #1025 by monkey path the xmlrpc lib
xmlrpc.monkey_patch()


class GlancesClientBrowser:
    """This class creates and manages the TCP client browser (servers list)."""

    def __init__(self, config=None, args=None):
        # Store the arg/config
        self.args = args
        self.config = config
        self.static_server = None
        self.password = None

        # Load the configuration file
        self.load()

        # Start the autodiscover mode (Zeroconf listener)
        if not self.args.disable_autodiscover:
            self.autodiscover_server = GlancesAutoDiscoverServer()
        else:
            self.autodiscover_server = None

        # Init screen
        self.screen = GlancesCursesBrowser(args=self.args)

    def load(self):
        """Load server and password list from the configuration file."""
        # Init the static server list (if defined)
        self.static_server = GlancesStaticServer(config=self.config)

        # Init the password list (if defined)
        self.password = GlancesPassword(config=self.config)

    def get_servers_list(self):
        """Return the current server list (list of dict).

        Merge of static + autodiscover servers list.
        """
        ret = []

        if self.args.browser:
            ret = self.static_server.get_servers_list()
            if self.autodiscover_server is not None:
                ret = self.static_server.get_servers_list() + self.autodiscover_server.get_servers_list()

        return ret

    def __get_uri(self, server):
        """Return the URI for the given server dict."""
        # Select the connection mode (with or without password)
        if server['password'] != "":
            if server['status'] == 'PROTECTED':
                # Try with the preconfigure password (only if status is PROTECTED)
                clear_password = self.password.get_password(server['name'])
                if clear_password is not None:
                    server['password'] = self.password.get_hash(clear_password)
            return 'http://{}:{}@{}:{}'.format(server['username'], server['password'], server['ip'], server['port'])
        return 'http://{}:{}'.format(server['ip'], server['port'])

    def __update_stats(self, server):
        """Update stats for the given server (picked from the server list)"""
        # Get the server URI
        uri = self.__get_uri(server)

        # Try to connect to the server
        t = GlancesClientTransport()
        t.set_timeout(3)

        # Get common stats from Glances server
        try:
            s = xmlrpc.xmlrpc_client.ServerProxy(uri, transport=t)
        except Exception as e:
            logger.warning(f"Client browser couldn't create socket ({e})")
            return server

        # Get the stats
        for column in self.static_server.get_columns():
            server_key = column.get('plugin') + '_' + column.get('field')
            if 'key' in column:
                server_key += '_' + column.get('key')
            try:
                # Value
                v_json = json_loads(s.getPlugin(column['plugin']))
                if 'key' in column:
                    v_json = [i for i in v_json if i[i['key']].lower() == column['key'].lower()][0]
                server[server_key] = v_json[column['field']]
                # Decoration
                d_json = json_loads(s.getPluginView(column['plugin']))
                if 'key' in column:
                    d_json = d_json.get(column['key'])
                server[server_key + '_decoration'] = d_json[column['field']]['decoration']
            except (KeyError, IndexError, xmlrpc.xmlrpc_client.Fault) as e:
                logger.debug(f"Error while grabbing stats form server ({e})")
            except OSError as e:
                logger.debug(f"Error while grabbing stats form server ({e})")
                server['status'] = 'OFFLINE'
            except xmlrpc.xmlrpc_client.ProtocolError as e:
                if e.errcode == 401:
                    # Error 401 (Authentication failed)
                    # Password is not the good one...
                    server['password'] = None
                    server['status'] = 'PROTECTED'
                else:
                    server['status'] = 'OFFLINE'
                logger.debug(f"Cannot grab stats from server ({e.errcode} {e.errmsg})")
            else:
                # Status
                server['status'] = 'ONLINE'

        return server

    def __display_server(self, server):
        """Connect and display the given server"""
        # Display the Glances client for the selected server
        logger.debug(f"Selected server {server}")

        # Connection can take time
        # Display a popup
        self.screen.display_popup('Connect to {}:{}'.format(server['name'], server['port']), duration=1)

        # A password is needed to access to the server's stats
        if server['password'] is None:
            # First of all, check if a password is available in the [passwords] section
            clear_password = self.password.get_password(server['name'])
            if clear_password is None or self.get_servers_list()[self.screen.active_server]['status'] == 'PROTECTED':
                # Else, the password should be enter by the user
                # Display a popup to enter password
                clear_password = self.screen.display_popup(
                    'Password needed for {}: '.format(server['name']), popup_type='input', is_password=True
                )
            # Store the password for the selected server
            if clear_password is not None:
                self.set_in_selected('password', self.password.get_hash(clear_password))

        # Display the Glance client on the selected server
        logger.info("Connect Glances client to the {} server".format(server['key']))

        # Init the client
        args_server = self.args

        # Overwrite connection setting
        args_server.client = server['ip']
        args_server.port = server['port']
        args_server.username = server['username']
        args_server.password = server['password']
        client = GlancesClient(config=self.config, args=args_server, return_to_browser=True)

        # Test if client and server are in the same major version
        if not client.login():
            self.screen.display_popup(
                "Sorry, cannot connect to '{}'\n" "See '{}' for more details".format(server['name'], LOG_FILENAME)
            )

            # Set the ONLINE status for the selected server
            self.set_in_selected('status', 'OFFLINE')
        else:
            # Start the client loop
            # Return connection type: 'glances' or 'snmp'
            connection_type = client.serve_forever()

            try:
                logger.debug("Disconnect Glances client from the {} server".format(server['key']))
            except IndexError:
                # Server did not exist anymore
                pass
            else:
                # Set the ONLINE status for the selected server
                if connection_type == 'snmp':
                    self.set_in_selected('status', 'SNMP')
                else:
                    self.set_in_selected('status', 'ONLINE')

        # Return to the browser (no server selected)
        self.screen.active_server = None

    def __serve_forever(self):
        """Main client loop."""
        # No need to update the server list
        # It's done by the GlancesAutoDiscoverListener class (autodiscover.py)
        # Or define statically in the configuration file (module static_list.py)
        # For each server in the list, grab elementary stats (CPU, LOAD, MEM, OS...)
        thread_list = {}
        while not self.screen.is_end:
            logger.debug(f"Iter through the following server list: {self.get_servers_list()}")
            for v in self.get_servers_list():
                key = v["key"]
                thread = thread_list.get(key, None)
                if thread is None or thread.is_alive() is False:
                    thread = threading.Thread(target=self.__update_stats, args=[v])
                    thread_list[key] = thread
                    thread.start()

            # Update the screen (list or Glances client)
            if self.screen.active_server is None:
                #  Display the Glances browser
                self.screen.update(self.get_servers_list())
            else:
                # Display the active server
                self.__display_server(self.get_servers_list()[self.screen.active_server])

        # exit key pressed
        for thread in thread_list.values():
            thread.join()

    def serve_forever(self):
        """Wrapper to the serve_forever function.

        This function will restore the terminal to a sane state
        before re-raising the exception and generating a traceback.
        """
        try:
            return self.__serve_forever()
        finally:
            self.end()

    def set_in_selected(self, key, value):
        """Set the (key, value) for the selected server in the list."""
        # Static list then dynamic one
        if self.screen.active_server >= len(self.static_server.get_servers_list()):
            self.autodiscover_server.set_server(
                self.screen.active_server - len(self.static_server.get_servers_list()), key, value
            )
        else:
            self.static_server.set_server(self.screen.active_server, key, value)

    def end(self):
        """End of the client browser session."""
        self.screen.end()


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			"""Manage the Glances client browser (list of Glances server)."""
10
11			import threading
12
13			from defusedxml import xmlrpc
14
15			from glances.autodiscover import GlancesAutoDiscoverServer
16			from glances.client import GlancesClient, GlancesClientTransport
17			from glances.globals import json_loads
18			from glances.logger import LOG_FILENAME, logger
19			from glances.outputs.glances_curses_browser import GlancesCursesBrowser
20			from glances.password_list import GlancesPasswordList as GlancesPassword
21			from glances.static_list import GlancesStaticServer
22
23			# Correct issue #1025 by monkey path the xmlrpc lib
24			xmlrpc.monkey_patch()
25
26
27			class GlancesClientBrowser:
28			"""This class creates and manages the TCP client browser (servers list)."""
29
30			def __init__(self, config=None, args=None):
31			# Store the arg/config
32			self.args = args
33			self.config = config
34			self.static_server = None
35			self.password = None
36
37			# Load the configuration file
38			self.load()
39
40			# Start the autodiscover mode (Zeroconf listener)
41			if not self.args.disable_autodiscover:
42			self.autodiscover_server = GlancesAutoDiscoverServer()
43			else:
44			self.autodiscover_server = None
45
46			# Init screen
47			self.screen = GlancesCursesBrowser(args=self.args)
48
49			def load(self):
50			"""Load server and password list from the configuration file."""
51			# Init the static server list (if defined)
52			self.static_server = GlancesStaticServer(config=self.config)
53
54			# Init the password list (if defined)
55			self.password = GlancesPassword(config=self.config)
56
57			def get_servers_list(self):
58			"""Return the current server list (list of dict).
59
60			Merge of static + autodiscover servers list.
61			"""
62			ret = []
63
64			if self.args.browser:
65			ret = self.static_server.get_servers_list()
66			if self.autodiscover_server is not None:
67			ret = self.static_server.get_servers_list() + self.autodiscover_server.get_servers_list()
68
69			return ret
70
71			def __get_uri(self, server):
72			"""Return the URI for the given server dict."""
73			# Select the connection mode (with or without password)
74			if server['password'] != "":
75			if server['status'] == 'PROTECTED':
76			# Try with the preconfigure password (only if status is PROTECTED)
77			clear_password = self.password.get_password(server['name'])
78			if clear_password is not None:
79			server['password'] = self.password.get_hash(clear_password)
80			return 'http://{}:{}@{}:{}'.format(server['username'], server['password'], server['ip'], server['port'])
81			return 'http://{}:{}'.format(server['ip'], server['port'])
82
83			def __update_stats(self, server):
84			"""Update stats for the given server (picked from the server list)"""
85			# Get the server URI
86			uri = self.__get_uri(server)
87
88			# Try to connect to the server
89			t = GlancesClientTransport()
90			t.set_timeout(3)
91
92			# Get common stats from Glances server
93			try:
94			s = xmlrpc.xmlrpc_client.ServerProxy(uri, transport=t)
95			except Exception as e:
96			logger.warning(f"Client browser couldn't create socket ({e})")
97			return server
98
99			# Get the stats
100			for column in self.static_server.get_columns():
101			server_key = column.get('plugin') + '_' + column.get('field')
102			if 'key' in column:
103			server_key += '_' + column.get('key')
104			try:
105			# Value
106			v_json = json_loads(s.getPlugin(column['plugin']))
107			if 'key' in column:
108			v_json = [i for i in v_json if i[i['key']].lower() == column['key'].lower()][0]
109			server[server_key] = v_json[column['field']]
110			# Decoration
111			d_json = json_loads(s.getPluginView(column['plugin']))
112			if 'key' in column:
113			d_json = d_json.get(column['key'])
114			server[server_key + '_decoration'] = d_json[column['field']]['decoration']
115			except (KeyError, IndexError, xmlrpc.xmlrpc_client.Fault) as e:
116			logger.debug(f"Error while grabbing stats form server ({e})")
117			except OSError as e:
118			logger.debug(f"Error while grabbing stats form server ({e})")
119			server['status'] = 'OFFLINE'
120			except xmlrpc.xmlrpc_client.ProtocolError as e:
121			if e.errcode == 401:
122			# Error 401 (Authentication failed)
123			# Password is not the good one...
124			server['password'] = None
125			server['status'] = 'PROTECTED'
126			else:
127			server['status'] = 'OFFLINE'
128			logger.debug(f"Cannot grab stats from server ({e.errcode} {e.errmsg})")
129			else:
130			# Status
131			server['status'] = 'ONLINE'
132
133			return server
134
135			def __display_server(self, server):
136			"""Connect and display the given server"""
137			# Display the Glances client for the selected server
138			logger.debug(f"Selected server {server}")
139
140			# Connection can take time
141			# Display a popup
142			self.screen.display_popup('Connect to {}:{}'.format(server['name'], server['port']), duration=1)
143
144			# A password is needed to access to the server's stats
145			if server['password'] is None:
146			# First of all, check if a password is available in the [passwords] section
147			clear_password = self.password.get_password(server['name'])
148			if clear_password is None or self.get_servers_list()[self.screen.active_server]['status'] == 'PROTECTED':
149			# Else, the password should be enter by the user
150			# Display a popup to enter password
151			clear_password = self.screen.display_popup(
152			'Password needed for {}: '.format(server['name']), popup_type='input', is_password=True
153			)
154			# Store the password for the selected server
155			if clear_password is not None:
156			self.set_in_selected('password', self.password.get_hash(clear_password))
157
158			# Display the Glance client on the selected server
159			logger.info("Connect Glances client to the {} server".format(server['key']))
160
161			# Init the client
162			args_server = self.args
163
164			# Overwrite connection setting
165			args_server.client = server['ip']
166			args_server.port = server['port']
167			args_server.username = server['username']
168			args_server.password = server['password']
169			client = GlancesClient(config=self.config, args=args_server, return_to_browser=True)
170
171			# Test if client and server are in the same major version
172			if not client.login():
173			self.screen.display_popup(
174			"Sorry, cannot connect to '{}'\n" "See '{}' for more details".format(server['name'], LOG_FILENAME)
175			)
176
177			# Set the ONLINE status for the selected server
178			self.set_in_selected('status', 'OFFLINE')
179			else:
180			# Start the client loop
181			# Return connection type: 'glances' or 'snmp'
182			connection_type = client.serve_forever()
183
184			try:
185			logger.debug("Disconnect Glances client from the {} server".format(server['key']))
186			except IndexError:
187			# Server did not exist anymore
188			pass
189			else:
190			# Set the ONLINE status for the selected server
191			if connection_type == 'snmp':
192			self.set_in_selected('status', 'SNMP')
193			else:
194			self.set_in_selected('status', 'ONLINE')
195
196			# Return to the browser (no server selected)
197			self.screen.active_server = None
198
199			def __serve_forever(self):
200			"""Main client loop."""
201			# No need to update the server list
202			# It's done by the GlancesAutoDiscoverListener class (autodiscover.py)
203			# Or define statically in the configuration file (module static_list.py)
204			# For each server in the list, grab elementary stats (CPU, LOAD, MEM, OS...)
205			thread_list = {}
206			while not self.screen.is_end:
207			logger.debug(f"Iter through the following server list: {self.get_servers_list()}")
208			for v in self.get_servers_list():
209			key = v["key"]
210			thread = thread_list.get(key, None)
211			if thread is None or thread.is_alive() is False:
212			thread = threading.Thread(target=self.__update_stats, args=[v])
213			thread_list[key] = thread
214			thread.start()
215
216			# Update the screen (list or Glances client)
217			if self.screen.active_server is None:
218			# Display the Glances browser
219			self.screen.update(self.get_servers_list())
220			else:
221			# Display the active server
222			self.__display_server(self.get_servers_list()[self.screen.active_server])
223
224			# exit key pressed
225			for thread in thread_list.values():
226			thread.join()
227
228			def serve_forever(self):
229			"""Wrapper to the serve_forever function.
230
231			This function will restore the terminal to a sane state
232			before re-raising the exception and generating a traceback.
233			"""
234			try:
235			return self.__serve_forever()
236			finally:
237			self.end()
238
239			def set_in_selected(self, key, value):
240			"""Set the (key, value) for the selected server in the list."""
241			# Static list then dynamic one
242			if self.screen.active_server >= len(self.static_server.get_servers_list()):
243			self.autodiscover_server.set_server(
244			self.screen.active_server - len(self.static_server.get_servers_list()), key, value
245			)
246			else:
247			self.static_server.set_server(self.screen.active_server, key, value)
248
249			def end(self):
250			"""End of the client browser session."""
251			self.screen.end()
252

nicolargo / glances

Push — master ( cc9054...a8608f )

GlancesClientBrowser.__update_stats() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like