glances.plugins.glances_cloud.ThreadOpenStack.stopped() - Code Metrics - Inspection of "Merge branch 'master' into develop" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 05aaee...10b5c2 )

by Nicolas

created 2023-05-16 07:38 UTC

ThreadOpenStack.stopped() A

↳ Parent: glances.plugins.glances_cloud

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# -*- coding: utf-8 -*-
#
# 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.compat import iteritems, to_ascii
from glances.plugins.glances_plugin import GlancesPlugin
from glances.logger import logger

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


class Plugin(GlancesPlugin):
    """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(Plugin, self).__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(Plugin, self).exit()

    @GlancesPlugin._check_decorator
    @GlancesPlugin._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

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

nicolargo / glances

Push — master ( 05aaee...10b5c2 )

ThreadOpenStack.stopped() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like