Move some code into separate class in import/export daemon

[ganeti-local] / daemons / ganeti-noded

diff --git a/daemons/ganeti-noded b/daemons/ganeti-noded
index 7a17e4f..ef78513 100755 (executable)
--- a/daemons/ganeti-noded
+++ b/daemons/ganeti-noded
@@ -31,7 +31,6 @@
 
 import os
 import sys
 
 import os
 import sys

-import SocketServer

 import logging
 import signal
 
 import logging
 import signal
 

@@ -46,13 +45,33 @@ from ganeti import daemon
 from ganeti import http
 from ganeti import utils
 from ganeti import storage
 from ganeti import http
 from ganeti import utils
 from ganeti import storage

+from ganeti import serializer

 
 

-import ganeti.http.server
+import ganeti.http.server # pylint: disable-msg=W0611

 
 
 queue_lock = None
 
 
 
 
 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
+
+  # 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.
 
 def _RequireJobQueueLock(fn):
   """Decorator for job queue manipulating functions.
 

@@ -62,6 +81,9 @@ def _RequireJobQueueLock(fn):
   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.
   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)
     queue_lock.Exclusive(blocking=True, timeout=QUEUE_LOCK_TIMEOUT)
     try:
       return fn(*args, **kwargs)

@@ -71,6 +93,21 @@ def _RequireJobQueueLock(fn):
   return wrapper
 
 
   return wrapper
 
 

+def _DecodeImportExportIO(ieio, ieioargs):
+  """Decodes import/export I/O information.
+
+  """
+  if ieio == constants.IEIO_RAW_DISK:
+    assert len(ieioargs) == 1
+    return (objects.Disk.FromDict(ieioargs[0]), )
+
+  if ieio == constants.IEIO_SCRIPT:
+    assert len(ieioargs) == 2
+    return (objects.Disk.FromDict(ieioargs[0]), ieioargs[1])
+
+  return ieioargs
+
+

 class NodeHttpServer(http.server.HttpServer):
   """The server implementation.
 
 class NodeHttpServer(http.server.HttpServer):
   """The server implementation.
 

@@ -100,14 +137,13 @@ class NodeHttpServer(http.server.HttpServer):
       raise http.HttpNotFound()
 
     try:
       raise http.HttpNotFound()
 
     try:

-      rvalue = method(req.request_body)
-      return True, rvalue
+      result = (True, method(serializer.LoadJson(req.request_body)))

 
     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)
 
     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))
+      result = (False, str(err))

     except errors.QuitGanetiException, err:
       # Tell parent to quit
       logging.info("Shutting down the node daemon, arguments: %s",
     except errors.QuitGanetiException, err:
       # Tell parent to quit
       logging.info("Shutting down the node daemon, arguments: %s",

@@ -115,10 +151,12 @@ class NodeHttpServer(http.server.HttpServer):
       os.kill(self.noded_pid, signal.SIGTERM)
       # And return the error's arguments, which must be already in
       # correct tuple format
       os.kill(self.noded_pid, signal.SIGTERM)
       # And return the error's arguments, which must be already in
       # correct tuple format

-      return err.args
+      result = err.args

     except Exception, err:
       logging.exception("Error in RPC call")
     except Exception, err:
       logging.exception("Error in RPC call")

-      return False, "Error while executing backend function: %s" % str(err)
+      result = (False, "Error while executing backend function: %s" % str(err))
+
+    return serializer.DumpJson(result, indent=False)

 
   # the new block devices  --------------------------
 
 
   # the new block devices  --------------------------
 

@@ -315,26 +353,19 @@ class NodeHttpServer(http.server.HttpServer):
   # export/import  --------------------------
 
   @staticmethod
   # export/import  --------------------------
 
   @staticmethod

-  def perspective_snapshot_export(params):
-    """Export a given snapshot.
-
-    """
-    disk = objects.Disk.FromDict(params[0])
-    dest_node = params[1]
-    instance = objects.Instance.FromDict(params[2])
-    cluster_name = params[3]
-    dev_idx = params[4]
-    return backend.ExportSnapshot(disk, dest_node, instance,
-                                  cluster_name, dev_idx)
-
-  @staticmethod

   def perspective_finalize_export(params):
     """Expose the finalize export functionality.
 
     """
     instance = objects.Instance.FromDict(params[0])
   def perspective_finalize_export(params):
     """Expose the finalize export functionality.
 
     """
     instance = objects.Instance.FromDict(params[0])

-    snap_disks = [objects.Disk.FromDict(str_data)
-                  for str_data in params[1]]
+
+    snap_disks = []
+    for disk in params[1]:
+      if isinstance(disk, bool):
+        snap_disks.append(disk)
+      else:
+        snap_disks.append(objects.Disk.FromDict(disk))
+

     return backend.FinalizeExport(instance, snap_disks)
 
   @staticmethod
     return backend.FinalizeExport(instance, snap_disks)
 
   @staticmethod

@@ -430,26 +461,17 @@ class NodeHttpServer(http.server.HttpServer):
     inst_s = params[0]
     inst = objects.Instance.FromDict(inst_s)
     reinstall = params[1]
     inst_s = params[0]
     inst = objects.Instance.FromDict(inst_s)
     reinstall = params[1]

-    return backend.InstanceOsAdd(inst, reinstall)
+    debug = params[2]
+    return backend.InstanceOsAdd(inst, reinstall, debug)

 
   @staticmethod
   def perspective_instance_run_rename(params):
     """Runs the OS rename script for an instance.
 
     """
 
   @staticmethod
   def perspective_instance_run_rename(params):
     """Runs the OS rename script for an instance.
 
     """

-    inst_s, old_name = params
+    inst_s, old_name, debug = params

     inst = objects.Instance.FromDict(inst_s)
     inst = objects.Instance.FromDict(inst_s)

-    return backend.RunRenameInstance(inst, old_name)
-
-  @staticmethod
-  def perspective_instance_os_import(params):
-    """Run the import function of an OS onto a given instance.
-
-    """
-    inst_s, src_node, src_images, cluster_name = params
-    inst = objects.Instance.FromDict(inst_s)
-    return backend.ImportOSIntoInstance(inst, src_node, src_images,
-                                        cluster_name)
+    return backend.RunRenameInstance(inst, old_name, debug)

 
   @staticmethod
   def perspective_instance_shutdown(params):
 
   @staticmethod
   def perspective_instance_shutdown(params):

@@ -789,6 +811,75 @@ class NodeHttpServer(http.server.HttpServer):
     (hvname, hvparams) = params
     return backend.ValidateHVParams(hvname, hvparams)
 
     (hvname, hvparams) = params
     return backend.ValidateHVParams(hvname, hvparams)
 

+  # Crypto
+
+  @staticmethod
+  def perspective_x509_cert_create(params):
+    """Creates a new X509 certificate for SSL/TLS.
+
+    """
+    (validity, ) = params
+    return backend.CreateX509Certificate(validity)
+
+  @staticmethod
+  def perspective_x509_cert_remove(params):
+    """Removes a X509 certificate.
+
+    """
+    (name, ) = params
+    return backend.RemoveX509Certificate(name)
+
+  # Import and export
+
+  @staticmethod
+  def perspective_import_start(params):
+    """Starts an import daemon.
+
+    """
+    (x509_key_name, source_x509_ca, instance, dest, dest_args) = params
+    return backend.StartImportExportDaemon(constants.IEM_IMPORT,
+                                           x509_key_name, source_x509_ca,
+                                           None, None,
+                                           objects.Instance.FromDict(instance),
+                                           dest,
+                                           _DecodeImportExportIO(dest,
+                                                                 dest_args))
+  @staticmethod
+  def perspective_export_start(params):
+    """Starts an export daemon.
+
+    """
+    (x509_key_name, dest_x509_ca, host, port, instance,
+     source, source_args) = params
+    return backend.StartImportExportDaemon(constants.IEM_EXPORT,
+                                           x509_key_name, dest_x509_ca,
+                                           host, port,
+                                           objects.Instance.FromDict(instance),
+                                           source,
+                                           _DecodeImportExportIO(source,
+                                                                 source_args))
+
+  @staticmethod
+  def perspective_impexp_status(params):
+    """Retrieves the status of an import or export daemon.
+
+    """
+    return backend.GetImportExportStatus(params[0])
+
+  @staticmethod
+  def perspective_impexp_abort(params):
+    """Aborts an import or export.
+
+    """
+    return backend.AbortImportExport(params[0])
+
+  @staticmethod
+  def perspective_impexp_cleanup(params):
+    """Cleans up after an import or export.
+
+    """
+    return backend.CleanupImportExport(params[0])
+

 
 def CheckNoded(_, args):
   """Initial checks whether to run or exit with a failure.
 
 def CheckNoded(_, args):
   """Initial checks whether to run or exit with a failure.

@@ -804,8 +895,6 @@ def ExecNoded(options, _):
   """Main node daemon function, executed with the PID file held.
 
   """
   """Main node daemon function, executed with the PID file held.
 
   """

-  global queue_lock # pylint: disable-msg=W0603
-

   # Read SSL certificate
   if options.ssl:
     ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,
   # Read SSL certificate
   if options.ssl:
     ssl_params = http.HttpSslParams(ssl_key_path=options.ssl_key,

@@ -813,8 +902,12 @@ def ExecNoded(options, _):
   else:
     ssl_params = None
 
   else:
     ssl_params = None
 

-  # Prepare job queue
-  queue_lock = jstore.InitAndVerifyQueue(must_lock=False)
+  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,
 
   mainloop = daemon.Mainloop()
   server = NodeHttpServer(mainloop, options.bind_address, options.port,

@@ -837,7 +930,11 @@ def main():
   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))
   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)
+  dirs.append((constants.CRYPTO_KEYS_DIR, constants.CRYPTO_KEYS_DIR_MODE))
+  dirs.append((constants.IMPORT_EXPORT_DIR, constants.IMPORT_EXPORT_DIR_MODE))
+  daemon.GenericMain(constants.NODED, parser, dirs, CheckNoded, ExecNoded,
+                     default_ssl_cert=constants.NODED_CERT_FILE,
+                     default_ssl_key=constants.NODED_CERT_FILE)

 
 
 if __name__ == '__main__':
 
 
 if __name__ == '__main__':