glances.plugins.cloud.ThreadOpenStack.run() - 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

glances.plugins.cloud.ThreadOpenStack.run() B

↳ Parent: glances.plugins.cloud

Complexity

Conditions

Size

Total Lines	25
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	15
nop	1
dl	0
loc	25
rs	8
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
#

"""Cloud plugin.

Supported Cloud API:
- OpenStack meta data (class ThreadOpenStack) - Vanilla OpenStack
- OpenStackEC2 meta data (class ThreadOpenStackEC2) - Amazon EC2 compatible
"""

import threading

from glances.globals import to_ascii
from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel

# Import plugin specific dependency
try:
    import requests
except ImportError as e:
    import_error_tag = True
    # Display debug message if import error
    logger.warning(f"Missing Python Lib ({e}), Cloud plugin is disabled")
else:
    import_error_tag = False


class CloudPlugin(GlancesPluginModel):
    """Glances' cloud plugin.

    The goal of this plugin is to retrieve additional information
    concerning the datacenter where the host is connected.

    See https://github.com/nicolargo/glances/issues/1029

    stats is a dict
    """

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super().__init__(args=args, config=config)

        # We want to display the stat in the curse interface
        self.display_curse = True

        # Init the stats
        self.reset()

        # Enable threads only if the plugin is enabled
        self.OPENSTACK = None
        self.OPENSTACKEC2 = None
        if self.is_enabled():
            # Init thread to grab OpenStack stats asynchronously
            self.OPENSTACK = ThreadOpenStack()
            self.OPENSTACKEC2 = ThreadOpenStackEC2()

            # Run the thread
            self.OPENSTACK.start()
            self.OPENSTACKEC2.start()

    def exit(self):
        """Overwrite the exit method to close threads."""
        if self.OPENSTACK:
            self.OPENSTACK.stop()
        if self.OPENSTACKEC2:
            self.OPENSTACKEC2.stop()
        # Call the father class
        super().exit()

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update the cloud stats.

        Return the stats (dict)
        """
        # Init new stats
        stats = self.get_init_value()

        # Requests lib is needed to get stats from the Cloud API
        if import_error_tag:
            return stats

        # Update the stats
        if self.input_method == 'local' and (self.OPENSTACK or self.OPENSTACKEC2):
            stats = self.OPENSTACK.stats
            if not stats:
                stats = self.OPENSTACKEC2.stats
            # Example:
            # Uncomment to test on physical computer (only for test purpose)
            # stats = {'id': 'ami-id', 'name': 'My VM', 'type': 'Gold', 'region': 'France', 'platform': 'OpenStack'}

        # Update the stats
        self.stats = stats

        return self.stats

    def msg_curse(self, args=None, max_width=None):
        """Return the string to display in the curse interface."""
        # Init the return message
        ret = []

        if not self.stats or self.stats == {} or self.is_disabled():
            return ret

        # Do not display Unknown information in the cloud plugin #2485
        if not self.stats.get('platform') or not self.stats.get('name'):
            return ret

        # Generate the output
        msg = self.stats.get('platform', 'Unknown')
        ret.append(self.curse_add_line(msg, "TITLE"))
        msg = ' {} instance {} ({})'.format(
            self.stats.get('type', 'Unknown'), self.stats.get('name', 'Unknown'), self.stats.get('region', 'Unknown')
        )
        ret.append(self.curse_add_line(msg))

        # Return the message with decoration
        # logger.info(ret)
        return ret


class ThreadOpenStack(threading.Thread):
    """
    Specific thread to grab OpenStack stats.

    stats is a dict
    """

    # The metadata service provides a way for instances to retrieve
    # instance-specific data via a REST API. Instances access this
    # service at 169.254.169.254 or at fe80::a9fe:a9fe.
    # All types of metadata, be it user-, nova- or vendor-provided,
    # can be accessed via this service.
    # https://docs.openstack.org/nova/latest/user/metadata-service.html
    OPENSTACK_PLATFORM = "OpenStack"
    OPENSTACK_API_URL = 'http://169.254.169.254/openstack/latest/meta-data'
    OPENSTACK_API_METADATA = {
        'id': 'project_id',
        'name': 'name',
        'type': 'meta/role',
        'region': 'availability_zone',
    }

    def __init__(self):
        """Init the class."""
        logger.debug("cloud plugin - Create thread for OpenStack metadata")
        super().__init__()
        # Event needed to stop properly the thread
        self._stopper = threading.Event()
        # The class return the stats as a dict
        self._stats = {}

    def run(self):
        """Grab plugin's stats.

        Infinite loop, should be stopped by calling the stop() method
        """
        if import_error_tag:
            self.stop()
            return False

        for k, v in self.OPENSTACK_API_METADATA.items():
            r_url = f'{self.OPENSTACK_API_URL}/{v}'
            try:
                # Local request, a timeout of 3 seconds is OK
                r = requests.get(r_url, timeout=3)
            except Exception as e:
                logger.debug(f'cloud plugin - Cannot connect to the OpenStack metadata API {r_url}: {e}')
                break
            else:
                if r.ok:
                    self._stats[k] = to_ascii(r.content)
        else:
            # No break during the loop, so we can set the platform
            self._stats['platform'] = self.OPENSTACK_PLATFORM

        return True

    @property
    def stats(self):
        """Stats getter."""
        return self._stats

    @stats.setter
    def stats(self, value):
        """Stats setter."""
        self._stats = value

    def stop(self, timeout=None):
        """Stop the thread."""
        logger.debug("cloud plugin - Close thread for OpenStack metadata")
        self._stopper.set()

    def stopped(self):
        """Return True is the thread is stopped."""
        return self._stopper.is_set()


class ThreadOpenStackEC2(ThreadOpenStack):
    """
    Specific thread to grab OpenStack EC2 (Amazon cloud) stats.

    stats is a dict
    """

    # The metadata service provides a way for instances to retrieve
    # instance-specific data via a REST API. Instances access this
    # service at 169.254.169.254 or at fe80::a9fe:a9fe.
    # All types of metadata, be it user-, nova- or vendor-provided,
    # can be accessed via this service.
    # https://docs.openstack.org/nova/latest/user/metadata-service.html
    OPENSTACK_PLATFORM = "Amazon EC2"
    OPENSTACK_API_URL = 'http://169.254.169.254/latest/meta-data'
    OPENSTACK_API_METADATA = {
        'id': 'ami-id',
        'name': 'instance-id',
        'type': 'instance-type',
        'region': 'placement/availability-zone',
    }


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			"""Cloud plugin.
10
11			Supported Cloud API:
12			- OpenStack meta data (class ThreadOpenStack) - Vanilla OpenStack
13			- OpenStackEC2 meta data (class ThreadOpenStackEC2) - Amazon EC2 compatible
14			"""
15
16			import threading
17
18			from glances.globals import to_ascii
19			from glances.logger import logger
20			from glances.plugins.plugin.model import GlancesPluginModel
21
22			# Import plugin specific dependency
23			try:
24			import requests
25			except ImportError as e:
26			import_error_tag = True
27			# Display debug message if import error
28			logger.warning(f"Missing Python Lib ({e}), Cloud plugin is disabled")
29			else:
30			import_error_tag = False
31
32
33			class CloudPlugin(GlancesPluginModel):
34			"""Glances' cloud plugin.
35
36			The goal of this plugin is to retrieve additional information
37			concerning the datacenter where the host is connected.
38
39			See https://github.com/nicolargo/glances/issues/1029
40
41			stats is a dict
42			"""
43
44			def __init__(self, args=None, config=None):
45			"""Init the plugin."""
46			super().__init__(args=args, config=config)
47
48			# We want to display the stat in the curse interface
49			self.display_curse = True
50
51			# Init the stats
52			self.reset()
53
54			# Enable threads only if the plugin is enabled
55			self.OPENSTACK = None
56			self.OPENSTACKEC2 = None
57			if self.is_enabled():
58			# Init thread to grab OpenStack stats asynchronously
59			self.OPENSTACK = ThreadOpenStack()
60			self.OPENSTACKEC2 = ThreadOpenStackEC2()
61
62			# Run the thread
63			self.OPENSTACK.start()
64			self.OPENSTACKEC2.start()
65
66			def exit(self):
67			"""Overwrite the exit method to close threads."""
68			if self.OPENSTACK:
69			self.OPENSTACK.stop()
70			if self.OPENSTACKEC2:
71			self.OPENSTACKEC2.stop()
72			# Call the father class
73			super().exit()
74
75			@GlancesPluginModel._check_decorator
76			@GlancesPluginModel._log_result_decorator
77			def update(self):
78			"""Update the cloud stats.
79
80			Return the stats (dict)
81			"""
82			# Init new stats
83			stats = self.get_init_value()
84
85			# Requests lib is needed to get stats from the Cloud API
86			if import_error_tag:
87			return stats
88
89			# Update the stats
90			if self.input_method == 'local' and (self.OPENSTACK or self.OPENSTACKEC2):
91			stats = self.OPENSTACK.stats
92			if not stats:
93			stats = self.OPENSTACKEC2.stats
94			# Example:
95			# Uncomment to test on physical computer (only for test purpose)
96			# stats = {'id': 'ami-id', 'name': 'My VM', 'type': 'Gold', 'region': 'France', 'platform': 'OpenStack'}
97
98			# Update the stats
99			self.stats = stats
100
101			return self.stats
102
103			def msg_curse(self, args=None, max_width=None):
104			"""Return the string to display in the curse interface."""
105			# Init the return message
106			ret = []
107
108			if not self.stats or self.stats == {} or self.is_disabled():
109			return ret
110
111			# Do not display Unknown information in the cloud plugin #2485
112			if not self.stats.get('platform') or not self.stats.get('name'):
113			return ret
114
115			# Generate the output
116			msg = self.stats.get('platform', 'Unknown')
117			ret.append(self.curse_add_line(msg, "TITLE"))
118			msg = ' {} instance {} ({})'.format(
119			self.stats.get('type', 'Unknown'), self.stats.get('name', 'Unknown'), self.stats.get('region', 'Unknown')
120			)
121			ret.append(self.curse_add_line(msg))
122
123			# Return the message with decoration
124			# logger.info(ret)
125			return ret
126
127
128			class ThreadOpenStack(threading.Thread):
129			"""
130			Specific thread to grab OpenStack stats.
131
132			stats is a dict
133			"""
134
135			# The metadata service provides a way for instances to retrieve
136			# instance-specific data via a REST API. Instances access this
137			# service at 169.254.169.254 or at fe80::a9fe:a9fe.
138			# All types of metadata, be it user-, nova- or vendor-provided,
139			# can be accessed via this service.
140			# https://docs.openstack.org/nova/latest/user/metadata-service.html
141			OPENSTACK_PLATFORM = "OpenStack"
142			OPENSTACK_API_URL = 'http://169.254.169.254/openstack/latest/meta-data'
143			OPENSTACK_API_METADATA = {
144			'id': 'project_id',
145			'name': 'name',
146			'type': 'meta/role',
147			'region': 'availability_zone',
148			}
149
150			def __init__(self):
151			"""Init the class."""
152			logger.debug("cloud plugin - Create thread for OpenStack metadata")
153			super().__init__()
154			# Event needed to stop properly the thread
155			self._stopper = threading.Event()
156			# The class return the stats as a dict
157			self._stats = {}
158
159			def run(self):
160			"""Grab plugin's stats.
161
162			Infinite loop, should be stopped by calling the stop() method
163			"""
164			if import_error_tag:
165			self.stop()
166			return False
167
168			for k, v in self.OPENSTACK_API_METADATA.items():
169			r_url = f'{self.OPENSTACK_API_URL}/{v}'
170			try:
171			# Local request, a timeout of 3 seconds is OK
172			r = requests.get(r_url, timeout=3)
173			except Exception as e:
174			logger.debug(f'cloud plugin - Cannot connect to the OpenStack metadata API {r_url}: {e}')
175			break
176			else:
177			if r.ok:
178			self._stats[k] = to_ascii(r.content)
179			else:
180			# No break during the loop, so we can set the platform
181			self._stats['platform'] = self.OPENSTACK_PLATFORM
182
183			return True
184
185			@property
186			def stats(self):
187			"""Stats getter."""
188			return self._stats
189
190			@stats.setter
191			def stats(self, value):
192			"""Stats setter."""
193			self._stats = value
194
195			def stop(self, timeout=None):
196			"""Stop the thread."""
197			logger.debug("cloud plugin - Close thread for OpenStack metadata")
198			self._stopper.set()
199
200			def stopped(self):
201			"""Return True is the thread is stopped."""
202			return self._stopper.is_set()
203
204
205			class ThreadOpenStackEC2(ThreadOpenStack):
206			"""
207			Specific thread to grab OpenStack EC2 (Amazon cloud) stats.
208
209			stats is a dict
210			"""
211
212			# The metadata service provides a way for instances to retrieve
213			# instance-specific data via a REST API. Instances access this
214			# service at 169.254.169.254 or at fe80::a9fe:a9fe.
215			# All types of metadata, be it user-, nova- or vendor-provided,
216			# can be accessed via this service.
217			# https://docs.openstack.org/nova/latest/user/metadata-service.html
218			OPENSTACK_PLATFORM = "Amazon EC2"
219			OPENSTACK_API_URL = 'http://169.254.169.254/latest/meta-data'
220			OPENSTACK_API_METADATA = {
221			'id': 'ami-id',
222			'name': 'instance-id',
223			'type': 'instance-type',
224			'region': 'placement/availability-zone',
225			}
226

nicolargo / glances

Push — master ( 73a01d...ef36eb )

glances.plugins.cloud.ThreadOpenStack.run() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like