glances.exports.export.GlancesExport.plugins_to_export() - Code Metrics - Inspection of "Export individual processes stats #794" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 24a3ae...5f48d5 )

by Nicolas

created 2024-04-06 15:48 UTC

GlancesExport.plugins_to_export() A

↳ Parent: glances.exports.export

Complexity

Conditions

Size

Total Lines	7
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	2
dl	0
loc	7
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
#

"""
I am your father...

...for all Glances exports IF.
"""

from glances.globals import json_dumps

from glances.globals import NoOptionError, NoSectionError, iteritems, iterkeys
from glances.logger import logger


class GlancesExport(object):

    """Main class for Glances export IF."""

    # List of non exportable plugins
    # @TODO: remove this part and make all plugins exportable (see issue #1556)
    # @TODO: also make this list configurable by the user (see issue #1443)
    non_exportable_plugins = [
        'alert',
        'amps',
        'help',
        'now',
        'plugin',
        'ports',
        # 'processlist',
        'psutilversion',
        'quicklook',
        'version',
    ]

    def __init__(self, config=None, args=None):
        """Init the export class."""
        # Export name
        self.export_name = self.__class__.__module__
        logger.debug("Init export module %s" % self.export_name)

        # Init the config & args
        self.config = config
        self.args = args

        # By default export is disabled
        # Needs to be set to True in the __init__ class of child
        self.export_enable = False

        # Mandatory for (most of) the export module
        self.host = None
        self.port = None

        # Save last export list
        self._last_exported_list = None

    def exit(self):
        """Close the export module."""
        logger.debug("Finalise export interface %s" % self.export_name)

    def load_conf(self, section, mandatories=['host', 'port'], options=None):
        """Load the export <section> configuration in the Glances configuration file.

        :param section: name of the export section to load
        :param mandatories: a list of mandatory parameters to load
        :param options: a list of optional parameters to load

        :returns: Boolean -- True if section is found
        """
        options = options or []

        if self.config is None:
            return False

        # By default read the mandatory host:port items
        try:
            for opt in mandatories:
                setattr(self, opt, self.config.get_value(section, opt))
        except NoSectionError:
            logger.error("No {} configuration found".format(section))
            return False
        except NoOptionError as e:
            logger.error("Error in the {} configuration ({})".format(section, e))
            return False

        # Load options
        for opt in options:
            try:
                setattr(self, opt, self.config.get_value(section, opt))
            except NoOptionError:
                pass

        logger.debug("Load {} from the Glances configuration file".format(section))
        logger.debug("{} parameters: {}".format(section, {opt: getattr(self, opt) for opt in mandatories + options}))

        return True

    def get_item_key(self, item):
        """Return the value of the item 'key'."""
        ret = None
        try:
            ret = item[item['key']]
        except KeyError:
            logger.error("No 'key' available in {}".format(item))
        if isinstance(ret, list):
            return ret[0]
        else:
            return ret

    def parse_tags(self, tags):
        """Parse tags into a dict.

        :param tags: a comma-separated list of 'key:value' pairs. Example: foo:bar,spam:eggs
        :return: a dict of tags. Example: {'foo': 'bar', 'spam': 'eggs'}
        """
        d_tags = {}
        if tags:
            try:
                d_tags = dict([x.split(':') for x in tags.split(',')])
            except ValueError:
                # one of the 'key:value' pairs was missing
                logger.info('Invalid tags passed: %s', tags)
                d_tags = {}

        return d_tags

    def plugins_to_export(self, stats):
        """Return the list of plugins to export.

        :param stats: the stats object
        :return: a list of plugins to export
        """
        return [p for p in stats.getPluginsList() if p not in self.non_exportable_plugins]

    def last_exported_list(self):
        """Return the list of plugins last exported."""
        return self._last_exported_list

    def update(self, stats):
        """Update stats to a server.

        The method builds two lists: names and values and calls the export method to export the stats.

        Note: this class can be overwritten (for example in CSV and Graph).
        """
        if not self.export_enable:
            return False

        # Get all the stats & limits
        self._last_exported_list = self.plugins_to_export(stats)
        all_stats = stats.getAllExportsAsDict(plugin_list=self.last_exported_list())
        all_limits = stats.getAllLimitsAsDict(plugin_list=self.last_exported_list())

        # Loop over plugins to export
        for plugin in self.last_exported_list():
            if isinstance(all_stats[plugin], dict):
                all_stats[plugin].update(all_limits[plugin])
            elif isinstance(all_stats[plugin], list):
                # TypeError: string indices must be integers (Network plugin) #1054
                for i in all_stats[plugin]:
                    i.update(all_limits[plugin])
            else:
                continue
            export_names, export_values = self.build_export(all_stats[plugin])
            self.export(plugin, export_names, export_values)

        return True

    def build_export(self, stats):
        """Build the export lists."""
        export_names = []
        export_values = []

        if isinstance(stats, dict):
            # Stats is a dict
            # Is there a key ?
            if 'key' in iterkeys(stats) and stats['key'] in iterkeys(stats):
                pre_key = '{}.'.format(stats[stats['key']])
            else:
                pre_key = ''
            # Walk through the dict
            for key, value in iteritems(stats):
                if isinstance(value, bool):
                    value = json_dumps(value)
                if isinstance(value, list):
                    try:
                        value = value[0]
                    except IndexError:
                        value = ''
                if isinstance(value, dict):
                    item_names, item_values = self.build_export(value)
                    item_names = [pre_key + key.lower() + str(i) for i in item_names]
                    export_names += item_names
                    export_values += item_values
                else:
                    export_names.append(pre_key + key.lower())
                    export_values.append(value)
        elif isinstance(stats, list):
            # Stats is a list (of dict)
            # Recursive loop through the list
            for item in stats:
                item_names, item_values = self.build_export(item)
                export_names += item_names
                export_values += item_values
        return export_names, export_values

    def export(self, name, columns, points):
        # This method should be implemented by each exporter
        pass


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			"""
11			I am your father...
12
13			...for all Glances exports IF.
14			"""
15
16			from glances.globals import json_dumps
17
18			from glances.globals import NoOptionError, NoSectionError, iteritems, iterkeys
19			from glances.logger import logger
20
21
22			class GlancesExport(object):
23
24			"""Main class for Glances export IF."""
25
26			# List of non exportable plugins
27			# @TODO: remove this part and make all plugins exportable (see issue #1556)
28			# @TODO: also make this list configurable by the user (see issue #1443)
29			non_exportable_plugins = [
30			'alert',
31			'amps',
32			'help',
33			'now',
34			'plugin',
35			'ports',
36			# 'processlist',
37			'psutilversion',
38			'quicklook',
39			'version',
40			]
41
42			def __init__(self, config=None, args=None):
43			"""Init the export class."""
44			# Export name
45			self.export_name = self.__class__.__module__
46			logger.debug("Init export module %s" % self.export_name)
47
48			# Init the config & args
49			self.config = config
50			self.args = args
51
52			# By default export is disabled
53			# Needs to be set to True in the __init__ class of child
54			self.export_enable = False
55
56			# Mandatory for (most of) the export module
57			self.host = None
58			self.port = None
59
60			# Save last export list
61			self._last_exported_list = None
62
63			def exit(self):
64			"""Close the export module."""
65			logger.debug("Finalise export interface %s" % self.export_name)
66
67			def load_conf(self, section, mandatories=['host', 'port'], options=None):
68			"""Load the export <section> configuration in the Glances configuration file.
69
70			:param section: name of the export section to load
71			:param mandatories: a list of mandatory parameters to load
72			:param options: a list of optional parameters to load
73
74			:returns: Boolean -- True if section is found
75			"""
76			options = options or []
77
78			if self.config is None:
79			return False
80
81			# By default read the mandatory host:port items
82			try:
83			for opt in mandatories:
84			setattr(self, opt, self.config.get_value(section, opt))
85			except NoSectionError:
86			logger.error("No {} configuration found".format(section))
87			return False
88			except NoOptionError as e:
89			logger.error("Error in the {} configuration ({})".format(section, e))
90			return False
91
92			# Load options
93			for opt in options:
94			try:
95			setattr(self, opt, self.config.get_value(section, opt))
96			except NoOptionError:
97			pass
98
99			logger.debug("Load {} from the Glances configuration file".format(section))
100			logger.debug("{} parameters: {}".format(section, {opt: getattr(self, opt) for opt in mandatories + options}))
101
102			return True
103
104			def get_item_key(self, item):
105			"""Return the value of the item 'key'."""
106			ret = None
107			try:
108			ret = item[item['key']]
109			except KeyError:
110			logger.error("No 'key' available in {}".format(item))
111			if isinstance(ret, list):
112			return ret[0]
113			else:
114			return ret
115
116			def parse_tags(self, tags):
117			"""Parse tags into a dict.
118
119			:param tags: a comma-separated list of 'key:value' pairs. Example: foo:bar,spam:eggs
120			:return: a dict of tags. Example: {'foo': 'bar', 'spam': 'eggs'}
121			"""
122			d_tags = {}
123			if tags:
124			try:
125			d_tags = dict([x.split(':') for x in tags.split(',')])
126			except ValueError:
127			# one of the 'key:value' pairs was missing
128			logger.info('Invalid tags passed: %s', tags)
129			d_tags = {}
130
131			return d_tags
132
133			def plugins_to_export(self, stats):
134			"""Return the list of plugins to export.
135
136			:param stats: the stats object
137			:return: a list of plugins to export
138			"""
139			return [p for p in stats.getPluginsList() if p not in self.non_exportable_plugins]
140
141			def last_exported_list(self):
142			"""Return the list of plugins last exported."""
143			return self._last_exported_list
144
145			def update(self, stats):
146			"""Update stats to a server.
147
148			The method builds two lists: names and values and calls the export method to export the stats.
149
150			Note: this class can be overwritten (for example in CSV and Graph).
151			"""
152			if not self.export_enable:
153			return False
154
155			# Get all the stats & limits
156			self._last_exported_list = self.plugins_to_export(stats)
157			all_stats = stats.getAllExportsAsDict(plugin_list=self.last_exported_list())
158			all_limits = stats.getAllLimitsAsDict(plugin_list=self.last_exported_list())
159
160			# Loop over plugins to export
161			for plugin in self.last_exported_list():
162			if isinstance(all_stats[plugin], dict):
163			all_stats[plugin].update(all_limits[plugin])
164			elif isinstance(all_stats[plugin], list):
165			# TypeError: string indices must be integers (Network plugin) #1054
166			for i in all_stats[plugin]:
167			i.update(all_limits[plugin])
168			else:
169			continue
170			export_names, export_values = self.build_export(all_stats[plugin])
171			self.export(plugin, export_names, export_values)
172
173			return True
174
175			def build_export(self, stats):
176			"""Build the export lists."""
177			export_names = []
178			export_values = []
179
180			if isinstance(stats, dict):
181			# Stats is a dict
182			# Is there a key ?
183			if 'key' in iterkeys(stats) and stats['key'] in iterkeys(stats):
184			pre_key = '{}.'.format(stats[stats['key']])
185			else:
186			pre_key = ''
187			# Walk through the dict
188			for key, value in iteritems(stats):
189			if isinstance(value, bool):
190			value = json_dumps(value)
191			if isinstance(value, list):
192			try:
193			value = value[0]
194			except IndexError:
195			value = ''
196			if isinstance(value, dict):
197			item_names, item_values = self.build_export(value)
198			item_names = [pre_key + key.lower() + str(i) for i in item_names]
199			export_names += item_names
200			export_values += item_values
201			else:
202			export_names.append(pre_key + key.lower())
203			export_values.append(value)
204			elif isinstance(stats, list):
205			# Stats is a list (of dict)
206			# Recursive loop through the list
207			for item in stats:
208			item_names, item_values = self.build_export(item)
209			export_names += item_names
210			export_values += item_values
211			return export_names, export_values
212
213			def export(self, name, columns, points):
214			# This method should be implemented by each exporter
215			pass
216

nicolargo / glances

Push — develop ( 24a3ae...5f48d5 )

GlancesExport.plugins_to_export() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like