glances.plugins.system._linux_os_release() - Code Metrics - Inspection of "Merge pull request #2884 from ariel-anieli/issue-2..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 6b036e...1153d1 )

by Nicolas

created 2024-07-24 13:45 UTC

glances.plugins.system._linux_os_release() C

↳ Parent: glances.plugins.system

Complexity

Conditions

Size

Total Lines	26
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	18
nop	0
dl	0
loc	26
rs	6.6666
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 = [
167			stats['platform'] == '32bit',
168			'PROCESSOR_ARCHITEW6432' in os.environ
169			]
170
171			return {
172			'os_version' : ' '.join(os_version[::2]),
173			'platform' : '64bit' if all(conditions) else stats['platform']
174			}
175
176			def get_linux_version_and_distro(self):
177			try:
178			linux_distro = platform.linux_distribution()
179			except AttributeError:
180			distro = _linux_os_release()
181			else:
182			if linux_distro[0] == '':
183			distro = _linux_os_release()
184			else:
185			distro = ' '.join(linux_distro[:2])
186
187			return {
188			'os_version' : platform.release(),
189			'linux_distro' : distro
190			}
191
192			def get_stats_from_std_sys_lib(self, stats):
193			stats['os_name'] = platform.system()
194			stats['hostname'] = platform.node()
195			stats['platform'] = platform.architecture()[0]
196			if stats['os_name'] == "Linux":
197			stats.update(self.get_linux_version_and_distro())
198			elif stats['os_name'].endswith('BSD') or stats['os_name'] == 'SunOS':
199			stats['os_version'] = platform.release()
200			elif stats['os_name'] == "Darwin":
201			stats['os_version'] = platform.mac_ver()[0]
202			elif stats['os_name'] == "Windows":
203			stats.update(self.get_win_version_and_platform(stats))
204			else:
205			stats['os_version'] = ""
206
207			return stats
208
209			@GlancesPluginModel._check_decorator
210			@GlancesPluginModel._log_result_decorator
211			def update(self):
212			"""Update the host/system info using the input method.
213
214			:return: the stats dict
215			"""
216			# Init new stats
217			stats = self.get_init_value()
218
219			if self.input_method == 'local':
220			# Update stats using the standard system library
221			stats = self.get_stats_from_std_sys_lib(stats)
222
223			# Add human readable name
224			stats['hr_name'] = self.add_human_readable_name(stats)
225
226			elif self.input_method == 'snmp':
227			# Update stats using SNMP
228			stats = self.update_stats_with_snmp()
229
230			# Add human readable name
231			stats['hr_name'] = stats['os_name']
232
233			# Update the stats
234			self.stats = stats
235
236			return self.stats
237
238			def msg_curse(self, args=None, max_width=None):
239			"""Return the string to display in the curse interface."""
240			# Init the return message
241			ret = []
242
243			# Only process if stats exist and plugin not disabled
244			if not self.stats or self.is_disabled():
245			return ret
246
247			# Build the string message
248			if args.client:
249			# Client mode
250			if args.cs_status.lower() == "connected":
251			msg = 'Connected to '
252			ret.append(self.curse_add_line(msg, 'OK'))
253			elif args.cs_status.lower() == "snmp":
254			msg = 'SNMP from '
255			ret.append(self.curse_add_line(msg, 'OK'))
256			elif args.cs_status.lower() == "disconnected":
257			msg = 'Disconnected from '
258			ret.append(self.curse_add_line(msg, 'CRITICAL'))
259
260			# Hostname is mandatory
261			msg = self.stats['hostname']
262			ret.append(self.curse_add_line(msg, "TITLE"))
263
264			# System info
265			msg = ' ' + self.stats['hr_name']
266			ret.append(self.curse_add_line(msg, optional=True))
267
268			# Return the message with decoration
269			return ret
270

nicolargo / glances

Push — develop ( 6b036e...1153d1 )

glances.plugins.system._linux_os_release() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like