Synnefo » snf-ganeti » ganeti-local

#

#

# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Xen hypervisors

"""

import logging

import string # pylint: disable=W0402

from cStringIO import StringIO

from ganeti import constants

from ganeti import errors

from ganeti import utils

from ganeti.hypervisor import hv_base

from ganeti import netutils

from ganeti import objects

from ganeti import pathutils

from ganeti import ssconf

XEND_CONFIG_FILE = utils.PathJoin(pathutils.XEN_CONFIG_DIR, "xend-config.sxp")

XL_CONFIG_FILE = utils.PathJoin(pathutils.XEN_CONFIG_DIR, "xen/xl.conf")

VIF_BRIDGE_SCRIPT = utils.PathJoin(pathutils.XEN_CONFIG_DIR,

                                   "scripts/vif-bridge")

_DOM0_NAME = "Domain-0"

_DISK_LETTERS = string.ascii_lowercase

_FILE_DRIVER_MAP = {

  constants.FD_LOOP: "file",

  constants.FD_BLKTAP: "tap:aio",

  constants.FD_BLKTAP2: "tap2:tapdisk:aio",

  }

def _CreateConfigCpus(cpu_mask):

  """Create a CPU config string for Xen's config file.

  """

  # Convert the string CPU mask to a list of list of int's

  cpu_list = utils.ParseMultiCpuMask(cpu_mask)

  if len(cpu_list) == 1:

    all_cpu_mapping = cpu_list[0]

    if all_cpu_mapping == constants.CPU_PINNING_OFF:

      # If CPU pinning has 1 entry that's "all", then remove the

      # parameter from the config file

      return None

    else:

      # If CPU pinning has one non-all entry, mapping all vCPUS (the entire

      # VM) to one physical CPU, using format 'cpu = "C"'

      return "cpu = \"%s\"" % ",".join(map(str, all_cpu_mapping))

  else:

    def _GetCPUMap(vcpu):

      if vcpu[0] == constants.CPU_PINNING_ALL_VAL:

        cpu_map = constants.CPU_PINNING_ALL_XEN

      else:

        cpu_map = ",".join(map(str, vcpu))

      return "\"%s\"" % cpu_map

    # build the result string in format 'cpus = [ "c", "c", "c" ]',

    # where each c is a physical CPU number, a range, a list, or any

    # combination

    return "cpus = [ %s ]" % ", ".join(map(_GetCPUMap, cpu_list))

def _RunXmList(fn, xmllist_errors):

  """Helper function for L{_GetXmList} to run "xm list".

  @type fn: callable

  @param fn: Function returning result of running C{xm list}

  @type xmllist_errors: list

  @param xmllist_errors: Error list

  @rtype: list

  """

  result = fn()

  if result.failed:

    logging.error("xm list failed (%s): %s", result.fail_reason,

                  result.output)

    xmllist_errors.append(result)

    raise utils.RetryAgain()

  # skip over the heading

  return result.stdout.splitlines()

def _ParseXmList(lines, include_node):

  """Parses the output of C{xm list}.

  @type lines: list

  @param lines: Output lines of C{xm list}

  @type include_node: boolean

  @param include_node: If True, return information for Dom0

  @return: list of tuple containing (name, id, memory, vcpus, state, time

    spent)

  """

  result = []

  # Iterate through all lines while ignoring header

  for line in lines[1:]:

    # The format of lines is:

    # Name      ID Mem(MiB) VCPUs State  Time(s)

    # Domain-0   0  3418     4 r-----    266.2

    data = line.split()

    if len(data) != 6:

      raise errors.HypervisorError("Can't parse output of xm list,"

                                   " line: %s" % line)

    try:

      data[1] = int(data[1])

      data[2] = int(data[2])

      data[3] = int(data[3])

      data[5] = float(data[5])

    except (TypeError, ValueError), err:

      raise errors.HypervisorError("Can't parse output of xm list,"

                                   " line: %s, error: %s" % (line, err))

    # skip the Domain-0 (optional)

    if include_node or data[0] != _DOM0_NAME:

      result.append(data)

  return result

def _GetXmList(fn, include_node, _timeout=5):

  """Return the list of running instances.

  See L{_RunXmList} and L{_ParseXmList} for parameter details.

  """

  xmllist_errors = []

  try:

    lines = utils.Retry(_RunXmList, (0.3, 1.5, 1.0), _timeout,

                        args=(fn, xmllist_errors))

  except utils.RetryTimeout:

    if xmllist_errors:

      xmlist_result = xmllist_errors.pop()

      errmsg = ("xm list failed, timeout exceeded (%s): %s" %

                (xmlist_result.fail_reason, xmlist_result.output))

    else:

      errmsg = "xm list failed"

    raise errors.HypervisorError(errmsg)

  return _ParseXmList(lines, include_node)

def _IsInstanceRunning(instance_info):

  return instance_info == "r-----" \

      or instance_info == "-b----"

def _IsInstanceShutdown(instance_info):

  return instance_info == "---s--"

def _ParseNodeInfo(info):

  """Return information about the node.

  @return: a dict with the following keys (memory values in MiB):

        - memory_total: the total memory size on the node

        - memory_free: the available memory on the node for instances

        - nr_cpus: total number of CPUs

        - nr_nodes: in a NUMA system, the number of domains

        - nr_sockets: the number of physical CPU sockets in the node

        - hv_version: the hypervisor version in the form (major, minor)

  """

  result = {}

  cores_per_socket = threads_per_core = nr_cpus = None

  xen_major, xen_minor = None, None

  memory_total = None

  memory_free = None

  for line in info.splitlines():

    fields = line.split(":", 1)

    if len(fields) < 2:

      continue

    (key, val) = map(lambda s: s.strip(), fields)

    # Note: in Xen 3, memory has changed to total_memory

    if key in ("memory", "total_memory"):

      memory_total = int(val)

    elif key == "free_memory":

      memory_free = int(val)

    elif key == "nr_cpus":

      nr_cpus = result["cpu_total"] = int(val)

    elif key == "nr_nodes":

      result["cpu_nodes"] = int(val)

    elif key == "cores_per_socket":

      cores_per_socket = int(val)

    elif key == "threads_per_core":

      threads_per_core = int(val)

    elif key == "xen_major":

      xen_major = int(val)

    elif key == "xen_minor":

      xen_minor = int(val)

  if None not in [cores_per_socket, threads_per_core, nr_cpus]:

    result["cpu_sockets"] = nr_cpus / (cores_per_socket * threads_per_core)

  if memory_free is not None:

    result["memory_free"] = memory_free

  if memory_total is not None:

    result["memory_total"] = memory_total

  if not (xen_major is None or xen_minor is None):

    result[constants.HV_NODEINFO_KEY_VERSION] = (xen_major, xen_minor)

  return result

def _MergeInstanceInfo(info, fn):

  """Updates node information from L{_ParseNodeInfo} with instance info.

  @type info: dict

  @param info: Result from L{_ParseNodeInfo}

  @type fn: callable

  @param fn: Function returning result of running C{xm list}

  @rtype: dict

  """

  total_instmem = 0

  for (name, _, mem, vcpus, _, _) in fn(True):

    if name == _DOM0_NAME:

      info["memory_dom0"] = mem

      info["dom0_cpus"] = vcpus

    # Include Dom0 in total memory usage

    total_instmem += mem

  memory_free = info.get("memory_free")

  memory_total = info.get("memory_total")

  # Calculate memory used by hypervisor

  if None not in [memory_total, memory_free, total_instmem]:

    info["memory_hv"] = memory_total - memory_free - total_instmem

  return info

def _GetNodeInfo(info, fn):

  """Combines L{_MergeInstanceInfo} and L{_ParseNodeInfo}.

  """

  return _MergeInstanceInfo(_ParseNodeInfo(info), fn)

def _GetConfigFileDiskData(block_devices, blockdev_prefix,

                           _letters=_DISK_LETTERS):

  """Get disk directives for Xen config file.

  This method builds the xen config disk directive according to the

  given disk_template and block_devices.

  @param block_devices: list of tuples (cfdev, rldev):

      - cfdev: dict containing ganeti config disk part

      - rldev: ganeti.bdev.BlockDev object

  @param blockdev_prefix: a string containing blockdevice prefix,

                          e.g. "sd" for /dev/sda

  @return: string containing disk directive for xen instance config file

  """

  if len(block_devices) > len(_letters):

    raise errors.HypervisorError("Too many disks")

  disk_data = []

  for sd_suffix, (cfdev, dev_path) in zip(_letters, block_devices):

    sd_name = blockdev_prefix + sd_suffix

    if cfdev.mode == constants.DISK_RDWR:

      mode = "w"

    else:

      mode = "r"

    if cfdev.dev_type == constants.LD_FILE:

      driver = _FILE_DRIVER_MAP[cfdev.physical_id[0]]

    else:

      driver = "phy"

    disk_data.append("'%s:%s,%s,%s'" % (driver, dev_path, sd_name, mode))

  return disk_data

class XenHypervisor(hv_base.BaseHypervisor):

  """Xen generic hypervisor interface

  This is the Xen base class used for both Xen PVM and HVM. It contains

  all the functionality that is identical for both.

  """

  CAN_MIGRATE = True

  REBOOT_RETRY_COUNT = 60

  REBOOT_RETRY_INTERVAL = 10

  ANCILLARY_FILES = [

    XEND_CONFIG_FILE,

    XL_CONFIG_FILE,

    VIF_BRIDGE_SCRIPT,

    ]

  ANCILLARY_FILES_OPT = [

    XL_CONFIG_FILE,

    ]

  def __init__(self, _cfgdir=None, _run_cmd_fn=None, _cmd=None):

    hv_base.BaseHypervisor.__init__(self)

    if _cfgdir is None:

      self._cfgdir = pathutils.XEN_CONFIG_DIR

    else:

      self._cfgdir = _cfgdir

    if _run_cmd_fn is None:

      self._run_cmd_fn = utils.RunCmd

    else:

      self._run_cmd_fn = _run_cmd_fn

    self._cmd = _cmd

  def _GetCommand(self):

    """Returns Xen command to use.

    """

    if self._cmd is None:

      # TODO: Make command a hypervisor parameter

      cmd = constants.XEN_CMD

    else:

      cmd = self._cmd

    if cmd not in constants.KNOWN_XEN_COMMANDS:

      raise errors.ProgrammerError("Unknown Xen command '%s'" % cmd)

    return cmd

  def _RunXen(self, args):

    """Wrapper around L{utils.process.RunCmd} to run Xen command.

    @see: L{utils.process.RunCmd}

    """

    cmd = [self._GetCommand()]

    cmd.extend(args)

    return self._run_cmd_fn(cmd)

  def _ConfigFileName(self, instance_name):

    """Get the config file name for an instance.

    @param instance_name: instance name

    @type instance_name: str

    @return: fully qualified path to instance config file

    @rtype: str

    """

    return utils.PathJoin(self._cfgdir, instance_name)

  @classmethod

  def _GetConfig(cls, instance, startup_memory, block_devices):

    """Build Xen configuration for an instance.

    """

    raise NotImplementedError

  def _WriteConfigFile(self, instance_name, data):

    """Write the Xen config file for the instance.

    This version of the function just writes the config file from static data.

    """

    # just in case it exists

    utils.RemoveFile(utils.PathJoin(self._cfgdir, "auto", instance_name))

    cfg_file = self._ConfigFileName(instance_name)

    try:

      utils.WriteFile(cfg_file, data=data)

    except EnvironmentError, err:

      raise errors.HypervisorError("Cannot write Xen instance configuration"

                                   " file %s: %s" % (cfg_file, err))

  def _ReadConfigFile(self, instance_name):

    """Returns the contents of the instance config file.

    """

    filename = self._ConfigFileName(instance_name)

    try:

      file_content = utils.ReadFile(filename)

    except EnvironmentError, err:

      raise errors.HypervisorError("Failed to load Xen config file: %s" % err)

    return file_content

  def _RemoveConfigFile(self, instance_name):

    """Remove the xen configuration file.

    """

    utils.RemoveFile(self._ConfigFileName(instance_name))

  def _StashConfigFile(self, instance_name):

    """Move the Xen config file to the log directory and return its new path.

    """

    old_filename = self._ConfigFileName(instance_name)

    base = ("%s-%s" %

            (instance_name, utils.TimestampForFilename()))

    new_filename = utils.PathJoin(pathutils.LOG_XEN_DIR, base)

    utils.RenameFile(old_filename, new_filename)

    return new_filename

  def _GetXmList(self, include_node):

    """Wrapper around module level L{_GetXmList}.

    """

    return _GetXmList(lambda: self._RunXen(["list"]), include_node)

  def ListInstances(self):

    """Get the list of running instances.

    """

    xm_list = self._GetXmList(False)

    names = [info[0] for info in xm_list]

    return names

  def GetInstanceInfo(self, instance_name):

    """Get instance properties.

    @param instance_name: the instance name

    @return: tuple (name, id, memory, vcpus, stat, times)

    """

    xm_list = self._GetXmList(instance_name == _DOM0_NAME)

    result = None

    for data in xm_list:

      if data[0] == instance_name:

        result = data

        break

    return result

  def GetAllInstancesInfo(self):

    """Get properties of all instances.

    @return: list of tuples (name, id, memory, vcpus, stat, times)

    """

    xm_list = self._GetXmList(False)

    return xm_list

  def _MakeConfigFile(self, instance, startup_memory, block_devices):

    """Gather configuration details and write to disk.

    See L{_GetConfig} for arguments.

    """

    buf = StringIO()

    buf.write("# Automatically generated by Ganeti. Do not edit!\n")

    buf.write("\n")

    buf.write(self._GetConfig(instance, startup_memory, block_devices))

    buf.write("\n")

    self._WriteConfigFile(instance.name, buf.getvalue())

  def StartInstance(self, instance, block_devices, startup_paused):

    """Start an instance.

    """

    startup_memory = self._InstanceStartupMemory(instance)

    self._MakeConfigFile(instance, startup_memory, block_devices)

    cmd = ["create"]

    if startup_paused:

      cmd.append("-p")

    cmd.append(self._ConfigFileName(instance.name))

    result = self._RunXen(cmd)

    if result.failed:

      # Move the Xen configuration file to the log directory to avoid

      # leaving a stale config file behind.

      stashed_config = self._StashConfigFile(instance.name)

      raise errors.HypervisorError("Failed to start instance %s: %s (%s). Moved"

                                   " config file to %s" %

                                   (instance.name, result.fail_reason,

                                    result.output, stashed_config))

  def StopInstance(self, instance, force=False, retry=False, name=None):

    """Stop an instance.

    """

    if name is None:

      name = instance.name

    return self._StopInstance(name, force)

  def _ShutdownInstance(self, name):

    """Shutdown an instance if the instance is running.

    @type name: string

    @param name: name of the instance to stop

    The '-w' flag waits for shutdown to complete which avoids the need

    to poll in the case where we want to destroy the domain

    immediately after shutdown.

    """

    instance_info = self.GetInstanceInfo(name)

    if instance_info is None or _IsInstanceShutdown(instance_info[4]):

      logging.info("Failed to shutdown instance %s, not running", name)

      return None

    return self._RunXen(["shutdown", "-w", name])

  def _DestroyInstance(self, name):

    """Destroy an instance if the instance if the instance exists.

    @type name: string

    @param name: name of the instance to destroy

    """

    instance_info = self.GetInstanceInfo(name)

    if instance_info is None:

      logging.info("Failed to destroy instance %s, does not exist", name)

      return None

    return self._RunXen(["destroy", name])

  def _StopInstance(self, name, force):

    """Stop an instance.

    @type name: string

    @param name: name of the instance to destroy

    @type force: boolean

    @param force: whether to do a "hard" stop (destroy)

    """

    if force:

      result = self._DestroyInstance(name)

    else:

      self._ShutdownInstance(name)

      result = self._DestroyInstance(name)

    if result is not None and result.failed and \

          self.GetInstanceInfo(name) is not None:

      raise errors.HypervisorError("Failed to stop instance %s: %s, %s" %

                                   (name, result.fail_reason, result.output))

    # Remove configuration file if stopping/starting instance was successful

    self._RemoveConfigFile(name)

  def RebootInstance(self, instance):

    """Reboot an instance.

    """

    ini_info = self.GetInstanceInfo(instance.name)

    if ini_info is None:

      raise errors.HypervisorError("Failed to reboot instance %s,"

                                   " not running" % instance.name)

    result = self._RunXen(["reboot", instance.name])

    if result.failed:

      raise errors.HypervisorError("Failed to reboot instance %s: %s, %s" %

                                   (instance.name, result.fail_reason,

                                    result.output))

    def _CheckInstance():

      new_info = self.GetInstanceInfo(instance.name)

      # check if the domain ID has changed or the run time has decreased

      if (new_info is not None and

          (new_info[1] != ini_info[1] or new_info[5] < ini_info[5])):

        return

      raise utils.RetryAgain()

    try:

      utils.Retry(_CheckInstance, self.REBOOT_RETRY_INTERVAL,

                  self.REBOOT_RETRY_INTERVAL * self.REBOOT_RETRY_COUNT)

    except utils.RetryTimeout:

      raise errors.HypervisorError("Failed to reboot instance %s: instance"

                                   " did not reboot in the expected interval" %

                                   (instance.name, ))

  def BalloonInstanceMemory(self, instance, mem):

    """Balloon an instance memory to a certain value.

    @type instance: L{objects.Instance}

    @param instance: instance to be accepted

    @type mem: int

    @param mem: actual memory size to use for instance runtime

    """

    result = self._RunXen(["mem-set", instance.name, mem])

    if result.failed:

      raise errors.HypervisorError("Failed to balloon instance %s: %s (%s)" %

                                   (instance.name, result.fail_reason,

                                    result.output))

    # Update configuration file

    cmd = ["sed", "-ie", "s/^memory.*$/memory = %s/" % mem]

    cmd.append(self._ConfigFileName(instance.name))

    result = utils.RunCmd(cmd)

    if result.failed:

      raise errors.HypervisorError("Failed to update memory for %s: %s (%s)" %

                                   (instance.name, result.fail_reason,

                                    result.output))

  def GetNodeInfo(self):

    """Return information about the node.

    @see: L{_GetNodeInfo} and L{_ParseNodeInfo}

    """

    result = self._RunXen(["info"])

    if result.failed:

      logging.error("Can't run 'xm info' (%s): %s", result.fail_reason,

                    result.output)

      return None

    return _GetNodeInfo(result.stdout, self._GetXmList)

  @classmethod

  def GetInstanceConsole(cls, instance, hvparams, beparams):

    """Return a command for connecting to the console of an instance.

    """

    return objects.InstanceConsole(instance=instance.name,

                                   kind=constants.CONS_SSH,

                                   host=instance.primary_node,

                                   user=constants.SSH_CONSOLE_USER,

                                   command=[pathutils.XEN_CONSOLE_WRAPPER,

                                            constants.XEN_CMD, instance.name])

  def Verify(self):

    """Verify the hypervisor.

    For Xen, this verifies that the xend process is running.

    @return: Problem description if something is wrong, C{None} otherwise

    """

    result = self._RunXen(["info"])

    if result.failed:

      return "'xm info' failed: %s, %s" % (result.fail_reason, result.output)

    return None

  def MigrationInfo(self, instance):

    """Get instance information to perform a migration.

    @type instance: L{objects.Instance}

    @param instance: instance to be migrated

    @rtype: string

    @return: content of the xen config file

    """

    return self._ReadConfigFile(instance.name)

  def AcceptInstance(self, instance, info, target):

    """Prepare to accept an instance.

    @type instance: L{objects.Instance}

    @param instance: instance to be accepted

    @type info: string

    @param info: content of the xen config file on the source node

    @type target: string

    @param target: target host (usually ip), on this node

    """

    pass

  def FinalizeMigrationDst(self, instance, info, success):

    """Finalize an instance migration.

    After a successful migration we write the xen config file.

    We do nothing on a failure, as we did not change anything at accept time.

    @type instance: L{objects.Instance}

    @param instance: instance whose migration is being finalized

    @type info: string

    @param info: content of the xen config file on the source node

    @type success: boolean

    @param success: whether the migration was a success or a failure

    """

    if success:

      self._WriteConfigFile(instance.name, info)

  def MigrateInstance(self, instance, target, live):

    """Migrate an instance to a target node.

    The migration will not be attempted if the instance is not

    currently running.

    @type instance: L{objects.Instance}

    @param instance: the instance to be migrated

    @type target: string

    @param target: ip address of the target node

    @type live: boolean

    @param live: perform a live migration

    """

    port = instance.hvparams[constants.HV_MIGRATION_PORT]

    # TODO: Pass cluster name via RPC

    cluster_name = ssconf.SimpleStore().GetClusterName()

    return self._MigrateInstance(cluster_name, instance.name, target, port,

                                 live)

  def _MigrateInstance(self, cluster_name, instance_name, target, port, live,

                       _ping_fn=netutils.TcpPing):

    """Migrate an instance to a target node.

    @see: L{MigrateInstance} for details

    """

    if self.GetInstanceInfo(instance_name) is None:

      raise errors.HypervisorError("Instance not running, cannot migrate")

    cmd = self._GetCommand()