#
#
-# Copyright (C) 2006, 2007 Google Inc.
+# Copyright (C) 2006, 2007, 2008, 2009, 2010 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
from ganeti import rpc
from ganeti import ssh
from ganeti import compat
+from ganeti import netutils
from optparse import (OptionParser, TitledHelpFormatter,
Option, OptionValueError)
"DISK_OPT",
"DISK_TEMPLATE_OPT",
"DRAINED_OPT",
+ "DRBD_HELPER_OPT",
"EARLY_RELEASE_OPT",
"ENABLED_HV_OPT",
"ERROR_CODES_OPT",
"HVOPTS_OPT",
"HYPERVISOR_OPT",
"IALLOCATOR_OPT",
+ "DEFAULT_IALLOCATOR_OPT",
"IDENTIFY_DEFAULTS_OPT",
"IGNORE_CONSIST_OPT",
"IGNORE_FAILURES_OPT",
"MAINTAIN_NODE_HEALTH_OPT",
"MASTER_NETDEV_OPT",
"MC_OPT",
+ "MIGRATION_MODE_OPT",
"NET_OPT",
"NEW_CLUSTER_CERT_OPT",
"NEW_CLUSTER_DOMAIN_SECRET_OPT",
"NIC_PARAMS_OPT",
"NODE_LIST_OPT",
"NODE_PLACEMENT_OPT",
+ "NODRBD_STORAGE_OPT",
"NOHDR_OPT",
"NOIPCHECK_OPT",
"NO_INSTALL_OPT",
"ON_PRIMARY_OPT",
"ON_SECONDARY_OPT",
"OFFLINE_OPT",
+ "OSPARAMS_OPT",
"OS_OPT",
"OS_SIZE_OPT",
+ "PRIMARY_IP_VERSION_OPT",
"RAPI_CERT_OPT",
"READD_OPT",
"REBOOT_TYPE_OPT",
"REMOVE_INSTANCE_OPT",
"REMOVE_UIDS_OPT",
+ "RESERVED_LVS_OPT",
"ROMAN_OPT",
"SECONDARY_IP_OPT",
"SELECT_OS_OPT",
"GenerateTable",
"AskUser",
"FormatTimestamp",
+ "FormatLogMessage",
# Tags functions
"ListTags",
"AddTags",
default=None, type="string",
completion_suggest=OPT_COMPL_ONE_IALLOCATOR)
+DEFAULT_IALLOCATOR_OPT = cli_option("-I", "--default-iallocator",
+ metavar="<NAME>",
+ help="Set the default instance allocator plugin",
+ default=None, type="string",
+ completion_suggest=OPT_COMPL_ONE_IALLOCATOR)
+
OS_OPT = cli_option("-o", "--os-type", dest="os", help="What OS to run",
metavar="<os>",
completion_suggest=OPT_COMPL_ONE_OS)
+OSPARAMS_OPT = cli_option("-O", "--os-parameters", dest="osparams",
+ type="keyval", default={},
+ help="OS parameters")
+
FORCE_VARIANT_OPT = cli_option("--force-variant", dest="force_variant",
action="store_true", default=False,
help="Force an unknown variant")
" freeze the instance, save the state, transfer and"
" only then resume running on the secondary node)")
+MIGRATION_MODE_OPT = cli_option("--migration-mode", dest="migration_mode",
+ default=None,
+ choices=list(constants.HT_MIGRATION_MODES),
+ help="Override default migration mode (choose"
+ " either live or non-live")
+
NODE_PLACEMENT_OPT = cli_option("-n", "--node", dest="node",
help="Target node and optional secondary node",
metavar="<pnode>[:<snode>]",
" ranges separated by commas, to be"
" removed from the user-id pool"))
+RESERVED_LVS_OPT = cli_option("--reserved-lvs", default=None,
+ action="store", dest="reserved_lvs",
+ help=("A comma-separated list of reserved"
+ " logical volumes names, that will be"
+ " ignored by cluster verify"))
+
ROMAN_OPT = cli_option("--roman",
dest="roman_integers", default=False,
action="store_true",
help="Use roman numbers for positive integers")
+DRBD_HELPER_OPT = cli_option("--drbd-usermode-helper", dest="drbd_helper",
+ action="store", default=None,
+ help="Specifies usermode helper for DRBD")
+
+NODRBD_STORAGE_OPT = cli_option("--no-drbd-storage", dest="drbd_storage",
+ action="store_false", default=True,
+ help="Disable support for DRBD")
+
+PRIMARY_IP_VERSION_OPT = \
+ cli_option("--primary-ip-version", default=constants.IP4_VERSION,
+ action="store", dest="primary_ip_version",
+ metavar="%d|%d" % (constants.IP4_VERSION,
+ constants.IP6_VERSION),
+ help="Cluster-wide IP version for primary IP")
def _ParseArgs(argv, commands, aliases):
binary = argv[0].split("/")[-1]
if len(argv) > 1 and argv[1] == "--version":
- ToStdout("%s (ganeti) %s", binary, constants.RELEASE_VERSION)
+ ToStdout("%s (ganeti %s) %s", binary, constants.VCS_VERSION,
+ constants.RELEASE_VERSION)
# Quit right away. That way we don't have to care about this special
# argument. optparse.py does it the same.
sys.exit(0)
return [os_name]
-def UsesRPC(fn):
- def wrapper(*args, **kwargs):
- rpc.Init()
- try:
- return fn(*args, **kwargs)
- finally:
- rpc.Shutdown()
- return wrapper
+UsesRPC = rpc.RunWithRPC
def AskUser(text, choices=None):
"""
ToStdout("%s %s", time.ctime(utils.MergeTime(timestamp)),
- utils.SafeEncode(log_msg))
+ FormatLogMessage(log_type, log_msg))
def ReportNotChanged(self, job_id, status):
"""Called if a job hasn't changed in a while.
self.notified_waitlock = True
-def PollJob(job_id, cl=None, feedback_fn=None):
+def FormatLogMessage(log_type, log_msg):
+ """Formats a job message according to its type.
+
+ """
+ if log_type != constants.ELOG_MESSAGE:
+ log_msg = str(log_msg)
+
+ return utils.SafeEncode(log_msg)
+
+
+def PollJob(job_id, cl=None, feedback_fn=None, reporter=None):
"""Function to poll for the result of a job.
@type job_id: job identified
if cl is None:
cl = GetClient()
- if feedback_fn:
- reporter = FeedbackFnJobPollReportCb(feedback_fn)
- else:
- reporter = StdioJobPollReportCb()
+ if reporter is None:
+ if feedback_fn:
+ reporter = FeedbackFnJobPollReportCb(feedback_fn)
+ else:
+ reporter = StdioJobPollReportCb()
+ elif feedback_fn:
+ raise errors.ProgrammerError("Can't specify reporter and feedback function")
return GenericPollJob(job_id, _LuxiJobPollCb(cl), reporter)
-def SubmitOpCode(op, cl=None, feedback_fn=None, opts=None):
+def SubmitOpCode(op, cl=None, feedback_fn=None, opts=None, reporter=None):
"""Legacy function to submit an opcode.
This is just a simple wrapper over the construction of the processor
SetGenericOpcodeOpts([op], opts)
- job_id = SendJob([op], cl)
+ job_id = SendJob([op], cl=cl)
- op_results = PollJob(job_id, cl=cl, feedback_fn=feedback_fn)
+ op_results = PollJob(job_id, cl=cl, feedback_fn=feedback_fn,
+ reporter=reporter)
return op_results[0]
elif isinstance(err, errors.HooksFailure):
obuf.write("Failure: hooks general failure: %s" % msg)
elif isinstance(err, errors.ResolverError):
- this_host = utils.HostInfo.SysName()
+ this_host = netutils.Hostname.GetSysName()
if err.args[0] == this_host:
msg = "Failure: can't resolve my own hostname ('%s')"
else:
elif isinstance(err, luxi.TimeoutError):
obuf.write("Timeout while talking to the master daemon. Error:\n"
"%s" % msg)
+ elif isinstance(err, luxi.PermissionError):
+ obuf.write("It seems you don't have permissions to connect to the"
+ " master daemon.\nPlease retry as a different user.")
elif isinstance(err, luxi.ProtocolError):
obuf.write("Unhandled protocol error while talking to the master daemon:\n"
"%s" % msg)
+ elif isinstance(err, errors.JobLost):
+ obuf.write("Error checking job status: %s" % msg)
elif isinstance(err, errors.GenericError):
obuf.write("Unhandled Ganeti error: %s" % msg)
elif isinstance(err, JobSubmittedException):
if mode == constants.INSTANCE_CREATE:
start = opts.start
os_type = opts.os
+ force_variant = opts.force_variant
src_node = None
src_path = None
no_install = opts.no_install
elif mode == constants.INSTANCE_IMPORT:
start = False
os_type = None
+ force_variant = False
src_node = opts.src_node
src_path = opts.src_dir
no_install = None
hypervisor=hypervisor,
hvparams=hvparams,
beparams=opts.beparams,
+ osparams=opts.osparams,
mode=mode,
start=start,
os_type=os_type,
+ force_variant=force_variant,
src_node=src_node,
src_path=src_path,
no_install=no_install,
if separator is None:
mlens = [0 for name in fields]
- format = ' '.join(format_fields)
+ format_str = ' '.join(format_fields)
else:
- format = separator.replace("%", "%%").join(format_fields)
+ format_str = separator.replace("%", "%%").join(format_fields)
for row in data:
if row is None:
mlens[idx] = max(mlens[idx], len(hdr))
args.append(mlens[idx])
args.append(hdr)
- result.append(format % tuple(args))
+ result.append(format_str % tuple(args))
if separator is None:
assert len(mlens) == len(fields)
if separator is None:
args.append(mlens[idx])
args.append(line[idx])
- result.append(format % tuple(args))
+ result.append(format_str % tuple(args))
return result
SetGenericOpcodeOpts(ops, self.opts)
self.queue.append((name, ops))
- def SubmitPending(self):
+ def SubmitPending(self, each=False):
"""Submit all pending jobs.
"""
- results = self.cl.SubmitManyJobs([row[1] for row in self.queue])
+ if each:
+ results = []
+ for row in self.queue:
+ # SubmitJob will remove the success status, but raise an exception if
+ # the submission fails, so we'll notice that anyway.
+ results.append([True, self.cl.SubmitJob(row[1])])
+ else:
+ results = self.cl.SubmitManyJobs([row[1] for row in self.queue])
for (idx, ((status, data), (name, _))) in enumerate(zip(results,
self.queue)):
self.jobs.append((idx, status, data, name))
assert result
for job_data, status in zip(self.jobs, result):
- if status[0] in (constants.JOB_STATUS_QUEUED,
- constants.JOB_STATUS_WAITLOCK,
- constants.JOB_STATUS_CANCELING):
- # job is still waiting
+ if (isinstance(status, list) and status and
+ status[0] in (constants.JOB_STATUS_QUEUED,
+ constants.JOB_STATUS_WAITLOCK,
+ constants.JOB_STATUS_CANCELING)):
+ # job is still present and waiting
continue
- # good candidate found
+ # good candidate found (either running job or lost job)
self.jobs.remove(job_data)
return job_data
try:
job_result = PollJob(jid, cl=self.cl, feedback_fn=self.feedback_fn)
success = True
+ except errors.JobLost, err:
+ _, job_result = FormatError(err)
+ ToStderr("Job %s for %s has been archived, cannot check its result",
+ jid, name)
+ success = False
except (errors.GenericError, luxi.ProtocolError), err:
_, job_result = FormatError(err)
success = False
ToStdout("%s: %s", result, name)
else:
ToStderr("Failure for %s: %s", name, result)
+ return [row[1:3] for row in self.jobs]
projects / ganeti-local / blobdiff