Initial copy of RAPI filebase to the trunk
[ganeti-local] / lib / backend.py
index ac84801..382eec5 100644 (file)
@@ -26,11 +26,11 @@ import os
 import os.path
 import shutil
 import time
-import tempfile
 import stat
 import errno
 import re
 import subprocess
+import random
 
 from ganeti import logger
 from ganeti import errors
@@ -43,6 +43,10 @@ from ganeti import objects
 from ganeti import ssconf
 
 
+def _GetSshRunner():
+  return ssh.SshRunner()
+
+
 def StartMaster():
   """Activate local node as master node.
 
@@ -64,9 +68,7 @@ def StartMaster():
 def StopMaster():
   """Deactivate this node as master.
 
-  This does two things:
-    - run the master stop script
-    - remove link to master cron script.
+  This runs the master stop script.
 
   """
   result = utils.RunCmd([constants.MASTER_SCRIPT, "-d", "stop"])
@@ -122,7 +124,6 @@ def LeaveCluster():
       if os.path.isfile(full_name) and not os.path.islink(full_name):
         utils.RemoveFile(full_name)
 
-
   try:
     priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
   except errors.OpExecError, err:
@@ -138,6 +139,9 @@ def LeaveCluster():
   utils.RemoveFile(priv_key)
   utils.RemoveFile(pub_key)
 
+  # Return a reassuring string to the caller, and quit
+  raise errors.QuitGanetiException(False, 'Shutdown scheduled')
+
 
 def GetNodeInfo(vgname):
   """Gives back a hash with different informations about the node.
@@ -200,10 +204,38 @@ def VerifyNode(what):
 
   if 'nodelist' in what:
     result['nodelist'] = {}
+    random.shuffle(what['nodelist'])
     for node in what['nodelist']:
-      success, message = ssh.VerifyNodeHostname(node)
+      success, message = _GetSshRunner().VerifyNodeHostname(node)
       if not success:
         result['nodelist'][node] = message
+  if 'node-net-test' in what:
+    result['node-net-test'] = {}
+    my_name = utils.HostInfo().name
+    my_pip = my_sip = None
+    for name, pip, sip in what['node-net-test']:
+      if name == my_name:
+        my_pip = pip
+        my_sip = sip
+        break
+    if not my_pip:
+      result['node-net-test'][my_name] = ("Can't find my own"
+                                          " primary/secondary IP"
+                                          " in the node list")
+    else:
+      port = ssconf.SimpleStore().GetNodeDaemonPort()
+      for name, pip, sip in what['node-net-test']:
+        fail = []
+        if not utils.TcpPing(pip, port, source=my_pip):
+          fail.append("primary")
+        if sip != pip:
+          if not utils.TcpPing(sip, port, source=my_sip):
+            fail.append("secondary")
+        if fail:
+          result['node-net-test'][name] = ("failure using the %s"
+                                           " interface(s)" %
+                                           " and ".join(fail))
+
   return result
 
 
@@ -211,19 +243,31 @@ def GetVolumeList(vg_name):
   """Compute list of logical volumes and their size.
 
   Returns:
-    dictionary of all partions (key) with their size:
-    test1: 20.06MiB
+    dictionary of all partions (key) with their size (in MiB), inactive
+    and online status:
+    {'test1': ('20.06', True, True)}
 
   """
-  result = utils.RunCmd(["lvs", "--noheadings", "--units=m",
-                         "-oname,size", vg_name])
+  lvs = {}
+  sep = '|'
+  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
+                         "--separator=%s" % sep,
+                         "-olv_name,lv_size,lv_attr", vg_name])
   if result.failed:
     logger.Error("Failed to list logical volumes, lvs output: %s" %
                  result.output)
-    return {}
+    return result.output
 
-  lvlist = [line.split() for line in result.output.splitlines()]
-  return dict(lvlist)
+  for line in result.stdout.splitlines():
+    line = line.strip().rstrip(sep)
+    name, size, attr = line.split(sep)
+    if len(attr) != 6:
+      attr = '------'
+    inactive = attr[4] == '-'
+    online = attr[5] == 'o'
+    lvs[name] = (size, inactive, online)
+
+  return lvs
 
 
 def ListVolumeGroups():
@@ -262,7 +306,7 @@ def NodeVolumes():
       'vg': line[3].strip(),
     }
 
-  return [map_line(line.split('|')) for line in result.output.splitlines()]
+  return [map_line(line.split('|')) for line in result.stdout.splitlines()]
 
 
 def BridgesExist(bridges_list):
@@ -400,11 +444,10 @@ def AddOSToInstance(instance, os_disk, swap_disk):
                                 logfile)
 
   result = utils.RunCmd(command)
-
   if result.failed:
-    logger.Error("os create command '%s' returned error: %s"
+    logger.Error("os create command '%s' returned error: %s, logfile: %s,"
                  " output: %s" %
-                 (command, result.fail_reason, result.output))
+                 (command, result.fail_reason, logfile, result.output))
     return False
 
   return True
@@ -481,20 +524,32 @@ def _GetVGInfo(vg_name):
     vg_free is the free size of the volume group in MiB
     pv_count are the number of physical disks in that vg
 
+  If an error occurs during gathering of data, we return the same dict
+  with keys all set to None.
+
   """
+  retdic = dict.fromkeys(["vg_size", "vg_free", "pv_count"])
+
   retval = utils.RunCmd(["vgs", "-ovg_size,vg_free,pv_count", "--noheadings",
                          "--nosuffix", "--units=m", "--separator=:", vg_name])
 
   if retval.failed:
     errmsg = "volume group %s not present" % vg_name
     logger.Error(errmsg)
-    raise errors.LVMError(errmsg)
-  valarr = retval.stdout.strip().split(':')
-  retdic = {
-    "vg_size": int(round(float(valarr[0]), 0)),
-    "vg_free": int(round(float(valarr[1]), 0)),
-    "pv_count": int(valarr[2]),
-    }
+    return retdic
+  valarr = retval.stdout.strip().rstrip(':').split(':')
+  if len(valarr) == 3:
+    try:
+      retdic = {
+        "vg_size": int(round(float(valarr[0]), 0)),
+        "vg_free": int(round(float(valarr[1]), 0)),
+        "pv_count": int(valarr[2]),
+        }
+    except ValueError, err:
+      logger.Error("Fail to parse vgs output: %s" % str(err))
+  else:
+    logger.Error("vgs output has the wrong number of fields (expected"
+                 " three): %s" % str(valarr))
   return retdic
 
 
@@ -620,14 +675,30 @@ def RebootInstance(instance, reboot_type, extra_args):
   return True
 
 
+def MigrateInstance(instance, target, live):
+  """Migrates an instance to another node.
+
+  """
+  hyper = hypervisor.GetHypervisor()
+
+  try:
+    hyper.MigrateInstance(instance, target, live)
+  except errors.HypervisorError, err:
+    msg = "Failed to migrate instance: %s" % str(err)
+    logger.Error(msg)
+    return (False, msg)
+  return (True, "Migration successfull")
+
+
 def CreateBlockDevice(disk, size, owner, on_primary, info):
   """Creates a block device for an instance.
 
   Args:
-   bdev: a ganeti.objects.Disk object
-   size: the size of the physical underlying devices
-   do_open: if the device should be `Assemble()`-d and
-            `Open()`-ed after creation
+   disk: a ganeti.objects.Disk object
+   size: the size of the physical underlying device
+   owner: a string with the name of the instance
+   on_primary: a boolean indicating if it is the primary node or not
+   info: string that will be sent to the physical device creation
 
   Returns:
     the new unique_id of the device (this can sometime be
@@ -643,8 +714,6 @@ def CreateBlockDevice(disk, size, owner, on_primary, info):
         # we need the children open in case the device itself has to
         # be assembled
         crdev.Open()
-      else:
-        crdev.Close()
       clist.append(crdev)
   try:
     device = bdev.FindDevice(disk.dev_type, disk.physical_id, clist)
@@ -661,9 +730,10 @@ def CreateBlockDevice(disk, size, owner, on_primary, info):
                      (disk, size))
   if on_primary or disk.AssembleOnSecondary():
     if not device.Assemble():
-      raise errors.BlockDeviceError("Can't assemble device after creation,"
-                                    " very unusual event - check the node"
-                                    " daemon logs")
+      errorstring = "Can't assemble device after creation"
+      logger.Error(errorstring)
+      raise errors.BlockDeviceError("%s, very unusual event - check the node"
+                                    " daemon logs" % errorstring)
     device.SetSyncSpeed(constants.SYNC_SPEED)
     if on_primary or disk.OpenOnSecondary():
       device.Open(force=True)
@@ -731,7 +801,7 @@ def _RecursiveAssembleBD(disk, owner, as_primary):
       try:
         cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
       except errors.BlockDeviceError, err:
-        if children.count(None) > mcn:
+        if children.count(None) >= mcn:
           raise
         cdev = None
         logger.Debug("Error in child activation: %s" % str(err))
@@ -743,8 +813,6 @@ def _RecursiveAssembleBD(disk, owner, as_primary):
     result = r_dev
     if as_primary or disk.OpenOnSecondary():
       r_dev.Open()
-    else:
-      r_dev.Close()
     DevCacheManager.UpdateCache(r_dev.dev_path, owner,
                                 as_primary, disk.iv_name)
 
@@ -910,28 +978,20 @@ def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
                  file_name)
     return False
 
-  allowed_files = [constants.CLUSTER_CONF_FILE, "/etc/hosts",
-                   constants.SSH_KNOWN_HOSTS_FILE]
+  allowed_files = [
+    constants.CLUSTER_CONF_FILE,
+    constants.ETC_HOSTS,
+    constants.SSH_KNOWN_HOSTS_FILE,
+    constants.VNC_PASSWORD_FILE,
+    ]
   allowed_files.extend(ssconf.SimpleStore().GetFileList())
   if file_name not in allowed_files:
     logger.Error("Filename passed to UploadFile not in allowed"
                  " upload targets: '%s'" % file_name)
     return False
 
-  dir_name, small_name = os.path.split(file_name)
-  fd, new_name = tempfile.mkstemp('.new', small_name, dir_name)
-  # here we need to make sure we remove the temp file, if any error
-  # leaves it in place
-  try:
-    os.chown(new_name, uid, gid)
-    os.chmod(new_name, mode)
-    os.write(fd, data)
-    os.fsync(fd)
-    os.utime(new_name, (atime, mtime))
-    os.rename(new_name, file_name)
-  finally:
-    os.close(fd)
-    utils.RemoveFile(new_name)
+  utils.WriteFile(file_name, data=data, mode=mode, uid=uid, gid=gid,
+                  atime=atime, mtime=mtime)
   return True
 
 
@@ -950,28 +1010,6 @@ def _ErrnoOrStr(err):
   return detail
 
 
-def _OSSearch(name, search_path=None):
-  """Search for OSes with the given name in the search_path.
-
-  Args:
-    name: The name of the OS to look for
-    search_path: List of dirs to search (defaults to constants.OS_SEARCH_PATH)
-
-  Returns:
-    The base_dir the OS resides in
-
-  """
-  if search_path is None:
-    search_path = constants.OS_SEARCH_PATH
-
-  for dir_name in search_path:
-    t_os_dir = os.path.sep.join([dir_name, name])
-    if os.path.isdir(t_os_dir):
-      return dir_name
-
-  return None
-
-
 def _OSOndiskVersion(name, os_dir):
   """Compute and return the API version of a given OS.
 
@@ -1061,12 +1099,12 @@ def OSFromDisk(name, base_dir=None):
   """
 
   if base_dir is None:
-    base_dir = _OSSearch(name)
-
-  if base_dir is None:
-    raise errors.InvalidOS(name, None, "OS dir not found in search path")
+    os_dir = utils.FindFile(name, constants.OS_SEARCH_PATH, os.path.isdir)
+    if os_dir is None:
+      raise errors.InvalidOS(name, None, "OS dir not found in search path")
+  else:
+    os_dir = os.path.sep.join([base_dir, name])
 
-  os_dir = os.path.sep.join([base_dir, name])
   api_version = _OSOndiskVersion(name, os_dir)
 
   if api_version != constants.OS_API_VERSION:
@@ -1103,6 +1141,32 @@ def OSFromDisk(name, base_dir=None):
                     api_version=api_version)
 
 
+def GrowBlockDevice(disk, amount):
+  """Grow a stack of block devices.
+
+  This function is called recursively, with the childrens being the
+  first one resize.
+
+  Args:
+    disk: the disk to be grown
+
+  Returns: a tuple of (status, result), with:
+    status: the result (true/false) of the operation
+    result: the error message if the operation failed, otherwise not used
+
+  """
+  r_dev = _RecursiveFindBD(disk)
+  if r_dev is None:
+    return False, "Cannot find block device %s" % (disk,)
+
+  try:
+    r_dev.Grow(amount)
+  except errors.BlockDeviceError, err:
+    return False, str(err)
+
+  return True, None
+
+
 def SnapshotBlockDevice(disk):
   """Create a snapshot copy of a block device.
 
@@ -1135,7 +1199,7 @@ def SnapshotBlockDevice(disk):
       return None
   else:
     raise errors.ProgrammerError("Cannot snapshot non-lvm block device"
-                                 "'%s' of type '%s'" %
+                                 " '%s' of type '%s'" %
                                  (disk.unique_id, disk.dev_type))
 
 
@@ -1180,9 +1244,8 @@ def ExportSnapshot(disk, dest_node, instance):
 
   destcmd = utils.BuildShellCmd("mkdir -p %s && cat > %s/%s",
                                 destdir, destdir, destfile)
-  remotecmd = ssh.BuildSSHCmd(dest_node, constants.GANETI_RUNAS, destcmd)
-
-
+  remotecmd = _GetSshRunner().BuildCmd(dest_node, constants.GANETI_RUNAS,
+                                       destcmd)
 
   # all commands have been checked, so we're safe to combine them
   command = '|'.join([expcmd, comprcmd, utils.ShellQuoteArgs(remotecmd)])
@@ -1226,13 +1289,17 @@ def FinalizeExport(instance, snap_disks):
   config.set(constants.INISECT_INS, 'memory', '%d' % instance.memory)
   config.set(constants.INISECT_INS, 'vcpus', '%d' % instance.vcpus)
   config.set(constants.INISECT_INS, 'disk_template', instance.disk_template)
+
+  nic_count = 0
   for nic_count, nic in enumerate(instance.nics):
     config.set(constants.INISECT_INS, 'nic%d_mac' %
                nic_count, '%s' % nic.mac)
     config.set(constants.INISECT_INS, 'nic%d_ip' % nic_count, '%s' % nic.ip)
+    config.set(constants.INISECT_INS, 'nic%d_bridge' % nic_count, '%s' % nic.bridge)
   # TODO: redundant: on load can read nics until it doesn't exist
   config.set(constants.INISECT_INS, 'nic_count' , '%d' % nic_count)
 
+  disk_count = 0
   for disk_count, disk in enumerate(snap_disks):
     config.set(constants.INISECT_INS, 'disk%d_ivname' % disk_count,
                ('%s' % disk.iv_name))
@@ -1322,7 +1389,8 @@ def ImportOSIntoInstance(instance, os_disk, swap_disk, src_node, src_image):
     os.mkdir(constants.LOG_OS_DIR, 0750)
 
   destcmd = utils.BuildShellCmd('cat %s', src_image)
-  remotecmd = ssh.BuildSSHCmd(src_node, constants.GANETI_RUNAS, destcmd)
+  remotecmd = _GetSshRunner().BuildCmd(src_node, constants.GANETI_RUNAS,
+                                       destcmd)
 
   comprcmd = "gunzip"
   impcmd = utils.BuildShellCmd("(cd %s; %s -i %s -b %s -s %s &>%s)",
@@ -1404,6 +1472,155 @@ def RenameBlockDevices(devlist):
   return result
 
 
+def _TransformFileStorageDir(file_storage_dir):
+  """Checks whether given file_storage_dir is valid.
+
+  Checks wheter the given file_storage_dir is within the cluster-wide
+  default file_storage_dir stored in SimpleStore. Only paths under that
+  directory are allowed.
+
+  Args:
+    file_storage_dir: string with path
+
+  Returns:
+    normalized file_storage_dir (string) if valid, None otherwise
+
+  """
+  file_storage_dir = os.path.normpath(file_storage_dir)
+  base_file_storage_dir = ssconf.SimpleStore().GetFileStorageDir()
+  if (not os.path.commonprefix([file_storage_dir, base_file_storage_dir]) ==
+      base_file_storage_dir):
+    logger.Error("file storage directory '%s' is not under base file"
+                 " storage directory '%s'" %
+                 (file_storage_dir, base_file_storage_dir))
+    return None
+  return file_storage_dir
+
+
+def CreateFileStorageDir(file_storage_dir):
+  """Create file storage directory.
+
+  Args:
+    file_storage_dir: string containing the path
+
+  Returns:
+    tuple with first element a boolean indicating wheter dir
+    creation was successful or not
+
+  """
+  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
+  result = True,
+  if not file_storage_dir:
+    result = False,
+  else:
+    if os.path.exists(file_storage_dir):
+      if not os.path.isdir(file_storage_dir):
+        logger.Error("'%s' is not a directory" % file_storage_dir)
+        result = False,
+    else:
+      try:
+        os.makedirs(file_storage_dir, 0750)
+      except OSError, err:
+        logger.Error("Cannot create file storage directory '%s': %s" %
+                     (file_storage_dir, err))
+        result = False,
+  return result
+
+
+def RemoveFileStorageDir(file_storage_dir):
+  """Remove file storage directory.
+
+  Remove it only if it's empty. If not log an error and return.
+
+  Args:
+    file_storage_dir: string containing the path
+
+  Returns:
+    tuple with first element a boolean indicating wheter dir
+    removal was successful or not
+
+  """
+  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
+  result = True,
+  if not file_storage_dir:
+    result = False,
+  else:
+    if os.path.exists(file_storage_dir):
+      if not os.path.isdir(file_storage_dir):
+        logger.Error("'%s' is not a directory" % file_storage_dir)
+        result = False,
+      # deletes dir only if empty, otherwise we want to return False
+      try:
+        os.rmdir(file_storage_dir)
+      except OSError, err:
+        logger.Error("Cannot remove file storage directory '%s': %s" %
+                     (file_storage_dir, err))
+        result = False,
+  return result
+
+
+def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
+  """Rename the file storage directory.
+
+  Args:
+    old_file_storage_dir: string containing the old path
+    new_file_storage_dir: string containing the new path
+
+  Returns:
+    tuple with first element a boolean indicating wheter dir
+    rename was successful or not
+
+  """
+  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
+  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
+  result = True,
+  if not old_file_storage_dir or not new_file_storage_dir:
+    result = False,
+  else:
+    if not os.path.exists(new_file_storage_dir):
+      if os.path.isdir(old_file_storage_dir):
+        try:
+          os.rename(old_file_storage_dir, new_file_storage_dir)
+        except OSError, err:
+          logger.Error("Cannot rename '%s' to '%s': %s"
+                       % (old_file_storage_dir, new_file_storage_dir, err))
+          result =  False,
+      else:
+        logger.Error("'%s' is not a directory" % old_file_storage_dir)
+        result = False,
+    else:
+      if os.path.exists(old_file_storage_dir):
+        logger.Error("Cannot rename '%s' to '%s'. Both locations exist." %
+                     old_file_storage_dir, new_file_storage_dir)
+        result = False,
+  return result
+
+
+def CloseBlockDevices(disks):
+  """Closes the given block devices.
+
+  This means they will be switched to secondary mode (in case of DRBD).
+
+  """
+  bdevs = []
+  for cf in disks:
+    rd = _RecursiveFindBD(cf)
+    if rd is None:
+      return (False, "Can't find device %s" % cf)
+    bdevs.append(rd)
+
+  msg = []
+  for rd in bdevs:
+    try:
+      rd.Close()
+    except errors.BlockDeviceError, err:
+      msg.append(str(err))
+  if msg:
+    return (False, "Can't make devices secondary: %s" % ",".join(msg))
+  else:
+    return (True, "All devices secondary")
+
+
 class HooksRunner(object):
   """Hook runner.
 
@@ -1419,9 +1636,6 @@ class HooksRunner(object):
     Args:
       - hooks_base_dir: if not None, this overrides the
         constants.HOOKS_BASE_DIR (useful for unittests)
-      - logs_base_dir: if not None, this overrides the
-        constants.LOG_HOOKS_DIR (useful for unittests)
-      - logging: enable or disable logging of script output
 
     """
     if hooks_base_dir is None:
@@ -1433,7 +1647,6 @@ class HooksRunner(object):
     """Exec one hook script.
 
     Args:
-     - phase: the phase
      - script: the full path to the script
      - env: the environment with which to exec the script
 
@@ -1444,7 +1657,7 @@ class HooksRunner(object):
       fdstdin = open("/dev/null", "r")
       child = subprocess.Popen([script], stdin=fdstdin, stdout=subprocess.PIPE,
                                stderr=subprocess.STDOUT, close_fds=True,
-                               shell=False, cwd="/",env=env)
+                               shell=False, cwd="/", env=env)
       output = ""
       try:
         output = child.stdout.read(4096)
@@ -1515,8 +1728,44 @@ class HooksRunner(object):
     return rr
 
 
+class IAllocatorRunner(object):
+  """IAllocator runner.
+
+  This class is instantiated on the node side (ganeti-noded) and not on
+  the master side.
+
+  """
+  def Run(self, name, idata):
+    """Run an iallocator script.
+
+    Return value: tuple of:
+       - run status (one of the IARUN_ constants)
+       - stdout
+       - stderr
+       - fail reason (as from utils.RunResult)
+
+    """
+    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
+                                  os.path.isfile)
+    if alloc_script is None:
+      return (constants.IARUN_NOTFOUND, None, None, None)
+
+    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
+    try:
+      os.write(fd, idata)
+      os.close(fd)
+      result = utils.RunCmd([alloc_script, fin_name])
+      if result.failed:
+        return (constants.IARUN_FAILURE, result.stdout, result.stderr,
+                result.fail_reason)
+    finally:
+      os.unlink(fin_name)
+
+    return (constants.IARUN_SUCCESS, result.stdout, result.stderr, None)
+
+
 class DevCacheManager(object):
-  """Simple class for managing a chache of block device information.
+  """Simple class for managing a cache of block device information.
 
   """
   _DEV_PREFIX = "/dev/"