diff --git a/doc/conf.py b/doc/conf.py index c5d5b7063..867fd6cfd 100644 --- a/doc/conf.py +++ b/doc/conf.py @@ -368,6 +368,8 @@ u'Manage tags on a qube', _man_pages_author, 1), ('manpages/qvm-template', 'qvm-template', u'Manage templates', _man_pages_author, 1), + ('manpages/qvm-top', 'qvm-top', + u'Top-like monitoring tool', _man_pages_author, 1), ('manpages/qvm-unpause', 'qvm-unpause', u'Pause a qube', _man_pages_author, 1), ('manpages/qvm-volume', 'qvm-volume', diff --git a/doc/manpages/qvm-top.rst b/doc/manpages/qvm-top.rst new file mode 100644 index 000000000..7de190362 --- /dev/null +++ b/doc/manpages/qvm-top.rst @@ -0,0 +1,361 @@ +.. program:: qvm-top + +:program:`qvm-top` -- top-like monitoring tool +============================================== + +Synopsis +-------- + +:command:`qvm-top` [--verbose] [--quiet] [--help] [--version] [--all] [--exclude *EXCLUDE*] [--show-halted] [--filter *FILTER*,...] [--columns *COLUMN*,... | --format *FORMAT*] [--help-columns] [--help-formats] [--sort-column *COLUMN*] [--reverse] [--no-color] [*VMNAME* ...] + +Top-like info for Qubes. Defaults is to show all non-halted qubes. + +Positional arguments +-------------------- + +.. option:: VMNAME ... + + Zero or more domain names. If none spcified, the default is to show all. + +General Options +--------------- + +.. option:: --verbose, -v + + Increase verbosity. + +.. option:: --quiet, -q + + Decrease verbosity. + +.. option:: --help, -h + + Show this help message and exit. + +.. option:: --version + + Show program's version number and exit. + +.. option:: --all + + Perform the action on all qubes + +.. option:: --exclude EXCLUDE + + Exclude the qube from ``--all``. + +.. option:: --show-halted, -S + + Don't hide halted qubes. + +.. option:: --filter, -f FILTER,... + + Filter domains name matching each fixed string separated by comma. + +.. option:: --thin-columns + + Columns will to be as large as the its header or current lengthiest data. It + is not the default because columns will move every time the length of the + lengthiest data change. + +.. option:: --memory-unit UNIT + + Unit to use for displaying memory. The digit after the dot is the precision. + Available: KiB, MiB, GiB.0, GiB.1, GiB.2, GiB.3. + +.. option:: --no-color + + Do not colorize the screen. If this option is not provided, but environment + variable `NO_COLOR` is set, the screen will not be colorized. + +Formatting Options +------------------ + +.. option:: --columns, -C COLUMN,... + + Show only specified columns. + +.. option:: --format, -F FORMAT + + Show only columns declared by format: `min`, `default`, `max-no-internal`, + `max`. + +.. option:: --help-columns + + List all available columns with short descriptions and exit. + +.. option:: --help-formats + + List all available formats with short descriptions and exit. + +Sorting Options +--------------- + +.. option:: --sort-column, -k COLUMN + + Sort by specified column. + +.. option:: --reverse, -r + + Reverse sorting. + +Interaction - Navigation +------------------------ + +.. option:: Up, k, ^N + + Scroll one row up. + +.. option:: Down, j, ^P + + Scroll one row down. + +.. option:: ^B, ^F + + Scroll one page up or down. + +.. option:: ^U, ^D + + Scroll half page up or down. + +.. option:: Home, ^A, g + + Scroll to the first page. + +.. option:: End, ^E, G + + Scroll to the last page. + +Interaction - Visualization +--------------------------- + +.. option:: Left, h + + Sort the column to the left. + +.. option:: Right, l + + Sort the column to the right. + +.. option:: r + + Reverse sorting order. + +.. option:: S + + Toggle showing halted qubes. + +.. option:: ^L + + Redraw the screen on the next refresh. + +.. option:: q, Q, ESC, ^C + + Quit the application. + +Interaction - Filter +-------------------- + +.. option:: / + + Filter by qube name, CSV. + +.. option:: Enter + + Apply filter and return to main screen. + +.. option:: q, Q, ESC, ^C + + Quit filter mode. + +Interaction - Execution +----------------------- + +.. option:: Double-Left-Click, Space + + Tag a row and move one line, down if row was not previously selected, else + up. + +.. option:: T + + Toggle tag all visible rows. Great when used with filter. + +.. option:: U + + Untag all rows. + +.. option:: a + + Toggle action column, which shows only actions that can be executed to all + tagged qubes. If none is tagged, act based on the current selection. + +.. option:: Numbers + + If action column is active, typing a number will select the corresponding + action. + +.. option:: Enter + + If action column is active, apply action and return to main screen. + +.. option:: q, Q, ESC, ^C + + Quit action mode. + +Headers +------- + +.. option:: qvm-top + + The clock shows the last time the screen was refreshed. + +.. option:: Domains + + Indicates the state of all filtered qubes, and if there are any selected + qube. + +.. option:: Memory + + Host indicators: + + - ``total``: How much memory the host has. + - ``assigned``: How much memory is made available from the host to the + domains. + - ``unassigned``: How much memory from the host was not assigned to any + domain. + + Domain indicators (qubes can lie): + + - ``used``: How much memory the domains are using from the total. + - ``swap``: How much swap the the domains are using from the total. + +Columns +------- + +Columns are identified by a machine header followed by a pretty header if +necessary. + +.. option:: name - NAME + + Qube's name. + +.. option:: state - STATE + + Qube's power state. + +.. option:: memory_init -> MI + + How much memory the system must reserve for the qube to be able to + initialize. On non-memory-balanced qubes, this is the maximum amount of + memory a domain will ever have while it is running. + +.. option:: memory_max -> MM + + How much memory the qube can use from the system. Part of this value is + reserved to videoram, while the rest is up to qmemman to balloon up this qube + when there is enough free memory on the host. + +.. option:: memory_assigned_max_total -> MAMT + + ``MAM`` + ``MAMi``. + +.. option:: memory_assigned_usable_total -> MAUT + + ``MAU`` + ``MAUi``. + +.. option:: memory_used_total -> MUT + + ``MU`` + ``MUi``. + +.. option:: cpu_time_total -> CPUsecT + + ``CPUsec`` + ``CPUisec``. + +.. option:: online_vcpus_total -> VCT + + ``VC`` + ``VCi``. + +.. option:: memory_assigned_max -> MAM + + How much memory has been assigned to the qube, including overhead. + +.. option:: memory_assigned_usable -> MAU + + How much memory has been assigned to the qube and can be used. A qube is + allowed to claim this amount at any time, and it cannot use more memory than + what has been assigned to it. When the system is under no memory pressure, + this value is close to ``MM``, while when the system is under memory + pressure, the value can be as low as enough for the qube to survive. + +.. option:: memory_used_noswap -> MU + + How much memory the qube alleges to use. This value or part of it is + broadcast by the qube, it can be a lie. + +.. option:: memory_used_swap -> MUS + + How much memory the qube alleges to use for swap. This value or part of it is + broadcast by the qube, it can be a lie. + +.. option:: memory_usage_assigned -> MAM/MM + + How much memory the qube has assigned compared to the maximum it can have + assigned, in percentage. A high percentage means the system is not pressuring + the qube to release memory. + +.. option:: memory_usage_used -> MU/MAU + + How much memory the qube alleges to use from the assigned amount, in + percentage. A high percentage on non-memory-balanced qubes is irrelevant. On + memory balanced qubes, a higher value indicates the qube is using a lot of + the memory it has assigned, which might be near exhaustion, if ``MAU`` can't + be ballooned up anymore. + +.. option:: memory_usage_swap_over_used -> MUS/MU + + How much memory the qube alleges to be swaping from what it alleges touse, in + percentage. When it is over 10%, the qube might be swaping too much. + +.. option:: cpu_time -> CPUsec + + How many seconds the qube has used from the CPU. + +.. option:: cpu_usage -> CPU% + + How much CPU the qube is using, in percentage. + +.. option:: online_vcpus -> VC + + How many Virtual CPUs are online. + +.. option:: memory_assigned_max_internal -> MAMi + + Same as ``MAM``, but internal usage. + +.. option:: memory_assigned_usable_internal -> MAUi + + Same as ``MAU``, but internal usage. + +.. option:: memory_used_noswap_internal -> MUi + + Same as ``MU``, but internal usage. + +.. option:: cpu_time_internal -> CPUisec + + Same as ``CPUsec``, but internal usage. + +.. option:: cpu_usage_internal -> CPUi% + + Same as ``CPU%``, but internal usage. + +.. option:: online_vcpus_internal -> VCi + + Same as ``VC``, but internal usage. + +Authors +------- + +| Benjamin Grande +| For complete author list see: https://github.com/QubesOS/qubes-core-admin-client.git + +.. vim: ts=3 sw=3 et tw=80 diff --git a/doc/qubesadmin.tools.rst b/doc/qubesadmin.tools.rst index 416767385..408947e7f 100644 --- a/doc/qubesadmin.tools.rst +++ b/doc/qubesadmin.tools.rst @@ -196,6 +196,14 @@ qubesadmin\.tools\.qvm\_template\_postprocess module :undoc-members: :show-inheritance: +qubesadmin\.tools\.qvm\_top module +---------------------------------------------------- + +.. automodule:: qubesadmin.tools.qvm_top + :members: + :undoc-members: + :show-inheritance: + qubesadmin\.tools\.qvm\_unpause module -------------------------------------- diff --git a/qubesadmin/app.py b/qubesadmin/app.py index 132ba74f3..7b2b9796c 100644 --- a/qubesadmin/app.py +++ b/qubesadmin/app.py @@ -784,13 +784,8 @@ def _update_power_state_cache(self, subject: QubesVM, event: str, **kwargs) -> None: """Update cached VM power state. - This method is designed to be hooked as an event handler for: - - domain-pre-start - - domain-start - - domain-shutdown - - domain-paused - - domain-unpaused - - domain-start-failed + This method is designed to be hooked as an event handler for + :py:data:`qubesadmin.events.POWER_EVENTS`. This is done in :py:class:`qubesadmin.events.EventsDispatcher` class directly, before calling other handlers. @@ -805,17 +800,28 @@ def _update_power_state_cache(self, subject: QubesVM, return if event == "domain-pre-start": - power_state = "Transient" + power_state = "Starting" + elif event == "domain-start-failed": + power_state = "Halted" elif event == "domain-start": power_state = "Running" - elif event == "domain-shutdown": - power_state = "Halted" + elif event == "domain-paused": power_state = "Paused" elif event == "domain-unpaused": power_state = "Running" - elif event == "domain-start-failed": + elif event == "domain-suspended": + power_state = "Suspended" + elif event == "domain-resumed": + power_state = "Running" + + elif event == "domain-pre-shutdown": + power_state = "Halting" + elif event == "domain-shutdown-failed": + power_state = "Running" + elif event == "domain-shutdown": power_state = "Halted" + else: # unknown power state change, drop cached power state power_state = None diff --git a/qubesadmin/events/__init__.py b/qubesadmin/events/__init__.py index d2581867b..16cd03083 100644 --- a/qubesadmin/events/__init__.py +++ b/qubesadmin/events/__init__.py @@ -36,6 +36,20 @@ Handler: typing.TypeAlias\ = Callable[[QubesVM | None, str, ...], Any] # noqa: ANN401 +POWER_EVENTS = [ + "domain-pre-start", + "domain-start-failed", + "domain-start", + "domain-paused", + "domain-unpaused", + "domain-suspended", + "domain-resumed", + "domain-pre-shutdown", + "domain-shutdown-failed", + "domain-shutdown", +] + + class EventsDispatcher: ''' Events dispatcher, responsible for receiving events and calling appropriate handlers''' @@ -243,9 +257,7 @@ def handle(self, subject_name: str | None, event: str, **kwargs) -> None: if event.startswith('property-set:') or \ event.startswith('property-reset:'): self.app._invalidate_cache(subject, event, **kwargs) - elif event in ('domain-pre-start', 'domain-start', 'domain-shutdown', - 'domain-paused', 'domain-unpaused', - 'domain-start-failed'): + elif event in POWER_EVENTS: assert subject is not None self.app._update_power_state_cache(subject, event, **kwargs) subject.devices.clear_cache() diff --git a/qubesadmin/tests/tools/qvm_ls.py b/qubesadmin/tests/tools/qvm_ls.py index 6a8377940..653144f8b 100644 --- a/qubesadmin/tests/tools/qvm_ls.py +++ b/qubesadmin/tests/tools/qvm_ls.py @@ -306,36 +306,42 @@ def setUp(self): self.app.domains = TestVMCollection( [ ('a', TestVM('a', power_state='Halted')), - ('b', TestVM('b', power_state='Transient')), - ('c', TestVM('c', power_state='Running')) + ('b', TestVM('b', power_state='Starting')), + ('c', TestVM('c', power_state='Running')), + ('d', TestVM('d', power_state='Halting')), ] ) def test_100_nofilter(self): with qubesadmin.tests.tools.StdoutBuffer() as stdout: - qubesadmin.tools.qvm_ls.main([], app=self.app) + qubesadmin.tools.qvm_ls.main(['--raw-list'], app=self.app) self.assertEqual(stdout.getvalue(), - 'NAME STATE CLASS LABEL TEMPLATE NETVM\n' - 'a Halted TestVM - - -\n' - 'b Transient TestVM - - -\n' - 'c Running TestVM - - -\n') + 'a\n' + 'b\n' + 'c\n' + 'd\n' + ) def test_100_running(self): with qubesadmin.tests.tools.StdoutBuffer() as stdout: - qubesadmin.tools.qvm_ls.main(['--running'], app=self.app) + qubesadmin.tools.qvm_ls.main( + ['--raw-list', '--running'], + app=self.app + ) self.assertEqual(stdout.getvalue(), - 'NAME STATE CLASS LABEL TEMPLATE NETVM\n' - 'c Running TestVM - - -\n') + 'c\n' + ) def test_100_running_or_halted(self): with qubesadmin.tests.tools.StdoutBuffer() as stdout: - qubesadmin.tools.qvm_ls.main(['--running', '--halted'], - app=self.app) + qubesadmin.tools.qvm_ls.main( + ['--raw-list', '--running', '--halted'], + app=self.app + ) self.assertEqual(stdout.getvalue(), - 'NAME STATE CLASS LABEL TEMPLATE NETVM\n' - 'a Halted TestVM - - -\n' - 'c Running TestVM - - -\n') - + 'a\n' + 'c\n' + ) class TC_90_List_with_qubesd_calls(qubesadmin.tests.QubesTestCase): def test_100_list_with_status(self): diff --git a/qubesadmin/tests/vm/properties.py b/qubesadmin/tests/vm/properties.py index 261aa9367..68cc5a38a 100644 --- a/qubesadmin/tests/vm/properties.py +++ b/qubesadmin/tests/vm/properties.py @@ -281,11 +281,21 @@ def test_012_power_state_halted(self): self.assertFalse(self.vm.is_paused()) self.assertFalse(self.vm.is_suspended()) - def test_012_power_state_transient(self): + def test_012_power_state_starting(self): self.app.expected_calls[ ('test-vm', 'admin.vm.CurrentState', None, None)] = \ - b'0\x00power_state=Transient' - self.assertEqual(self.vm.get_power_state(), 'Transient') + b'0\x00power_state=Starting' + self.assertEqual(self.vm.get_power_state(), 'Starting') + self.assertTrue(self.vm.is_running()) + self.assertFalse(self.vm.is_halted()) + self.assertFalse(self.vm.is_paused()) + self.assertFalse(self.vm.is_suspended()) + + def test_012_power_state_halting(self): + self.app.expected_calls[ + ('test-vm', 'admin.vm.CurrentState', None, None)] = \ + b'0\x00power_state=Halting' + self.assertEqual(self.vm.get_power_state(), 'Halting') self.assertTrue(self.vm.is_running()) self.assertFalse(self.vm.is_halted()) self.assertFalse(self.vm.is_paused()) diff --git a/qubesadmin/tools/__init__.py b/qubesadmin/tools/__init__.py index 85cfa7fad..2994f0fd6 100644 --- a/qubesadmin/tools/__init__.py +++ b/qubesadmin/tools/__init__.py @@ -170,14 +170,21 @@ def parse_qubes_app(self, parser, namespace): assert hasattr(namespace, 'app') setattr(namespace, 'domains', []) app = namespace.app - if hasattr(namespace, 'all_domains') and namespace.all_domains: + if ( + hasattr(namespace, 'all_domains') + and namespace.all_domains + and not getattr(namespace, 'VMNAME') + ): namespace.domains = [ vm for vm in app.domains - if not vm.klass == 'AdminVM' and - vm.name not in namespace.exclude + if ( + namespace.all_domains == "global" + or not vm.klass == 'AdminVM' + ) and vm.name not in namespace.exclude ] else: + setattr(namespace, 'all_domains', False) if hasattr(namespace, 'exclude') and namespace.exclude: parser.error('--exclude can only be used with --all') @@ -333,7 +340,7 @@ class QubesArgumentParser(argparse.ArgumentParser): ''' def __init__(self, vmname_nargs=None, show_forceroot=False, version=None, \ - **kwargs): + all_default=False, all_include_adminvm=False, **kwargs): super().__init__(add_help=False, **kwargs) @@ -370,7 +377,8 @@ def __init__(self, vmname_nargs=None, show_forceroot=False, version=None, \ self.add_argument('--version', action='version') if self._vmname_nargs in [argparse.ZERO_OR_MORE, argparse.ONE_OR_MORE]: - vm_name_group = VmNameGroup(self, + vm_name_group = VmNameGroup(self, all_default=all_default, + all_include_adminvm=all_include_adminvm, required=(self._vmname_nargs not in [argparse.ZERO_OR_MORE, argparse.OPTIONAL])) self._mutually_exclusive_groups.append(vm_name_group) @@ -546,13 +554,16 @@ class VmNameGroup(argparse._MutuallyExclusiveGroup): :py:class:``argparse.ArgumentParser```. ''' - def __init__(self, container, required, vm_action=VmNameAction, help=None): + def __init__(self, container, required, vm_action=VmNameAction,\ + all_default=False, all_include_adminvm=False, help=None): # pylint: disable=redefined-builtin super().__init__(container, required=required) if not help: help = 'perform the action on all qubes' + if all_default and all_include_adminvm: + all_default = "global" self.add_argument('--all', action='store_true', dest='all_domains', - help=help) + default=all_default, help=help) container.add_argument('--exclude', action='append', default=[], help='exclude the qube from --all') diff --git a/qubesadmin/tools/qvm_ls.py b/qubesadmin/tools/qvm_ls.py index 5134c8a8a..1f1844de7 100644 --- a/qubesadmin/tools/qvm_ls.py +++ b/qubesadmin/tools/qvm_ls.py @@ -128,7 +128,7 @@ def _format_flags(vm: QubesVM) -> str: 1 type: 0=AdminVM, a/A=AppVM, d/D=DispVM, s/S=StandaloneVM, t/T=TemplateVM (uppercase = HVM) - 2 power state: r=running, t=transient, p=paused, s=suspended, + 2 power state: r=running, S=starting, p=paused, s=suspended, h=halting, d=dying, c=crashed, ?=unknown 3 U updateable 4 N provides_network @@ -149,7 +149,7 @@ def _format_flags(vm: QubesVM) -> str: state = vm.get_power_state().lower() if state == 'unknown': power_letter = '?' - elif state in ('running', 'transient', 'paused', 'suspended', + elif state in ('running', 'starting', 'paused', 'suspended', 'halting', 'dying', 'crashed'): power_letter = state[0] else: diff --git a/qubesadmin/tools/qvm_start_daemon.py b/qubesadmin/tools/qvm_start_daemon.py index 6983ad484..e1ca271e4 100644 --- a/qubesadmin/tools/qvm_start_daemon.py +++ b/qubesadmin/tools/qvm_start_daemon.py @@ -949,7 +949,7 @@ def on_connection_established(self, _subject, _event, **_kwargs): if "audiovm" in self.enabled_services: asyncio.ensure_future(self.start_audio(vm)) self.xid_cache[vm.name] = vm.xid, vm.stubdom_xid - elif power_state == "Transient": + elif power_state == "Starting": # it is still starting, we'll get 'domain-start' # event when fully started if ( diff --git a/qubesadmin/tools/qvm_top.py b/qubesadmin/tools/qvm_top.py new file mode 100644 index 000000000..9ebc5288a --- /dev/null +++ b/qubesadmin/tools/qvm_top.py @@ -0,0 +1,2716 @@ +#!/usr/bin/env python3 +# +# SPDX-FileCopyrightText: 2025 - 2026 Benjamin Grande +# +# SPDX-License-Identifier: GPL-2.0-only + +""" +Top-like info for Qubes. + +Visuals from htop and xentop. +""" + +import curses + +from argparse import Action, SUPPRESS, ArgumentTypeError +from asyncio import ( + CancelledError, + create_subprocess_exec, + create_task, + ensure_future, + gather, + run, + sleep, + to_thread, +) +from asyncio.subprocess import PIPE, STDOUT, DEVNULL +from asyncio.subprocess import subprocess as async_subprocess +from collections import Counter +from curses.ascii import ( + ESC, + STX, + NAK, + EOT, + ACK, + SOH, + ENQ, + DLE, + SO, + FF, + SP, + CR, +) +from importlib.metadata import metadata +from logging import getLogger, Formatter, Logger, DEBUG, INFO +from logging.handlers import SysLogHandler +from os import environ +from sys import stderr, exc_info, exit as sys_exit +from time import strftime +from textwrap import TextWrapper +from traceback import print_exception +from typing import TypedDict, NotRequired, Callable, Awaitable, Any + +from qubesadmin.app import QubesBase +from qubesadmin.events import EventsDispatcher +from qubesadmin.events import POWER_EVENTS +from qubesadmin.exc import ( + QubesPropertyAccessError, + QubesVMNotFoundError, +) +from qubesadmin.label import Label +from qubesadmin.tools import QubesArgumentParser +from qubesadmin.utils import start, pause, unpause, shutdown, kill +from qubesadmin.vm import QubesVM, POWER_STATES + + +class ActionEntry(TypedDict): + # pylint: disable=missing-class-docstring + identity: int + action: Callable[[list[QubesVM]], Awaitable[object]] + on_state: list[str] + when: NotRequired[list[str]] + + +def log_failures(qubes, results) -> None: + """ + Filter results for failures and log them. + """ + failed: dict[QubesVM, BaseException] = {} + for qube, res in zip(qubes, results): + if not isinstance(res, BaseException): + continue + failed[qube] = res + for qube, exc in failed.items(): + qube.log.exception(exc, exc_info=exc) + + +async def exec_and_check(*args): + """ + Execute subprocess asynchronously and check for its result. + """ + proc = await create_subprocess_exec( + *args, + stdin=DEVNULL, + stdout=PIPE, + stderr=STDOUT, + ) + outerr, _ = await proc.communicate() + if proc.returncode != 0: + raise async_subprocess.CalledProcessError( + returncode=proc.returncode, cmd=[*args], output=outerr + ) + + +async def console(qubes: list[QubesVM]) -> None: + """ + Get debug console for provided qubes. + """ + tasks = [ + exec_and_check("qvm-console-dispvm", "--autostart", qube.name) + for qube in qubes + ] + results = await gather(*tasks, return_exceptions=True) + log_failures(qubes, results) + + +async def terminal(qubes: list[QubesVM], user: str | None = None) -> None: + """ + Get GUI terminal for provided qubes. + """ + tasks = [ + to_thread( + qube.run_service_for_stdio, + "qubes.StartApp+qubes-run-terminal", + user=user, + ) + for qube in qubes + ] + results = await gather(*tasks, return_exceptions=True) + log_failures(qubes, results) + + +async def action_wrapper( + qubes: list[QubesVM], action: Callable, **kwargs +) -> None: + """ + Run action from utils and log failures. + """ + results = await action(domains=qubes, **kwargs) + log_failures(qubes, results) + + +ACTIONS: dict[str, ActionEntry] = { + "Run terminal": { + "identity": 0, + "action": lambda qubes: terminal(qubes=qubes), + "on_state": ["Halted", "Running"], + "when": ["can_gui"], + }, + "Run root terminal": { + "identity": 1, + "action": lambda qubes: terminal(qubes=qubes, user="root"), + "on_state": ["Halted", "Running"], + "when": ["can_gui"], + }, + "Debug console": { + "identity": 2, + "action": lambda qubes: console(qubes=qubes), + "on_state": ["Halted", "Running"], + "when": ["can_console"], + }, + "Start": { + "identity": 3, + "action": lambda qubes: action_wrapper(qubes=qubes, action=start), + "on_state": ["Halted"], + }, + "Pause": { + "identity": 4, + "action": lambda qubes: action_wrapper(qubes=qubes, action=pause), + "on_state": ["Running"], + }, + "Unpause": { + "identity": 5, + "action": lambda qubes: action_wrapper(qubes=qubes, action=unpause), + "on_state": ["Paused"], + }, + "Shutdown": { + "identity": 6, + "action": lambda qubes: action_wrapper( + qubes=qubes, action=shutdown, wait=True + ), + "on_state": ["Transient", "Running"], + }, + "Force shutdown": { + "identity": 7, + "action": lambda qubes: action_wrapper( + qubes=qubes, action=shutdown, force=True, wait=True + ), + "on_state": ["Transient", "Running"], + }, + # "Restart": { + # "identity": 8, + # "action": lambda qubes: action_wrapper(qubes=qubes, action=restart), + # "on_state": ["Transient", "Running", "Paused", "Halting"], + # "when": ["can_restart"], + # }, + "Kill": { + "identity": 9, + "action": lambda qubes: action_wrapper(qubes=qubes, action=kill), + "on_state": ["Transient", "Running", "Paused", "Halting"], + }, +} +ACTION_NUMBERS: list[int] = [action["identity"] for action in ACTIONS.values()] +ACTION_WIDTH = max(len(action) for action in ACTIONS) +ACTION_NUMBER_WIDTH = max(len(str(number)) for number in ACTION_NUMBERS) +ACTION_ALL_WIDTHS = ACTION_WIDTH + ACTION_NUMBER_WIDTH + 1 + +QUBES_TOP_DEBUG = bool(environ.get("QUBES_TOP_DEBUG")) +LOGGING_LEVEL = DEBUG if QUBES_TOP_DEBUG else INFO +NO_COLOR = bool(environ.get("NO_COLOR")) +UNICODE = bool("utf-8" in environ.get("LC_ALL", "").lower()) +if UNICODE: + SORT_SIGN_UP = "\u2193" + SORT_SIGN_DOWN = "\u2191" + ELLIPSIS = "\u2026" +else: + SORT_SIGN_UP = "^" + SORT_SIGN_DOWN = "v" + ELLIPSIS = "..." +UNTRUSTED_NUMBER_MAX_LEN = 10 +MEMORY_UNIT = "MiB" +MEMORY_UNIT_PRECISION = 0 +MEMORY_UNIT_AVAILABLE = ["KiB", "MiB", "GiB.0", "GiB.1", "GiB.2", "GiB.3"] + + +def convert_html_color_to_curses(hexadecimal: str): + """ + Convert HTML color codes to RGB accepted by curses. + """ + number = int(hexadecimal, 16) + red = (number >> 16) & 255 + green = (number >> 8) & 255 + blue = number & 255 + return ( + int(red * 1000 / 255), + int(green * 1000 / 255), + int(blue * 1000 / 255), + ) + + +def gen_logger(name: str) -> Logger: + """ + Return logger for a given name. Use syslog, as stdout and stderr are being + used by curses. + """ + logger = getLogger(name) + logger.setLevel(LOGGING_LEVEL) + formatter = Formatter( + f" %(levelname)s: {name} %(funcName)s:%(lineno)d: %(message)s" + ) + handler = SysLogHandler(address="/dev/log") + handler.setFormatter(formatter) + handler.ident = "qvm-top" + logger.addHandler(handler) + logger.propagate = False + return logger + + +class Stats: + """ + Storage qube statistics. + """ + + # pylint: disable=too-many-instance-attributes + + outdated_by_removal: list[str] = [] + host_checked: bool = False + host_memory_max: int | float | str = "NA" + host_no_cpus: int | str = "NA" + host_cpu_mhz: int | str = "NA" + + def __init__(self, vm: QubesVM) -> None: + super().__init__() + self.vm: QubesVM = vm + + self.uptodate: bool = False + self.name = str(self.vm) + self.log = gen_logger(f"{self.__class__.__qualname__}.{self.name}") + + self.features_default = {"gui": True, "internal": False} + self.state: str = "NA" + self.label: Label | str = "" + self.is_preload: bool = False + self.gui: bool = True + self.internal: bool = False + self.guivm: QubesVM | None = None + self.management_dispvm: QubesVM | None = None + self.auto_cleanup: bool = False + self.memory_max: int | str = "NA" + self.memory_init: int | str = "NA" + self.update_cache() + + self.memory_assigned_max: int | str = "NA" + self.memory_assigned_usable: int | str = "NA" + self.memory_usage_assigned: float | str = "NA" + self.memory_used_noswap: int | str = "NA" + self.memory_used_swap: int | str = "NA" + self.memory_usage_used: float | str = "NA" + self.memory_usage_swap_over_used: float | str = "NA" + self.cpu_time: int | str = "NA" + self.cpu_usage: int | str = "NA" + self.online_vcpus: int | str = "NA" + + self.memory_assigned_max_internal: int | str = "NA" + self.memory_assigned_usable_internal: int | str = "NA" + self.memory_used_noswap_internal: int | str = "NA" + self.cpu_time_internal: int | str = "NA" + self.cpu_usage_internal: float | str = "NA" + self.online_vcpus_internal: int | str = "NA" + + self.memory_assigned_max_total: int | str = "NA" + self.memory_assigned_usable_total: int | str = "NA" + self.memory_used_noswap_total: int | str = "NA" + self.memory_used_swap_total: int | str = "NA" + self.memory_used_total: int | str = "NA" + self.cpu_time_total: int | str = "NA" + self.cpu_usage_total: float | str = "NA" + self.online_vcpus_total: int | str = "NA" + + if self.__class__.host_checked is False: + self.__class__.host_checked = True + host_memory_max: int | str = getattr(self.vm.app, "maxmem", "NA") + if isinstance(host_memory_max, int): + host_memory_max = self.convert_memory(int(host_memory_max)) + self.__class__.host_memory_max = host_memory_max + self.__class__.host_no_cpus = getattr(self.vm.app, "no_cpus", "NA") + self.__class__.host_cpu_mhz = getattr(self.vm.app, "cpu_mhz", "NA") + + def convert_memory(self, number: int) -> int | float: + """ + Convert memory data to the expected unit. + """ + match MEMORY_UNIT: + case "KiB": + return number + case "GiB": + return round(float(number / 1024**2), MEMORY_UNIT_PRECISION) + case _: + return number // 1024 + + def can_gui(self) -> bool: + """ + Returns boolean if qube can show graphical applications. + """ + return bool( + self.guivm + and self.gui + and not self.is_preload + and not self.internal + ) + + def can_console(self) -> bool: + """ + Returns boolean if qube can have a debug console. + """ + return bool( + self.guivm + and self.management_dispvm + and not self.is_preload + and not self.internal + ) + + def can_restart(self) -> bool: + """ + Returns boolean if qube can be restarted. + """ + return bool(not self.auto_cleanup) + + def filter_actions(self) -> list[str]: + """ + Only show actions that passes all filters. + """ + actions: list[str] = [] + for action, entry in ACTIONS.items(): + state = entry["on_state"] + when = entry.get("when", []) + if self.state not in state: + continue + if not (not when or all(getattr(self, meth)() for meth in when)): + continue + actions.append(action) + return actions + + def get_actions(self) -> list[str]: + """ + List available actions depending on the state. + """ + if self.vm.klass in ["AdminVM", "RemoteVM"]: + actions = [] + else: + actions = self.filter_actions() + return actions + + def _setter(self, name: str, newvalue) -> dict | None: + """ + Set instance attribute, inform if status is outdated and return changes. + """ + oldvalue = getattr(self, name) + if oldvalue == newvalue: + return None + self.uptodate = False + setattr(self, name, newvalue) + return {name: [oldvalue, newvalue]} + + def set_verbose(self, items_to_update: dict) -> None: + """ + Set multiple properties at a time and log outdated ones. + """ + updates = [ + res + for prop, value in items_to_update.items() + if (res := self._setter(prop, value)) + ] + if updates: + self.log.debug("update=%s", updates) + + def update_cache(self) -> None: + """ + Regenerate cache upon (re)connection. + """ + data = { + "state": self.vm.get_power_state(), + "label": self.vm.label, + "memory_init": getattr(self.vm, "memory", "NA"), + "memory_max": getattr(self.vm, "maxmem", "NA"), + "auto_cleanup": getattr(self.vm, "auto_cleanup", False), + "is_preload": getattr(self.vm, "is_preload", False), + "guivm": getattr(self.vm, "guivm", None), + "management_dispvm": getattr(self.vm, "management_dispvm", None), + "gui": ( + self.vm.features.check_with_template( + "gui", self.features_default["gui"] + ) + if self.vm.klass != "RemoteVM" + else self.features_default["gui"] + ), + "internal": ( + self.vm.features.check_with_template( + "internal", self.features_default["internal"] + ) + if self.vm.klass != "RemoteVM" + else self.features_default["internal"] + ), + } + for k in ["memory_init", "memory_max"]: + if isinstance(data[k], int): + data[k] = self.convert_memory(data[k] * 1024) + + self.set_verbose(data) + + def update_stats(self, **kwargs: Any) -> None: + """ + Update memory and CPU statistics. + """ + memory_assigned_max = self.convert_memory( + int(kwargs["memory_assigned_max_KiB"]) + ) + memory_assigned_usable = self.convert_memory( + int(kwargs["memory_assigned_usable_KiB"]) + ) + memory_assigned_max_internal = kwargs.get( + "memory_assigned_max_KiB_internal" + ) + memory_assigned_usable_internal = kwargs.get( + "memory_assigned_usable_KiB_internal" + ) + if memory_assigned_max_internal is None: + memory_assigned_max_internal = "NA" + memory_assigned_max_total = memory_assigned_max + else: + memory_assigned_max_internal = self.convert_memory( + int(memory_assigned_max_internal) + ) + memory_assigned_max_total = ( + memory_assigned_max + memory_assigned_max_internal + ) + if memory_assigned_usable_internal is None: + memory_assigned_usable_internal = "NA" + memory_assigned_usable_total = memory_assigned_usable + else: + memory_assigned_usable_internal = self.convert_memory( + int(memory_assigned_usable_internal) + ) + memory_assigned_usable_total = ( + memory_assigned_usable + memory_assigned_usable_internal + ) + + memory_used_swap = self.convert_memory( + int(kwargs["memory_used_swap_KiB"]) + ) + memory_used_noswap = self.convert_memory( + int(kwargs["memory_used_noswap_KiB"]) + ) + memory_used_noswap_internal = kwargs.get( + "memory_used_noswap_KiB_internal" + ) + if memory_used_noswap_internal is None: + memory_used_noswap_internal = "NA" + memory_used_noswap_total = memory_used_noswap + else: + memory_used_noswap_internal = self.convert_memory( + int(memory_used_noswap_internal) + ) + memory_used_noswap_total = ( + memory_used_noswap + memory_used_noswap_internal + ) + memory_used_swap_total = memory_used_swap + memory_used_total = memory_used_swap_total + memory_used_noswap_total + + if isinstance(self.memory_max, int) and self.memory_max > 0: + memory_usage_assigned: float | str = round( + float(memory_assigned_max / self.memory_max) * 100, 1 + ) + else: + memory_usage_assigned = "NA" + if memory_assigned_usable > 0: + memory_usage_used: float | str = round( + float(memory_used_noswap / memory_assigned_usable) * 100, 1 + ) + else: + memory_usage_used = 0.0 + if memory_used_noswap > 0: + memory_usage_swap_over_used: float | str = round( + float(memory_used_swap / memory_used_noswap) * 100, 1 + ) + else: + memory_usage_swap_over_used = 0.0 + + cpu_time = int(kwargs["cpu_time"]) // 10**3 + cpu_time_internal = kwargs.get("cpu_time_internal") + if cpu_time_internal is None: + cpu_time_internal = "NA" + cpu_time_total = cpu_time + else: + cpu_time_internal = int(cpu_time_internal) // 10**3 + cpu_time_total = cpu_time + cpu_time_internal + + cpu_usage = int(kwargs["cpu_usage"]) + cpu_usage_internal = kwargs.get("cpu_usage_internal") + if cpu_usage_internal is None: + cpu_usage_internal = "NA" + else: + cpu_usage_internal = float(cpu_usage_internal) + + online_vcpus = int(kwargs["online_vcpus"]) + online_vcpus_internal = kwargs.get("online_vcpus_internal") + if online_vcpus_internal is None: + online_vcpus_internal = "NA" + online_vcpus_total = online_vcpus + else: + online_vcpus_internal = int(online_vcpus_internal) + online_vcpus_total = online_vcpus + online_vcpus_internal + + data = { + "memory_assigned_max": memory_assigned_max, + "memory_assigned_usable": memory_assigned_usable, + "memory_used_noswap": memory_used_noswap, + "memory_used_swap": memory_used_swap, + "memory_usage_used": memory_usage_used, + "memory_usage_assigned": memory_usage_assigned, + "memory_usage_swap_over_used": memory_usage_swap_over_used, + "cpu_time": cpu_time, + "cpu_usage": cpu_usage, + "online_vcpus": online_vcpus, + "memory_assigned_usable_internal": memory_assigned_usable_internal, + "memory_assigned_max_internal": memory_assigned_max_internal, + "memory_used_noswap_internal": memory_used_noswap_internal, + "cpu_time_internal": cpu_time_internal, + "cpu_usage_internal": cpu_usage_internal, + "online_vcpus_internal": online_vcpus_internal, + "memory_assigned_usable_total": memory_assigned_usable_total, + "memory_assigned_max_total": memory_assigned_max_total, + "memory_used_noswap_total": memory_used_noswap_total, + "memory_used_swap_total": memory_used_swap_total, + "memory_used_total": memory_used_total, + "cpu_time_total": cpu_time_total, + "online_vcpus_total": online_vcpus_total, + } + self.set_verbose(data) + + def update_memory_init(self, value: int) -> None: + """ + Update initial memory. + """ + self.set_verbose({"memory_init": self.convert_memory(value * 1024)}) + + def update_memory_max(self, value: int) -> None: + """ + Update maximum memory and it's related properties. + """ + memory_max = self.convert_memory(int(value) * 1024) + if memory_max > 0: + memory_usage_assigned: float | str = "NA" + if isinstance(self.memory_assigned_max, int): + memory_usage_assigned = round( + float(self.memory_assigned_max / memory_max) * 100, 1 + ) + else: + memory_usage_assigned = "NA" + data = { + "memory_max": memory_max, + "memory_usage_assigned": memory_usage_assigned, + } + self.set_verbose(data) + + def update_generic(self, name: str, value: Any) -> None: + """ + Update instance attributes. + """ + self.set_verbose({name: value}) + + def update_feat_from_template(self, name: str) -> None: + """ + Update instance features according to template features. + """ + default = self.features_default[name] + if self.vm.klass != "RemoteVM": + value = self.vm.features.check_with_template(name, default) + else: + value = default + self.set_verbose({name: value}) + + +class Screen: + """ + Organize the screen for drawing, scrolling, typing. + """ + + # pylint: disable=too-many-instance-attributes + def __init__( + self, + app: QubesBase, + show_halted: bool, + allow_color: bool, + filter_query: str, + sort_reverse: bool, + sort_col_index: int | None, + columns: dict[str, "Column"], + thin_columns: bool, + ): + # pylint: disable=too-many-positional-arguments + self.allow_color = allow_color + self.app = app + self.show_halted = show_halted + self.filter_query = filter_query + self.filter = False + self.sort_reverse = sort_reverse + self.sort_col_index = sort_col_index + self.columns = columns + self.thin_columns = thin_columns + + self.log = gen_logger(f"{self.__class__.__qualname__}") + self.getch_timeout = 1000 + self.version: str = metadata("qubesadmin")["version"] + + self.selected_stats: list[Stats] = [] + self._stats: list[Stats] | None = None + + self.header_row = 4 + self.old_cursor_row = 0 + self.visible_stats: dict[int, Stats] = {} + self.visible_actions: dict[int, str] = {} + self.column_width: dict[str, int] = {} + self.act = False + self.actions: list[str] = [] + self.label_index = 0 + + self.scroll_offset = 0 + self.max_offset = 0 + self.action_scroll_offset = 0 + self.action_max_offset = 0 + self.body_height = 0 + self.body_top = 0 + self.body_bottom = 0 + self.cursor_row = 0 + self.outdated_sleep = 0.010 + self.uptodate_sleep = 0.1 + self._old_cursor_visibility = 0 + + self.last_selected_stat: Stats | None = None + self.last_selected_row: int | None = None + + def init_label(self, label: Label | str) -> Label | None: + """ + Initialize colors from qube labels. + """ + if not curses.can_change_color(): + return None + if isinstance(label, str): + label = self.app.labels[label] + name: str = label.name + color: str = label.color + if name in self.label_colors: + return label + index = self.label_index + if name == "black": + self.label_colors[name] = self.colors["FG_DEFAULT"] + return label + r1000, g1000, b1000 = convert_html_color_to_curses(color) + curses.init_color(index, r1000, g1000, b1000) + curses.init_pair(index, index, -1) + self.label_colors[name] = curses.color_pair(index) + self.label_index += 1 + return label + + def init_screen(self) -> None: + """ + Initialize curses. + + Not a color expert, relied on how existing applications such as Vim and + htop use the 8 default colors. + + - Black: only as reverse video + - Red: critical + - Green: header + - Blue: not very used, use with discretion + - Yellow: tag, warning + - Magenta: not very used, use with discretion + - Cyan: filter, cursor + - White: only as reverse video + + The problematic colors: + + - White and black: Never use them without reverse video, as they might + be the default font and background color. + - Blue: Some shades of blue can be too dark to read on dark background. + Not widely used. + - Magenta: Not widely used, unknown reason, might be too alarming? + """ + # pylint: disable=attribute-defined-outside-init + term_fallback = "xterm-256color" + term = environ.get("TERM") + curses_unknown_colored_term = ["tmux-256color", "tmux-direct"] + if term in curses_unknown_colored_term: + self.log.debug( + "TERM is known to support colors but curses renders it poorly, " + "falling back to sane TERM" + ) + environ["TERM"] = term_fallback + self.stdscr = curses.initscr() + self.stdscr.timeout(self.getch_timeout) + self.stdscr.keypad(True) + colors = { + "BLACK": 0, + "RED": 1, + "GREEN": 2, + "YELLOW": 3, + "BLUE": 4, + "MAGENTA": 5, + "CYAN": 6, + "WHITE": 7, + } + self.label_colors: dict[str, int] = {} + self.colors: dict[str, int] = {} + if self.allow_color: + curses.start_color() + curses.use_default_colors() + for curses_color_name, index in colors.items(): + color_code = getattr(curses, f"COLOR_{curses_color_name}") + curses.init_pair(index, color_code, -1) + self.colors[f"FG_{curses_color_name}"] = curses.color_pair( + index + ) + curses.init_pair(index + len(colors), -1, color_code) + self.colors[f"BG_{curses_color_name}"] = curses.color_pair( + index + ) + self.colors.update( + { + "FG_DEFAULT": curses.color_pair(0), + "BG_DEFAULT": curses.color_pair(0), + } + ) + self.label_index = len(list(self.colors.keys())) + 1 + + if curses.can_change_color(): + for label in self.app.labels.values(): + self.init_label(label=label) + else: + self.label_colors = { + "red": self.colors["FG_RED"], + "orange": self.colors["FG_YELLOW"] | curses.A_DIM, + "yellow": self.colors["FG_YELLOW"], + "green": self.colors["FG_GREEN"], + "gray": self.colors["FG_WHITE"] | curses.A_DIM, + "blue": self.colors["FG_BLUE"], + "purple": self.colors["FG_MAGENTA"], + "black": self.colors["FG_DEFAULT"], + } + self.label_color_backup = ( + self.colors["FG_BLACK"] | curses.A_UNDERLINE + ) + self.header_summary_attr: int | None = self.colors["FG_GREEN"] + self.header_attr = curses.A_REVERSE | self.colors["BG_GREEN"] + self.header_sel_attr: int | None = ( + curses.A_REVERSE | self.colors["BG_CYAN"] + ) + self.footer_attr: int | None = ( + curses.A_REVERSE | self.colors["BG_GREEN"] + ) + self.footer_sel_attr: int | None = ( + curses.A_REVERSE | self.colors["BG_CYAN"] + ) + self.sel_attr: int | None = ( + self.colors["FG_YELLOW"] | curses.A_REVERSE + ) + self.cursor_attr: int | None = ( + curses.A_REVERSE | self.colors["BG_CYAN"] + ) + self.sel_and_cursor_attr = self.cursor_attr | curses.A_DIM + else: + self.header_summary_attr = None + self.header_attr = curses.A_REVERSE + self.header_sel_attr = curses.A_REVERSE | curses.A_DIM + self.footer_attr = curses.A_REVERSE + self.footer_sel_attr = curses.A_REVERSE | curses.A_DIM + self.sel_attr = curses.A_DIM + self.cursor_attr = curses.A_REVERSE + self.sel_and_cursor_attr = curses.A_REVERSE | curses.A_DIM + curses.noecho() + curses.cbreak() + curses.nonl() + self._old_cursor_visibility = curses.curs_set(0) + curses.mousemask(curses.ALL_MOUSE_EVENTS) + + def restore_screen(self) -> None: + """ + Restore sane window. + """ + curses.nl() + curses.nocbreak() + curses.echo() + try: + curses.curs_set(self._old_cursor_visibility) + except curses.error: + pass + curses.endwin() + + def write(self, *args: Any, **kwargs: Any) -> None: + """ + Set screen text and attributes and deal with errors without panic. + """ + attr = kwargs.pop("attr", None) + max_width = kwargs.pop("max_width", None) + row, col, text = args + try: + if attr is not None: + self.stdscr.attron(attr) + if max_width: + self.stdscr.addnstr(row, col, text, max_width) + else: + self.stdscr.addstr(row, col, text) + except curses.error: + pass + finally: + if attr is not None: + self.stdscr.attroff(attr) + + def get_selection(self) -> list[Stats]: + """ + Get selected qube. If none is selected, consider the last one the + cursor was in. + """ + stats = self.selected_stats + if not stats and self.last_selected_stat: + stats = [self.last_selected_stat] + if not stats: + stats = [] + return stats + + def get_action_from_selection(self) -> list[str]: + """ + List common actions that is valid for all selected qubes. + """ + stats = self.get_selection() + if not stats: + return [] + actions = [] + all_actions = [] + for stat in stats: + if not (stat_actions := stat.get_actions()): + return [] + all_actions.append(stat_actions) + actions = list(set.intersection(*(set(x) for x in all_actions))) + actions = sorted(actions, key=lambda k: int(ACTIONS[k]["identity"])) + self.log.debug("available-actions=%s", actions) + return actions + + def get_stats(self): + """ + Qubes that can be shown. + """ + if self._stats is not None: + return self._stats + + self._stats = sorted( + [ + stats + for stats in Monitor.entries.values() + if ( + self.show_halted + or ( + stats.state != "Halted" and stats.vm.klass != "RemoteVM" + ) + ) + and stats.vm in Monitor.domains + and ( + (not self.filter_query and not self.filter) + or any( + string in stats.vm.name + for string in self.filter_query.split(",") + ) + ) + ], + key=self.sort_stats_helper, + reverse=self.sort_reverse, + ) + return self._stats + + def sort_stats_helper(self, row) -> tuple[int, float]: + """ + Helper to sort a column according to the values it owns. + """ + if self.sort_col_index is None or ( + (header := list(self.columns.keys())[self.sort_col_index]) + and header == "name" + ): + if row.vm.klass == "AdminVM": + return (0, row.vm.name) + return (1, row.vm.name) + val = getattr(row, header.lower()) + try: + return (1, float(val)) + except (TypeError, ValueError): + return (0, val.lower()) + + def mark_uptodate(self) -> None: + """ + Mark everything as up-to-date. + """ + for stats in self.get_stats(): + stats.uptodate = True + Stats.outdated_by_removal = [] + + def is_uptodate(self) -> bool: + """ + Check if statistics are live or expired. + """ + self._stats = None + outdated = ( + [stats.vm.name for stats in self.get_stats() if not stats.uptodate], + ) + uptodate = not Stats.outdated_by_removal and not outdated + self.log.debug( + "removed=%s outdated=%s", + Stats.outdated_by_removal, + outdated, + ) + return uptodate + + def draw_table(self) -> None: + """ + Define screen regions. + """ + # pylint: disable=too-many-locals,too-many-statements,too-many-branches + self.stdscr.erase() + height, width = self.stdscr.getmaxyx() + sums_row = self.header_row - 1 + self.body_top = self.header_row + 1 + self.body_top = self.header_row + 1 + self.body_bottom = height - 1 + self.body_height = max(0, self.body_bottom - self.body_top) + + if not self.cursor_row: + self.cursor_row = self.body_top + + wanted = self.get_stats() + self.mark_uptodate() + + total_items = len(wanted) + self.max_offset = max(0, total_items - self.body_height) + self.scroll_offset = min(self.max_offset, max(0, self.scroll_offset)) + scroll_start = self.scroll_offset + scroll_end = min(total_items, scroll_start + self.body_height) + visible = wanted[scroll_start:scroll_end] + + self.actions = self.get_action_from_selection() + total_actions = len(self.actions) + self.action_max_offset = max(0, total_actions - self.body_height) + self.action_scroll_offset = min( + self.action_max_offset, max(0, self.action_scroll_offset) + ) + action_scroll_start = self.action_scroll_offset + action_scroll_end = min( + total_actions, action_scroll_start + self.body_height + ) + action_visible = self.actions[action_scroll_start:action_scroll_end] + + memory_used_noswap_total: list[int] = [ + stats.memory_used_noswap_total + for stats in wanted + if isinstance(stats.memory_used_noswap_total, int) + ] + memory_used_swap_total: list[int] = [ + stats.memory_used_swap_total + for stats in wanted + if isinstance(stats.memory_used_swap_total, int) + ] + memory_assigned_max_total: list[int] = [ + stats.memory_assigned_max_total + for stats in wanted + if isinstance(stats.memory_assigned_max_total, int) + ] + states = [stats.state for stats in wanted] + sums: dict[str, list[int]] = {} + for machine_header, column in self.columns.items(): + if not column.summable_for_overall: + continue + sums[machine_header] = [ + val + for stats in wanted + if (val := getattr(stats, machine_header, "NA")) + and isinstance(val, (int, float)) + ] + + if self.sort_reverse: + sort_sign = SORT_SIGN_UP + else: + sort_sign = SORT_SIGN_DOWN + + self.visible_stats = {} + self.visible_actions = {} + for row_index in range(self.body_height): + curr_height = self.body_top + row_index + if row_index >= len(visible): + continue + self.visible_stats[curr_height] = visible[row_index] + + if ( + not self.act + and self.cursor_row > len(self.visible_stats) + self.header_row + ): + self.cursor_row = self.body_top + elif self.cursor_row == -1: + if self.last_selected_stat in visible: + self.cursor_row = ( + visible.index(self.last_selected_stat) + self.body_top + ) + else: + self.cursor_row = self.body_top + + line_content: dict[str, dict[int, tuple]] = {} + line_start_after_action = 0 + + for row_index in range(self.body_height): + line_start = 0 + curr_height = self.body_top + row_index + if self.act and self.actions: + act_attr = None + if curr_height == self.cursor_row and not self.filter: + act_attr = self.cursor_attr + if row_index < len(action_visible): + curr_action = action_visible[row_index] + self.visible_actions[curr_height] = curr_action + action = str(curr_action).ljust(ACTION_WIDTH) + action_number = str( + ACTIONS[action.strip()]["identity"] + ).rjust(ACTION_NUMBER_WIDTH) + content = action_number + " " + action + self.write( + curr_height, + line_start, + content, + attr=act_attr, + max_width=width - line_start, + ) + line_start = ACTION_ALL_WIDTHS + self.write( + curr_height, + line_start, + " ", + max_width=width - line_start, + ) + line_start += 1 + line_start_after_action = line_start + if row_index >= len(visible): + if row_index not in wanted: + continue + for column in self.columns.values(): + stats = wanted[row_index] + data = getattr(stats, column.machine_header.lower()) + line_content.setdefault( + column.machine_header, {} + ).setdefault(curr_height, (data, None)) + continue + stats = self.visible_stats[curr_height] + sel_attr = None + if ( + curr_height == self.cursor_row + and not self.filter + and not self.act + ): + if stats in self.selected_stats: + sel_attr = self.sel_and_cursor_attr + else: + sel_attr = self.cursor_attr + elif stats in self.selected_stats: + sel_attr = self.sel_attr + elif ( + not self.selected_stats + and curr_height == self.last_selected_row + and not self.filter + and self.act + ): + sel_attr = self.cursor_attr + for column in self.columns.values(): + color_attr = None + attr = column.machine_header.lower() + data = getattr(stats, attr) + if not sel_attr and self.allow_color: + if attr == "name": + try: + assert isinstance(stats.label, Label) + color_attr = self.label_colors[stats.label.name] + except KeyError: + color_attr = self.label_color_backup + elif attr == "state": + if stats.state in ["Paused", "Suspended"]: + color_attr = self.colors["FG_YELLOW"] + elif stats.state != "Running": + color_attr = self.colors["FG_RED"] + elif ( + column.percentage + and data != "NA" + and column.percentage_intensity + ): + if data > max(column.percentage_intensity): + color_attr = self.colors["FG_RED"] + elif data > min(column.percentage_intensity): + color_attr = self.colors["FG_YELLOW"] + elif isinstance(data, (int, float)) and data == 0: + color_attr = self.label_colors["gray"] + elif data == "NA" or ( + isinstance(data, (int, float)) and data == 0 + ): + color_attr = self.label_colors["gray"] + + if attr == "state" and self.thin_columns: + data = POWER_STATES[data]["short"] + + line_content.setdefault(column.machine_header, {}).setdefault( + curr_height, (data, sel_attr or color_attr) + ) + + del column + + self.column_width = {} + col_line_start: dict[int, int] = {} + for machine_header, column in self.columns.items(): + self.column_width[machine_header] = 0 + header_sum: int | float = 0 + if machine_header in sums: + header_sum = sum(sums[machine_header]) + if ( + not column.percentage + and isinstance(header_sum, float) + and MEMORY_UNIT_PRECISION > 0 + ): + header_sum = round(header_sum, MEMORY_UNIT_PRECISION) + header_sum_formatted = ( + f"{header_sum:.{MEMORY_UNIT_PRECISION}f}" + ) + else: + if MEMORY_UNIT_PRECISION == 0: + header_sum = round(header_sum) + header_sum_formatted = str(header_sum) + header_sum_len = len(str(header_sum_formatted)) + if not column.untrusted or ( + column.untrusted + and header_sum_len <= UNTRUSTED_NUMBER_MAX_LEN + ): + self.column_width[machine_header] = header_sum_len + + if machine_header not in line_content: + continue + + for curr_height, height_data in line_content[ + machine_header + ].items(): + col_line_start.setdefault(curr_height, line_start_after_action) + col_width = len(str(height_data[0])) + if col_width > self.column_width[machine_header]: + self.column_width[machine_header] = col_width + + for curr_height, height_data in line_content[ + machine_header + ].items(): + data, data_attr = height_data + + if not column.percentage and isinstance(data, float): + content_str = f"{data:.{MEMORY_UNIT_PRECISION}f}" + else: + content_str = str(data) + curr_width = self.column_width[machine_header] + content = column.justify(content_str, curr_width) + if ( + column.untrusted + and len(content_str) > UNTRUSTED_NUMBER_MAX_LEN + ): + content = column.justify(ELLIPSIS, curr_width) + + if column.summable_for_overall: + sum_content = column.justify( + header_sum_formatted, curr_width + ) + else: + sum_content = " " * len(content) + + content += " " + sum_content += " " + self.write( + sums_row, + col_line_start[curr_height], + sum_content, + max_width=width - col_line_start[curr_height], + ) + self.write( + curr_height, + col_line_start[curr_height], + content, + attr=data_attr, + max_width=width - col_line_start[curr_height], + ) + col_line_start[curr_height] += ( + max(curr_width, column.min_width) + 1 + ) + + state_counts = Counter(states) + total_states = len(states) + total_states_len = len(str(total_states)) + all_states = list(POWER_STATES) + + if self.thin_columns: + top_text = "top" + domain_text = "D" + mem_text = "M" + mem_unit = MEMORY_UNIT[0] + cpu_text = "C" + selected_text = "*" + assigned_text = "a" + used_text = "u" + swap_text = "s" + else: + top_text = "qvm-top" + domain_text = "Domain{}".format("s" if total_states > 0 else "") + mem_text = "Mem" + mem_unit = MEMORY_UNIT + cpu_text = "CPUs" + selected_text = "selected" + assigned_text = "assigned" + used_text = "used" + swap_text = "swap" + + memory_total = Stats.host_memory_max + sum_memory_used_noswap = sum(memory_used_noswap_total) + sum_memory_used_swap = sum(memory_used_swap_total) + sum_memory_assigned_max_total = sum(memory_assigned_max_total) + pct_memory_used_noswap: float | str = "NA" + pct_memory_used_swap: float | str = "NA" + pct_memory_assigned_max_total: float | str = "NA" + if isinstance(memory_total, int): + pct_memory_used_noswap = round( + sum_memory_used_noswap / memory_total * 100, 1 + ) + pct_memory_used_swap = round( + sum_memory_used_swap / memory_total * 100, 1 + ) + pct_memory_assigned_max_total = round( + sum_memory_assigned_max_total / memory_total * 100, 1 + ) + + no_cpus = Stats.host_no_cpus + cpu_mhz = Stats.host_cpu_mhz + if isinstance(cpu_mhz, int): + cpu_mhz = str(cpu_mhz) + "MHz" + header_cpu_prefix = " {}".format(cpu_text) + header_cpu_suffix = ": {} @ {}".format(no_cpus, cpu_mhz) + if MEMORY_UNIT_PRECISION == 0: + header_mem_prefix = "{}({})".format(mem_text, mem_unit) + else: + header_mem_prefix = "{}({}.{})".format( + mem_text, mem_unit, MEMORY_UNIT_PRECISION + ) + total_mem_len = len(str(memory_total)) + header_mem_total = "{}".format(memory_total) + + header_mem_assigned_max_total = "{}({}%) {}".format( + str(sum_memory_assigned_max_total).rjust(total_mem_len), + pct_memory_assigned_max_total, + assigned_text, + ) + header_mem_used_noswap = "{}({}%) {}".format( + str(sum_memory_used_noswap).rjust(total_mem_len), + pct_memory_used_noswap, + used_text, + ) + header_mem_used_swap = "{}({}%) {}".format( + str(sum_memory_used_swap).rjust(total_mem_len), + pct_memory_used_swap, + swap_text, + ) + header_mem_suffix = ": " + header_mem_values = [ + header_mem_total, + header_mem_assigned_max_total, + header_mem_used_noswap, + header_mem_used_swap, + ] + header_mem_suffix += ", ".join(header_mem_values) + + state_parts = ["{}".format(total_states)] + state_parts.append( + "({} {})".format( + str(len(self.selected_stats)).rjust(total_states_len), selected_text + ) + ) + for state in all_states: + if self.thin_columns: + pretty_state = POWER_STATES[state]["short"] + else: + pretty_state = state.lower() if state != "NA" else "NA" + state_parts.append( + "{} {}".format( + str(state_counts.get(state, 0)).rjust(total_states_len), + pretty_state, + ) + ) + extra = [state for state in state_counts if state not in all_states] + for state in sorted(extra): + state_parts.append( + "{} {}".format( + str(state_counts[state]).rjust(total_states_len), + state.lower(), + ) + ) + header_dom_prefix = domain_text + header_dom_suffix = ": " + ", ".join(state_parts) + + current_time = strftime("%H:%M:%S") + if self.sort_col_index is not None: + sort_col_index = self.sort_col_index + else: + sort_col_index = 0 + sorted_col = list(self.columns.values())[sort_col_index] + sorted_header = str( + sorted_col.justify( + sorted_col.header, self.column_width[sorted_col.machine_header] + ) + ) + sorted_header += sort_sign + + header_desc_prefix = top_text + header_desc_suffix = f": {self.version} - {current_time}" + scroll_hint = ( + f"{scroll_start + 1}-{scroll_end}/{total_items}" + if total_items > 0 + else "0/0" + ) + + self.write( + 0, + 0, + header_desc_prefix, + max_width=width, + attr=self.header_summary_attr, + ) + self.write( + 0, + len(header_desc_prefix), + header_desc_suffix, + max_width=width - len(header_desc_prefix), + ) + + self.write( + 1, + 0, + header_dom_prefix, + max_width=width, + attr=self.header_summary_attr, + ) + self.write( + 1, + len(header_dom_prefix), + header_dom_suffix, + max_width=width - len(header_dom_prefix), + ) + + header_resources_start = 0 + self.write( + 2, + 0, + header_mem_prefix, + max_width=width, + attr=self.header_summary_attr, + ) + header_resources_start += len(header_mem_prefix) + self.write( + 2, + header_resources_start, + header_mem_suffix, + max_width=width - header_resources_start, + ) + header_resources_start += len(header_mem_suffix) + self.write( + 2, + header_resources_start, + header_cpu_prefix, + max_width=width - header_resources_start, + attr=self.header_summary_attr, + ) + header_resources_start += len(header_cpu_prefix) + self.write( + 2, + header_resources_start, + header_cpu_suffix, + max_width=width - header_resources_start, + ) + + action_headers = "" + if self.act and self.actions: + action_headers = "ACTION".ljust(ACTION_ALL_WIDTHS) + pre_headers = " ".join( + col.justify(col.header, self.column_width[col.machine_header]) + for col in list(self.columns.values())[:sort_col_index] + ) + if sort_col_index > 0: + pre_headers += " " + post_headers = " ".join( + col.justify(col.header, self.column_width[col.machine_header]) + for col in list(self.columns.values())[sort_col_index + 1 :] + ) + if post_headers: + post_headers += " " + self.log.debug("draw-header-pre=%r", pre_headers) + self.log.debug("draw-header-sorted=%r", sorted_header) + self.log.debug("draw-header-post=%r", post_headers) + header_start = 0 + if action_headers: + self.write( + self.header_row, + 0, + action_headers, + max_width=width - header_start, + attr=self.header_attr, + ) + header_start += len(action_headers) + self.write( + self.header_row, + header_start, + " ", + max_width=width - header_start, + ) + header_start += 1 + self.write( + self.header_row, + header_start, + pre_headers, + max_width=width - header_start, + attr=self.header_attr, + ) + header_start += len(pre_headers) + self.log.debug("pre-sort-start=%s", header_start) + if header_start < width: + self.write( + self.header_row, + header_start, + sorted_header, + max_width=width - header_start, + attr=self.header_sel_attr, + ) + header_start += len(sorted_header) + self.log.debug("post-sort-start=%s", header_start) + if len(post_headers.strip()) > 0 and header_start < width: + self.write( + self.header_row, + header_start, + post_headers, + max_width=width - header_start, + attr=self.header_attr, + ) + header_end = header_start + len(post_headers) + else: + header_end = header_start + + current_headers_len = header_end + if self.actions: + current_headers_len += len(action_headers) + filter_prefix = "Filter: " + if self.filter or self.filter_query: + if self.filter: + footer_base = filter_prefix + else: + footer_base = filter_prefix.upper() + footer_filter = self.filter_query + + footer_start = 0 + self.write( + height - 1, + 0, + footer_base, + max_width=width, + attr=self.footer_attr, + ) + footer_start += len(footer_base) + self.write( + height - 1, + footer_start, + footer_filter, + max_width=width - footer_start, + attr=self.footer_sel_attr, + ) + footer_start += len(footer_filter) + if self.filter: + self.write( + height - 1, + footer_start, + " ", + max_width=width - footer_start, + attr=curses.A_REVERSE, + ) + footer_start += 1 + space_between = ( + min(width, current_headers_len) + - footer_start + - len(scroll_hint) + ) + footer_suffix = " " * space_between + scroll_hint + self.write( + height - 1, + footer_start, + footer_suffix, + max_width=width - footer_start, + attr=self.footer_attr, + ) + + else: + footer_base = "" + space_between = ( + min(width, current_headers_len) + - len(footer_base) + - len(scroll_hint) + ) + footer = footer_base + (" " * space_between) + scroll_hint + self.write( + height - 1, + 0, + footer, + max_width=width, + attr=self.footer_attr, + ) + + self.stdscr.refresh() + + def select_row(self, row) -> bool | None: + """ + React to selecting and deselecting a row. + """ + if ( + row in self.visible_stats + and self.visible_stats[row] in self.selected_stats + ): + self.log.debug("unlick-row=%s", row) + self.selected_stats.remove(self.visible_stats[row]) + return True + + if ( + row != self.header_row + and row in self.visible_stats + and self.visible_stats[row].get_actions() + ): + self.log.debug("click-row=%s", row) + self.selected_stats.append(self.visible_stats[row]) + return True + return None + + def line_scroll(self, upward: bool): + """ + React to clicking on a row, selecting an deselecting line. + """ + if upward: + self.cursor_row -= 1 + if self.cursor_row == self.header_row: + self.cursor_row = self.old_cursor_row + if self.act: + self.action_scroll_offset = max( + 0, self.action_scroll_offset - 1 + ) + else: + self.scroll_offset = max(0, self.scroll_offset - 1) + else: + self.cursor_row += 1 + if self.act: + bottom = self.body_top + len(self.visible_actions) + visible: dict[int, str] | dict[int, Stats] = ( + self.visible_actions + ) + else: + bottom = self.body_bottom + visible = self.visible_stats + if self.cursor_row == bottom: + self.cursor_row = self.old_cursor_row + if self.act: + self.action_scroll_offset = min( + self.action_max_offset, self.action_scroll_offset + 1 + ) + else: + self.scroll_offset = min( + self.max_offset, self.scroll_offset + 1 + ) + elif self.cursor_row > len(visible) + self.header_row: + self.cursor_row = self.old_cursor_row + + def page_scroll(self, upward: bool, half: bool = False): + """ + Scroll a page up to down and even half in the chosen direction. + """ + if upward: + self.scroll_offset = max(0, self.scroll_offset - self.body_height) + if half: + self.scroll_offset = self.scroll_offset // 2 + else: + self.scroll_offset = min( + self.max_offset, self.scroll_offset + self.body_height + ) + if half: + self.scroll_offset = self.scroll_offset // 2 + + def cancel_filter(self) -> None: + """ + Cancel filter mode. + """ + self.filter = False + self.cursor_row = self.body_top + + def cancel_act(self) -> None: + """ + Cancel action mode. + """ + self.act = False + self.cursor_row = -1 + + def getch(self) -> bool | None: + """ + Act based on received key, if any. Returns ``True`` if screen should be + refreshed. + """ + # pylint: disable=too-many-return-statements,too-many-statements,too-many-branches + + char = self.stdscr.getch() + if char == -1: + return None + + self.log.debug("char: %s", char) + + if self.filter: + if 32 <= char <= 126: + self.filter_query += chr(char) + elif char in (curses.KEY_BACKSPACE,): + self.filter_query = self.filter_query[:-1] + else: + if char not in ( + curses.KEY_ENTER, + CR, + ): + self.filter_query = "" + self.cancel_filter() + return True + + if self.act: + if char in (curses.KEY_ENTER, CR): + action: str = self.actions[self.cursor_row - self.body_top] + qubes = [stat.vm for stat in self.get_selection()] + self.log.info( + "Running action '%s' on domain%s: %s", + action, + "s" if len(qubes) > 1 else "", + ", ".join(qube.name for qube in qubes), + ) + command = ACTIONS[action]["action"] + self.cancel_act() + ensure_future(command(qubes)) + return True + if char in (ord("a"),): + self.cancel_act() + return True + if char in [ord(str(n)) for n in ACTION_NUMBERS]: + action = [ + action + for action in self.actions + if str(ACTIONS[action]["identity"]) == chr(char) + ][0] + action_index = self.actions.index(action) + self.cursor_row = self.body_top + action_index + return True + + if char in (ord("q"), ord("Q"), ESC): + raise KeyboardInterrupt + + old_sort_col_index = self.sort_col_index + old_scroll_offset = self.scroll_offset + self.old_cursor_row = self.cursor_row + old_selected_stats = self.selected_stats + + if char in (curses.KEY_UP, ord("k"), DLE): + self.line_scroll(upward=True) + + elif char in (curses.KEY_DOWN, ord("j"), SO): + self.line_scroll(upward=False) + + elif char in (curses.KEY_PPAGE, NAK, STX): + self.page_scroll(upward=True, half=char == NAK) + + elif char in (curses.KEY_NPAGE, ACK, EOT): + self.page_scroll(upward=False, half=char == EOT) + + elif char in (curses.KEY_HOME, SOH, ord("g")): + self.scroll_offset = 0 + + elif char in (curses.KEY_END, ENQ, ord("G")): + self.scroll_offset = self.max_offset + + elif char in (FF,): + self.stdscr.clearok(True) + + elif char in (curses.KEY_LEFT, ord("h")): + if self.sort_col_index is None: + self.sort_col_index = -1 + else: + self.sort_col_index -= 1 + if self.sort_col_index == -1: + self.sort_col_index = len(self.columns) - 1 + if not bool(self.sort_col_index != old_sort_col_index): + return None + self.log.debug("sort-col=%s", self.sort_col_index) + return True + + elif char in (curses.KEY_RIGHT, ord("l")): + if self.sort_col_index is None: + self.sort_col_index = 1 + else: + self.sort_col_index += 1 + if self.sort_col_index > len(self.columns) - 1: + self.sort_col_index = 0 + if not bool(self.sort_col_index != old_sort_col_index): + return None + self.log.debug("sort-col=%s", self.sort_col_index) + return True + + elif char in (ord("r"),): + self.sort_reverse = not self.sort_reverse + + elif char in (ord("S"),): + self.show_halted = not self.show_halted + + elif char in (SP,) and not self.act: + return self.select_row(self.cursor_row) + + elif char in (ord("U"),): + self.selected_stats = [] + + elif char in (ord("T"),): + if self.visible_stats: + first_visible = list(self.visible_stats.values())[0] + if first_visible in self.selected_stats: + self.selected_stats = [ + stat + for stat in self.selected_stats + if stat in self.visible_stats + ] + else: + self.selected_stats += [ + stat + for stat in self.visible_stats.values() + if stat not in self.selected_stats + ] + + elif char in (ord("a"),): + if self.cursor_row in self.visible_stats: + self.last_selected_row = self.cursor_row + self.last_selected_stat = self.visible_stats[ + self.last_selected_row + ] + self.log.debug( + "last-stat(row)=%s(%s)", + self.last_selected_stat.vm.name, + self.last_selected_row, + ) + if self.get_action_from_selection(): + self.cursor_row = self.body_top + self.act = True + return True + if self.act: + self.act = False + return True + + elif not self.act and char in (ord("/"),): + self.filter = True + + elif char == curses.KEY_MOUSE: + try: + _, mouse_col, mouse_row, _, button_state = curses.getmouse() + except curses.error: + return None + + if button_state & curses.BUTTON4_PRESSED: + self.page_scroll(upward=True, half=True) + + elif button_state & curses.BUTTON5_PRESSED: + self.page_scroll(upward=False, half=True) + + elif button_state & curses.BUTTON1_DOUBLE_CLICKED: + return self.select_row(mouse_row) + + elif button_state & curses.BUTTON1_CLICKED: + if self.body_top <= mouse_row <= self.body_bottom: + self.cursor_row = mouse_row + else: + col_index = self.get_col_index(mouse_col) + if col_index is None: + return None + if self.sort_col_index == col_index: + self.sort_reverse = not self.sort_reverse + else: + if self.sort_col_index is None: + reverse = not self.sort_reverse + else: + reverse = False + self.sort_col_index = col_index + self.sort_reverse = reverse + self.log.debug( + "sort-col=%s, reverse=%s", + self.sort_col_index, + self.sort_reverse, + ) + return True + + scrolled = bool( + self.scroll_offset != old_scroll_offset + or self.cursor_row != self.old_cursor_row + or self.selected_stats != old_selected_stats + ) + return scrolled + + def get_col_index(self, col) -> int | None: + """ + Returns the index of the specified column. + """ + pos = 0 + # The +1 is to consider space between columns. + for i, width in enumerate( + max(column.min_width, self.column_width[machine_header]) + 1 + for machine_header, column in self.columns.items() + ): + if pos <= col < pos + width: + return i + pos += width + return None + + async def paint(self) -> None: + """ + Draw a top-like table when necessary, and get input to react on quit or + body scrolling. + """ + scrolled: bool | None = False + while True: + try: + self.log.debug("loop") + uptodate = self.is_uptodate() + self.log.debug("uptodate=%s, scrolled=%s", uptodate, scrolled) + if scrolled or not uptodate: + self.log.debug("draw") + self.draw_table() + self.log.debug("getch") + scrolled = self.getch() + self.log.debug("scrolled=%s", scrolled) + if not uptodate or scrolled or scrolled is False: + self.log.debug( + "urgent need redraw, sleeping for %ss", + self.outdated_sleep, + ) + # Sleep a bit to avoid unhappy CPU. + await sleep(self.outdated_sleep) + continue + self.log.debug( + "not rushing redraw, sleeping for %ss", + self.outdated_sleep, + ) + await sleep(self.uptodate_sleep) + continue + except (KeyboardInterrupt, CancelledError): + if self.filter: + self.filter_query = "" + self.cancel_filter() + continue + if self.act: + self.cancel_act() + continue + break + + +class Monitor: + """ + Monitor and react to events. + """ + + entries: dict[QubesVM, Stats] = {} + domains: list[QubesVM] + + def __init__( + self, + app: QubesBase, + domains: list[QubesVM], + dispatcher: EventsDispatcher, + stats_dispatcher: EventsDispatcher, + show_halted: bool, + all_domains: bool, + allow_color: bool, + columns: dict[str, "Column"], + sort_reverse: bool = False, + sort_column: str | None = None, + filter_query: str = "", + thin_columns: bool = False, + ) -> None: + # pylint: disable=too-many-positional-arguments + self.app = app + self.__class__.domains = domains + self.dispatcher = dispatcher + self.stats_dispatcher = stats_dispatcher + self.all_domains = all_domains + sort_col_index: int | None = None + if sort_column is not None: + sort_col_index = list(columns.keys()).index(sort_column) + self.screen = Screen( + app=self.app, + show_halted=show_halted, + allow_color=allow_color, + filter_query=filter_query, + sort_reverse=sort_reverse, + sort_col_index=sort_col_index, + columns=columns, + thin_columns=thin_columns, + ) + self.log = gen_logger(f"{self.__class__.__qualname__}") + root_logger = getLogger() + root_logger.setLevel(LOGGING_LEVEL) + root_logger.handlers.clear() + root_logger.addHandler(self.log.handlers[0]) + + def init_entries(self) -> None: + """ + Initialize statistics holder for all qubes. + """ + for vm in sorted( + [vm for vm in self.app.domains if vm.klass == "AdminVM"] + ): + self.add_domain(submitter=None, vm=vm, event=None) + for vm in sorted( + [vm for vm in self.app.domains if vm not in self.__class__.entries] + ): + self.add_domain(submitter=None, vm=vm, event=None) + + def update_stats(self, vm: QubesVM, _event: str, **kwargs: Any) -> None: + """ + Read vm-stats and update qube statistics. + """ + if vm not in self.__class__.entries: + return + self.__class__.entries[vm].update_stats(**kwargs) + + def refresh_all_items( + self, submitter: QubesVM | None, _event: str, **_kwargs: Any + ) -> None: + """ + Reconnected to the API, check if all entries are up to date. + """ + # pylint: disable=unused-argument + items_to_delete = [ + vm for vm in self.__class__.entries if vm not in self.app.domains + ] + for vm in items_to_delete: + self.remove_domain(submitter=None, vm=vm, event=None) + for vm in self.app.domains: + self.update_domain_item(vm=vm, event=None) + + def add_domain( + self, + submitter: QubesVM | None, + event: str | None, + vm: QubesVM, + **kwargs: Any, + ) -> None: + """ + Add qube to the statistics holder. + """ + # pylint: disable=unused-argument + if str(vm) in self.__class__.entries: + return + try: + vm = self.app.domains[str(vm)] + except KeyError: + return + + self.log.debug("add=%s", str(vm)) + try: + self.__class__.entries[vm] = Stats(vm) + except QubesVMNotFoundError: + self.log.debug("Qube removed while adding its entry: %s", str(vm)) + self.remove_domain(submitter=None, vm=vm, event=event) + if self.all_domains: + self.__class__.domains.append(vm) + + def remove_domain( + self, + submitter: QubesVM | None, + event: str | None, + vm: QubesVM, + **kwargs: Any, + ) -> None: + """ + Remove qube from the statistics holder. + """ + # pylint: disable=unused-argument + if str(vm) not in self.__class__.entries: + return + self.log.debug("remove=%s", str(vm)) + if vm in self.__class__.domains: + Stats.outdated_by_removal.append(str(vm)) + if self.__class__.entries[vm] in self.screen.selected_stats: + self.screen.selected_stats.remove(self.__class__.entries[vm]) + del self.__class__.entries[vm] + + def get_entry(self, vm: QubesVM, event: str | None) -> Stats | None: + """ + Safely get ``Stats`` if qube can be accessed, else, remove the entry and + return ``None``. + """ + try: + item = self.__class__.entries[vm] + except QubesPropertyAccessError as e: + self.log.exception(e) + self.remove_domain(submitter=None, vm=vm, event=event) + return None + return item + + def update_domain_item( + self, vm: QubesVM, event: str | None, **kwargs: Any + ) -> None: + """ + Fully update the statistics holder of the qube. + """ + # pylint: disable=unused-argument + try: + item = self.get_entry(vm=vm, event=event) + if item is None: + return + except KeyError: + self.add_domain(submitter=None, vm=vm, event=event) + if not vm in self.__class__.entries: + return + item = self.__class__.entries[vm] + try: + item.update_cache() + except QubesVMNotFoundError: + self.log.debug("Qube removed while updating cache: %s", str(vm)) + self.remove_domain(submitter=None, vm=vm, event=event) + + def set_generic( + self, vm, event, name, newvalue=None, oldvalue=None + ) -> None: + """ + Update attributes of the qube's statistics holder. + """ + # pylint: disable=unused-argument + if not (item := self.get_entry(vm=vm, event=event)): + return + self.log.debug("update -> %s=%s(%s)", name, vm.name, newvalue) + if name == "maxmem": + item.update_memory_max(value=newvalue) + return + if name == "memory": + item.update_memory_init(value=newvalue) + return + if name == "label": + newvalue = self.screen.init_label(label=newvalue) + item.update_generic(name=name, value=newvalue) + + def set_feat_generic( + self, vm, event, feature, value, oldvalue=None + ) -> None: + """ + Update feature attributes of the qube's statistics holder. + """ + # pylint: disable=unused-argument + if not (item := self.get_entry(vm=vm, event=event)): + return + self.log.debug("update -> %s=%s(%s)", feature, vm.name, value) + item.update_generic(name=feature, value=value) + for qube in vm.derived_vms: + if qube not in self.__class__.entries: + continue + self.__class__.entries[qube].update_feat_from_template(name=feature) + + def del_feat_generic(self, vm, event, feature) -> None: + """ + Update feature attributes of the qube's statistics holder. + """ + # pylint: disable=unused-argument + if not (item := self.get_entry(vm=vm, event=event)): + return + self.log.debug("update del -> %s=%s", feature, vm.name) + item.update_feat_from_template(name=feature) + for qube in vm.derived_vms: + if qube not in self.__class__.entries: + continue + self.__class__.entries[qube].update_feat_from_template(name=feature) + + async def run(self) -> None: + """ + Run initial setup. + """ + self.register_events() + self.init_entries() + exc_data = None + try: + # Allow registering events. + await sleep(0) + self.screen.init_screen() + await self.screen.paint() + except BaseException as e: + exc_data = exc_info() + self.log.exception(e) + finally: + self.unregister_events() + self.screen.restore_screen() + if exc_data: + print_exception(*exc_data, file=stderr) + + def register_events(self): # type: ignore[list-item] + """ + Track and set event handlers. + """ + stats_handlers = [("vm-stats", self.update_stats)] + handlers = [ + ("connection-established", self.refresh_all_items), + ("domain-add", self.add_domain), + ("domain-delete", self.remove_domain), + ("property-set:maxmem", self.set_generic), + ("property-set:memory", self.set_generic), + ("property-set:label", self.set_generic), + ("property-set:auto_cleanup", self.set_generic), + ("property-reset:is_preload", self.set_generic), + ("property-set:guivm", self.set_generic), + ("domain-feature-set:gui", self.set_feat_generic), + ("domain-feature-delete:gui", self.del_feat_generic), + ("domain-feature-set:internal", self.set_feat_generic), + ("domain-feature-delete:internal", self.del_feat_generic), + ] + for event in POWER_EVENTS: + handlers.append((event, self.update_domain_item)) + for event, handler in stats_handlers: + self.stats_dispatcher.add_handler(event=event, handler=handler) + for event, handler in handlers: + self.dispatcher.add_handler(event=event, handler=handler) + + def unregister_events(self) -> None: + """ + Stop watchers. + """ + self.stats_dispatcher.stop() + self.dispatcher.stop() + + +class Column: + """ + Column store. + """ + + columns: dict[str, "Column"] = {} + + def __init__( + self, + header: str, + min_header: str | None = None, + min_width: int | Callable = 0, + internal: bool = False, + total: bool = False, + machine_header: str = "", + doc: str | None = None, + right_justify: bool = True, + percentage: bool = False, + percentage_intensity: list[int] | None = None, + summable_for_overall: bool = False, + untrusted: bool = False, + ): + # pylint: disable=too-many-positional-arguments + self.internal = internal + self.total = total + self.percentage = percentage + self.summable_for_overall = summable_for_overall + self.untrusted = untrusted + self.min_header = min_header + if percentage_intensity is None: + self.percentage_intensity = [75, 50] + else: + self.percentage_intensity = percentage_intensity + self.__doc__ = doc + + self.header = header + if machine_header: + self.machine_header = machine_header + else: + self.machine_header = self.header.strip() + if isinstance(min_width, int): + self.min_width = min_width + else: + self.min_width = min_width(header) + self.min_width = max(self.min_width, len(header)) + if right_justify: + self.justify = lambda x, w: x.rjust(max(w, self.min_width)) + else: + self.justify = lambda x, w: x.ljust(max(w, self.min_width)) + self.columns[self.machine_header] = self + + +class _HelpColumnsAction(Action): + """Action for argument parser that displays all columns and exits.""" + + # pylint: disable=redefined-builtin + def __init__( + self, + option_strings, + dest=SUPPRESS, + default=SUPPRESS, + help="list all available columns with short descriptions and exit", + ): + super().__init__( + option_strings=option_strings, + dest=dest, + default=default, + nargs=0, + help=help, + ) + + def __call__(self, _parser, _namespace, _values, option_string=None): + width = max(len(column.header) for column in Column.columns.values()) + wrapper = TextWrapper( + width=800, initial_indent=" ", subsequent_indent=" " * (width + 6) + ) + + text = ( + "Available columns (machine_header -> header Description):\n" + + "\n".join( + wrapper.fill( + "{header:{width}s} {doc}".format( + header="{} -> {}".format( + column.machine_header, column.header.strip() + ), + doc=column.__doc__ or "", + width=width, + ) + ) + for column in Column.columns.values() + ) + ) + print(text) + sys_exit(0) + + +class _HelpFormatsAction(Action): + """Action for argument parser that displays all formats and exits.""" + + # pylint: disable=redefined-builtin + def __init__( + self, + option_strings, + dest=SUPPRESS, + default=SUPPRESS, + help="list all available formats with short descriptions and exit", + ): + super().__init__( + option_strings=option_strings, + dest=dest, + default=default, + nargs=0, + help=help, + ) + + def __call__(self, _parser, _namespace, _values, option_string=None): + width = max(len(fmt) for fmt in FORMATS) + text = "Available formats:\n" + "".join( + " {fmt:{width}s} {columns}\n".format( + fmt=fmt, columns=",".join(columns), width=width + ) + for fmt, columns in FORMATS.items() + ) + print(text, end="") + sys_exit(0) + + +UNTRUSTED_COLUMN = ( + "This value or part of it is broadcast by the qube, it can be a lie." +) + +# Basic +Column( + header="NAME", + machine_header="name", + min_width=31, + doc="Qube's name.", + right_justify=False, +) +Column( + header="STATE", + min_header="S", + machine_header="state", + min_width=max(len(state) for state in POWER_STATES), + doc="Qube's power state.", + right_justify=False, +) + +# Totals +Column( + header="MI", + machine_header="memory_init", + min_width=lambda header: max(len(header), 5), + doc="How much memory the system must reserve for the qube to be able to " + "initialize. On non-memory-balanced qubes, this is the maximum amount of " + "memory a domain will ever have while it is running.", +) +Column( + header="MM", + machine_header="memory_max", + min_width=lambda header: max(len(header), 5), + doc="How much memory the qube can use from the system. Part of this value " + "is reserved to videoram, while the rest is up to qmemman to balloon up " + "this qube when there is enough free memory on the host.", +) + +Column( + header="MAMT", + machine_header="memory_assigned_max_total", + min_width=lambda header: max(len(header), 5), + doc="``MAM`` + ``MAMi``.", + total=True, + summable_for_overall=True, +) +Column( + header="MAUT", + machine_header="memory_assigned_usable_total", + min_width=lambda header: max(len(header), 5), + doc="``MAU`` + ``MAUi``.", + total=True, + summable_for_overall=True, +) +Column( + header="MUT", + machine_header="memory_used_total", + min_width=lambda header: max(len(header), 5), + doc="``MU`` + ``MUi``.", + total=True, + summable_for_overall=True, +) +Column( + header="CPUsecT", + machine_header="cpu_time_total", + doc="``CPUsec`` + ``CPUisec``.", + total=True, + summable_for_overall=True, +) +Column( + header="VCT", + machine_header="online_vcpus_total", + min_width=lambda header: max(len(header), 3), + doc="``VC`` + ``VCi``.", + total=True, + summable_for_overall=True, +) + +# Standard +Column( + header="MAM", + machine_header="memory_assigned_max", + min_width=lambda header: max(len(header), 5), + doc="How much memory has been assigned to the qube, including overhead.", + summable_for_overall=True, +) +Column( + header="MAU", + machine_header="memory_assigned_usable", + min_width=lambda header: max(len(header), 5), + doc="How much memory has been assigned to the qube and can be used. A qube " + "is allowed to claim this amount at any time, and it cannot use more memory" + " than what has been assigned to it. When the system is under no memory " + "pressure, this value is close to ``MM``, while when the system is under " + "memory pressure, the value can be as low as enough for the qube to " + "survive.", + summable_for_overall=True, +) +Column( + header="MU", + machine_header="memory_used_noswap", + min_width=lambda header: max(len(header), 5), + doc="How much memory the qube alleges to use. " + UNTRUSTED_COLUMN, + summable_for_overall=True, + untrusted=True, +) +Column( + header="MUS", + machine_header="memory_used_swap", + min_width=lambda header: max(len(header), 5), + doc="How much memory the qube alleges to use for swap. " + UNTRUSTED_COLUMN, + summable_for_overall=True, + untrusted=True, +) +Column( + header="MAM/MM", + machine_header="memory_usage_assigned", + min_width=lambda header: max(len(header), 5), + doc="How much memory the qube has assigned compared to the maximum it can " + "have assigned, in percentage. A high percentage means the system is not " + "pressuring the qube to release memory.", + percentage=True, + percentage_intensity=[], +) +Column( + header="MU/MAU", + machine_header="memory_usage_used", + min_width=lambda header: max(len(header), 5), + doc="How much memory the qube alleges to use from the assigned amount, in " + "percentage. A high percentage on non-memory-balanced qubes is irrelevant. " + "On memory balanced qubes, a higher value indicates the qube is using a lot" + " of the memory it has assigned, which might be near exhaustion, if ``MAU``" + " can't be ballooned up anymore.", + percentage=True, + untrusted=True, +) +Column( + header="MUS/MU", + machine_header="memory_usage_swap_over_used", + min_width=lambda header: max(len(header), 5), + doc="How much memory the qube alleges to be swaping from what it alleges to" + "use, in percentage. When it is over 10%, the qube might be swaping too " + "much.", + percentage=True, + percentage_intensity=[30, 10], + untrusted=True, +) +Column( + header="CPUsec", + machine_header="cpu_time", + doc="How many seconds the qube has used from the CPU.", + summable_for_overall=True, +) +Column( + header="CPU%", + machine_header="cpu_usage", + min_width=lambda header: max(len(header), 5), + doc="How much CPU the qube is using, in percentage.", + percentage=True, +) +Column( + header="VC", + machine_header="online_vcpus", + min_width=lambda header: max(len(header), 3), + doc="How many Virtual CPUs are online.", + summable_for_overall=True, +) + +# Internal +Column( + header="MAMi", + machine_header="memory_assigned_max_internal", + min_width=lambda header: max(len(header), 5), + doc="Same as ``MAM``, but internal usage.", + internal=True, + summable_for_overall=True, +) +Column( + header="MAUi", + machine_header="memory_assigned_usable_internal", + min_width=lambda header: max(len(header), 5), + doc="Same as ``MAU``, but internal usage.", + internal=True, + summable_for_overall=True, +) +Column( + header="MUi", + machine_header="memory_used_noswap_internal", + min_width=lambda header: max(len(header), 5), + doc="Same as ``MU``, but internal usage.", + internal=True, + summable_for_overall=True, +) +Column( + header="CPUisec", + machine_header="cpu_time_internal", + doc="Same as ``CPUsec``, but internal usage.", + internal=True, + summable_for_overall=True, +) +Column( + header="CPUi%", + machine_header="cpu_usage_internal", + min_width=lambda header: max(len(header), 5), + doc="Same as ``CPU%``, but internal usage.", + percentage=True, + internal=True, +) +Column( + header="VCi", + machine_header="online_vcpus_internal", + min_width=lambda header: max(len(header), 3), + doc="Same as ``VC``, but internal usage.", + internal=True, + summable_for_overall=True, +) + + +FORMATS: dict[str, list[str]] = { + "min": ["name", "state", "memory_usage_used", "cpu_usage"], + "default": [ + "name", + "state", + "memory_assigned_usable", + "memory_used_noswap", + "memory_used_swap", + "memory_usage_used", + "online_vcpus", + "cpu_usage", + ], + "max-no-internal": [ + machine_header + for machine_header, column in list(Column.columns.items()) + if not column.internal + ], + "max-no-total": [ + machine_header + for machine_header, column in list(Column.columns.items()) + if not column.total + ], + "max-no-total-no-internal": [ + machine_header + for machine_header, column in list(Column.columns.items()) + if not column.internal and not column.total + ], + "max": list(Column.columns.keys()), +} + + +def column_multiple_choice(value: str): + """ + Validate CSV column values. + """ + if bad := [x for x in value.split(",") if x not in Column.columns]: + raise ArgumentTypeError("Invalid choice(s): {}".format(", ".join(bad))) + return value + + +parser = QubesArgumentParser( + description=__doc__ + " Defaults is to show all non-halted qubes.", + vmname_nargs="*", + all_default=True, + all_include_adminvm=True, +) +parser.add_argument( + "--show-halted", + "-S", + action="store_true", + default=False, + help="don't hide halted qubes", +) +parser.add_argument( + "--filter", + "-f", + metavar="FILTER,...", + action="store", + default="", + help="filter domains name matching each fixed string separated by comma", +) +parser.add_argument( + "--thin-columns", + action="store_true", + default=False, + help="Columns will to be as large as the its header or current lengthiest " + "data. It is not the default because columns will move every time the " + "length of the lengthiest data change.", +) +parser.add_argument( + "--memory-unit", + action="store", + choices=MEMORY_UNIT_AVAILABLE, + default=MEMORY_UNIT, + metavar="UNIT", + help="Unit to use for displaying memory. The digit after the dot is the " + "precision. Available: " + ", ".join(MEMORY_UNIT_AVAILABLE), +) + +parser_format = parser.add_argument_group(title="formatting options") +parser_format_exclusive = parser_format.add_mutually_exclusive_group() + +parser_format_exclusive.add_argument( + "--columns", + "-C", + metavar="COLUMN,...", + action="store", + type=column_multiple_choice, + help="show only specified columns", +) +parser_format_exclusive.add_argument( + "--format", + "-F", + metavar="FORMAT", + action="store", + choices=FORMATS.keys(), + help="show only columns declared by format: " + ", ".join(FORMATS), +) +parser_format.add_argument("--help-columns", action=_HelpColumnsAction) +parser_format.add_argument("--help-formats", action=_HelpFormatsAction) + +parser_sort = parser.add_argument_group(title="sorting options") +parser_sort.add_argument( + "--sort-column", + "-k", + metavar="COLUMN", + action="store", + choices=Column.columns, + help="sort by specified column", +) +parser_sort.add_argument( + "--reverse", + "-r", + action="store_true", + help="reverse sorting", +) +parser.add_argument( + "--no-color", + action="store_true", + help="do not colorize the screen", +) + + +async def run_async(args) -> int: + """ + Dispatch events, display statistics and wait for exit. + """ + global MEMORY_UNIT # pylint: disable=global-statement + global MEMORY_UNIT_PRECISION # pylint: disable=global-statement + MEMORY_UNIT = args.memory_unit.split(".")[0] + if "." in args.memory_unit: + MEMORY_UNIT_PRECISION = int(args.memory_unit.split(".")[1]) + if args.columns: + user_columns = [col.strip() for col in args.columns.split(",")] + elif args.format: + user_columns = [col.strip() for col in FORMATS[args.format]] + else: + user_columns = [col.strip() for col in FORMATS["default"]] + columns = {col: Column.columns[col] for col in user_columns} + if args.thin_columns: + for col, column in columns.items(): + if column.min_header: + columns[col].header = column.min_header + columns[col].min_width = len(column.header) + + sort_column = args.sort_column or None + + app = args.app + app.cache_enabled = True + dispatcher = EventsDispatcher(app) + stats_dispatcher = EventsDispatcher(app, api_method="admin.vm.Stats") + top = Monitor( + all_domains=args.all_domains, + app=app, + domains=args.domains, + dispatcher=dispatcher, + stats_dispatcher=stats_dispatcher, + show_halted=args.show_halted, + sort_column=sort_column, + sort_reverse=args.reverse, + columns=columns, + allow_color=not (args.no_color or NO_COLOR), + filter_query=args.filter, + thin_columns=args.thin_columns + ) + tasks = [ + create_task(dispatcher.listen_for_events()), + create_task(stats_dispatcher.listen_for_events()), + create_task(top.run()), + ] + await gather(*tasks) + return 0 + + +def main(args=None, app=None) -> int: + """ + Show top-like statistics for all qubes by default, else, show for the + specified qubes. + """ + try: + args = parser.parse_args(args, app=app) + run(run_async(args=args)) + except KeyboardInterrupt: + pass + return 0 + + +if __name__ == "__main__": + sys_exit(main()) diff --git a/qubesadmin/vm/__init__.py b/qubesadmin/vm/__init__.py index e9d353590..acde6cdb0 100644 --- a/qubesadmin/vm/__init__.py +++ b/qubesadmin/vm/__init__.py @@ -44,8 +44,17 @@ # ["AppVM", "AdminVM", "TemplateVM", "DispVM", "StandaloneVM"] # but can be extended Klass = str -PowerState = Literal["Transient", "Running", "Halted", "Paused", -"Suspended", "Halting", "Dying", "Crashed", "NA"] +POWER_STATES = { + "Running": {"short": "r"}, + "Starting": {"short": "S"}, + "Paused": {"short": "p"}, + "Suspended": {"short": "s"}, + "Halting": {"short": "H"}, + "Halted": {"short": "h"}, + "Crashed": {"short": "c"}, + "NA": {"short": "-"}, +} +PowerState = Literal[POWER_STATES] class QubesVM(qubesadmin.base.PropertyHolder): @@ -196,13 +205,11 @@ def get_power_state(self): return value meaning =============== ======================================================== ``'Halted'`` Machine is not active. - ``'Transient'`` Machine is running, but does not have :program:`guid` - or :program:`qrexec` available. + ``'Starting'`` Machine is in progress of starting. ``'Running'`` Machine is ready and running. ``'Paused'`` Machine is paused. ``'Suspended'`` Machine is S3-suspended. ``'Halting'`` Machine is in process of shutting down (OS shutdown). - ``'Dying'`` Machine is in process of shutting down (cleanup). ``'Crashed'`` Machine crashed and is unusable. ``'NA'`` Machine is in unknown state. =============== ======================================================== @@ -444,7 +451,7 @@ def _get_derived_vms(vm): at any level of inheritance. """ result = set(vm.appvms) - for appvm in vm.appvms: + for appvm in result.copy(): result.update(QubesVM._get_derived_vms(appvm)) return result