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)
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):
"""
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
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):
"""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,
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,