glances.plugins.vms.PluginModel.msg_curse() - 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.vms.PluginModel.msg_curse() F

↳ Parent: glances.plugins.vms

Complexity

Conditions

Size

Total Lines	93
Code Lines	70

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	20
eloc	70
nop	3
dl	0
loc	93
rs	0
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
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
        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"))
        if len(self.stats) > 1:
            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
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		self.watchers['multipass'] = VmExtension()
113
114		# Sort key
115		self.sort_key = None
116
117		def get_key(self) -> str:
118		"""Return the key of the list."""
119		return 'name'
120
121		def get_export(self) -> List[Dict]:
122		"""Overwrite the default export method.
123
124		- Only exports vms
125		- The key is the first vm name
126		"""
127		try:
128		ret = deepcopy(self.stats)
129		except KeyError as e:
130		logger.debug(f"vm plugin - Vm export error {e}")
131		ret = []
132
133		# Remove fields uses to compute rate
134		for vm in ret:
135		for i in export_exclude_list:
136		vm.pop(i)
137
138		return ret
139
140		def _all_tag(self) -> bool:
141		"""Return the all tag of the Glances/Vm configuration file.
142
143		# By default, Glances only display running vms
144		# Set the following key to True to display all vms
145		all=True
146		"""
147		all_tag = self.get_conf_value('all')
148		if len(all_tag) == 0:
149		return False
150		return all_tag[0].lower() == 'true'
151
152		@GlancesPluginModel._check_decorator
153		@GlancesPluginModel._log_result_decorator
154		def update(self) -> List[Dict]:
155		"""Update VMs stats using the input method."""
156		# Connection should be ok
157		if not self.watchers or self.input_method != 'local':
158		return self.get_init_value()
159
160		# Update stats
161		stats = []
162		for engine, watcher in iteritems(self.watchers):
163		version, vms = watcher.update(all_tag=self._all_tag())
164		for vm in vms:
165		vm["engine"] = engine
166		vm["engine_version"] = version
167		stats.extend(vms)
168
169		# Sort and update the stats
170		# TODO: test
171		self.sort_key, self.stats = sort_vm_stats(stats)
172		return self.stats
173
174		def update_views(self) -> bool:
175		"""Update stats views."""
176		# Call the father's method
177		super().update_views()
178
179		if not self.stats:
180		return False
181
182		# Display Engine ?
183		show_engine_name = False
184		if len({ct["engine"] for ct in self.stats}) > 1:
185		show_engine_name = True
186		self.views['show_engine_name'] = show_engine_name
187
188		return True
189
190		def msg_curse(self, args=None, max_width: Optional[int] = None) -> List[str]:
191		"""Return the dict to display in the curse interface."""
192		# Init the return message
193		ret = []
194
195		# Only process if stats exist (and non null) and display plugin enable...
196		if not self.stats or len(self.stats) == 0 or self.is_disabled():
197		return ret
198
199		# Build the string message
200		# Title
201		msg = '{}'.format('VMs')
202		ret.append(self.curse_add_line(msg, "TITLE"))
203		if len(self.stats) > 1:
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 — master ( cc9054...a8608f )

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