glances.servers_list_dynamic.GlancesAutoDiscoverClient.__init__() - 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

GlancesAutoDiscoverClient.init() C

↳ Parent: glances.servers_list_dynamic

Complexity

Conditions

Size

Total Lines	58
Code Lines	41

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	11
eloc	41
nop	3
dl	0
loc	58
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.servers_list_dynamic.GlancesAutoDiscoverClient.__init__() 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 autodiscover Glances server (thk to the ZeroConf protocol)."""

import socket
import sys

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

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 zeroconf_bind_address == '0.0.0.0':
                        zeroconf_bind_address = self.find_active_ip_address()
                except KeyError:
                    # Issue #528 (no network interface available)
                    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."""
        import netifaces

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

nicolargo / glances

Push — master ( ce0fc3...e09530 )

GlancesAutoDiscoverClient.__init__() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

GlancesAutoDiscoverClient.init() C