"""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 traceback
-import SocketServer
-import errno
import logging
import signal
from ganeti import daemon
from ganeti import http
from ganeti import utils
+from ganeti import storage
-import ganeti.http.server
+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
+
+ # 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 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)
This class holds all methods exposed over the RPC interface.
"""
+ # 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()
"""
disks = [objects.Disk.FromDict(dsk_s)
- for dsk_s in params]
- return backend.BlockdevGetmirrorstatus(disks)
+ for dsk_s in params]
+ return [status.ToDict()
+ for status in backend.BlockdevGetmirrorstatus(disks)]
@staticmethod
def perspective_blockdev_find(params):
"""
disk = objects.Disk.FromDict(params[0])
- return backend.BlockdevFind(disk)
+
+ result = backend.BlockdevFind(disk)
+ if result is None:
+ return None
+
+ return result.ToDict()
@staticmethod
def perspective_blockdev_snapshot(params):
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
instance = objects.Instance.FromDict(params[2])
cluster_name = params[3]
dev_idx = params[4]
+ debug = params[5]
return backend.ExportSnapshot(disk, dest_node, instance,
- cluster_name, dev_idx)
+ cluster_name, dev_idx, debug)
@staticmethod
def perspective_finalize_export(params):
"""
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
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.
"""
- inst_s, old_name = params
+ inst_s, old_name, debug = params
inst = objects.Instance.FromDict(inst_s)
- return backend.RunRenameInstance(inst, old_name)
+ 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, src_node, src_images, cluster_name = params
+ inst_s, src_node, src_images, cluster_name, debug = params
inst = objects.Instance.FromDict(inst_s)
return backend.ImportOSIntoInstance(inst, src_node, src_images,
- cluster_name)
+ cluster_name, debug)
@staticmethod
def perspective_instance_shutdown(params):
"""
instance = objects.Instance.FromDict(params[0])
- return backend.InstanceShutdown(instance)
+ timeout = params[1]
+ return backend.InstanceShutdown(instance, timeout)
@staticmethod
def perspective_instance_start(params):
"""
instance = objects.Instance.FromDict(params[0])
reboot_type = params[1]
- return backend.InstanceReboot(instance, reboot_type)
+ shutdown_timeout = params[2]
+ return backend.InstanceReboot(instance, reboot_type, shutdown_timeout)
@staticmethod
def perspective_instance_info(params):
"""Cleanup after leaving a cluster.
"""
- return backend.LeaveCluster()
+ return backend.LeaveCluster(params[0])
@staticmethod
def perspective_node_volumes(params):
return backend.ValidateHVParams(hvname, hvparams)
-def ParseOptions():
- """Parse the command line options.
-
- @return: (options, args) as from OptionParser.parse_args()
+def CheckNoded(_, args):
+ """Initial checks whether to run or exit with a failure.
"""
- parser = OptionParser(description="Ganeti node daemon",
- usage="%prog [-f] [-d] [-b ADDRESS]",
- version="%%prog (ganeti) %s" %
- constants.RELEASE_VERSION)
+ 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)
- 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")
- parser.add_option("-b", "--bind", dest="bind_address",
- help="Bind address",
- default="", metavar="ADDRESS")
- options, args = parser.parse_args()
- return options, args
+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:
+ 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:
+ mainloop.Run()
+ finally:
+ server.Stop()
def main():
"""Main function for the node daemon.
"""
- global queue_lock
- daemon_name = constants.NODED
-
- options, args = ParseOptions()
-
- if options.fork:
- utils.CloseFDs()
-
- for fname in (constants.SSL_CERT_FILE,):
- if not os.path.isfile(fname):
- print "config %s not there, will not run." % fname
- sys.exit(constants.EXIT_NOTCLUSTER)
-
- port = utils.GetDaemonPort(constants.NODED)
-
+ 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))
- utils.EnsureDirs(dirs)
-
- # become a daemon
- if options.fork:
- utils.Daemonize(logfile=constants.DAEMONS_LOGFILES[daemon_name])
-
- utils.WritePidFile(constants.NODED_PID)
- try:
- utils.SetupLogging(logfile=constants.DAEMONS_LOGFILES[daemon_name],
- debug=options.debug,
- stderr_logging=not options.fork)
- logging.info("ganeti node daemon startup")
-
- # Read SSL certificate
- ssl_params = http.HttpSslParams(ssl_key_path=constants.SSL_CERT_FILE,
- ssl_cert_path=constants.SSL_CERT_FILE)
-
- # Prepare job queue
- queue_lock = jstore.InitAndVerifyQueue(must_lock=False)
-
- mainloop = daemon.Mainloop()
- server = NodeHttpServer(mainloop, options.bind_address, port,
- ssl_params=ssl_params, ssl_verify_peer=True)
- server.Start()
- try:
- mainloop.Run()
- finally:
- server.Stop()
- finally:
- utils.RemovePidFile(constants.NODED_PID)
+ daemon.GenericMain(constants.NODED, parser, dirs, CheckNoded, ExecNoded)
if __name__ == '__main__':
projects / ganeti-local / blobdiff