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

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2010, 2011, 2012 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 """Cluster related commands"""
  22
  23 # pylint: disable=W0401,W0613,W0614,C0103
  24 # W0401: Wildcard import ganeti.cli
  25 # W0613: Unused argument, since all functions follow the same API
  26 # W0614: Unused import %s from wildcard import (since we need cli)
  27 # C0103: Invalid name gnt-cluster
  28
  29 import os.path
  30 import time
  31 import OpenSSL
  32 import itertools
  33
  34 from ganeti.cli import *
  35 from ganeti import opcodes
  36 from ganeti import constants
  37 from ganeti import errors
  38 from ganeti import utils
  39 from ganeti import bootstrap
  40 from ganeti import ssh
  41 from ganeti import objects
  42 from ganeti import uidpool
  43 from ganeti import compat
  44 from ganeti import netutils
  45 from ganeti import pathutils
  46
  47
  48 ON_OPT = cli_option("--on", default=False,
  49                     action="store_true", dest="on",
  50                     help="Recover from an EPO")
  51
  52 GROUPS_OPT = cli_option("--groups", default=False,
  53                         action="store_true", dest="groups",
  54                         help="Arguments are node groups instead of nodes")
  55
  56 SHOW_MACHINE_OPT = cli_option("-M", "--show-machine-names", default=False,
  57                               action="store_true",
  58                               help="Show machine name for every line in output")
  59
  60 _EPO_PING_INTERVAL = 30 # 30 seconds between pings
  61 _EPO_PING_TIMEOUT = 1 # 1 second
  62 _EPO_REACHABLE_TIMEOUT = 15 * 60 # 15 minutes
  63
  64
  65 @UsesRPC
  66 def InitCluster(opts, args):
  67   """Initialize the cluster.
  68
  69   @param opts: the command line options selected by the user
  70   @type args: list
  71   @param args: should contain only one element, the desired
  72       cluster name
  73   @rtype: int
  74   @return: the desired exit code
  75
  76   """
  77   if not opts.lvm_storage and opts.vg_name:
  78     ToStderr("Options --no-lvm-storage and --vg-name conflict.")
  79     return 1
  80
  81   vg_name = opts.vg_name
  82   if opts.lvm_storage and not opts.vg_name:
  83     vg_name = constants.DEFAULT_VG
  84
  85   if not opts.drbd_storage and opts.drbd_helper:
  86     ToStderr("Options --no-drbd-storage and --drbd-usermode-helper conflict.")
  87     return 1
  88
  89   drbd_helper = opts.drbd_helper
  90   if opts.drbd_storage and not opts.drbd_helper:
  91     drbd_helper = constants.DEFAULT_DRBD_HELPER
  92
  93   master_netdev = opts.master_netdev
  94   if master_netdev is None:
  95     master_netdev = constants.DEFAULT_BRIDGE
  96
  97   hvlist = opts.enabled_hypervisors
  98   if hvlist is None:
  99     hvlist = constants.DEFAULT_ENABLED_HYPERVISOR
 100   hvlist = hvlist.split(",")
 101
 102   hvparams = dict(opts.hvparams)
 103   beparams = opts.beparams
 104   nicparams = opts.nicparams
 105
 106   diskparams = dict(opts.diskparams)
 107
 108   # check the disk template types here, as we cannot rely on the type check done
 109   # by the opcode parameter types
 110   diskparams_keys = set(diskparams.keys())
 111   if not (diskparams_keys <= constants.DISK_TEMPLATES):
 112     unknown = utils.NiceSort(diskparams_keys - constants.DISK_TEMPLATES)
 113     ToStderr("Disk templates unknown: %s" % utils.CommaJoin(unknown))
 114     return 1
 115
 116   # prepare beparams dict
 117   beparams = objects.FillDict(constants.BEC_DEFAULTS, beparams)
 118   utils.ForceDictType(beparams, constants.BES_PARAMETER_COMPAT)
 119
 120   # prepare nicparams dict
 121   nicparams = objects.FillDict(constants.NICC_DEFAULTS, nicparams)
 122   utils.ForceDictType(nicparams, constants.NICS_PARAMETER_TYPES)
 123
 124   # prepare ndparams dict
 125   if opts.ndparams is None:
 126     ndparams = dict(constants.NDC_DEFAULTS)
 127   else:
 128     ndparams = objects.FillDict(constants.NDC_DEFAULTS, opts.ndparams)
 129     utils.ForceDictType(ndparams, constants.NDS_PARAMETER_TYPES)
 130
 131   # prepare hvparams dict
 132   for hv in constants.HYPER_TYPES:
 133     if hv not in hvparams:
 134       hvparams[hv] = {}
 135     hvparams[hv] = objects.FillDict(constants.HVC_DEFAULTS[hv], hvparams[hv])
 136     utils.ForceDictType(hvparams[hv], constants.HVS_PARAMETER_TYPES)
 137
 138   # prepare diskparams dict
 139   for templ in constants.DISK_TEMPLATES:
 140     if templ not in diskparams:
 141       diskparams[templ] = {}
 142     diskparams[templ] = objects.FillDict(constants.DISK_DT_DEFAULTS[templ],
 143                                          diskparams[templ])
 144     utils.ForceDictType(diskparams[templ], constants.DISK_DT_TYPES)
 145
 146   # prepare ipolicy dict
 147   ipolicy_raw = CreateIPolicyFromOpts(
 148     ispecs_mem_size=opts.ispecs_mem_size,
 149     ispecs_cpu_count=opts.ispecs_cpu_count,
 150     ispecs_disk_count=opts.ispecs_disk_count,
 151     ispecs_disk_size=opts.ispecs_disk_size,
 152     ispecs_nic_count=opts.ispecs_nic_count,
 153     ipolicy_disk_templates=opts.ipolicy_disk_templates,
 154     ipolicy_vcpu_ratio=opts.ipolicy_vcpu_ratio,
 155     ipolicy_spindle_ratio=opts.ipolicy_spindle_ratio,
 156     fill_all=True)
 157   ipolicy = objects.FillIPolicy(constants.IPOLICY_DEFAULTS, ipolicy_raw)
 158
 159   if opts.candidate_pool_size is None:
 160     opts.candidate_pool_size = constants.MASTER_POOL_SIZE_DEFAULT
 161
 162   if opts.mac_prefix is None:
 163     opts.mac_prefix = constants.DEFAULT_MAC_PREFIX
 164
 165   uid_pool = opts.uid_pool
 166   if uid_pool is not None:
 167     uid_pool = uidpool.ParseUidPool(uid_pool)
 168
 169   if opts.prealloc_wipe_disks is None:
 170     opts.prealloc_wipe_disks = False
 171
 172   external_ip_setup_script = opts.use_external_mip_script
 173   if external_ip_setup_script is None:
 174     external_ip_setup_script = False
 175
 176   try:
 177     primary_ip_version = int(opts.primary_ip_version)
 178   except (ValueError, TypeError), err:
 179     ToStderr("Invalid primary ip version value: %s" % str(err))
 180     return 1
 181
 182   master_netmask = opts.master_netmask
 183   try:
 184     if master_netmask is not None:
 185       master_netmask = int(master_netmask)
 186   except (ValueError, TypeError), err:
 187     ToStderr("Invalid master netmask value: %s" % str(err))
 188     return 1
 189
 190   if opts.disk_state:
 191     disk_state = utils.FlatToDict(opts.disk_state)
 192   else:
 193     disk_state = {}
 194
 195   hv_state = dict(opts.hv_state)
 196
 197   bootstrap.InitCluster(cluster_name=args[0],
 198                         secondary_ip=opts.secondary_ip,
 199                         vg_name=vg_name,
 200                         mac_prefix=opts.mac_prefix,
 201                         master_netmask=master_netmask,
 202                         master_netdev=master_netdev,
 203                         file_storage_dir=opts.file_storage_dir,
 204                         shared_file_storage_dir=opts.shared_file_storage_dir,
 205                         enabled_hypervisors=hvlist,
 206                         hvparams=hvparams,
 207                         beparams=beparams,
 208                         nicparams=nicparams,
 209                         ndparams=ndparams,
 210                         diskparams=diskparams,
 211                         ipolicy=ipolicy,
 212                         candidate_pool_size=opts.candidate_pool_size,
 213                         modify_etc_hosts=opts.modify_etc_hosts,
 214                         modify_ssh_setup=opts.modify_ssh_setup,
 215                         maintain_node_health=opts.maintain_node_health,
 216                         drbd_helper=drbd_helper,
 217                         uid_pool=uid_pool,
 218                         default_iallocator=opts.default_iallocator,
 219                         primary_ip_version=primary_ip_version,
 220                         prealloc_wipe_disks=opts.prealloc_wipe_disks,
 221                         use_external_mip_script=external_ip_setup_script,
 222                         hv_state=hv_state,
 223                         disk_state=disk_state,
 224                         )
 225   op = opcodes.OpClusterPostInit()
 226   SubmitOpCode(op, opts=opts)
 227   return 0
 228
 229
 230 @UsesRPC
 231 def DestroyCluster(opts, args):
 232   """Destroy the cluster.
 233
 234   @param opts: the command line options selected by the user
 235   @type args: list
 236   @param args: should be an empty list
 237   @rtype: int
 238   @return: the desired exit code
 239
 240   """
 241   if not opts.yes_do_it:
 242     ToStderr("Destroying a cluster is irreversible. If you really want"
 243              " destroy this cluster, supply the --yes-do-it option.")
 244     return 1
 245
 246   op = opcodes.OpClusterDestroy()
 247   master = SubmitOpCode(op, opts=opts)
 248   # if we reached this, the opcode didn't fail; we can proceed to
 249   # shutdown all the daemons
 250   bootstrap.FinalizeClusterDestroy(master)
 251   return 0
 252
 253
 254 def RenameCluster(opts, args):
 255   """Rename the cluster.
 256
 257   @param opts: the command line options selected by the user
 258   @type args: list
 259   @param args: should contain only one element, the new cluster name
 260   @rtype: int
 261   @return: the desired exit code
 262
 263   """
 264   cl = GetClient()
 265
 266   (cluster_name, ) = cl.QueryConfigValues(["cluster_name"])
 267
 268   new_name = args[0]
 269   if not opts.force:
 270     usertext = ("This will rename the cluster from '%s' to '%s'. If you are"
 271                 " connected over the network to the cluster name, the"
 272                 " operation is very dangerous as the IP address will be"
 273                 " removed from the node and the change may not go through."
 274                 " Continue?") % (cluster_name, new_name)
 275     if not AskUser(usertext):
 276       return 1
 277
 278   op = opcodes.OpClusterRename(name=new_name)
 279   result = SubmitOpCode(op, opts=opts, cl=cl)
 280
 281   if result:
 282     ToStdout("Cluster renamed from '%s' to '%s'", cluster_name, result)
 283
 284   return 0
 285
 286
 287 def ActivateMasterIp(opts, args):
 288   """Activates the master IP.
 289
 290   """
 291   op = opcodes.OpClusterActivateMasterIp()
 292   SubmitOpCode(op)
 293   return 0
 294
 295
 296 def DeactivateMasterIp(opts, args):
 297   """Deactivates the master IP.
 298
 299   """
 300   if not opts.confirm:
 301     usertext = ("This will disable the master IP. All the open connections to"
 302                 " the master IP will be closed. To reach the master you will"
 303                 " need to use its node IP."
 304                 " Continue?")
 305     if not AskUser(usertext):
 306       return 1
 307
 308   op = opcodes.OpClusterDeactivateMasterIp()
 309   SubmitOpCode(op)
 310   return 0
 311
 312
 313 def RedistributeConfig(opts, args):
 314   """Forces push of the cluster configuration.
 315
 316   @param opts: the command line options selected by the user
 317   @type args: list
 318   @param args: empty list
 319   @rtype: int
 320   @return: the desired exit code
 321
 322   """
 323   op = opcodes.OpClusterRedistConf()
 324   SubmitOrSend(op, opts)
 325   return 0
 326
 327
 328 def ShowClusterVersion(opts, args):
 329   """Write version of ganeti software to the standard output.
 330
 331   @param opts: the command line options selected by the user
 332   @type args: list
 333   @param args: should be an empty list
 334   @rtype: int
 335   @return: the desired exit code
 336
 337   """
 338   cl = GetClient(query=True)
 339   result = cl.QueryClusterInfo()
 340   ToStdout("Software version: %s", result["software_version"])
 341   ToStdout("Internode protocol: %s", result["protocol_version"])
 342   ToStdout("Configuration format: %s", result["config_version"])
 343   ToStdout("OS api version: %s", result["os_api_version"])
 344   ToStdout("Export interface: %s", result["export_version"])
 345   return 0
 346
 347
 348 def ShowClusterMaster(opts, args):
 349   """Write name of master node to the standard output.
 350
 351   @param opts: the command line options selected by the user
 352   @type args: list
 353   @param args: should be an empty list
 354   @rtype: int
 355   @return: the desired exit code
 356
 357   """
 358   master = bootstrap.GetMaster()
 359   ToStdout(master)
 360   return 0
 361
 362
 363 def _PrintGroupedParams(paramsdict, level=1, roman=False):
 364   """Print Grouped parameters (be, nic, disk) by group.
 365
 366   @type paramsdict: dict of dicts
 367   @param paramsdict: {group: {param: value, ...}, ...}
 368   @type level: int
 369   @param level: Level of indention
 370
 371   """
 372   indent = "  " * level
 373   for item, val in sorted(paramsdict.items()):
 374     if isinstance(val, dict):
 375       ToStdout("%s- %s:", indent, item)
 376       _PrintGroupedParams(val, level=level + 1, roman=roman)
 377     elif roman and isinstance(val, int):
 378       ToStdout("%s  %s: %s", indent, item, compat.TryToRoman(val))
 379     else:
 380       ToStdout("%s  %s: %s", indent, item, val)
 381
 382
 383 def ShowClusterConfig(opts, args):
 384   """Shows cluster information.
 385
 386   @param opts: the command line options selected by the user
 387   @type args: list
 388   @param args: should be an empty list
 389   @rtype: int
 390   @return: the desired exit code
 391
 392   """
 393   cl = GetClient(query=True)
 394   result = cl.QueryClusterInfo()
 395
 396   ToStdout("Cluster name: %s", result["name"])
 397   ToStdout("Cluster UUID: %s", result["uuid"])
 398
 399   ToStdout("Creation time: %s", utils.FormatTime(result["ctime"]))
 400   ToStdout("Modification time: %s", utils.FormatTime(result["mtime"]))
 401
 402   ToStdout("Master node: %s", result["master"])
 403
 404   ToStdout("Architecture (this node): %s (%s)",
 405            result["architecture"][0], result["architecture"][1])
 406
 407   if result["tags"]:
 408     tags = utils.CommaJoin(utils.NiceSort(result["tags"]))
 409   else:
 410     tags = "(none)"
 411
 412   ToStdout("Tags: %s", tags)
 413
 414   ToStdout("Default hypervisor: %s", result["default_hypervisor"])
 415   ToStdout("Enabled hypervisors: %s",
 416            utils.CommaJoin(result["enabled_hypervisors"]))
 417
 418   ToStdout("Hypervisor parameters:")
 419   _PrintGroupedParams(result["hvparams"])
 420
 421   ToStdout("OS-specific hypervisor parameters:")
 422   _PrintGroupedParams(result["os_hvp"])
 423
 424   ToStdout("OS parameters:")
 425   _PrintGroupedParams(result["osparams"])
 426
 427   ToStdout("Hidden OSes: %s", utils.CommaJoin(result["hidden_os"]))
 428   ToStdout("Blacklisted OSes: %s", utils.CommaJoin(result["blacklisted_os"]))
 429
 430   ToStdout("Cluster parameters:")
 431   ToStdout("  - candidate pool size: %s",
 432             compat.TryToRoman(result["candidate_pool_size"],
 433                               convert=opts.roman_integers))
 434   ToStdout("  - master netdev: %s", result["master_netdev"])
 435   ToStdout("  - master netmask: %s", result["master_netmask"])
 436   ToStdout("  - use external master IP address setup script: %s",
 437            result["use_external_mip_script"])
 438   ToStdout("  - lvm volume group: %s", result["volume_group_name"])
 439   if result["reserved_lvs"]:
 440     reserved_lvs = utils.CommaJoin(result["reserved_lvs"])
 441   else:
 442     reserved_lvs = "(none)"
 443   ToStdout("  - lvm reserved volumes: %s", reserved_lvs)
 444   ToStdout("  - drbd usermode helper: %s", result["drbd_usermode_helper"])
 445   ToStdout("  - file storage path: %s", result["file_storage_dir"])
 446   ToStdout("  - shared file storage path: %s",
 447            result["shared_file_storage_dir"])
 448   ToStdout("  - maintenance of node health: %s",
 449            result["maintain_node_health"])
 450   ToStdout("  - uid pool: %s",
 451             uidpool.FormatUidPool(result["uid_pool"],
 452                                   roman=opts.roman_integers))
 453   ToStdout("  - default instance allocator: %s", result["default_iallocator"])
 454   ToStdout("  - primary ip version: %d", result["primary_ip_version"])
 455   ToStdout("  - preallocation wipe disks: %s", result["prealloc_wipe_disks"])
 456   ToStdout("  - OS search path: %s", utils.CommaJoin(pathutils.OS_SEARCH_PATH))
 457
 458   ToStdout("Default node parameters:")
 459   _PrintGroupedParams(result["ndparams"], roman=opts.roman_integers)
 460
 461   ToStdout("Default instance parameters:")
 462   _PrintGroupedParams(result["beparams"], roman=opts.roman_integers)
 463
 464   ToStdout("Default nic parameters:")
 465   _PrintGroupedParams(result["nicparams"], roman=opts.roman_integers)
 466
 467   ToStdout("Default disk parameters:")
 468   _PrintGroupedParams(result["diskparams"], roman=opts.roman_integers)
 469
 470   ToStdout("Instance policy - limits for instances:")
 471   for key in constants.IPOLICY_ISPECS:
 472     ToStdout("  - %s", key)
 473     _PrintGroupedParams(result["ipolicy"][key], roman=opts.roman_integers)
 474   ToStdout("  - enabled disk templates: %s",
 475            utils.CommaJoin(result["ipolicy"][constants.IPOLICY_DTS]))
 476   for key in constants.IPOLICY_PARAMETERS:
 477     ToStdout("  - %s: %s", key, result["ipolicy"][key])
 478
 479   return 0
 480
 481
 482 def ClusterCopyFile(opts, args):
 483   """Copy a file from master to some nodes.
 484
 485   @param opts: the command line options selected by the user
 486   @type args: list
 487   @param args: should contain only one element, the path of
 488       the file to be copied
 489   @rtype: int
 490   @return: the desired exit code
 491
 492   """
 493   filename = args[0]
 494   if not os.path.exists(filename):
 495     raise errors.OpPrereqError("No such filename '%s'" % filename,
 496                                errors.ECODE_INVAL)
 497
 498   cl = GetClient()
 499
 500   cluster_name = cl.QueryConfigValues(["cluster_name"])[0]
 501
 502   results = GetOnlineNodes(nodes=opts.nodes, cl=cl, filter_master=True,
 503                            secondary_ips=opts.use_replication_network,
 504                            nodegroup=opts.nodegroup)
 505
 506   srun = ssh.SshRunner(cluster_name=cluster_name)
 507   for node in results:
 508     if not srun.CopyFileToNode(node, filename):
 509       ToStderr("Copy of file %s to node %s failed", filename, node)
 510
 511   return 0
 512
 513
 514 def RunClusterCommand(opts, args):
 515   """Run a command on some nodes.
 516
 517   @param opts: the command line options selected by the user
 518   @type args: list
 519   @param args: should contain the command to be run and its arguments
 520   @rtype: int
 521   @return: the desired exit code
 522
 523   """
 524   cl = GetClient()
 525
 526   command = " ".join(args)
 527
 528   nodes = GetOnlineNodes(nodes=opts.nodes, cl=cl, nodegroup=opts.nodegroup)
 529
 530   cluster_name, master_node = cl.QueryConfigValues(["cluster_name",
 531                                                     "master_node"])
 532
 533   srun = ssh.SshRunner(cluster_name=cluster_name)
 534
 535   # Make sure master node is at list end
 536   if master_node in nodes:
 537     nodes.remove(master_node)
 538     nodes.append(master_node)
 539
 540   for name in nodes:
 541     result = srun.Run(name, constants.SSH_LOGIN_USER, command)
 542     ToStdout("------------------------------------------------")
 543     if opts.show_machine_names:
 544       for line in result.output.splitlines():
 545         ToStdout("%s: %s", name, line)
 546     else:
 547       ToStdout("node: %s", name)
 548       ToStdout("%s", result.output)
 549     ToStdout("return code = %s", result.exit_code)
 550
 551   return 0
 552
 553
 554 def VerifyCluster(opts, args):
 555   """Verify integrity of cluster, performing various test on nodes.
 556
 557   @param opts: the command line options selected by the user
 558   @type args: list
 559   @param args: should be an empty list
 560   @rtype: int
 561   @return: the desired exit code
 562
 563   """
 564   skip_checks = []
 565
 566   if opts.skip_nplusone_mem:
 567     skip_checks.append(constants.VERIFY_NPLUSONE_MEM)
 568
 569   cl = GetClient()
 570
 571   op = opcodes.OpClusterVerify(verbose=opts.verbose,
 572                                error_codes=opts.error_codes,
 573                                debug_simulate_errors=opts.simulate_errors,
 574                                skip_checks=skip_checks,
 575                                ignore_errors=opts.ignore_errors,
 576                                group_name=opts.nodegroup)
 577   result = SubmitOpCode(op, cl=cl, opts=opts)
 578
 579   # Keep track of submitted jobs
 580   jex = JobExecutor(cl=cl, opts=opts)
 581
 582   for (status, job_id) in result[constants.JOB_IDS_KEY]:
 583     jex.AddJobId(None, status, job_id)
 584
 585   results = jex.GetResults()
 586
 587   (bad_jobs, bad_results) = \
 588     map(len,
 589         # Convert iterators to lists
 590         map(list,
 591             # Count errors
 592             map(compat.partial(itertools.ifilterfalse, bool),
 593                 # Convert result to booleans in a tuple
 594                 zip(*((job_success, len(op_results) == 1 and op_results[0])
 595                       for (job_success, op_results) in results)))))
 596
 597   if bad_jobs == 0 and bad_results == 0:
 598     rcode = constants.EXIT_SUCCESS
 599   else:
 600     rcode = constants.EXIT_FAILURE
 601     if bad_jobs > 0:
 602       ToStdout("%s job(s) failed while verifying the cluster.", bad_jobs)
 603
 604   return rcode
 605
 606
 607 def VerifyDisks(opts, args):
 608   """Verify integrity of cluster disks.
 609
 610   @param opts: the command line options selected by the user
 611   @type args: list
 612   @param args: should be an empty list
 613   @rtype: int
 614   @return: the desired exit code
 615
 616   """
 617   cl = GetClient()
 618
 619   op = opcodes.OpClusterVerifyDisks()
 620
 621   result = SubmitOpCode(op, cl=cl, opts=opts)
 622
 623   # Keep track of submitted jobs
 624   jex = JobExecutor(cl=cl, opts=opts)
 625
 626   for (status, job_id) in result[constants.JOB_IDS_KEY]:
 627     jex.AddJobId(None, status, job_id)
 628
 629   retcode = constants.EXIT_SUCCESS
 630
 631   for (status, result) in jex.GetResults():
 632     if not status:
 633       ToStdout("Job failed: %s", result)
 634       continue
 635
 636     ((bad_nodes, instances, missing), ) = result
 637
 638     for node, text in bad_nodes.items():
 639       ToStdout("Error gathering data on node %s: %s",
 640                node, utils.SafeEncode(text[-400:]))
 641       retcode = constants.EXIT_FAILURE
 642       ToStdout("You need to fix these nodes first before fixing instances")
 643
 644     for iname in instances:
 645       if iname in missing:
 646         continue
 647       op = opcodes.OpInstanceActivateDisks(instance_name=iname)
 648       try:
 649         ToStdout("Activating disks for instance '%s'", iname)
 650         SubmitOpCode(op, opts=opts, cl=cl)
 651       except errors.GenericError, err:
 652         nret, msg = FormatError(err)
 653         retcode |= nret
 654         ToStderr("Error activating disks for instance %s: %s", iname, msg)
 655
 656     if missing:
 657       for iname, ival in missing.iteritems():
 658         all_missing = compat.all(x[0] in bad_nodes for x in ival)
 659         if all_missing:
 660           ToStdout("Instance %s cannot be verified as it lives on"
 661                    " broken nodes", iname)
 662         else:
 663           ToStdout("Instance %s has missing logical volumes:", iname)
 664           ival.sort()
 665           for node, vol in ival:
 666             if node in bad_nodes:
 667               ToStdout("\tbroken node %s /dev/%s", node, vol)
 668             else:
 669               ToStdout("\t%s /dev/%s", node, vol)
 670
 671       ToStdout("You need to replace or recreate disks for all the above"
 672                " instances if this message persists after fixing broken nodes.")
 673       retcode = constants.EXIT_FAILURE
 674     elif not instances:
 675       ToStdout("No disks need to be activated.")
 676
 677   return retcode
 678
 679
 680 def RepairDiskSizes(opts, args):
 681   """Verify sizes of cluster disks.
 682
 683   @param opts: the command line options selected by the user
 684   @type args: list
 685   @param args: optional list of instances to restrict check to
 686   @rtype: int
 687   @return: the desired exit code
 688
 689   """
 690   op = opcodes.OpClusterRepairDiskSizes(instances=args)
 691   SubmitOpCode(op, opts=opts)
 692
 693
 694 @UsesRPC
 695 def MasterFailover(opts, args):
 696   """Failover the master node.
 697
 698   This command, when run on a non-master node, will cause the current
 699   master to cease being master, and the non-master to become new
 700   master.
 701
 702   @param opts: the command line options selected by the user
 703   @type args: list
 704   @param args: should be an empty list
 705   @rtype: int
 706   @return: the desired exit code
 707
 708   """
 709   if opts.no_voting:
 710     usertext = ("This will perform the failover even if most other nodes"
 711                 " are down, or if this node is outdated. This is dangerous"
 712                 " as it can lead to a non-consistent cluster. Check the"
 713                 " gnt-cluster(8) man page before proceeding. Continue?")
 714     if not AskUser(usertext):
 715       return 1
 716
 717   return bootstrap.MasterFailover(no_voting=opts.no_voting)
 718
 719
 720 def MasterPing(opts, args):
 721   """Checks if the master is alive.
 722
 723   @param opts: the command line options selected by the user
 724   @type args: list
 725   @param args: should be an empty list
 726   @rtype: int
 727   @return: the desired exit code
 728
 729   """
 730   try:
 731     cl = GetClient()
 732     cl.QueryClusterInfo()
 733     return 0
 734   except Exception: # pylint: disable=W0703
 735     return 1
 736
 737
 738 def SearchTags(opts, args):
 739   """Searches the tags on all the cluster.
 740
 741   @param opts: the command line options selected by the user
 742   @type args: list
 743   @param args: should contain only one element, the tag pattern
 744   @rtype: int
 745   @return: the desired exit code
 746
 747   """
 748   op = opcodes.OpTagsSearch(pattern=args[0])
 749   result = SubmitOpCode(op, opts=opts)
 750   if not result:
 751     return 1
 752   result = list(result)
 753   result.sort()
 754   for path, tag in result:
 755     ToStdout("%s %s", path, tag)
 756
 757
 758 def _ReadAndVerifyCert(cert_filename, verify_private_key=False):
 759   """Reads and verifies an X509 certificate.
 760
 761   @type cert_filename: string
 762   @param cert_filename: the path of the file containing the certificate to
 763                         verify encoded in PEM format
 764   @type verify_private_key: bool
 765   @param verify_private_key: whether to verify the private key in addition to
 766                              the public certificate
 767   @rtype: string
 768   @return: a string containing the PEM-encoded certificate.
 769
 770   """
 771   try:
 772     pem = utils.ReadFile(cert_filename)
 773   except IOError, err:
 774     raise errors.X509CertError(cert_filename,
 775                                "Unable to read certificate: %s" % str(err))
 776
 777   try:
 778     OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, pem)
 779   except Exception, err:
 780     raise errors.X509CertError(cert_filename,
 781                                "Unable to load certificate: %s" % str(err))
 782
 783   if verify_private_key:
 784     try:
 785       OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, pem)
 786     except Exception, err:
 787       raise errors.X509CertError(cert_filename,
 788                                  "Unable to load private key: %s" % str(err))
 789
 790   return pem
 791
 792
 793 def _RenewCrypto(new_cluster_cert, new_rapi_cert, # pylint: disable=R0911
 794                  rapi_cert_filename, new_spice_cert, spice_cert_filename,
 795                  spice_cacert_filename, new_confd_hmac_key, new_cds,
 796                  cds_filename, force):
 797   """Renews cluster certificates, keys and secrets.
 798
 799   @type new_cluster_cert: bool
 800   @param new_cluster_cert: Whether to generate a new cluster certificate
 801   @type new_rapi_cert: bool
 802   @param new_rapi_cert: Whether to generate a new RAPI certificate
 803   @type rapi_cert_filename: string
 804   @param rapi_cert_filename: Path to file containing new RAPI certificate
 805   @type new_spice_cert: bool
 806   @param new_spice_cert: Whether to generate a new SPICE certificate
 807   @type spice_cert_filename: string
 808   @param spice_cert_filename: Path to file containing new SPICE certificate
 809   @type spice_cacert_filename: string
 810   @param spice_cacert_filename: Path to file containing the certificate of the
 811                                 CA that signed the SPICE certificate
 812   @type new_confd_hmac_key: bool
 813   @param new_confd_hmac_key: Whether to generate a new HMAC key
 814   @type new_cds: bool
 815   @param new_cds: Whether to generate a new cluster domain secret
 816   @type cds_filename: string
 817   @param cds_filename: Path to file containing new cluster domain secret
 818   @type force: bool
 819   @param force: Whether to ask user for confirmation
 820
 821   """
 822   if new_rapi_cert and rapi_cert_filename:
 823     ToStderr("Only one of the --new-rapi-certificate and --rapi-certificate"
 824              " options can be specified at the same time.")
 825     return 1
 826
 827   if new_cds and cds_filename:
 828     ToStderr("Only one of the --new-cluster-domain-secret and"
 829              " --cluster-domain-secret options can be specified at"
 830              " the same time.")
 831     return 1
 832
 833   if new_spice_cert and (spice_cert_filename or spice_cacert_filename):
 834     ToStderr("When using --new-spice-certificate, the --spice-certificate"
 835              " and --spice-ca-certificate must not be used.")
 836     return 1
 837
 838   if bool(spice_cacert_filename) ^ bool(spice_cert_filename):
 839     ToStderr("Both --spice-certificate and --spice-ca-certificate must be"
 840              " specified.")
 841     return 1
 842
 843   rapi_cert_pem, spice_cert_pem, spice_cacert_pem = (None, None, None)
 844   try:
 845     if rapi_cert_filename:
 846       rapi_cert_pem = _ReadAndVerifyCert(rapi_cert_filename, True)
 847     if spice_cert_filename:
 848       spice_cert_pem = _ReadAndVerifyCert(spice_cert_filename, True)
 849       spice_cacert_pem = _ReadAndVerifyCert(spice_cacert_filename)
 850   except errors.X509CertError, err:
 851     ToStderr("Unable to load X509 certificate from %s: %s", err[0], err[1])
 852     return 1
 853
 854   if cds_filename:
 855     try:
 856       cds = utils.ReadFile(cds_filename)
 857     except Exception, err: # pylint: disable=W0703
 858       ToStderr("Can't load new cluster domain secret from %s: %s" %
 859                (cds_filename, str(err)))
 860       return 1
 861   else:
 862     cds = None
 863
 864   if not force:
 865     usertext = ("This requires all daemons on all nodes to be restarted and"
 866                 " may take some time. Continue?")
 867     if not AskUser(usertext):
 868       return 1
 869
 870   def _RenewCryptoInner(ctx):
 871     ctx.feedback_fn("Updating certificates and keys")
 872     bootstrap.GenerateClusterCrypto(new_cluster_cert,
 873                                     new_rapi_cert,
 874                                     new_spice_cert,
 875                                     new_confd_hmac_key,
 876                                     new_cds,
 877                                     rapi_cert_pem=rapi_cert_pem,
 878                                     spice_cert_pem=spice_cert_pem,
 879                                     spice_cacert_pem=spice_cacert_pem,
 880                                     cds=cds)
 881
 882     files_to_copy = []
 883
 884     if new_cluster_cert:
 885       files_to_copy.append(pathutils.NODED_CERT_FILE)
 886
 887     if new_rapi_cert or rapi_cert_pem:
 888       files_to_copy.append(pathutils.RAPI_CERT_FILE)
 889
 890     if new_spice_cert or spice_cert_pem:
 891       files_to_copy.append(pathutils.SPICE_CERT_FILE)
 892       files_to_copy.append(pathutils.SPICE_CACERT_FILE)
 893
 894     if new_confd_hmac_key:
 895       files_to_copy.append(pathutils.CONFD_HMAC_KEY)
 896
 897     if new_cds or cds:
 898       files_to_copy.append(pathutils.CLUSTER_DOMAIN_SECRET_FILE)
 899
 900     if files_to_copy:
 901       for node_name in ctx.nonmaster_nodes:
 902         ctx.feedback_fn("Copying %s to %s" %
 903                         (", ".join(files_to_copy), node_name))
 904         for file_name in files_to_copy:
 905           ctx.ssh.CopyFileToNode(node_name, file_name)
 906
 907   RunWhileClusterStopped(ToStdout, _RenewCryptoInner)
 908
 909   ToStdout("All requested certificates and keys have been replaced."
 910            " Running \"gnt-cluster verify\" now is recommended.")
 911
 912   return 0
 913
 914
 915 def RenewCrypto(opts, args):
 916   """Renews cluster certificates, keys and secrets.
 917
 918   """
 919   return _RenewCrypto(opts.new_cluster_cert,
 920                       opts.new_rapi_cert,
 921                       opts.rapi_cert,
 922                       opts.new_spice_cert,
 923                       opts.spice_cert,
 924                       opts.spice_cacert,
 925                       opts.new_confd_hmac_key,
 926                       opts.new_cluster_domain_secret,
 927                       opts.cluster_domain_secret,
 928                       opts.force)
 929
 930
 931 def SetClusterParams(opts, args):
 932   """Modify the cluster.
 933
 934   @param opts: the command line options selected by the user
 935   @type args: list
 936   @param args: should be an empty list
 937   @rtype: int
 938   @return: the desired exit code
 939
 940   """
 941   if not (not opts.lvm_storage or opts.vg_name or
 942           not opts.drbd_storage or opts.drbd_helper or
 943           opts.enabled_hypervisors or opts.hvparams or
 944           opts.beparams or opts.nicparams or
 945           opts.ndparams or opts.diskparams or
 946           opts.candidate_pool_size is not None or
 947           opts.uid_pool is not None or
 948           opts.maintain_node_health is not None or
 949           opts.add_uids is not None or
 950           opts.remove_uids is not None or
 951           opts.default_iallocator is not None or
 952           opts.reserved_lvs is not None or
 953           opts.master_netdev is not None or
 954           opts.master_netmask is not None or
 955           opts.use_external_mip_script is not None or
 956           opts.prealloc_wipe_disks is not None or
 957           opts.hv_state or
 958           opts.disk_state or
 959           opts.ispecs_mem_size or
 960           opts.ispecs_cpu_count or
 961           opts.ispecs_disk_count or
 962           opts.ispecs_disk_size or
 963           opts.ispecs_nic_count or
 964           opts.ipolicy_disk_templates is not None or
 965           opts.ipolicy_vcpu_ratio is not None or
 966           opts.ipolicy_spindle_ratio is not None):
 967     ToStderr("Please give at least one of the parameters.")
 968     return 1
 969
 970   vg_name = opts.vg_name
 971   if not opts.lvm_storage and opts.vg_name:
 972     ToStderr("Options --no-lvm-storage and --vg-name conflict.")
 973     return 1
 974
 975   if not opts.lvm_storage:
 976     vg_name = ""
 977
 978   drbd_helper = opts.drbd_helper
 979   if not opts.drbd_storage and opts.drbd_helper:
 980     ToStderr("Options --no-drbd-storage and --drbd-usermode-helper conflict.")
 981     return 1
 982
 983   if not opts.drbd_storage:
 984     drbd_helper = ""
 985
 986   hvlist = opts.enabled_hypervisors
 987   if hvlist is not None:
 988     hvlist = hvlist.split(",")
 989
 990   # a list of (name, dict) we can pass directly to dict() (or [])
 991   hvparams = dict(opts.hvparams)
 992   for hv_params in hvparams.values():
 993     utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
 994
 995   diskparams = dict(opts.diskparams)
 996
 997   for dt_params in diskparams.values():
 998     utils.ForceDictType(dt_params, constants.DISK_DT_TYPES)
 999
1000   beparams = opts.beparams
1001   utils.ForceDictType(beparams, constants.BES_PARAMETER_COMPAT)
1002
1003   nicparams = opts.nicparams
1004   utils.ForceDictType(nicparams, constants.NICS_PARAMETER_TYPES)
1005
1006   ndparams = opts.ndparams
1007   if ndparams is not None:
1008     utils.ForceDictType(ndparams, constants.NDS_PARAMETER_TYPES)
1009
1010   ipolicy = CreateIPolicyFromOpts(
1011     ispecs_mem_size=opts.ispecs_mem_size,
1012     ispecs_cpu_count=opts.ispecs_cpu_count,
1013     ispecs_disk_count=opts.ispecs_disk_count,
1014     ispecs_disk_size=opts.ispecs_disk_size,
1015     ispecs_nic_count=opts.ispecs_nic_count,
1016     ipolicy_disk_templates=opts.ipolicy_disk_templates,
1017     ipolicy_vcpu_ratio=opts.ipolicy_vcpu_ratio,
1018     ipolicy_spindle_ratio=opts.ipolicy_spindle_ratio,
1019     )
1020
1021   mnh = opts.maintain_node_health
1022
1023   uid_pool = opts.uid_pool
1024   if uid_pool is not None:
1025     uid_pool = uidpool.ParseUidPool(uid_pool)
1026
1027   add_uids = opts.add_uids
1028   if add_uids is not None:
1029     add_uids = uidpool.ParseUidPool(add_uids)
1030
1031   remove_uids = opts.remove_uids
1032   if remove_uids is not None:
1033     remove_uids = uidpool.ParseUidPool(remove_uids)
1034
1035   if opts.reserved_lvs is not None:
1036     if opts.reserved_lvs == "":
1037       opts.reserved_lvs = []
1038     else:
1039       opts.reserved_lvs = utils.UnescapeAndSplit(opts.reserved_lvs, sep=",")
1040
1041   if opts.master_netmask is not None:
1042     try:
1043       opts.master_netmask = int(opts.master_netmask)
1044     except ValueError:
1045       ToStderr("The --master-netmask option expects an int parameter.")
1046       return 1
1047
1048   ext_ip_script = opts.use_external_mip_script
1049
1050   if opts.disk_state:
1051     disk_state = utils.FlatToDict(opts.disk_state)
1052   else:
1053     disk_state = {}
1054
1055   hv_state = dict(opts.hv_state)
1056
1057   op = opcodes.OpClusterSetParams(vg_name=vg_name,
1058                                   drbd_helper=drbd_helper,
1059                                   enabled_hypervisors=hvlist,
1060                                   hvparams=hvparams,
1061                                   os_hvp=None,
1062                                   beparams=beparams,
1063                                   nicparams=nicparams,
1064                                   ndparams=ndparams,
1065                                   diskparams=diskparams,
1066                                   ipolicy=ipolicy,
1067                                   candidate_pool_size=opts.candidate_pool_size,
1068                                   maintain_node_health=mnh,
1069                                   uid_pool=uid_pool,
1070                                   add_uids=add_uids,
1071                                   remove_uids=remove_uids,
1072                                   default_iallocator=opts.default_iallocator,
1073                                   prealloc_wipe_disks=opts.prealloc_wipe_disks,
1074                                   master_netdev=opts.master_netdev,
1075                                   master_netmask=opts.master_netmask,
1076                                   reserved_lvs=opts.reserved_lvs,
1077                                   use_external_mip_script=ext_ip_script,
1078                                   hv_state=hv_state,
1079                                   disk_state=disk_state,
1080                                   )
1081   SubmitOrSend(op, opts)
1082   return 0
1083
1084
1085 def QueueOps(opts, args):
1086   """Queue operations.
1087
1088   @param opts: the command line options selected by the user
1089   @type args: list
1090   @param args: should contain only one element, the subcommand
1091   @rtype: int
1092   @return: the desired exit code
1093
1094   """
1095   command = args[0]
1096   client = GetClient()
1097   if command in ("drain", "undrain"):
1098     drain_flag = command == "drain"
1099     client.SetQueueDrainFlag(drain_flag)
1100   elif command == "info":
1101     result = client.QueryConfigValues(["drain_flag"])
1102     if result[0]:
1103       val = "set"
1104     else:
1105       val = "unset"
1106     ToStdout("The drain flag is %s" % val)
1107   else:
1108     raise errors.OpPrereqError("Command '%s' is not valid." % command,
1109                                errors.ECODE_INVAL)
1110
1111   return 0
1112
1113
1114 def _ShowWatcherPause(until):
1115   if until is None or until < time.time():
1116     ToStdout("The watcher is not paused.")
1117   else:
1118     ToStdout("The watcher is paused until %s.", time.ctime(until))
1119
1120
1121 def WatcherOps(opts, args):
1122   """Watcher operations.
1123
1124   @param opts: the command line options selected by the user
1125   @type args: list
1126   @param args: should contain only one element, the subcommand
1127   @rtype: int
1128   @return: the desired exit code
1129
1130   """
1131   command = args[0]
1132   client = GetClient()
1133
1134   if command == "continue":
1135     client.SetWatcherPause(None)
1136     ToStdout("The watcher is no longer paused.")
1137
1138   elif command == "pause":
1139     if len(args) < 2:
1140       raise errors.OpPrereqError("Missing pause duration", errors.ECODE_INVAL)
1141
1142     result = client.SetWatcherPause(time.time() + ParseTimespec(args[1]))
1143     _ShowWatcherPause(result)
1144
1145   elif command == "info":
1146     result = client.QueryConfigValues(["watcher_pause"])
1147     _ShowWatcherPause(result[0])
1148
1149   else:
1150     raise errors.OpPrereqError("Command '%s' is not valid." % command,
1151                                errors.ECODE_INVAL)
1152
1153   return 0
1154
1155
1156 def _OobPower(opts, node_list, power):
1157   """Puts the node in the list to desired power state.
1158
1159   @param opts: The command line options selected by the user
1160   @param node_list: The list of nodes to operate on
1161   @param power: True if they should be powered on, False otherwise
1162   @return: The success of the operation (none failed)
1163
1164   """
1165   if power:
1166     command = constants.OOB_POWER_ON
1167   else:
1168     command = constants.OOB_POWER_OFF
1169
1170   op = opcodes.OpOobCommand(node_names=node_list,
1171                             command=command,
1172                             ignore_status=True,
1173                             timeout=opts.oob_timeout,
1174                             power_delay=opts.power_delay)
1175   result = SubmitOpCode(op, opts=opts)
1176   errs = 0
1177   for node_result in result:
1178     (node_tuple, data_tuple) = node_result
1179     (_, node_name) = node_tuple
1180     (data_status, _) = data_tuple
1181     if data_status != constants.RS_NORMAL:
1182       assert data_status != constants.RS_UNAVAIL
1183       errs += 1
1184       ToStderr("There was a problem changing power for %s, please investigate",
1185                node_name)
1186
1187   if errs > 0:
1188     return False
1189
1190   return True
1191
1192
1193 def _InstanceStart(opts, inst_list, start, no_remember=False):
1194   """Puts the instances in the list to desired state.
1195
1196   @param opts: The command line options selected by the user
1197   @param inst_list: The list of instances to operate on
1198   @param start: True if they should be started, False for shutdown
1199   @param no_remember: If the instance state should be remembered
1200   @return: The success of the operation (none failed)
1201
1202   """
1203   if start:
1204     opcls = opcodes.OpInstanceStartup
1205     text_submit, text_success, text_failed = ("startup", "started", "starting")
1206   else:
1207     opcls = compat.partial(opcodes.OpInstanceShutdown,
1208                            timeout=opts.shutdown_timeout,
1209                            no_remember=no_remember)
1210     text_submit, text_success, text_failed = ("shutdown", "stopped", "stopping")
1211
1212   jex = JobExecutor(opts=opts)
1213
1214   for inst in inst_list:
1215     ToStdout("Submit %s of instance %s", text_submit, inst)
1216     op = opcls(instance_name=inst)
1217     jex.QueueJob(inst, op)
1218
1219   results = jex.GetResults()
1220   bad_cnt = len([1 for (success, _) in results if not success])
1221
1222   if bad_cnt == 0:
1223     ToStdout("All instances have been %s successfully", text_success)
1224   else:
1225     ToStderr("There were errors while %s instances:\n"
1226              "%d error(s) out of %d instance(s)", text_failed, bad_cnt,
1227              len(results))
1228     return False
1229
1230   return True
1231
1232
1233 class _RunWhenNodesReachableHelper:
1234   """Helper class to make shared internal state sharing easier.
1235
1236   @ivar success: Indicates if all action_cb calls were successful
1237
1238   """
1239   def __init__(self, node_list, action_cb, node2ip, port, feedback_fn,
1240                _ping_fn=netutils.TcpPing, _sleep_fn=time.sleep):
1241     """Init the object.
1242
1243     @param node_list: The list of nodes to be reachable
1244     @param action_cb: Callback called when a new host is reachable
1245     @type node2ip: dict
1246     @param node2ip: Node to ip mapping
1247     @param port: The port to use for the TCP ping
1248     @param feedback_fn: The function used for feedback
1249     @param _ping_fn: Function to check reachabilty (for unittest use only)
1250     @param _sleep_fn: Function to sleep (for unittest use only)
1251
1252     """
1253     self.down = set(node_list)
1254     self.up = set()
1255     self.node2ip = node2ip
1256     self.success = True
1257     self.action_cb = action_cb
1258     self.port = port
1259     self.feedback_fn = feedback_fn
1260     self._ping_fn = _ping_fn
1261     self._sleep_fn = _sleep_fn
1262
1263   def __call__(self):
1264     """When called we run action_cb.
1265
1266     @raises utils.RetryAgain: When there are still down nodes
1267
1268     """
1269     if not self.action_cb(self.up):
1270       self.success = False
1271
1272     if self.down:
1273       raise utils.RetryAgain()
1274     else:
1275       return self.success
1276
1277   def Wait(self, secs):
1278     """Checks if a host is up or waits remaining seconds.
1279
1280     @param secs: The secs remaining
1281
1282     """
1283     start = time.time()
1284     for node in self.down:
1285       if self._ping_fn(self.node2ip[node], self.port, timeout=_EPO_PING_TIMEOUT,
1286                        live_port_needed=True):
1287         self.feedback_fn("Node %s became available" % node)
1288         self.up.add(node)
1289         self.down -= self.up
1290         # If we have a node available there is the possibility to run the
1291         # action callback successfully, therefore we don't wait and return
1292         return
1293
1294     self._sleep_fn(max(0.0, start + secs - time.time()))
1295
1296
1297 def _RunWhenNodesReachable(node_list, action_cb, interval):
1298   """Run action_cb when nodes become reachable.
1299
1300   @param node_list: The list of nodes to be reachable
1301   @param action_cb: Callback called when a new host is reachable
1302   @param interval: The earliest time to retry
1303
1304   """
1305   client = GetClient()
1306   cluster_info = client.QueryClusterInfo()
1307   if cluster_info["primary_ip_version"] == constants.IP4_VERSION:
1308     family = netutils.IPAddress.family
1309   else:
1310     family = netutils.IP6Address.family
1311
1312   node2ip = dict((node, netutils.GetHostname(node, family=family).ip)
1313                  for node in node_list)
1314
1315   port = netutils.GetDaemonPort(constants.NODED)
1316   helper = _RunWhenNodesReachableHelper(node_list, action_cb, node2ip, port,
1317                                         ToStdout)
1318
1319   try:
1320     return utils.Retry(helper, interval, _EPO_REACHABLE_TIMEOUT,
1321                        wait_fn=helper.Wait)
1322   except utils.RetryTimeout:
1323     ToStderr("Time exceeded while waiting for nodes to become reachable"
1324              " again:\n  - %s", "  - ".join(helper.down))
1325     return False
1326
1327
1328 def _MaybeInstanceStartup(opts, inst_map, nodes_online,
1329                           _instance_start_fn=_InstanceStart):
1330   """Start the instances conditional based on node_states.
1331
1332   @param opts: The command line options selected by the user
1333   @param inst_map: A dict of inst -> nodes mapping
1334   @param nodes_online: A list of nodes online
1335   @param _instance_start_fn: Callback to start instances (unittest use only)
1336   @return: Success of the operation on all instances
1337
1338   """
1339   start_inst_list = []
1340   for (inst, nodes) in inst_map.items():
1341     if not (nodes - nodes_online):
1342       # All nodes the instance lives on are back online
1343       start_inst_list.append(inst)
1344
1345   for inst in start_inst_list:
1346     del inst_map[inst]
1347
1348   if start_inst_list:
1349     return _instance_start_fn(opts, start_inst_list, True)
1350
1351   return True
1352
1353
1354 def _EpoOn(opts, full_node_list, node_list, inst_map):
1355   """Does the actual power on.
1356
1357   @param opts: The command line options selected by the user
1358   @param full_node_list: All nodes to operate on (includes nodes not supporting
1359                          OOB)
1360   @param node_list: The list of nodes to operate on (all need to support OOB)
1361   @param inst_map: A dict of inst -> nodes mapping
1362   @return: The desired exit status
1363
1364   """
1365   if node_list and not _OobPower(opts, node_list, False):
1366     ToStderr("Not all nodes seem to get back up, investigate and start"
1367              " manually if needed")
1368
1369   # Wait for the nodes to be back up
1370   action_cb = compat.partial(_MaybeInstanceStartup, opts, dict(inst_map))
1371
1372   ToStdout("Waiting until all nodes are available again")
1373   if not _RunWhenNodesReachable(full_node_list, action_cb, _EPO_PING_INTERVAL):
1374     ToStderr("Please investigate and start stopped instances manually")
1375     return constants.EXIT_FAILURE
1376
1377   return constants.EXIT_SUCCESS
1378
1379
1380 def _EpoOff(opts, node_list, inst_map):
1381   """Does the actual power off.
1382
1383   @param opts: The command line options selected by the user
1384   @param node_list: The list of nodes to operate on (all need to support OOB)
1385   @param inst_map: A dict of inst -> nodes mapping
1386   @return: The desired exit status
1387
1388   """
1389   if not _InstanceStart(opts, inst_map.keys(), False, no_remember=True):
1390     ToStderr("Please investigate and stop instances manually before continuing")
1391     return constants.EXIT_FAILURE
1392
1393   if not node_list:
1394     return constants.EXIT_SUCCESS
1395
1396   if _OobPower(opts, node_list, False):
1397     return constants.EXIT_SUCCESS
1398   else:
1399     return constants.EXIT_FAILURE
1400
1401
1402 def Epo(opts, args, cl=None, _on_fn=_EpoOn, _off_fn=_EpoOff,
1403         _confirm_fn=ConfirmOperation,
1404         _stdout_fn=ToStdout, _stderr_fn=ToStderr):
1405   """EPO operations.
1406
1407   @param opts: the command line options selected by the user
1408   @type args: list
1409   @param args: should contain only one element, the subcommand
1410   @rtype: int
1411   @return: the desired exit code
1412
1413   """
1414   if opts.groups and opts.show_all:
1415     _stderr_fn("Only one of --groups or --all are allowed")
1416     return constants.EXIT_FAILURE
1417   elif args and opts.show_all:
1418     _stderr_fn("Arguments in combination with --all are not allowed")
1419     return constants.EXIT_FAILURE
1420
1421   if cl is None:
1422     cl = GetClient()
1423
1424   if opts.groups:
1425     node_query_list = \
1426       itertools.chain(*cl.QueryGroups(args, ["node_list"], False))
1427   else:
1428     node_query_list = args
1429
1430   result = cl.QueryNodes(node_query_list, ["name", "master", "pinst_list",
1431                                            "sinst_list", "powered", "offline"],
1432                          False)
1433
1434   all_nodes = map(compat.fst, result)
1435   node_list = []
1436   inst_map = {}
1437   for (node, master, pinsts, sinsts, powered, offline) in result:
1438     if not offline:
1439       for inst in (pinsts + sinsts):
1440         if inst in inst_map:
1441           if not master:
1442             inst_map[inst].add(node)
1443         elif master:
1444           inst_map[inst] = set()
1445         else:
1446           inst_map[inst] = set([node])
1447
1448     if master and opts.on:
1449       # We ignore the master for turning on the machines, in fact we are
1450       # already operating on the master at this point :)
1451       continue
1452     elif master and not opts.show_all:
1453       _stderr_fn("%s is the master node, please do a master-failover to another"
1454                  " node not affected by the EPO or use --all if you intend to"
1455                  " shutdown the whole cluster", node)
1456       return constants.EXIT_FAILURE
1457     elif powered is None:
1458       _stdout_fn("Node %s does not support out-of-band handling, it can not be"
1459                  " handled in a fully automated manner", node)
1460     elif powered == opts.on:
1461       _stdout_fn("Node %s is already in desired power state, skipping", node)
1462     elif not offline or (offline and powered):
1463       node_list.append(node)
1464
1465   if not (opts.force or _confirm_fn(all_nodes, "nodes", "epo")):
1466     return constants.EXIT_FAILURE
1467
1468   if opts.on:
1469     return _on_fn(opts, all_nodes, node_list, inst_map)
1470   else:
1471     return _off_fn(opts, node_list, inst_map)
1472
1473
1474 commands = {
1475   "init": (
1476     InitCluster, [ArgHost(min=1, max=1)],
1477     [BACKEND_OPT, CP_SIZE_OPT, ENABLED_HV_OPT, GLOBAL_FILEDIR_OPT,
1478      HVLIST_OPT, MAC_PREFIX_OPT, MASTER_NETDEV_OPT, MASTER_NETMASK_OPT,
1479      NIC_PARAMS_OPT, NOLVM_STORAGE_OPT, NOMODIFY_ETCHOSTS_OPT,
1480      NOMODIFY_SSH_SETUP_OPT, SECONDARY_IP_OPT, VG_NAME_OPT,
1481      MAINTAIN_NODE_HEALTH_OPT, UIDPOOL_OPT, DRBD_HELPER_OPT, NODRBD_STORAGE_OPT,
1482      DEFAULT_IALLOCATOR_OPT, PRIMARY_IP_VERSION_OPT, PREALLOC_WIPE_DISKS_OPT,
1483      NODE_PARAMS_OPT, GLOBAL_SHARED_FILEDIR_OPT, USE_EXTERNAL_MIP_SCRIPT,
1484      DISK_PARAMS_OPT, HV_STATE_OPT, DISK_STATE_OPT] + INSTANCE_POLICY_OPTS,
1485     "[opts...] <cluster_name>", "Initialises a new cluster configuration"),
1486   "destroy": (
1487     DestroyCluster, ARGS_NONE, [YES_DOIT_OPT],
1488     "", "Destroy cluster"),
1489   "rename": (
1490     RenameCluster, [ArgHost(min=1, max=1)],
1491     [FORCE_OPT, DRY_RUN_OPT],
1492     "<new_name>",
1493     "Renames the cluster"),
1494   "redist-conf": (
1495     RedistributeConfig, ARGS_NONE, [SUBMIT_OPT, DRY_RUN_OPT, PRIORITY_OPT],
1496     "", "Forces a push of the configuration file and ssconf files"
1497     " to the nodes in the cluster"),
1498   "verify": (
1499     VerifyCluster, ARGS_NONE,
1500     [VERBOSE_OPT, DEBUG_SIMERR_OPT, ERROR_CODES_OPT, NONPLUS1_OPT,
1501      DRY_RUN_OPT, PRIORITY_OPT, NODEGROUP_OPT, IGNORE_ERRORS_OPT],
1502     "", "Does a check on the cluster configuration"),
1503   "verify-disks": (
1504     VerifyDisks, ARGS_NONE, [PRIORITY_OPT],
1505     "", "Does a check on the cluster disk status"),
1506   "repair-disk-sizes": (
1507     RepairDiskSizes, ARGS_MANY_INSTANCES, [DRY_RUN_OPT, PRIORITY_OPT],
1508     "[instance...]", "Updates mismatches in recorded disk sizes"),
1509   "master-failover": (
1510     MasterFailover, ARGS_NONE, [NOVOTING_OPT],
1511     "", "Makes the current node the master"),
1512   "master-ping": (
1513     MasterPing, ARGS_NONE, [],
1514     "", "Checks if the master is alive"),
1515   "version": (
1516     ShowClusterVersion, ARGS_NONE, [],
1517     "", "Shows the cluster version"),
1518   "getmaster": (
1519     ShowClusterMaster, ARGS_NONE, [],
1520     "", "Shows the cluster master"),
1521   "copyfile": (
1522     ClusterCopyFile, [ArgFile(min=1, max=1)],
1523     [NODE_LIST_OPT, USE_REPL_NET_OPT, NODEGROUP_OPT],
1524     "[-n node...] <filename>", "Copies a file to all (or only some) nodes"),
1525   "command": (
1526     RunClusterCommand, [ArgCommand(min=1)],
1527     [NODE_LIST_OPT, NODEGROUP_OPT, SHOW_MACHINE_OPT],
1528     "[-n node...] <command>", "Runs a command on all (or only some) nodes"),
1529   "info": (
1530     ShowClusterConfig, ARGS_NONE, [ROMAN_OPT],
1531     "[--roman]", "Show cluster configuration"),
1532   "list-tags": (
1533     ListTags, ARGS_NONE, [], "", "List the tags of the cluster"),
1534   "add-tags": (
1535     AddTags, [ArgUnknown()], [TAG_SRC_OPT, PRIORITY_OPT, SUBMIT_OPT],
1536     "tag...", "Add tags to the cluster"),
1537   "remove-tags": (
1538     RemoveTags, [ArgUnknown()], [TAG_SRC_OPT, PRIORITY_OPT, SUBMIT_OPT],
1539     "tag...", "Remove tags from the cluster"),
1540   "search-tags": (
1541     SearchTags, [ArgUnknown(min=1, max=1)], [PRIORITY_OPT], "",
1542     "Searches the tags on all objects on"
1543     " the cluster for a given pattern (regex)"),
1544   "queue": (
1545     QueueOps,
1546     [ArgChoice(min=1, max=1, choices=["drain", "undrain", "info"])],
1547     [], "drain|undrain|info", "Change queue properties"),
1548   "watcher": (
1549     WatcherOps,
1550     [ArgChoice(min=1, max=1, choices=["pause", "continue", "info"]),
1551      ArgSuggest(min=0, max=1, choices=["30m", "1h", "4h"])],
1552     [],
1553     "{pause <timespec>|continue|info}", "Change watcher properties"),
1554   "modify": (
1555     SetClusterParams, ARGS_NONE,
1556     [BACKEND_OPT, CP_SIZE_OPT, ENABLED_HV_OPT, HVLIST_OPT, MASTER_NETDEV_OPT,
1557      MASTER_NETMASK_OPT, NIC_PARAMS_OPT, NOLVM_STORAGE_OPT, VG_NAME_OPT,
1558      MAINTAIN_NODE_HEALTH_OPT, UIDPOOL_OPT, ADD_UIDS_OPT, REMOVE_UIDS_OPT,
1559      DRBD_HELPER_OPT, NODRBD_STORAGE_OPT, DEFAULT_IALLOCATOR_OPT,
1560      RESERVED_LVS_OPT, DRY_RUN_OPT, PRIORITY_OPT, PREALLOC_WIPE_DISKS_OPT,
1561      NODE_PARAMS_OPT, USE_EXTERNAL_MIP_SCRIPT, DISK_PARAMS_OPT, HV_STATE_OPT,
1562      DISK_STATE_OPT, SUBMIT_OPT] +
1563     INSTANCE_POLICY_OPTS,
1564     "[opts...]",
1565     "Alters the parameters of the cluster"),
1566   "renew-crypto": (
1567     RenewCrypto, ARGS_NONE,
1568     [NEW_CLUSTER_CERT_OPT, NEW_RAPI_CERT_OPT, RAPI_CERT_OPT,
1569      NEW_CONFD_HMAC_KEY_OPT, FORCE_OPT,
1570      NEW_CLUSTER_DOMAIN_SECRET_OPT, CLUSTER_DOMAIN_SECRET_OPT,
1571      NEW_SPICE_CERT_OPT, SPICE_CERT_OPT, SPICE_CACERT_OPT],
1572     "[opts...]",
1573     "Renews cluster certificates, keys and secrets"),
1574   "epo": (
1575     Epo, [ArgUnknown()],
1576     [FORCE_OPT, ON_OPT, GROUPS_OPT, ALL_OPT, OOB_TIMEOUT_OPT,
1577      SHUTDOWN_TIMEOUT_OPT, POWER_DELAY_OPT],
1578     "[opts...] [args]",
1579     "Performs an emergency power-off on given args"),
1580   "activate-master-ip": (
1581     ActivateMasterIp, ARGS_NONE, [], "", "Activates the master IP"),
1582   "deactivate-master-ip": (
1583     DeactivateMasterIp, ARGS_NONE, [CONFIRM_OPT], "",
1584     "Deactivates the master IP"),
1585   }
1586
1587
1588 #: dictionary with aliases for commands
1589 aliases = {
1590   "masterfailover": "master-failover",
1591   "show": "info",
1592 }
1593
1594
1595 def Main():
1596   return GenericMain(commands, override={"tag_type": constants.TAG_CLUSTER},
1597                      aliases=aliases)