glances.plugins.cloud.PluginModel.msg_curse() - Code Metrics - Inspection of "version 4.2.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( cc9054...a8608f )

by Nicolas

created 2024-10-20 16:02 UTC

glances.plugins.cloud.PluginModel.msg_curse() B

↳ Parent: glances.plugins.cloud

Complexity

Conditions

Size

Total Lines	23
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	12
nop	3
dl	0
loc	23
rs	8.6666
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 iteritems, 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 PluginModel(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()

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

nicolargo / glances

Push — master ( cc9054...a8608f )

glances.plugins.cloud.PluginModel.msg_curse() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like