glances.plugins.vms.PluginModel.msg_curse() - Code Metrics - Inspection of "beta04" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 1153d1...e38e2c )

by Nicolas

created 2024-07-24 13:53 UTC

glances.plugins.vms.PluginModel.msg_curse() F

↳ Parent: glances.plugins.vms

Complexity

Conditions

Size

Total Lines	92
Code Lines	69

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	19
eloc	69
nop	3
dl	0
loc	92
rs	0.5999
c	0
b	0
f	0

How to fix Long Method Complexity

#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""Vms plugin."""

from copy import deepcopy
from typing import Any, Dict, List, Optional, Tuple

from glances.globals import iteritems
from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel
from glances.plugins.vms.engines import VmsExtension
from glances.plugins.vms.engines.multipass import VmExtension, import_multipass_error_tag
from glances.processes import glances_processes
from glances.processes import sort_stats as sort_stats_processes

# 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 = {
    'name': {
        'description': 'Vm name',
    },
    'id': {
        'description': 'Vm ID',
    },
    'release': {
        'description': 'Vm release',
    },
    'status': {
        'description': 'Vm status',
    },
    'cpu_count': {
        'description': 'Vm CPU count',
    },
    'memory_usage': {
        'description': 'Vm memory usage',
        'unit': 'byte',
    },
    'memory_total': {
        'description': 'Vm memory total',
        'unit': 'byte',
    },
    'load_1min': {
        'description': 'Vm Load last 1 min',
    },
    'load_5min': {
        'description': 'Vm Load last 5 mins',
    },
    'load_15min': {
        'description': 'Vm Load last 15 mins',
    },
    'ipv4': {
        'description': 'Vm IP v4 address',
    },
    'engine': {
        'description': 'VM engine name (only Mutlipass is currently supported)',
    },
    'engine_version': {
        'description': 'VM engine version',
    },
}

# Define the items history list (list of items to add to history)
items_history_list = [{'name': 'memory_usage', 'description': 'Vm MEM usage', 'y_unit': 'byte'}]

# List of key to remove before export
export_exclude_list = []

# Sort dictionary for human
sort_for_human = {
    'cpu_count': 'CPU count',
    'memory_usage': 'memory consumption',
    'load_1min': 'load',
    'name': 'VM name',
    None: 'None',
}


class PluginModel(GlancesPluginModel):
    """Glances Vm plugin.

    stats is a dict: {'version': {...}, 'vms': [{}, {}]}
    """

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

        # The plugin can be disabled using: args.disable_vm
        self.args = args

        # Default config keys
        self.config = config

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

        self.watchers: Dict[str, VmsExtension] = {}

        # Init the Multipass API
        if not import_multipass_error_tag:
            self.watchers['multipass'] = VmExtension()

        # Sort key
        self.sort_key = None

    def get_key(self) -> str:
        """Return the key of the list."""
        return 'name'

    def get_export(self) -> List[Dict]:
        """Overwrite the default export method.

        - Only exports vms
        - The key is the first vm name
        """
        try:
            ret = deepcopy(self.stats)
        except KeyError as e:
            logger.debug(f"vm plugin - Vm export error {e}")
            ret = []

        # Remove fields uses to compute rate
        for vm in ret:
            for i in export_exclude_list:
                vm.pop(i)

        return ret

    def _all_tag(self) -> bool:
        """Return the all tag of the Glances/Vm configuration file.

        # By default, Glances only display running vms
        # Set the following key to True to display all vms
        all=True
        """
        all_tag = self.get_conf_value('all')
        if len(all_tag) == 0:
            return False
        return all_tag[0].lower() == 'true'

    @GlancesPluginModel._check_decorator

    @GlancesPluginModel._log_result_decorator
    def update(self) -> List[Dict]:
        """Update VMs stats using the input method."""
        # Connection should be ok
        if not self.watchers or self.input_method != 'local':
            return self.get_init_value()

        # Update stats
        stats = []
        for engine, watcher in iteritems(self.watchers):
            version, vms = watcher.update(all_tag=self._all_tag())
            for vm in vms:
                vm["engine"] = engine
                vm["engine_version"] = version
            stats.extend(vms)

        # Sort and update the stats
        # TODO: test
        self.sort_key, self.stats = sort_vm_stats(stats)
        return self.stats

    def update_views(self) -> bool:
        """Update stats views."""
        # Call the father's method
        super().update_views()

        if not self.stats:
            return False

        # Display Engine ?
        show_engine_name = False
        if len({ct["engine"] for ct in self.stats}) > 1:
            show_engine_name = True
        self.views['show_engine_name'] = show_engine_name

        return True

    def msg_curse(self, args=None, max_width: Optional[int] = None) -> List[str]:
        """Return the dict to display in the curse interface."""
        # Init the return message
        ret = []

        # Only process if stats exist (and non null) and display plugin enable...
        if not self.stats or len(self.stats) == 0 or self.is_disabled():
            return ret

        # Build the string message
        # Title
        msg = '{}'.format('VMs')
        ret.append(self.curse_add_line(msg, "TITLE"))
        msg = f' {len(self.stats)}'
        ret.append(self.curse_add_line(msg))
        msg = f' sorted by {sort_for_human[self.sort_key]}'
        ret.append(self.curse_add_line(msg))
        if not self.views['show_engine_name']:
            msg = f' (served by {self.stats[0].get("engine", "")})'
        ret.append(self.curse_add_line(msg))
        ret.append(self.curse_new_line())
        # Header
        ret.append(self.curse_new_line())
        # Get the maximum VMs name
        # Max size is configurable. See feature request #1723.
        name_max_width = min(
            self.config.get_int_value('vms', 'max_name_size', default=20) if self.config is not None else 20,
            len(max(self.stats, key=lambda x: len(x['name']))['name']),
        )

        if self.views['show_engine_name']:
            msg = ' {:{width}}'.format('Engine', width=8)
            ret.append(self.curse_add_line(msg))
        msg = ' {:{width}}'.format('Name', width=name_max_width)
        ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'name' else 'DEFAULT'))
        msg = '{:>10}'.format('Status')
        ret.append(self.curse_add_line(msg))
        msg = '{:>6}'.format('Core')
        ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'cpu_count' else 'DEFAULT'))
        msg = '{:>7}'.format('MEM')
        ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'memory_usage' else 'DEFAULT'))
        msg = '/{:<7}'.format('MAX')
        ret.append(self.curse_add_line(msg))
        msg = '{:>17}'.format('LOAD 1/5/15min')
        ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'load_1min' else 'DEFAULT'))
        msg = '{:>10}'.format('Release')
        ret.append(self.curse_add_line(msg))

        # Data
        for vm in self.stats:
            ret.append(self.curse_new_line())
            if self.views['show_engine_name']:
                ret.append(self.curse_add_line(' {:{width}}'.format(vm["engine"], width=8)))
            # Name
            ret.append(self.curse_add_line(' {:{width}}'.format(vm['name'][:name_max_width], width=name_max_width)))
            # Status
            status = self.vm_alert(vm['status'])
            msg = '{:>10}'.format(vm['status'][0:10])
            ret.append(self.curse_add_line(msg, status))
            # CPU (count)
            try:
                msg = '{:>6}'.format(vm['cpu_count'])
            except (KeyError, TypeError):
                msg = '{:>6}'.format('-')
            ret.append(self.curse_add_line(msg, self.get_views(item=vm['name'], key='cpu_count', option='decoration')))
            # MEM
            try:
                msg = '{:>7}'.format(self.auto_unit(vm['memory_usage']))
            except KeyError:
                msg = '{:>7}'.format('-')
            ret.append(
                self.curse_add_line(msg, self.get_views(item=vm['name'], key='memory_usage', option='decoration'))
            )
            try:
                msg = '/{:<7}'.format(self.auto_unit(vm['memory_total']))
            except (KeyError, TypeError):
                msg = '/{:<7}'.format('-')
            ret.append(self.curse_add_line(msg))
            # LOAD
            try:
                msg = '{:>5.1f}/{:>5.1f}/{:>5.1f}'.format(vm['load_1min'], vm['load_5min'], vm['load_15min'])
            except (KeyError, TypeError):
                msg = '{:>5}/{:>5}/{:>5}'.format('-', '-', '-')
            ret.append(self.curse_add_line(msg, self.get_views(item=vm['name'], key='load_1min', option='decoration')))
            # Release
            if vm['release'] is not None:
                msg = '   {}'.format(vm['release'])
            else:
                msg = '   {}'.format('-')
            ret.append(self.curse_add_line(msg, splittable=True))

        return ret

    @staticmethod
    def vm_alert(status: str) -> str:
        """Analyse the vm status.
        For multipass: https://multipass.run/docs/instance-states
        """
        if status == 'running':
            return 'OK'
        if status in ['starting', 'restarting', 'delayed shutdown']:
            return 'WARNING'
        return 'INFO'


def sort_vm_stats(stats: List[Dict[str, Any]]) -> Tuple[str, List[Dict[str, Any]]]:

    # Make VM sort related to process sort
    if glances_processes.sort_key == 'memory_percent':
        sort_by = 'memory_usage'
        sort_by_secondary = 'load_1min'
    elif glances_processes.sort_key == 'name':
        sort_by = 'name'
        sort_by_secondary = 'load_1min'
    else:
        sort_by = 'load_1min'
        sort_by_secondary = 'memory_usage'

    # Sort vm stats
    sort_stats_processes(
        stats,
        sorted_by=sort_by,
        sorted_by_secondary=sort_by_secondary,
        # Reverse for all but name
        reverse=glances_processes.sort_key != 'name',
    )

    # Return the main sort key and the sorted stats
    return sort_by, stats


1		#
2		# This file is part of Glances.
3		#
4		# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
5		#
6		# SPDX-License-Identifier: LGPL-3.0-only
7		#
8
9		"""Vms plugin."""
10
11		from copy import deepcopy
12		from typing import Any, Dict, List, Optional, Tuple
13
14		from glances.globals import iteritems
15		from glances.logger import logger
16		from glances.plugins.plugin.model import GlancesPluginModel
17		from glances.plugins.vms.engines import VmsExtension
18		from glances.plugins.vms.engines.multipass import VmExtension, import_multipass_error_tag
19		from glances.processes import glances_processes
20		from glances.processes import sort_stats as sort_stats_processes
21
22		# Fields description
23		# description: human readable description
24		# short_name: shortname to use un UI
25		# unit: unit type
26		# rate: is it a rate ? If yes, // by time_since_update when displayed,
27		# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
28		fields_description = {
29		'name': {
30		'description': 'Vm name',
31		},
32		'id': {
33		'description': 'Vm ID',
34		},
35		'release': {
36		'description': 'Vm release',
37		},
38		'status': {
39		'description': 'Vm status',
40		},
41		'cpu_count': {
42		'description': 'Vm CPU count',
43		},
44		'memory_usage': {
45		'description': 'Vm memory usage',
46		'unit': 'byte',
47		},
48		'memory_total': {
49		'description': 'Vm memory total',
50		'unit': 'byte',
51		},
52		'load_1min': {
53		'description': 'Vm Load last 1 min',
54		},
55		'load_5min': {
56		'description': 'Vm Load last 5 mins',
57		},
58		'load_15min': {
59		'description': 'Vm Load last 15 mins',
60		},
61		'ipv4': {
62		'description': 'Vm IP v4 address',
63		},
64		'engine': {
65		'description': 'VM engine name (only Mutlipass is currently supported)',
66		},
67		'engine_version': {
68		'description': 'VM engine version',
69		},
70		}
71
72		# Define the items history list (list of items to add to history)
73		items_history_list = [{'name': 'memory_usage', 'description': 'Vm MEM usage', 'y_unit': 'byte'}]
74
75		# List of key to remove before export
76		export_exclude_list = []
77
78		# Sort dictionary for human
79		sort_for_human = {
80		'cpu_count': 'CPU count',
81		'memory_usage': 'memory consumption',
82		'load_1min': 'load',
83		'name': 'VM name',
84		None: 'None',
85		}
86
87
88		class PluginModel(GlancesPluginModel):
89		"""Glances Vm plugin.
90
91		stats is a dict: {'version': {...}, 'vms': [{}, {}]}
92		"""
93
94		def __init__(self, args=None, config=None):
95		"""Init the plugin."""
96		super().__init__(
97		args=args, config=config, items_history_list=items_history_list, fields_description=fields_description
98		)
99
100		# The plugin can be disabled using: args.disable_vm
101		self.args = args
102
103		# Default config keys
104		self.config = config
105
106		# We want to display the stat in the curse interface
107		self.display_curse = True
108
109		self.watchers: Dict[str, VmsExtension] = {}
110
111		# Init the Multipass API
112		if not import_multipass_error_tag:
113		self.watchers['multipass'] = VmExtension()
114
115		# Sort key
116		self.sort_key = None
117
118		def get_key(self) -> str:
119		"""Return the key of the list."""
120		return 'name'
121
122		def get_export(self) -> List[Dict]:
123		"""Overwrite the default export method.
124
125		- Only exports vms
126		- The key is the first vm name
127		"""
128		try:
129		ret = deepcopy(self.stats)
130		except KeyError as e:
131		logger.debug(f"vm plugin - Vm export error {e}")
132		ret = []
133
134		# Remove fields uses to compute rate
135		for vm in ret:
136		for i in export_exclude_list:
137		vm.pop(i)
138
139		return ret
140
141		def _all_tag(self) -> bool:
142		"""Return the all tag of the Glances/Vm configuration file.
143
144		# By default, Glances only display running vms
145		# Set the following key to True to display all vms
146		all=True
147		"""
148		all_tag = self.get_conf_value('all')
149		if len(all_tag) == 0:
150		return False
151		return all_tag[0].lower() == 'true'
152
153	View Code Duplication	@GlancesPluginModel._check_decorator
		0 ignored issues – show Duplication introduced 2024-07-24 13:56 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
154		@GlancesPluginModel._log_result_decorator
155		def update(self) -> List[Dict]:
156		"""Update VMs stats using the input method."""
157		# Connection should be ok
158		if not self.watchers or self.input_method != 'local':
159		return self.get_init_value()
160
161		# Update stats
162		stats = []
163		for engine, watcher in iteritems(self.watchers):
164		version, vms = watcher.update(all_tag=self._all_tag())
165		for vm in vms:
166		vm["engine"] = engine
167		vm["engine_version"] = version
168		stats.extend(vms)
169
170		# Sort and update the stats
171		# TODO: test
172		self.sort_key, self.stats = sort_vm_stats(stats)
173		return self.stats
174
175		def update_views(self) -> bool:
176		"""Update stats views."""
177		# Call the father's method
178		super().update_views()
179
180		if not self.stats:
181		return False
182
183		# Display Engine ?
184		show_engine_name = False
185		if len({ct["engine"] for ct in self.stats}) > 1:
186		show_engine_name = True
187		self.views['show_engine_name'] = show_engine_name
188
189		return True
190
191		def msg_curse(self, args=None, max_width: Optional[int] = None) -> List[str]:
192		"""Return the dict to display in the curse interface."""
193		# Init the return message
194		ret = []
195
196		# Only process if stats exist (and non null) and display plugin enable...
197		if not self.stats or len(self.stats) == 0 or self.is_disabled():
198		return ret
199
200		# Build the string message
201		# Title
202		msg = '{}'.format('VMs')
203		ret.append(self.curse_add_line(msg, "TITLE"))
204		msg = f' {len(self.stats)}'
205		ret.append(self.curse_add_line(msg))
206		msg = f' sorted by {sort_for_human[self.sort_key]}'
207		ret.append(self.curse_add_line(msg))
208		if not self.views['show_engine_name']:
209		msg = f' (served by {self.stats[0].get("engine", "")})'
210		ret.append(self.curse_add_line(msg))
211		ret.append(self.curse_new_line())
212		# Header
213		ret.append(self.curse_new_line())
214		# Get the maximum VMs name
215		# Max size is configurable. See feature request #1723.
216		name_max_width = min(
217		self.config.get_int_value('vms', 'max_name_size', default=20) if self.config is not None else 20,
218		len(max(self.stats, key=lambda x: len(x['name']))['name']),
219		)
220
221		if self.views['show_engine_name']:
222		msg = ' {:{width}}'.format('Engine', width=8)
223		ret.append(self.curse_add_line(msg))
224		msg = ' {:{width}}'.format('Name', width=name_max_width)
225		ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'name' else 'DEFAULT'))
226		msg = '{:>10}'.format('Status')
227		ret.append(self.curse_add_line(msg))
228		msg = '{:>6}'.format('Core')
229		ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'cpu_count' else 'DEFAULT'))
230		msg = '{:>7}'.format('MEM')
231		ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'memory_usage' else 'DEFAULT'))
232		msg = '/{:<7}'.format('MAX')
233		ret.append(self.curse_add_line(msg))
234		msg = '{:>17}'.format('LOAD 1/5/15min')
235		ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'load_1min' else 'DEFAULT'))
236		msg = '{:>10}'.format('Release')
237		ret.append(self.curse_add_line(msg))
238
239		# Data
240		for vm in self.stats:
241		ret.append(self.curse_new_line())
242		if self.views['show_engine_name']:
243		ret.append(self.curse_add_line(' {:{width}}'.format(vm["engine"], width=8)))
244		# Name
245		ret.append(self.curse_add_line(' {:{width}}'.format(vm['name'][:name_max_width], width=name_max_width)))
246		# Status
247		status = self.vm_alert(vm['status'])
248		msg = '{:>10}'.format(vm['status'][0:10])
249		ret.append(self.curse_add_line(msg, status))
250		# CPU (count)
251		try:
252		msg = '{:>6}'.format(vm['cpu_count'])
253		except (KeyError, TypeError):
254		msg = '{:>6}'.format('-')
255		ret.append(self.curse_add_line(msg, self.get_views(item=vm['name'], key='cpu_count', option='decoration')))
256		# MEM
257		try:
258		msg = '{:>7}'.format(self.auto_unit(vm['memory_usage']))
259		except KeyError:
260		msg = '{:>7}'.format('-')
261		ret.append(
262		self.curse_add_line(msg, self.get_views(item=vm['name'], key='memory_usage', option='decoration'))
263		)
264		try:
265		msg = '/{:<7}'.format(self.auto_unit(vm['memory_total']))
266		except (KeyError, TypeError):
267		msg = '/{:<7}'.format('-')
268		ret.append(self.curse_add_line(msg))
269		# LOAD
270		try:
271		msg = '{:>5.1f}/{:>5.1f}/{:>5.1f}'.format(vm['load_1min'], vm['load_5min'], vm['load_15min'])
272		except (KeyError, TypeError):
273		msg = '{:>5}/{:>5}/{:>5}'.format('-', '-', '-')
274		ret.append(self.curse_add_line(msg, self.get_views(item=vm['name'], key='load_1min', option='decoration')))
275		# Release
276		if vm['release'] is not None:
277		msg = ' {}'.format(vm['release'])
278		else:
279		msg = ' {}'.format('-')
280		ret.append(self.curse_add_line(msg, splittable=True))
281
282		return ret
283
284		@staticmethod
285		def vm_alert(status: str) -> str:
286		"""Analyse the vm status.
287		For multipass: https://multipass.run/docs/instance-states
288		"""
289		if status == 'running':
290		return 'OK'
291		if status in ['starting', 'restarting', 'delayed shutdown']:
292		return 'WARNING'
293		return 'INFO'
294
295
296	View Code Duplication	def sort_vm_stats(stats: List[Dict[str, Any]]) -> Tuple[str, List[Dict[str, Any]]]:
		0 ignored issues – show Duplication introduced 2024-07-24 13:56 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
297		# Make VM sort related to process sort
298		if glances_processes.sort_key == 'memory_percent':
299		sort_by = 'memory_usage'
300		sort_by_secondary = 'load_1min'
301		elif glances_processes.sort_key == 'name':
302		sort_by = 'name'
303		sort_by_secondary = 'load_1min'
304		else:
305		sort_by = 'load_1min'
306		sort_by_secondary = 'memory_usage'
307
308		# Sort vm stats
309		sort_stats_processes(
310		stats,
311		sorted_by=sort_by,
312		sorted_by_secondary=sort_by_secondary,
313		# Reverse for all but name
314		reverse=glances_processes.sort_key != 'name',
315		)
316
317		# Return the main sort key and the sorted stats
318		return sort_by, stats
319

nicolargo / glances

Push — develop ( 1153d1...e38e2c )

glances.plugins.vms.PluginModel.msg_curse() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like