glances.stats.GlancesStats.load_additional_plugins() - Code Metrics - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

GlancesStats.load_additional_plugins() F
last analyzed 2025-06-14 08:50 UTC

↳ Parent: glances.stats

Complexity

Conditions

Size

Total Lines	55
Code Lines	37

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	19
eloc	37
nop	3
dl	0
loc	55
rs	0.5999
c	0
b	0
f	0

How to fix Long Method Complexity

#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""The stats manager."""

import collections
import os
import pathlib
import sys
import threading
import traceback
from importlib import import_module

from glances.globals import exports_path, plugins_path, sys_path
from glances.logger import logger
from glances.timer import Counter


class GlancesStats:
    """This class stores, updates and gives stats."""

    # Script header constant
    header = "glances_"

    def __init__(self, config=None, args=None):
        # Set the config instance
        self.config = config

        # Set the argument instance
        self.args = args

        # Load plugins and exports modules
        self.first_export = True
        self.load_modules(self.args)

    def __getattr__(self, item):
        """Overwrite the getattr method in case of attribute is not found.

        The goal is to dynamically generate the following methods:
        - getPlugname(): return Plugname stat in JSON format
        - getViewsPlugname(): return views of the Plugname stat in JSON format
        """
        # Check if the attribute starts with 'get'
        if item.startswith('getViews'):

            # Get the plugin name
            plugname = item[len('getViews') :].lower()
            # Get the plugin instance
            plugin = self._plugins[plugname]
            if hasattr(plugin, 'get_json_views'):
                # The method get_json_views exist, return it
                return getattr(plugin, 'get_json_views')
            # The method get_views is not found for the plugin
            raise AttributeError(item)
        if item.startswith('get'):

            # Get the plugin name
            plugname = item[len('get') :].lower()
            # Get the plugin instance
            plugin = self._plugins[plugname]
            if hasattr(plugin, 'get_json'):
                # The method get_json exist, return it
                return getattr(plugin, 'get_json')
            # The method get_stats is not found for the plugin
            raise AttributeError(item)
        # Default behavior
        raise AttributeError(item)

    def load_modules(self, args):
        """Wrapper to load: plugins and export modules."""

        # Init the plugins dict
        # Active plugins dictionary
        self._plugins = collections.defaultdict(dict)
        # Load the plugins
        self.load_plugins(args=args)

        # Load addititional plugins
        self.load_additional_plugins(args=args, config=self.config)

        # Init the export modules dict
        # Active exporters dictionary
        self._exports = collections.defaultdict(dict)
        # All available exporters dictionary
        self._exports_all = collections.defaultdict(dict)
        # Load the export modules
        self.load_exports(args=args)

        # Restoring system path
        sys.path = sys_path

    def _load_plugin(self, plugin_path, args=None, config=None):
        """Load the plugin, init it and add to the _plugin dict."""
        # Load the plugin class
        try:
            # Import the plugin
            plugin = import_module('glances.plugins.' + plugin_path)
            # Init and add the plugin to the dictionary
            if hasattr(plugin, 'PluginModel'):
                # Old fashion way to load the plugin (before Glances 5.0)
                # Should be removed in Glances 5.0 - see #3170
                self._plugins[plugin_path] = getattr(plugin, 'PluginModel')(args=args, config=config)
                logger.warning(
                    f'The {plugin_path} plugin class name is "PluginModel" and it is deprecated, \
please rename it to "{plugin_path.capitalize()}Plugin"'
                )
            elif hasattr(plugin, plugin_path.capitalize() + 'Plugin'):
                # New fashion way to load the plugin (after Glances 5.0)
                self._plugins[plugin_path] = getattr(plugin, plugin_path.capitalize() + 'Plugin')(
                    args=args, config=config
                )
        except Exception as e:
            # If a plugin can not be loaded, display a critical message
            # on the console but do not crash
            logger.critical(f"Error while initializing the {plugin_path} plugin ({e})")
            logger.error(traceback.format_exc())
            # An error occurred, disable the plugin
            if args is not None:
                setattr(args, 'disable_' + plugin_path, False)
        else:
            # Manage the default status of the plugin (enable or disable)
            if args is not None:
                # If the all keys are set in the disable_plugin option then look in the enable_plugin option
                if getattr(args, 'disable_all', False):
                    logger.debug('%s => %s', plugin_path, getattr(args, 'enable_' + plugin_path, False))
                    setattr(args, 'disable_' + plugin_path, not getattr(args, 'enable_' + plugin_path, False))
                else:
                    setattr(args, 'disable_' + plugin_path, getattr(args, 'disable_' + plugin_path, False))

    def load_plugins(self, args=None):
        """Load all plugins in the 'plugins' folder."""
        start_duration = Counter()

        for item in os.listdir(plugins_path):
            if os.path.isdir(os.path.join(plugins_path, item)) and not item.startswith('__') and item != 'plugin':
                # Load the plugin
                start_duration.reset()
                self._load_plugin(os.path.basename(item), args=args, config=self.config)
                logger.debug(f"Plugin {item} started in {start_duration.get()} seconds")

        # Log plugins list
        logger.debug(f"Active plugins list: {self.getPluginsList()}")

    def load_additional_plugins(self, args=None, config=None):
        """Load additional plugins if defined"""

        def get_addl_plugins(self, plugin_path):
            """Get list of additional plugins"""
            _plugin_list = []
            for plugin in os.listdir(plugin_path):
                path = os.path.join(plugin_path, plugin)
                if os.path.isdir(path) and not path.startswith('__'):
                    # Poor man's walk_pkgs - can't use pkgutil as the module would be already imported here!
                    for fil in pathlib.Path(path).glob('*.py'):
                        if fil.is_file():
                            with open(fil) as fd:
                                # The first test should be removed in Glances 5.x - see #3170
                                if 'PluginModel' in fd.read() or plugin.capitalize() + 'Plugin' in fd.read():
                                    _plugin_list.append(plugin)
                                    break

            return _plugin_list

        path = None
        # Skip section check as implied by has_option
        if config and config.parser.has_option('global', 'plugin_dir'):
            path = config.parser['global']['plugin_dir']

        if args and 'plugin_dir' in args and args.plugin_dir:
            path = args.plugin_dir

        if path:
            # Get list before starting the counter
            _sys_path = sys.path
            start_duration = Counter()
            # Ensure that plugins can be found in plugin_dir
            sys.path.insert(0, path)
            for plugin in get_addl_plugins(self, path):
                if plugin in sys.modules:
                    logger.warn(f"Plugin {plugin} already in sys.modules, skipping (workaround: rename plugin)")
                else:
                    start_duration.reset()
                    try:
                        _mod_loaded = import_module(plugin + '.model')
                        self._plugins[plugin] = _mod_loaded.PluginModel(args=args, config=config)
                        logger.debug(f"Plugin {plugin} started in {start_duration.get()} seconds")
                    except Exception as e:
                        # If a plugin can not be loaded, display a critical message
                        # on the console but do not crash
                        logger.critical(f"Error while initializing the {plugin} plugin ({e})")
                        logger.error(traceback.format_exc())
                        # An error occurred, disable the plugin
                        if args:
                            setattr(args, 'disable_' + plugin, False)

            sys.path = _sys_path
            # Log plugins list
            logger.debug(f"Active additional plugins list: {self.getPluginsList()}")

    def load_exports(self, args=None):
        """Load all exporters in the 'exports' folder."""
        start_duration = Counter()

        if args is None:
            return False

        for item in os.listdir(exports_path):
            if os.path.isdir(os.path.join(exports_path, item)) and not item.startswith('__'):
                # Load the exporter
                start_duration.reset()
                if item.startswith('glances_'):
                    # Avoid circular loop when Glances exporter uses lib with same name
                    # Example: influxdb should be named to glances_influxdb
                    exporter_name = os.path.basename(item).split('glances_')[1]
                else:
                    exporter_name = os.path.basename(item)
                # Set the disable_<name> to False by default
                setattr(self.args, 'export_' + exporter_name, getattr(self.args, 'export_' + exporter_name, False))
                # We should import the module
                if getattr(self.args, 'export_' + exporter_name, False):
                    # Import the export module
                    export_module = import_module(item)
                    # Add the exporter instance to the active exporters dictionary
                    self._exports[exporter_name] = export_module.Export(args=args, config=self.config)
                    # Add the exporter instance to the available exporters dictionary
                    self._exports_all[exporter_name] = self._exports[exporter_name]
                else:
                    # Add the exporter name to the available exporters dictionary
                    self._exports_all[exporter_name] = exporter_name
                logger.debug(f"Exporter {exporter_name} started in {start_duration.get()} seconds")

        # Log plugins list
        logger.debug(f"Active exports modules list: {self.getExportsList()}")
        return True

    def getPluginsList(self, enable=True):
        """Return the plugins list.

        if enable is True, only return the active plugins (default)
        if enable is False, return all the plugins

        Return: list of plugin name
        """
        if enable:
            return [p for p in self._plugins if self._plugins[p].is_enabled()]
        return list(self._plugins)

    def getExportsList(self, enable=True):
        """Return the exports list.

        if enable is True, only return the active exporters (default)
        if enable is False, return all the exporters

        :return: list of export module names
        """
        if enable:
            return list(self._exports)
        return list(self._exports_all)

    def load_limits(self, config=None):
        """Load the stats limits (except the one in the exclude list)."""
        # For each plugins (enable or not), call the load_limits method
        for p in self.getPluginsList(enable=False):
            self._plugins[p].load_limits(config)

    def __update_plugin(self, p):
        """Update stats, history and views for the given plugin name p"""
        self._plugins[p].update()
        self._plugins[p].update_views()
        self._plugins[p].update_stats_history()

    def update(self):
        """Wrapper method to update all stats.

        Only called by standalone and server modes
        """
        # Start update of all enable plugins
        for p in self.getPluginsList(enable=True):
            self.__update_plugin(p)

    def export(self, input_stats=None):
        """Export all the stats.

        Each export module is ran in a dedicated thread.
        """
        if self.first_export:
            # Init fields description
            # Why ? Because some exports modules need to know what is exported (name, type...)
            for e in self.getExportsList():
                logger.debug(f"Init exported stats using the {e} module")
                # @TODO: is it a good idea to thread this call ?
                self._exports[e].init_fields(input_stats)
            # In this first loop, data are not exported because some information are missing (rate for example)
            logger.debug("Do not export stats during the first iteration because some information are missing")
            self.first_export = False
            return False

        input_stats = input_stats or {}

        for e in self.getExportsList():
            logger.debug(f"Export stats using the {e} module")
            thread = threading.Thread(target=self._exports[e].update, args=(input_stats,))
            thread.start()

        return True

    def getAll(self):
        """Return all the stats (list).
        This method is called byt the XML/RPC API.
        It should return all the plugins (enable or not) because filtering can be done by the client.
        """
        return [self._plugins[p].get_raw() for p in self.getPluginsList(enable=False)]

    def getAllAsDict(self, plugin_list=None):
        """Return all the stats (as dict).
        This method is called by the RESTFul API.
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return {p: self._plugins[p].get_raw() for p in plugin_list}

    def getAllFieldsDescription(self):
        """Return all fields description (as list)."""
        return [self._plugins[p].fields_description for p in self.getPluginsList(enable=False)]

    def getAllFieldsDescriptionAsDict(self, plugin_list=None):
        """Return all fields description (as dict)."""
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return {p: self._plugins[p].fields_description for p in plugin_list}

    def getAllExports(self, plugin_list=None):
        """Return all the stats to be exported as a list.

        Default behavior is to export all the stat
        if plugin_list is provided (list), only export stats of given plugins
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return [self._plugins[p].get_export() for p in plugin_list]

    def getAllExportsAsDict(self, plugin_list=None):
        """Return all the stats to be exported as a dict.

        Default behavior is to export all the stat
        if plugin_list is provided (list), only export stats of given plugins
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return {p: self._plugins[p].get_export() for p in plugin_list}

    def getAllLimits(self, plugin_list=None):
        """Return the plugins limits list.

        Default behavior is to export all the limits
        if plugin_list is provided, only export limits of given plugin (list)
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return [self._plugins[p].limits for p in plugin_list]

    def getAllLimitsAsDict(self, plugin_list=None):
        """Return all the stats limits (dict).

        Default behavior is to export all the limits
        if plugin_list is provided, only export limits of given plugin (list)
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return {p: self._plugins[p].limits for p in plugin_list}

    def getAllViews(self, plugin_list=None):
        """Return the plugins views.
        This method is called byt the XML/RPC API.
        It should return all the plugins views (enable or not) because filtering can be done by the client.
        """
        if plugin_list is None:
            plugin_list = self.getPluginsList(enable=False)
        return [self._plugins[p].get_views() for p in plugin_list]

    def getAllViewsAsDict(self, plugin_list=None):
        """Return all the stats views (dict).
        This method is called by the RESTFul API.
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return {p: self._plugins[p].get_views() for p in plugin_list}

    def get_plugin(self, plugin_name):
        """Return the plugin stats."""
        if plugin_name in self._plugins:
            return self._plugins[plugin_name]
        return None

    def get_plugin_view(self, plugin_name):
        """Return the plugin views."""
        if plugin_name in self._plugins:
            return self._plugins[plugin_name].get_views()
        return None

    def end(self):
        """End of the Glances stats."""
        # Close export modules
        for e in self._exports:
            self._exports[e].exit()
        # Close plugins
        for p in self._plugins:
            self._plugins[p].exit()


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		"""The stats manager."""
10
11		import collections
12		import os
13		import pathlib
14		import sys
15		import threading
16		import traceback
17		from importlib import import_module
18
19		from glances.globals import exports_path, plugins_path, sys_path
20		from glances.logger import logger
21		from glances.timer import Counter
22
23
24		class GlancesStats:
25		"""This class stores, updates and gives stats."""
26
27		# Script header constant
28		header = "glances_"
29
30		def __init__(self, config=None, args=None):
31		# Set the config instance
32		self.config = config
33
34		# Set the argument instance
35		self.args = args
36
37		# Load plugins and exports modules
38		self.first_export = True
39		self.load_modules(self.args)
40
41		def __getattr__(self, item):
42		"""Overwrite the getattr method in case of attribute is not found.
43
44		The goal is to dynamically generate the following methods:
45		- getPlugname(): return Plugname stat in JSON format
46		- getViewsPlugname(): return views of the Plugname stat in JSON format
47		"""
48		# Check if the attribute starts with 'get'
49	View Code Duplication	if item.startswith('getViews'):
		0 ignored issues – show Duplication introduced 2024-05-16 22:50 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
50		# Get the plugin name
51		plugname = item[len('getViews') :].lower()
52		# Get the plugin instance
53		plugin = self._plugins[plugname]
54		if hasattr(plugin, 'get_json_views'):
55		# The method get_json_views exist, return it
56		return getattr(plugin, 'get_json_views')
57		# The method get_views is not found for the plugin
58		raise AttributeError(item)
59	View Code Duplication	if item.startswith('get'):
		0 ignored issues – show Duplication introduced 2024-05-16 22:50 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
60		# Get the plugin name
61		plugname = item[len('get') :].lower()
62		# Get the plugin instance
63		plugin = self._plugins[plugname]
64		if hasattr(plugin, 'get_json'):
65		# The method get_json exist, return it
66		return getattr(plugin, 'get_json')
67		# The method get_stats is not found for the plugin
68		raise AttributeError(item)
69		# Default behavior
70		raise AttributeError(item)
71
72		def load_modules(self, args):
73		"""Wrapper to load: plugins and export modules."""
74
75		# Init the plugins dict
76		# Active plugins dictionary
77		self._plugins = collections.defaultdict(dict)
78		# Load the plugins
79		self.load_plugins(args=args)
80
81		# Load addititional plugins
82		self.load_additional_plugins(args=args, config=self.config)
83
84		# Init the export modules dict
85		# Active exporters dictionary
86		self._exports = collections.defaultdict(dict)
87		# All available exporters dictionary
88		self._exports_all = collections.defaultdict(dict)
89		# Load the export modules
90		self.load_exports(args=args)
91
92		# Restoring system path
93		sys.path = sys_path
94
95		def _load_plugin(self, plugin_path, args=None, config=None):
96		"""Load the plugin, init it and add to the _plugin dict."""
97		# Load the plugin class
98		try:
99		# Import the plugin
100		plugin = import_module('glances.plugins.' + plugin_path)
101		# Init and add the plugin to the dictionary
102		if hasattr(plugin, 'PluginModel'):
103		# Old fashion way to load the plugin (before Glances 5.0)
104		# Should be removed in Glances 5.0 - see #3170
105		self._plugins[plugin_path] = getattr(plugin, 'PluginModel')(args=args, config=config)
106		logger.warning(
107		f'The {plugin_path} plugin class name is "PluginModel" and it is deprecated, \
108		please rename it to "{plugin_path.capitalize()}Plugin"'
109		)
110		elif hasattr(plugin, plugin_path.capitalize() + 'Plugin'):
111		# New fashion way to load the plugin (after Glances 5.0)
112		self._plugins[plugin_path] = getattr(plugin, plugin_path.capitalize() + 'Plugin')(
113		args=args, config=config
114		)
115		except Exception as e:
116		# If a plugin can not be loaded, display a critical message
117		# on the console but do not crash
118		logger.critical(f"Error while initializing the {plugin_path} plugin ({e})")
119		logger.error(traceback.format_exc())
120		# An error occurred, disable the plugin
121		if args is not None:
122		setattr(args, 'disable_' + plugin_path, False)
123		else:
124		# Manage the default status of the plugin (enable or disable)
125		if args is not None:
126		# If the all keys are set in the disable_plugin option then look in the enable_plugin option
127		if getattr(args, 'disable_all', False):
128		logger.debug('%s => %s', plugin_path, getattr(args, 'enable_' + plugin_path, False))
129		setattr(args, 'disable_' + plugin_path, not getattr(args, 'enable_' + plugin_path, False))
130		else:
131		setattr(args, 'disable_' + plugin_path, getattr(args, 'disable_' + plugin_path, False))
132
133		def load_plugins(self, args=None):
134		"""Load all plugins in the 'plugins' folder."""
135		start_duration = Counter()
136
137		for item in os.listdir(plugins_path):
138		if os.path.isdir(os.path.join(plugins_path, item)) and not item.startswith('__') and item != 'plugin':
139		# Load the plugin
140		start_duration.reset()
141		self._load_plugin(os.path.basename(item), args=args, config=self.config)
142		logger.debug(f"Plugin {item} started in {start_duration.get()} seconds")
143
144		# Log plugins list
145		logger.debug(f"Active plugins list: {self.getPluginsList()}")
146
147		def load_additional_plugins(self, args=None, config=None):
148		"""Load additional plugins if defined"""
149
150		def get_addl_plugins(self, plugin_path):
151		"""Get list of additional plugins"""
152		_plugin_list = []
153		for plugin in os.listdir(plugin_path):
154		path = os.path.join(plugin_path, plugin)
155		if os.path.isdir(path) and not path.startswith('__'):
156		# Poor man's walk_pkgs - can't use pkgutil as the module would be already imported here!
157		for fil in pathlib.Path(path).glob('*.py'):
158		if fil.is_file():
159		with open(fil) as fd:
160		# The first test should be removed in Glances 5.x - see #3170
161		if 'PluginModel' in fd.read() or plugin.capitalize() + 'Plugin' in fd.read():
162		_plugin_list.append(plugin)
163		break
164
165		return _plugin_list
166
167		path = None
168		# Skip section check as implied by has_option
169		if config and config.parser.has_option('global', 'plugin_dir'):
170		path = config.parser['global']['plugin_dir']
171
172		if args and 'plugin_dir' in args and args.plugin_dir:
173		path = args.plugin_dir
174
175		if path:
176		# Get list before starting the counter
177		_sys_path = sys.path
178		start_duration = Counter()
179		# Ensure that plugins can be found in plugin_dir
180		sys.path.insert(0, path)
181		for plugin in get_addl_plugins(self, path):
182		if plugin in sys.modules:
183		logger.warn(f"Plugin {plugin} already in sys.modules, skipping (workaround: rename plugin)")
184		else:
185		start_duration.reset()
186		try:
187		_mod_loaded = import_module(plugin + '.model')
188		self._plugins[plugin] = _mod_loaded.PluginModel(args=args, config=config)
189		logger.debug(f"Plugin {plugin} started in {start_duration.get()} seconds")
190		except Exception as e:
191		# If a plugin can not be loaded, display a critical message
192		# on the console but do not crash
193		logger.critical(f"Error while initializing the {plugin} plugin ({e})")
194		logger.error(traceback.format_exc())
195		# An error occurred, disable the plugin
196		if args:
197		setattr(args, 'disable_' + plugin, False)
198
199		sys.path = _sys_path
200		# Log plugins list
201		logger.debug(f"Active additional plugins list: {self.getPluginsList()}")
202
203		def load_exports(self, args=None):
204		"""Load all exporters in the 'exports' folder."""
205		start_duration = Counter()
206
207		if args is None:
208		return False
209
210		for item in os.listdir(exports_path):
211		if os.path.isdir(os.path.join(exports_path, item)) and not item.startswith('__'):
212		# Load the exporter
213		start_duration.reset()
214		if item.startswith('glances_'):
215		# Avoid circular loop when Glances exporter uses lib with same name
216		# Example: influxdb should be named to glances_influxdb
217		exporter_name = os.path.basename(item).split('glances_')[1]
218		else:
219		exporter_name = os.path.basename(item)
220		# Set the disable_<name> to False by default
221		setattr(self.args, 'export_' + exporter_name, getattr(self.args, 'export_' + exporter_name, False))
222		# We should import the module
223		if getattr(self.args, 'export_' + exporter_name, False):
224		# Import the export module
225		export_module = import_module(item)
226		# Add the exporter instance to the active exporters dictionary
227		self._exports[exporter_name] = export_module.Export(args=args, config=self.config)
228		# Add the exporter instance to the available exporters dictionary
229		self._exports_all[exporter_name] = self._exports[exporter_name]
230		else:
231		# Add the exporter name to the available exporters dictionary
232		self._exports_all[exporter_name] = exporter_name
233		logger.debug(f"Exporter {exporter_name} started in {start_duration.get()} seconds")
234
235		# Log plugins list
236		logger.debug(f"Active exports modules list: {self.getExportsList()}")
237		return True
238
239		def getPluginsList(self, enable=True):
240		"""Return the plugins list.
241
242		if enable is True, only return the active plugins (default)
243		if enable is False, return all the plugins
244
245		Return: list of plugin name
246		"""
247		if enable:
248		return [p for p in self._plugins if self._plugins[p].is_enabled()]
249		return list(self._plugins)
250
251		def getExportsList(self, enable=True):
252		"""Return the exports list.
253
254		if enable is True, only return the active exporters (default)
255		if enable is False, return all the exporters
256
257		:return: list of export module names
258		"""
259		if enable:
260		return list(self._exports)
261		return list(self._exports_all)
262
263		def load_limits(self, config=None):
264		"""Load the stats limits (except the one in the exclude list)."""
265		# For each plugins (enable or not), call the load_limits method
266		for p in self.getPluginsList(enable=False):
267		self._plugins[p].load_limits(config)
268
269		def __update_plugin(self, p):
270		"""Update stats, history and views for the given plugin name p"""
271		self._plugins[p].update()
272		self._plugins[p].update_views()
273		self._plugins[p].update_stats_history()
274
275		def update(self):
276		"""Wrapper method to update all stats.
277
278		Only called by standalone and server modes
279		"""
280		# Start update of all enable plugins
281		for p in self.getPluginsList(enable=True):
282		self.__update_plugin(p)
283
284		def export(self, input_stats=None):
285		"""Export all the stats.
286
287		Each export module is ran in a dedicated thread.
288		"""
289		if self.first_export:
290		# Init fields description
291		# Why ? Because some exports modules need to know what is exported (name, type...)
292		for e in self.getExportsList():
293		logger.debug(f"Init exported stats using the {e} module")
294		# @TODO: is it a good idea to thread this call ?
295		self._exports[e].init_fields(input_stats)
296		# In this first loop, data are not exported because some information are missing (rate for example)
297		logger.debug("Do not export stats during the first iteration because some information are missing")
298		self.first_export = False
299		return False
300
301		input_stats = input_stats or {}
302
303		for e in self.getExportsList():
304		logger.debug(f"Export stats using the {e} module")
305		thread = threading.Thread(target=self._exports[e].update, args=(input_stats,))
306		thread.start()
307
308		return True
309
310		def getAll(self):
311		"""Return all the stats (list).
312		This method is called byt the XML/RPC API.
313		It should return all the plugins (enable or not) because filtering can be done by the client.
314		"""
315		return [self._plugins[p].get_raw() for p in self.getPluginsList(enable=False)]
316
317		def getAllAsDict(self, plugin_list=None):
318		"""Return all the stats (as dict).
319		This method is called by the RESTFul API.
320		"""
321		if plugin_list is None:
322		# All enabled plugins should be exported
323		plugin_list = self.getPluginsList()
324		return {p: self._plugins[p].get_raw() for p in plugin_list}
325
326		def getAllFieldsDescription(self):
327		"""Return all fields description (as list)."""
328		return [self._plugins[p].fields_description for p in self.getPluginsList(enable=False)]
329
330		def getAllFieldsDescriptionAsDict(self, plugin_list=None):
331		"""Return all fields description (as dict)."""
332		if plugin_list is None:
333		# All enabled plugins should be exported
334		plugin_list = self.getPluginsList()
335		return {p: self._plugins[p].fields_description for p in plugin_list}
336
337		def getAllExports(self, plugin_list=None):
338		"""Return all the stats to be exported as a list.
339
340		Default behavior is to export all the stat
341		if plugin_list is provided (list), only export stats of given plugins
342		"""
343		if plugin_list is None:
344		# All enabled plugins should be exported
345		plugin_list = self.getPluginsList()
346		return [self._plugins[p].get_export() for p in plugin_list]
347
348		def getAllExportsAsDict(self, plugin_list=None):
349		"""Return all the stats to be exported as a dict.
350
351		Default behavior is to export all the stat
352		if plugin_list is provided (list), only export stats of given plugins
353		"""
354		if plugin_list is None:
355		# All enabled plugins should be exported
356		plugin_list = self.getPluginsList()
357		return {p: self._plugins[p].get_export() for p in plugin_list}
358
359		def getAllLimits(self, plugin_list=None):
360		"""Return the plugins limits list.
361
362		Default behavior is to export all the limits
363		if plugin_list is provided, only export limits of given plugin (list)
364		"""
365		if plugin_list is None:
366		# All enabled plugins should be exported
367		plugin_list = self.getPluginsList()
368		return [self._plugins[p].limits for p in plugin_list]
369
370		def getAllLimitsAsDict(self, plugin_list=None):
371		"""Return all the stats limits (dict).
372
373		Default behavior is to export all the limits
374		if plugin_list is provided, only export limits of given plugin (list)
375		"""
376		if plugin_list is None:
377		# All enabled plugins should be exported
378		plugin_list = self.getPluginsList()
379		return {p: self._plugins[p].limits for p in plugin_list}
380
381		def getAllViews(self, plugin_list=None):
382		"""Return the plugins views.
383		This method is called byt the XML/RPC API.
384		It should return all the plugins views (enable or not) because filtering can be done by the client.
385		"""
386		if plugin_list is None:
387		plugin_list = self.getPluginsList(enable=False)
388		return [self._plugins[p].get_views() for p in plugin_list]
389
390		def getAllViewsAsDict(self, plugin_list=None):
391		"""Return all the stats views (dict).
392		This method is called by the RESTFul API.
393		"""
394		if plugin_list is None:
395		# All enabled plugins should be exported
396		plugin_list = self.getPluginsList()
397		return {p: self._plugins[p].get_views() for p in plugin_list}
398
399		def get_plugin(self, plugin_name):
400		"""Return the plugin stats."""
401		if plugin_name in self._plugins:
402		return self._plugins[plugin_name]
403		return None
404
405		def get_plugin_view(self, plugin_name):
406		"""Return the plugin views."""
407		if plugin_name in self._plugins:
408		return self._plugins[plugin_name].get_views()
409		return None
410
411		def end(self):
412		"""End of the Glances stats."""
413		# Close export modules
414		for e in self._exports:
415		self._exports[e].exit()
416		# Close plugins
417		for p in self._plugins:
418		self._plugins[p].exit()
419

nicolargo / glances

GlancesStats.load_additional_plugins() F last analyzed 2025-06-14 08:50 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

GlancesStats.load_additional_plugins() F
last analyzed 2025-06-14 08:50 UTC