xref: /openbmc/qemu/python/qemu/machine/machine.py (revision 684750ab)

"""
QEMU machine module:

The machine module primarily provides the QEMUMachine class,
which provides facilities for managing the lifetime of a QEMU VM.
"""

# Copyright (C) 2015-2016 Red Hat Inc.
# Copyright (C) 2012 IBM Corp.
#
# Authors:
#  Fam Zheng <famz@redhat.com>
#
# This work is licensed under the terms of the GNU GPL, version 2.  See
# the COPYING file in the top-level directory.
#
# Based on qmp.py.
#

import errno
from itertools import chain
import locale
import logging
import os
import shutil
import signal
import socket
import subprocess
import tempfile
from types import TracebackType
from typing import (
    Any,
    BinaryIO,
    Dict,
    List,
    Optional,
    Sequence,
    Tuple,
    Type,
    TypeVar,
)

from qemu.qmp import SocketAddrT
from qemu.qmp.legacy import (
    QEMUMonitorProtocol,
    QMPMessage,
    QMPReturnValue,
)

from . import console_socket


LOG = logging.getLogger(__name__)


class QEMUMachineError(Exception):
    """
    Exception called when an error in QEMUMachine happens.
    """


class QEMUMachineAddDeviceError(QEMUMachineError):
    """
    Exception raised when a request to add a device can not be fulfilled

    The failures are caused by limitations, lack of information or conflicting
    requests on the QEMUMachine methods.  This exception does not represent
    failures reported by the QEMU binary itself.
    """


class VMLaunchFailure(QEMUMachineError):
    """
    Exception raised when a VM launch was attempted, but failed.
    """
    def __init__(self, exitcode: Optional[int],
                 command: str, output: Optional[str]):
        super().__init__(exitcode, command, output)
        self.exitcode = exitcode
        self.command = command
        self.output = output

    def __str__(self) -> str:
        ret = ''
        if self.__cause__ is not None:
            name = type(self.__cause__).__name__
            reason = str(self.__cause__)
            if reason:
                ret += f"{name}: {reason}"
            else:
                ret += f"{name}"
        ret += '\n'

        if self.exitcode is not None:
            ret += f"\tExit code: {self.exitcode}\n"
        ret += f"\tCommand: {self.command}\n"
        ret += f"\tOutput: {self.output}\n"
        return ret


class AbnormalShutdown(QEMUMachineError):
    """
    Exception raised when a graceful shutdown was requested, but not performed.
    """


_T = TypeVar('_T', bound='QEMUMachine')


class QEMUMachine:
    """
    A QEMU VM.

    Use this object as a context manager to ensure
    the QEMU process terminates::

        with VM(binary) as vm:
            ...
        # vm is guaranteed to be shut down here
    """
    # pylint: disable=too-many-instance-attributes, too-many-public-methods

    def __init__(self,
                 binary: str,
                 args: Sequence[str] = (),
                 wrapper: Sequence[str] = (),
                 name: Optional[str] = None,
                 base_temp_dir: str = "/var/tmp",
                 monitor_address: Optional[SocketAddrT] = None,
                 drain_console: bool = False,
                 console_log: Optional[str] = None,
                 log_dir: Optional[str] = None,
                 qmp_timer: Optional[float] = 30):
        '''
        Initialize a QEMUMachine

        @param binary: path to the qemu binary
        @param args: list of extra arguments
        @param wrapper: list of arguments used as prefix to qemu binary
        @param name: prefix for socket and log file names (default: qemu-PID)
        @param base_temp_dir: default location where temp files are created
        @param monitor_address: address for QMP monitor
        @param drain_console: (optional) True to drain console socket to buffer
        @param console_log: (optional) path to console log file
        @param log_dir: where to create and keep log files
        @param qmp_timer: (optional) default QMP socket timeout
        @note: Qemu process is not started until launch() is used.
        '''
        # pylint: disable=too-many-arguments

        # Direct user configuration

        self._binary = binary
        self._args = list(args)
        self._wrapper = wrapper
        self._qmp_timer = qmp_timer

        self._name = name or f"{id(self):x}"
        self._sock_pair: Optional[Tuple[socket.socket, socket.socket]] = None
        self._cons_sock_pair: Optional[
            Tuple[socket.socket, socket.socket]] = None
        self._temp_dir: Optional[str] = None
        self._base_temp_dir = base_temp_dir
        self._log_dir = log_dir

        self._monitor_address = monitor_address

        self._console_log_path = console_log
        if self._console_log_path:
            # In order to log the console, buffering needs to be enabled.
            self._drain_console = True
        else:
            self._drain_console = drain_console

        # Runstate
        self._qemu_log_path: Optional[str] = None
        self._qemu_log_file: Optional[BinaryIO] = None
        self._popen: Optional['subprocess.Popen[bytes]'] = None
        self._events: List[QMPMessage] = []
        self._iolog: Optional[str] = None
        self._qmp_set = True   # Enable QMP monitor by default.
        self._qmp_connection: Optional[QEMUMonitorProtocol] = None
        self._qemu_full_args: Tuple[str, ...] = ()
        self._launched = False
        self._machine: Optional[str] = None
        self._console_index = 0
        self._console_set = False
        self._console_device_type: Optional[str] = None
        self._console_socket: Optional[socket.socket] = None
        self._console_file: Optional[socket.SocketIO] = None
        self._remove_files: List[str] = []
        self._user_killed = False
        self._quit_issued = False

    def __enter__(self: _T) -> _T:
        return self

    def __exit__(self,
                 exc_type: Optional[Type[BaseException]],
                 exc_val: Optional[BaseException],
                 exc_tb: Optional[TracebackType]) -> None:
        self.shutdown()

    def add_monitor_null(self) -> None:
        """
        This can be used to add an unused monitor instance.
        """
        self._args.append('-monitor')
        self._args.append('null')

    def add_fd(self: _T, fd: int, fdset: int,
               opaque: str, opts: str = '') -> _T:
        """
        Pass a file descriptor to the VM
        """
        options = ['fd=%d' % fd,
                   'set=%d' % fdset,
                   'opaque=%s' % opaque]
        if opts:
            options.append(opts)

        # This did not exist before 3.4, but since then it is
        # mandatory for our purpose
        if hasattr(os, 'set_inheritable'):
            os.set_inheritable(fd, True)

        self._args.append('-add-fd')
        self._args.append(','.join(options))
        return self

    def send_fd_scm(self, fd: Optional[int] = None,
                    file_path: Optional[str] = None) -> int:
        """
        Send an fd or file_path to the remote via SCM_RIGHTS.

        Exactly one of fd and file_path must be given.  If it is
        file_path, the file will be opened read-only and the new file
        descriptor will be sent to the remote.
        """
        if file_path is not None:
            assert fd is None
            with open(file_path, "rb") as passfile:
                fd = passfile.fileno()
                self._qmp.send_fd_scm(fd)
        else:
            assert fd is not None
            self._qmp.send_fd_scm(fd)

        return 0

    @staticmethod
    def _remove_if_exists(path: str) -> None:
        """
        Remove file object at path if it exists
        """
        try:
            os.remove(path)
        except OSError as exception:
            if exception.errno == errno.ENOENT:
                return
            raise

    def is_running(self) -> bool:
        """Returns true if the VM is running."""
        return self._popen is not None and self._popen.poll() is None

    @property
    def _subp(self) -> 'subprocess.Popen[bytes]':
        if self._popen is None:
            raise QEMUMachineError('Subprocess pipe not present')
        return self._popen

    def exitcode(self) -> Optional[int]:
        """Returns the exit code if possible, or None."""
        if self._popen is None:
            return None
        return self._popen.poll()

    def get_pid(self) -> Optional[int]:
        """Returns the PID of the running process, or None."""
        if not self.is_running():
            return None
        return self._subp.pid

    def _load_io_log(self) -> None:
        # Assume that the output encoding of QEMU's terminal output is
        # defined by our locale. If indeterminate, allow open() to fall
        # back to the platform default.
        _, encoding = locale.getlocale()
        if self._qemu_log_path is not None:
            with open(self._qemu_log_path, "r", encoding=encoding) as iolog:
                self._iolog = iolog.read()

    @property
    def _base_args(self) -> List[str]:
        args = ['-display', 'none', '-vga', 'none']

        if self._qmp_set:
            if self._sock_pair:
                moncdev = f"socket,id=mon,fd={self._sock_pair[0].fileno()}"
            elif isinstance(self._monitor_address, tuple):
                moncdev = "socket,id=mon,host={},port={}".format(
                    *self._monitor_address
                )
            else:
                moncdev = f"socket,id=mon,path={self._monitor_address}"
            args.extend(['-chardev', moncdev, '-mon',
                         'chardev=mon,mode=control'])

        if self._machine is not None:
            args.extend(['-machine', self._machine])
        for _ in range(self._console_index):
            args.extend(['-serial', 'null'])
        if self._console_set:
            assert self._cons_sock_pair is not None
            fd = self._cons_sock_pair[0].fileno()
            chardev = f"socket,id=console,fd={fd}"
            args.extend(['-chardev', chardev])
            if self._console_device_type is None:
                args.extend(['-serial', 'chardev:console'])
            else:
                device = '%s,chardev=console' % self._console_device_type
                args.extend(['-device', device])
        return args

    @property
    def args(self) -> List[str]:
        """Returns the list of arguments given to the QEMU binary."""
        return self._args

    def _pre_launch(self) -> None:
        if self._qmp_set:
            if self._monitor_address is None:
                self._sock_pair = socket.socketpair()
                os.set_inheritable(self._sock_pair[0].fileno(), True)
                sock = self._sock_pair[1]
            if isinstance(self._monitor_address, str):
                self._remove_files.append(self._monitor_address)

            sock_or_addr = self._monitor_address or sock
            assert sock_or_addr is not None

            self._qmp_connection = QEMUMonitorProtocol(
                sock_or_addr,
                server=bool(self._monitor_address),
                nickname=self._name
            )

        if self._console_set:
            self._cons_sock_pair = socket.socketpair()
            os.set_inheritable(self._cons_sock_pair[0].fileno(), True)

        # NOTE: Make sure any opened resources are *definitely* freed in
        # _post_shutdown()!
        # pylint: disable=consider-using-with
        self._qemu_log_path = os.path.join(self.log_dir, self._name + ".log")
        self._qemu_log_file = open(self._qemu_log_path, 'wb')

        self._iolog = None
        self._qemu_full_args = tuple(chain(
            self._wrapper,
            [self._binary],
            self._base_args,
            self._args
        ))

    def _post_launch(self) -> None:
        if self._sock_pair:
            self._sock_pair[0].close()
        if self._cons_sock_pair:
            self._cons_sock_pair[0].close()

        if self._qmp_connection:
            if self._sock_pair:
                self._qmp.connect()
            else:
                self._qmp.accept(self._qmp_timer)

    def _close_qemu_log_file(self) -> None:
        if self._qemu_log_file is not None:
            self._qemu_log_file.close()
            self._qemu_log_file = None

    def _post_shutdown(self) -> None:
        """
        Called to cleanup the VM instance after the process has exited.
        May also be called after a failed launch.
        """
        LOG.debug("Cleaning up after VM process")
        try:
            self._close_qmp_connection()
        except Exception as err:  # pylint: disable=broad-except
            LOG.warning(
                "Exception closing QMP connection: %s",
                str(err) if str(err) else type(err).__name__
            )
        finally:
            assert self._qmp_connection is None

        if self._sock_pair:
            self._sock_pair[0].close()
            self._sock_pair[1].close()
            self._sock_pair = None

        self._close_qemu_log_file()

        self._load_io_log()

        self._qemu_log_path = None

        if self._temp_dir is not None:
            shutil.rmtree(self._temp_dir)
            self._temp_dir = None

        while len(self._remove_files) > 0:
            self._remove_if_exists(self._remove_files.pop())

        exitcode = self.exitcode()
        if (exitcode is not None and exitcode < 0
                and not (self._user_killed and exitcode == -signal.SIGKILL)):
            msg = 'qemu received signal %i; command: "%s"'
            if self._qemu_full_args:
                command = ' '.join(self._qemu_full_args)
            else:
                command = ''
            LOG.warning(msg, -int(exitcode), command)

        self._quit_issued = False
        self._user_killed = False
        self._launched = False

    def launch(self) -> None:
        """
        Launch the VM and make sure we cleanup and expose the
        command line/output in case of exception
        """

        if self._launched:
            raise QEMUMachineError('VM already launched')

        try:
            self._launch()
        except BaseException as exc:
            # We may have launched the process but it may
            # have exited before we could connect via QMP.
            # Assume the VM didn't launch or is exiting.
            # If we don't wait for the process, exitcode() may still be
            # 'None' by the time control is ceded back to the caller.
            if self._launched:
                self.wait()
            else:
                self._post_shutdown()

            if isinstance(exc, Exception):
                raise VMLaunchFailure(
                    exitcode=self.exitcode(),
                    command=' '.join(self._qemu_full_args),
                    output=self._iolog
                ) from exc

            # Don't wrap 'BaseException'; doing so would downgrade
            # that exception. However, we still want to clean up.
            raise

    def _launch(self) -> None:
        """
        Launch the VM and establish a QMP connection
        """
        self._pre_launch()
        LOG.debug('VM launch command: %r', ' '.join(self._qemu_full_args))

        # Cleaning up of this subprocess is guaranteed by _do_shutdown.
        # pylint: disable=consider-using-with
        self._popen = subprocess.Popen(self._qemu_full_args,
                                       stdin=subprocess.DEVNULL,
                                       stdout=self._qemu_log_file,
                                       stderr=subprocess.STDOUT,
                                       shell=False,
                                       close_fds=False)
        self._launched = True
        self._post_launch()

    def _close_qmp_connection(self) -> None:
        """
        Close the underlying QMP connection, if any.

        Dutifully report errors that occurred while closing, but assume
        that any error encountered indicates an abnormal termination
        process and not a failure to close.
        """
        if self._qmp_connection is None:
            return

        try:
            self._qmp.close()
        except EOFError:
            # EOF can occur as an Exception here when using the Async
            # QMP backend. It indicates that the server closed the
            # stream. If we successfully issued 'quit' at any point,
            # then this was expected. If the remote went away without
            # our permission, it's worth reporting that as an abnormal
            # shutdown case.
            if not (self._user_killed or self._quit_issued):
                raise
        finally:
            self._qmp_connection = None

    def _early_cleanup(self) -> None:
        """
        Perform any cleanup that needs to happen before the VM exits.

        This method may be called twice upon shutdown, once each by soft
        and hard shutdown in failover scenarios.
        """
        # If we keep the console socket open, we may deadlock waiting
        # for QEMU to exit, while QEMU is waiting for the socket to
        # become writable.
        if self._console_file is not None:
            LOG.debug("Closing console file")
            self._console_file.close()
            self._console_file = None

        if self._console_socket is not None:
            LOG.debug("Closing console socket")
            self._console_socket.close()
            self._console_socket = None

        if self._cons_sock_pair:
            self._cons_sock_pair[0].close()
            self._cons_sock_pair[1].close()
            self._cons_sock_pair = None

    def _hard_shutdown(self) -> None:
        """
        Perform early cleanup, kill the VM, and wait for it to terminate.

        :raise subprocess.Timeout: When timeout is exceeds 60 seconds
            waiting for the QEMU process to terminate.
        """
        LOG.debug("Performing hard shutdown")
        self._early_cleanup()
        self._subp.kill()
        self._subp.wait(timeout=60)

    def _soft_shutdown(self, timeout: Optional[int]) -> None:
        """
        Perform early cleanup, attempt to gracefully shut down the VM, and wait
        for it to terminate.

        :param timeout: Timeout in seconds for graceful shutdown.
                        A value of None is an infinite wait.

        :raise ConnectionReset: On QMP communication errors
        :raise subprocess.TimeoutExpired: When timeout is exceeded waiting for
            the QEMU process to terminate.
        """
        LOG.debug("Attempting graceful termination")

        self._early_cleanup()

        if self._quit_issued:
            LOG.debug(
                "Anticipating QEMU termination due to prior 'quit' command, "
                "or explicit call to wait()"
            )
        else:
            LOG.debug("Politely asking QEMU to terminate")

        if self._qmp_connection:
            try:
                if not self._quit_issued:
                    # May raise ExecInterruptedError or StateError if the
                    # connection dies or has *already* died.
                    self.qmp('quit')
            finally:
                # Regardless, we want to quiesce the connection.
                self._close_qmp_connection()
        elif not self._quit_issued:
            LOG.debug(
                "Not anticipating QEMU quit and no QMP connection present, "
                "issuing SIGTERM"
            )
            self._subp.terminate()

        # May raise subprocess.TimeoutExpired
        LOG.debug(
            "Waiting (timeout=%s) for QEMU process (pid=%s) to terminate",
            timeout, self._subp.pid
        )
        self._subp.wait(timeout=timeout)

    def _do_shutdown(self, timeout: Optional[int]) -> None:
        """
        Attempt to shutdown the VM gracefully; fallback to a hard shutdown.

        :param timeout: Timeout in seconds for graceful shutdown.
                        A value of None is an infinite wait.

        :raise AbnormalShutdown: When the VM could not be shut down gracefully.
            The inner exception will likely be ConnectionReset or
            subprocess.TimeoutExpired. In rare cases, non-graceful termination
            may result in its own exceptions, likely subprocess.TimeoutExpired.
        """
        try:
            self._soft_shutdown(timeout)
        except Exception as exc:
            if isinstance(exc, subprocess.TimeoutExpired):
                LOG.debug("Timed out waiting for QEMU process to exit")
            LOG.debug("Graceful shutdown failed", exc_info=True)
            LOG.debug("Falling back to hard shutdown")
            self._hard_shutdown()
            raise AbnormalShutdown("Could not perform graceful shutdown") \
                from exc

    def shutdown(self,
                 hard: bool = False,
                 timeout: Optional[int] = 30) -> None:
        """
        Terminate the VM (gracefully if possible) and perform cleanup.
        Cleanup will always be performed.

        If the VM has not yet been launched, or shutdown(), wait(), or kill()
        have already been called, this method does nothing.

        :param hard: When true, do not attempt graceful shutdown, and
                     suppress the SIGKILL warning log message.
        :param timeout: Optional timeout in seconds for graceful shutdown.
                        Default 30 seconds, A `None` value is an infinite wait.
        """
        if not self._launched:
            return

        LOG.debug("Shutting down VM appliance; timeout=%s", timeout)
        if hard:
            LOG.debug("Caller requests immediate termination of QEMU process.")

        try:
            if hard:
                self._user_killed = True
                self._hard_shutdown()
            else:
                self._do_shutdown(timeout)
        finally:
            self._post_shutdown()

    def kill(self) -> None:
        """
        Terminate the VM forcefully, wait for it to exit, and perform cleanup.
        """
        self.shutdown(hard=True)

    def wait(self, timeout: Optional[int] = 30) -> None:
        """
        Wait for the VM to power off and perform post-shutdown cleanup.

        :param timeout: Optional timeout in seconds. Default 30 seconds.
                        A value of `None` is an infinite wait.
        """
        self._quit_issued = True
        self.shutdown(timeout=timeout)

    def set_qmp_monitor(self, enabled: bool = True) -> None:
        """
        Set the QMP monitor.

        @param enabled: if False, qmp monitor options will be removed from
                        the base arguments of the resulting QEMU command
                        line. Default is True.

        .. note:: Call this function before launch().
        """
        self._qmp_set = enabled

    @property
    def _qmp(self) -> QEMUMonitorProtocol:
        if self._qmp_connection is None:
            raise QEMUMachineError("Attempt to access QMP with no connection")
        return self._qmp_connection

    @classmethod
    def _qmp_args(cls, conv_keys: bool,
                  args: Dict[str, Any]) -> Dict[str, object]:
        if conv_keys:
            return {k.replace('_', '-'): v for k, v in args.items()}

        return args

    def qmp(self, cmd: str,
            args_dict: Optional[Dict[str, object]] = None,
            conv_keys: Optional[bool] = None,
            **args: Any) -> QMPMessage:
        """
        Invoke a QMP command and return the response dict
        """
        if args_dict is not None:
            assert not args
            assert conv_keys is None
            args = args_dict
            conv_keys = False

        if conv_keys is None:
            conv_keys = True

        qmp_args = self._qmp_args(conv_keys, args)
        ret = self._qmp.cmd_raw(cmd, args=qmp_args)
        if cmd == 'quit' and 'error' not in ret and 'return' in ret:
            self._quit_issued = True
        return ret

    def cmd(self, cmd: str,
            conv_keys: bool = True,
            **args: Any) -> QMPReturnValue:
        """
        Invoke a QMP command.
        On success return the response dict.
        On failure raise an exception.
        """
        qmp_args = self._qmp_args(conv_keys, args)
        ret = self._qmp.cmd(cmd, **qmp_args)
        if cmd == 'quit':
            self._quit_issued = True
        return ret

    def get_qmp_event(self, wait: bool = False) -> Optional[QMPMessage]:
        """
        Poll for one queued QMP events and return it
        """
        if self._events:
            return self._events.pop(0)
        return self._qmp.pull_event(wait=wait)

    def get_qmp_events(self, wait: bool = False) -> List[QMPMessage]:
        """
        Poll for queued QMP events and return a list of dicts
        """
        events = self._qmp.get_events(wait=wait)
        events.extend(self._events)
        del self._events[:]
        return events

    @staticmethod
    def event_match(event: Any, match: Optional[Any]) -> bool:
        """
        Check if an event matches optional match criteria.

        The match criteria takes the form of a matching subdict. The event is
        checked to be a superset of the subdict, recursively, with matching
        values whenever the subdict values are not None.

        This has a limitation that you cannot explicitly check for None values.

        Examples, with the subdict queries on the left:
         - None matches any object.
         - {"foo": None} matches {"foo": {"bar": 1}}
         - {"foo": None} matches {"foo": 5}
         - {"foo": {"abc": None}} does not match {"foo": {"bar": 1}}
         - {"foo": {"rab": 2}} matches {"foo": {"bar": 1, "rab": 2}}
        """
        if match is None:
            return True

        try:
            for key in match:
                if key in event:
                    if not QEMUMachine.event_match(event[key], match[key]):
                        return False
                else:
                    return False
            return True
        except TypeError:
            # either match or event wasn't iterable (not a dict)
            return bool(match == event)

    def event_wait(self, name: str,
                   timeout: float = 60.0,
                   match: Optional[QMPMessage] = None) -> Optional[QMPMessage]:
        """
        event_wait waits for and returns a named event from QMP with a timeout.

        name: The event to wait for.
        timeout: QEMUMonitorProtocol.pull_event timeout parameter.
        match: Optional match criteria. See event_match for details.
        """
        return self.events_wait([(name, match)], timeout)

    def events_wait(self,
                    events: Sequence[Tuple[str, Any]],
                    timeout: float = 60.0) -> Optional[QMPMessage]:
        """
        events_wait waits for and returns a single named event from QMP.
        In the case of multiple qualifying events, this function returns the
        first one.

        :param events: A sequence of (name, match_criteria) tuples.
                       The match criteria are optional and may be None.
                       See event_match for details.
        :param timeout: Optional timeout, in seconds.
                        See QEMUMonitorProtocol.pull_event.

        :raise asyncio.TimeoutError:
            If timeout was non-zero and no matching events were found.

        :return: A QMP event matching the filter criteria.
                 If timeout was 0 and no event matched, None.
        """
        def _match(event: QMPMessage) -> bool:
            for name, match in events:
                if event['event'] == name and self.event_match(event, match):
                    return True
            return False

        event: Optional[QMPMessage]

        # Search cached events
        for event in self._events:
            if _match(event):
                self._events.remove(event)
                return event

        # Poll for new events
        while True:
            event = self._qmp.pull_event(wait=timeout)
            if event is None:
                # NB: None is only returned when timeout is false-ish.
                # Timeouts raise asyncio.TimeoutError instead!
                break
            if _match(event):
                return event
            self._events.append(event)

        return None

    def get_log(self) -> Optional[str]:
        """
        After self.shutdown or failed qemu execution, this returns the output
        of the qemu process.
        """
        return self._iolog

    def add_args(self, *args: str) -> None:
        """
        Adds to the list of extra arguments to be given to the QEMU binary
        """
        self._args.extend(args)

    def set_machine(self, machine_type: str) -> None:
        """
        Sets the machine type

        If set, the machine type will be added to the base arguments
        of the resulting QEMU command line.
        """
        self._machine = machine_type

    def set_console(self,
                    device_type: Optional[str] = None,
                    console_index: int = 0) -> None:
        """
        Sets the device type for a console device

        If set, the console device and a backing character device will
        be added to the base arguments of the resulting QEMU command
        line.

        This is a convenience method that will either use the provided
        device type, or default to a "-serial chardev:console" command
        line argument.

        The actual setting of command line arguments will be be done at
        machine launch time, as it depends on the temporary directory
        to be created.

        @param device_type: the device type, such as "isa-serial".  If
                            None is given (the default value) a "-serial
                            chardev:console" command line argument will
                            be used instead, resorting to the machine's
                            default device type.
        @param console_index: the index of the console device to use.
                              If not zero, the command line will create
                              'index - 1' consoles and connect them to
                              the 'null' backing character device.
        """
        self._console_set = True
        self._console_device_type = device_type
        self._console_index = console_index

    @property
    def console_socket(self) -> socket.socket:
        """
        Returns a socket connected to the console
        """
        if self._console_socket is None:
            LOG.debug("Opening console socket")
            if not self._console_set:
                raise QEMUMachineError(
                    "Attempt to access console socket with no connection")
            assert self._cons_sock_pair is not None
            # os.dup() is used here for sock_fd because otherwise we'd
            # have two rich python socket objects that would each try to
            # close the same underlying fd when either one gets garbage
            # collected.
            self._console_socket = console_socket.ConsoleSocket(
                sock_fd=os.dup(self._cons_sock_pair[1].fileno()),
                file=self._console_log_path,
                drain=self._drain_console)
            self._cons_sock_pair[1].close()
        return self._console_socket

    @property
    def console_file(self) -> socket.SocketIO:
        """
        Returns a file associated with the console socket
        """
        if self._console_file is None:
            LOG.debug("Opening console file")
            self._console_file = self.console_socket.makefile(mode='rb',
                                                              buffering=0,
                                                              encoding='utf-8')
        return self._console_file

    @property
    def temp_dir(self) -> str:
        """
        Returns a temporary directory to be used for this machine
        """
        if self._temp_dir is None:
            self._temp_dir = tempfile.mkdtemp(prefix="qemu-machine-",
                                              dir=self._base_temp_dir)
        return self._temp_dir

    @property
    def log_dir(self) -> str:
        """
        Returns a directory to be used for writing logs
        """
        if self._log_dir is None:
            return self.temp_dir
        return self._log_dir