glances.exports.export.GlancesExport.__init__() - Code Metrics - Inspection of "version 4.4.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 73a01d...ef36eb )

by Nicolas

created 2025-11-02 16:07 UTC

glances.exports.export.GlancesExport.init() A

↳ Parent: glances.exports.export

Complexity

Conditions

Size

Total Lines	27
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	12
nop	3
dl	0
loc	27
rs	9.8
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.
"""

import re

from glances.globals import NoOptionError, NoSectionError, 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

        # Load the default common export configuration
        if self.config is not None:
            self.load_common_conf()

    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_common_conf(self):
        """Load the common export configuration in the Glances configuration file.

        :returns: Boolean -- True if section is found
        """
        # Read the common [export] section
        section = "export"

        opt = "exclude_fields"
        try:
            setattr(self, opt, self.config.get_list_value(section, opt))
        except NoOptionError:
            logger.debug(f"{opt} option not found in the {section} configuration section")

        logger.debug(f"Load common {section} from the Glances configuration file")

        return True

    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:
                logger.debug(f"{opt} option not found in the {section} configuration section")

        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(":", 1) 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 is_excluded(self, field):
        """Return true if the field is excluded."""
        return any(re.fullmatch(i, field, re.I) for i in (getattr(self, 'exclude_fields') or ()))

    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: if needed this class can be overwritten.
        """
        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])
                    # Remove the <plugin>_disable field
                    i.pop(f"{plugin}_disable", None)
            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.
        This method builds two lists: names and values.
        """

        # Initialize export lists
        export_names = []
        export_values = []

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

nicolargo / glances

Push — master ( 73a01d...ef36eb )

glances.exports.export.GlancesExport.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

glances.exports.export.GlancesExport.init() A