glances.client_browser.GlancesClientBrowser.serve_forever() - Code Metrics - Inspection of "Merge branch 'issue1289' into develop" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 6e485a...10cc1d )

by Nicolas

created 2024-10-06 09:20 UTC

GlancesClientBrowser.serve_forever() A

↳ Parent: glances.client_browser

Complexity

Conditions

Size

Total Lines	10
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	1
dl	0
loc	10
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
#

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

import threading

from glances.autodiscover import GlancesAutoDiscoverServer
from glances.client import GlancesClient, GlancesClientTransport
from glances.globals import Fault, ProtocolError, ServerProxy, 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


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

nicolargo / glances

Push — develop ( 6e485a...10cc1d )

GlancesClientBrowser.serve_forever() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like