import os
import sys
-import resource
import traceback
-import BaseHTTPServer
-import simplejson
+import SocketServer
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 jstore
from ganeti import ssconf
+from ganeti import http
from ganeti import utils
-class ServerObject(BaseHTTPServer.BaseHTTPRequestHandler):
+queue_lock = None
+
+
+class NodeDaemonRequestHandler(http.HTTPRequestHandler):
"""The server implementation.
This class holds all methods exposed over the RPC interface.
"""
- def do_PUT(self):
- """Handle a post request.
+ def HandleRequest(self):
+ """Handle a request.
"""
+ if self.command.upper() != "PUT":
+ raise http.HTTPBadRequest()
+
path = self.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 httperror.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)
- 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\n" %
- (self.address_string(),
- self.log_date_time_string(),
- format % args))
- logger.Debug(msg)
+ try:
+ return method(self.post_data)
+ except:
+ logging.exception("Error in RPC call")
+ raise
+ except errors.QuitGanetiException, err:
+ # Tell parent to quit
+ os.kill(self.server.noded_pid, signal.SIGTERM)
# the new block devices --------------------------
cfbd = objects.Disk.FromDict(params[0])
return backend.SnapshotBlockDevice(cfbd)
+ @staticmethod
+ def perspective_blockdev_grow(params):
+ """Grow a stack of devices.
+
+ """
+ cfbd = objects.Disk.FromDict(params[0])
+ amount = params[1]
+ return backend.GrowBlockDevice(cfbd, amount)
+
+ @staticmethod
+ def perspective_blockdev_close(params):
+ """Closes the given block devices.
+
+ """
+ disks = [objects.Disk.FromDict(cf) for cf in params]
+ return backend.CloseBlockDevices(disks)
+
# export/import --------------------------
@staticmethod
return backend.StartInstance(instance, extra_args)
@staticmethod
+ def perspective_instance_migrate(params):
+ """Migrates an instance.
+
+ """
+ instance, target, live = params
+ return backend.MigrateInstance(instance, target, live)
+
+ @staticmethod
def perspective_instance_reboot(params):
"""Reboot an instance.
"""Promote this node to master status.
"""
- return backend.StartMaster()
+ return backend.StartMaster(params[0])
@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):
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)
+ @staticmethod
+ def perspective_file_storage_dir_create(params):
+ """Create the file storage directory.
+
+ """
+ file_storage_dir = params[0]
+ return backend.CreateFileStorageDir(file_storage_dir)
+
+ @staticmethod
+ def perspective_file_storage_dir_remove(params):
+ """Remove the file storage directory.
+
+ """
+ file_storage_dir = params[0]
+ return backend.RemoveFileStorageDir(file_storage_dir)
+
+ @staticmethod
+ def perspective_file_storage_dir_rename(params):
+ """Rename the file storage directory.
+
+ """
+ old_file_storage_dir = params[0]
+ new_file_storage_dir = params[1]
+ return backend.RenameFileStorageDir(old_file_storage_dir,
+ new_file_storage_dir)
+
+ @staticmethod
+ def perspective_jobqueue_update(params):
+ """Update job queue.
+
+ """
+ (file_name, content) = params
+
+ # Locking in exclusive, blocking mode because there could be several
+ # children running at the same time.
+ # TODO: Implement nonblocking locking with retries?
+ queue_lock.Exclusive(blocking=True)
+ try:
+ return backend.JobQueueUpdate(file_name, content)
+ finally:
+ queue_lock.Unlock()
+
+ @staticmethod
+ def perspective_jobqueue_purge(params):
+ """Purge job queue.
+
+ """
+ return backend.JobQueuePurge()
+
+
+class NodeDaemonHttpServer(http.HTTPServer):
+ def __init__(self, server_address):
+ http.HTTPServer.__init__(self, server_address, NodeDaemonRequestHandler)
+ self.noded_pid = os.getpid()
+
+ def serve_forever(self):
+ """Handle requests until told to quit."""
+ sighandler = utils.SignalHandler([signal.SIGINT, signal.SIGTERM])
+ try:
+ while not sighandler.called:
+ self.handle_request()
+ # TODO: There could be children running at this point
+ finally:
+ sighandler.Reset()
+
+
+class ForkingHTTPServer(SocketServer.ForkingMixIn, NodeDaemonHttpServer):
+ """Forking HTTP Server.
+
+ This inherits from ForkingMixIn and HTTPServer in order to fork for each
+ request we handle. This allows more requests to be handled concurrently.
+
+ """
+
def ParseOptions():
"""Parse the command line options.
"""Main function for the node daemon.
"""
+ global queue_lock
+
options, args = ParseOptions()
utils.debug = options.debug
for fname in (constants.SSL_CERT_FILE,):
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)
+ # create the various SUB_RUN_DIRS, if not existing, so that we handle the
+ # situation where RUN_DIR is tmpfs
+ for dir_name in constants.SUB_RUN_DIRS:
+ if not os.path.exists(dir_name):
+ try:
+ os.mkdir(dir_name, 0755)
+ except EnvironmentError, err:
+ if err.errno != errno.EEXIST:
+ print ("Node setup wrong, cannot create directory %s: %s" %
+ (dir_name, err))
+ sys.exit(5)
+ if not os.path.isdir(dir_name):
+ print ("Node setup wrong, %s is not a directory" % dir_name)
+ sys.exit(5)
# become a daemon
if options.fork:
- createDaemon()
-
- logger.SetupLogging(twisted_workaround=True, debug=options.debug,
- program="ganeti-noded")
+ utils.Daemonize(logfile=constants.LOG_NODESERVER)
- httpd = BaseHTTPServer.HTTPServer(('', port), ServerObject)
- httpd.serve_forever()
+ utils.WritePidFile(constants.NODED_PID)
+ logger.SetupLogging(logfile=constants.LOG_NODESERVER, debug=options.debug,
+ stderr_logging=not options.fork)
+ logging.info("ganeti node daemon startup")
-def createDaemon():
- """Detach a process from the controlling terminal and run it in the
- background as a daemon.
+ # Prepare job queue
+ queue_lock = jstore.InitAndVerifyQueue(must_lock=False)
- """
- 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
- 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
- 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.
+ if options.fork:
+ server = ForkingHTTPServer(('', port))
else:
- os._exit(0) # Exit parent of the first child.
- maxfd = resource.getrlimit(resource.RLIMIT_NOFILE)[1]
- if (maxfd == resource.RLIM_INFINITY):
- maxfd = MAXFD
+ server = NodeDaemonHttpServer(('', port))
- # 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)
+ try:
+ server.serve_forever()
+ finally:
+ utils.RemovePidFile(constants.NODED_PID)
if __name__ == '__main__':
projects / ganeti-local / blobdiff