glances.servers_list_dynamic.AutoDiscovered.set_server() - Code Metrics - Inspection of "version 4.4.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 73a01d...ef36eb )

by Nicolas

created 2025-11-02 16:07 UTC

AutoDiscovered.set_server() A

↳ Parent: glances.servers_list_dynamic

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	4
dl	0
loc	3
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 autodiscover Glances server (thk to the ZeroConf protocol)."""

import socket
import sys

from glances.globals import BSD
from glances.logger import logger

try:
    import netifaces

    netifaces_tag = True
except ImportError:
    logger.warning("Servers list - Netifaces2 lib not found, active IP address not possible")
    netifaces_tag = False

try:
    netifaces.gateways()
except Exception:
    netifaces_tag = True
else:
    logger.warning("Servers list - Netifaces2 do not support gateways() method, active IP address not possible")
    netifaces_tag = False


try:
    from zeroconf import ServiceBrowser, ServiceInfo, Zeroconf
    from zeroconf import __version__ as __zeroconf_version

    zeroconf_tag = True
except ImportError:
    zeroconf_tag = False

# Zeroconf 0.17 or higher is needed
if zeroconf_tag:
    zeroconf_min_version = (0, 17, 0)
    zeroconf_version = tuple([int(num) for num in __zeroconf_version.split('.')])
    logger.debug(f"Zeroconf version {__zeroconf_version} detected.")
    if zeroconf_version < zeroconf_min_version:
        logger.critical("Please install zeroconf 0.17 or higher.")
        sys.exit(1)

# Global var
# Recent versions of the zeroconf python package doesn't like a zeroconf type that ends with '._tcp.'.
# Correct issue: zeroconf problem with zeroconf_type = "_%s._tcp." % 'glances' #888
zeroconf_type = "_{}._tcp.local.".format('glances')


class AutoDiscovered:
    """Class to manage the auto discovered servers dict."""

    def __init__(self):
        # server_dict is a list of dict (JSON compliant)
        # [ {'key': 'zeroconf name', ip': '172.1.2.3', 'port': 61209, 'protocol': 'rpc', 'cpu': 3, 'mem': 34 ...} ... ]
        self._server_list = []

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

    def set_server(self, server_pos, key, value):
        """Set the key to the value for the server_pos (position in the list)."""
        self._server_list[server_pos][key] = value

    def add_server(self, name, ip, port, protocol='rpc'):
        """Add a new server to the list."""
        new_server = {
            'key': name,  # Zeroconf name with both hostname and port
            'name': name.split(':')[0],  # Short name
            'ip': ip,  # IP address seen by the client
            'port': port,  # TCP port
            'protocol': str(protocol),  # RPC or RESTFUL protocol
            'username': 'glances',  # Default username
            'password': '',  # Default password
            'status': 'UNKNOWN',  # Server status: 'UNKNOWN', 'OFFLINE', 'ONLINE', 'PROTECTED'
            'type': 'DYNAMIC',
        }  # Server type: 'STATIC' or 'DYNAMIC'
        self._server_list.append(new_server)
        logger.debug(f"Updated servers list ({len(self._server_list)} servers): {self._server_list}")

    def remove_server(self, name):
        """Remove a server from the dict."""
        for i in self._server_list:
            if i['key'] == name:
                try:
                    self._server_list.remove(i)
                    logger.debug(f"Remove server {name} from the list")
                    logger.debug(f"Updated servers list ({len(self._server_list)} servers): {self._server_list}")
                except ValueError:
                    logger.error(f"Cannot remove server {name} from the list")


class GlancesAutoDiscoverListener:
    """Zeroconf listener for Glances server."""

    def __init__(self):
        # Create an instance of the servers list
        self.servers = AutoDiscovered()

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

    def set_server(self, server_pos, key, value):
        """Set the key to the value for the server_pos (position in the list)."""
        self.servers.set_server(server_pos, key, value)

    def add_service(self, zeroconf, srv_type, srv_name):
        """Method called when a new Zeroconf client is detected.

        Note: the return code will never be used

        :return: True if the zeroconf client is a Glances server
        """
        if srv_type != zeroconf_type:
            return False
        logger.debug(f"Check new Zeroconf server: {srv_type} / {srv_name}")
        info = zeroconf.get_service_info(srv_type, srv_name)
        if info and (info.addresses or info.parsed_addresses):
            address = info.addresses[0] if info.addresses else info.parsed_addresses[0]
            new_server_ip = socket.inet_ntoa(address)
            new_server_port = info.port
            new_server_protocol = info.properties[b'protocol'].decode() if b'protocol' in info.properties else 'rpc'

            # Add server to the global dict
            self.servers.add_server(
                srv_name,
                new_server_ip,
                new_server_port,
                protocol=new_server_protocol,
            )
            logger.info(
                f"New {new_server_protocol} Glances server detected ({srv_name} from {new_server_ip}:{new_server_port})"
            )
        else:
            logger.warning("New Glances server detected, but failed to be get Zeroconf ServiceInfo ")
        return True

    def remove_service(self, zeroconf, srv_type, srv_name):
        """Remove the server from the list."""
        self.servers.remove_server(srv_name)
        logger.info(f"Glances server {srv_name} removed from the autodetect list")

    def update_service(self, zeroconf, srv_type, srv_name):
        """Update the server from the list.
        Done by a dirty hack (remove + add).
        """
        self.remove_service(zeroconf, srv_type, srv_name)
        self.add_service(zeroconf, srv_type, srv_name)
        logger.info(f"Glances server {srv_name} updated from the autodetect list")


class GlancesAutoDiscoverServer:
    """Implementation of the Zeroconf protocol (server side for the Glances client)."""

    def __init__(self, args=None):
        if zeroconf_tag:
            logger.info("Init autodiscover mode (Zeroconf protocol)")
            try:
                self.zeroconf = Zeroconf()
            except OSError as e:
                logger.error(f"Cannot start Zeroconf ({e})")
                self.zeroconf_enable_tag = False
            else:
                self.listener = GlancesAutoDiscoverListener()
                self.browser = ServiceBrowser(self.zeroconf, zeroconf_type, self.listener)
                self.zeroconf_enable_tag = True
        else:
            logger.error("Cannot start autodiscover mode (Zeroconf lib is not installed)")
            self.zeroconf_enable_tag = False

    def get_servers_list(self):
        """Return the current server list (dict of dict)."""
        if zeroconf_tag and self.zeroconf_enable_tag:
            return self.listener.get_servers_list()
        return []

    def set_server(self, server_pos, key, value):
        """Set the key to the value for the server_pos (position in the list)."""
        if zeroconf_tag and self.zeroconf_enable_tag:
            self.listener.set_server(server_pos, key, value)

    def close(self):
        if zeroconf_tag and self.zeroconf_enable_tag:
            self.zeroconf.close()


class GlancesAutoDiscoverClient:
    """Implementation of the zeroconf protocol (client side for the Glances server)."""

    def __init__(self, hostname, args=None):
        if zeroconf_tag:
            zeroconf_bind_address = args.bind_address
            try:
                self.zeroconf = Zeroconf()
            except OSError as e:
                logger.error(f"Cannot start zeroconf: {e}")

            # XXX *BSDs: Segmentation fault (core dumped)
            # -- https://bitbucket.org/al45tair/netifaces/issues/15
            if not BSD:
                try:
                    # -B @ overwrite the dynamic IPv4 choice
                    if netifaces_tag and zeroconf_bind_address == '0.0.0.0':
                        zeroconf_bind_address = self.find_active_ip_address()
                except Exception:
                    # Issue #528 (no network interface available)
                    # Issue #3219 (no implementation for gateway())
                    pass

            # Ensure zeroconf_bind_address is an IP address not an host
            zeroconf_bind_address = socket.gethostbyname(zeroconf_bind_address)

            # Check IP v4/v6
            address_family = socket.getaddrinfo(zeroconf_bind_address, args.port)[0][0]

            # Start the zeroconf service
            try:
                self.info = ServiceInfo(
                    zeroconf_type,
                    f'{hostname}:{args.port}.{zeroconf_type}',
                    address=socket.inet_pton(address_family, zeroconf_bind_address),
                    port=args.port,
                    weight=0,
                    priority=0,
                    properties={'protocol': 'rest' if args.webserver else 'rpc'},
                    server=hostname,
                )
            except TypeError:
                # Manage issue 1663 with breaking change on ServiceInfo method
                # address (only one address) is replaced by addresses (list of addresses)
                self.info = ServiceInfo(
                    zeroconf_type,
                    name=f'{hostname}:{args.port}.{zeroconf_type}',
                    addresses=[socket.inet_pton(address_family, zeroconf_bind_address)],
                    port=args.port,
                    weight=0,
                    priority=0,
                    properties={'protocol': 'rest' if args.webserver else 'rpc'},
                    server=hostname,
                )
            try:
                self.zeroconf.register_service(self.info)
            except Exception as e:
                logger.error(f"Error while announcing Glances server: {e}")
            else:
                print(f"Announce the Glances server on the LAN (using {zeroconf_bind_address} IP address)")
        else:
            logger.error("Cannot announce Glances server on the network: zeroconf library not found.")

    @staticmethod
    def find_active_ip_address():
        """Try to find the active IP addresses."""
        # Interface of the default gateway
        gateway_itf = netifaces.gateways()[netifaces.AF_INET][0][1]
        # IP address for the interface
        return netifaces.ifaddresses(gateway_itf)[netifaces.AF_INET][0]['addr']

    def close(self):
        if zeroconf_tag:
            self.zeroconf.unregister_service(self.info)
            self.zeroconf.close()


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 autodiscover Glances server (thk to the ZeroConf protocol)."""
10
11			import socket
12			import sys
13
14			from glances.globals import BSD
15			from glances.logger import logger
16
17			try:
18			import netifaces
19
20			netifaces_tag = True
21			except ImportError:
22			logger.warning("Servers list - Netifaces2 lib not found, active IP address not possible")
23			netifaces_tag = False
24
25			try:
26			netifaces.gateways()
27			except Exception:
28			netifaces_tag = True
29			else:
30			logger.warning("Servers list - Netifaces2 do not support gateways() method, active IP address not possible")
31			netifaces_tag = False
32
33
34			try:
35			from zeroconf import ServiceBrowser, ServiceInfo, Zeroconf
36			from zeroconf import __version__ as __zeroconf_version
37
38			zeroconf_tag = True
39			except ImportError:
40			zeroconf_tag = False
41
42			# Zeroconf 0.17 or higher is needed
43			if zeroconf_tag:
44			zeroconf_min_version = (0, 17, 0)
45			zeroconf_version = tuple([int(num) for num in __zeroconf_version.split('.')])
46			logger.debug(f"Zeroconf version {__zeroconf_version} detected.")
47			if zeroconf_version < zeroconf_min_version:
48			logger.critical("Please install zeroconf 0.17 or higher.")
49			sys.exit(1)
50
51			# Global var
52			# Recent versions of the zeroconf python package doesn't like a zeroconf type that ends with '._tcp.'.
53			# Correct issue: zeroconf problem with zeroconf_type = "_%s._tcp." % 'glances' #888
54			zeroconf_type = "_{}._tcp.local.".format('glances')
55
56
57			class AutoDiscovered:
58			"""Class to manage the auto discovered servers dict."""
59
60			def __init__(self):
61			# server_dict is a list of dict (JSON compliant)
62			# [ {'key': 'zeroconf name', ip': '172.1.2.3', 'port': 61209, 'protocol': 'rpc', 'cpu': 3, 'mem': 34 ...} ... ]
63			self._server_list = []
64
65			def get_servers_list(self):
66			"""Return the current server list (list of dict)."""
67			return self._server_list
68
69			def set_server(self, server_pos, key, value):
70			"""Set the key to the value for the server_pos (position in the list)."""
71			self._server_list[server_pos][key] = value
72
73			def add_server(self, name, ip, port, protocol='rpc'):
74			"""Add a new server to the list."""
75			new_server = {
76			'key': name, # Zeroconf name with both hostname and port
77			'name': name.split(':')[0], # Short name
78			'ip': ip, # IP address seen by the client
79			'port': port, # TCP port
80			'protocol': str(protocol), # RPC or RESTFUL protocol
81			'username': 'glances', # Default username
82			'password': '', # Default password
83			'status': 'UNKNOWN', # Server status: 'UNKNOWN', 'OFFLINE', 'ONLINE', 'PROTECTED'
84			'type': 'DYNAMIC',
85			} # Server type: 'STATIC' or 'DYNAMIC'
86			self._server_list.append(new_server)
87			logger.debug(f"Updated servers list ({len(self._server_list)} servers): {self._server_list}")
88
89			def remove_server(self, name):
90			"""Remove a server from the dict."""
91			for i in self._server_list:
92			if i['key'] == name:
93			try:
94			self._server_list.remove(i)
95			logger.debug(f"Remove server {name} from the list")
96			logger.debug(f"Updated servers list ({len(self._server_list)} servers): {self._server_list}")
97			except ValueError:
98			logger.error(f"Cannot remove server {name} from the list")
99
100
101			class GlancesAutoDiscoverListener:
102			"""Zeroconf listener for Glances server."""
103
104			def __init__(self):
105			# Create an instance of the servers list
106			self.servers = AutoDiscovered()
107
108			def get_servers_list(self):
109			"""Return the current server list (list of dict)."""
110			return self.servers.get_servers_list()
111
112			def set_server(self, server_pos, key, value):
113			"""Set the key to the value for the server_pos (position in the list)."""
114			self.servers.set_server(server_pos, key, value)
115
116			def add_service(self, zeroconf, srv_type, srv_name):
117			"""Method called when a new Zeroconf client is detected.
118
119			Note: the return code will never be used
120
121			:return: True if the zeroconf client is a Glances server
122			"""
123			if srv_type != zeroconf_type:
124			return False
125			logger.debug(f"Check new Zeroconf server: {srv_type} / {srv_name}")
126			info = zeroconf.get_service_info(srv_type, srv_name)
127			if info and (info.addresses or info.parsed_addresses):
128			address = info.addresses[0] if info.addresses else info.parsed_addresses[0]
129			new_server_ip = socket.inet_ntoa(address)
130			new_server_port = info.port
131			new_server_protocol = info.properties[b'protocol'].decode() if b'protocol' in info.properties else 'rpc'
132
133			# Add server to the global dict
134			self.servers.add_server(
135			srv_name,
136			new_server_ip,
137			new_server_port,
138			protocol=new_server_protocol,
139			)
140			logger.info(
141			f"New {new_server_protocol} Glances server detected ({srv_name} from {new_server_ip}:{new_server_port})"
142			)
143			else:
144			logger.warning("New Glances server detected, but failed to be get Zeroconf ServiceInfo ")
145			return True
146
147			def remove_service(self, zeroconf, srv_type, srv_name):
148			"""Remove the server from the list."""
149			self.servers.remove_server(srv_name)
150			logger.info(f"Glances server {srv_name} removed from the autodetect list")
151
152			def update_service(self, zeroconf, srv_type, srv_name):
153			"""Update the server from the list.
154			Done by a dirty hack (remove + add).
155			"""
156			self.remove_service(zeroconf, srv_type, srv_name)
157			self.add_service(zeroconf, srv_type, srv_name)
158			logger.info(f"Glances server {srv_name} updated from the autodetect list")
159
160
161			class GlancesAutoDiscoverServer:
162			"""Implementation of the Zeroconf protocol (server side for the Glances client)."""
163
164			def __init__(self, args=None):
165			if zeroconf_tag:
166			logger.info("Init autodiscover mode (Zeroconf protocol)")
167			try:
168			self.zeroconf = Zeroconf()
169			except OSError as e:
170			logger.error(f"Cannot start Zeroconf ({e})")
171			self.zeroconf_enable_tag = False
172			else:
173			self.listener = GlancesAutoDiscoverListener()
174			self.browser = ServiceBrowser(self.zeroconf, zeroconf_type, self.listener)
175			self.zeroconf_enable_tag = True
176			else:
177			logger.error("Cannot start autodiscover mode (Zeroconf lib is not installed)")
178			self.zeroconf_enable_tag = False
179
180			def get_servers_list(self):
181			"""Return the current server list (dict of dict)."""
182			if zeroconf_tag and self.zeroconf_enable_tag:
183			return self.listener.get_servers_list()
184			return []
185
186			def set_server(self, server_pos, key, value):
187			"""Set the key to the value for the server_pos (position in the list)."""
188			if zeroconf_tag and self.zeroconf_enable_tag:
189			self.listener.set_server(server_pos, key, value)
190
191			def close(self):
192			if zeroconf_tag and self.zeroconf_enable_tag:
193			self.zeroconf.close()
194
195
196			class GlancesAutoDiscoverClient:
197			"""Implementation of the zeroconf protocol (client side for the Glances server)."""
198
199			def __init__(self, hostname, args=None):
200			if zeroconf_tag:
201			zeroconf_bind_address = args.bind_address
202			try:
203			self.zeroconf = Zeroconf()
204			except OSError as e:
205			logger.error(f"Cannot start zeroconf: {e}")
206
207			# XXX *BSDs: Segmentation fault (core dumped)
208			# -- https://bitbucket.org/al45tair/netifaces/issues/15
209			if not BSD:
210			try:
211			# -B @ overwrite the dynamic IPv4 choice
212			if netifaces_tag and zeroconf_bind_address == '0.0.0.0':
213			zeroconf_bind_address = self.find_active_ip_address()
214			except Exception:
215			# Issue #528 (no network interface available)
216			# Issue #3219 (no implementation for gateway())
217			pass
218
219			# Ensure zeroconf_bind_address is an IP address not an host
220			zeroconf_bind_address = socket.gethostbyname(zeroconf_bind_address)
221
222			# Check IP v4/v6
223			address_family = socket.getaddrinfo(zeroconf_bind_address, args.port)[0][0]
224
225			# Start the zeroconf service
226			try:
227			self.info = ServiceInfo(
228			zeroconf_type,
229			f'{hostname}:{args.port}.{zeroconf_type}',
230			address=socket.inet_pton(address_family, zeroconf_bind_address),
231			port=args.port,
232			weight=0,
233			priority=0,
234			properties={'protocol': 'rest' if args.webserver else 'rpc'},
235			server=hostname,
236			)
237			except TypeError:
238			# Manage issue 1663 with breaking change on ServiceInfo method
239			# address (only one address) is replaced by addresses (list of addresses)
240			self.info = ServiceInfo(
241			zeroconf_type,
242			name=f'{hostname}:{args.port}.{zeroconf_type}',
243			addresses=[socket.inet_pton(address_family, zeroconf_bind_address)],
244			port=args.port,
245			weight=0,
246			priority=0,
247			properties={'protocol': 'rest' if args.webserver else 'rpc'},
248			server=hostname,
249			)
250			try:
251			self.zeroconf.register_service(self.info)
252			except Exception as e:
253			logger.error(f"Error while announcing Glances server: {e}")
254			else:
255			print(f"Announce the Glances server on the LAN (using {zeroconf_bind_address} IP address)")
256			else:
257			logger.error("Cannot announce Glances server on the network: zeroconf library not found.")
258
259			@staticmethod
260			def find_active_ip_address():
261			"""Try to find the active IP addresses."""
262			# Interface of the default gateway
263			gateway_itf = netifaces.gateways()[netifaces.AF_INET][0][1]
264			# IP address for the interface
265			return netifaces.ifaddresses(gateway_itf)[netifaces.AF_INET][0]['addr']
266
267			def close(self):
268			if zeroconf_tag:
269			self.zeroconf.unregister_service(self.info)
270			self.zeroconf.close()
271

nicolargo / glances

Push — master ( 73a01d...ef36eb )

AutoDiscovered.set_server() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like