glances.plugins.vms.VmsPlugin.msg_curse() - Code Metrics - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

glances.plugins.vms.VmsPlugin.msg_curse() F
last analyzed 2025-07-05 08:48 UTC

↳ Parent: glances.plugins.vms

Complexity

Conditions

Size

Total Lines	108
Code Lines	80

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	23
eloc	80
nop	3
dl	0
loc	108
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, Optional

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 as MultipassVmExtension
from glances.plugins.vms.engines.virsh import VmExtension as VirshVmExtension
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',
    },
    'cpu_time': {
        'description': 'Vm CPU time',
        'rate': True,
        'unit': 'percent',
    },
    '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',
    },
    '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',
    'cpu_time': 'CPU time',
    'memory_usage': 'memory consumption',
    'load_1min': 'load',
    'name': 'VM name',
    None: 'None',
}


class VmsPlugin(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'] = MultipassVmExtension()

        # Init the Virsh API
        self.watchers['virsh'] = VirshVmExtension()

        # 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 not all_tag:
            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 = self.update_local()

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

    @GlancesPluginModel._manage_rate
    def update_local(self):
        """Update stats localy"""
        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)
        return 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 length
        # 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']:
            # Get the maximum engine length
            engine_max_width = max(len(max(self.stats, key=lambda x: len(x['engine']))['engine']), 8)
            msg = ' {:{width}}'.format('Engine', width=engine_max_width)
            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 = '{:>6}'.format('CPU%')
        ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'cpu_time' 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=engine_max_width)))

            # 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')))
            # CPU (time)
            try:
                msg = '{:>6}'.format(vm['cpu_time_rate_per_sec'])
            except (KeyError, TypeError):
                msg = '{:>6}'.format('-')
            ret.append(
                self.curse_add_line(
                    msg, self.get_views(item=vm['name'], key='cpu_time_rate_per_sec', 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 = 'cpu_time'
    elif glances_processes.sort_key == 'name':
        sort_by = 'name'
        sort_by_secondary = 'load_1min'
    else:
        sort_by = 'cpu_time'
        sort_by_secondary = 'load_1min'

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

nicolargo / glances

glances.plugins.vms.VmsPlugin.msg_curse() F last analyzed 2025-07-05 08:48 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

glances.plugins.vms.VmsPlugin.msg_curse() F
last analyzed 2025-07-05 08:48 UTC