linux_virtual_machine.py

# Copyright 2019 PerfKitBenchmarker Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Module containing mixin classes for linux virtual machines.

These classes allow installation on both Debian and RHEL based linuxes.
They also handle some initial setup (especially on RHEL based linuxes
since by default sudo commands without a tty don't work) and
can restore the VM to the state it was in before packages were
installed.

To install a package on a VM, just call vm.Install(package_name).
The package name is just the name of the package module (i.e. the
file name minus .py). The framework will take care of all cleanup
for you.
"""

import abc
import collections
import copy
import json
import logging
import os
import pipes
import posixpath
import re
import threading
import time
from typing import Any, Dict, Optional, Set, Tuple, Union
import uuid

from absl import flags
from packaging import version as packaging_version
from perfkitbenchmarker import background_tasks
from perfkitbenchmarker import disk
from perfkitbenchmarker import errors
from perfkitbenchmarker import linux_packages
from perfkitbenchmarker import os_types
from perfkitbenchmarker import regex_util
from perfkitbenchmarker import sample
from perfkitbenchmarker import virtual_machine
from perfkitbenchmarker import vm_util
import yaml

FLAGS = flags.FLAGS


OS_PRETTY_NAME_REGEXP = r'PRETTY_NAME="(.*)"'
_EPEL_URL = (
    'https://dl.fedoraproject.org/pub/epel/epel-release-latest-{}.noarch.rpm'
)
CLEAR_BUILD_REGEXP = r'Installed version:\s*(.*)\s*'
UPDATE_RETRIES = 5
DEFAULT_SSH_PORT = 22
REMOTE_KEY_PATH = '~/.ssh/id_rsa'
CONTAINER_MOUNT_DIR = '/mnt'
CONTAINER_WORK_DIR = '/root'

# This pair of scripts used for executing long-running commands, which will be
# resilient in the face of SSH connection errors.
# EXECUTE_COMMAND runs a command, streaming stdout / stderr to a file, then
# writing the return code to a file. An exclusive lock is acquired on the return
# code file, so that other processes may wait for completion.
EXECUTE_COMMAND = 'execute_command.py'
# WAIT_FOR_COMMAND waits on the file lock created by EXECUTE_COMMAND,
# then copies the stdout and stderr, exiting with the status of the command run
# by EXECUTE_COMMAND.
WAIT_FOR_COMMAND = 'wait_for_command.py'

_DEFAULT_DISK_FS_TYPE = 'ext4'
_DEFAULT_DISK_MOUNT_OPTIONS = 'discard'
_DEFAULT_DISK_FSTAB_OPTIONS = 'defaults'

# regex for parsing lscpu and /proc/cpuinfo
_COLON_SEPARATED_RE = re.compile(r'^\s*(?P<key>.*?)\s*:\s*(?P<value>.*?)\s*$')

_SYSFS_CPU_PATH = '/sys/devices/system/cpu'

# TODO(user): update these to use a flag holder as recommended
# in go/python-tips/051
flags.DEFINE_bool(
    'setup_remote_firewall',
    False,
    'Whether PKB should configure the firewall of each remote'
    'VM to make sure it accepts all internal connections.',
)

flags.DEFINE_list(
    'sysctl',
    [],
    'Sysctl values to set. This flag should be a comma-separated '
    'list of path=value pairs. Each pair will be appended to'
    '/etc/sysctl.conf.  '
    'For example, if you pass '
    '--sysctls=vm.dirty_background_ratio=10,vm.dirty_ratio=25, '
    'PKB will append "vm.dirty_background_ratio=10" and'
    '"vm.dirty_ratio=25" on separate lines to /etc/sysctrl.conf',
)

flags.DEFINE_bool(
    'reboot_after_changing_sysctl',
    False,
    'Whether PKB should reboot after applying sysctl changes',
)

flags.DEFINE_list(
    'set_files',
    [],
    'Arbitrary filesystem configuration. This flag should be a '
    'comma-separated list of path=value pairs. Each value will '
    'be written to the corresponding path. For example, if you '
    'pass --set_files=/sys/kernel/mm/transparent_hugepage/enabled=always, '
    'then PKB will write "always" to '
    '/sys/kernel/mm/transparent_hugepage/enabled before starting '
    'the benchmark.',
)

flags.DEFINE_bool(
    'network_enable_BBR',
    False,
    'A shortcut to enable BBR congestion control on the network. '
    'equivalent to appending to --sysctls the following values '
    '"net.core.default_qdisc=fq, '
    '"net.ipv4.tcp_congestion_control=bbr" ',
)

flags.DEFINE_integer(
    'num_disable_cpus',
    None,
    'Number of CPUs to disable on the virtual machine.'
    'If the VM has n CPUs, you can disable at most n-1.',
    lower_bound=1,
)
flags.DEFINE_integer('disk_fill_size', 0, 'Size of file to create in GBs.')
flags.DEFINE_enum(
    'disk_fs_type',
    _DEFAULT_DISK_FS_TYPE,
    [_DEFAULT_DISK_FS_TYPE, 'xfs'],
    'File system type used to format disk.',
)
flags.DEFINE_integer(
    'disk_block_size',
    None,
    'Block size to format disk with.Defaults to 4096 for ext4.',
)

flags.DEFINE_bool(
    'enable_transparent_hugepages',
    None,
    'Whether to enable or '
    'disable transparent hugepages. If unspecified, the setting '
    'is unchanged from the default in the OS.',
)

flags.DEFINE_integer(
    'ssh_retries', 10, 'Default number of times to retry SSH.', lower_bound=0
)

flags.DEFINE_integer(
    'scp_connect_timeout', 30, 'timeout for SCP connection.', lower_bound=0
)

flags.DEFINE_string(
    'append_kernel_command_line',
    None,
    'String to append to the kernel command line. The presence of any '
    'non-empty string will cause a reboot to occur after VM prepare. '
    'If unspecified, the kernel command line will be unmodified.',
)

flags.DEFINE_integer(
    'tcp_max_receive_buffer',
    None,
    'Changes the third component of the sysctl value net.ipv4.tcp_rmem. '
    'This sets the maximum receive buffer for TCP socket connections in bytes. '
    'Increasing this value may increase single stream TCP throughput '
    'for high latency connections',
)

flags.DEFINE_integer(
    'tcp_max_send_buffer',
    None,
    'Changes the third component of the sysctl value net.ipv4.tcp_wmem. '
    'This sets the maximum send buffer for TCP socket connections in bytes. '
    'Increasing this value may increase single stream TCP throughput '
    'for high latency connections',
)

_TCP_MAX_NOTSENT_BYTES = flags.DEFINE_integer(
    'tcp_max_notsent_bytes',
    None,
    'Changes the third component of the sysctl value '
    'net.ipv4.tcp_notsent_lowat. This sets the maximum number of unsent bytes '
    'for TCP socket connections. Decreasing this value may to reduce usage '
    'of kernel memory.',
)

flags.DEFINE_integer(
    'rmem_max',
    None,
    'Sets the sysctl value net.core.rmem_max. This sets the max OS '
    'receive buffer size in bytes for all types of connections',
)

flags.DEFINE_integer(
    'wmem_max',
    None,
    'Sets the sysctl value net.core.wmem_max. This sets the max OS '
    'send buffer size in bytes for all types of connections',
)

flags.DEFINE_boolean(
    'gce_hpc_tools', False, 'Whether to apply the hpc-tools environment script.'
)

flags.DEFINE_boolean(
    'disable_smt',
    False,
    'Whether to disable SMT (Simultaneous Multithreading) in BIOS.',
)

flags.DEFINE_boolean(
    'use_numcpu_multi_files',
    False,
    'Whether to use /sys/fs/cgroup/cpuset.cpus.effective, '
    '/dev/cgroup/cpuset.cpus.effective, /proc/self/status, '
    '/proc/cpuinfo to extract the number of CPUs.',
)

flags.DEFINE_boolean(
    'use_cgroup_memory_limits',
    False,
    'Whether to use the cgroup memory limits, read from '
    '/sys/fs/cgroup/memory/{container_name}/memory.limit_in_bytes, '
    'to extract the total available memory capacity in the container.',
)

_DISABLE_YUM_CRON = flags.DEFINE_boolean(
    'disable_yum_cron', True, 'Whether to disable the cron-run yum service.'
)
_KERNEL_MODULES_TO_ADD = flags.DEFINE_list(
    'kernel_modules_to_add', [], 'Kernel modules to add to Linux VMs'
)
_KERNEL_MODULES_TO_REMOVE = flags.DEFINE_list(
    'kernel_modules_to_remove', [], 'Kernel modules to remove from Linux VMs'
)

_DISABLE_CSTATE_BY_NAME_AND_DEEPER = flags.DEFINE_string(
    'disable_cstate_by_name_and_deeper',
    None,
    'When specified, cstates that either match the given string or lower are'
    ' disabled. For instance, if C1E is specified for a VM running the'
    ' intel_idle driver, then C1E and C6 states would all be disabled, but C1'
    ' will remain enabled.',
)

_ENABLE_NVME_INTERRUPT_COALEASING = flags.DEFINE_bool(
    'enable_nvme_interrupt_coaleasing',
    False,
    'Attempt to enable interrupt coaleasing for all the NVMe disks on this VM. '
    'Currently only implemented for local disks. '
    'Depending on the Guest, this command may or may not actually '
    'modify the interrupt coaleasing behavior.',
)


# RHEL package managers
YUM = 'yum'
DNF = 'dnf'

RETRYABLE_SSH_RETCODE = 255

# Using root logger removes one function call logging.info otherwise adds to
# the stack level. Can remove after python 11; see:
# https://bugs.python.org/issue45171
logger = logging.getLogger()


class CpuVulnerabilities:
  """The 3 different vulnerability statuses from vm.cpu_vulernabilities.

  Example input:
    /sys/devices/system/cpu/vulnerabilities/itlb_multihit:KVM: Vulnerable
  Is put into vulnerability with a key of "itlb_multihit" and value "KVM"

  Unparsed lines are put into the unknown dict.
  """

  def __init__(self):
    self.mitigations: Dict[str, str] = {}
    self.vulnerabilities: Dict[str, str] = {}
    self.notaffecteds: Set[str] = set()
    self.unknowns: Dict[str, str] = {}

  def AddLine(self, full_line: str) -> None:
    """Parses a line of output from the cpu/vulnerabilities/* files."""
    if not full_line:
      return
    file_path, line = full_line.split(':', 1)
    file_name = posixpath.basename(file_path)
    if self._AddMitigation(file_name, line):
      return
    if self._AddVulnerability(file_name, line):
      return
    if self._AddNotAffected(file_name, line):
      return
    self.unknowns[file_name] = line

  def _AddMitigation(self, file_name, line):
    match = re.match('^Mitigation: (.*)', line) or re.match(
        '^([^:]+): Mitigation: (.*)$', line
    )
    if match:
      self.mitigations[file_name] = ':'.join(match.groups())
      return True

  def _AddVulnerability(self, file_name, line):
    match = (
        re.match('^Vulnerable: (.*)', line)
        or re.match('^Vulnerable$', line)
        or re.match('^([^:]+): Vulnerable$', line)
    )
    if match:
      self.vulnerabilities[file_name] = ':'.join(match.groups())
      return True

  def _AddNotAffected(self, file_name, line):
    match = re.match('^Not affected$', line)
    if match:
      self.notaffecteds.add(file_name)
      return True

  @property
  def asdict(self) -> Dict[str, str]:
    """Returns the parsed CPU vulnerabilities as a dict."""
    ret = {}
    if self.mitigations:
      ret['mitigations'] = ','.join(sorted(self.mitigations))
      for key, value in self.mitigations.items():
        ret[f'mitigation_{key}'] = value
    if self.vulnerabilities:
      ret['vulnerabilities'] = ','.join(sorted(self.vulnerabilities))
      for key, value in self.vulnerabilities.items():
        ret[f'vulnerability_{key}'] = value
    if self.unknowns:
      ret['unknowns'] = ','.join(self.unknowns)
      for key, value in self.unknowns.items():
        ret[f'unknown_{key}'] = value
    if self.notaffecteds:
      ret['notaffecteds'] = ','.join(sorted(self.notaffecteds))
    return ret


class KernelRelease(object):
  """Holds the contents of the linux kernel version returned from uname -r."""

  def __init__(self, uname: str):
    """KernelVersion Constructor.

    Args:
      uname: A string in the format of "uname -r" command
    """

    # example format would be: "4.5.0-96-generic"
    # or "3.10.0-514.26.2.el7.x86_64" for centos
    # major.minor.Rest
    # in this example, major = 4, minor = 5
    self.name = uname
    major_string, minor_string, _ = uname.split('.', 2)
    self.major = int(major_string)
    self.minor = int(minor_string)

  def AtLeast(self, major, minor):
    """Check If the kernel version meets a minimum bar.

    The kernel version needs to be at least as high as the major.minor
    specified in args.

    Args:
      major: The major number to test, as an integer
      minor: The minor number to test, as an integer

    Returns:
      True if the kernel version is at least as high as major.minor,
      False otherwise
    """
    if self.major < major:
      return False
    if self.major > major:
      return True
    return self.minor >= minor

  def __repr__(self) -> str:
    return self.name


class BaseLinuxMixin(virtual_machine.BaseOsMixin):
  """Class that holds Linux related VM methods and attributes."""

  # If multiple ssh calls are made in parallel using -t it will mess
  # the stty settings up and the terminal will become very hard to use.
  # Serializing calls to ssh with the -t option fixes the problem.
  _pseudo_tty_lock = threading.Lock()

  # TODO(user): Remove all uses of Python 2.
  PYTHON_2_PACKAGE = 'python2'

  # this command might change depending on the OS, but most linux distributions
  # can use the following command
  INIT_RAM_FS_CMD = 'sudo update-initramfs -u'

  # regex to get the network devices from "ip link show"
  _IP_LINK_RE_DEVICE_MTU = re.compile(
      r'^\d+: (?P<device_name>\S+):.*mtu (?P<mtu>\d+)'
  )
  # device prefixes to ignore from "ip link show"
  # TODO(spencerkim): Record ib device metadata.
  _IGNORE_NETWORK_DEVICE_PREFIXES = ('lo', 'docker', 'ib')

  def __init__(self, *args, **kwargs):
    super(BaseLinuxMixin, self).__init__(*args, **kwargs)
    # N.B. If you override ssh_port you must override remote_access_ports and
    # primary_remote_access_port.
    self.ssh_port = DEFAULT_SSH_PORT
    self.remote_access_ports = [self.ssh_port]
    self.primary_remote_access_port = self.ssh_port
    self.has_private_key = False
    self.ssh_external_time = None
    self.ssh_internal_time = None

    self._remote_command_script_upload_lock = threading.Lock()
    self._has_remote_command_script = False
    self._needs_reboot = False
    self._lscpu_cache = None
    self._partition_table = {}
    self._proccpu_cache = None
    self._smp_affinity_script = None
    self.name: str
    self._os_info: Optional[str] = None
    self._kernel_release: Optional[KernelRelease] = None
    self._cpu_arch: Optional[str] = None
    self._kernel_command_line: Optional[str] = None
    self._network_device_mtus = None

  def _Suspend(self):
    """Suspends a VM."""
    raise NotImplementedError()

  def _Resume(self):
    """Resumes a VM."""
    raise NotImplementedError()

  def _BeforeSuspend(self):
    pass

  def _CreateVmTmpDir(self):
    self.RemoteCommand('mkdir -p %s' % vm_util.VM_TMP_DIR)

  def _SetTransparentHugepages(self):
    """Sets transparent hugepages based on --enable_transparent_hugepages.

    If the flag is unset (None), this is a nop.
    """
    if FLAGS.enable_transparent_hugepages is None:
      return
    setting = 'always' if FLAGS.enable_transparent_hugepages else 'never'
    self.RemoteCommand(
        'echo %s | sudo tee /sys/kernel/mm/transparent_hugepage/enabled'
        % setting
    )
    self.os_metadata['transparent_hugepage'] = setting

  def _DisableCstates(self):
    """Disable cstates that either match or deeper than the given cstate."""
    cstate = _DISABLE_CSTATE_BY_NAME_AND_DEEPER.value
    if not cstate:
      return
    cstates = self._GetOrderedCstates()
    if not cstates:
      raise ValueError(
          'No cstates found, the system does not support disabling cstates'
      )
    logging.info('Available cstates listed in order: %s', cstates)
    if cstate not in cstates:
      raise ValueError(f'Requested cstate {cstate} is not present in {cstates}')
    num_cpus = self.num_cpus or self._GetNumCpus()
    start_index = cstates.index(cstate)
    disabled_cstates = []
    for index in range(start_index, len(cstates)):
      for cpu_id in range(num_cpus):
        config_path = (
            f'{_SYSFS_CPU_PATH}/cpu{cpu_id}/cpuidle/state{index}/disable'
        )
        self.RemoteCommand(f'echo 1 | sudo tee {config_path}')
      disabled_cstates.append(cstates[index])
    self.os_metadata['disabled_cstates'] = ','.join(disabled_cstates)

  def _GetOrderedCstates(self) -> Optional[list[str]]:
    """Returns the ordered cstates by querying the sysfs cpuidle path.

    The ordering is obtained by the alphabetical wildcard expansion.
    """
    query_paths = f'{_SYSFS_CPU_PATH}/cpu0/cpuidle/state*/name'
    if not self._RemoteFileExists(query_paths):
      return None
    out, _ = self.RemoteCommand(f'cat {query_paths}')
    return list(filter(None, out.split('\n')))

  def _SetupRobustCommand(self):
    """Sets up the RobustRemoteCommand tooling.

    This includes installing python3 and pushing scripts required by
    RobustRemoteCommand to this VM.  There is a check to skip if previously
    installed.
    """
    with self._remote_command_script_upload_lock:
      if not self._has_remote_command_script:
        # Python3 is needed for RobustRemoteCommands
        self.Install('python3')

        for f in (EXECUTE_COMMAND, WAIT_FOR_COMMAND):
          remote_path = os.path.join(vm_util.VM_TMP_DIR, os.path.basename(f))
          if os.path.basename(remote_path):
            self.RemoteCommand('sudo rm -f ' + remote_path)
          self.PushDataFile(f, remote_path)
        self._has_remote_command_script = True

  def RobustRemoteCommand(
      self,
      command: str,
      timeout: Optional[float] = None,
      ignore_failure: bool = False,
  ) -> Tuple[str, str]:
    """Runs a command on the VM in a more robust way than RemoteCommand.

    This is used for long-running commands that might experience network issues
    that would normally interrupt a RemoteCommand and fail to provide results.
    Executes a command via a pair of scripts on the VM:

    * EXECUTE_COMMAND, which runs 'command' in a nohupped background process.
    * WAIT_FOR_COMMAND, which first waits on confirmation that EXECUTE_COMMAND
      has acquired an exclusive lock on a file with the command's status. This
      is done by waiting for the existence of a file written by EXECUTE_COMMAND
      once it successfully acquires an exclusive lock. Once confirmed,
      WAIT_COMMAND waits to acquire the file lock held by EXECUTE_COMMAND until
      'command' completes, then returns with the stdout, stderr, and exit status
      of 'command'.

    Temporary SSH failures (where ssh returns a 255) while waiting for the
    command to complete will be tolerated and safely retried. However, if
    remote command actually returns 255, SSH will return 1 instead to bypass
    retry behavior.

    Args:
      command: The command to run.
      timeout: The timeout for the command in seconds.
      ignore_failure: Ignore any failure if set to true.

    Returns:
      A tuple of stdout, stderr from running the command.

    Raises:
      RemoteCommandError: If there was a problem establishing the connection, or
          the command fails.
    """
    self._SetupRobustCommand()
    logger.info(
        'Running RobustRemoteCommand on %s: %s',
        self.name,
        command,
        stacklevel=2,
    )

    execute_path = os.path.join(
        vm_util.VM_TMP_DIR, os.path.basename(EXECUTE_COMMAND)
    )
    wait_path = os.path.join(
        vm_util.VM_TMP_DIR, os.path.basename(WAIT_FOR_COMMAND)
    )

    uid = uuid.uuid4()
    file_base = os.path.join(vm_util.VM_TMP_DIR, 'cmd%s' % uid)
    wrapper_log = file_base + '.log'
    stdout_file = file_base + '.stdout'
    stderr_file = file_base + '.stderr'
    status_file = file_base + '.status'
    exclusive_file = file_base + '.exclusive'

    if not isinstance(command, str):
      command = ' '.join(command)

    start_command = ['nohup', 'python3', execute_path,
                     '--stdout', stdout_file,
                     '--stderr', stderr_file,
                     '--status', status_file,
                     '--exclusive', exclusive_file,
                     '--command', pipes.quote(command)]  # pyformat: disable
    if timeout:
      start_command.extend(['--timeout', str(timeout)])

    start_command = '%s 1> %s 2>&1 &' % (' '.join(start_command), wrapper_log)
    self.RemoteCommand(start_command, stack_level=2)

    def _WaitForCommand():
      wait_command = ['python3', wait_path,
                      '--status', status_file,
                      '--exclusive', exclusive_file]  # pyformat: disable
      stdout = ''
      while 'Command finished.' not in stdout:
        logging.info('Waiting for original cmd: %s', command, stacklevel=3)
        stdout, _ = self.RemoteCommand(
            ' '.join(wait_command),
            timeout=1800,
            should_pre_log=False,
            stack_level=3,
        )
      wait_command.extend([
          '--stdout', stdout_file,
          '--stderr', stderr_file,
          '--delete',
      ])  # pyformat: disable
      logging.info(
          'Finished waiting, printing stdout & stderr for cmd: %s',
          command,
          stacklevel=3,
      )
      return self.RemoteCommand(
          ' '.join(wait_command),
          ignore_failure=ignore_failure,
          should_pre_log=False,
          stack_level=3,
      )

    try:
      return _WaitForCommand()
    except errors.VirtualMachine.RemoteCommandError:
      # In case the error was with the wrapper script itself, print the log.
      stdout, _ = self.RemoteCommand('cat %s' % wrapper_log, stack_level=2)
      if stdout.strip():
        logging.warning(
            'Exception during RobustRemoteCommand. Wrapper script log:\n%s',
            stdout,
        )
      raise

  def SetupRemoteFirewall(self):
    """Sets up IP table configurations on the VM."""
    self.RemoteHostCommand('sudo iptables -A INPUT -j ACCEPT')
    self.RemoteHostCommand('sudo iptables -A OUTPUT -j ACCEPT')

  def SetupProxy(self):
    """Sets up proxy configuration variables for the cloud environment."""
    env_file = '/etc/environment'
    commands = []

    if FLAGS.http_proxy:
      commands.append(
          "echo 'http_proxy=%s' | sudo tee -a %s" % (FLAGS.http_proxy, env_file)
      )

    if FLAGS.https_proxy:
      commands.append(
          "echo 'https_proxy=%s' | sudo tee -a %s"
          % (FLAGS.https_proxy, env_file)
      )

    if FLAGS.ftp_proxy:
      commands.append(
          "echo 'ftp_proxy=%s' | sudo tee -a %s" % (FLAGS.ftp_proxy, env_file)
      )

    if commands:
      self.RemoteCommand(';'.join(commands))

  def SetupPackageManager(self):
    """Specific Linux flavors should override this."""
    pass

  def PrepareVMEnvironment(self):
    super(BaseLinuxMixin, self).PrepareVMEnvironment()
    self.SetupProxy()
    self._CreateVmTmpDir()
    self._SetTransparentHugepages()
    self._DisableCstates()
    if FLAGS.setup_remote_firewall:
      self.SetupRemoteFirewall()
    if self.install_packages:
      self._CreateInstallDir()
      if self.is_static:
        self.SnapshotPackages()
      # TODO(user): Only setup if necessary.
      # Call SetupPackageManager lazily from HasPackage/InstallPackages like
      # ShouldDownloadPreprovisionedData sets up object storage CLIs.
      self.SetupPackageManager()
    self.SetFiles()
    self.DoSysctls()
    self._DoAppendKernelCommandLine()
    self.ModifyKernelModules()
    self.DoConfigureNetworkForBBR()
    self.DoConfigureTCPWindow()
    self.UpdateEnvironmentPath()
    self._DisableCpus()
    self._RebootIfNecessary()
    self.BurnCpu()
    self.FillDisk()

  def _CreateInstallDir(self):
    self.RemoteCommand(
        ('sudo mkdir -p {0}; sudo chmod a+rwxt {0}').format(
            linux_packages.INSTALL_DIR
        )
    )

  # LinuxMixins do not implement _Start or _Stop
  def _Start(self):
    """Starts the VM."""
    raise NotImplementedError()

  def _Stop(self):
    """Stops the VM."""
    raise NotImplementedError()

  def SetFiles(self):
    """Apply --set_files to the VM."""

    for pair in FLAGS.set_files:
      path, value = pair.split('=')
      self.RemoteCommand('echo "%s" | sudo tee %s' % (value, path))

  def _DisableCpus(self):
    """Apply num_disable_cpus to the VM.

    Raises:
      ValueError: if num_disable_cpus is outside of (0 ... num_cpus-1)
                  inclusive
    """
    if not FLAGS.num_disable_cpus:
      return

    self.num_disable_cpus = FLAGS.num_disable_cpus

    if self.num_disable_cpus <= 0 or self.num_disable_cpus >= self.num_cpus:
      raise ValueError(
          'num_disable_cpus must be between 1 and '
          '(num_cpus - 1) inclusive.  '
          'num_disable_cpus: %i, num_cpus: %i'
          % (self.num_disable_cpus, self.num_cpus)
      )

    # We can't disable cpu 0, starting from the last cpu in /proc/cpuinfo.
    # On multiprocessor systems, we also attempt to disable cpus on each
    # physical processor based on "physical id" in order to keep a similar
    # number of cpus on each physical processor.
    # In addition, for each cpu we disable, we will look for cpu with same
    # "core id" in order to disable vcpu pairs.
    cpus = copy.deepcopy(self.CheckProcCpu().mappings)
    cpu_mapping = collections.defaultdict(list)
    for cpu, info in cpus.items():
      numa = info.get('physical id')
      cpu_mapping[int(numa)].append((cpu, int(info.get('core id'))))

    # Sort cpus based on 'core id' on each numa node
    for numa in cpu_mapping:
      cpu_mapping[numa] = sorted(
          cpu_mapping[numa], key=lambda cpu_info: (cpu_info[1], cpu_info[0])
      )

    def _GetNextCPUToDisable(num_disable_cpus):
      """Get the next CPU id to disable."""
      numa_nodes = list(cpu_mapping)
      while num_disable_cpus:
        for numa in sorted(numa_nodes, reverse=True):
          cpu_id, _ = cpu_mapping[numa].pop()
          num_disable_cpus -= 1
          yield cpu_id
          if not num_disable_cpus:
            break

    for cpu_id in _GetNextCPUToDisable(self.num_disable_cpus):
      self.RemoteCommand(
          f'sudo bash -c "echo 0 > /sys/devices/system/cpu/cpu{cpu_id}/online"'
      )
    self._proccpu_cache = None
    self._lscpu_cache = None

  def UpdateEnvironmentPath(self):
    """Specific Linux flavors should override this."""
    pass

  def FillDisk(self):
    """Fills the primary scratch disk with a zeros file."""
    if FLAGS.disk_fill_size:
      out_file = posixpath.join(self.scratch_disks[0].mount_point, 'fill_file')
      self.RobustRemoteCommand(
          'dd if=/dev/zero of={out_file} bs=1G count={fill_size}'.format(
              out_file=out_file, fill_size=FLAGS.disk_fill_size
          )
      )

  def _ApplySysctlPersistent(self, sysctl_params):
    """Apply "key=value" pairs to /etc/sysctl.conf and load via sysctl -p.

    These values should remain persistent across future reboots.

    Args:
      sysctl_params: dict - the keys and values to write
    """
    if not sysctl_params:
      return

    for key, value in sysctl_params.items():
      self.RemoteCommand(
          'sudo bash -c \'echo "%s=%s" >> /etc/sysctl.conf\'' % (key, value)
      )

    # See https://www.golinuxcloud.com/sysctl-reload-without-reboot/
    self.RemoteCommand('sudo sysctl -p')

  def ApplySysctlPersistent(self, sysctl_params, should_reboot=False):
    """Apply "key=value" pairs to /etc/sysctl.conf and load via sysctl -p.

    These values should remain persistent across future reboots.

    Args:
      sysctl_params: dict - the keys and values to write
      should_reboot: bool - whether to reboot after applying sysctl changes
    """
    self._ApplySysctlPersistent(sysctl_params)
    if should_reboot or FLAGS.reboot_after_changing_sysctl:
      self.Reboot()

  def DoSysctls(self):
    """Apply --sysctl to the VM.

    The Sysctl pairs are written persistently so that if a reboot
    occurs, the flags are not lost.
    """
    sysctl_params = {}
    for pair in FLAGS.sysctl:
      key, value = pair.split('=')
      sysctl_params[key] = value
    self._ApplySysctlPersistent(sysctl_params)

  def DoConfigureNetworkForBBR(self):
    """Apply --network_enable_BBR to the VM."""
    if not FLAGS.network_enable_BBR:
      return

    if not self.kernel_release.AtLeast(4, 9):
      raise flags.ValidationError(
          'BBR requires a linux image with kernel 4.9 or newer'
      )

    # if the current congestion control mechanism is already BBR
    # then nothing needs to be done (avoid unnecessary reboot)
    if self.TcpCongestionControl() == 'bbr':
      return

    self._ApplySysctlPersistent({
        'net.core.default_qdisc': 'fq',
        'net.ipv4.tcp_congestion_control': 'bbr',
    })

  def DoConfigureTCPWindow(self):
    """Change TCP window parameters in sysctl."""

    possible_tcp_flags = [
        FLAGS.tcp_max_receive_buffer,
        FLAGS.tcp_max_send_buffer,
        _TCP_MAX_NOTSENT_BYTES.value,
        FLAGS.rmem_max,
        FLAGS.wmem_max,
    ]
    # Return if none of these flags are set
    if all(x is None for x in possible_tcp_flags):
      return

    # Get current values from VM
    stdout, _ = self.RemoteCommand('cat /proc/sys/net/ipv4/tcp_rmem')
    rmem_values = stdout.split()
    stdout, _ = self.RemoteCommand('cat /proc/sys/net/ipv4/tcp_wmem')
    wmem_values = stdout.split()
    stdout, _ = self.RemoteCommand('cat /proc/sys/net/ipv4/tcp_notsent_lowat')
    notsent_lowat_values = stdout.split()
    stdout, _ = self.RemoteCommand('cat /proc/sys/net/core/rmem_max')
    rmem_max = int(stdout)
    stdout, _ = self.RemoteCommand('cat /proc/sys/net/core/wmem_max')
    wmem_max = int(stdout)

    # third number is max receive/send
    max_receive = rmem_values[2]
    max_send = wmem_values[2]
    max_not_sent = notsent_lowat_values[0]
    logging.info('notsent[0]: %s', notsent_lowat_values[0])
    # if flags are set, override current values from vm
    if FLAGS.tcp_max_receive_buffer:
      max_receive = FLAGS.tcp_max_receive_buffer
    if FLAGS.tcp_max_send_buffer:
      max_send = FLAGS.tcp_max_send_buffer
    if _TCP_MAX_NOTSENT_BYTES.value:
      max_not_sent = _TCP_MAX_NOTSENT_BYTES.value
    if FLAGS.rmem_max:
      rmem_max = FLAGS.rmem_max
    if FLAGS.wmem_max:
      wmem_max = FLAGS.wmem_max

    # Add values to metadata
    self.os_metadata['tcp_max_receive_buffer'] = max_receive
    self.os_metadata['tcp_max_send_buffer'] = max_send
    self.os_metadata['tcp_max_notsent_bytes'] = max_not_sent
    self.os_metadata['rmem_max'] = rmem_max
    self.os_metadata['wmem_max'] = wmem_max

    rmem_string = '{} {} {}'.format(rmem_values[0], rmem_values[1], max_receive)
    wmem_string = '{} {} {}'.format(wmem_values[0], wmem_values[1], max_send)
    logging.info('rmem_string: ' + rmem_string + ' wmem_string: ' + wmem_string)
    not_sent_string = '{}'.format(max_not_sent)

    self._ApplySysctlPersistent({
        'net.ipv4.tcp_rmem': rmem_string,
        'net.ipv4.tcp_wmem': wmem_string,
        'net.ipv4.tcp_notsent_lowat': not_sent_string,
        'net.core.rmem_max': rmem_max,
        'net.core.wmem_max': wmem_max,
    })

  def _RebootIfNecessary(self):
    """Will reboot the VM if self._needs_reboot has been set."""
    if self._needs_reboot:
      self.Reboot()
      self._needs_reboot = False

  def TcpCongestionControl(self):
    """Return the congestion control used for tcp."""
    try:
      resp, _ = self.RemoteCommand(
          'cat /proc/sys/net/ipv4/tcp_congestion_control'
      )
      return resp.rstrip('\n')
    except errors.VirtualMachine.RemoteCommandError:
      return 'unknown'

  def GetCPUVersion(self):
    """Get the CPU version of the VM.

    Refer to flag definition '--required_cpu_version' in virtual_machine.py for
    more details.

    Returns:
      guest_arch: str.
    """
    proccpu_results = self.CheckProcCpu(check_cache=False).GetValues()
    if 'vendor_id' in proccpu_results:
      vendor = proccpu_results.get('vendor_id', 'UnknownVendor')
      family = proccpu_results.get('cpu family', 'UnknownFamily')
      model = proccpu_results.get('model', 'UnknownModel')
      stepping = proccpu_results.get('stepping', 'UnknownStepping')
      guest_arch = f'{vendor}_{family}_{model}_{stepping}'
    else:
      implementer = proccpu_results.get('CPU implementer', 'UnknownVendor')
      arch = proccpu_results.get('CPU architecture', 'UnknownArchitecture')
      variant = proccpu_results.get('CPU variant', 'UnknownVariant')
      part = proccpu_results.get('CPU part', 'UnknownPart')
      guest_arch = f'{implementer}_{arch}_{variant}_{part}'
    return guest_arch

  def CheckLsCpu(self):
    """Returns a LsCpuResults from the host VM."""
    if not self._lscpu_cache:
      lscpu, _ = self.RemoteCommand('lscpu')
      self._lscpu_cache = LsCpuResults(lscpu)
    return self._lscpu_cache

  def CheckProcCpu(self, check_cache=True):
    """Returns a ProcCpuResults from the host VM."""
    if not self._proccpu_cache or not check_cache:
      proccpu, _ = self.RemoteCommand('cat /proc/cpuinfo')
      self._proccpu_cache = ProcCpuResults(proccpu)
    return self._proccpu_cache

  def GetOsInfo(self) -> str:
    """Returns information regarding OS type and version."""
    stdout, _ = self.RemoteCommand('grep PRETTY_NAME /etc/os-release')
    return regex_util.ExtractGroup(OS_PRETTY_NAME_REGEXP, stdout)

  @property
  def os_info(self) -> str:
    """Get distribution-specific information."""
    if not self._os_info: