glances.plugins.system.PluginModel.update() - Code Metrics - Inspection of "version 4.2.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( cc9054...a8608f )

by Nicolas

created 2024-10-20 16:02 UTC

glances.plugins.system.PluginModel.update() A

↳ Parent: glances.plugins.system

Complexity

Conditions

Size

Total Lines	28
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

"""System plugin."""

import builtins
import os
import platform
import re

from glances.globals import iteritems
from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel

# {
#   "os_name": "Linux",
#   "hostname": "XPS13-9333",
#   "platform": "64bit",
#   "linux_distro": "Ubuntu 22.04",
#   "os_version": "5.15.0-88-generic",
#   "hr_name": "Ubuntu 22.04 64bit"
# }
# Fields description
# description: human readable description
# short_name: shortname to use un UI
# unit: unit type
# rate: is it a rate ? If yes, // by time_since_update when displayed,
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'os_name': {
        'description': 'Operating system name',
    },
    'hostname': {
        'description': 'Hostname',
    },
    'platform': {
        'description': 'Platform (32 or 64 bits)',
    },
    'linux_distro': {
        'description': 'Linux distribution',
    },
    'os_version': {
        'description': 'Operating system version',
    },
    'hr_name': {
        'description': 'Human readable operating system name',
    },
}

# SNMP OID
snmp_oid = {
    'default': {'hostname': '1.3.6.1.2.1.1.5.0', 'system_name': '1.3.6.1.2.1.1.1.0'},
    'netapp': {
        'hostname': '1.3.6.1.2.1.1.5.0',
        'system_name': '1.3.6.1.2.1.1.1.0',
        'platform': '1.3.6.1.4.1.789.1.1.5.0',
    },
}

# SNMP to human read
# Dict (key: OS short name) of dict (reg exp OID to human)
# Windows:
# http://msdn.microsoft.com/en-us/library/windows/desktop/ms724832%28v=vs.85%29.aspx
snmp_to_human = {
    'windows': {
        'Windows Version 10.0': 'Windows 10|11 or Server 2016|2019|2022',
        'Windows Version 6.3': 'Windows 8.1 or Server 2012R2',
        'Windows Version 6.2': 'Windows 8 or Server 2012',
        'Windows Version 6.1': 'Windows 7 or Server 2008R2',
        'Windows Version 6.0': 'Windows Vista or Server 2008',
        'Windows Version 5.2': 'Windows XP 64bits or 2003 server',
        'Windows Version 5.1': 'Windows XP',
        'Windows Version 5.0': 'Windows 2000',
    }
}


def _linux_os_release():
    """Try to determine the name of a Linux distribution.

    This function checks for the /etc/os-release file.
    It takes the name from the 'NAME' field and the version from 'VERSION_ID'.
    An empty string is returned if the above values cannot be determined.
    """
    pretty_name = ''
    ashtray = {}
    keys = ['NAME', 'VERSION_ID']
    try:
        with builtins.open(os.path.join('/etc', 'os-release')) as f:
            for line in f:
                for key in keys:
                    if line.startswith(key):
                        ashtray[key] = re.sub(r'^"|"$', '', line.strip().split('=')[1])
    except OSError:
        return pretty_name

    if ashtray:
        if 'NAME' in ashtray:
            pretty_name = ashtray['NAME']
        if 'VERSION_ID' in ashtray:
            pretty_name += ' {}'.format(ashtray['VERSION_ID'])

    return pretty_name


class PluginModel(GlancesPluginModel):
    """Glances' host/system plugin.

    stats is a dict
    """

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super().__init__(args=args, config=config, fields_description=fields_description)

        # We want to display the stat in the curse interface
        self.display_curse = True

        # Set default rate to 60 seconds
        if self.get_refresh():
            self.set_refresh(60)

        # Get the default message (if defined)
        self.system_info_msg = config.get_value('system', 'system_info_msg') if config else None

    def update_stats_with_snmp(self):
        try:
            stats = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name])
        except KeyError:
            stats = self.get_stats_snmp(snmp_oid=snmp_oid['default'])

        # Default behavior: display all the information
        stats['os_name'] = stats['system_name']

        # Windows OS tips
        if self.short_system_name == 'windows':
            for key, value in iteritems(snmp_to_human['windows']):
                if re.search(key, stats['system_name']):
                    stats['os_name'] = value
                    break

        return stats

    def add_human_readable_name(self, stats):
        if self.system_info_msg:
            try:
                hr_name = self.system_info_msg.format(**stats)
            except KeyError as e:
                logger.debug(f'Error in system_info_msg ({e})')
                hr_name = '{os_name} {os_version} {platform}'.format(**stats)
        elif stats['os_name'] == "Linux":
            hr_name = '{linux_distro} {platform} / {os_name} {os_version}'.format(**stats)
        else:
            hr_name = '{os_name} {os_version} {platform}'.format(**stats)
        return hr_name

    def get_win_version_and_platform(self, stats):
        os_version = platform.win32_ver()
        # if the python version is 32 bit perhaps the windows operating
        # system is 64bit
        conditions = [stats['platform'] == '32bit', 'PROCESSOR_ARCHITEW6432' in os.environ]

        return {'os_version': ' '.join(os_version[::2]), 'platform': '64bit' if all(conditions) else stats['platform']}

    def get_linux_version_and_distro(self):
        try:
            linux_distro = platform.linux_distribution()
        except AttributeError:
            distro = _linux_os_release()
        else:
            if linux_distro[0] == '':
                distro = _linux_os_release()
            else:
                distro = ' '.join(linux_distro[:2])

        return {'os_version': platform.release(), 'linux_distro': distro}

    def get_stats_from_std_sys_lib(self, stats):
        stats['os_name'] = platform.system()
        stats['hostname'] = platform.node()
        stats['platform'] = platform.architecture()[0]
        if stats['os_name'] == "Linux":
            stats.update(self.get_linux_version_and_distro())
        elif stats['os_name'].endswith('BSD') or stats['os_name'] == 'SunOS':
            stats['os_version'] = platform.release()
        elif stats['os_name'] == "Darwin":
            stats['os_version'] = platform.mac_ver()[0]
        elif stats['os_name'] == "Windows":
            stats.update(self.get_win_version_and_platform(stats))
        else:
            stats['os_version'] = ""

        return stats

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update the host/system info using the input method.

        :return: the stats dict
        """
        # Init new stats
        stats = self.get_init_value()

        if self.input_method == 'local':
            # Update stats using the standard system library
            stats = self.get_stats_from_std_sys_lib(stats)

            # Add human readable name
            stats['hr_name'] = self.add_human_readable_name(stats)

        elif self.input_method == 'snmp':
            # Update stats using SNMP
            stats = self.update_stats_with_snmp()

            # Add human readable name
            stats['hr_name'] = stats['os_name']

        # Update the stats
        self.stats = stats

        return self.stats

    def msg_curse(self, args=None, max_width=None):
        """Return the string to display in the curse interface."""
        # Init the return message
        ret = []

        # Only process if stats exist and plugin not disabled
        if not self.stats or self.is_disabled():
            return ret

        # Build the string message
        if args.client:
            # Client mode
            if args.cs_status.lower() == "connected":
                msg = 'Connected to '
                ret.append(self.curse_add_line(msg, 'OK'))
            elif args.cs_status.lower() == "snmp":
                msg = 'SNMP from '
                ret.append(self.curse_add_line(msg, 'OK'))
            elif args.cs_status.lower() == "disconnected":
                msg = 'Disconnected from '
                ret.append(self.curse_add_line(msg, 'CRITICAL'))

        # Hostname is mandatory
        msg = self.stats['hostname']
        ret.append(self.curse_add_line(msg, "TITLE"))

        # System info
        msg = ' ' + self.stats['hr_name']
        ret.append(self.curse_add_line(msg, optional=True))

        # Return the message with decoration
        return ret


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			"""System plugin."""
10
11			import builtins
12			import os
13			import platform
14			import re
15
16			from glances.globals import iteritems
17			from glances.logger import logger
18			from glances.plugins.plugin.model import GlancesPluginModel
19
20			# {
21			# "os_name": "Linux",
22			# "hostname": "XPS13-9333",
23			# "platform": "64bit",
24			# "linux_distro": "Ubuntu 22.04",
25			# "os_version": "5.15.0-88-generic",
26			# "hr_name": "Ubuntu 22.04 64bit"
27			# }
28			# Fields description
29			# description: human readable description
30			# short_name: shortname to use un UI
31			# unit: unit type
32			# rate: is it a rate ? If yes, // by time_since_update when displayed,
33			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
34			fields_description = {
35			'os_name': {
36			'description': 'Operating system name',
37			},
38			'hostname': {
39			'description': 'Hostname',
40			},
41			'platform': {
42			'description': 'Platform (32 or 64 bits)',
43			},
44			'linux_distro': {
45			'description': 'Linux distribution',
46			},
47			'os_version': {
48			'description': 'Operating system version',
49			},
50			'hr_name': {
51			'description': 'Human readable operating system name',
52			},
53			}
54
55			# SNMP OID
56			snmp_oid = {
57			'default': {'hostname': '1.3.6.1.2.1.1.5.0', 'system_name': '1.3.6.1.2.1.1.1.0'},
58			'netapp': {
59			'hostname': '1.3.6.1.2.1.1.5.0',
60			'system_name': '1.3.6.1.2.1.1.1.0',
61			'platform': '1.3.6.1.4.1.789.1.1.5.0',
62			},
63			}
64
65			# SNMP to human read
66			# Dict (key: OS short name) of dict (reg exp OID to human)
67			# Windows:
68			# http://msdn.microsoft.com/en-us/library/windows/desktop/ms724832%28v=vs.85%29.aspx
69			snmp_to_human = {
70			'windows': {
71			'Windows Version 10.0': 'Windows 10\|11 or Server 2016\|2019\|2022',
72			'Windows Version 6.3': 'Windows 8.1 or Server 2012R2',
73			'Windows Version 6.2': 'Windows 8 or Server 2012',
74			'Windows Version 6.1': 'Windows 7 or Server 2008R2',
75			'Windows Version 6.0': 'Windows Vista or Server 2008',
76			'Windows Version 5.2': 'Windows XP 64bits or 2003 server',
77			'Windows Version 5.1': 'Windows XP',
78			'Windows Version 5.0': 'Windows 2000',
79			}
80			}
81
82
83			def _linux_os_release():
84			"""Try to determine the name of a Linux distribution.
85
86			This function checks for the /etc/os-release file.
87			It takes the name from the 'NAME' field and the version from 'VERSION_ID'.
88			An empty string is returned if the above values cannot be determined.
89			"""
90			pretty_name = ''
91			ashtray = {}
92			keys = ['NAME', 'VERSION_ID']
93			try:
94			with builtins.open(os.path.join('/etc', 'os-release')) as f:
95			for line in f:
96			for key in keys:
97			if line.startswith(key):
98			ashtray[key] = re.sub(r'^"\|"$', '', line.strip().split('=')[1])
99			except OSError:
100			return pretty_name
101
102			if ashtray:
103			if 'NAME' in ashtray:
104			pretty_name = ashtray['NAME']
105			if 'VERSION_ID' in ashtray:
106			pretty_name += ' {}'.format(ashtray['VERSION_ID'])
107
108			return pretty_name
109
110
111			class PluginModel(GlancesPluginModel):
112			"""Glances' host/system plugin.
113
114			stats is a dict
115			"""
116
117			def __init__(self, args=None, config=None):
118			"""Init the plugin."""
119			super().__init__(args=args, config=config, fields_description=fields_description)
120
121			# We want to display the stat in the curse interface
122			self.display_curse = True
123
124			# Set default rate to 60 seconds
125			if self.get_refresh():
126			self.set_refresh(60)
127
128			# Get the default message (if defined)
129			self.system_info_msg = config.get_value('system', 'system_info_msg') if config else None
130
131			def update_stats_with_snmp(self):
132			try:
133			stats = self.get_stats_snmp(snmp_oid=snmp_oid[self.short_system_name])
134			except KeyError:
135			stats = self.get_stats_snmp(snmp_oid=snmp_oid['default'])
136
137			# Default behavior: display all the information
138			stats['os_name'] = stats['system_name']
139
140			# Windows OS tips
141			if self.short_system_name == 'windows':
142			for key, value in iteritems(snmp_to_human['windows']):
143			if re.search(key, stats['system_name']):
144			stats['os_name'] = value
145			break
146
147			return stats
148
149			def add_human_readable_name(self, stats):
150			if self.system_info_msg:
151			try:
152			hr_name = self.system_info_msg.format(**stats)
153			except KeyError as e:
154			logger.debug(f'Error in system_info_msg ({e})')
155			hr_name = '{os_name} {os_version} {platform}'.format(**stats)
156			elif stats['os_name'] == "Linux":
157			hr_name = '{linux_distro} {platform} / {os_name} {os_version}'.format(**stats)
158			else:
159			hr_name = '{os_name} {os_version} {platform}'.format(**stats)
160			return hr_name
161
162			def get_win_version_and_platform(self, stats):
163			os_version = platform.win32_ver()
164			# if the python version is 32 bit perhaps the windows operating
165			# system is 64bit
166			conditions = [stats['platform'] == '32bit', 'PROCESSOR_ARCHITEW6432' in os.environ]
167
168			return {'os_version': ' '.join(os_version[::2]), 'platform': '64bit' if all(conditions) else stats['platform']}
169
170			def get_linux_version_and_distro(self):
171			try:
172			linux_distro = platform.linux_distribution()
173			except AttributeError:
174			distro = _linux_os_release()
175			else:
176			if linux_distro[0] == '':
177			distro = _linux_os_release()
178			else:
179			distro = ' '.join(linux_distro[:2])
180
181			return {'os_version': platform.release(), 'linux_distro': distro}
182
183			def get_stats_from_std_sys_lib(self, stats):
184			stats['os_name'] = platform.system()
185			stats['hostname'] = platform.node()
186			stats['platform'] = platform.architecture()[0]
187			if stats['os_name'] == "Linux":
188			stats.update(self.get_linux_version_and_distro())
189			elif stats['os_name'].endswith('BSD') or stats['os_name'] == 'SunOS':
190			stats['os_version'] = platform.release()
191			elif stats['os_name'] == "Darwin":
192			stats['os_version'] = platform.mac_ver()[0]
193			elif stats['os_name'] == "Windows":
194			stats.update(self.get_win_version_and_platform(stats))
195			else:
196			stats['os_version'] = ""
197
198			return stats
199
200			@GlancesPluginModel._check_decorator
201			@GlancesPluginModel._log_result_decorator
202			def update(self):
203			"""Update the host/system info using the input method.
204
205			:return: the stats dict
206			"""
207			# Init new stats
208			stats = self.get_init_value()
209
210			if self.input_method == 'local':
211			# Update stats using the standard system library
212			stats = self.get_stats_from_std_sys_lib(stats)
213
214			# Add human readable name
215			stats['hr_name'] = self.add_human_readable_name(stats)
216
217			elif self.input_method == 'snmp':
218			# Update stats using SNMP
219			stats = self.update_stats_with_snmp()
220
221			# Add human readable name
222			stats['hr_name'] = stats['os_name']
223
224			# Update the stats
225			self.stats = stats
226
227			return self.stats
228
229			def msg_curse(self, args=None, max_width=None):
230			"""Return the string to display in the curse interface."""
231			# Init the return message
232			ret = []
233
234			# Only process if stats exist and plugin not disabled
235			if not self.stats or self.is_disabled():
236			return ret
237
238			# Build the string message
239			if args.client:
240			# Client mode
241			if args.cs_status.lower() == "connected":
242			msg = 'Connected to '
243			ret.append(self.curse_add_line(msg, 'OK'))
244			elif args.cs_status.lower() == "snmp":
245			msg = 'SNMP from '
246			ret.append(self.curse_add_line(msg, 'OK'))
247			elif args.cs_status.lower() == "disconnected":
248			msg = 'Disconnected from '
249			ret.append(self.curse_add_line(msg, 'CRITICAL'))
250
251			# Hostname is mandatory
252			msg = self.stats['hostname']
253			ret.append(self.curse_add_line(msg, "TITLE"))
254
255			# System info
256			msg = ' ' + self.stats['hr_name']
257			ret.append(self.curse_add_line(msg, optional=True))
258
259			# Return the message with decoration
260			return ret
261

nicolargo / glances

Push — master ( cc9054...a8608f )

glances.plugins.system.PluginModel.update() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like