diff --git a/conftest.py b/conftest.py index 1f38e95ac..96ba96328 100644 --- a/conftest.py +++ b/conftest.py @@ -5,12 +5,12 @@ import argparse import dataclasses import itertools -import logging import os import tempfile from collections import defaultdict import git +import structlog from cryptography.hazmat.primitives.serialization import SSHCertPrivateKeyTypes from packaging import version @@ -31,6 +31,7 @@ wait_for, ) from lib.host import Host +from lib.logging import configure_logging from lib.netutil import is_ipv6 from lib.pool import Pool from lib.sr import SR @@ -153,6 +154,8 @@ def pytest_configure(config: pytest.Config) -> None: assert write_volume_align is not None global_config.write_volume_align = parse_size(write_volume_align) + configure_logging() + def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: if "vm_ref" in metafunc.fixturenames: vms = metafunc.config.getoption("vm") @@ -308,22 +311,24 @@ def setup_host(hostname_or_ip: str, *, config: pytest.Config | None = None) -> H vif = host_vm.vifs()[0] mac_address = vif.mac_address() - logging.info("Nested host has MAC %s", mac_address) + host_vm.logger.info("Nested host MAC address retrieved", mac_address=mac_address) host_vm.start() - wait_for(host_vm.is_running, "Wait for nested host VM running") + wait_for(host_vm.is_running, "Wait for nested host VM running", logger=host_vm.logger) # catch host-vm IP address wait_for(lambda: pxe.arp_addresses_for(mac_address), "Wait for DHCP server to see nested host in ARP tables", - timeout_secs=10 * 60) + timeout_secs=10 * 60, + logger=host_vm.logger.bind(mac_address=mac_address)) ips = pxe.arp_addresses_for(mac_address) - logging.info("Nested host has IPs %s", ips) + host_vm.logger.info("Nested host IPs retrieved", ips=ips) assert len(ips) == 1 host_vm.ip = ips[0] wait_for(lambda: not os.system(f"nc -zw5 {host_vm.ip} 22"), - "Wait for ssh up on nested host", retry_delay_secs=5) + "Wait for ssh up on nested host", retry_delay_secs=5, + logger=host_vm.logger) hostname_or_ip = host_vm.ip @@ -333,7 +338,7 @@ def setup_host(hostname_or_ip: str, *, config: pytest.Config | None = None) -> H def cleanup_hosts() -> None: for vm in nested_list: - logging.info("Destroying nested host VM %s", vm.uuid) + vm.logger.info("Destroying nested host VM") vm.destroy(verify=True) # a list of master hosts, each from a different pool @@ -375,17 +380,17 @@ def registered_xo_cli() -> None: @pytest.fixture(scope='session') def hosts_with_xo(hosts: list[Host], registered_xo_cli: None) -> Generator[list[Host], None, None]: for h in hosts: - logging.info(">>> Connect host %s" % h) + h.logger.info(">>> Connect host") if not h.skip_xo_config: h.xo_server_add(h.user, h.password) else: h.xo_get_server_id(store=True) - wait_for(h.xo_server_connected, timeout_secs=10) + wait_for(h.xo_server_connected, timeout_secs=10, logger=h.logger) yield hosts # teardown for h in hosts: if not h.skip_xo_config: - logging.info("<<< Disconnect host %s" % h) + h.logger.info("<<< Disconnect host") h.xo_server_remove() @pytest.fixture(scope='session') @@ -403,7 +408,7 @@ def hostA2(hostA1: Host) -> Generator[Host, None, None]: """ Second host of pool A. """ assert len(hostA1.pool.hosts) > 1, "A second host in first pool is required" _hostA2 = hostA1.pool.hosts[1] - logging.info(">>> hostA2 present: %s" % _hostA2) + _hostA2.logger.info(">>> hostA2 present") yield _hostA2 @pytest.fixture(scope='session') @@ -412,7 +417,7 @@ def hostB1(hosts: list[Host]) -> Generator[Host, None, None]: assert len(hosts) > 1, "A second pool is required" assert hosts[0].pool.uuid != hosts[1].pool.uuid _hostB1 = hosts[1] - logging.info(">>> hostB1 present: %s" % _hostB1) + _hostB1.logger.info(">>> hostB1 present") yield _hostB1 @pytest.fixture(scope='session') @@ -448,7 +453,7 @@ def host_no_ipv6(host: Host) -> None: def shared_sr(host: Host) -> Generator[SR, None, None]: sr = host.pool.first_shared_sr() assert sr, "No shared SR available on hosts" - logging.info(">> Shared SR on host present: {} of type {}".format(sr.uuid, sr.get_type())) + sr.logger.info(">> Shared SR on host present: {} of type {}".format(sr.uuid, sr.get_type())) yield sr @pytest.fixture(scope='session') @@ -458,7 +463,7 @@ def local_sr_on_hostA1(hostA1: Host) -> Generator[SR, None, None]: assert len(srs) > 0, "a local SR is required on the pool's master" # use the first local SR found sr = srs[0] - logging.info(">> local SR on hostA1 present: {} of type {}".format(sr.uuid, sr.get_type())) + sr.logger.info(">> local SR on hostA1 present: {} of type {}".format(sr.uuid, sr.get_type())) yield sr @pytest.fixture(scope='session') @@ -468,7 +473,7 @@ def local_sr_on_hostA2(hostA2: Host) -> Generator[SR, None, None]: assert len(srs) > 0, "a local SR is required on the pool's second host" # use the first local SR found sr = srs[0] - logging.info(">> local SR on hostA2 present: {} of type {}".format(sr.uuid, sr.get_type())) + sr.logger.info(">> local SR on hostA2 present") yield sr @pytest.fixture(scope='session') @@ -478,13 +483,15 @@ def local_sr_on_hostB1(hostB1: Host) -> Generator[SR, None, None]: assert len(srs) > 0, "a local SR is required on the second pool's master" # use the first local SR found sr = srs[0] - logging.info(">> local SR on hostB1 present: {} of type {}".format(sr.uuid, sr.get_type())) + sr.logger.info(">> local SR on hostB1 present") yield sr @pytest.fixture(scope='session') def disks(pytestconfig: pytest.Config, pools_hosts_by_name_or_ip: dict[HostAddress, Host] ) -> dict[Host, list[Host.BlockDeviceInfo]]: """Dict identifying names of all disks for on all hosts of first pool.""" + logger = structlog.get_logger("disks").bind(pools=pools_hosts_by_name_or_ip) + def _parse_disk_option(option_text: str) -> tuple[HostAddress, list[DiskDevName]]: parsed = option_text.split(sep=":", maxsplit=1) assert len(parsed) == 2, f"--disks option {option_text!r} is not :[,]*" @@ -538,7 +545,7 @@ def _host_disks(host: Host, hosts_cli_disks: list[DiskDevName] | None) -> Iterab if disk.wwn and not disk.available and not disk.wwn.startswith("uuid.00000000-0000-0000-0000-") } if used_wwns: - logging.debug("cross-host used WWNs: %s", used_wwns) + logger.debug("Resport cross-host used WWNs", used_wwns=used_wwns) ret = { host: [ dataclasses.replace(disk, available=False) if (disk.wwn and disk.wwn in used_wwns) else disk @@ -560,34 +567,42 @@ def _host_disks(host: Host, hosts_cli_disks: list[DiskDevName] | None) -> Iterab except ImportError: pass if reserved_wwns: - logging.debug("reserved WWNs (lvmohba/lvmoiscsi): %s", reserved_wwns) + logger.debug("Report reserved WWNs (lvmohba/lvmoiscsi)", reserved_wwns=reserved_wwns) ret = { host: sorted(host_disks, key=lambda d: d.wwn in reserved_wwns) for host, host_disks in ret.items() } - logging.debug("disks collected: %s", {host.hostname_or_ip: value for host, value in ret.items()}) + logger.debug("Disks collected", collected_disks={host.hostname_or_ip: value for host, value in ret.items()}) return ret @pytest.fixture(scope='session') def unused_512B_disks(disks: dict[Host, list[Host.BlockDeviceInfo]] ) -> dict[Host, list[Host.BlockDeviceInfo]]: """Dict identifying names of all 512-bytes-blocks disks for on all hosts of first pool.""" + logger = structlog.get_logger("unused_512B_disks").bind(disks=disks) ret = {host: [disk for disk in host_disks if disk.log_sec == 512 and disk.available] for host, host_disks in disks.items() } - logging.debug("available disks collected: %s", {host.hostname_or_ip: value for host, value in ret.items()}) + logger.debug( + "Available disks collected", + available_disks={host.hostname_or_ip: value for host, value in ret.items()} + ) return ret @pytest.fixture(scope='session') def unused_4k_disks(disks: dict[Host, list[Host.BlockDeviceInfo]] ) -> dict[Host, list[Host.BlockDeviceInfo]]: """Dict identifying names of all 4K-blocks disks for on all hosts of first pool.""" + logger = structlog.get_logger("unused_4k_disks").bind(disks=disks) ret = {host: [disk for disk in host_disks if disk.log_sec == 4096 and disk.available] for host, host_disks in disks.items() } - logging.debug("available 4k disks collected: %s", {host.hostname_or_ip: value for host, value in ret.items()}) + logger.debug( + "Available 4k disks collected", + available_4k_disks={host.hostname_or_ip: value for host, value in ret.items()} + ) return ret @pytest.fixture(scope='session') @@ -600,6 +615,7 @@ def pool_with_unused_512B_disk(host: Host, unused_512B_disks: dict[Host, list[Ho @pytest.fixture(scope='module') def vm_ref(request: pytest.FixtureRequest) -> str: + logger = structlog.get_logger("vm_ref") ref = request.param if ref is None: @@ -607,11 +623,14 @@ def vm_ref(request: pytest.FixtureRequest) -> str: marker = request.node.get_closest_marker("default_vm") if marker is not None: ref = marker.args[0] - logging.info(">> No VM specified on CLI. Using default: %s.", ref) + logger.info(">> No VM specified on CLI. Using default", default_vm=ref) else: # global default - logging.info(">> No VM specified on CLI, and no default found in test definition. Using global default.") ref = 'mini-linux-x86_64-bios' + logger.info( + ">> No VM specified on CLI, and no default found in test definition. Using global default.", + default_vm=ref, + ) if is_uuid(ref) or ref.startswith('http'): return ref @@ -623,13 +642,13 @@ def imported_vm(host: Host, vm_ref: str) -> Generator[VM, None, None]: if is_uuid(vm_ref): vm_orig = VM(vm_ref, host) name = vm_orig.name() - logging.info(">> Reuse VM %s (%s) on host %s" % (vm_ref, name, host)) + vm_orig.logger.info(">> Reuse VM", vm_name=name) else: vm_orig = host.import_vm(vm_ref, host.main_sr_uuid(), use_cache=CACHE_IMPORTED_VM) if CACHE_IMPORTED_VM: # Clone the VM before running tests, so that the original VM remains untouched - logging.info(">> Clone cached VM before running tests") + vm_orig.logger.info(">> Clone cached VM before running tests") vm = vm_orig.clone() # Remove the description, which may contain a cache identifier vm.param_set('name-description', "") @@ -639,7 +658,7 @@ def imported_vm(host: Host, vm_ref: str) -> Generator[VM, None, None]: yield vm # teardown if CACHE_IMPORTED_VM or not is_uuid(vm_ref): - logging.info("<< Destroy VM") + vm.logger.info("<< Destroy VM") vm.destroy(verify=True) @pytest.fixture(scope="session") @@ -730,11 +749,11 @@ def create_vms(request: pytest.FixtureRequest, host: Host, tests_git_revision: s report = request.node.stash.get(PHASE_REPORT_KEY, None) if report is None: # user interruption during setup - logging.warning("test setup result not available: not exporting VMs") + host.logger.warning("test setup result not available: not exporting VMs") elif report["setup"].failed: - logging.warning("setting up a test failed or skipped: not exporting VMs") + host.logger.warning("setting up a test failed or skipped: not exporting VMs") elif ("call" not in report) or report["call"].failed: - logging.warning("executing test failed or skipped: not exporting VMs") + host.logger.warning("executing test failed or skipped: not exporting VMs") else: # record this state for vm_def, vm in zip(vm_defs, vms): @@ -742,18 +761,18 @@ def create_vms(request: pytest.FixtureRequest, host: Host, tests_git_revision: s vm.save_to_cache(f"{nodeid}-{vm_def['name']}-{tests_git_revision}") except Exception: - logging.error("exception caught...") + host.logger.exception() raise finally: for vbd in vbds: - logging.info("<< Destroy VBD %s", vbd.uuid) + vbd.logger.info("<< Destroy VBD") vbd.destroy() for vdi in vdis: - logging.info("<< Destroy VDI %s", vdi.uuid) + vdi.logger.info("<< Destroy VDI") vdi.destroy() for vm in vms: - logging.info("<< Destroy VM %s", vm.uuid) + vm.logger.info("<< Destroy VM") vm.destroy(verify=True) def _vm_name(request: pytest.FixtureRequest, vm_def: dict[str, Any]) -> str: @@ -765,7 +784,7 @@ def _create_vm( vm_name = _vm_name(request, vm_def) vm_template = vm_def["template"] - logging.info("Installing VM %r from template %r", vm_name, vm_template) + host.logger.info("Installing VM from template", vm_name=vm_name, vm_template=vm_template) vm = host.vm_from_template(vm_name, vm_template) @@ -793,7 +812,7 @@ def _create_vm( if "params" in vm_def: for param_def in vm_def["params"]: - logging.info("Setting param %s", param_def) + vm.logger.info("Setting parameter", param_def=param_def) vm.param_set(**param_def) def _vm_from_cache( @@ -805,7 +824,7 @@ def _vm_from_cache( raise RuntimeError("No cache found") # Clone the VM before running tests, so that the original VM remains untouched - logging.info("Cloning VM from cache") + base_vm.logger.info("Cloning VM from cache") vm = base_vm.clone(name=prefix_object_name(_vm_name(request, vm_def))) # Remove the description, which may contain a cache identifier vm.param_set('name-description', "") @@ -818,15 +837,15 @@ def started_vm(imported_vm: VM) -> VM: # may be already running if we skipped the import to use an existing VM if not vm.is_running(): vm.start() - wait_for(vm.is_running, '> Wait for VM running') - wait_for(vm.try_get_and_store_ip, "> Wait for VM IP", timeout_secs=5 * 60) + wait_for(vm.is_running, 'Wait for VM running', logger=vm.logger) + wait_for(vm.try_get_and_store_ip, "Wait for VM IP", timeout_secs=5 * 60, logger=vm.logger) return vm # no teardown @pytest.fixture(scope="module") def running_vm(started_vm: VM) -> VM: vm = started_vm - wait_for(vm.is_ssh_up, "> Wait for VM SSH up") + wait_for(vm.is_ssh_up, "Wait for VM SSH up", logger=vm.logger) return vm @pytest.fixture(scope='module') @@ -921,10 +940,10 @@ def nfs_iso_sr(host: Host, nfs_iso_device_config: dict[str, Any]) -> Generator[S @pytest.fixture(scope='function') def exit_on_fistpoint(host: Host) -> Generator[None, None, None]: from lib.fistpoint import FistPoint - logging.info(">> Enabling exit on fistpoint") + host.logger.info(">> Enabling exit on fistpoint") FistPoint.enable_exit_on_fistpoint(host) yield - logging.info("<< Disabling exit on fistpoint") + host.logger.info("<< Disabling exit on fistpoint") FistPoint.disable_exit_on_fistpoint(host) @pytest.fixture(scope='module') diff --git a/jobs.py b/jobs.py index dd1d807af..695e97b5c 100755 --- a/jobs.py +++ b/jobs.py @@ -7,6 +7,7 @@ import sys from lib.commands import ssh +from lib.logging import configure_logging from typing import NotRequired, TypedDict, cast @@ -823,6 +824,7 @@ def action_run(args: argparse.Namespace) -> None: def main() -> None: + configure_logging() parser = argparse.ArgumentParser(description="Manage test jobs") subparsers = parser.add_subparsers(dest="action", metavar="action") subparsers.required = True diff --git a/lib/basevm.py b/lib/basevm.py index 9bd91651c..fc0a9c295 100644 --- a/lib/basevm.py +++ b/lib/basevm.py @@ -1,12 +1,12 @@ from __future__ import annotations -import logging - from typing import TYPE_CHECKING, List, Literal, overload if TYPE_CHECKING: from lib.host import Host +import structlog + from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set from lib.sr import SR @@ -15,9 +15,9 @@ class BaseVM: xe_prefix = "vm" uuid: str + logger: structlog.BoundLogger def __init__(self, uuid: str, host: Host): - logging.info("New %s: %s", type(self).__name__, uuid) self.uuid = uuid self.host = host @@ -96,7 +96,7 @@ def get_sr(self) -> SR: return sr def export(self, filepath: str, compress: str = 'none', use_cache: bool = False) -> None: - logging.info("Export VM %s to %s with compress=%s" % (self.uuid, filepath, compress)) + self.logger.info("Export VM", filepath=filepath, compress=compress) params: dict[str, str | bool | dict[str, str]] = { 'uuid': self.uuid, 'compress': compress, diff --git a/lib/commands.py b/lib/commands.py index 967acf83b..31cf6acf8 100644 --- a/lib/commands.py +++ b/lib/commands.py @@ -1,12 +1,13 @@ from __future__ import annotations import base64 -import logging import os import platform import subprocess import tempfile +import structlog + import lib.config as config from lib.netutil import wrap_ip @@ -86,7 +87,12 @@ def _ssh( decode: bool, options: list[str], multiplexing: bool, + logger: structlog.BoundLogger | None = None, ) -> SSHResult[str] | SSHResult[bytes] | SSHCommandFailed | str | bytes | None: + if logger is None: + logger = structlog.get_logger("ssh").bind(host=hostname_or_ip) + assert logger is not None + logger = logger.bind(command=cmd) opts = list(options) opts += ['-o', 'BatchMode yes'] opts += ['-o', 'PubkeyAcceptedKeyTypes +ssh-rsa'] @@ -125,7 +131,7 @@ def _ssh( if background: ssh_cmd = ['ssh', f'root@{hostname_or_ip}'] + opts + [cmd] - logging.debug(f"[{hostname_or_ip}] {cmd}") + logger.debug("Run command in background") subprocess.Popen( ssh_cmd, stdout=subprocess.PIPE, @@ -137,7 +143,7 @@ def _ssh( opts += ['-E', ssh_log_file.name] ssh_cmd = ['ssh', f'root@{hostname_or_ip}'] + opts + [cmd] - logging.debug(f"[{hostname_or_ip}] {cmd}") + logger.debug("Run SSH command", ssh_command=True) process = subprocess.Popen( ssh_cmd, stdout=subprocess.PIPE, @@ -149,14 +155,17 @@ def _ssh( for line in iter(process.stdout.readline, b''): readable_line = line.decode(errors='replace').strip() stdout.append(line) - logging.debug("> %s", readable_line) + logger.debug("New line on stdout", stdout=readable_line, ssh_output=True) _, stderr = process.communicate() res = subprocess.CompletedProcess(ssh_cmd, process.returncode, b''.join(stdout), stderr) ssherr = ssh_log_file.read() - if ssherr: - logging.debug("[%s] ssh stderr: %s", hostname_or_ip, ssherr) + kwargs = {"ssh_error": ssherr} if ssherr else {} + if res.returncode != 0: + logger.debug("SSH command failed", returncode=res.returncode, ssh_result=True, **kwargs) + else: + logger.debug("SSH command succeeded", ssh_result=True, **kwargs) # Get a decoded version of the output in any case, replacing potential errors output_for_errors = res.stdout.decode(errors='replace').strip() @@ -191,45 +200,52 @@ def _ssh( def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True, simple_output: Literal[True] = True, suppress_fingerprint_warnings: bool = True, background: Literal[False] = False, - decode: Literal[True] = True, options: List[str] = [], multiplexing: bool = True) -> str: + decode: Literal[True] = True, options: List[str] = [], multiplexing: bool = True, + logger: structlog.BoundLogger | None = None) -> str: ... @overload def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True, simple_output: Literal[True] = True, suppress_fingerprint_warnings: bool = True, background: Literal[False] = False, - decode: Literal[False], options: List[str] = [], multiplexing: bool = True) -> bytes: + decode: Literal[False], options: List[str] = [], multiplexing: bool = True, + logger: structlog.BoundLogger | None = None) -> bytes: ... @overload def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True, simple_output: Literal[False], suppress_fingerprint_warnings: bool = True, background: Literal[False] = False, - decode: Literal[True] = True, options: List[str] = [], multiplexing: bool = True) -> SSHResult[str]: + decode: Literal[True] = True, options: List[str] = [], multiplexing: bool = True, + logger: structlog.BoundLogger | None = None) -> SSHResult[str]: ... @overload def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True, simple_output: Literal[False], suppress_fingerprint_warnings: bool = True, background: Literal[False] = False, - decode: Literal[False], options: List[str] = [], multiplexing: bool = True) -> SSHResult[bytes]: + decode: Literal[False], options: List[str] = [], multiplexing: bool = True, + logger: structlog.BoundLogger | None = None) -> SSHResult[bytes]: ... @overload def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True, simple_output: Literal[False], suppress_fingerprint_warnings: bool = True, background: Literal[True], - decode: bool = True, options: List[str] = [], multiplexing: bool = True) -> None: + decode: bool = True, options: List[str] = [], multiplexing: bool = True, + logger: structlog.BoundLogger | None = None) -> None: ... @overload def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True, simple_output: bool = True, suppress_fingerprint_warnings: bool = True, background: bool = False, - decode: bool = True, options: List[str] = [], multiplexing: bool = True) \ + decode: bool = True, options: List[str] = [], multiplexing: bool = True, + logger: structlog.BoundLogger | None = None) \ -> str | bytes | SSHResult[str] | SSHResult[bytes] | None: ... def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True, simple_output: bool = True, suppress_fingerprint_warnings: bool = True, - background: bool = False, decode: bool = True, options: List[str] = [], multiplexing: bool = True) \ + background: bool = False, decode: bool = True, options: List[str] = [], multiplexing: bool = True, + logger: structlog.BoundLogger | None = None) \ -> str | bytes | SSHResult[str] | SSHResult[bytes] | None: result_or_exc = _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warnings, - background, decode, options, multiplexing) + background, decode, options, multiplexing, logger) if isinstance(result_or_exc, SSHCommandFailed): raise result_or_exc else: @@ -239,19 +255,22 @@ def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True, simple_out def ssh_with_result(hostname_or_ip: HostAddress, cmd: str, *, decode: Literal[True] = True, suppress_fingerprint_warnings: bool = True, background: bool = False, options: List[str] = [], - multiplexing: bool = True) -> SSHResult[str]: + multiplexing: bool = True, + logger: structlog.BoundLogger | None = None) -> SSHResult[str]: ... @overload def ssh_with_result(hostname_or_ip: HostAddress, cmd: str, *, decode: Literal[False], suppress_fingerprint_warnings: bool = True, background: bool = False, options: List[str] = [], - multiplexing: bool = True) -> SSHResult[bytes]: + multiplexing: bool = True, + logger: structlog.BoundLogger | None = None) -> SSHResult[bytes]: ... def ssh_with_result(hostname_or_ip: HostAddress, cmd: str, *, suppress_fingerprint_warnings: bool = True, background: bool = False, decode: bool = True, options: List[str] = [], - multiplexing: bool = True) -> SSHResult[str] | SSHResult[bytes]: + multiplexing: bool = True, + logger: structlog.BoundLogger | None = None) -> SSHResult[str] | SSHResult[bytes]: result_or_exc = _ssh(hostname_or_ip, cmd, False, False, suppress_fingerprint_warnings, - background, decode, options, multiplexing) + background, decode, options, multiplexing, logger) if isinstance(result_or_exc, SSHCommandFailed): raise result_or_exc elif isinstance(result_or_exc, SSHResult): @@ -261,6 +280,7 @@ def ssh_with_result(hostname_or_ip: HostAddress, cmd: str, *, suppress_fingerpri def scp(hostname_or_ip: HostAddress, src: str, dest: str, check: bool = True, suppress_fingerprint_warnings: bool = True, local_dest: bool = False) -> subprocess.CompletedProcess[bytes]: opts = ['-o', 'BatchMode=yes'] + logger = structlog.get_logger("scp").bind(host=hostname_or_ip, source=src, destination=dest) if suppress_fingerprint_warnings: # Suppress warnings and questions related to host key fingerprints # because on a test network IPs get reused, VMs are reinstalled, etc. @@ -273,6 +293,7 @@ def scp(hostname_or_ip: HostAddress, src: str, dest: str, check: bool = True, else: dest = 'root@{}:{}'.format(ip, dest) + logger.debug("Run SCP command") command = ['scp'] + opts + [src, dest] res = subprocess.run( command, @@ -280,9 +301,7 @@ def scp(hostname_or_ip: HostAddress, src: str, dest: str, check: bool = True, stderr=subprocess.STDOUT, check=False ) - - errorcode_msg = "" if res.returncode == 0 else " - Got error code: %s" % res.returncode - logging.debug(f"[{hostname_or_ip}] scp: {src} => {dest}{errorcode_msg}") + logger.debug("SCP command returned", returncode=res.returncode) if check and res.returncode: raise SSHCommandFailed(res.returncode, res.stdout.decode(), ' '.join(command)) @@ -325,7 +344,9 @@ def local_cmd( cmd: List[str], *, check: bool = True, decode: bool = True ) -> LocalCommandResult[str] | LocalCommandResult[bytes]: """ Run a command locally on tester end. """ - logging.debug("[local] %s", (cmd,)) + command = " ".join(cmd) + logger = structlog.get_logger("cmd").bind(command=command) + logger.debug("Run local command") res = subprocess.run( cmd, stdout=subprocess.PIPE, @@ -336,9 +357,7 @@ def local_cmd( # get a decoded version of the output in any case, replacing potential errors output_for_logs = res.stdout.decode(errors='replace').strip() - errorcode_msg = "" if res.returncode == 0 else " - Got error code: %s" % res.returncode - command = " ".join(cmd) - logging.debug(f"[local] {command}{errorcode_msg}{_ellide_log_lines(output_for_logs)}") + logger.debug("Local command returned", returncode=res.returncode, stdout=output_for_logs) if res.returncode and check: raise LocalCommandFailed(res.returncode, output_for_logs, command) diff --git a/lib/common.py b/lib/common.py index 58cf612c8..75b2488f4 100644 --- a/lib/common.py +++ b/lib/common.py @@ -18,6 +18,7 @@ from uuid import UUID import requests +import structlog from pydantic import TypeAdapter from typing import ( @@ -188,9 +189,9 @@ def callable_marker(value: T | Callable[..., T], request: pytest.FixtureRequest) return value def wait_for(fn: Callable[[], object], msg: str | None = None, timeout_secs: int = 2 * 60, retry_delay_secs: int = 2, - invert: bool = False) -> None: + invert: bool = False, logger: structlog.BoundLogger | None = None) -> None: if msg is not None: - logging.info(msg) + logging.info(msg) if logger is None else logger.info(msg) start_time = time.perf_counter() while True: ret = fn() @@ -206,9 +207,10 @@ def wait_for(fn: Callable[[], object], msg: str | None = None, timeout_secs: int time.sleep(retry_delay_secs) def wait_for_not( - fn: Callable[[], Any], msg: str | None = None, timeout_secs: int = 2 * 60, retry_delay_secs: int = 2 + fn: Callable[[], Any], msg: str | None = None, timeout_secs: int = 2 * 60, retry_delay_secs: int = 2, + logger: structlog.BoundLogger | None = None, ) -> None: - return wait_for(fn, msg, timeout_secs, retry_delay_secs, True) + return wait_for(fn, msg, timeout_secs, retry_delay_secs, True, logger=logger) def is_uuid(maybe_uuid: str) -> bool: try: diff --git a/lib/host.py b/lib/host.py index 5a2d34593..8191a83d1 100644 --- a/lib/host.py +++ b/lib/host.py @@ -1,6 +1,5 @@ from __future__ import annotations -import logging import os import re import shlex @@ -9,6 +8,7 @@ import uuid from dataclasses import dataclass +import structlog from packaging import version import lib.commands as commands @@ -73,6 +73,7 @@ def __init__(self, pool: Pool, hostname_or_ip: str): self.pool = pool self.hostname_or_ip = hostname_or_ip self.xo_srv_id: str | None = None + self.logger = structlog.get_logger("Host").bind(pool=self.pool.master_hostname_or_ip, host=hostname_or_ip) h_data = host_data(self.hostname_or_ip) self.user = h_data['user'] @@ -140,11 +141,12 @@ def ssh(self, cmd: str, *, check: bool = True, simple_output: bool = True, multiplexing: bool = True) -> str | bytes | commands.SSHResult[str] | commands.SSHResult[bytes] | None: return commands.ssh(self.hostname_or_ip, cmd, check=check, simple_output=simple_output, suppress_fingerprint_warnings=suppress_fingerprint_warnings, - background=background, decode=decode, multiplexing=multiplexing) + background=background, decode=decode, multiplexing=multiplexing, + logger=self.logger) def ssh_with_result(self, cmd: str) -> commands.SSHResult[str]: # doesn't raise if the command's return is nonzero, unless there's a SSH error - return commands.ssh_with_result(self.hostname_or_ip, cmd) + return commands.ssh_with_result(self.hostname_or_ip, cmd, logger=self.logger) def scp(self, src: str, dest: str, check: bool = True, suppress_fingerprint_warnings: bool = True, local_dest: bool = False) -> subprocess.CompletedProcess[bytes]: @@ -261,12 +263,12 @@ def execute_script(self, script_contents: str, shebang: str = 'sh', remote_path = self.ssh("mktemp").strip() self.scp(script.name, remote_path) self.ssh(f'chmod 0755 {remote_path}') - except Exception as e: - logging.error("Failed to create temporary file. %s", e) + except Exception: + self.logger.exception("Failed to create temporary file") raise try: - logging.debug(f"[{self}] # Will execute this temporary script:\n{script_contents.strip()}") + self.logger.debug("Will execute this temporary script", script_contents=script_contents) return self.ssh(remote_path, simple_output=simple_output) finally: self.ssh(f'rm -f {remote_path}') @@ -360,13 +362,17 @@ def xo_server_connected(self) -> bool: def xo_server_reconnect(self) -> None: assert self.xo_srv_id is not None - logging.info(f"[{self}] Reconnect XO to host") + self.logger.info("Reconnect XO to host %s" % self) xo_cli('server.disable', {'id': self.xo_srv_id}) xo_cli('server.enable', {'id': self.xo_srv_id}) - wait_for(self.xo_server_connected, timeout_secs=10) + wait_for(self.xo_server_connected, timeout_secs=10, logger=self.logger) # wait for XO to know about the host. Apparently a connected server status # is not enough to guarantee that the host object exists yet. - wait_for(lambda: xo_object_exists(self.uuid), f"[{self}] Wait for XO to know about HOST {self.uuid}") + wait_for( + lambda: xo_object_exists(self.uuid), + "Wait for XO to know about host", + logger=self.logger, + ) @staticmethod def vm_cache_key(uri: str) -> str: @@ -384,9 +390,9 @@ def cached_vm(self, uri: str, sr_uuid: str) -> VM | None: # Assumption: if the first disk is on the SR, the VM is. # If there's no VDI at all, then it is virtually on any SR. if not vm.vdi_uuids() or vm.get_sr().uuid == sr_uuid: - logging.info(f"[{self}] Reusing cached VM {vm.uuid} for {uri}") + self.logger.info("Reusing cached VM", vm_uuid=vm.uuid, uri=uri) return vm - logging.info(f"[{self}] Could not find a VM in cache for {uri!r}") + self.logger.info("Could not find a VM in cache for %r", uri) return None def import_vm(self, uri: str, sr_uuid: str | None = None, use_cache: bool = False) -> VM: @@ -403,7 +409,7 @@ def import_vm(self, uri: str, sr_uuid: str | None = None, use_cache: bool = Fals vm.param_clear('name-description') if uri.startswith("clone+start"): vm.start() - wait_for(vm.is_running, f"[{self}] Wait for VM running ({vm.uuid})") + wait_for(vm.is_running, "Wait for VM running", logger=self.logger) else: vm = self.cached_vm(uri, sr_uuid) if vm: @@ -412,15 +418,14 @@ def import_vm(self, uri: str, sr_uuid: str | None = None, use_cache: bool = Fals assert not ('://' in uri and uri.startswith("clone")), "clone URIs require cache enabled" params: dict[str, str | bool | dict[str, str]] = {} - msg = f"[{self}] Import VM {uri}" + msg = "Import VM" if '://' in uri: params['url'] = uri else: params['filename'] = uri if sr_uuid is not None: - msg += " (SR: %s)" % sr_uuid params['sr-uuid'] = sr_uuid - logging.info(msg) + self.logger.info(msg, uri=uri, sr_uuid=sr_uuid) vm_uuid = self.xe('vm-import', params) vm_name = prefix_object_name(self.xe('vm-param-get', {'uuid': vm_uuid, 'param-name': 'name-label'})) vm = VM(vm_uuid, self) @@ -430,7 +435,7 @@ def import_vm(self, uri: str, sr_uuid: str | None = None, use_cache: bool = Fals vif.move(self.management_network()) if use_cache: cache_key = self.vm_cache_key(uri) - logging.info(f"[{self}] Marking VM {vm.uuid} as cached") + self.logger.info("Marking VM as cached", vm_uuid=vm_uuid) vm.param_set('name-description', cache_key) return vm @@ -450,13 +455,13 @@ def import_iso(self, uri: str, sr: SR) -> VDI: try: params: dict[str, str | bool | dict[str, str]] = {'uuid': vdi_uuid} if '://' in uri: - logging.info(f"[{self}] Download ISO {uri}") + self.logger.info("Download ISO", uri=uri) download_path = f'/tmp/{vdi_uuid}' self.ssh(f"curl -o '{download_path}' '{uri}'") params['filename'] = download_path else: params['filename'] = uri - logging.info(f"[{self}] Import ISO {uri}: name {random_name}, uuid {vdi_uuid}") + self.logger.info("Import ISO", uri=uri, name=random_name, vdi_uuid=vdi_uuid) self.xe('vdi-import', params) finally: @@ -508,7 +513,7 @@ def yum_clean_metadata(self) -> str: yum clean metadata -q """ - logging.info(f"[{self}] Removing cache metadata...") + self.logger.info("Removing cache metadata...") return self.ssh("yum clean metadata -q") def yum_update(self, enablerepos: list[str] = []) -> str: @@ -524,7 +529,7 @@ def yum_update(self, enablerepos: list[str] = []) -> str: """ base_command = "yum update -y" - logging.info(f"[{self}] Updating packages...") + self.logger.info("Updating packages...") if enablerepos: extra = " ".join(f"--enablerepo={r}" for r in enablerepos) base_command = f"{base_command} {extra}" @@ -541,7 +546,7 @@ def update(self, enablerepos: list[str] = [], reboot: bool = True) -> None: :param bool reboot: Choose to reboot or not after update (default: True). """ - logging.info(f"[{self}] Updating...") + self.logger.info(f"[{self}] Updating...") self.yum_clean_metadata() self.yum_update(enablerepos=enablerepos) @@ -549,10 +554,10 @@ def update(self, enablerepos: list[str] = [], reboot: bool = True) -> None: # Everything's ok, just reboot self.reboot(verify=True) - logging.info(f"[{self}] Updated successfully!") + self.logger.info("Updated successfully!") def restart_toolstack(self, verify: bool = False) -> None: - logging.info(f"[{self}] Restart toolstack on host") + self.logger.info("Restart toolstack") self.ssh('xe-toolstack-restart') if verify: self.wait_for_xapi_enabled() @@ -571,6 +576,7 @@ def wait_for_host_up(self, timeout_secs: int = 10 * 60) -> None: f"[{self}] Wait for host up", timeout_secs=timeout_secs, retry_delay_secs=10, + logger=self.logger, ) def wait_for_ssh_reachable(self, timeout_secs: int = 10 * 60) -> None: @@ -578,11 +584,12 @@ def wait_for_ssh_reachable(self, timeout_secs: int = 10 * 60) -> None: lambda: commands.local_cmd(["nc", "-zw5", self.hostname_or_ip, "22"], check=False).returncode == 0, f"[{self}] Wait for ssh up on host", timeout_secs=timeout_secs, - retry_delay_secs=5 + retry_delay_secs=5, + logger=self.logger, ) def wait_for_xapi_enabled(self, timeout_secs: int = 30 * 60) -> None: - logging.info(f"[{self}] Wait for XAPI to complete initialization") + self.logger.info("Wait for XAPI to complete initialization") self.ssh(f"xapi-wait-init-complete {timeout_secs}") assert self.is_enabled() @@ -625,7 +632,7 @@ def get_last_yum_history_tid(self) -> int: # yum history list fails if the list is empty, and it's also not possible to rollback # to before the first transaction, so "0" would not be appropriate as last transaction. # To workaround this, create transactions: install and remove a small package. - logging.info(f"[{self}] Install and remove a small package to workaround empty yum history.") + self.logger.info('Install and remove a small package to workaround empty yum history.') self.yum_install(['gpm-libs']) self.yum_remove(['gpm-libs']) history_str = self.ssh('yum history list --noplugins') @@ -646,14 +653,14 @@ def get_last_yum_history_tid(self) -> int: raise Exception('Unable to parse correctly last yum history tid. Output:\n' + history_str) def yum_install(self, packages: list[str], enablerepo: str | None = None) -> str: - logging.info(f"[{self}] Install packages: {' '.join(packages)} on host") + self.logger.info('Install packages', packages=packages) cmd = 'yum install --setopt=skip_missing_names_on_install=False -y' if enablerepo is not None: cmd = f'{cmd} --enablerepo={enablerepo}' return self.ssh(f'{cmd} {" ".join(packages)}') def yum_remove(self, packages: list[str]) -> str: - logging.info(f"[{self}] Remove packages: {' '.join(packages)} from host") + self.logger.info('Remove packages', packages) return self.ssh(f'yum remove -y {" ".join(packages)}') def packages(self) -> list[str]: @@ -671,14 +678,14 @@ def is_package_installed(self, package: str) -> bool: return self.ssh_with_result(f'rpm -q {package}').returncode == 0 def yum_save_state(self) -> None: - logging.info(f"[{self}] Save yum state for host") + self.logger.info("Save yum state") # For now, that saved state feature does not support several saved states assert self.saved_packages_list is None, "There is already a saved package list set" self.saved_packages_list = self.packages() self.saved_rollback_id = self.get_last_yum_history_tid() def yum_restore_saved_state(self) -> None: - logging.info(f"[{self}] Restore yum state for host") + self.logger.info("Restore yum state") """ Restore yum state to saved state. """ assert self.saved_packages_list is not None, \ "Can't restore previous state without a package list: no saved packages list" @@ -703,7 +710,7 @@ def yum_restore_saved_state(self) -> None: self.saved_rollback_id = None def reboot(self, verify: bool = False) -> None: - logging.info(f"[{self}] Reboot host") + self.logger.info("Reboot host") # Running `reboot` directly immediately disconnects the ssh session and makes the ssh client return with an # error code. Instead, we schedule the reboot a few seconds later to let the ssh command return properly. self.ssh('systemd-run --on-active=2s reboot') @@ -860,7 +867,7 @@ def _all_available(kname: str) -> bool: )) self.block_devices_info = sorted(devices, key=lambda d: d.size, reverse=True) - logging.debug(f"[{self}] blockdevs found: {[d.name for d in self.block_devices_info]}") + self.logger.debug("Block devices collected", block_devices=[d.name for d in self.block_devices_info]) def disks(self) -> list[Host.BlockDeviceInfo]: """ List of all block devices (local disks, mdadm arrays, multipath devices). """ @@ -888,13 +895,16 @@ def sr_create(self, sr_type: str, label: str, device_config: dict[str, str], sha for key, value in device_config.items(): params['device-config:{}'.format(key)] = value - logging.info( - f"[{self}] Create {sr_type} SR on host with label '{label}' and device-config: {str(device_config)}" + self.logger.info( + "Create SR", + sr_type=sr_type, + device_config=device_config, + label=label ) sr_uuid = self.xe('sr-create', params) sr = SR(sr_uuid, self.pool) if verify: - wait_for(sr.exists, f"[{self}] Wait for SR {sr_uuid} to exist") + wait_for(sr.exists, "Wait for SR to exist", logger=self.logger) return sr def is_master(self) -> bool: @@ -964,13 +974,15 @@ def join_pool(self, pool: Pool) -> None: }) wait_for( lambda: self.uuid in pool.hosts_uuids(), - f"Wait for joining host {self} to appear in joined pool {master}." + f"Wait for joining host {self} to appear in joined pool {master}.", + logger=self.logger, ) pool.hosts.append(Host(pool, pool.host_ip(self.uuid))) # Do not use `self.is_enabled` since it'd ask the XAPI of hostB1 before the join... wait_for( lambda: strtobool(master.xe('host-param-get', {'uuid': self.uuid, 'param-name': 'enabled'})), - f"Wait for pool {master} to see joined host {self} as enabled." + f"Wait for pool {master} to see joined host {self} as enabled.", + logger=self.logger, ) self.pool = pool @@ -1047,7 +1059,7 @@ def create_bond(self, network: Network, pifs: list[PIF], mode: str | None = None args['mode'] = mode uuid = self.xe("bond-create", args, minimal=True) - logging.info(f"[{self}] New Bond: {uuid}") + self.logger.info("New Bond", bond_uuid=uuid) return Bond(self, uuid) @@ -1059,8 +1071,8 @@ def create_network(self, label: str, description: str | None = None) -> Network: if description is not None: args['name-description'] = description - logging.info(f"[{self}] Creating network '{label}'") + self.logger.info("Creating network", network_label=label) uuid = self.xe("network-create", args, minimal=True) - logging.info(f"[{self}] New Network: {uuid}") + self.logger.info("New Network", network_label=label, network_uuid=uuid) return Network(self, uuid) diff --git a/lib/logging.py b/lib/logging.py new file mode 100644 index 000000000..3fc94629b --- /dev/null +++ b/lib/logging.py @@ -0,0 +1,262 @@ +from datetime import datetime + +import structlog + +def drop_column(key: str, value: object) -> str: + return "" + + +def millisecond_timestamper( + logger: structlog.typing.WrappedLogger, + method_name: str, + event_dict: structlog.typing.EventDict, +) -> structlog.typing.EventDict: + now = datetime.now() + base_time = now.strftime("%b %d %H:%M:%S") + milliseconds = f"{now.microsecond // 1000:03d}" + event_dict["timestamp"] = f"{base_time}.{milliseconds}" + return event_dict + + +def make_kv_identifier_column(key: str, styles: structlog.dev.ColumnStyles) -> structlog.dev.Column: + width = 16 - len(key) + value_format = f"{{:>{width}}}" + col = structlog.dev.Column( + key, + structlog.dev.KeyValueColumnFormatter( + key_style=styles.kv_key, + value_style=styles.kv_value, + reset_style=styles.reset, + value_repr=value_format.format, + prefix="[", + postfix="]", + width=16 - len(key), + ), + ) + return col + + +def make_ssh_output_renderer( + default_renderer: structlog.dev.ConsoleRenderer, +) -> structlog.dev.ConsoleRenderer: + styles = default_renderer.get_default_column_styles(colors=True) + col_timestamp = default_renderer.columns[1] + col_level = default_renderer.columns[2] + return structlog.dev.ConsoleRenderer( + sort_keys=False, + columns=[ + structlog.dev.Column("", drop_column), + col_timestamp, + col_level, + structlog.dev.Column( + "ssh_prefix", + structlog.dev.KeyValueColumnFormatter( + key_style=None, + value_style=styles.bright, + reset_style=styles.reset, + value_repr=str, + ), + ), + structlog.dev.Column( + "stdout", + structlog.dev.KeyValueColumnFormatter( + key_style=None, + value_style=styles.timestamp, + reset_style=styles.reset, + value_repr=str, + ), + ), + ], + ) + +def make_ssh_result_renderer( + default_renderer: structlog.dev.ConsoleRenderer, +) -> structlog.dev.ConsoleRenderer: + styles = default_renderer.get_default_column_styles(colors=True) + col_timestamp = default_renderer.columns[1] + col_level = default_renderer.columns[2] + return structlog.dev.ConsoleRenderer( + sort_keys=False, + columns=[ + structlog.dev.Column("", drop_column), + col_timestamp, + col_level, + structlog.dev.Column( + "ssh_prefix", + structlog.dev.KeyValueColumnFormatter( + key_style=None, + value_style=styles.bright, + reset_style=styles.reset, + value_repr=str, + ), + ), + structlog.dev.Column( + "returncode", + structlog.dev.KeyValueColumnFormatter( + key_style=None, + value_style=styles.level_warn, + reset_style=styles.reset, + value_repr=str, + ), + ), + structlog.dev.Column( + "ssh_error", + structlog.dev.KeyValueColumnFormatter( + key_style=styles.kv_key, + value_style=styles.kv_value, + reset_style=styles.reset, + value_repr=str, + ), + ), + ], + ) + + +def make_pretty_renderer_processor(left_identifier: bool = True) -> structlog.typing.Processor: + pad_event_to = 0 if left_identifier else 40 + + # Default renderer + default_renderer = structlog.dev.ConsoleRenderer(colors=True, pad_event_to=pad_event_to, sort_keys=False) + col_extras = default_renderer.columns[0] + col_timestamp = default_renderer.columns[1] + col_level = default_renderer.columns[2] + col_event = default_renderer.columns[3] + col_logger = default_renderer.columns[4] + col_level.formatter.width = 5 # type: ignore + col_logger.formatter.width = 17 # type: ignore + + # Make identified renderer + def make_identified_renderer(identifier_column: structlog.dev.Column) -> structlog.dev.ConsoleRenderer: + columns = ( + [col_timestamp, col_level, identifier_column, col_event, col_extras] + if left_identifier + else [col_timestamp, col_level, col_event, identifier_column, col_extras] + ) + return structlog.dev.ConsoleRenderer(sort_keys=False, columns=columns) + + styles = default_renderer.get_default_column_styles(colors=True) + identified_renderers = { + "Pool": make_identified_renderer(make_kv_identifier_column("pool", styles)), + "Host": make_identified_renderer(make_kv_identifier_column("host", styles)), + "VM": make_identified_renderer(make_kv_identifier_column("vm", styles)), + "Snapshot": make_identified_renderer(make_kv_identifier_column("snapshot", styles)), + "VDI": make_identified_renderer(make_kv_identifier_column("vdi", styles)), + } + + # Make ssh command renderer + def make_identified_ssh_command_renderer(identifier_column: structlog.dev.Column) -> structlog.dev.ConsoleRenderer: + catchall = structlog.dev.Column("", drop_column) + command_column = structlog.dev.Column( + "command", + structlog.dev.KeyValueColumnFormatter( + key_style=None, + value_style=styles.level_warn, + reset_style=styles.reset, + value_repr=str, + ), + ) + return structlog.dev.ConsoleRenderer( + sort_keys=False, + columns=[catchall, col_timestamp, col_level, identifier_column, command_column], + ) + + ssh_command_identified_renderers = { + "Pool": make_identified_ssh_command_renderer(make_kv_identifier_column("pool", styles)), + "Host": make_identified_ssh_command_renderer(make_kv_identifier_column("host", styles)), + "VM": make_identified_ssh_command_renderer(make_kv_identifier_column("vm", styles)), + } + default_ssh_command_identified_renderer = make_identified_ssh_command_renderer( + make_kv_identifier_column("host", styles) + ) + + ssh_output_renderer = make_ssh_output_renderer(default_renderer) + ssh_result_renderer = make_ssh_result_renderer(default_renderer) + + noise_keys = { + "Pool": [], + "Host": ["pool"], + "VM": ["pool", "ip"], + "Snapshot": ["pool"], + "VDI": [], + } + info_noise_keys = { + "Pool": [], + "Host": [], + "VM": ["host", "vm_uuid", "ip"], + "Snapshot": ["host", "vm_uuid", "snapshot_uuid"], + "VDI": ["vdi_uuid", "sr_uuid"], + } + + def processor( + logger: structlog.typing.WrappedLogger, + method_name: str, + event_dict: structlog.typing.EventDict, + ) -> str: + logger_name = event_dict.get("logger", None) + + # Prepare identifier + if logger_name == "VM": + ip = event_dict.get("ip") + vm_uuid = event_dict.get("vm_uuid", "") + event_dict["vm"] = ip if ip else vm_uuid[:8] + elif logger_name == "Snapshot": + snapshot_uuid = event_dict.get("snapshot_uuid", "") + event_dict["snapshot"] = snapshot_uuid[:8] + elif logger_name == "VDI": + vdi_uuid = event_dict.get("vdi_uuid", "") + event_dict["vdi"] = vdi_uuid[:8] + + # SSH command renderer + if event_dict.pop("ssh_command", False): + renderer = ssh_command_identified_renderers.get( + logger_name, + default_ssh_command_identified_renderer, + ) + return renderer(logger, method_name, event_dict) + + # SSH output renderer + if event_dict.pop("ssh_output", False): + event_dict["ssh_prefix"] = ">" + return ssh_output_renderer(logger, method_name, event_dict) + + # SSH output renderer + if event_dict.pop("ssh_result", False): + returncode = event_dict.get("returncode") + if returncode is None: + raise structlog.DropEvent + event_dict["ssh_prefix"] = "$?" + return ssh_result_renderer(logger, method_name, event_dict) + + # Default renderer + logger_name = event_dict.get("logger", None) + if logger_name not in identified_renderers: + return default_renderer(logger, method_name, event_dict) + + # Remove noisy key-value pairs + event_dict.pop("logger") + for key in noise_keys[logger_name]: + event_dict.pop(key, None) + if event_dict.get("level") == "info": + for key in info_noise_keys[logger_name]: + event_dict.pop(key, None) + + # Pool / host / VM / snapshot renderer + renderer = identified_renderers[logger_name] + return renderer(logger, method_name, event_dict) + + return processor + + +def configure_logging(): + structlog.configure( + processors=[ + structlog.stdlib.add_log_level, + structlog.stdlib.add_logger_name, + millisecond_timestamper, + structlog.stdlib.PositionalArgumentsFormatter(), + structlog.processors.StackInfoRenderer(), + make_pretty_renderer_processor(), + ], + logger_factory=structlog.stdlib.LoggerFactory(), + cache_logger_on_first_use=True, + ) diff --git a/lib/pool.py b/lib/pool.py index 5c1edeaaa..7f94417f9 100644 --- a/lib/pool.py +++ b/lib/pool.py @@ -1,9 +1,8 @@ from __future__ import annotations -import logging import os -import traceback +import structlog from packaging import version import lib.commands as commands @@ -22,6 +21,8 @@ class Pool: xe_prefix = "pool" def __init__(self, master_hostname_or_ip: HostAddress) -> None: + self.master_hostname_or_ip = master_hostname_or_ip + self.logger = structlog.get_logger("Pool").bind(pool=master_hostname_or_ip) master = Host(self, master_hostname_or_ip) if not master.is_master(): raise NotAMasterHostError(f"Host {master_hostname_or_ip} is not a master host. Pool not created.") @@ -32,7 +33,7 @@ def __init__(self, master_hostname_or_ip: HostAddress) -> None: # refused (calling connect )" when calling self.hosts_uuids() self.master.wait_for_xapi_enabled() - logging.info("Getting Pool info for %r", master_hostname_or_ip) + self.logger.info("Getting Pool info") for host_uuid in self.hosts_uuids(): if host_uuid != self.hosts[0].uuid: host = Host(self, self.host_ip(host_uuid)) @@ -41,6 +42,9 @@ def __init__(self, master_hostname_or_ip: HostAddress) -> None: self.saved_uefi_certs: dict[str, str] | None = None self.pre_existing_sr_uuids = safe_split(self.master.xe('sr-list', {'minimal': 'true'}), ',') + def __repr__(self) -> str: + return f"Pool({self.master_hostname_or_ip!r})" + def param_get(self, param_name: str, key: str | None = None, accept_unknown_key: bool = False) -> str | None: return _param_get(self.master, Pool.xe_prefix, self.uuid, param_name, key, accept_unknown_key) @@ -65,14 +69,16 @@ def exec_on_hosts_on_error_rollback(self, func: Callable[[Host], Any], hosts_done.append(h) except Exception as e: if rollback_func: - logging.warning( - f"An error occurred in `exec_on_hosts_on_error_rollback` for host {h}\n" - f"Backtrace:\n{traceback.format_exc()}" + self.logger.exception( + "An error occurred in `exec_on_hosts_on_error_rollback`", + host=h, ) rollback_hosts = hosts_done + [h] - logging.info("Attempting to run the rollback function on host(s) " - f"{', '.join([str(h) for h in rollback_hosts])}...") + self.logger.info( + "Attempting to run the rollback function on host(s)", + rollback_hosts=rollback_hosts, + ) try: self.exec_on_hosts_on_error_continue(rollback_func, rollback_hosts) except Exception: @@ -92,11 +98,13 @@ def exec_on_hosts_on_error_continue(self, func: Callable[[Host], Any], host_list try: func(h) except Exception as e: - logging.warning( - f"An error occurred in `exec_on_hosts_on_error_continue` for host {h}\n" - f"Backtrace:\n{traceback.format_exc()}" + self.logger.exception( + "An error occurred in `exec_on_hosts_on_error_continue`", + host=h, + ) + self.logger.info( + "Attempting to run the function on the next hosts of the pool if there are any left..." ) - logging.info("Attempting to run the function on the next hosts of the pool if there are any left...") errors[h.hostname_or_ip] = e if errors: raise Exception(f"One or more exceptions were raised in `exec_on_hosts_on_error_continue`: {errors}") @@ -144,7 +152,7 @@ def push_iso(self, local_file: str, remote_filename: str | None = None) -> str: remote_filename = self.master.ssh(f'mktemp --suffix=.iso -p {mountpoint}') self.master.ssh(f'chmod 644 {remote_filename}') - logging.info("Uploading to ISO-SR %s as %s", local_file, remote_filename) + self.logger.info("Uploading local file to ISO-SR", local_file=local_file, remote_filename=remote_filename) self.master.scp(local_file, remote_filename) iso_sr.scan() return os.path.basename(remote_filename) @@ -152,7 +160,7 @@ def push_iso(self, local_file: str, remote_filename: str | None = None) -> str: def remove_iso(self, remote_filename: str) -> None: iso_sr = self.get_iso_sr() fullpath = f"/run/sr-mount/{iso_sr.uuid}/{remote_filename}" - logging.info("Removing %s from ISO-SR server", remote_filename) + self.logger.info("Removing ISO from ISO-SR server", remote_filename=remote_filename) self.master.ssh(f'rm {fullpath}') def save_uefi_certs(self) -> None: @@ -174,7 +182,7 @@ def save_uefi_certs(self) -> None: This can be revised later if a need for saving custom certificates in 8.3+ arises. """ assert self.master.xcp_version < version.parse("8.3"), "this function should only be needed on XCP-ng 8.2" - logging.info('Saving pool UEFI certificates') + self.logger.info('Saving pool UEFI certificates') if int(self.master.ssh("secureboot-certs --version").split(".")[0]) < 1: raise RuntimeError("The host must have secureboot-certs version >= 1.0.0") @@ -199,8 +207,7 @@ def save_uefi_certs(self) -> None: # else we won't be able to restore the exact same state if len(saved_certs) == 0 or ('PK' in saved_certs and 'KEK' in saved_certs and 'db' in saved_certs): self.saved_uefi_certs = saved_certs - logging.info('Pool UEFI certificates state saved: %s' - % (' '.join(saved_certs.keys()) if saved_certs else 'no certs')) + self.logger.info('Pool UEFI certificates state saved', saved_certs=saved_certs) else: for tmp_file in saved_certs.values(): self.master.ssh(f'rm -f {tmp_file}') @@ -217,11 +224,11 @@ def restore_uefi_certs(self) -> None: assert self.master.xcp_version < version.parse("8.3"), "this function should only be needed on XCP-ng 8.2" assert self.saved_uefi_certs is not None if len(self.saved_uefi_certs) == 0: - logging.info('We need to clear pool UEFI certificates to restore initial state') + self.logger.info('We need to clear pool UEFI certificates to restore initial state') self.clear_uefi_certs() else: assert 'PK' in self.saved_uefi_certs and 'KEK' in self.saved_uefi_certs and 'db' in self.saved_uefi_certs - logging.info('Restoring pool UEFI certificates: ' + ' '.join(self.saved_uefi_certs.keys())) + self.logger.info('Restoring pool UEFI certificates', saved_uefi_certs=self.saved_uefi_certs) # restore certs params = [self.saved_uefi_certs['PK'], self.saved_uefi_certs['KEK'], self.saved_uefi_certs['db']] if 'dbx' in self.saved_uefi_certs: @@ -246,7 +253,7 @@ def clear_uefi_certs(self) -> None: For XCP-ng 8.3+, see clear_custom_uefi_certificates() """ assert self.master.xcp_version < version.parse("8.3"), "function only relevant on XCP-ng 8.2" - logging.info('Clearing pool UEFI certificates in XAPI and on hosts disks') + self.logger.info('Clearing pool UEFI certificates in XAPI and on hosts disks') self.master.ssh('secureboot-certs clear') # remove files on each host for host in self.hosts: @@ -255,7 +262,7 @@ def clear_uefi_certs(self) -> None: def clear_custom_uefi_certs(self) -> None: """ Clear Custom UEFI certificates on XCP-ng 8.3+. """ assert self.master.xcp_version >= version.parse("8.3"), "function only relevant on XCP-ng 8.3+" - logging.info('Clearing custom pool UEFI certificates') + self.logger.info('Clearing custom pool UEFI certificates') self.master.ssh('secureboot-certs clear') def install_custom_uefi_certs(self, auths: Iterable[EFIAuth]) -> None: @@ -272,10 +279,10 @@ def install_custom_uefi_certs(self, auths: Iterable[EFIAuth]) -> None: assert 'KEK' in auths_dict assert 'db' in auths_dict - logging.info('Installing auths to pool: %s' % list(auths_dict.keys())) + self.logger.info('Installing auths to pool', auths=list(auths_dict.keys())) for key in auths_dict: value = host.ssh(f'md5sum {auths_dict[key]} | cut -d " " -f 1') - logging.debug('Key: %s, value: %s' % (key, value)) + self.logger.debug('Report auth values', auth_key=key, auth_value=value) params = [auths_dict['PK'], auths_dict['KEK'], auths_dict['db']] if 'dbx' in auths_dict: params.append(auths_dict['dbx']) diff --git a/lib/snapshot.py b/lib/snapshot.py index 4c88104ed..7c5841ff6 100644 --- a/lib/snapshot.py +++ b/lib/snapshot.py @@ -1,6 +1,6 @@ from __future__ import annotations -import logging +import structlog from lib.basevm import BaseVM @@ -14,25 +14,32 @@ class Snapshot(BaseVM): basevm: VM def __init__(self, uuid: str, host: Host, vm: VM): + self.logger = structlog.get_logger("Snapshot").bind( + pool=host.pool.master_hostname_or_ip, + host=host.hostname_or_ip, + vm_uuid=vm.uuid, + snapshot_uuid=uuid + ) self.basevm = vm super(Snapshot, self).__init__(uuid, host) + self.logger.info("New snapshot") def _disk_list(self) -> str: return self.host.xe('snapshot-disk-list', {'uuid': self.uuid, 'vbd-params': ''}, minimal=True) def destroy(self, verify: bool = False) -> None: - logging.info("Delete snapshot " + self.uuid) + self.logger.info("Delete snapshot") # that uninstall command apparently works better for snapshots than for VMs self.host.xe('snapshot-uninstall', {'uuid': self.uuid, 'force': True}) if verify: - logging.info("Check snapshot doesn't exist anymore") + self.logger.info("Check snapshot doesn't exist anymore") assert not self.exists() def exists(self) -> bool: return self.host.pool_has_vm(self.uuid, vm_type='snapshot') def revert(self) -> None: - logging.info("Revert to snapshot %s", self.uuid) + self.logger.info("Revert to snapshot") self.host.xe('snapshot-revert', {'uuid': self.uuid}) self.basevm.create_vdis_list() # We reset the base VM object VDIs list because it changed following the revert diff --git a/lib/sr.py b/lib/sr.py index 20e9d2d86..bbf19b036 100644 --- a/lib/sr.py +++ b/lib/sr.py @@ -1,8 +1,9 @@ from __future__ import annotations -import logging import time +import structlog + import lib.commands as commands from lib.common import ( GiB, @@ -35,6 +36,7 @@ def __init__(self, uuid: str, pool: Pool): self._is_shared: bool | None = None # cached value for is_shared() self._main_host: Host | None = None # cached value for main_host() self._type: str | None = None # cache value for get_type() + self.logger = structlog.get_logger("SR").bind(sr_uuid=uuid, pool_uuid=pool.uuid) def pbd_uuids(self) -> list[str]: return safe_split(self.pool.master.xe('pbd-list', {'sr-uuid': self.uuid}, minimal=True)) @@ -54,10 +56,10 @@ def unplug_pbd(self, pbd_uuid: str, force: bool = False) -> None: # if force is set. if not force: raise - logging.warning('Ignore exception during PBD unplug: {}'.format(e)) + self.logger.warning('Ignore exception during PBD unplug', exception=e) def unplug_pbds(self, force: bool = False) -> None: - logging.info(f"Unplug PBDs for SR {self.uuid}") + self.logger.info("Unplug PBDs") for pbd_uuid in self.pbd_uuids(): self.unplug_pbd(pbd_uuid, force=force) @@ -74,7 +76,7 @@ def plug_pbd(self, pbd_uuid: str) -> None: self.pool.master.xe('pbd-plug', {'uuid': pbd_uuid}) def plug_pbds(self, verify: bool = True) -> None: - logging.info("Attach PBDs") + self.logger.info("Attach PBDs") for pbd_uuid in self.pbd_uuids(): self.plug_pbd(pbd_uuid) if verify: @@ -90,14 +92,14 @@ def vdi_uuids(self, managed: bool = False, name_label: str | None = None) -> lis return safe_split(self.pool.master.xe('vdi-list', args, minimal=True)) def destroy(self, verify: bool = False, force: bool = False) -> None: - logging.info(f"Will attempt SR destroy on {self.uuid}...") + self.logger.info(f"Will attempt SR destroy on {self.uuid}...") # Rescan SR to improve the chances of the forced GC run triggered by sr-destroy # remove all VDIs in one pass and such have sr-destroy working on first try. self.scan() max_tries = 5 for i in range(1, max_tries + 1): # [1, 2, ..., max_tries] self.unplug_pbds(force) - logging.info(f"Destroy SR {self.uuid} (attempt {i})") + self.logger.info("Destroy SR", attempt=i) try: # Note: sr-destroy triggers ONE forced GC run # This may not be enough in some cases @@ -107,7 +109,7 @@ def destroy(self, verify: bool = False, force: bool = False) -> None: if "the SR is not empty" not in e.stdout: raise else: - logging.info(f"SR destroy failed with message: {e.stdout}") + self.logger.info("SR destroy failed", stdout=e.stdout) try: self.plug_pbds() # rescan for an up to date list of VDIs @@ -119,7 +121,8 @@ def destroy(self, verify: bool = False, force: bool = False) -> None: raise Exception("SR destroy failed due to SR not empty, " "and there are indeed managed VDIs left on the SR.") else: - logging.info("SR destroy failed due to SR not empty but there aren't any managed VDIs left.") + self.logger.info( + "SR destroy failed due to SR not empty but there aren't any managed VDIs left.") if i < max_tries: if i == max_tries - 1: # We tried already 4 times to destroy the SR, and there still are hidden VDIs that @@ -128,10 +131,10 @@ def destroy(self, verify: bool = False, force: bool = False) -> None: # The GC should kick approximately 5 minutes after the last operation we did, so let's # give it these 5 minutes plus extra time to complete. gc_delay = 600 - logging.warning(f"SR destroy failed {i} times in a row. " - f"Wait for {gc_delay}s, hoping GC fully runs before next try") + self.logger.warning(f"SR destroy failed {i} times in a row. " + f"Wait for {gc_delay}s, hoping GC fully runs before next try") time.sleep(gc_delay) - logging.info("Retrying sr-destroy in case it previously failed due to incomplete GC.") + self.logger.info("Retrying sr-destroy in case it previously failed due to incomplete GC.") continue else: raise Exception(f"Could not destroy the SR even after {i} attempts.") @@ -142,14 +145,14 @@ def destroy(self, verify: bool = False, force: bool = False) -> None: def forget(self, force: bool = False) -> None: self.unplug_pbds(force) - logging.info("Forget SR " + self.uuid) + self.logger.info("Forget SR") self.pool.master.xe('sr-forget', {'uuid': self.uuid}) def exists(self) -> bool: return self.pool.master.xe('sr-list', {'uuid': self.uuid}, minimal=True) == self.uuid def scan(self) -> None: - logging.info("Scan SR " + self.uuid) + self.logger.info("Scan SR") self.pool.master.xe('sr-scan', {'uuid': self.uuid}) def hosts_uuids(self) -> list[str]: @@ -203,6 +206,7 @@ def is_shared(self) -> bool: def get_type(self) -> str: if self._type is None: self._type = self.param_get('type') + self.logger = self.logger.bind(sr_type=type) return self._type def get_name_label(self) -> str: @@ -212,7 +216,7 @@ def create_vdi( self, name_label: str | None = None, virtual_size: int = 1 * GiB, image_format: ImageFormat | None = None ) -> VDI: name_label = name_label or f'test-vdi-{randid()}' - logging.info("Create VDI %r on SR %s", name_label, self.uuid) + self.logger.info("Create VDI", name_label=name_label) args: dict[str, str | bool | dict[str, str]] = { 'name-label': prefix_object_name(name_label), 'virtual-size': str(virtual_size), @@ -224,14 +228,14 @@ def create_vdi( return VDI(vdi_uuid, sr=self) def run_quicktest(self) -> None: - logging.info(f"Run quicktest on SR {self.uuid}") + self.logger.info(f"Run quicktest on SR {self.uuid}") # Always display the output of quicktest, failed or not. # This will duplicate the output in some cases, but it ensures we always have it for failure analysis, # even when quicktest leaves SRs in a state which makes teardown fail (in this case, pytest often doesn't # manage to display the details of the failed command, for a reason unknown - no usable reproducer found) try: output = self.pool.master.ssh(f'/opt/xensource/debug/quicktest -sr {self.uuid}') - logging.info(f"Quicktest output: {output}") + self.logger.info("Quicktest finished", output=output) except commands.SSHCommandFailed as e: - logging.error(f"Quicktest output: {e.stdout}") + self.logger.error("Quicktest failed", output=e.stdout) raise diff --git a/lib/vbd.py b/lib/vbd.py index b194433e0..95e8e3dfc 100644 --- a/lib/vbd.py +++ b/lib/vbd.py @@ -1,6 +1,6 @@ from __future__ import annotations -import logging +import structlog from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set @@ -19,6 +19,7 @@ def __init__(self, uuid: str, vm: VM, device: str): self.uuid = uuid self.vm = vm self.device = device + self.logger = structlog.get_logger("VBD").bind(vbd_uuid=uuid, vm_uuid=vm.uuid, device=device) def plug(self) -> None: self.vm.host.xe("vbd-plug", {'uuid': self.uuid}) @@ -47,7 +48,7 @@ def param_clear(self, param_name: str) -> None: param_name) def destroy(self) -> None: - logging.info("Destroy %s", self) + self.logger.info("Destroy VBD") self.vm.host.pool.master.xe('vbd-destroy', {'uuid': self.uuid}) def __str__(self) -> str: diff --git a/lib/vdi.py b/lib/vdi.py index 0ad83a580..fe4739854 100644 --- a/lib/vdi.py +++ b/lib/vdi.py @@ -1,6 +1,6 @@ from __future__ import annotations -import logging +import structlog from lib.common import ( _param_add, @@ -45,12 +45,13 @@ def __init__(self, uuid: str, *, host: Host | None = None, sr: SR | None = None) self.sr = sr else: self.sr = sr + self.logger = structlog.get_logger("VDI").bind(vdi_uuid=uuid, sr_uuid=self.sr.uuid) def name(self) -> str: return self.param_get('name-label') def destroy(self) -> None: - logging.info("Destroy %s", self) + self.logger.info("Destroy VDI") self.sr.pool.master.xe('vdi-destroy', {'uuid': self.uuid}) def clone(self) -> VDI: @@ -68,7 +69,7 @@ def get_virtual_size(self) -> int: return int(self.param_get("virtual-size")) def resize(self, new_size: int) -> None: - logging.info(f"Resizing VDI {self.uuid} to {new_size}") + self.logger.info("Resizing VDI", new_size=new_size) self.sr.pool.master.xe("vdi-resize", {"uuid": self.uuid, "disk-size": str(new_size)}) def __str__(self) -> str: @@ -122,5 +123,5 @@ def wait_for_coalesce(self, fn: Callable[[], R] | None = None) -> R | None: # It is necessary to wait a long time because the GC can be paused for more than 5 minutes. # And it is also necessary to allow a sufficiently long merge time which depends on the amount of data. wait_for(lambda: self.get_parent() != previous_parent, msg="Waiting for coalesce", timeout_secs=10 * 60) - logging.info("Coalesce done") + self.logger.info("Coalesce done") return ret diff --git a/lib/vm.py b/lib/vm.py index 1483f5f03..96fcbc01a 100644 --- a/lib/vm.py +++ b/lib/vm.py @@ -2,12 +2,13 @@ import pytest -import logging import os import subprocess import tempfile import uuid +import structlog + import lib.commands as commands import lib.efi as efi from lib.basevm import BaseVM @@ -33,8 +34,15 @@ from lib.host import Host class VM(BaseVM): + def __init__(self, uuid: str, host: Host) -> None: + self.logger = structlog.get_logger("VM").bind( + pool=host.pool.master_hostname_or_ip, + host=host.hostname_or_ip, + vm_uuid=uuid + ) super().__init__(uuid, host) + self.logger.info("New VM instantiated") self.ip: str | None = None self.previous_host: Host | None = None # previous host when migrated or being migrated self.is_windows = self.param_get('platform', 'device_id', accept_unknown_key=True) == '0002' @@ -58,8 +66,7 @@ def is_paused(self) -> bool: # `on` can be an host name-label or UUID def start(self, on: str | None = None) -> str: - msg_starts_on = f" (on host {on})" if on else "" - logging.info("Start VM" + msg_starts_on) + self.logger.info("Start VM", on=on) args: dict[str, str | bool | dict[str, str]] = {'uuid': self.uuid} if on is not None: args['on'] = on @@ -67,15 +74,15 @@ def start(self, on: str | None = None) -> str: def shutdown(self, force: bool = False, verify: bool = False, force_if_fails: bool = False) -> str: assert not (force and force_if_fails), "force and force_if_fails cannot be both True" - logging.info("Shutdown VM" + (" (force)" if force else "")) + self.logger.info("Shutdown VM", force=force) try: ret = self.host.xe('vm-shutdown', {'uuid': self.uuid, 'force': force}) if verify: - wait_for(self.is_halted, "Wait for VM halted") + wait_for(self.is_halted, "Wait for VM halted", logger=self.logger) except Exception as e: if force_if_fails: - logging.warning("Shutdown failed: %s" % e) + self.logger.warning("Shutdown failed", exception=e) ret = self.shutdown(force=True, verify=verify) else: raise @@ -83,7 +90,7 @@ def shutdown(self, force: bool = False, verify: bool = False, force_if_fails: bo return str(ret) # Ensure return type matches hint, xe can return non-string def reboot(self, force: bool = False, verify: bool = False) -> str: - logging.info("Reboot VM") + self.logger.info("Reboot VM") ret = self.host.xe('vm-reboot', {'uuid': self.uuid, 'force': force}) if verify: # No need to verify that the reboot actually happened because the xe command @@ -100,8 +107,9 @@ def try_get_and_store_ip(self) -> bool: if not ip or ip.startswith('169.254.'): return False else: - logging.info("VM IP: %s" % ip) self.ip = ip + self.logger = self.logger.bind(ip=ip) + self.logger.info("Set VM IP") return True @overload @@ -139,12 +147,12 @@ def ssh(self, cmd: str, *, check: bool = True, simple_output: bool = True, backg # raises by default for any nonzero return code assert self.ip is not None return commands.ssh(self.ip, cmd, check=check, simple_output=simple_output, background=background, - decode=decode) + decode=decode, logger=self.logger) def ssh_with_result(self, cmd: str) -> commands.SSHResult[str]: # doesn't raise if the command's return is nonzero, unless there's a SSH error assert self.ip is not None - return commands.ssh_with_result(self.ip, cmd) + return commands.ssh_with_result(self.ip, cmd, logger=self.logger) def scp(self, src: str, dest: str, check: bool = True, suppress_fingerprint_warnings: bool = True, local_dest: bool = False) -> subprocess.CompletedProcess[bytes]: @@ -191,44 +199,44 @@ def is_management_agent_up(self) -> bool: ) def wait_for_os_booted(self) -> None: - wait_for(self.is_running, "Wait for VM running") + wait_for(self.is_running, "Wait for VM running", logger=self.logger) # waiting for the IP: # - allows to make sure the OS actually started (on VMs that have the management agent) # - allows to store the IP for future use in the VM object - wait_for(self.try_get_and_store_ip, "Wait for VM IP", timeout_secs=5 * 60) + wait_for(self.try_get_and_store_ip, "Wait for VM IP", timeout_secs=5 * 60, logger=self.logger) # now wait also for the management agent to have started - wait_for(self.is_management_agent_up, "Wait for management agent up") + wait_for(self.is_management_agent_up, "Wait for management agent up", logger=self.logger) def wait_for_vm_running_and_ssh_up(self) -> None: self.wait_for_os_booted() - wait_for(self.is_ssh_up, "Wait for SSH up") + wait_for(self.is_ssh_up, "Wait for SSH up", logger=self.logger) def ssh_touch_file(self, filepath: str) -> None: - logging.info("Create file on VM (%s)" % filepath) + self.logger.info("Create file on VM", filepath=filepath) self.ssh(f'touch {filepath}') if not self.is_windows: self.ssh(f'sync {filepath}') - logging.info("Check file created") + self.logger.info("Check file created") self.ssh(f'test -f {filepath}') def suspend(self, verify: bool = False) -> None: - logging.info("Suspend VM") + self.logger.info("Suspend VM") self.host.xe('vm-suspend', {'uuid': self.uuid}) if verify: - wait_for(self.is_suspended, "Wait for VM suspended") + wait_for(self.is_suspended, "Wait for VM suspended", logger=self.logger) def resume(self) -> None: - logging.info("Resume VM") + self.logger.info("Resume VM") self.host.xe('vm-resume', {'uuid': self.uuid}) def pause(self, verify: bool = False) -> None: - logging.info("Pause VM") + self.logger.info("Pause VM") self.host.xe('vm-pause', {'uuid': self.uuid}) if verify: - wait_for(self.is_paused, "Wait for VM paused") + wait_for(self.is_paused, "Wait for VM paused", logger=self.logger) def unpause(self) -> None: - logging.info("Unpause VM") + self.logger.info("Unpause VM") self.host.xe('vm-unpause', {'uuid': self.uuid}) def _disk_list(self) -> str: @@ -245,7 +253,7 @@ def destroy(self, verify: bool = False) -> None: self.host.xe('vm-destroy', {'uuid': self.uuid}) if verify: - wait_for_not(self.exists, "Wait for VM destroyed") + wait_for_not(self.exists, "Wait for VM destroyed", logger=self.logger) def exists(self) -> bool: return self.host.pool_has_vm(self.uuid) @@ -262,15 +270,10 @@ def migrate(self, target_host: Host, sr: SR | None = None, network: str | None = 'live': self.is_running() } cross_pool = self.host.pool.uuid != target_host.pool.uuid - if sr is not None: - if self.get_sr().uuid == sr.uuid: - # Same SR, no need to migrate storage - sr = None - else: - msg += " (SR: %s)" % sr.uuid - if network is not None: - msg += " (Network: %s)" % network - logging.info(msg) + if sr is not None and self.get_sr().uuid == sr.uuid: + # Same SR, no need to migrate storage + sr = None + self.logger.info(msg, network=network, sr_uuid=None if sr is None else sr.uuid) storage_motion = cross_pool or sr is not None or network is not None if storage_motion: @@ -305,7 +308,7 @@ def migrate(self, target_host: Host, sr: SR | None = None, network: str | None = self.create_vdis_list() def snapshot(self, ignore_vdis: List[str] | None = None, name: str | None = None) -> Snapshot: - logging.info("Snapshot VM") + self.logger.info("Snapshot VM") name_label = name or f"Snapshot of {self.uuid}" args: dict[str, str | bool | dict[str, str]] = {'uuid': self.uuid, 'new-name-label': name_label} @@ -315,13 +318,13 @@ def snapshot(self, ignore_vdis: List[str] | None = None, name: str | None = None return Snapshot(snap_uuid, self.host, self) def checkpoint(self) -> Snapshot: - logging.info("Checkpoint VM") + self.logger.info("Checkpoint VM") return Snapshot(self.host.xe('vm-checkpoint', {'uuid': self.uuid, 'new-name-label': 'Checkpoint of %s' % self.uuid}), self.host, self) def connect_vdi(self, vdi: VDI, device: str = "autodetect") -> VBD: - logging.info(f">> Plugging VDI {vdi.uuid} on VM {self.uuid}") + self.logger.info(">> Plugging VDI", vdi_uuid=vdi.uuid) vbd_uuid = self.host.xe("vbd-create", { "vdi-uuid": vdi.uuid, "vm-uuid": self.uuid, @@ -339,7 +342,7 @@ def connect_vdi(self, vdi: VDI, device: str = "autodetect") -> VBD: return VBD(vbd_uuid, self, vdi.name()) def disconnect_vdi(self, vdi: VDI) -> None: - logging.info(f"<< Unplugging VDI {vdi.uuid} from VM {self.uuid}") + self.logger.info("<< Unplugging VDI", vdi_uuid=vdi.uuid) assert vdi in self.vdis, f"VDI {vdi.uuid} not in VM {self.uuid} VDI list" vbd_uuid = self.host.xe("vbd-list", { "vdi-uuid": vdi.uuid, @@ -350,7 +353,7 @@ def disconnect_vdi(self, vdi: VDI) -> None: self.host.xe("vbd-unplug", {"uuid": vbd_uuid}) except commands.SSHCommandFailed as e: if e.stdout == f"The device is not currently attached\ndevice: {vbd_uuid}": - logging.info(f"VBD {vbd_uuid} already unplugged") + self.logger.info("VBD already unplugged", vbd_uuid=vbd_uuid) else: raise self.host.xe("vbd-destroy", {"uuid": vbd_uuid}) @@ -378,7 +381,7 @@ def create_vdis_list(self) -> None: except commands.SSHCommandFailed as e: # Doesn't work with Dom0 since `vm-disk-list` doesn't work on it so we create empty list if e.stdout == "Error: No matching VMs found": - logging.info("Couldn't get disks list. We are Dom0. Continuing...") + self.logger.info("Couldn't get disks list. We are Dom0. Continuing...") self.vdis = [] else: raise @@ -396,7 +399,7 @@ def create_vif(self, vif_num: int, *, network_uuid: str | None = None, if network_name: network_uuid = self.host.pool.network_named(network_name) assert network_uuid, f"No UUID given, and network name {network_name!r} not found" - logging.info("Create VIF %d to network %r on VM %s", vif_num, network_uuid, self.uuid) + self.logger.info("Create VIF", vif=vif_num, network_uuid=network_uuid) vif_uuid = self.host.xe('vif-create', {'vm-uuid': self.uuid, 'device': str(vif_num), 'network-uuid': network_uuid, @@ -413,7 +416,7 @@ def get_residence_host(self) -> Host: def start_background_process(self, cmd: str) -> str: if self.is_windows: - logging.warning('start_background_process is not reliable on Windows') + self.logger.warning('start_background_process is not reliable on Windows') script = "/tmp/bg_process.sh" pidfile = "/tmp/bg_process.pid" with tempfile.NamedTemporaryFile('w') as f: @@ -439,7 +442,7 @@ def start_background_process(self, cmd: str) -> str: self.ssh(remote_cmd, background=True) wait_for(lambda: self.ssh_with_result(f'test -f {pidfile}').returncode == 0, - "wait for pid file %s to exist" % pidfile) + "wait for pid file %s to exist" % pidfile, logger=self.logger) pid = self.ssh(f'cat {pidfile}') self.ssh(f'rm -f {script}') self.ssh(f'rm -f {pidfile}') @@ -473,7 +476,7 @@ def execute_script(self, script_contents: str, simple_output: bool = True) -> st f.flush() self.scp(f.name, f.name) try: - logging.debug(f"[{self.ip}] # Will execute this temporary script:\n{script_contents.strip()}") + self.logger.debug("Will execute this temporary script", script_contents=script_contents) # Use bash to run the script, to avoid being hit by differences between shells, for example on FreeBSD # It is a documented requirement that bash is present on all test VMs. res = self.ssh(f'bash {f.name}', simple_output=simple_output) @@ -520,14 +523,14 @@ def detect_package_manager(self) -> PackageManagerEnum: return PackageManagerEnum.UNKNOWN def insert_cd(self, vdi_name: str) -> None: - logging.info("Insert CD %r in VM %s", vdi_name, self.uuid) + self.logger.info("Insert CD", vdi_name=vdi_name) self.host.xe('vm-cd-insert', {'uuid': self.uuid, 'cd-name': vdi_name}) def insert_guest_tools_iso(self) -> None: self.insert_cd('guest-tools.iso') def eject_cd(self) -> None: - logging.info("Ejecting CD from VM %s", self.uuid) + self.logger.info("Ejecting CD") self.host.xe('vm-cd-eject', {'uuid': self.uuid}) # *** Common reusable test fragments @@ -540,7 +543,7 @@ def test_snapshot_on_running_vm(self) -> None: snapshot.revert() self.start() self.wait_for_vm_running_and_ssh_up() - logging.info("Check file does not exist anymore") + self.logger.info("Check file does not exist anymore") self.ssh(f'test ! -f {filepath}') finally: snapshot.destroy(verify=True) @@ -640,26 +643,26 @@ def get_vtpm_uuid(self) -> str: return self.host.xe('vtpm-list', {'vm-uuid': self.uuid}, minimal=True) def create_vtpm(self) -> str: - logging.info("Creating vTPM for vm %s" % self.uuid) + self.logger.info("Creating vTPM") return self.host.xe('vtpm-create', {'vm-uuid': self.uuid}) def destroy_vtpm(self) -> str: vtpm_uuid = self.get_vtpm_uuid() assert vtpm_uuid, "A vTPM must be present" - logging.info("Destroying vTPM %s" % vtpm_uuid) + self.logger.info("Destroying vTPM") return self.host.xe('vtpm-destroy', {'uuid': vtpm_uuid}, force=True) def create_vbd(self, device: str, vdi_uuid: str) -> VBD: - logging.info("Create VBD %r for VDI %r on VM %s", device, vdi_uuid, self.uuid) + self.logger.info("Create VBD", vbd=device, vdi_uuid=vdi_uuid) vbd_uuid = self.host.xe('vbd-create', {'vm-uuid': self.uuid, 'device': device, 'vdi-uuid': vdi_uuid, }) - logging.info("New VBD %s", vbd_uuid) + self.logger.info("New VBD", vbd_uuid=vbd_uuid) return VBD(vbd_uuid, self, device) def create_cd_vbd(self, device: str, userdevice: str) -> VBD: - logging.info("Create CD VBD %r on VM %s", device, self.uuid) + self.logger.info("Create CD VBD", vbd=device) vbd_uuid = self.host.xe('vbd-create', {'vm-uuid': self.uuid, 'device': device, 'type': 'CD', @@ -667,13 +670,13 @@ def create_cd_vbd(self, device: str, userdevice: str) -> VBD: }) vbd = VBD(vbd_uuid, self, device) vbd.param_set(param_name="userdevice", value=userdevice) - logging.info("New VBD %s", vbd_uuid) + self.logger.info("New VBD", vbd_uuid=vbd_uuid) return vbd def clone(self, *, name: str | None = None) -> "VM": if name is None: name = self.name() + '_clone_for_tests' - logging.info("Clone VM") + self.logger.info("Clone VM") uuid = self.host.xe('vm-clone', {'uuid': self.uuid, 'new-name-label': name}) return VM(uuid, self.host) @@ -698,7 +701,7 @@ def install_uefi_certs(self, auths: Iterable[efi.EFIAuth]) -> None: """ for auth in auths: assert auth.name in ['PK', 'KEK', 'db', 'dbx'] - logging.info(f"Installing UEFI certs to VM {self.uuid}: {[auth.name for auth in auths]}") + self.logger.info("Installing UEFI certs to VM", auth_names=[auth.name for auth in auths]) for auth in auths: self.set_variable_from_file(auth.auth(), auth.guid.as_str(), auth.name, efi.EFI_AT_ATTRS) @@ -762,11 +765,12 @@ def is_in_uefi_shell(self) -> bool: wait_for( lambda: "UEFI Interactive Shell" in res_host.ssh(f'cat -v {tmp_file}'), "Wait for UEFI shell response in pty output", - 10 + 10, + logger=self.logger, ) ret = True except TimeoutError as e: - logging.debug(e) + self.logger.debug(e) pass finally: res_host.ssh(f'screen -S {session} -X quit', check=False) @@ -776,12 +780,12 @@ def is_in_uefi_shell(self) -> bool: def set_uefi_setup_mode(self) -> None: # Note that in XCP-ng 8.2, the VM won't stay in setup mode, because uefistored # will add PK and other certs if available when the guest boots. - logging.info(f"Set VM {self.uuid} to UEFI setup mode") + self.logger.info("Set UEFI setup mode") self.host.ssh(f'varstore-sb-state {self.uuid} setup') def set_uefi_user_mode(self) -> None: # Setting user mode propagates the host's certificates to the VM - logging.info(f"Set VM {self.uuid} to UEFI user mode") + self.logger.info("Set UEFI user mode") self.host.ssh(f'varstore-sb-state {self.uuid} user') def is_uefi_var_present(self, varname: str) -> bool: @@ -858,7 +862,7 @@ def is_windows_pv_device_installed(self) -> bool: # devices may have different statuses (default = installed, vendor = not installed). # For now, make sure all of them share the same status since our tools do not support vendor devices anyway. statuses = output.splitlines() - logging.debug(f"Installed Xen device status: {statuses}") + self.logger.debug("Installed Xen device status", statuses) if all(x == "CM_PROB_NONE" for x in statuses): return True elif all(x == "CM_PROB_FAILED_INSTALL" for x in statuses): @@ -894,17 +898,17 @@ def are_windows_tools_uninstalled(self) -> bool: ) def save_to_cache(self, cache_id: str) -> None: - logging.info("Save VM %s to cache for %r as a clone" % (self.uuid, cache_id)) + self.logger.info("Save VM to cache", cache_id=cache_id) while True: old_vm = self.host.cached_vm(cache_id, sr_uuid=self.host.main_sr_uuid()) if old_vm is None: break - logging.info("Destroying old cache %s first", old_vm.uuid) + self.logger.info("Destroying old cache first", old_vm_uuid=old_vm.uuid) old_vm.destroy() clone = self.clone(name=f"{self.name()} cache") - logging.info(f"Marking VM {clone.uuid} as cached") + self.logger.info("Marking VM as cached", clone_uuid=clone.uuid) clone.param_set('name-description', self.host.vm_cache_key(cache_id)) @overload diff --git a/pyproject.toml b/pyproject.toml index d0664a722..81ae52a53 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,8 @@ dependencies = [ "pytest>=8.0.0", "pytest-dependency", "requests", + "rich>=15.0.0", + "structlog>=26.1.0", ] [dependency-groups] diff --git a/pytest.ini b/pytest.ini index bf08f838d..ccaed403d 100644 --- a/pytest.ini +++ b/pytest.ini @@ -44,8 +44,8 @@ markers = log_level = debug log_cli = 1 log_cli_level = info -log_format = %(asctime)s.%(msecs)03d %(levelname)s %(message)s -log_date_format = %b %d %H:%M:%S +log_format = %(message)s +log_cli_format = %(message)s filterwarnings = error ignore::DeprecationWarning diff --git a/requirements/base.txt b/requirements/base.txt index 5992b12f6..0e9275aea 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -8,3 +8,5 @@ pydantic pytest>=8.0.0 pytest-dependency requests +rich>=15.0.0 +structlog>=26.1.0 diff --git a/tests/pci_passthrough/test_pci_passthrough.py b/tests/pci_passthrough/test_pci_passthrough.py index ac4e20876..b8dd2eff0 100644 --- a/tests/pci_passthrough/test_pci_passthrough.py +++ b/tests/pci_passthrough/test_pci_passthrough.py @@ -29,7 +29,6 @@ def test_access_status_manual_modification(self, host: Host, enabled_pci_uuid: s if hidden_devices == "": hidden_devices = "xen-pciback.hide=" devices = hidden_devices + f"({device_id})" - host.ssh(f'{XEN_CMDLINE} --set-dom0 "{devices}"') assert host.xe("pci-get-dom0-access-status", {"uuid": enabled_pci_uuid}) == "disable_on_reboot" host.reboot(verify=True) diff --git a/tests/unit/test_rescan_block_devices_info.py b/tests/unit/test_rescan_block_devices_info.py index 0a359b56f..8690a757f 100644 --- a/tests/unit/test_rescan_block_devices_info.py +++ b/tests/unit/test_rescan_block_devices_info.py @@ -3,6 +3,8 @@ # flake8: noqa: E501 - lsblk output lines are intentionally long from unittest.mock import MagicMock +import structlog + from lib.common import KiB from lib.host import Host @@ -171,6 +173,7 @@ def _rescan(lsblk_output: str) -> list[Host.BlockDeviceInfo]: host = MagicMock(spec=Host) + host.logger = structlog.get_logger() host.ssh.return_value = lsblk_output Host.rescan_block_devices_info(host) return host.block_devices_info diff --git a/uv.lock b/uv.lock index f79d82da8..91afa7501 100644 --- a/uv.lock +++ b/uv.lock @@ -445,6 +445,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ce/62/b40b382fa0c66fee1478073eb8db352a4a6beda4a1adccf1df911d8c289c/librt-0.11.0-cp314-cp314t-win_arm64.whl", hash = "sha256:dee008f20b542e3cd162ba338a7f9ec0f6d23d395f66fe8aeeec3c9d067ea253", size = 102572, upload-time = "2026-05-10T18:17:06.809Z" }, ] +[[package]] +name = "markdown-it-py" +version = "4.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "mdurl" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/ff/7841249c247aa650a76b9ee4bbaeae59370dc8bfd2f6c01f3630c35eb134/markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49", size = 82454, upload-time = "2026-05-07T12:08:28.36Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/81/4da04ced5a082363ecfa159c010d200ecbd959ae410c10c0264a38cac0f5/markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a", size = 91687, upload-time = "2026-05-07T12:08:27.182Z" }, +] + [[package]] name = "mccabe" version = "0.7.0" @@ -454,6 +466,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/27/1a/1f68f9ba0c207934b35b86a8ca3aad8395a3d6dd7921c0686e23853ff5a9/mccabe-0.7.0-py2.py3-none-any.whl", hash = "sha256:6c2d30ab6be0e4a46919781807b4f0d834ebdd6c6e3dca0bda5a15f863427b6e", size = 7350, upload-time = "2022-01-24T01:14:49.62Z" }, ] +[[package]] +name = "mdurl" +version = "0.1.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, +] + [[package]] name = "mypy" version = "2.1.0" @@ -788,6 +809,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/56/5d/c814546c2333ceea4ba42262d8c4d55763003e767fa169adc693bd524478/requests-2.33.0-py3-none-any.whl", hash = "sha256:3324635456fa185245e24865e810cecec7b4caf933d7eb133dcde67d48cee69b", size = 65017, upload-time = "2026-03-25T15:10:40.382Z" }, ] +[[package]] +name = "rich" +version = "15.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c0/8f/0722ca900cc807c13a6a0c696dacf35430f72e0ec571c4275d2371fca3e9/rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36", size = 230680, upload-time = "2026-04-12T08:24:00.75Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" }, +] + [[package]] name = "ruff" version = "0.15.17" @@ -849,6 +883,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e7/9c/0e6afc12c269578be5c0c1c9f4b49a8d32770a080260c333ac04cc1c832d/soupsieve-2.7-py3-none-any.whl", hash = "sha256:6e60cc5c1ffaf1cebcc12e8188320b72071e922c2e897f737cadce79ad5d30c4", size = 36677, upload-time = "2025-04-20T18:50:07.196Z" }, ] +[[package]] +name = "structlog" +version = "26.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/5e/89/b4a0bcfdf4f71a3dea31379f095929613d7e4528a0996bca6aa964cd0dca/structlog-26.1.0.tar.gz", hash = "sha256:f63a716cbd1b1291cf7661de7794b455acfa4c43c5bcf1630e6ad5ddc1adb3b7", size = 1459881, upload-time = "2026-06-06T07:33:39.348Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a9/18/489c97b834dfff9cf2fc2507cede4bcd4b11e67f84bc462acd1992496f86/structlog-26.1.0-py3-none-any.whl", hash = "sha256:e081a26d6c373e6d201eca24eede26d8ffab07f88f477822e679183428d3d91e", size = 73764, upload-time = "2026-06-06T07:33:38.046Z" }, +] + [[package]] name = "types-colorama" version = "0.4.15.20260508" @@ -944,6 +987,8 @@ dependencies = [ { name = "pytest" }, { name = "pytest-dependency" }, { name = "requests" }, + { name = "rich" }, + { name = "structlog" }, ] [package.dev-dependencies] @@ -977,6 +1022,8 @@ requires-dist = [ { name = "pytest", specifier = ">=8.0.0" }, { name = "pytest-dependency" }, { name = "requests" }, + { name = "rich", specifier = ">=15.0.0" }, + { name = "structlog", specifier = ">=26.1.0" }, ] [package.metadata.requires-dev]