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

nicolargo / glances

glances.plugins.vms.VmsPlugin.msg_curse() F last analyzed 2025-08-17 15:07 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-08-17 15:07 UTC