glances.exports.glances_influxdb.Export._normalize() - 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.glances_influxdb.Export._normalize() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	58
Code Lines	32

Duplication

Lines	58
Ratio	100 %

Importance

Changes

Metric	Value
cc	12
eloc	32
nop	4
dl	58
loc	58
rs	4.8
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
#

"""InfluxDB (up to InfluxDB 1.7.x) interface class."""

import sys
from platform import node

from influxdb import InfluxDBClient
from influxdb.client import InfluxDBClientError

from glances.exports.export import GlancesExport
from glances.logger import logger


class Export(GlancesExport):
    """This class manages the InfluxDB export module."""

    def __init__(self, config=None, args=None):
        """Init the InfluxDB export IF."""
        super().__init__(config=config, args=args)

        # Mandatory configuration keys (additional to host and port)
        self.user = None
        self.password = None
        self.db = None

        # Optional configuration keys
        self.protocol = "http"
        self.prefix = None
        self.tags = None
        self.hostname = None

        # Load the InfluxDB configuration file
        self.export_enable = self.load_conf(
            "influxdb",
            mandatories=["host", "port", "user", "password", "db"],
            options=["protocol", "prefix", "tags"],
        )
        if not self.export_enable:
            exit("Missing influxdb config")

        # The hostname is always add as a tag
        self.hostname = node().split(".")[0]

        # Init the InfluxDB client
        self.client = self.init()

    def init(self):
        """Init the connection to the InfluxDB server."""
        if not self.export_enable:
            return None

        # Correct issue #1530
        if self.protocol is not None and (self.protocol.lower() == "https"):
            ssl = True
        else:
            ssl = False

        try:
            db = InfluxDBClient(
                host=self.host,
                port=self.port,
                ssl=ssl,
                verify_ssl=False,
                username=self.user,
                password=self.password,
                database=self.db,
            )
            get_all_db = [i["name"] for i in db.get_list_database()]
        except InfluxDBClientError as e:
            logger.critical(f"Cannot connect to InfluxDB database '{self.db}' ({e})")
            sys.exit(2)

        if self.db in get_all_db:

            logger.info(f"Stats will be exported to InfluxDB server: {db._baseurl}")
        else:
            logger.critical(f"InfluxDB database '{self.db}' did not exist. Please create it")
            sys.exit(2)

        return db

    def export(self, name, columns, points):
        """Write the points to the InfluxDB server."""
        # Manage prefix
        if self.prefix is not None:
            name = self.prefix + "." + name
        # Write input to the InfluxDB database
        if not points:
            logger.debug(f"Cannot export empty {name} stats to InfluxDB")
        else:
            try:
                self.client.write_points(
                    self.normalize_for_influxdb(name, columns, points),
                    time_precision="s",
                )
            except Exception as e:
                # Log level set to warning instead of error (see: issue #1561)
                logger.warning(f"Cannot export {name} stats to InfluxDB ({e})")
            else:
                logger.debug(f"Export {name} stats to InfluxDB")


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			"""InfluxDB (up to InfluxDB 1.7.x) interface class."""
10
11			import sys
12			from platform import node
13
14			from influxdb import InfluxDBClient
15			from influxdb.client import InfluxDBClientError
16
17			from glances.exports.export import GlancesExport
18			from glances.logger import logger
19
20
21			class Export(GlancesExport):
22			"""This class manages the InfluxDB export module."""
23
24			def __init__(self, config=None, args=None):
25			"""Init the InfluxDB export IF."""
26			super().__init__(config=config, args=args)
27
28			# Mandatory configuration keys (additional to host and port)
29			self.user = None
30			self.password = None
31			self.db = None
32
33			# Optional configuration keys
34			self.protocol = "http"
35			self.prefix = None
36			self.tags = None
37			self.hostname = None
38
39			# Load the InfluxDB configuration file
40			self.export_enable = self.load_conf(
41			"influxdb",
42			mandatories=["host", "port", "user", "password", "db"],
43			options=["protocol", "prefix", "tags"],
44			)
45			if not self.export_enable:
46			exit("Missing influxdb config")
47
48			# The hostname is always add as a tag
49			self.hostname = node().split(".")[0]
50
51			# Init the InfluxDB client
52			self.client = self.init()
53
54			def init(self):
55			"""Init the connection to the InfluxDB server."""
56			if not self.export_enable:
57			return None
58
59			# Correct issue #1530
60			if self.protocol is not None and (self.protocol.lower() == "https"):
61			ssl = True
62			else:
63			ssl = False
64
65			try:
66			db = InfluxDBClient(
67			host=self.host,
68			port=self.port,
69			ssl=ssl,
70			verify_ssl=False,
71			username=self.user,
72			password=self.password,
73			database=self.db,
74			)
75			get_all_db = [i["name"] for i in db.get_list_database()]
76			except InfluxDBClientError as e:
77			logger.critical(f"Cannot connect to InfluxDB database '{self.db}' ({e})")
78			sys.exit(2)
79
80			if self.db in get_all_db:
			0 ignored issues – show introduced 2018-11-11 15:28 UTC by Report Bug Copy Issue Report The variable `get_all_db` does not seem to be defined for all execution paths. Loading history...
81			logger.info(f"Stats will be exported to InfluxDB server: {db._baseurl}")
82			else:
83			logger.critical(f"InfluxDB database '{self.db}' did not exist. Please create it")
84			sys.exit(2)
85
86			return db
87
88			def export(self, name, columns, points):
89			"""Write the points to the InfluxDB server."""
90			# Manage prefix
91			if self.prefix is not None:
92			name = self.prefix + "." + name
93			# Write input to the InfluxDB database
94			if not points:
95			logger.debug(f"Cannot export empty {name} stats to InfluxDB")
96			else:
97			try:
98			self.client.write_points(
99			self.normalize_for_influxdb(name, columns, points),
100			time_precision="s",
101			)
102			except Exception as e:
103			# Log level set to warning instead of error (see: issue #1561)
104			logger.warning(f"Cannot export {name} stats to InfluxDB ({e})")
105			else:
106			logger.debug(f"Export {name} stats to InfluxDB")
107

nicolargo / glances

Push — develop ( 24eb6c...ad5e7f )

glances.exports.glances_influxdb.Export._normalize() D

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like