Add a generic 'debug_level' attribute to opcodes
[ganeti-local] / daemons / ganeti-noded
index d2f4055..30087f1 100755 (executable)
 
 """Ganeti node daemon"""
 
-# functions in this module need to have a given name structure, so:
-# pylint: disable-msg=C0103
+# pylint: disable-msg=C0103,W0142
+
+# C0103: Functions in this module need to have a given name structure,
+# and the name of the daemon doesn't match
+
+# W0142: Used * or ** magic, since we do use it extensively in this
+# module
 
 import os
 import sys
-import resource
-import traceback
-import BaseHTTPServer
-import simplejson
-import errno
+import logging
+import signal
 
 from optparse import OptionParser
 
-
 from ganeti import backend
-from ganeti import logger
 from ganeti import constants
 from ganeti import objects
 from ganeti import errors
-from ganeti import ssconf
+from ganeti import jstore
+from ganeti import daemon
+from ganeti import http
 from ganeti import utils
+from ganeti import storage
+
+import ganeti.http.server # pylint: disable-msg=W0611
+
+
+queue_lock = None
+
+
+def _PrepareQueueLock():
+  """Try to prepare the queue lock.
+
+  @return: None for success, otherwise an exception object
+
+  """
+  global queue_lock # pylint: disable-msg=W0603
 
+  if queue_lock is not None:
+    return None
 
-class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
+  # Prepare job queue
+  try:
+    queue_lock = jstore.InitAndVerifyQueue(must_lock=False)
+    return None
+  except EnvironmentError, err:
+    return err
+
+
+def _RequireJobQueueLock(fn):
+  """Decorator for job queue manipulating functions.
+
+  """
+  QUEUE_LOCK_TIMEOUT = 10
+
+  def wrapper(*args, **kwargs):
+    # Locking in exclusive, blocking mode because there could be several
+    # children running at the same time. Waiting up to 10 seconds.
+    if _PrepareQueueLock() is not None:
+      raise errors.JobQueueError("Job queue failed initialization,"
+                                 " cannot update jobs")
+    queue_lock.Exclusive(blocking=True, timeout=QUEUE_LOCK_TIMEOUT)
+    try:
+      return fn(*args, **kwargs)
+    finally:
+      queue_lock.Unlock()
+
+  return wrapper
+
+
+class NodeHttpServer(http.server.HttpServer):
   """The server implementation.
 
   This class holds all methods exposed over the RPC interface.
 
   """
-  def do_PUT(self):
-    """Handle a post request.
+  # too many public methods, and unused args - all methods get params
+  # due to the API
+  # pylint: disable-msg=R0904,W0613
+  def __init__(self, *args, **kwargs):
+    http.server.HttpServer.__init__(self, *args, **kwargs)
+    self.noded_pid = os.getpid()
+
+  def HandleRequest(self, req):
+    """Handle a request.
 
     """
-    path = self.path
+    if req.request_method.upper() != http.HTTP_PUT:
+      raise http.HttpBadRequest()
+
+    path = req.request_path
     if path.startswith("/"):
       path = path[1:]
-    mname = "perspective_%s" % path
-    if not hasattr(self, mname):
-      self.send_error(404)
-      return False
 
-    method = getattr(self, mname)
-    try:
-      body_length = int(self.headers.get('Content-Length', '0'))
-    except ValueError:
-      self.send_error(400, 'No Content-Length header or invalid format')
-      return False
+    method = getattr(self, "perspective_%s" % path, None)
+    if method is None:
+      raise http.HttpNotFound()
 
     try:
-      body = self.rfile.read(body_length)
-    except socket.error, err:
-      logger.Error("Socket error while reading: %s" % str(err))
-      return
-    try:
-      params = simplejson.loads(body)
-      result = method(params)
-      payload = simplejson.dumps(result)
+      rvalue = method(req.request_body)
+      return True, rvalue
+
+    except backend.RPCFail, err:
+      # our custom failure exception; str(err) works fine if the
+      # exception was constructed with a single argument, and in
+      # this case, err.message == err.args[0] == str(err)
+      return (False, str(err))
+    except errors.QuitGanetiException, err:
+      # Tell parent to quit
+      logging.info("Shutting down the node daemon, arguments: %s",
+                   str(err.args))
+      os.kill(self.noded_pid, signal.SIGTERM)
+      # And return the error's arguments, which must be already in
+      # correct tuple format
+      return err.args
     except Exception, err:
-      self.send_error(500, "Error: %s" % str(err))
-      return False
-    self.send_response(200)
-    self.send_header('Content-Length', str(len(payload)))
-    self.end_headers()
-    self.wfile.write(payload)
-    return True
-
-  def log_message(self, format, *args):
-    """Log a request to the log.
-
-    This is the same as the parent, we just log somewhere else.
-
-    """
-    msg = ("%s - - [%s] %s" %
-           (self.address_string(),
-            self.log_date_time_string(),
-            format % args))
-    logger.Debug(msg)
+      logging.exception("Error in RPC call")
+      return False, "Error while executing backend function: %s" % str(err)
 
   # the new block devices  --------------------------
 
@@ -110,7 +152,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     bdev = objects.Disk.FromDict(bdev_s)
     if bdev is None:
       raise ValueError("can't unserialize data!")
-    return backend.CreateBlockDevice(bdev, size, owner, on_primary, info)
+    return backend.BlockdevCreate(bdev, size, owner, on_primary, info)
 
   @staticmethod
   def perspective_blockdev_remove(params):
@@ -119,7 +161,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     """
     bdev_s = params[0]
     bdev = objects.Disk.FromDict(bdev_s)
-    return backend.RemoveBlockDevice(bdev)
+    return backend.BlockdevRemove(bdev)
 
   @staticmethod
   def perspective_blockdev_rename(params):
@@ -127,7 +169,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
 
     """
     devlist = [(objects.Disk.FromDict(ds), uid) for ds, uid in params]
-    return backend.RenameBlockDevices(devlist)
+    return backend.BlockdevRename(devlist)
 
   @staticmethod
   def perspective_blockdev_assemble(params):
@@ -138,7 +180,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     bdev = objects.Disk.FromDict(bdev_s)
     if bdev is None:
       raise ValueError("can't unserialize data!")
-    return backend.AssembleBlockDevice(bdev, owner, on_primary)
+    return backend.BlockdevAssemble(bdev, owner, on_primary)
 
   @staticmethod
   def perspective_blockdev_shutdown(params):
@@ -149,7 +191,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     bdev = objects.Disk.FromDict(bdev_s)
     if bdev is None:
       raise ValueError("can't unserialize data!")
-    return backend.ShutdownBlockDevice(bdev)
+    return backend.BlockdevShutdown(bdev)
 
   @staticmethod
   def perspective_blockdev_addchildren(params):
@@ -164,7 +206,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
     if bdev is None or ndevs.count(None) > 0:
       raise ValueError("can't unserialize data!")
-    return backend.MirrorAddChildren(bdev, ndevs)
+    return backend.BlockdevAddchildren(bdev, ndevs)
 
   @staticmethod
   def perspective_blockdev_removechildren(params):
@@ -179,7 +221,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     ndevs = [objects.Disk.FromDict(disk_s) for disk_s in ndev_s]
     if bdev is None or ndevs.count(None) > 0:
       raise ValueError("can't unserialize data!")
-    return backend.MirrorRemoveChildren(bdev, ndevs)
+    return backend.BlockdevRemovechildren(bdev, ndevs)
 
   @staticmethod
   def perspective_blockdev_getmirrorstatus(params):
@@ -187,8 +229,9 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
 
     """
     disks = [objects.Disk.FromDict(dsk_s)
-            for dsk_s in params]
-    return backend.GetMirrorStatus(disks)
+             for dsk_s in params]
+    return [status.ToDict()
+            for status in backend.BlockdevGetmirrorstatus(disks)]
 
   @staticmethod
   def perspective_blockdev_find(params):
@@ -198,7 +241,12 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
 
     """
     disk = objects.Disk.FromDict(params[0])
-    return backend.FindBlockDevice(disk)
+
+    result = backend.BlockdevFind(disk)
+    if result is None:
+      return None
+
+    return result.ToDict()
 
   @staticmethod
   def perspective_blockdev_snapshot(params):
@@ -210,7 +258,80 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
 
     """
     cfbd = objects.Disk.FromDict(params[0])
-    return backend.SnapshotBlockDevice(cfbd)
+    return backend.BlockdevSnapshot(cfbd)
+
+  @staticmethod
+  def perspective_blockdev_grow(params):
+    """Grow a stack of devices.
+
+    """
+    cfbd = objects.Disk.FromDict(params[0])
+    amount = params[1]
+    return backend.BlockdevGrow(cfbd, amount)
+
+  @staticmethod
+  def perspective_blockdev_close(params):
+    """Closes the given block devices.
+
+    """
+    disks = [objects.Disk.FromDict(cf) for cf in params[1]]
+    return backend.BlockdevClose(params[0], disks)
+
+  @staticmethod
+  def perspective_blockdev_getsize(params):
+    """Compute the sizes of the given block devices.
+
+    """
+    disks = [objects.Disk.FromDict(cf) for cf in params[0]]
+    return backend.BlockdevGetsize(disks)
+
+  @staticmethod
+  def perspective_blockdev_export(params):
+    """Compute the sizes of the given block devices.
+
+    """
+    disk = objects.Disk.FromDict(params[0])
+    dest_node, dest_path, cluster_name = params[1:]
+    return backend.BlockdevExport(disk, dest_node, dest_path, cluster_name)
+
+  # blockdev/drbd specific methods ----------
+
+  @staticmethod
+  def perspective_drbd_disconnect_net(params):
+    """Disconnects the network connection of drbd disks.
+
+    Note that this is only valid for drbd disks, so the members of the
+    disk list must all be drbd devices.
+
+    """
+    nodes_ip, disks = params
+    disks = [objects.Disk.FromDict(cf) for cf in disks]
+    return backend.DrbdDisconnectNet(nodes_ip, disks)
+
+  @staticmethod
+  def perspective_drbd_attach_net(params):
+    """Attaches the network connection of drbd disks.
+
+    Note that this is only valid for drbd disks, so the members of the
+    disk list must all be drbd devices.
+
+    """
+    nodes_ip, disks, instance_name, multimaster = params
+    disks = [objects.Disk.FromDict(cf) for cf in disks]
+    return backend.DrbdAttachNet(nodes_ip, disks,
+                                     instance_name, multimaster)
+
+  @staticmethod
+  def perspective_drbd_wait_sync(params):
+    """Wait until DRBD disks are synched.
+
+    Note that this is only valid for drbd disks, so the members of the
+    disk list must all be drbd devices.
+
+    """
+    nodes_ip, disks = params
+    disks = [objects.Disk.FromDict(cf) for cf in disks]
+    return backend.DrbdWaitSync(nodes_ip, disks)
 
   # export/import  --------------------------
 
@@ -222,7 +343,11 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     disk = objects.Disk.FromDict(params[0])
     dest_node = params[1]
     instance = objects.Instance.FromDict(params[2])
-    return backend.ExportSnapshot(disk, dest_node, instance)
+    cluster_name = params[3]
+    dev_idx = params[4]
+    debug = params[5]
+    return backend.ExportSnapshot(disk, dest_node, instance,
+                                  cluster_name, dev_idx, debug)
 
   @staticmethod
   def perspective_finalize_export(params):
@@ -243,10 +368,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
 
     """
     path = params[0]
-    einfo = backend.ExportInfo(path)
-    if einfo is None:
-      return einfo
-    return einfo.Dumps()
+    return backend.ExportInfo(path)
 
   @staticmethod
   def perspective_export_list(params):
@@ -270,7 +392,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
   # volume  --------------------------
 
   @staticmethod
-  def perspective_volume_list(params):
+  def perspective_lv_list(params):
     """Query the list of logical volumes in a given volume group.
 
     """
@@ -284,6 +406,32 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     """
     return backend.ListVolumeGroups()
 
+  # Storage --------------------------
+
+  @staticmethod
+  def perspective_storage_list(params):
+    """Get list of storage units.
+
+    """
+    (su_name, su_args, name, fields) = params
+    return storage.GetStorage(su_name, *su_args).List(name, fields)
+
+  @staticmethod
+  def perspective_storage_modify(params):
+    """Modify a storage unit.
+
+    """
+    (su_name, su_args, name, changes) = params
+    return storage.GetStorage(su_name, *su_args).Modify(name, changes)
+
+  @staticmethod
+  def perspective_storage_execute(params):
+    """Execute an operation on a storage unit.
+
+    """
+    (su_name, su_args, name, op) = params
+    return storage.GetStorage(su_name, *su_args).Execute(name, op)
+
   # bridge  --------------------------
 
   @staticmethod
@@ -301,28 +449,30 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     """Install an OS on a given instance.
 
     """
-    inst_s, os_disk, swap_disk = params
+    inst_s = params[0]
     inst = objects.Instance.FromDict(inst_s)
-    return backend.AddOSToInstance(inst, os_disk, swap_disk)
+    reinstall = params[1]
+    debug = params[2]
+    return backend.InstanceOsAdd(inst, reinstall, debug)
 
   @staticmethod
   def perspective_instance_run_rename(params):
     """Runs the OS rename script for an instance.
 
     """
-    inst_s, old_name, os_disk, swap_disk = params
+    inst_s, old_name, debug = params
     inst = objects.Instance.FromDict(inst_s)
-    return backend.RunRenameInstance(inst, old_name, os_disk, swap_disk)
+    return backend.RunRenameInstance(inst, old_name, debug)
 
   @staticmethod
   def perspective_instance_os_import(params):
     """Run the import function of an OS onto a given instance.
 
     """
-    inst_s, os_disk, swap_disk, src_node, src_image = params
+    inst_s, src_node, src_images, cluster_name, debug = params
     inst = objects.Instance.FromDict(inst_s)
-    return backend.ImportOSIntoInstance(inst, os_disk, swap_disk,
-                                        src_node, src_image)
+    return backend.ImportOSIntoInstance(inst, src_node, src_images,
+                                        cluster_name, debug)
 
   @staticmethod
   def perspective_instance_shutdown(params):
@@ -330,7 +480,8 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
 
     """
     instance = objects.Instance.FromDict(params[0])
-    return backend.ShutdownInstance(instance)
+    timeout = params[1]
+    return backend.InstanceShutdown(instance, timeout)
 
   @staticmethod
   def perspective_instance_start(params):
@@ -338,8 +489,42 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
 
     """
     instance = objects.Instance.FromDict(params[0])
-    extra_args = params[1]
-    return backend.StartInstance(instance, extra_args)
+    return backend.StartInstance(instance)
+
+  @staticmethod
+  def perspective_migration_info(params):
+    """Gather information about an instance to be migrated.
+
+    """
+    instance = objects.Instance.FromDict(params[0])
+    return backend.MigrationInfo(instance)
+
+  @staticmethod
+  def perspective_accept_instance(params):
+    """Prepare the node to accept an instance.
+
+    """
+    instance, info, target = params
+    instance = objects.Instance.FromDict(instance)
+    return backend.AcceptInstance(instance, info, target)
+
+  @staticmethod
+  def perspective_finalize_migration(params):
+    """Finalize the instance migration.
+
+    """
+    instance, info, success = params
+    instance = objects.Instance.FromDict(instance)
+    return backend.FinalizeMigration(instance, info, success)
+
+  @staticmethod
+  def perspective_instance_migrate(params):
+    """Migrates an instance.
+
+    """
+    instance, target, live = params
+    instance = objects.Instance.FromDict(instance)
+    return backend.MigrateInstance(instance, target, live)
 
   @staticmethod
   def perspective_instance_reboot(params):
@@ -348,29 +533,37 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     """
     instance = objects.Instance.FromDict(params[0])
     reboot_type = params[1]
-    extra_args = params[2]
-    return backend.RebootInstance(instance, reboot_type, extra_args)
+    shutdown_timeout = params[2]
+    return backend.InstanceReboot(instance, reboot_type, shutdown_timeout)
 
   @staticmethod
   def perspective_instance_info(params):
     """Query instance information.
 
     """
-    return backend.GetInstanceInfo(params[0])
+    return backend.GetInstanceInfo(params[0], params[1])
+
+  @staticmethod
+  def perspective_instance_migratable(params):
+    """Query whether the specified instance can be migrated.
+
+    """
+    instance = objects.Instance.FromDict(params[0])
+    return backend.GetInstanceMigratable(instance)
 
   @staticmethod
   def perspective_all_instances_info(params):
     """Query information about all instances.
 
     """
-    return backend.GetAllInstancesInfo()
+    return backend.GetAllInstancesInfo(params[0])
 
   @staticmethod
   def perspective_instance_list(params):
     """Query the list of running instances.
 
     """
-    return backend.GetInstanceList()
+    return backend.GetInstanceList(params[0])
 
   # node --------------------------
 
@@ -383,12 +576,19 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
                          live_port_needed=params[4], source=params[0])
 
   @staticmethod
+  def perspective_node_has_ip_address(params):
+    """Checks if a node has the given ip address.
+
+    """
+    return utils.OwnIpAddress(params[0])
+
+  @staticmethod
   def perspective_node_info(params):
     """Query node information.
 
     """
-    vgname = params[0]
-    return backend.GetNodeInfo(vgname)
+    vgname, hypervisor_type = params
+    return backend.GetNodeInfo(vgname, hypervisor_type)
 
   @staticmethod
   def perspective_node_add(params):
@@ -403,28 +603,28 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     """Run a verify sequence on this node.
 
     """
-    return backend.VerifyNode(params[0])
+    return backend.VerifyNode(params[0], params[1])
 
   @staticmethod
   def perspective_node_start_master(params):
     """Promote this node to master status.
 
     """
-    return backend.StartMaster()
+    return backend.StartMaster(params[0], params[1])
 
   @staticmethod
   def perspective_node_stop_master(params):
     """Demote this node from master status.
 
     """
-    return backend.StopMaster()
+    return backend.StopMaster(params[0])
 
   @staticmethod
   def perspective_node_leave_cluster(params):
     """Cleanup after leaving a cluster.
 
     """
-    return backend.LeaveCluster()
+    return backend.LeaveCluster(params[0])
 
   @staticmethod
   def perspective_node_volumes(params):
@@ -433,6 +633,23 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     """
     return backend.NodeVolumes()
 
+  @staticmethod
+  def perspective_node_demote_from_mc(params):
+    """Demote a node from the master candidate role.
+
+    """
+    return backend.DemoteFromMC()
+
+
+  @staticmethod
+  def perspective_node_powercycle(params):
+    """Tries to powercycle the nod.
+
+    """
+    hypervisor_type = params[0]
+    return backend.PowercycleNode(hypervisor_type)
+
+
   # cluster --------------------------
 
   @staticmethod
@@ -452,6 +669,20 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     """
     return backend.UploadFile(*params)
 
+  @staticmethod
+  def perspective_master_info(params):
+    """Query master information.
+
+    """
+    return backend.GetMasterInfo()
+
+  @staticmethod
+  def perspective_write_ssconf_files(params):
+    """Write ssconf files.
+
+    """
+    (values,) = params
+    return backend.WriteSsconfFiles(values)
 
   # os -----------------------
 
@@ -460,7 +691,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     """Query detailed information about existing OSes.
 
     """
-    return [os.ToDict() for os in backend.DiagnoseOS()]
+    return backend.DiagnoseOS()
 
   @staticmethod
   def perspective_os_get(params):
@@ -468,10 +699,7 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
 
     """
     name = params[0]
-    try:
-      os_obj = backend.OSFromDisk(name)
-    except errors.InvalidOS, err:
-      os_obj = objects.OS.FromInvalidOS(err)
+    os_obj = backend.OSFromDisk(name)
     return os_obj.ToDict()
 
   # hooks -----------------------
@@ -485,6 +713,17 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
     hr = backend.HooksRunner()
     return hr.RunHooks(hpath, phase, env)
 
+  # iallocator -----------------
+
+  @staticmethod
+  def perspective_iallocator_runner(params):
+    """Run an iallocator script.
+
+    """
+    name, idata = params
+    iar = backend.IAllocatorRunner()
+    return iar.Run(name, idata)
+
   # test -----------------------
 
   @staticmethod
@@ -493,129 +732,137 @@ class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
 
     """
     duration = params[0]
-    return utils.TestDelay(duration)
+    status, rval = utils.TestDelay(duration)
+    if not status:
+      raise backend.RPCFail(rval)
+    return rval
 
+  # file storage ---------------
 
-def ParseOptions():
-  """Parse the command line options.
+  @staticmethod
+  def perspective_file_storage_dir_create(params):
+    """Create the file storage directory.
 
-  Returns:
-    (options, args) as from OptionParser.parse_args()
+    """
+    file_storage_dir = params[0]
+    return backend.CreateFileStorageDir(file_storage_dir)
 
-  """
-  parser = OptionParser(description="Ganeti node daemon",
-                        usage="%prog [-f] [-d]",
-                        version="%%prog (ganeti) %s" %
-                        constants.RELEASE_VERSION)
+  @staticmethod
+  def perspective_file_storage_dir_remove(params):
+    """Remove the file storage directory.
 
-  parser.add_option("-f", "--foreground", dest="fork",
-                    help="Don't detach from the current terminal",
-                    default=True, action="store_false")
-  parser.add_option("-d", "--debug", dest="debug",
-                    help="Enable some debug messages",
-                    default=False, action="store_true")
-  options, args = parser.parse_args()
-  return options, args
+    """
+    file_storage_dir = params[0]
+    return backend.RemoveFileStorageDir(file_storage_dir)
 
+  @staticmethod
+  def perspective_file_storage_dir_rename(params):
+    """Rename the file storage directory.
 
-def main():
-  """Main function for the node daemon.
+    """
+    old_file_storage_dir = params[0]
+    new_file_storage_dir = params[1]
+    return backend.RenameFileStorageDir(old_file_storage_dir,
+                                        new_file_storage_dir)
 
-  """
-  options, args = ParseOptions()
-  utils.debug = options.debug
-  for fname in (constants.SSL_CERT_FILE,):
-    if not os.path.isfile(fname):
-      print "config %s not there, will not run." % fname
-      sys.exit(5)
+  # jobs ------------------------
 
-  try:
-    ss = ssconf.SimpleStore()
-    port = ss.GetNodeDaemonPort()
-    pwdata = ss.GetNodeDaemonPassword()
-  except errors.ConfigurationError, err:
-    print "Cluster configuration incomplete: '%s'" % str(err)
-    sys.exit(5)
-
-  # create /var/run/ganeti if not existing, in order to take care of
-  # tmpfs /var/run
-  if not os.path.exists(constants.BDEV_CACHE_DIR):
-    try:
-      os.mkdir(constants.BDEV_CACHE_DIR, 0755)
-    except EnvironmentError, err:
-      if err.errno != errno.EEXIST:
-        print ("Node setup wrong, cannot create directory %s: %s" %
-               (constants.BDEV_CACHE_DIR, err))
-        sys.exit(5)
-  if not os.path.isdir(constants.BDEV_CACHE_DIR):
-    print ("Node setup wrong, %s is not a directory" %
-           constants.BDEV_CACHE_DIR)
-    sys.exit(5)
+  @staticmethod
+  @_RequireJobQueueLock
+  def perspective_jobqueue_update(params):
+    """Update job queue.
+
+    """
+    (file_name, content) = params
+    return backend.JobQueueUpdate(file_name, content)
+
+  @staticmethod
+  @_RequireJobQueueLock
+  def perspective_jobqueue_purge(params):
+    """Purge job queue.
+
+    """
+    return backend.JobQueuePurge()
+
+  @staticmethod
+  @_RequireJobQueueLock
+  def perspective_jobqueue_rename(params):
+    """Rename a job queue file.
+
+    """
+    # TODO: What if a file fails to rename?
+    return [backend.JobQueueRename(old, new) for old, new in params]
+
+  @staticmethod
+  def perspective_jobqueue_set_drain(params):
+    """Set/unset the queue drain flag.
+
+    """
+    drain_flag = params[0]
+    return backend.JobQueueSetDrainFlag(drain_flag)
 
-  # become a daemon
-  if options.fork:
-    createDaemon()
 
-  logger.SetupLogging(twisted_workaround=True, debug=options.debug,
-                      program="ganeti-noded")
+  # hypervisor ---------------
 
-  httpd = BaseHTTPServer.HTTPServer(('', port), ServerObject)
-  httpd.serve_forever()
+  @staticmethod
+  def perspective_hypervisor_validate_params(params):
+    """Validate the hypervisor parameters.
+
+    """
+    (hvname, hvparams) = params
+    return backend.ValidateHVParams(hvname, hvparams)
 
 
-def createDaemon():
-  """Detach a process from the controlling terminal and run it in the
-  background as a daemon.
+def CheckNoded(_, args):
+  """Initial checks whether to run or exit with a failure.
 
   """
-  UMASK = 077
-  WORKDIR = "/"
-  # Default maximum for the number of available file descriptors.
-  if 'SC_OPEN_MAX' in os.sysconf_names:
-    try:
-      MAXFD = os.sysconf('SC_OPEN_MAX')
-      if MAXFD < 0:
-        MAXFD = 1024
-    except OSError:
-      MAXFD = 1024
+  if args: # noded doesn't take any arguments
+    print >> sys.stderr, ("Usage: %s [-f] [-d] [-p port] [-b ADDRESS]" %
+                          sys.argv[0])
+    sys.exit(constants.EXIT_FAILURE)
+
+
+def ExecNoded(options, _):
+  """Main node daemon function, executed with the PID file held.
+
+  """
+  # Read SSL certificate
+  if options.ssl:
+    ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
+                                    ssl_cert_path=options.ssl_cert)
   else:
-    MAXFD = 1024
-  # The standard I/O file descriptors are redirected to /dev/null by default.
-  #REDIRECT_TO = getattr(os, "devnull", "/dev/null")
-  REDIRECT_TO = constants.LOG_NODESERVER
+    ssl_params = None
+
+  err = _PrepareQueueLock()
+  if err is not None:
+    # this might be some kind of file-system/permission error; while
+    # this breaks the job queue functionality, we shouldn't prevent
+    # startup of the whole node daemon because of this
+    logging.critical("Can't init/verify the queue, proceeding anyway: %s", err)
+
+  mainloop = daemon.Mainloop()
+  server = NodeHttpServer(mainloop, options.bind_address, options.port,
+                          ssl_params=ssl_params, ssl_verify_peer=True)
+  server.Start()
   try:
-    pid = os.fork()
-  except OSError, e:
-    raise Exception("%s [%d]" % (e.strerror, e.errno))
-  if (pid == 0):  # The first child.
-    os.setsid()
-    try:
-      pid = os.fork() # Fork a second child.
-    except OSError, e:
-      raise Exception("%s [%d]" % (e.strerror, e.errno))
-    if (pid == 0):  # The second child.
-      os.chdir(WORKDIR)
-      os.umask(UMASK)
-    else:
-      # exit() or _exit()?  See below.
-      os._exit(0) # Exit parent (the first child) of the second child.
-  else:
-    os._exit(0) # Exit parent of the first child.
-  maxfd = resource.getrlimit(resource.RLIMIT_NOFILE)[1]
-  if (maxfd == resource.RLIM_INFINITY):
-    maxfd = MAXFD
+    mainloop.Run()
+  finally:
+    server.Stop()
 
-  # Iterate through and close all file descriptors.
-  for fd in range(0, maxfd):
-    try:
-      os.close(fd)
-    except OSError: # ERROR, fd wasn't open to begin with (ignored)
-      pass
-  os.open(REDIRECT_TO, os.O_RDWR|os.O_CREAT|os.O_APPEND, 0600)
-  # Duplicate standard input to standard output and standard error.
-  os.dup2(0, 1)     # standard output (1)
-  os.dup2(0, 2)     # standard error (2)
-  return(0)
+
+def main():
+  """Main function for the node daemon.
+
+  """
+  parser = OptionParser(description="Ganeti node daemon",
+                        usage="%prog [-f] [-d] [-p port] [-b ADDRESS]",
+                        version="%%prog (ganeti) %s" %
+                        constants.RELEASE_VERSION)
+  dirs = [(val, constants.RUN_DIRS_MODE) for val in constants.SUB_RUN_DIRS]
+  dirs.append((constants.LOG_OS_DIR, 0750))
+  dirs.append((constants.LOCK_DIR, 1777))
+  daemon.GenericMain(constants.NODED, parser, dirs, CheckNoded, ExecNoded)
 
 
 if __name__ == '__main__':