"""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 --------------------------
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):
"""
bdev_s = params[0]
bdev = objects.Disk.FromDict(bdev_s)
- return backend.RemoveBlockDevice(bdev)
+ return backend.BlockdevRemove(bdev)
@staticmethod
def perspective_blockdev_rename(params):
"""
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):
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):
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):
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):
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):
"""
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):
"""
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):
"""
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 --------------------------
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):
"""
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):
# volume --------------------------
@staticmethod
- def perspective_volume_list(params):
+ def perspective_lv_list(params):
"""Query the list of logical volumes in a given volume group.
"""
"""
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
"""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):
"""
instance = objects.Instance.FromDict(params[0])
- return backend.ShutdownInstance(instance)
+ timeout = params[1]
+ return backend.InstanceShutdown(instance, timeout)
@staticmethod
def perspective_instance_start(params):
"""
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):
"""
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 --------------------------
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):
"""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):
"""
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
"""
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 -----------------------
"""Query detailed information about existing OSes.
"""
- return [os.ToDict() for os in backend.DiagnoseOS()]
+ return backend.DiagnoseOS()
@staticmethod
def perspective_os_get(params):
"""
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 -----------------------
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
"""
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__':