glances.servers_list.GlancesServersList.__update_stats_rest() - Code Metrics - Inspection of "version 4.3.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( ce0fc3...e09530 )

by Nicolas

created 2024-12-27 14:16 UTC

GlancesServersList.__update_stats_rest() D

↳ Parent: glances.servers_list

Complexity

Conditions

Size

Total Lines	35
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	13
eloc	27
nop	3
dl	0
loc	35
rs	4.2
c	0
b	0
f	0

How to fix Complexity

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

"""Manage the servers list used in TUI and WEBUI Central Browser mode"""

import threading

from defusedxml import xmlrpc

from glances import __apiversion__
from glances.client import GlancesClientTransport
from glances.globals import json_loads
from glances.logger import logger
from glances.password_list import GlancesPasswordList as GlancesPassword
from glances.servers_list_dynamic import GlancesAutoDiscoverServer
from glances.servers_list_static import GlancesStaticServer

try:
    import requests
except ImportError as e:
    import_requests_error_tag = True
    # Display debug message if import error
    logger.warning(f"Missing Python Lib ({e}), Client browser will not grab stats from Glances REST server")
else:
    import_requests_error_tag = False

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

DEFAULT_COLUMNS = "system:hr_name,load:min5,cpu:total,mem:percent"


class GlancesServersList:
    _section = "serverlist"

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

        # Init the servers and passwords list defined in the Glances configuration file
        self.static_server = None
        self.password = None
        self.load()

        # Init the dynamic servers list by starting a Zeroconf listener
        self.autodiscover_server = None
        if self.args.browser or not self.args.disable_autodiscover:
            self.autodiscover_server = GlancesAutoDiscoverServer()

        # Stats are updated in thread
        # Create a dict of threads
        self.threads_list = {}

    def load(self):
        """Load server and password list from the configuration file."""
        # Init the static server list
        self.static_server = GlancesStaticServer(config=self.config)
        # Init the password list (if defined)
        self.password = GlancesPassword(config=self.config)
        # Load columns to grab/display
        self._columns = self.load_columns()

    def load_columns(self):
        """Load columns definition from the configuration file.
        Read:   'system:hr_name,load:min5,cpu:total,mem:percent,sensors:value:Ambient'
        Return: [{'plugin': 'system', 'field': 'hr_name'},
                 {'plugin': 'load', 'field': 'min5'},
                 {'plugin': 'cpu', 'field': 'total'},
                 {'plugin': 'mem', 'field': 'percent'},
                 {'plugin': 'sensors', 'field': 'value', 'key': 'Ambient'}]
        """
        if self.config is None:
            logger.debug("No configuration file available. Cannot load columns definition.")
        elif not self.config.has_section(self._section):
            logger.warning(f"No [{self._section}] section in the configuration file. Cannot load columns definition.")

        columns_def = (
            self.config.get_value(self._section, 'columns')
            if self.config.get_value(self._section, 'columns')
            else DEFAULT_COLUMNS
        )

        return [dict(zip(['plugin', 'field', 'key'], i.split(':'))) for i in columns_def.split(',')]

    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_columns(self):
        """Return the columns definitions"""
        return self._columns

    def update_servers_stats(self):
        """For each server in the servers list, update the stats"""
        for v in self.get_servers_list():
            key = v["key"]
            thread = self.threads_list.get(key, None)
            if thread is None or thread.is_alive() is False:
                thread = threading.Thread(target=self.__update_stats, args=[v])
                self.threads_list[key] = thread
                thread.start()

    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)
            uri = 'http://{}:{}@{}:{}'.format(server['username'], server['password'], server['ip'], server['port'])
        else:
            uri = 'http://{}:{}'.format(server['ip'], server['port'])
        return uri

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

    def __update_stats(self, server):
        """Update stats for the given server"""
        server['uri'] = self.get_uri(server)
        server['columns'] = [self.__get_key(column) for column in self.get_columns()]
        if server['protocol'].lower() == 'rpc':
            self.__update_stats_rpc(server['uri'], server)
        elif server['protocol'].lower() == 'rest' and not import_requests_error_tag:
            self.__update_stats_rest(f"{server['uri']}/api/{__apiversion__}", server)

        return server

    def __update_stats_rpc(self, 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.get_columns():
            server_key = self.__get_key(column)
            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 __update_stats_rest(self, uri, server):
        try:
            requests.get(f'{uri}/status', timeout=3)
        except requests.exceptions.RequestException as e:
            logger.debug(f"Error while grabbing stats form server ({e})")
            server['status'] = 'OFFLINE'
            return server
        else:
            server['status'] = 'ONLINE'

        for column in self.get_columns():
            server_key = self.__get_key(column)
            # Build the URI (URL encoded)
            request_uri = f"{column['plugin']}/{column['field']}"
            if 'key' in column:
                request_uri += f"/{column['key']}"
            request_uri = f'{uri}/' + requests.utils.quote(request_uri)
            # Value
            try:
                r = requests.get(request_uri, timeout=3)
            except requests.exceptions.RequestException as e:
                logger.debug(f"Error while grabbing stats form server ({e})")
            else:
                if r.json() and column['field'] in r.json():
                    server[server_key] = r.json()[column['field']]
            # Decoration
            try:
                d = requests.get(request_uri + '/views', timeout=3)
            except requests.exceptions.RequestException as e:
                logger.debug(f"Error while grabbing stats view form server ({e})")
            else:
                if d.json() and 'decoration' in d.json():
                    server[server_key + '_decoration'] = d.json()['decoration']

        return server

    def __get_key(self, column):
        server_key = column.get('plugin') + '_' + column.get('field')
        if 'key' in column:
            server_key += '_' + column.get('key')
        return server_key


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			"""Manage the servers list used in TUI and WEBUI Central Browser mode"""
10
11			import threading
12
13			from defusedxml import xmlrpc
14
15			from glances import __apiversion__
16			from glances.client import GlancesClientTransport
17			from glances.globals import json_loads
18			from glances.logger import logger
19			from glances.password_list import GlancesPasswordList as GlancesPassword
20			from glances.servers_list_dynamic import GlancesAutoDiscoverServer
21			from glances.servers_list_static import GlancesStaticServer
22
23			try:
24			import requests
25			except ImportError as e:
26			import_requests_error_tag = True
27			# Display debug message if import error
28			logger.warning(f"Missing Python Lib ({e}), Client browser will not grab stats from Glances REST server")
29			else:
30			import_requests_error_tag = False
31
32			# Correct issue #1025 by monkey path the xmlrpc lib
33			xmlrpc.monkey_patch()
34
35			DEFAULT_COLUMNS = "system:hr_name,load:min5,cpu:total,mem:percent"
36
37
38			class GlancesServersList:
39			_section = "serverlist"
40
41			def __init__(self, config=None, args=None):
42			# Store the arg/config
43			self.args = args
44			self.config = config
45
46			# Init the servers and passwords list defined in the Glances configuration file
47			self.static_server = None
48			self.password = None
49			self.load()
50
51			# Init the dynamic servers list by starting a Zeroconf listener
52			self.autodiscover_server = None
53			if self.args.browser or not self.args.disable_autodiscover:
54			self.autodiscover_server = GlancesAutoDiscoverServer()
55
56			# Stats are updated in thread
57			# Create a dict of threads
58			self.threads_list = {}
59
60			def load(self):
61			"""Load server and password list from the configuration file."""
62			# Init the static server list
63			self.static_server = GlancesStaticServer(config=self.config)
64			# Init the password list (if defined)
65			self.password = GlancesPassword(config=self.config)
66			# Load columns to grab/display
67			self._columns = self.load_columns()
68
69			def load_columns(self):
70			"""Load columns definition from the configuration file.
71			Read: 'system:hr_name,load:min5,cpu:total,mem:percent,sensors:value:Ambient'
72			Return: [{'plugin': 'system', 'field': 'hr_name'},
73			{'plugin': 'load', 'field': 'min5'},
74			{'plugin': 'cpu', 'field': 'total'},
75			{'plugin': 'mem', 'field': 'percent'},
76			{'plugin': 'sensors', 'field': 'value', 'key': 'Ambient'}]
77			"""
78			if self.config is None:
79			logger.debug("No configuration file available. Cannot load columns definition.")
80			elif not self.config.has_section(self._section):
81			logger.warning(f"No [{self._section}] section in the configuration file. Cannot load columns definition.")
82
83			columns_def = (
84			self.config.get_value(self._section, 'columns')
85			if self.config.get_value(self._section, 'columns')
86			else DEFAULT_COLUMNS
87			)
88
89			return [dict(zip(['plugin', 'field', 'key'], i.split(':'))) for i in columns_def.split(',')]
90
91			def get_servers_list(self):
92			"""Return the current server list (list of dict).
93
94			Merge of static + autodiscover servers list.
95			"""
96			ret = []
97
98			if self.args.browser:
99			ret = self.static_server.get_servers_list()
100			if self.autodiscover_server is not None:
101			ret = self.static_server.get_servers_list() + self.autodiscover_server.get_servers_list()
102
103			return ret
104
105			def get_columns(self):
106			"""Return the columns definitions"""
107			return self._columns
108
109			def update_servers_stats(self):
110			"""For each server in the servers list, update the stats"""
111			for v in self.get_servers_list():
112			key = v["key"]
113			thread = self.threads_list.get(key, None)
114			if thread is None or thread.is_alive() is False:
115			thread = threading.Thread(target=self.__update_stats, args=[v])
116			self.threads_list[key] = thread
117			thread.start()
118
119			def get_uri(self, server):
120			"""Return the URI for the given server dict."""
121			# Select the connection mode (with or without password)
122			if server['password'] != "":
123			if server['status'] == 'PROTECTED':
124			# Try with the preconfigure password (only if status is PROTECTED)
125			clear_password = self.password.get_password(server['name'])
126			if clear_password is not None:
127			server['password'] = self.password.get_hash(clear_password)
128			uri = 'http://{}:{}@{}:{}'.format(server['username'], server['password'], server['ip'], server['port'])
129			else:
130			uri = 'http://{}:{}'.format(server['ip'], server['port'])
131			return uri
132
133			def set_in_selected(self, selected, key, value):
134			"""Set the (key, value) for the selected server in the list."""
135			# Static list then dynamic one
136			if selected >= len(self.static_server.get_servers_list()):
137			self.autodiscover_server.set_server(selected - len(self.static_server.get_servers_list()), key, value)
138			else:
139			self.static_server.set_server(selected, key, value)
140
141			def __update_stats(self, server):
142			"""Update stats for the given server"""
143			server['uri'] = self.get_uri(server)
144			server['columns'] = [self.__get_key(column) for column in self.get_columns()]
145			if server['protocol'].lower() == 'rpc':
146			self.__update_stats_rpc(server['uri'], server)
147			elif server['protocol'].lower() == 'rest' and not import_requests_error_tag:
148			self.__update_stats_rest(f"{server['uri']}/api/{__apiversion__}", server)
149
150			return server
151
152			def __update_stats_rpc(self, uri, server):
153			# Try to connect to the server
154			t = GlancesClientTransport()
155			t.set_timeout(3)
156
157			# Get common stats from Glances server
158			try:
159			s = xmlrpc.xmlrpc_client.ServerProxy(uri, transport=t)
160			except Exception as e:
161			logger.warning(f"Client browser couldn't create socket ({e})")
162			return server
163
164			# Get the stats
165			for column in self.get_columns():
166			server_key = self.__get_key(column)
167			try:
168			# Value
169			v_json = json_loads(s.getPlugin(column['plugin']))
170			if 'key' in column:
171			v_json = [i for i in v_json if i[i['key']].lower() == column['key'].lower()][0]
172			server[server_key] = v_json[column['field']]
173			# Decoration
174			d_json = json_loads(s.getPluginView(column['plugin']))
175			if 'key' in column:
176			d_json = d_json.get(column['key'])
177			server[server_key + '_decoration'] = d_json[column['field']]['decoration']
178			except (KeyError, IndexError, xmlrpc.xmlrpc_client.Fault) as e:
179			logger.debug(f"Error while grabbing stats form server ({e})")
180			except OSError as e:
181			logger.debug(f"Error while grabbing stats form server ({e})")
182			server['status'] = 'OFFLINE'
183			except xmlrpc.xmlrpc_client.ProtocolError as e:
184			if e.errcode == 401:
185			# Error 401 (Authentication failed)
186			# Password is not the good one...
187			server['password'] = None
188			server['status'] = 'PROTECTED'
189			else:
190			server['status'] = 'OFFLINE'
191			logger.debug(f"Cannot grab stats from server ({e.errcode} {e.errmsg})")
192			else:
193			# Status
194			server['status'] = 'ONLINE'
195
196			return server
197
198			def __update_stats_rest(self, uri, server):
199			try:
200			requests.get(f'{uri}/status', timeout=3)
201			except requests.exceptions.RequestException as e:
202			logger.debug(f"Error while grabbing stats form server ({e})")
203			server['status'] = 'OFFLINE'
204			return server
205			else:
206			server['status'] = 'ONLINE'
207
208			for column in self.get_columns():
209			server_key = self.__get_key(column)
210			# Build the URI (URL encoded)
211			request_uri = f"{column['plugin']}/{column['field']}"
212			if 'key' in column:
213			request_uri += f"/{column['key']}"
214			request_uri = f'{uri}/' + requests.utils.quote(request_uri)
215			# Value
216			try:
217			r = requests.get(request_uri, timeout=3)
218			except requests.exceptions.RequestException as e:
219			logger.debug(f"Error while grabbing stats form server ({e})")
220			else:
221			if r.json() and column['field'] in r.json():
222			server[server_key] = r.json()[column['field']]
223			# Decoration
224			try:
225			d = requests.get(request_uri + '/views', timeout=3)
226			except requests.exceptions.RequestException as e:
227			logger.debug(f"Error while grabbing stats view form server ({e})")
228			else:
229			if d.json() and 'decoration' in d.json():
230			server[server_key + '_decoration'] = d.json()['decoration']
231
232			return server
233
234			def __get_key(self, column):
235			server_key = column.get('plugin') + '_' + column.get('field')
236			if 'key' in column:
237			server_key += '_' + column.get('key')
238			return server_key
239

nicolargo / glances

Push — master ( ce0fc3...e09530 )

GlancesServersList.__update_stats_rest() D

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like