glances.exports.export.GlancesExport.load_conf() - Code Metrics - Inspection of "Merge branch 'issue3182-InfluxDB3' into develop" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 24eb6c...ad5e7f )

by Nicolas

created 2025-05-27 08:47 UTC

glances.exports.export.GlancesExport.load_conf() B

↳ Parent: glances.exports.export

Complexity

Conditions

Size

Total Lines	36
Code Lines	21

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	21
nop	4
dl	0
loc	36
rs	7.9759
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
#

"""
I am your father...
...for all Glances exports IF.
"""

from glances.globals import (
    NoOptionError,
    NoSectionError,
    iteritems,
    iterkeys,
    json_dumps,
)
from glances.logger import logger
from glances.timer import Counter


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

    # List of non exportable internal plugins
    non_exportable_plugins = [
        "alert",
        "help",
        "plugin",
        "psutilversion",
        "quicklook",
        "version",
    ]

    def __init__(self, config=None, args=None):
        """Init the export class."""
        # Export name
        self.export_name = self.__class__.__module__
        logger.debug(f"Init export module {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

        # Fields description
        self._fields_description = None

    def _log_result_decorator(fct):

        """Log (DEBUG) the result of the function fct."""

        def wrapper(*args, **kw):
            counter = Counter()
            ret = fct(*args, **kw)
            duration = counter.get()
            class_name = args[0].__class__.__name__
            class_module = args[0].__class__.__module__
            logger.debug(f"{class_name} {class_module} {fct.__name__} return {ret} in {duration} seconds")
            return ret

        return wrapper

    def exit(self):
        """Close the export module."""
        logger.debug(f"Finalise export interface {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(f"No {section} configuration found")
            return False
        except NoOptionError as e:
            logger.error(f"Error in the {section} configuration ({e})")
            return False

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

        logger.debug(f"Load {section} from the Glances configuration file")
        logger.debug(f"{section} parameters: { ({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(f"No 'key' available in {item}")
        if isinstance(ret, list):
            return ret[0]
        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 normalize_for_influxdb(self, name, columns, points):
        """Normalize data for the InfluxDB's data model.

        :return: a list of measurements.
        """
        FIELD_TO_TAG = ["name", "cmdline", "type"]
        ret = []

        # Build initial dict by crossing columns and point
        data_dict = dict(zip(columns, points))

        # issue1871 - Check if a key exist. If a key exist, the value of
        # the key should be used as a tag to identify the measurement.
        keys_list = [k.split(".")[0] for k in columns if k.endswith(".key")]
        if not keys_list:
            keys_list = [None]

        for measurement in keys_list:
            # Manage field
            if measurement is not None:
                fields = {
                    k.replace(f"{measurement}.", ""): data_dict[k] for k in data_dict if k.startswith(f"{measurement}.")
                }
            else:
                fields = data_dict
            # Transform to InfluxDB data model
            # https://docs.influxdata.com/influxdb/v1.8/write_protocols/line_protocol_reference/
            for k in fields:
                #  Do not export empty (None) value
                if fields[k] is None:
                    continue
                # Convert numerical to float
                try:
                    fields[k] = float(fields[k])
                except (TypeError, ValueError):
                    # Convert others to string
                    try:
                        fields[k] = str(fields[k])
                    except (TypeError, ValueError):
                        pass
            # Manage tags
            tags = self.parse_tags(self.tags)
            # Add the hostname as a tag
            tags["hostname"] = self.hostname
            if "hostname" in fields:
                fields.pop("hostname")
            # Others tags...
            if "key" in fields and fields["key"] in fields:
                # Create a tag from the key
                # Tag should be an string (see InfluxDB data model)
                tags[fields["key"]] = str(fields[fields["key"]])
                # Remove it from the field list (can not be a field and a tag)
                fields.pop(fields["key"])
            # Add name as a tag (example for the process list)
            for k in FIELD_TO_TAG:
                if k in fields:
                    tags[k] = str(fields[k])
                    # Remove it from the field list (can not be a field and a tag)
                    fields.pop(k)
            # Add the measurement to the list
            ret.append({"measurement": name, "tags": tags, "fields": fields})
        return ret

    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 init_fields(self, stats):
        """Return fields description in order to init stats in a server."""
        if not self.export_enable:
            return False

        self._last_exported_list = self.plugins_to_export(stats)
        self._fields_description = stats.getAllFieldsDescriptionAsDict(plugin_list=self.last_exported_list())
        return self._fields_description

    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])
                # Remove the <plugin>_disable field
                all_stats[plugin].pop(f"{plugin}_disable", None)
            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 sorted(iteritems(stats)):
                if isinstance(value, bool):
                    value = json_dumps(value).decode()

                if isinstance(value, list):
                    value = " ".join([str(v) for v in 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		#
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		"""
10		I am your father...
11		...for all Glances exports IF.
12		"""
13
14		from glances.globals import (
15		NoOptionError,
16		NoSectionError,
17		iteritems,
18		iterkeys,
19		json_dumps,
20		)
21		from glances.logger import logger
22		from glances.timer import Counter
23
24
25		class GlancesExport:
26		"""Main class for Glances export IF."""
27
28		# List of non exportable internal plugins
29		non_exportable_plugins = [
30		"alert",
31		"help",
32		"plugin",
33		"psutilversion",
34		"quicklook",
35		"version",
36		]
37
38		def __init__(self, config=None, args=None):
39		"""Init the export class."""
40		# Export name
41		self.export_name = self.__class__.__module__
42		logger.debug(f"Init export module {self.export_name}")
43
44		# Init the config & args
45		self.config = config
46		self.args = args
47
48		# By default export is disabled
49		# Needs to be set to True in the __init__ class of child
50		self.export_enable = False
51
52		# Mandatory for (most of) the export module
53		self.host = None
54		self.port = None
55
56		# Save last export list
57		self._last_exported_list = None
58
59		# Fields description
60		self._fields_description = None
61
62	View Code Duplication	def _log_result_decorator(fct):
		0 ignored issues – show Duplication introduced 2024-05-02 16:23 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
63		"""Log (DEBUG) the result of the function fct."""
64
65		def wrapper(args, *kw):
66		counter = Counter()
67		ret = fct(args, *kw)
68		duration = counter.get()
69		class_name = args[0].__class__.__name__
70		class_module = args[0].__class__.__module__
71		logger.debug(f"{class_name} {class_module} {fct.__name__} return {ret} in {duration} seconds")
72		return ret
73
74		return wrapper
75
76		def exit(self):
77		"""Close the export module."""
78		logger.debug(f"Finalise export interface {self.export_name}")
79
80		def load_conf(self, section, mandatories=["host", "port"], options=None):
81		"""Load the export <section> configuration in the Glances configuration file.
82
83		:param section: name of the export section to load
84		:param mandatories: a list of mandatory parameters to load
85		:param options: a list of optional parameters to load
86
87		:returns: Boolean -- True if section is found
88		"""
89		options = options or []
90
91		if self.config is None:
92		return False
93
94		# By default read the mandatory host:port items
95		try:
96		for opt in mandatories:
97		setattr(self, opt, self.config.get_value(section, opt))
98		except NoSectionError:
99		logger.error(f"No {section} configuration found")
100		return False
101		except NoOptionError as e:
102		logger.error(f"Error in the {section} configuration ({e})")
103		return False
104
105		# Load options
106		for opt in options:
107		try:
108		setattr(self, opt, self.config.get_value(section, opt))
109		except NoOptionError:
110		pass
111
112		logger.debug(f"Load {section} from the Glances configuration file")
113		logger.debug(f"{section} parameters: { ({opt: getattr(self, opt) for opt in mandatories + options}) }")
114
115		return True
116
117		def get_item_key(self, item):
118		"""Return the value of the item 'key'."""
119		ret = None
120		try:
121		ret = item[item["key"]]
122		except KeyError:
123		logger.error(f"No 'key' available in {item}")
124		if isinstance(ret, list):
125		return ret[0]
126		return ret
127
128		def parse_tags(self, tags):
129		"""Parse tags into a dict.
130
131		:param tags: a comma-separated list of 'key:value' pairs. Example: foo:bar,spam:eggs
132		:return: a dict of tags. Example: {'foo': 'bar', 'spam': 'eggs'}
133		"""
134		d_tags = {}
135		if tags:
136		try:
137		d_tags = dict([x.split(":") for x in tags.split(",")])
138		except ValueError:
139		# one of the 'key:value' pairs was missing
140		logger.info("Invalid tags passed: %s", tags)
141		d_tags = {}
142
143		return d_tags
144
145		def normalize_for_influxdb(self, name, columns, points):
146		"""Normalize data for the InfluxDB's data model.
147
148		:return: a list of measurements.
149		"""
150		FIELD_TO_TAG = ["name", "cmdline", "type"]
151		ret = []
152
153		# Build initial dict by crossing columns and point
154		data_dict = dict(zip(columns, points))
155
156		# issue1871 - Check if a key exist. If a key exist, the value of
157		# the key should be used as a tag to identify the measurement.
158		keys_list = [k.split(".")[0] for k in columns if k.endswith(".key")]
159		if not keys_list:
160		keys_list = [None]
161
162		for measurement in keys_list:
163		# Manage field
164		if measurement is not None:
165		fields = {
166		k.replace(f"{measurement}.", ""): data_dict[k] for k in data_dict if k.startswith(f"{measurement}.")
167		}
168		else:
169		fields = data_dict
170		# Transform to InfluxDB data model
171		# https://docs.influxdata.com/influxdb/v1.8/write_protocols/line_protocol_reference/
172		for k in fields:
173		# Do not export empty (None) value
174		if fields[k] is None:
175		continue
176		# Convert numerical to float
177		try:
178		fields[k] = float(fields[k])
179		except (TypeError, ValueError):
180		# Convert others to string
181		try:
182		fields[k] = str(fields[k])
183		except (TypeError, ValueError):
184		pass
185		# Manage tags
186		tags = self.parse_tags(self.tags)
187		# Add the hostname as a tag
188		tags["hostname"] = self.hostname
189		if "hostname" in fields:
190		fields.pop("hostname")
191		# Others tags...
192		if "key" in fields and fields["key"] in fields:
193		# Create a tag from the key
194		# Tag should be an string (see InfluxDB data model)
195		tags[fields["key"]] = str(fields[fields["key"]])
196		# Remove it from the field list (can not be a field and a tag)
197		fields.pop(fields["key"])
198		# Add name as a tag (example for the process list)
199		for k in FIELD_TO_TAG:
200		if k in fields:
201		tags[k] = str(fields[k])
202		# Remove it from the field list (can not be a field and a tag)
203		fields.pop(k)
204		# Add the measurement to the list
205		ret.append({"measurement": name, "tags": tags, "fields": fields})
206		return ret
207
208		def plugins_to_export(self, stats):
209		"""Return the list of plugins to export.
210
211		:param stats: the stats object
212		:return: a list of plugins to export
213		"""
214		return [p for p in stats.getPluginsList() if p not in self.non_exportable_plugins]
215
216		def last_exported_list(self):
217		"""Return the list of plugins last exported."""
218		return self._last_exported_list
219
220		def init_fields(self, stats):
221		"""Return fields description in order to init stats in a server."""
222		if not self.export_enable:
223		return False
224
225		self._last_exported_list = self.plugins_to_export(stats)
226		self._fields_description = stats.getAllFieldsDescriptionAsDict(plugin_list=self.last_exported_list())
227		return self._fields_description
228
229		def update(self, stats):
230		"""Update stats to a server.
231
232		The method builds two lists: names and values and calls the export method to export the stats.
233
234		Note: this class can be overwritten (for example in CSV and Graph).
235		"""
236		if not self.export_enable:
237		return False
238
239		# Get all the stats & limits
240		self._last_exported_list = self.plugins_to_export(stats)
241		all_stats = stats.getAllExportsAsDict(plugin_list=self.last_exported_list())
242		all_limits = stats.getAllLimitsAsDict(plugin_list=self.last_exported_list())
243
244		# Loop over plugins to export
245		for plugin in self.last_exported_list():
246		if isinstance(all_stats[plugin], dict):
247		all_stats[plugin].update(all_limits[plugin])
248		# Remove the <plugin>_disable field
249		all_stats[plugin].pop(f"{plugin}_disable", None)
250		elif isinstance(all_stats[plugin], list):
251		# TypeError: string indices must be integers (Network plugin) #1054
252		for i in all_stats[plugin]:
253		i.update(all_limits[plugin])
254		else:
255		continue
256		export_names, export_values = self.build_export(all_stats[plugin])
257		self.export(plugin, export_names, export_values)
258
259		return True
260
261		def build_export(self, stats):
262		"""Build the export lists."""
263		export_names = []
264		export_values = []
265
266		if isinstance(stats, dict):
267		# Stats is a dict
268		# Is there a key ?
269		if "key" in iterkeys(stats) and stats["key"] in iterkeys(stats):
270		pre_key = "{}.".format(stats[stats["key"]])
271		else:
272		pre_key = ""
273		# Walk through the dict
274		for key, value in sorted(iteritems(stats)):
275		if isinstance(value, bool):
276		value = json_dumps(value).decode()
277
278		if isinstance(value, list):
279		value = " ".join([str(v) for v in value])
280
281		if isinstance(value, dict):
282		item_names, item_values = self.build_export(value)
283		item_names = [pre_key + key.lower() + str(i) for i in item_names]
284		export_names += item_names
285		export_values += item_values
286		else:
287		export_names.append(pre_key + key.lower())
288		export_values.append(value)
289		elif isinstance(stats, list):
290		# Stats is a list (of dict)
291		# Recursive loop through the list
292		for item in stats:
293		item_names, item_values = self.build_export(item)
294		export_names += item_names
295		export_values += item_values
296		return export_names, export_values
297
298		def export(self, name, columns, points):
299		# This method should be implemented by each exporter
300		pass
301

nicolargo / glances

Push — develop ( 24eb6c...ad5e7f )

glances.exports.export.GlancesExport.load_conf() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like