First version of the Curses UI, miss engine name

conf/glances.conf

@@ -461,6 +461,14 @@ port_default_gateway=True
 #web_4_url=https://blog.nicolargo.com/nonexist
 #web_4_description=Intranet
 
+[vms]
+disable=False
+# Define the maximum VMs size name (default is 20 chars)
+max_name_size=20
+# By default, Glances only display running VMs
+# Set the following key to True to display all VMs
+all=False
+
 [containers]
 disable=False
 # Only show specific containers (comma-separated list of container name or regular expression)

glances/main.py

@@ -692,6 +692,7 @@ Examples of use:
             disable(args, 'alert')
             disable(args, 'amps')
             disable(args, 'containers')
+            disable(args, 'vms')
 
         # Manage full quicklook option
         if getattr(args, 'full_quicklook', False):

glances/outputs/glances_curses.py

@@ -84,6 +84,7 @@ class _GlancesCurses:
         'T': {'switch': 'network_sum'},
         'u': {'sort_key': 'username'},
         'U': {'switch': 'network_cumul'},
+        'V': {'switch': 'disable_vms'},
         'w': {'handler': '_handle_clean_logs'},
         'W': {'switch': 'disable_wifi'},
         'x': {'handler': '_handle_clean_critical_logs'},
@@ -124,7 +125,7 @@ class _GlancesCurses:
     _left_sidebar_max_width = 34
 
     # Define right sidebar
-    _right_sidebar = ['containers', 'processcount', 'amps', 'processlist', 'alert']
+    _right_sidebar = ['vms', 'containers', 'processcount', 'amps', 'processlist', 'alert']
 
     def __init__(self, config=None, args=None):
         # Init

glances/plugins/containers/engines/__init__.py

@@ -1,3 +1,11 @@
+#
+# This file is part of Glances.
+#
+# SPDX-FileCopyrightText: 2024 Nicolas Hennion <nicolas@nicolargo.com>
+#
+# SPDX-License-Identifier: LGPL-3.0-only
+#
+
 from typing import Any, Dict, Protocol, Tuple
 
 

glances/plugins/containers/engines/docker.py

@@ -13,7 +13,7 @@ from typing import Any, Dict, List, Optional, Tuple
 
 from glances.globals import iterkeys, itervalues, nativestr, pretty_date, replace_special_chars
 from glances.logger import logger
-from glances.plugins.containers.stats_streamer import ThreadedIterableStreamer
+from glances.stats_streamer import ThreadedIterableStreamer
 
 # Docker-py library (optional and Linux-only)
 # https://github.com/docker/docker-py

glances/plugins/containers/engines/podman.py

@@ -13,7 +13,7 @@ from typing import Any, Dict, Optional, Tuple
 
 from glances.globals import iterkeys, itervalues, nativestr, pretty_date, replace_special_chars, string_value_to_float
 from glances.logger import logger
-from glances.plugins.containers.stats_streamer import ThreadedIterableStreamer
+from glances.stats_streamer import ThreadedIterableStreamer
 
 # Podman library (optional and Linux-only)
 # https://pypi.org/project/podman/

glances/plugins/vms/__init__.py

@@ -0,0 +1,341 @@
+#
+# This file is part of Glances.
+#
+# SPDX-FileCopyrightText: 2024 Nicolas Hennion <nicolas@nicolargo.com>
+#
+# 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',
+    },
+}
+
+# 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',
+    '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())
+            # print(engine, version, vms)
+            for vm in vms:
+                vm["engine"] = 'vm'
+            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
+
+        # Add specifics information
+        # Alert
+        # TODO
+        # for i in self.stats:
+        #     # Init the views for the current vm (key = vm name)
+        #     self.views[i[self.get_key()]] = {'cpu': {}, 'mem': {}}
+        #     # CPU alert
+        #     if 'cpu' in i and 'total' in i['cpu']:
+        #         # Looking for specific CPU vm threshold in the conf file
+        #         alert = self.get_alert(i['cpu']['total'], header=i['name'] + '_cpu', action_key=i['name'])
+        #         if alert == 'DEFAULT':
+        #             # Not found ? Get back to default CPU threshold value
+        #             alert = self.get_alert(i['cpu']['total'], header='cpu')
+        #         self.views[i[self.get_key()]]['cpu']['decoration'] = alert
+        #     # MEM alert
+        #     if 'memory' in i and 'usage' in i['memory']:
+        #         # Looking for specific MEM vm threshold in the conf file
+        #         alert = self.get_alert(
+        #             self.memory_usage_no_cache(i['memory']),
+        #             maximum=i['memory']['limit'],
+        #             header=i['name'] + '_mem',
+        #             action_key=i['name'],
+        #         )
+        #         if alert == 'DEFAULT':
+        #             # Not found ? Get back to default MEM threshold value
+        #             alert = self.get_alert(
+        #                 self.memory_usage_no_cache(i['memory']), maximum=i['memory']['limit'], header='mem'
+        #             )
+        #         self.views[i[self.get_key()]]['mem']['decoration'] = alert
+
+        # 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))
+        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=6)
+            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('CPU')
+        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=6)))
+            # 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.1f}'.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 'INFO'
+        if status in ['stopped', 'deleted', 'suspending', 'suspended']:
+            return 'CRITICAL'
+        return 'CAREFUL'
+
+
+def sort_vm_stats(stats: List[Dict[str, Any]]) -> Tuple[str, List[Dict[str, Any]]]:
+    # Sort Vm stats using the same function than processes
+    sort_by = glances_processes.sort_key
+    if sort_by == 'cpu_percent':
+        sort_by = 'cpu_count'
+        sort_by_secondary = 'memory_usage'
+    elif sort_by == 'memory_percent':
+        sort_by = 'memory_usage'
+        sort_by_secondary = 'cpu_count'
+    elif sort_by in ['username', 'io_counters', 'cpu_times']:
+        sort_by = 'cpu_count'
+        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

glances/plugins/vms/engines/__init__.py

@@ -0,0 +1,17 @@
+#
+# This file is part of Glances.
+#
+# SPDX-FileCopyrightText: 2024 Nicolas Hennion <nicolas@nicolargo.com>
+#
+# SPDX-License-Identifier: LGPL-3.0-only
+#
+
+from typing import Any, Dict, Protocol, Tuple
+
+
+class VmsExtension(Protocol):
+    def stop(self) -> None:
+        raise NotImplementedError
+
+    def update(self, all_tag) -> Tuple[Dict, list[Dict[str, Any]]]:
+        raise NotImplementedError

glances/plugins/vms/engines/multipass.py

@@ -0,0 +1,118 @@
+#
+# This file is part of Glances.
+#
+# SPDX-FileCopyrightText: 2024 Nicolas Hennion <nicolas@nicolargo.com>
+#
+# SPDX-License-Identifier: LGPL-3.0-only
+#
+
+"""Multipass Extension unit for Glances' Vms plugin."""
+
+import os
+from typing import Any, Dict, List, Tuple
+
+import orjson
+
+from glances.globals import nativestr
+from glances.secure import secure_popen
+
+# Check if multipass binary exist
+# TODO: make this path configurable from the Glances configuration file
+MULTIPASS_PATH = '/snap/bin/multipass'
+MULTIPASS_VERSION_OPTIONS = 'version --format json'
+MULTIPASS_INFO_OPTIONS = 'info --format json'
+import_multipass_error_tag = not os.path.exists(MULTIPASS_PATH)
+
+
+class VmExtension:
+    """Glances' Vms Plugin's Vm Extension unit"""
+
+    CONTAINER_ACTIVE_STATUS = ['running']
+
+    def __init__(self):
+        if import_multipass_error_tag:
+            raise Exception(f"Multipass binary ({MULTIPASS_PATH})is mandatory to get Vm stats")
+
+        self.ext_name = "Multipass (Vm)"
+
+    def update_version(self):
+        # > multipass version --format json
+        # {
+        #     "multipass": "1.13.1",
+        #     "multipassd": "1.13.1"
+        # }
+        return orjson.loads(secure_popen(f'{MULTIPASS_PATH} {MULTIPASS_VERSION_OPTIONS}'))
+
+    def update_info(self):
+        # > multipass info  --format json
+        # {
+        #     "errors": [
+        #     ],
+        #     "info": {
+        #         "adapted-budgerigar": {
+        #             "cpu_count": "1",
+        #             "disks": {
+        #                 "sda1": {
+        #                     "total": "5116440064",
+        #                     "used": "2287162880"
+        #                 }
+        #             },
+        #             "image_hash": "182dc760bfca26c45fb4e4668049ecd4d0ecdd6171b3bae81d0135e8f1e9d93e",
+        #             "image_release": "24.04 LTS",
+        #             "ipv4": [
+        #                 "10.160.166.174"
+        #             ],
+        #             "load": [
+        #                 0,
+        #                 0.03,
+        #                 0
+        #             ],
+        #             "memory": {
+        #                 "total": 1002500096,
+        #                 "used": 432058368
+        #             },
+        #             "mounts": {
+        #             },
+        #             "release": "Ubuntu 24.04 LTS",
+        #             "snapshot_count": "0",
+        #             "state": "Running"
+        #         }
+        #     }
+        # }
+        return orjson.loads(secure_popen(f'{MULTIPASS_PATH} {MULTIPASS_INFO_OPTIONS}')).get('info')
+
+    def update(self, all_tag) -> Tuple[Dict, List[Dict]]:
+        """Update Vm stats using the input method."""
+        version_stats = self.update_version()
+
+        # TODO: manage all_tag option
+        info_stats = self.update_info()
+
+        returned_stats = []
+        for k, v in info_stats.items():
+            returned_stats.append(self.generate_stats(k, v))
+
+        return version_stats, returned_stats
+
+    @property
+    def key(self) -> str:
+        """Return the key of the list."""
+        return 'name'
+
+    def generate_stats(self, vm_name, vm_stats) -> Dict[str, Any]:
+        # Init the stats for the current vm
+        return {
+            'key': self.key,
+            'name': nativestr(vm_name),
+            'id': vm_stats.get('image_hash'),
+            'status': vm_stats.get('state').lower() if vm_stats.get('state') else None,
+            'release': vm_stats.get('release') if len(vm_stats.get('release')) > 0 else vm_stats.get('image_release'),
+            'cpu_count': int(vm_stats.get('cpu_count', 1)) if len(vm_stats.get('cpu_count', 1)) > 0 else None,
+            'memory_usage': vm_stats.get('memory').get('used') if vm_stats.get('memory') else None,
+            'memory_total': vm_stats.get('memory').get('total') if vm_stats.get('memory') else None,
+            'load_1min': vm_stats.get('load')[0] if vm_stats.get('load') else None,
+            'load_5min': vm_stats.get('load')[1] if vm_stats.get('load') else None,
+            'load_15min': vm_stats.get('load')[2] if vm_stats.get('load') else None,
+            'ipv4': vm_stats.get('ipv4')[0] if len(vm_stats.get('ipv4')) > 0 else None,
+            # TODO: disk
+        }