code.grnet.gr Git - ganeti-local/blob - lib/cli.py

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007 Google Inc.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Module dealing with command line parsing"""
  23
  24
  25 import sys
  26 import textwrap
  27 import os.path
  28 import time
  29 import logging
  30 from cStringIO import StringIO
  31
  32 from ganeti import utils
  33 from ganeti import errors
  34 from ganeti import constants
  35 from ganeti import opcodes
  36 from ganeti import luxi
  37 from ganeti import ssconf
  38 from ganeti import rpc
  39 from ganeti import ssh
  40
  41 from optparse import (OptionParser, TitledHelpFormatter,
  42                       Option, OptionValueError)
  43
  44
  45 __all__ = [
  46   # Command line options
  47   "ALLOCATABLE_OPT",
  48   "ALL_OPT",
  49   "AUTO_PROMOTE_OPT",
  50   "AUTO_REPLACE_OPT",
  51   "BACKEND_OPT",
  52   "CLEANUP_OPT",
  53   "CLUSTER_DOMAIN_SECRET_OPT",
  54   "CONFIRM_OPT",
  55   "CP_SIZE_OPT",
  56   "DEBUG_OPT",
  57   "DEBUG_SIMERR_OPT",
  58   "DISKIDX_OPT",
  59   "DISK_OPT",
  60   "DISK_TEMPLATE_OPT",
  61   "DRAINED_OPT",
  62   "EARLY_RELEASE_OPT",
  63   "ENABLED_HV_OPT",
  64   "ERROR_CODES_OPT",
  65   "FIELDS_OPT",
  66   "FILESTORE_DIR_OPT",
  67   "FILESTORE_DRIVER_OPT",
  68   "FORCE_OPT",
  69   "FORCE_VARIANT_OPT",
  70   "GLOBAL_FILEDIR_OPT",
  71   "HVLIST_OPT",
  72   "HVOPTS_OPT",
  73   "HYPERVISOR_OPT",
  74   "IALLOCATOR_OPT",
  75   "IGNORE_CONSIST_OPT",
  76   "IGNORE_FAILURES_OPT",
  77   "IGNORE_REMOVE_FAILURES_OPT",
  78   "IGNORE_SECONDARIES_OPT",
  79   "IGNORE_SIZE_OPT",
  80   "MAC_PREFIX_OPT",
  81   "MASTER_NETDEV_OPT",
  82   "MC_OPT",
  83   "NET_OPT",
  84   "NEW_CLUSTER_CERT_OPT",
  85   "NEW_CLUSTER_DOMAIN_SECRET_OPT",
  86   "NEW_CONFD_HMAC_KEY_OPT",
  87   "NEW_RAPI_CERT_OPT",
  88   "NEW_SECONDARY_OPT",
  89   "NIC_PARAMS_OPT",
  90   "NODE_LIST_OPT",
  91   "NODE_PLACEMENT_OPT",
  92   "NOHDR_OPT",
  93   "NOIPCHECK_OPT",
  94   "NO_INSTALL_OPT",
  95   "NONAMECHECK_OPT",
  96   "NOLVM_STORAGE_OPT",
  97   "NOMODIFY_ETCHOSTS_OPT",
  98   "NOMODIFY_SSH_SETUP_OPT",
  99   "NONICS_OPT",
 100   "NONLIVE_OPT",
 101   "NONPLUS1_OPT",
 102   "NOSHUTDOWN_OPT",
 103   "NOSTART_OPT",
 104   "NOSSH_KEYCHECK_OPT",
 105   "NOVOTING_OPT",
 106   "NWSYNC_OPT",
 107   "ON_PRIMARY_OPT",
 108   "ON_SECONDARY_OPT",
 109   "OFFLINE_OPT",
 110   "OS_OPT",
 111   "OS_SIZE_OPT",
 112   "RAPI_CERT_OPT",
 113   "READD_OPT",
 114   "REBOOT_TYPE_OPT",
 115   "REMOVE_INSTANCE_OPT",
 116   "SECONDARY_IP_OPT",
 117   "SELECT_OS_OPT",
 118   "SEP_OPT",
 119   "SHOWCMD_OPT",
 120   "SHUTDOWN_TIMEOUT_OPT",
 121   "SINGLE_NODE_OPT",
 122   "SRC_DIR_OPT",
 123   "SRC_NODE_OPT",
 124   "SUBMIT_OPT",
 125   "STATIC_OPT",
 126   "SYNC_OPT",
 127   "TAG_SRC_OPT",
 128   "TIMEOUT_OPT",
 129   "USEUNITS_OPT",
 130   "USE_REPL_NET_OPT",
 131   "VERBOSE_OPT",
 132   "VG_NAME_OPT",
 133   "YES_DOIT_OPT",
 134   # Generic functions for CLI programs
 135   "GenericMain",
 136   "GenericInstanceCreate",
 137   "GetClient",
 138   "GetOnlineNodes",
 139   "JobExecutor",
 140   "JobSubmittedException",
 141   "ParseTimespec",
 142   "RunWhileClusterStopped",
 143   "SubmitOpCode",
 144   "SubmitOrSend",
 145   "UsesRPC",
 146   # Formatting functions
 147   "ToStderr", "ToStdout",
 148   "FormatError",
 149   "GenerateTable",
 150   "AskUser",
 151   "FormatTimestamp",
 152   # Tags functions
 153   "ListTags",
 154   "AddTags",
 155   "RemoveTags",
 156   # command line options support infrastructure
 157   "ARGS_MANY_INSTANCES",
 158   "ARGS_MANY_NODES",
 159   "ARGS_NONE",
 160   "ARGS_ONE_INSTANCE",
 161   "ARGS_ONE_NODE",
 162   "ARGS_ONE_OS",
 163   "ArgChoice",
 164   "ArgCommand",
 165   "ArgFile",
 166   "ArgHost",
 167   "ArgInstance",
 168   "ArgJobId",
 169   "ArgNode",
 170   "ArgOs",
 171   "ArgSuggest",
 172   "ArgUnknown",
 173   "OPT_COMPL_INST_ADD_NODES",
 174   "OPT_COMPL_MANY_NODES",
 175   "OPT_COMPL_ONE_IALLOCATOR",
 176   "OPT_COMPL_ONE_INSTANCE",
 177   "OPT_COMPL_ONE_NODE",
 178   "OPT_COMPL_ONE_OS",
 179   "cli_option",
 180   "SplitNodeOption",
 181   "CalculateOSNames",
 182   ]
 183
 184 NO_PREFIX = "no_"
 185 UN_PREFIX = "-"
 186
 187
 188 class _Argument:
 189   def __init__(self, min=0, max=None): # pylint: disable-msg=W0622
 190     self.min = min
 191     self.max = max
 192
 193   def __repr__(self):
 194     return ("<%s min=%s max=%s>" %
 195             (self.__class__.__name__, self.min, self.max))
 196
 197
 198 class ArgSuggest(_Argument):
 199   """Suggesting argument.
 200
 201   Value can be any of the ones passed to the constructor.
 202
 203   """
 204   # pylint: disable-msg=W0622
 205   def __init__(self, min=0, max=None, choices=None):
 206     _Argument.__init__(self, min=min, max=max)
 207     self.choices = choices
 208
 209   def __repr__(self):
 210     return ("<%s min=%s max=%s choices=%r>" %
 211             (self.__class__.__name__, self.min, self.max, self.choices))
 212
 213
 214 class ArgChoice(ArgSuggest):
 215   """Choice argument.
 216
 217   Value can be any of the ones passed to the constructor. Like L{ArgSuggest},
 218   but value must be one of the choices.
 219
 220   """
 221
 222
 223 class ArgUnknown(_Argument):
 224   """Unknown argument to program (e.g. determined at runtime).
 225
 226   """
 227
 228
 229 class ArgInstance(_Argument):
 230   """Instances argument.
 231
 232   """
 233
 234
 235 class ArgNode(_Argument):
 236   """Node argument.
 237
 238   """
 239
 240 class ArgJobId(_Argument):
 241   """Job ID argument.
 242
 243   """
 244
 245
 246 class ArgFile(_Argument):
 247   """File path argument.
 248
 249   """
 250
 251
 252 class ArgCommand(_Argument):
 253   """Command argument.
 254
 255   """
 256
 257
 258 class ArgHost(_Argument):
 259   """Host argument.
 260
 261   """
 262
 263
 264 class ArgOs(_Argument):
 265   """OS argument.
 266
 267   """
 268
 269
 270 ARGS_NONE = []
 271 ARGS_MANY_INSTANCES = [ArgInstance()]
 272 ARGS_MANY_NODES = [ArgNode()]
 273 ARGS_ONE_INSTANCE = [ArgInstance(min=1, max=1)]
 274 ARGS_ONE_NODE = [ArgNode(min=1, max=1)]
 275 ARGS_ONE_OS = [ArgOs(min=1, max=1)]
 276
 277
 278 def _ExtractTagsObject(opts, args):
 279   """Extract the tag type object.
 280
 281   Note that this function will modify its args parameter.
 282
 283   """
 284   if not hasattr(opts, "tag_type"):
 285     raise errors.ProgrammerError("tag_type not passed to _ExtractTagsObject")
 286   kind = opts.tag_type
 287   if kind == constants.TAG_CLUSTER:
 288     retval = kind, kind
 289   elif kind == constants.TAG_NODE or kind == constants.TAG_INSTANCE:
 290     if not args:
 291       raise errors.OpPrereqError("no arguments passed to the command")
 292     name = args.pop(0)
 293     retval = kind, name
 294   else:
 295     raise errors.ProgrammerError("Unhandled tag type '%s'" % kind)
 296   return retval
 297
 298
 299 def _ExtendTags(opts, args):
 300   """Extend the args if a source file has been given.
 301
 302   This function will extend the tags with the contents of the file
 303   passed in the 'tags_source' attribute of the opts parameter. A file
 304   named '-' will be replaced by stdin.
 305
 306   """
 307   fname = opts.tags_source
 308   if fname is None:
 309     return
 310   if fname == "-":
 311     new_fh = sys.stdin
 312   else:
 313     new_fh = open(fname, "r")
 314   new_data = []
 315   try:
 316     # we don't use the nice 'new_data = [line.strip() for line in fh]'
 317     # because of python bug 1633941
 318     while True:
 319       line = new_fh.readline()
 320       if not line:
 321         break
 322       new_data.append(line.strip())
 323   finally:
 324     new_fh.close()
 325   args.extend(new_data)
 326
 327
 328 def ListTags(opts, args):
 329   """List the tags on a given object.
 330
 331   This is a generic implementation that knows how to deal with all
 332   three cases of tag objects (cluster, node, instance). The opts
 333   argument is expected to contain a tag_type field denoting what
 334   object type we work on.
 335
 336   """
 337   kind, name = _ExtractTagsObject(opts, args)
 338   cl = GetClient()
 339   result = cl.QueryTags(kind, name)
 340   result = list(result)
 341   result.sort()
 342   for tag in result:
 343     ToStdout(tag)
 344
 345
 346 def AddTags(opts, args):
 347   """Add tags on a given object.
 348
 349   This is a generic implementation that knows how to deal with all
 350   three cases of tag objects (cluster, node, instance). The opts
 351   argument is expected to contain a tag_type field denoting what
 352   object type we work on.
 353
 354   """
 355   kind, name = _ExtractTagsObject(opts, args)
 356   _ExtendTags(opts, args)
 357   if not args:
 358     raise errors.OpPrereqError("No tags to be added")
 359   op = opcodes.OpAddTags(kind=kind, name=name, tags=args)
 360   SubmitOpCode(op)
 361
 362
 363 def RemoveTags(opts, args):
 364   """Remove tags from a given object.
 365
 366   This is a generic implementation that knows how to deal with all
 367   three cases of tag objects (cluster, node, instance). The opts
 368   argument is expected to contain a tag_type field denoting what
 369   object type we work on.
 370
 371   """
 372   kind, name = _ExtractTagsObject(opts, args)
 373   _ExtendTags(opts, args)
 374   if not args:
 375     raise errors.OpPrereqError("No tags to be removed")
 376   op = opcodes.OpDelTags(kind=kind, name=name, tags=args)
 377   SubmitOpCode(op)
 378
 379
 380 def check_unit(option, opt, value): # pylint: disable-msg=W0613
 381   """OptParsers custom converter for units.
 382
 383   """
 384   try:
 385     return utils.ParseUnit(value)
 386   except errors.UnitParseError, err:
 387     raise OptionValueError("option %s: %s" % (opt, err))
 388
 389
 390 def _SplitKeyVal(opt, data):
 391   """Convert a KeyVal string into a dict.
 392
 393   This function will convert a key=val[,...] string into a dict. Empty
 394   values will be converted specially: keys which have the prefix 'no_'
 395   will have the value=False and the prefix stripped, the others will
 396   have value=True.
 397
 398   @type opt: string
 399   @param opt: a string holding the option name for which we process the
 400       data, used in building error messages
 401   @type data: string
 402   @param data: a string of the format key=val,key=val,...
 403   @rtype: dict
 404   @return: {key=val, key=val}
 405   @raises errors.ParameterError: if there are duplicate keys
 406
 407   """
 408   kv_dict = {}
 409   if data:
 410     for elem in utils.UnescapeAndSplit(data, sep=","):
 411       if "=" in elem:
 412         key, val = elem.split("=", 1)
 413       else:
 414         if elem.startswith(NO_PREFIX):
 415           key, val = elem[len(NO_PREFIX):], False
 416         elif elem.startswith(UN_PREFIX):
 417           key, val = elem[len(UN_PREFIX):], None
 418         else:
 419           key, val = elem, True
 420       if key in kv_dict:
 421         raise errors.ParameterError("Duplicate key '%s' in option %s" %
 422                                     (key, opt))
 423       kv_dict[key] = val
 424   return kv_dict
 425
 426
 427 def check_ident_key_val(option, opt, value):  # pylint: disable-msg=W0613
 428   """Custom parser for ident:key=val,key=val options.
 429
 430   This will store the parsed values as a tuple (ident, {key: val}). As such,
 431   multiple uses of this option via action=append is possible.
 432
 433   """
 434   if ":" not in value:
 435     ident, rest = value, ''
 436   else:
 437     ident, rest = value.split(":", 1)
 438
 439   if ident.startswith(NO_PREFIX):
 440     if rest:
 441       msg = "Cannot pass options when removing parameter groups: %s" % value
 442       raise errors.ParameterError(msg)
 443     retval = (ident[len(NO_PREFIX):], False)
 444   elif ident.startswith(UN_PREFIX):
 445     if rest:
 446       msg = "Cannot pass options when removing parameter groups: %s" % value
 447       raise errors.ParameterError(msg)
 448     retval = (ident[len(UN_PREFIX):], None)
 449   else:
 450     kv_dict = _SplitKeyVal(opt, rest)
 451     retval = (ident, kv_dict)
 452   return retval
 453
 454
 455 def check_key_val(option, opt, value):  # pylint: disable-msg=W0613
 456   """Custom parser class for key=val,key=val options.
 457
 458   This will store the parsed values as a dict {key: val}.
 459
 460   """
 461   return _SplitKeyVal(opt, value)
 462
 463
 464 def check_bool(option, opt, value): # pylint: disable-msg=W0613
 465   """Custom parser for yes/no options.
 466
 467   This will store the parsed value as either True or False.
 468
 469   """
 470   value = value.lower()
 471   if value == constants.VALUE_FALSE or value == "no":
 472     return False
 473   elif value == constants.VALUE_TRUE or value == "yes":
 474     return True
 475   else:
 476     raise errors.ParameterError("Invalid boolean value '%s'" % value)
 477
 478
 479 # completion_suggestion is normally a list. Using numeric values not evaluating
 480 # to False for dynamic completion.
 481 (OPT_COMPL_MANY_NODES,
 482  OPT_COMPL_ONE_NODE,
 483  OPT_COMPL_ONE_INSTANCE,
 484  OPT_COMPL_ONE_OS,
 485  OPT_COMPL_ONE_IALLOCATOR,
 486  OPT_COMPL_INST_ADD_NODES) = range(100, 106)
 487
 488 OPT_COMPL_ALL = frozenset([
 489   OPT_COMPL_MANY_NODES,
 490   OPT_COMPL_ONE_NODE,
 491   OPT_COMPL_ONE_INSTANCE,
 492   OPT_COMPL_ONE_OS,
 493   OPT_COMPL_ONE_IALLOCATOR,
 494   OPT_COMPL_INST_ADD_NODES,
 495   ])
 496
 497
 498 class CliOption(Option):
 499   """Custom option class for optparse.
 500
 501   """
 502   ATTRS = Option.ATTRS + [
 503     "completion_suggest",
 504     ]
 505   TYPES = Option.TYPES + (
 506     "identkeyval",
 507     "keyval",
 508     "unit",
 509     "bool",
 510     )
 511   TYPE_CHECKER = Option.TYPE_CHECKER.copy()
 512   TYPE_CHECKER["identkeyval"] = check_ident_key_val
 513   TYPE_CHECKER["keyval"] = check_key_val
 514   TYPE_CHECKER["unit"] = check_unit
 515   TYPE_CHECKER["bool"] = check_bool
 516
 517
 518 # optparse.py sets make_option, so we do it for our own option class, too
 519 cli_option = CliOption
 520
 521
 522 _YORNO = "yes|no"
 523
 524 DEBUG_OPT = cli_option("-d", "--debug", default=0, action="count",
 525                        help="Increase debugging level")
 526
 527 NOHDR_OPT = cli_option("--no-headers", default=False,
 528                        action="store_true", dest="no_headers",
 529                        help="Don't display column headers")
 530
 531 SEP_OPT = cli_option("--separator", default=None,
 532                      action="store", dest="separator",
 533                      help=("Separator between output fields"
 534                            " (defaults to one space)"))
 535
 536 USEUNITS_OPT = cli_option("--units", default=None,
 537                           dest="units", choices=('h', 'm', 'g', 't'),
 538                           help="Specify units for output (one of hmgt)")
 539
 540 FIELDS_OPT = cli_option("-o", "--output", dest="output", action="store",
 541                         type="string", metavar="FIELDS",
 542                         help="Comma separated list of output fields")
 543
 544 FORCE_OPT = cli_option("-f", "--force", dest="force", action="store_true",
 545                        default=False, help="Force the operation")
 546
 547 CONFIRM_OPT = cli_option("--yes", dest="confirm", action="store_true",
 548                          default=False, help="Do not require confirmation")
 549
 550 TAG_SRC_OPT = cli_option("--from", dest="tags_source",
 551                          default=None, help="File with tag names")
 552
 553 SUBMIT_OPT = cli_option("--submit", dest="submit_only",
 554                         default=False, action="store_true",
 555                         help=("Submit the job and return the job ID, but"
 556                               " don't wait for the job to finish"))
 557
 558 SYNC_OPT = cli_option("--sync", dest="do_locking",
 559                       default=False, action="store_true",
 560                       help=("Grab locks while doing the queries"
 561                             " in order to ensure more consistent results"))
 562
 563 _DRY_RUN_OPT = cli_option("--dry-run", default=False,
 564                           action="store_true",
 565                           help=("Do not execute the operation, just run the"
 566                                 " check steps and verify it it could be"
 567                                 " executed"))
 568
 569 VERBOSE_OPT = cli_option("-v", "--verbose", default=False,
 570                          action="store_true",
 571                          help="Increase the verbosity of the operation")
 572
 573 DEBUG_SIMERR_OPT = cli_option("--debug-simulate-errors", default=False,
 574                               action="store_true", dest="simulate_errors",
 575                               help="Debugging option that makes the operation"
 576                               " treat most runtime checks as failed")
 577
 578 NWSYNC_OPT = cli_option("--no-wait-for-sync", dest="wait_for_sync",
 579                         default=True, action="store_false",
 580                         help="Don't wait for sync (DANGEROUS!)")
 581
 582 DISK_TEMPLATE_OPT = cli_option("-t", "--disk-template", dest="disk_template",
 583                                help="Custom disk setup (diskless, file,"
 584                                " plain or drbd)",
 585                                default=None, metavar="TEMPL",
 586                                choices=list(constants.DISK_TEMPLATES))
 587
 588 NONICS_OPT = cli_option("--no-nics", default=False, action="store_true",
 589                         help="Do not create any network cards for"
 590                         " the instance")
 591
 592 FILESTORE_DIR_OPT = cli_option("--file-storage-dir", dest="file_storage_dir",
 593                                help="Relative path under default cluster-wide"
 594                                " file storage dir to store file-based disks",
 595                                default=None, metavar="<DIR>")
 596
 597 FILESTORE_DRIVER_OPT = cli_option("--file-driver", dest="file_driver",
 598                                   help="Driver to use for image files",
 599                                   default="loop", metavar="<DRIVER>",
 600                                   choices=list(constants.FILE_DRIVER))
 601
 602 IALLOCATOR_OPT = cli_option("-I", "--iallocator", metavar="<NAME>",
 603                             help="Select nodes for the instance automatically"
 604                             " using the <NAME> iallocator plugin",
 605                             default=None, type="string",
 606                             completion_suggest=OPT_COMPL_ONE_IALLOCATOR)
 607
 608 OS_OPT = cli_option("-o", "--os-type", dest="os", help="What OS to run",
 609                     metavar="<os>",
 610                     completion_suggest=OPT_COMPL_ONE_OS)
 611
 612 FORCE_VARIANT_OPT = cli_option("--force-variant", dest="force_variant",
 613                                action="store_true", default=False,
 614                                help="Force an unknown variant")
 615
 616 NO_INSTALL_OPT = cli_option("--no-install", dest="no_install",
 617                             action="store_true", default=False,
 618                             help="Do not install the OS (will"
 619                             " enable no-start)")
 620
 621 BACKEND_OPT = cli_option("-B", "--backend-parameters", dest="beparams",
 622                          type="keyval", default={},
 623                          help="Backend parameters")
 624
 625 HVOPTS_OPT =  cli_option("-H", "--hypervisor-parameters", type="keyval",
 626                          default={}, dest="hvparams",
 627                          help="Hypervisor parameters")
 628
 629 HYPERVISOR_OPT = cli_option("-H", "--hypervisor-parameters", dest="hypervisor",
 630                             help="Hypervisor and hypervisor options, in the"
 631                             " format hypervisor:option=value,option=value,...",
 632                             default=None, type="identkeyval")
 633
 634 HVLIST_OPT = cli_option("-H", "--hypervisor-parameters", dest="hvparams",
 635                         help="Hypervisor and hypervisor options, in the"
 636                         " format hypervisor:option=value,option=value,...",
 637                         default=[], action="append", type="identkeyval")
 638
 639 NOIPCHECK_OPT = cli_option("--no-ip-check", dest="ip_check", default=True,
 640                            action="store_false",
 641                            help="Don't check that the instance's IP"
 642                            " is alive")
 643
 644 NONAMECHECK_OPT = cli_option("--no-name-check", dest="name_check",
 645                              default=True, action="store_false",
 646                              help="Don't check that the instance's name"
 647                              " is resolvable")
 648
 649 NET_OPT = cli_option("--net",
 650                      help="NIC parameters", default=[],
 651                      dest="nics", action="append", type="identkeyval")
 652
 653 DISK_OPT = cli_option("--disk", help="Disk parameters", default=[],
 654                       dest="disks", action="append", type="identkeyval")
 655
 656 DISKIDX_OPT = cli_option("--disks", dest="disks", default=None,
 657                          help="Comma-separated list of disks"
 658                          " indices to act on (e.g. 0,2) (optional,"
 659                          " defaults to all disks)")
 660
 661 OS_SIZE_OPT = cli_option("-s", "--os-size", dest="sd_size",
 662                          help="Enforces a single-disk configuration using the"
 663                          " given disk size, in MiB unless a suffix is used",
 664                          default=None, type="unit", metavar="<size>")
 665
 666 IGNORE_CONSIST_OPT = cli_option("--ignore-consistency",
 667                                 dest="ignore_consistency",
 668                                 action="store_true", default=False,
 669                                 help="Ignore the consistency of the disks on"
 670                                 " the secondary")
 671
 672 NONLIVE_OPT = cli_option("--non-live", dest="live",
 673                          default=True, action="store_false",
 674                          help="Do a non-live migration (this usually means"
 675                          " freeze the instance, save the state, transfer and"
 676                          " only then resume running on the secondary node)")
 677
 678 NODE_PLACEMENT_OPT = cli_option("-n", "--node", dest="node",
 679                                 help="Target node and optional secondary node",
 680                                 metavar="<pnode>[:<snode>]",
 681                                 completion_suggest=OPT_COMPL_INST_ADD_NODES)
 682
 683 NODE_LIST_OPT = cli_option("-n", "--node", dest="nodes", default=[],
 684                            action="append", metavar="<node>",
 685                            help="Use only this node (can be used multiple"
 686                            " times, if not given defaults to all nodes)",
 687                            completion_suggest=OPT_COMPL_ONE_NODE)
 688
 689 SINGLE_NODE_OPT = cli_option("-n", "--node", dest="node", help="Target node",
 690                              metavar="<node>",
 691                              completion_suggest=OPT_COMPL_ONE_NODE)
 692
 693 NOSTART_OPT = cli_option("--no-start", dest="start", default=True,
 694                          action="store_false",
 695                          help="Don't start the instance after creation")
 696
 697 SHOWCMD_OPT = cli_option("--show-cmd", dest="show_command",
 698                          action="store_true", default=False,
 699                          help="Show command instead of executing it")
 700
 701 CLEANUP_OPT = cli_option("--cleanup", dest="cleanup",
 702                          default=False, action="store_true",
 703                          help="Instead of performing the migration, try to"
 704                          " recover from a failed cleanup. This is safe"
 705                          " to run even if the instance is healthy, but it"
 706                          " will create extra replication traffic and "
 707                          " disrupt briefly the replication (like during the"
 708                          " migration")
 709
 710 STATIC_OPT = cli_option("-s", "--static", dest="static",
 711                         action="store_true", default=False,
 712                         help="Only show configuration data, not runtime data")
 713
 714 ALL_OPT = cli_option("--all", dest="show_all",
 715                      default=False, action="store_true",
 716                      help="Show info on all instances on the cluster."
 717                      " This can take a long time to run, use wisely")
 718
 719 SELECT_OS_OPT = cli_option("--select-os", dest="select_os",
 720                            action="store_true", default=False,
 721                            help="Interactive OS reinstall, lists available"
 722                            " OS templates for selection")
 723
 724 IGNORE_FAILURES_OPT = cli_option("--ignore-failures", dest="ignore_failures",
 725                                  action="store_true", default=False,
 726                                  help="Remove the instance from the cluster"
 727                                  " configuration even if there are failures"
 728                                  " during the removal process")
 729
 730 IGNORE_REMOVE_FAILURES_OPT = cli_option("--ignore-remove-failures",
 731                                         dest="ignore_remove_failures",
 732                                         action="store_true", default=False,
 733                                         help="Remove the instance from the"
 734                                         " cluster configuration even if there"
 735                                         " are failures during the removal"
 736                                         " process")
 737
 738 REMOVE_INSTANCE_OPT = cli_option("--remove-instance", dest="remove_instance",
 739                                  action="store_true", default=False,
 740                                  help="Remove the instance from the cluster")
 741
 742 NEW_SECONDARY_OPT = cli_option("-n", "--new-secondary", dest="dst_node",
 743                                help="Specifies the new secondary node",
 744                                metavar="NODE", default=None,
 745                                completion_suggest=OPT_COMPL_ONE_NODE)
 746
 747 ON_PRIMARY_OPT = cli_option("-p", "--on-primary", dest="on_primary",
 748                             default=False, action="store_true",
 749                             help="Replace the disk(s) on the primary"
 750                             " node (only for the drbd template)")
 751
 752 ON_SECONDARY_OPT = cli_option("-s", "--on-secondary", dest="on_secondary",
 753                               default=False, action="store_true",
 754                               help="Replace the disk(s) on the secondary"
 755                               " node (only for the drbd template)")
 756
 757 AUTO_PROMOTE_OPT = cli_option("--auto-promote", dest="auto_promote",
 758                               default=False, action="store_true",
 759                               help="Lock all nodes and auto-promote as needed"
 760                               " to MC status")
 761
 762 AUTO_REPLACE_OPT = cli_option("-a", "--auto", dest="auto",
 763                               default=False, action="store_true",
 764                               help="Automatically replace faulty disks"
 765                               " (only for the drbd template)")
 766
 767 IGNORE_SIZE_OPT = cli_option("--ignore-size", dest="ignore_size",
 768                              default=False, action="store_true",
 769                              help="Ignore current recorded size"
 770                              " (useful for forcing activation when"
 771                              " the recorded size is wrong)")
 772
 773 SRC_NODE_OPT = cli_option("--src-node", dest="src_node", help="Source node",
 774                           metavar="<node>",
 775                           completion_suggest=OPT_COMPL_ONE_NODE)
 776
 777 SRC_DIR_OPT = cli_option("--src-dir", dest="src_dir", help="Source directory",
 778                          metavar="<dir>")
 779
 780 SECONDARY_IP_OPT = cli_option("-s", "--secondary-ip", dest="secondary_ip",
 781                               help="Specify the secondary ip for the node",
 782                               metavar="ADDRESS", default=None)
 783
 784 READD_OPT = cli_option("--readd", dest="readd",
 785                        default=False, action="store_true",
 786                        help="Readd old node after replacing it")
 787
 788 NOSSH_KEYCHECK_OPT = cli_option("--no-ssh-key-check", dest="ssh_key_check",
 789                                 default=True, action="store_false",
 790                                 help="Disable SSH key fingerprint checking")
 791
 792
 793 MC_OPT = cli_option("-C", "--master-candidate", dest="master_candidate",
 794                     type="bool", default=None, metavar=_YORNO,
 795                     help="Set the master_candidate flag on the node")
 796
 797 OFFLINE_OPT = cli_option("-O", "--offline", dest="offline", metavar=_YORNO,
 798                          type="bool", default=None,
 799                          help="Set the offline flag on the node")
 800
 801 DRAINED_OPT = cli_option("-D", "--drained", dest="drained", metavar=_YORNO,
 802                          type="bool", default=None,
 803                          help="Set the drained flag on the node")
 804
 805 ALLOCATABLE_OPT = cli_option("--allocatable", dest="allocatable",
 806                              type="bool", default=None, metavar=_YORNO,
 807                              help="Set the allocatable flag on a volume")
 808
 809 NOLVM_STORAGE_OPT = cli_option("--no-lvm-storage", dest="lvm_storage",
 810                                help="Disable support for lvm based instances"
 811                                " (cluster-wide)",
 812                                action="store_false", default=True)
 813
 814 ENABLED_HV_OPT = cli_option("--enabled-hypervisors",
 815                             dest="enabled_hypervisors",
 816                             help="Comma-separated list of hypervisors",
 817                             type="string", default=None)
 818
 819 NIC_PARAMS_OPT = cli_option("-N", "--nic-parameters", dest="nicparams",
 820                             type="keyval", default={},
 821                             help="NIC parameters")
 822
 823 CP_SIZE_OPT = cli_option("-C", "--candidate-pool-size", default=None,
 824                          dest="candidate_pool_size", type="int",
 825                          help="Set the candidate pool size")
 826
 827 VG_NAME_OPT = cli_option("-g", "--vg-name", dest="vg_name",
 828                          help="Enables LVM and specifies the volume group"
 829                          " name (cluster-wide) for disk allocation [xenvg]",
 830                          metavar="VG", default=None)
 831
 832 YES_DOIT_OPT = cli_option("--yes-do-it", dest="yes_do_it",
 833                           help="Destroy cluster", action="store_true")
 834
 835 NOVOTING_OPT = cli_option("--no-voting", dest="no_voting",
 836                           help="Skip node agreement check (dangerous)",
 837                           action="store_true", default=False)
 838
 839 MAC_PREFIX_OPT = cli_option("-m", "--mac-prefix", dest="mac_prefix",
 840                             help="Specify the mac prefix for the instance IP"
 841                             " addresses, in the format XX:XX:XX",
 842                             metavar="PREFIX",
 843                             default=None)
 844
 845 MASTER_NETDEV_OPT = cli_option("--master-netdev", dest="master_netdev",
 846                                help="Specify the node interface (cluster-wide)"
 847                                " on which the master IP address will be added "
 848                                " [%s]" % constants.DEFAULT_BRIDGE,
 849                                metavar="NETDEV",
 850                                default=constants.DEFAULT_BRIDGE)
 851
 852 GLOBAL_FILEDIR_OPT = cli_option("--file-storage-dir", dest="file_storage_dir",
 853                                 help="Specify the default directory (cluster-"
 854                                 "wide) for storing the file-based disks [%s]" %
 855                                 constants.DEFAULT_FILE_STORAGE_DIR,
 856                                 metavar="DIR",
 857                                 default=constants.DEFAULT_FILE_STORAGE_DIR)
 858
 859 NOMODIFY_ETCHOSTS_OPT = cli_option("--no-etc-hosts", dest="modify_etc_hosts",
 860                                    help="Don't modify /etc/hosts",
 861                                    action="store_false", default=True)
 862
 863 NOMODIFY_SSH_SETUP_OPT = cli_option("--no-ssh-init", dest="modify_ssh_setup",
 864                                     help="Don't initialize SSH keys",
 865                                     action="store_false", default=True)
 866
 867 ERROR_CODES_OPT = cli_option("--error-codes", dest="error_codes",
 868                              help="Enable parseable error messages",
 869                              action="store_true", default=False)
 870
 871 NONPLUS1_OPT = cli_option("--no-nplus1-mem", dest="skip_nplusone_mem",
 872                           help="Skip N+1 memory redundancy tests",
 873                           action="store_true", default=False)
 874
 875 REBOOT_TYPE_OPT = cli_option("-t", "--type", dest="reboot_type",
 876                              help="Type of reboot: soft/hard/full",
 877                              default=constants.INSTANCE_REBOOT_HARD,
 878                              metavar="<REBOOT>",
 879                              choices=list(constants.REBOOT_TYPES))
 880
 881 IGNORE_SECONDARIES_OPT = cli_option("--ignore-secondaries",
 882                                     dest="ignore_secondaries",
 883                                     default=False, action="store_true",
 884                                     help="Ignore errors from secondaries")
 885
 886 NOSHUTDOWN_OPT = cli_option("--noshutdown", dest="shutdown",
 887                             action="store_false", default=True,
 888                             help="Don't shutdown the instance (unsafe)")
 889
 890 TIMEOUT_OPT = cli_option("--timeout", dest="timeout", type="int",
 891                          default=constants.DEFAULT_SHUTDOWN_TIMEOUT,
 892                          help="Maximum time to wait")
 893
 894 SHUTDOWN_TIMEOUT_OPT = cli_option("--shutdown-timeout",
 895                          dest="shutdown_timeout", type="int",
 896                          default=constants.DEFAULT_SHUTDOWN_TIMEOUT,
 897                          help="Maximum time to wait for instance shutdown")
 898
 899 EARLY_RELEASE_OPT = cli_option("--early-release",
 900                                dest="early_release", default=False,
 901                                action="store_true",
 902                                help="Release the locks on the secondary"
 903                                " node(s) early")
 904
 905 NEW_CLUSTER_CERT_OPT = cli_option("--new-cluster-certificate",
 906                                   dest="new_cluster_cert",
 907                                   default=False, action="store_true",
 908                                   help="Generate a new cluster certificate")
 909
 910 RAPI_CERT_OPT = cli_option("--rapi-certificate", dest="rapi_cert",
 911                            default=None,
 912                            help="File containing new RAPI certificate")
 913
 914 NEW_RAPI_CERT_OPT = cli_option("--new-rapi-certificate", dest="new_rapi_cert",
 915                                default=None, action="store_true",
 916                                help=("Generate a new self-signed RAPI"
 917                                      " certificate"))
 918
 919 NEW_CONFD_HMAC_KEY_OPT = cli_option("--new-confd-hmac-key",
 920                                     dest="new_confd_hmac_key",
 921                                     default=False, action="store_true",
 922                                     help=("Create a new HMAC key for %s" %
 923                                           constants.CONFD))
 924
 925 CLUSTER_DOMAIN_SECRET_OPT = cli_option("--cluster-domain-secret",
 926                                        dest="cluster_domain_secret",
 927                                        default=None,
 928                                        help=("Load new new cluster domain"
 929                                              " secret from file"))
 930
 931 NEW_CLUSTER_DOMAIN_SECRET_OPT = cli_option("--new-cluster-domain-secret",
 932                                            dest="new_cluster_domain_secret",
 933                                            default=False, action="store_true",
 934                                            help=("Create a new cluster domain"
 935                                                  " secret"))
 936
 937 USE_REPL_NET_OPT = cli_option("--use-replication-network",
 938                               dest="use_replication_network",
 939                               help="Whether to use the replication network"
 940                               " for talking to the nodes",
 941                               action="store_true", default=False)
 942
 943
 944 def _ParseArgs(argv, commands, aliases):
 945   """Parser for the command line arguments.
 946
 947   This function parses the arguments and returns the function which
 948   must be executed together with its (modified) arguments.
 949
 950   @param argv: the command line
 951   @param commands: dictionary with special contents, see the design
 952       doc for cmdline handling
 953   @param aliases: dictionary with command aliases {'alias': 'target, ...}
 954
 955   """
 956   if len(argv) == 0:
 957     binary = "<command>"
 958   else:
 959     binary = argv[0].split("/")[-1]
 960
 961   if len(argv) > 1 and argv[1] == "--version":
 962     ToStdout("%s (ganeti) %s", binary, constants.RELEASE_VERSION)
 963     # Quit right away. That way we don't have to care about this special
 964     # argument. optparse.py does it the same.
 965     sys.exit(0)
 966
 967   if len(argv) < 2 or not (argv[1] in commands or
 968                            argv[1] in aliases):
 969     # let's do a nice thing
 970     sortedcmds = commands.keys()
 971     sortedcmds.sort()
 972
 973     ToStdout("Usage: %s {command} [options...] [argument...]", binary)
 974     ToStdout("%s <command> --help to see details, or man %s", binary, binary)
 975     ToStdout("")
 976
 977     # compute the max line length for cmd + usage
 978     mlen = max([len(" %s" % cmd) for cmd in commands])
 979     mlen = min(60, mlen) # should not get here...
 980
 981     # and format a nice command list
 982     ToStdout("Commands:")
 983     for cmd in sortedcmds:
 984       cmdstr = " %s" % (cmd,)
 985       help_text = commands[cmd][4]
 986       help_lines = textwrap.wrap(help_text, 79 - 3 - mlen)
 987       ToStdout("%-*s - %s", mlen, cmdstr, help_lines.pop(0))
 988       for line in help_lines:
 989         ToStdout("%-*s   %s", mlen, "", line)
 990
 991     ToStdout("")
 992
 993     return None, None, None
 994
 995   # get command, unalias it, and look it up in commands
 996   cmd = argv.pop(1)
 997   if cmd in aliases:
 998     if cmd in commands:
 999       raise errors.ProgrammerError("Alias '%s' overrides an existing"
1000                                    " command" % cmd)
1001
1002     if aliases[cmd] not in commands:
1003       raise errors.ProgrammerError("Alias '%s' maps to non-existing"
1004                                    " command '%s'" % (cmd, aliases[cmd]))
1005
1006     cmd = aliases[cmd]
1007
1008   func, args_def, parser_opts, usage, description = commands[cmd]
1009   parser = OptionParser(option_list=parser_opts + [_DRY_RUN_OPT, DEBUG_OPT],
1010                         description=description,
1011                         formatter=TitledHelpFormatter(),
1012                         usage="%%prog %s %s" % (cmd, usage))
1013   parser.disable_interspersed_args()
1014   options, args = parser.parse_args()
1015
1016   if not _CheckArguments(cmd, args_def, args):
1017     return None, None, None
1018
1019   return func, options, args
1020
1021
1022 def _CheckArguments(cmd, args_def, args):
1023   """Verifies the arguments using the argument definition.
1024
1025   Algorithm:
1026
1027     1. Abort with error if values specified by user but none expected.
1028
1029     1. For each argument in definition
1030
1031       1. Keep running count of minimum number of values (min_count)
1032       1. Keep running count of maximum number of values (max_count)
1033       1. If it has an unlimited number of values
1034
1035         1. Abort with error if it's not the last argument in the definition
1036
1037     1. If last argument has limited number of values
1038
1039       1. Abort with error if number of values doesn't match or is too large
1040
1041     1. Abort with error if user didn't pass enough values (min_count)
1042
1043   """
1044   if args and not args_def:
1045     ToStderr("Error: Command %s expects no arguments", cmd)
1046     return False
1047
1048   min_count = None
1049   max_count = None
1050   check_max = None
1051
1052   last_idx = len(args_def) - 1
1053
1054   for idx, arg in enumerate(args_def):
1055     if min_count is None:
1056       min_count = arg.min
1057     elif arg.min is not None:
1058       min_count += arg.min
1059
1060     if max_count is None:
1061       max_count = arg.max
1062     elif arg.max is not None:
1063       max_count += arg.max
1064
1065     if idx == last_idx:
1066       check_max = (arg.max is not None)
1067
1068     elif arg.max is None:
1069       raise errors.ProgrammerError("Only the last argument can have max=None")
1070
1071   if check_max:
1072     # Command with exact number of arguments
1073     if (min_count is not None and max_count is not None and
1074         min_count == max_count and len(args) != min_count):
1075       ToStderr("Error: Command %s expects %d argument(s)", cmd, min_count)
1076       return False
1077
1078     # Command with limited number of arguments
1079     if max_count is not None and len(args) > max_count:
1080       ToStderr("Error: Command %s expects only %d argument(s)",
1081                cmd, max_count)
1082       return False
1083
1084   # Command with some required arguments
1085   if min_count is not None and len(args) < min_count:
1086     ToStderr("Error: Command %s expects at least %d argument(s)",
1087              cmd, min_count)
1088     return False
1089
1090   return True
1091
1092
1093 def SplitNodeOption(value):
1094   """Splits the value of a --node option.
1095
1096   """
1097   if value and ':' in value:
1098     return value.split(':', 1)
1099   else:
1100     return (value, None)
1101
1102
1103 def CalculateOSNames(os_name, os_variants):
1104   """Calculates all the names an OS can be called, according to its variants.
1105
1106   @type os_name: string
1107   @param os_name: base name of the os
1108   @type os_variants: list or None
1109   @param os_variants: list of supported variants
1110   @rtype: list
1111   @return: list of valid names
1112
1113   """
1114   if os_variants:
1115     return ['%s+%s' % (os_name, v) for v in os_variants]
1116   else:
1117     return [os_name]
1118
1119
1120 def UsesRPC(fn):
1121   def wrapper(*args, **kwargs):
1122     rpc.Init()
1123     try:
1124       return fn(*args, **kwargs)
1125     finally:
1126       rpc.Shutdown()
1127   return wrapper
1128
1129
1130 def AskUser(text, choices=None):
1131   """Ask the user a question.
1132
1133   @param text: the question to ask
1134
1135   @param choices: list with elements tuples (input_char, return_value,
1136       description); if not given, it will default to: [('y', True,
1137       'Perform the operation'), ('n', False, 'Do no do the operation')];
1138       note that the '?' char is reserved for help
1139
1140   @return: one of the return values from the choices list; if input is
1141       not possible (i.e. not running with a tty, we return the last
1142       entry from the list
1143
1144   """
1145   if choices is None:
1146     choices = [('y', True, 'Perform the operation'),
1147                ('n', False, 'Do not perform the operation')]
1148   if not choices or not isinstance(choices, list):
1149     raise errors.ProgrammerError("Invalid choices argument to AskUser")
1150   for entry in choices:
1151     if not isinstance(entry, tuple) or len(entry) < 3 or entry[0] == '?':
1152       raise errors.ProgrammerError("Invalid choices element to AskUser")
1153
1154   answer = choices[-1][1]
1155   new_text = []
1156   for line in text.splitlines():
1157     new_text.append(textwrap.fill(line, 70, replace_whitespace=False))
1158   text = "\n".join(new_text)
1159   try:
1160     f = file("/dev/tty", "a+")
1161   except IOError:
1162     return answer
1163   try:
1164     chars = [entry[0] for entry in choices]
1165     chars[-1] = "[%s]" % chars[-1]
1166     chars.append('?')
1167     maps = dict([(entry[0], entry[1]) for entry in choices])
1168     while True:
1169       f.write(text)
1170       f.write('\n')
1171       f.write("/".join(chars))
1172       f.write(": ")
1173       line = f.readline(2).strip().lower()
1174       if line in maps:
1175         answer = maps[line]
1176         break
1177       elif line == '?':
1178         for entry in choices:
1179           f.write(" %s - %s\n" % (entry[0], entry[2]))
1180         f.write("\n")
1181         continue
1182   finally:
1183     f.close()
1184   return answer
1185
1186
1187 class JobSubmittedException(Exception):
1188   """Job was submitted, client should exit.
1189
1190   This exception has one argument, the ID of the job that was
1191   submitted. The handler should print this ID.
1192
1193   This is not an error, just a structured way to exit from clients.
1194
1195   """
1196
1197
1198 def SendJob(ops, cl=None):
1199   """Function to submit an opcode without waiting for the results.
1200
1201   @type ops: list
1202   @param ops: list of opcodes
1203   @type cl: luxi.Client
1204   @param cl: the luxi client to use for communicating with the master;
1205              if None, a new client will be created
1206
1207   """
1208   if cl is None:
1209     cl = GetClient()
1210
1211   job_id = cl.SubmitJob(ops)
1212
1213   return job_id
1214
1215
1216 def PollJob(job_id, cl=None, feedback_fn=None):
1217   """Function to poll for the result of a job.
1218
1219   @type job_id: job identified
1220   @param job_id: the job to poll for results
1221   @type cl: luxi.Client
1222   @param cl: the luxi client to use for communicating with the master;
1223              if None, a new client will be created
1224
1225   """
1226   if cl is None:
1227     cl = GetClient()
1228
1229   prev_job_info = None
1230   prev_logmsg_serial = None
1231
1232   status = None
1233
1234   notified_queued = False
1235   notified_waitlock = False
1236
1237   while True:
1238     result = cl.WaitForJobChangeOnce(job_id, ["status"], prev_job_info,
1239                                      prev_logmsg_serial)
1240     if not result:
1241       # job not found, go away!
1242       raise errors.JobLost("Job with id %s lost" % job_id)
1243     elif result == constants.JOB_NOTCHANGED:
1244       if status is not None and not callable(feedback_fn):
1245         if status == constants.JOB_STATUS_QUEUED and not notified_queued:
1246           ToStderr("Job %s is waiting in queue", job_id)
1247           notified_queued = True
1248         elif status == constants.JOB_STATUS_WAITLOCK and not notified_waitlock:
1249           ToStderr("Job %s is trying to acquire all necessary locks", job_id)
1250           notified_waitlock = True
1251
1252       # Wait again
1253       continue
1254
1255     # Split result, a tuple of (field values, log entries)
1256     (job_info, log_entries) = result
1257     (status, ) = job_info
1258
1259     if log_entries:
1260       for log_entry in log_entries:
1261         (serial, timestamp, _, message) = log_entry
1262         if callable(feedback_fn):
1263           feedback_fn(log_entry[1:])
1264         else:
1265           encoded = utils.SafeEncode(message)
1266           ToStdout("%s %s", time.ctime(utils.MergeTime(timestamp)), encoded)
1267         prev_logmsg_serial = max(prev_logmsg_serial, serial)
1268
1269     # TODO: Handle canceled and archived jobs
1270     elif status in (constants.JOB_STATUS_SUCCESS,
1271                     constants.JOB_STATUS_ERROR,
1272                     constants.JOB_STATUS_CANCELING,
1273                     constants.JOB_STATUS_CANCELED):
1274       break
1275
1276     prev_job_info = job_info
1277
1278   jobs = cl.QueryJobs([job_id], ["status", "opstatus", "opresult"])
1279   if not jobs:
1280     raise errors.JobLost("Job with id %s lost" % job_id)
1281
1282   status, opstatus, result = jobs[0]
1283   if status == constants.JOB_STATUS_SUCCESS:
1284     return result
1285   elif status in (constants.JOB_STATUS_CANCELING,
1286                   constants.JOB_STATUS_CANCELED):
1287     raise errors.OpExecError("Job was canceled")
1288   else:
1289     has_ok = False
1290     for idx, (status, msg) in enumerate(zip(opstatus, result)):
1291       if status == constants.OP_STATUS_SUCCESS:
1292         has_ok = True
1293       elif status == constants.OP_STATUS_ERROR:
1294         errors.MaybeRaise(msg)
1295         if has_ok:
1296           raise errors.OpExecError("partial failure (opcode %d): %s" %
1297                                    (idx, msg))
1298         else:
1299           raise errors.OpExecError(str(msg))
1300     # default failure mode
1301     raise errors.OpExecError(result)
1302
1303
1304 def SubmitOpCode(op, cl=None, feedback_fn=None, opts=None):
1305   """Legacy function to submit an opcode.
1306
1307   This is just a simple wrapper over the construction of the processor
1308   instance. It should be extended to better handle feedback and
1309   interaction functions.
1310
1311   """
1312   if cl is None:
1313     cl = GetClient()
1314
1315   SetGenericOpcodeOpts([op], opts)
1316
1317   job_id = SendJob([op], cl)
1318
1319   op_results = PollJob(job_id, cl=cl, feedback_fn=feedback_fn)
1320
1321   return op_results[0]
1322
1323
1324 def SubmitOrSend(op, opts, cl=None, feedback_fn=None):
1325   """Wrapper around SubmitOpCode or SendJob.
1326
1327   This function will decide, based on the 'opts' parameter, whether to
1328   submit and wait for the result of the opcode (and return it), or
1329   whether to just send the job and print its identifier. It is used in
1330   order to simplify the implementation of the '--submit' option.
1331
1332   It will also process the opcodes if we're sending the via SendJob
1333   (otherwise SubmitOpCode does it).
1334
1335   """
1336   if opts and opts.submit_only:
1337     job = [op]
1338     SetGenericOpcodeOpts(job, opts)
1339     job_id = SendJob(job, cl=cl)
1340     raise JobSubmittedException(job_id)
1341   else:
1342     return SubmitOpCode(op, cl=cl, feedback_fn=feedback_fn, opts=opts)
1343
1344
1345 def SetGenericOpcodeOpts(opcode_list, options):
1346   """Processor for generic options.
1347
1348   This function updates the given opcodes based on generic command
1349   line options (like debug, dry-run, etc.).
1350
1351   @param opcode_list: list of opcodes
1352   @param options: command line options or None
1353   @return: None (in-place modification)
1354
1355   """
1356   if not options:
1357     return
1358   for op in opcode_list:
1359     op.dry_run = options.dry_run
1360     op.debug_level = options.debug
1361
1362
1363 def GetClient():
1364   # TODO: Cache object?
1365   try:
1366     client = luxi.Client()
1367   except luxi.NoMasterError:
1368     ss = ssconf.SimpleStore()
1369
1370     # Try to read ssconf file
1371     try:
1372       ss.GetMasterNode()
1373     except errors.ConfigurationError:
1374       raise errors.OpPrereqError("Cluster not initialized or this machine is"
1375                                  " not part of a cluster")
1376
1377     master, myself = ssconf.GetMasterAndMyself(ss=ss)
1378     if master != myself:
1379       raise errors.OpPrereqError("This is not the master node, please connect"
1380                                  " to node '%s' and rerun the command" %
1381                                  master)
1382     raise
1383   return client
1384
1385
1386 def FormatError(err):
1387   """Return a formatted error message for a given error.
1388
1389   This function takes an exception instance and returns a tuple
1390   consisting of two values: first, the recommended exit code, and
1391   second, a string describing the error message (not
1392   newline-terminated).
1393
1394   """
1395   retcode = 1
1396   obuf = StringIO()
1397   msg = str(err)
1398   if isinstance(err, errors.ConfigurationError):
1399     txt = "Corrupt configuration file: %s" % msg
1400     logging.error(txt)
1401     obuf.write(txt + "\n")
1402     obuf.write("Aborting.")
1403     retcode = 2
1404   elif isinstance(err, errors.HooksAbort):
1405     obuf.write("Failure: hooks execution failed:\n")
1406     for node, script, out in err.args[0]:
1407       if out:
1408         obuf.write("  node: %s, script: %s, output: %s\n" %
1409                    (node, script, out))
1410       else:
1411         obuf.write("  node: %s, script: %s (no output)\n" %
1412                    (node, script))
1413   elif isinstance(err, errors.HooksFailure):
1414     obuf.write("Failure: hooks general failure: %s" % msg)
1415   elif isinstance(err, errors.ResolverError):
1416     this_host = utils.HostInfo.SysName()
1417     if err.args[0] == this_host:
1418       msg = "Failure: can't resolve my own hostname ('%s')"
1419     else:
1420       msg = "Failure: can't resolve hostname '%s'"
1421     obuf.write(msg % err.args[0])
1422   elif isinstance(err, errors.OpPrereqError):
1423     if len(err.args) == 2:
1424       obuf.write("Failure: prerequisites not met for this"
1425                " operation:\nerror type: %s, error details:\n%s" %
1426                  (err.args[1], err.args[0]))
1427     else:
1428       obuf.write("Failure: prerequisites not met for this"
1429                  " operation:\n%s" % msg)
1430   elif isinstance(err, errors.OpExecError):
1431     obuf.write("Failure: command execution error:\n%s" % msg)
1432   elif isinstance(err, errors.TagError):
1433     obuf.write("Failure: invalid tag(s) given:\n%s" % msg)
1434   elif isinstance(err, errors.JobQueueDrainError):
1435     obuf.write("Failure: the job queue is marked for drain and doesn't"
1436                " accept new requests\n")
1437   elif isinstance(err, errors.JobQueueFull):
1438     obuf.write("Failure: the job queue is full and doesn't accept new"
1439                " job submissions until old jobs are archived\n")
1440   elif isinstance(err, errors.TypeEnforcementError):
1441     obuf.write("Parameter Error: %s" % msg)
1442   elif isinstance(err, errors.ParameterError):
1443     obuf.write("Failure: unknown/wrong parameter name '%s'" % msg)
1444   elif isinstance(err, luxi.NoMasterError):
1445     obuf.write("Cannot communicate with the master daemon.\nIs it running"
1446                " and listening for connections?")
1447   elif isinstance(err, luxi.TimeoutError):
1448     obuf.write("Timeout while talking to the master daemon. Error:\n"
1449                "%s" % msg)
1450   elif isinstance(err, luxi.ProtocolError):
1451     obuf.write("Unhandled protocol error while talking to the master daemon:\n"
1452                "%s" % msg)
1453   elif isinstance(err, errors.GenericError):
1454     obuf.write("Unhandled Ganeti error: %s" % msg)
1455   elif isinstance(err, JobSubmittedException):
1456     obuf.write("JobID: %s\n" % err.args[0])
1457     retcode = 0
1458   else:
1459     obuf.write("Unhandled exception: %s" % msg)
1460   return retcode, obuf.getvalue().rstrip('\n')
1461
1462
1463 def GenericMain(commands, override=None, aliases=None):
1464   """Generic main function for all the gnt-* commands.
1465
1466   Arguments:
1467     - commands: a dictionary with a special structure, see the design doc
1468                 for command line handling.
1469     - override: if not None, we expect a dictionary with keys that will
1470                 override command line options; this can be used to pass
1471                 options from the scripts to generic functions
1472     - aliases: dictionary with command aliases {'alias': 'target, ...}
1473
1474   """
1475   # save the program name and the entire command line for later logging
1476   if sys.argv:
1477     binary = os.path.basename(sys.argv[0]) or sys.argv[0]
1478     if len(sys.argv) >= 2:
1479       binary += " " + sys.argv[1]
1480       old_cmdline = " ".join(sys.argv[2:])
1481     else:
1482       old_cmdline = ""
1483   else:
1484     binary = "<unknown program>"
1485     old_cmdline = ""
1486
1487   if aliases is None:
1488     aliases = {}
1489
1490   try:
1491     func, options, args = _ParseArgs(sys.argv, commands, aliases)
1492   except errors.ParameterError, err:
1493     result, err_msg = FormatError(err)
1494     ToStderr(err_msg)
1495     return 1
1496
1497   if func is None: # parse error
1498     return 1
1499
1500   if override is not None:
1501     for key, val in override.iteritems():
1502       setattr(options, key, val)
1503
1504   utils.SetupLogging(constants.LOG_COMMANDS, debug=options.debug,
1505                      stderr_logging=True, program=binary)
1506
1507   if old_cmdline:
1508     logging.info("run with arguments '%s'", old_cmdline)
1509   else:
1510     logging.info("run with no arguments")
1511
1512   try:
1513     result = func(options, args)
1514   except (errors.GenericError, luxi.ProtocolError,
1515           JobSubmittedException), err:
1516     result, err_msg = FormatError(err)
1517     logging.exception("Error during command processing")
1518     ToStderr(err_msg)
1519
1520   return result
1521
1522
1523 def GenericInstanceCreate(mode, opts, args):
1524   """Add an instance to the cluster via either creation or import.
1525
1526   @param mode: constants.INSTANCE_CREATE or constants.INSTANCE_IMPORT
1527   @param opts: the command line options selected by the user
1528   @type args: list
1529   @param args: should contain only one element, the new instance name
1530   @rtype: int
1531   @return: the desired exit code
1532
1533   """
1534   instance = args[0]
1535
1536   (pnode, snode) = SplitNodeOption(opts.node)
1537
1538   hypervisor = None
1539   hvparams = {}
1540   if opts.hypervisor:
1541     hypervisor, hvparams = opts.hypervisor
1542
1543   if opts.nics:
1544     try:
1545       nic_max = max(int(nidx[0]) + 1 for nidx in opts.nics)
1546     except ValueError, err:
1547       raise errors.OpPrereqError("Invalid NIC index passed: %s" % str(err))
1548     nics = [{}] * nic_max
1549     for nidx, ndict in opts.nics:
1550       nidx = int(nidx)
1551       if not isinstance(ndict, dict):
1552         msg = "Invalid nic/%d value: expected dict, got %s" % (nidx, ndict)
1553         raise errors.OpPrereqError(msg)
1554       nics[nidx] = ndict
1555   elif opts.no_nics:
1556     # no nics
1557     nics = []
1558   else:
1559     # default of one nic, all auto
1560     nics = [{}]
1561
1562   if opts.disk_template == constants.DT_DISKLESS:
1563     if opts.disks or opts.sd_size is not None:
1564       raise errors.OpPrereqError("Diskless instance but disk"
1565                                  " information passed")
1566     disks = []
1567   else:
1568     if not opts.disks and not opts.sd_size:
1569       raise errors.OpPrereqError("No disk information specified")
1570     if opts.disks and opts.sd_size is not None:
1571       raise errors.OpPrereqError("Please use either the '--disk' or"
1572                                  " '-s' option")
1573     if opts.sd_size is not None:
1574       opts.disks = [(0, {"size": opts.sd_size})]
1575     try:
1576       disk_max = max(int(didx[0]) + 1 for didx in opts.disks)
1577     except ValueError, err:
1578       raise errors.OpPrereqError("Invalid disk index passed: %s" % str(err))
1579     disks = [{}] * disk_max
1580     for didx, ddict in opts.disks:
1581       didx = int(didx)
1582       if not isinstance(ddict, dict):
1583         msg = "Invalid disk/%d value: expected dict, got %s" % (didx, ddict)
1584         raise errors.OpPrereqError(msg)
1585       elif "size" in ddict:
1586         if "adopt" in ddict:
1587           raise errors.OpPrereqError("Only one of 'size' and 'adopt' allowed"
1588                                      " (disk %d)" % didx)
1589         try:
1590           ddict["size"] = utils.ParseUnit(ddict["size"])
1591         except ValueError, err:
1592           raise errors.OpPrereqError("Invalid disk size for disk %d: %s" %
1593                                      (didx, err))
1594       elif "adopt" in ddict:
1595         if mode == constants.INSTANCE_IMPORT:
1596           raise errors.OpPrereqError("Disk adoption not allowed for instance"
1597                                      " import")
1598         ddict["size"] = 0
1599       else:
1600         raise errors.OpPrereqError("Missing size or adoption source for"
1601                                    " disk %d" % didx)
1602       disks[didx] = ddict
1603
1604   utils.ForceDictType(opts.beparams, constants.BES_PARAMETER_TYPES)
1605   utils.ForceDictType(hvparams, constants.HVS_PARAMETER_TYPES)
1606
1607   if mode == constants.INSTANCE_CREATE:
1608     start = opts.start
1609     os_type = opts.os
1610     src_node = None
1611     src_path = None
1612     no_install = opts.no_install
1613   elif mode == constants.INSTANCE_IMPORT:
1614     start = False
1615     os_type = None
1616     src_node = opts.src_node
1617     src_path = opts.src_dir
1618     no_install = None
1619   else:
1620     raise errors.ProgrammerError("Invalid creation mode %s" % mode)
1621
1622   op = opcodes.OpCreateInstance(instance_name=instance,
1623                                 disks=disks,
1624                                 disk_template=opts.disk_template,
1625                                 nics=nics,
1626                                 pnode=pnode, snode=snode,
1627                                 ip_check=opts.ip_check,
1628                                 name_check=opts.name_check,
1629                                 wait_for_sync=opts.wait_for_sync,
1630                                 file_storage_dir=opts.file_storage_dir,
1631                                 file_driver=opts.file_driver,
1632                                 iallocator=opts.iallocator,
1633                                 hypervisor=hypervisor,
1634                                 hvparams=hvparams,
1635                                 beparams=opts.beparams,
1636                                 mode=mode,
1637                                 start=start,
1638                                 os_type=os_type,
1639                                 src_node=src_node,
1640                                 src_path=src_path,
1641                                 no_install=no_install)
1642
1643   SubmitOrSend(op, opts)
1644   return 0
1645
1646
1647 class _RunWhileClusterStoppedHelper:
1648   """Helper class for L{RunWhileClusterStopped} to simplify state management
1649
1650   """
1651   def __init__(self, feedback_fn, cluster_name, master_node, online_nodes):
1652     """Initializes this class.
1653
1654     @type feedback_fn: callable
1655     @param feedback_fn: Feedback function
1656     @type cluster_name: string
1657     @param cluster_name: Cluster name
1658     @type master_node: string
1659     @param master_node Master node name
1660     @type online_nodes: list
1661     @param online_nodes: List of names of online nodes
1662
1663     """
1664     self.feedback_fn = feedback_fn
1665     self.cluster_name = cluster_name
1666     self.master_node = master_node
1667     self.online_nodes = online_nodes
1668
1669     self.ssh = ssh.SshRunner(self.cluster_name)
1670
1671     self.nonmaster_nodes = [name for name in online_nodes
1672                             if name != master_node]
1673
1674     assert self.master_node not in self.nonmaster_nodes
1675
1676   def _RunCmd(self, node_name, cmd):
1677     """Runs a command on the local or a remote machine.
1678
1679     @type node_name: string
1680     @param node_name: Machine name
1681     @type cmd: list
1682     @param cmd: Command
1683
1684     """
1685     if node_name is None or node_name == self.master_node:
1686       # No need to use SSH
1687       result = utils.RunCmd(cmd)
1688     else:
1689       result = self.ssh.Run(node_name, "root", utils.ShellQuoteArgs(cmd))
1690
1691     if result.failed:
1692       errmsg = ["Failed to run command %s" % result.cmd]
1693       if node_name:
1694         errmsg.append("on node %s" % node_name)
1695       errmsg.append(": exitcode %s and error %s" %
1696                     (result.exit_code, result.output))
1697       raise errors.OpExecError(" ".join(errmsg))
1698
1699   def Call(self, fn, *args):
1700     """Call function while all daemons are stopped.
1701
1702     @type fn: callable
1703     @param fn: Function to be called
1704
1705     """
1706     # Pause watcher by acquiring an exclusive lock on watcher state file
1707     self.feedback_fn("Blocking watcher")
1708     watcher_block = utils.FileLock.Open(constants.WATCHER_STATEFILE)
1709     try:
1710       # TODO: Currently, this just blocks. There's no timeout.
1711       # TODO: Should it be a shared lock?
1712       watcher_block.Exclusive(blocking=True)
1713
1714       # Stop master daemons, so that no new jobs can come in and all running
1715       # ones are finished
1716       self.feedback_fn("Stopping master daemons")
1717       self._RunCmd(None, [constants.DAEMON_UTIL, "stop-master"])
1718       try:
1719         # Stop daemons on all nodes
1720         for node_name in self.online_nodes:
1721           self.feedback_fn("Stopping daemons on %s" % node_name)
1722           self._RunCmd(node_name, [constants.DAEMON_UTIL, "stop-all"])
1723
1724         # All daemons are shut down now
1725         try:
1726           return fn(self, *args)
1727         except Exception, err:
1728           _, errmsg = FormatError(err)
1729           logging.exception("Caught exception")
1730           self.feedback_fn(errmsg)
1731           raise
1732       finally:
1733         # Start cluster again, master node last
1734         for node_name in self.nonmaster_nodes + [self.master_node]:
1735           self.feedback_fn("Starting daemons on %s" % node_name)
1736           self._RunCmd(node_name, [constants.DAEMON_UTIL, "start-all"])
1737     finally:
1738       # Resume watcher
1739       watcher_block.Close()
1740
1741
1742 def RunWhileClusterStopped(feedback_fn, fn, *args):
1743   """Calls a function while all cluster daemons are stopped.
1744
1745   @type feedback_fn: callable
1746   @param feedback_fn: Feedback function
1747   @type fn: callable
1748   @param fn: Function to be called when daemons are stopped
1749
1750   """
1751   feedback_fn("Gathering cluster information")
1752
1753   # This ensures we're running on the master daemon
1754   cl = GetClient()
1755
1756   (cluster_name, master_node) = \
1757     cl.QueryConfigValues(["cluster_name", "master_node"])
1758
1759   online_nodes = GetOnlineNodes([], cl=cl)
1760
1761   # Don't keep a reference to the client. The master daemon will go away.
1762   del cl
1763
1764   assert master_node in online_nodes
1765
1766   return _RunWhileClusterStoppedHelper(feedback_fn, cluster_name, master_node,
1767                                        online_nodes).Call(fn, *args)
1768
1769
1770 def GenerateTable(headers, fields, separator, data,
1771                   numfields=None, unitfields=None,
1772                   units=None):
1773   """Prints a table with headers and different fields.
1774
1775   @type headers: dict
1776   @param headers: dictionary mapping field names to headers for
1777       the table
1778   @type fields: list
1779   @param fields: the field names corresponding to each row in
1780       the data field
1781   @param separator: the separator to be used; if this is None,
1782       the default 'smart' algorithm is used which computes optimal
1783       field width, otherwise just the separator is used between
1784       each field
1785   @type data: list
1786   @param data: a list of lists, each sublist being one row to be output
1787   @type numfields: list
1788   @param numfields: a list with the fields that hold numeric
1789       values and thus should be right-aligned
1790   @type unitfields: list
1791   @param unitfields: a list with the fields that hold numeric
1792       values that should be formatted with the units field
1793   @type units: string or None
1794   @param units: the units we should use for formatting, or None for
1795       automatic choice (human-readable for non-separator usage, otherwise
1796       megabytes); this is a one-letter string
1797
1798   """
1799   if units is None:
1800     if separator:
1801       units = "m"
1802     else:
1803       units = "h"
1804
1805   if numfields is None:
1806     numfields = []
1807   if unitfields is None:
1808     unitfields = []
1809
1810   numfields = utils.FieldSet(*numfields)   # pylint: disable-msg=W0142
1811   unitfields = utils.FieldSet(*unitfields) # pylint: disable-msg=W0142
1812
1813   format_fields = []
1814   for field in fields:
1815     if headers and field not in headers:
1816       # TODO: handle better unknown fields (either revert to old
1817       # style of raising exception, or deal more intelligently with
1818       # variable fields)
1819       headers[field] = field
1820     if separator is not None:
1821       format_fields.append("%s")
1822     elif numfields.Matches(field):
1823       format_fields.append("%*s")
1824     else:
1825       format_fields.append("%-*s")
1826
1827   if separator is None:
1828     mlens = [0 for name in fields]
1829     format = ' '.join(format_fields)
1830   else:
1831     format = separator.replace("%", "%%").join(format_fields)
1832
1833   for row in data:
1834     if row is None:
1835       continue
1836     for idx, val in enumerate(row):
1837       if unitfields.Matches(fields[idx]):
1838         try:
1839           val = int(val)
1840         except (TypeError, ValueError):
1841           pass
1842         else:
1843           val = row[idx] = utils.FormatUnit(val, units)
1844       val = row[idx] = str(val)
1845       if separator is None:
1846         mlens[idx] = max(mlens[idx], len(val))
1847
1848   result = []
1849   if headers:
1850     args = []
1851     for idx, name in enumerate(fields):
1852       hdr = headers[name]
1853       if separator is None:
1854         mlens[idx] = max(mlens[idx], len(hdr))
1855         args.append(mlens[idx])
1856       args.append(hdr)
1857     result.append(format % tuple(args))
1858
1859   if separator is None:
1860     assert len(mlens) == len(fields)
1861
1862     if fields and not numfields.Matches(fields[-1]):
1863       mlens[-1] = 0
1864
1865   for line in data:
1866     args = []
1867     if line is None:
1868       line = ['-' for _ in fields]
1869     for idx in range(len(fields)):
1870       if separator is None:
1871         args.append(mlens[idx])
1872       args.append(line[idx])
1873     result.append(format % tuple(args))
1874
1875   return result
1876
1877
1878 def FormatTimestamp(ts):
1879   """Formats a given timestamp.
1880
1881   @type ts: timestamp
1882   @param ts: a timeval-type timestamp, a tuple of seconds and microseconds
1883
1884   @rtype: string
1885   @return: a string with the formatted timestamp
1886
1887   """
1888   if not isinstance (ts, (tuple, list)) or len(ts) != 2:
1889     return '?'
1890   sec, usec = ts
1891   return time.strftime("%F %T", time.localtime(sec)) + ".%06d" % usec
1892
1893
1894 def ParseTimespec(value):
1895   """Parse a time specification.
1896
1897   The following suffixed will be recognized:
1898
1899     - s: seconds
1900     - m: minutes
1901     - h: hours
1902     - d: day
1903     - w: weeks
1904
1905   Without any suffix, the value will be taken to be in seconds.
1906
1907   """
1908   value = str(value)
1909   if not value:
1910     raise errors.OpPrereqError("Empty time specification passed")
1911   suffix_map = {
1912     's': 1,
1913     'm': 60,
1914     'h': 3600,
1915     'd': 86400,
1916     'w': 604800,
1917     }
1918   if value[-1] not in suffix_map:
1919     try:
1920       value = int(value)
1921     except (TypeError, ValueError):
1922       raise errors.OpPrereqError("Invalid time specification '%s'" % value)
1923   else:
1924     multiplier = suffix_map[value[-1]]
1925     value = value[:-1]
1926     if not value: # no data left after stripping the suffix
1927       raise errors.OpPrereqError("Invalid time specification (only"
1928                                  " suffix passed)")
1929     try:
1930       value = int(value) * multiplier
1931     except (TypeError, ValueError):
1932       raise errors.OpPrereqError("Invalid time specification '%s'" % value)
1933   return value
1934
1935
1936 def GetOnlineNodes(nodes, cl=None, nowarn=False, secondary_ips=False,
1937                    filter_master=False):
1938   """Returns the names of online nodes.
1939
1940   This function will also log a warning on stderr with the names of
1941   the online nodes.
1942
1943   @param nodes: if not empty, use only this subset of nodes (minus the
1944       offline ones)
1945   @param cl: if not None, luxi client to use
1946   @type nowarn: boolean
1947   @param nowarn: by default, this function will output a note with the
1948       offline nodes that are skipped; if this parameter is True the
1949       note is not displayed
1950   @type secondary_ips: boolean
1951   @param secondary_ips: if True, return the secondary IPs instead of the
1952       names, useful for doing network traffic over the replication interface
1953       (if any)
1954   @type filter_master: boolean
1955   @param filter_master: if True, do not return the master node in the list
1956       (useful in coordination with secondary_ips where we cannot check our
1957       node name against the list)
1958
1959   """
1960   if cl is None:
1961     cl = GetClient()
1962
1963   if secondary_ips:
1964     name_idx = 2
1965   else:
1966     name_idx = 0
1967
1968   if filter_master:
1969     master_node = cl.QueryConfigValues(["master_node"])[0]
1970     filter_fn = lambda x: x != master_node
1971   else:
1972     filter_fn = lambda _: True
1973
1974   result = cl.QueryNodes(names=nodes, fields=["name", "offline", "sip"],
1975                          use_locking=False)
1976   offline = [row[0] for row in result if row[1]]
1977   if offline and not nowarn:
1978     ToStderr("Note: skipping offline node(s): %s" % utils.CommaJoin(offline))
1979   return [row[name_idx] for row in result if not row[1] and filter_fn(row[0])]
1980
1981
1982 def _ToStream(stream, txt, *args):
1983   """Write a message to a stream, bypassing the logging system
1984
1985   @type stream: file object
1986   @param stream: the file to which we should write
1987   @type txt: str
1988   @param txt: the message
1989
1990   """
1991   if args:
1992     args = tuple(args)
1993     stream.write(txt % args)
1994   else:
1995     stream.write(txt)
1996   stream.write('\n')
1997   stream.flush()
1998
1999
2000 def ToStdout(txt, *args):
2001   """Write a message to stdout only, bypassing the logging system
2002
2003   This is just a wrapper over _ToStream.
2004
2005   @type txt: str
2006   @param txt: the message
2007
2008   """
2009   _ToStream(sys.stdout, txt, *args)
2010
2011
2012 def ToStderr(txt, *args):
2013   """Write a message to stderr only, bypassing the logging system
2014
2015   This is just a wrapper over _ToStream.
2016
2017   @type txt: str
2018   @param txt: the message
2019
2020   """
2021   _ToStream(sys.stderr, txt, *args)
2022
2023
2024 class JobExecutor(object):
2025   """Class which manages the submission and execution of multiple jobs.
2026
2027   Note that instances of this class should not be reused between
2028   GetResults() calls.
2029
2030   """
2031   def __init__(self, cl=None, verbose=True, opts=None, feedback_fn=None):
2032     self.queue = []
2033     if cl is None:
2034       cl = GetClient()
2035     self.cl = cl
2036     self.verbose = verbose
2037     self.jobs = []
2038     self.opts = opts
2039     self.feedback_fn = feedback_fn
2040
2041   def QueueJob(self, name, *ops):
2042     """Record a job for later submit.
2043
2044     @type name: string
2045     @param name: a description of the job, will be used in WaitJobSet
2046     """
2047     SetGenericOpcodeOpts(ops, self.opts)
2048     self.queue.append((name, ops))
2049
2050   def SubmitPending(self):
2051     """Submit all pending jobs.
2052
2053     """
2054     results = self.cl.SubmitManyJobs([row[1] for row in self.queue])
2055     for (idx, ((status, data), (name, _))) in enumerate(zip(results,
2056                                                             self.queue)):
2057       self.jobs.append((idx, status, data, name))
2058
2059   def _ChooseJob(self):
2060     """Choose a non-waiting/queued job to poll next.
2061
2062     """
2063     assert self.jobs, "_ChooseJob called with empty job list"
2064
2065     result = self.cl.QueryJobs([i[2] for i in self.jobs], ["status"])
2066     assert result
2067
2068     for job_data, status in zip(self.jobs, result):
2069       if status[0] in (constants.JOB_STATUS_QUEUED,
2070                     constants.JOB_STATUS_WAITLOCK,
2071                     constants.JOB_STATUS_CANCELING):
2072         # job is still waiting
2073         continue
2074       # good candidate found
2075       self.jobs.remove(job_data)
2076       return job_data
2077
2078     # no job found
2079     return self.jobs.pop(0)
2080
2081   def GetResults(self):
2082     """Wait for and return the results of all jobs.
2083
2084     @rtype: list
2085     @return: list of tuples (success, job results), in the same order
2086         as the submitted jobs; if a job has failed, instead of the result
2087         there will be the error message
2088
2089     """
2090     if not self.jobs:
2091       self.SubmitPending()
2092     results = []
2093     if self.verbose:
2094       ok_jobs = [row[2] for row in self.jobs if row[1]]
2095       if ok_jobs:
2096         ToStdout("Submitted jobs %s", utils.CommaJoin(ok_jobs))
2097
2098     # first, remove any non-submitted jobs
2099     self.jobs, failures = utils.partition(self.jobs, lambda x: x[1])
2100     for idx, _, jid, name in failures:
2101       ToStderr("Failed to submit job for %s: %s", name, jid)
2102       results.append((idx, False, jid))
2103
2104     while self.jobs:
2105       (idx, _, jid, name) = self._ChooseJob()
2106       ToStdout("Waiting for job %s for %s...", jid, name)
2107       try:
2108         job_result = PollJob(jid, cl=self.cl, feedback_fn=self.feedback_fn)
2109         success = True
2110       except (errors.GenericError, luxi.ProtocolError), err:
2111         _, job_result = FormatError(err)
2112         success = False
2113         # the error message will always be shown, verbose or not
2114         ToStderr("Job %s for %s has failed: %s", jid, name, job_result)
2115
2116       results.append((idx, success, job_result))
2117
2118     # sort based on the index, then drop it
2119     results.sort()
2120     results = [i[1:] for i in results]
2121
2122     return results
2123
2124   def WaitOrShow(self, wait):
2125     """Wait for job results or only print the job IDs.
2126
2127     @type wait: boolean
2128     @param wait: whether to wait or not
2129
2130     """
2131     if wait:
2132       return self.GetResults()
2133     else:
2134       if not self.jobs:
2135         self.SubmitPending()
2136       for _, status, result, name in self.jobs:
2137         if status:
2138           ToStdout("%s: %s", result, name)
2139         else:
2140           ToStderr("Failure for %s: %s", name, result)