glances.autodiscover.GlancesAutoDiscoverServer.close() - Code Metrics - Inspection of "Update release note for Glances 3.2.5" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 7e7379...128504 )

by Nicolas

created 2022-04-10 14:14 UTC

GlancesAutoDiscoverServer.close() A

↳ Parent: glances.autodiscover

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	3
nop	1
dl	0
loc	3
rs	10
c	0
b	0
f	0

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2021 Nicolargo <[email protected]>
#
# Glances is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Glances is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

"""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 __version__ as __zeroconf_version, ServiceBrowser, ServiceInfo, Zeroconf

    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("Zeroconf version {} detected.".format(__zeroconf_version))
    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 doesnt like a zeroconf type that ends with '._tcp.'.
# Correct issue: zeroconf problem with zeroconf_type = "_%s._tcp." % 'glances' #888
zeroconf_type = "_%s._tcp.local." % 'glances'


class AutoDiscovered(object):

    """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, '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):
        """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
            '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("Updated servers list (%s servers): %s" % (len(self._server_list), 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("Remove server %s from the list" % name)
                    logger.debug("Updated servers list (%s servers): %s" % (len(self._server_list), self._server_list))
                except ValueError:
                    logger.error("Cannot remove server %s from the list" % name)


class GlancesAutoDiscoverListener(object):

    """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("Check new Zeroconf server: %s / %s" % (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

            # Add server to the global dict
            self.servers.add_server(srv_name, new_server_ip, new_server_port)
            logger.info("New Glances server detected (%s from %s:%s)" % (srv_name, 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("Glances server %s removed from the autodetect list" % srv_name)


class GlancesAutoDiscoverServer(object):

    """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 socket.error as e:
                logger.error("Cannot start Zeroconf (%s)" % 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()
        else:
            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(object):

    """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 socket.error as e:
                logger.error("Cannot start zeroconf: {}".format(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

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

nicolargo / glances

Push — master ( 7e7379...128504 )

GlancesAutoDiscoverServer.close() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like