code.grnet.gr Git - ganeti-local/blob - tools/cluster-merge

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 2010, 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 """Tool to merge two or more clusters together.
  22
  23 The clusters have to run the same version of Ganeti!
  24
  25 """
  26
  27 # pylint: disable=C0103
  28 # C0103: Invalid name cluster-merge
  29
  30 import logging
  31 import os
  32 import optparse
  33 import shutil
  34 import sys
  35 import tempfile
  36
  37 from ganeti import cli
  38 from ganeti import config
  39 from ganeti import constants
  40 from ganeti import errors
  41 from ganeti import ssh
  42 from ganeti import utils
  43 from ganeti import pathutils
  44 from ganeti import compat
  45
  46
  47 _GROUPS_MERGE = "merge"
  48 _GROUPS_RENAME = "rename"
  49 _CLUSTERMERGE_ECID = "clustermerge-ecid"
  50 _RESTART_ALL = "all"
  51 _RESTART_UP = "up"
  52 _RESTART_NONE = "none"
  53 _RESTART_CHOICES = (_RESTART_ALL, _RESTART_UP, _RESTART_NONE)
  54 _PARAMS_STRICT = "strict"
  55 _PARAMS_WARN = "warn"
  56 _PARAMS_CHOICES = (_PARAMS_STRICT, _PARAMS_WARN)
  57
  58
  59 PAUSE_PERIOD_OPT = cli.cli_option("-p", "--watcher-pause-period", default=1800,
  60                                   action="store", type="int",
  61                                   dest="pause_period",
  62                                   help=("Amount of time in seconds watcher"
  63                                         " should be suspended from running"))
  64 GROUPS_OPT = cli.cli_option("--groups", default=None, metavar="STRATEGY",
  65                             choices=(_GROUPS_MERGE, _GROUPS_RENAME),
  66                             dest="groups",
  67                             help=("How to handle groups that have the"
  68                                   " same name (One of: %s/%s)" %
  69                                   (_GROUPS_MERGE, _GROUPS_RENAME)))
  70 PARAMS_OPT = cli.cli_option("--parameter-conflicts", default=_PARAMS_STRICT,
  71                             metavar="STRATEGY",
  72                             choices=_PARAMS_CHOICES,
  73                             dest="params",
  74                             help=("How to handle params that have"
  75                                   " different values (One of: %s/%s)" %
  76                                   _PARAMS_CHOICES))
  77
  78 RESTART_OPT = cli.cli_option("--restart", default=_RESTART_ALL,
  79                              metavar="STRATEGY",
  80                              choices=_RESTART_CHOICES,
  81                              dest="restart",
  82                              help=("How to handle restarting instances"
  83                                    " same name (One of: %s/%s/%s)" %
  84                                    _RESTART_CHOICES))
  85
  86 SKIP_STOP_INSTANCES_OPT = \
  87   cli.cli_option("--skip-stop-instances", default=True, action="store_false",
  88                  dest="stop_instances",
  89                  help=("Don't stop the instances on the clusters, just check "
  90                        "that none is running"))
  91
  92
  93 def Flatten(unflattened_list):
  94   """Flattens a list.
  95
  96   @param unflattened_list: A list of unflattened list objects.
  97   @return: A flattened list
  98
  99   """
 100   flattened_list = []
 101
 102   for item in unflattened_list:
 103     if isinstance(item, list):
 104       flattened_list.extend(Flatten(item))
 105     else:
 106       flattened_list.append(item)
 107   return flattened_list
 108
 109
 110 class MergerData(object):
 111   """Container class to hold data used for merger.
 112
 113   """
 114   def __init__(self, cluster, key_path, nodes, instances, master_node,
 115                config_path=None):
 116     """Initialize the container.
 117
 118     @param cluster: The name of the cluster
 119     @param key_path: Path to the ssh private key used for authentication
 120     @param nodes: List of online nodes in the merging cluster
 121     @param instances: List of instances running on merging cluster
 122     @param master_node: Name of the master node
 123     @param config_path: Path to the merging cluster config
 124
 125     """
 126     self.cluster = cluster
 127     self.key_path = key_path
 128     self.nodes = nodes
 129     self.instances = instances
 130     self.master_node = master_node
 131     self.config_path = config_path
 132
 133
 134 class Merger(object):
 135   """Handling the merge.
 136
 137   """
 138   RUNNING_STATUSES = compat.UniqueFrozenset([
 139     constants.INSTST_RUNNING,
 140     constants.INSTST_ERRORUP,
 141     ])
 142
 143   def __init__(self, clusters, pause_period, groups, restart, params,
 144                stop_instances):
 145     """Initialize object with sane defaults and infos required.
 146
 147     @param clusters: The list of clusters to merge in
 148     @param pause_period: The time watcher shall be disabled for
 149     @param groups: How to handle group conflicts
 150     @param restart: How to handle instance restart
 151     @param stop_instances: Indicates whether the instances must be stopped
 152                            (True) or if the Merger must only check if no
 153                            instances are running on the mergee clusters (False)
 154
 155     """
 156     self.merger_data = []
 157     self.clusters = clusters
 158     self.pause_period = pause_period
 159     self.work_dir = tempfile.mkdtemp(suffix="cluster-merger")
 160     (self.cluster_name, ) = cli.GetClient().QueryConfigValues(["cluster_name"])
 161     self.ssh_runner = ssh.SshRunner(self.cluster_name)
 162     self.groups = groups
 163     self.restart = restart
 164     self.params = params
 165     self.stop_instances = stop_instances
 166     if self.restart == _RESTART_UP:
 167       raise NotImplementedError
 168
 169   def Setup(self):
 170     """Sets up our end so we can do the merger.
 171
 172     This method is setting us up as a preparation for the merger.
 173     It makes the initial contact and gathers information needed.
 174
 175     @raise errors.RemoteError: for errors in communication/grabbing
 176
 177     """
 178     (remote_path, _, _) = ssh.GetUserFiles("root")
 179
 180     if self.cluster_name in self.clusters:
 181       raise errors.CommandError("Cannot merge cluster %s with itself" %
 182                                 self.cluster_name)
 183
 184     # Fetch remotes private key
 185     for cluster in self.clusters:
 186       result = self._RunCmd(cluster, "cat %s" % remote_path, batch=False,
 187                             ask_key=False)
 188       if result.failed:
 189         raise errors.RemoteError("There was an error while grabbing ssh private"
 190                                  " key from %s. Fail reason: %s; output: %s" %
 191                                  (cluster, result.fail_reason, result.output))
 192
 193       key_path = utils.PathJoin(self.work_dir, cluster)
 194       utils.WriteFile(key_path, mode=0600, data=result.stdout)
 195
 196       result = self._RunCmd(cluster, "gnt-node list -o name,offline"
 197                             " --no-headers --separator=,", private_key=key_path)
 198       if result.failed:
 199         raise errors.RemoteError("Unable to retrieve list of nodes from %s."
 200                                  " Fail reason: %s; output: %s" %
 201                                  (cluster, result.fail_reason, result.output))
 202       nodes_statuses = [line.split(",") for line in result.stdout.splitlines()]
 203       nodes = [node_status[0] for node_status in nodes_statuses
 204                if node_status[1] == "N"]
 205
 206       result = self._RunCmd(cluster, "gnt-instance list -o name --no-headers",
 207                             private_key=key_path)
 208       if result.failed:
 209         raise errors.RemoteError("Unable to retrieve list of instances from"
 210                                  " %s. Fail reason: %s; output: %s" %
 211                                  (cluster, result.fail_reason, result.output))
 212       instances = result.stdout.splitlines()
 213
 214       path = utils.PathJoin(pathutils.DATA_DIR, "ssconf_%s" %
 215                             constants.SS_MASTER_NODE)
 216       result = self._RunCmd(cluster, "cat %s" % path, private_key=key_path)
 217       if result.failed:
 218         raise errors.RemoteError("Unable to retrieve the master node name from"
 219                                  " %s. Fail reason: %s; output: %s" %
 220                                  (cluster, result.fail_reason, result.output))
 221       master_node = result.stdout.strip()
 222
 223       self.merger_data.append(MergerData(cluster, key_path, nodes, instances,
 224                                          master_node))
 225
 226   def _PrepareAuthorizedKeys(self):
 227     """Prepare the authorized_keys on every merging node.
 228
 229     This method add our public key to remotes authorized_key for further
 230     communication.
 231
 232     """
 233     (_, pub_key_file, auth_keys) = ssh.GetUserFiles("root")
 234     pub_key = utils.ReadFile(pub_key_file)
 235
 236     for data in self.merger_data:
 237       for node in data.nodes:
 238         result = self._RunCmd(node, ("cat >> %s << '!EOF.'\n%s!EOF.\n" %
 239                                      (auth_keys, pub_key)),
 240                               private_key=data.key_path, max_attempts=3)
 241
 242         if result.failed:
 243           raise errors.RemoteError("Unable to add our public key to %s in %s."
 244                                    " Fail reason: %s; output: %s" %
 245                                    (node, data.cluster, result.fail_reason,
 246                                     result.output))
 247
 248   def _RunCmd(self, hostname, command, user="root", use_cluster_key=False,
 249               strict_host_check=False, private_key=None, batch=True,
 250               ask_key=False, max_attempts=1):
 251     """Wrapping SshRunner.Run with default parameters.
 252
 253     For explanation of parameters see L{ganeti.ssh.SshRunner.Run}.
 254
 255     """
 256     for _ in range(max_attempts):
 257       result = self.ssh_runner.Run(hostname=hostname, command=command,
 258                                    user=user, use_cluster_key=use_cluster_key,
 259                                    strict_host_check=strict_host_check,
 260                                    private_key=private_key, batch=batch,
 261                                    ask_key=ask_key)
 262       if not result.failed:
 263         break
 264
 265     return result
 266
 267   def _CheckRunningInstances(self):
 268     """Checks if on the clusters to be merged there are running instances
 269
 270     @rtype: boolean
 271     @return: True if there are running instances, False otherwise
 272
 273     """
 274     for cluster in self.clusters:
 275       result = self._RunCmd(cluster, "gnt-instance list -o status")
 276       if self.RUNNING_STATUSES.intersection(result.output.splitlines()):
 277         return True
 278
 279     return False
 280
 281   def _StopMergingInstances(self):
 282     """Stop instances on merging clusters.
 283
 284     """
 285     for cluster in self.clusters:
 286       result = self._RunCmd(cluster, "gnt-instance shutdown --all"
 287                                      " --force-multiple")
 288
 289       if result.failed:
 290         raise errors.RemoteError("Unable to stop instances on %s."
 291                                  " Fail reason: %s; output: %s" %
 292                                  (cluster, result.fail_reason, result.output))
 293
 294   def _DisableWatcher(self):
 295     """Disable watch on all merging clusters, including ourself.
 296
 297     """
 298     for cluster in ["localhost"] + self.clusters:
 299       result = self._RunCmd(cluster, "gnt-cluster watcher pause %d" %
 300                                      self.pause_period)
 301
 302       if result.failed:
 303         raise errors.RemoteError("Unable to pause watcher on %s."
 304                                  " Fail reason: %s; output: %s" %
 305                                  (cluster, result.fail_reason, result.output))
 306
 307   def _RemoveMasterIps(self):
 308     """Removes the master IPs from the master nodes of each cluster.
 309
 310     """
 311     for data in self.merger_data:
 312       result = self._RunCmd(data.master_node,
 313                             "gnt-cluster deactivate-master-ip --yes")
 314
 315       if result.failed:
 316         raise errors.RemoteError("Unable to remove master IP on %s."
 317                                  " Fail reason: %s; output: %s" %
 318                                  (data.master_node,
 319                                   result.fail_reason,
 320                                   result.output))
 321
 322   def _StopDaemons(self):
 323     """Stop all daemons on merging nodes.
 324
 325     """
 326     cmd = "%s stop-all" % pathutils.DAEMON_UTIL
 327     for data in self.merger_data:
 328       for node in data.nodes:
 329         result = self._RunCmd(node, cmd, max_attempts=3)
 330
 331         if result.failed:
 332           raise errors.RemoteError("Unable to stop daemons on %s."
 333                                    " Fail reason: %s; output: %s." %
 334                                    (node, result.fail_reason, result.output))
 335
 336   def _FetchRemoteConfig(self):
 337     """Fetches and stores remote cluster config from the master.
 338
 339     This step is needed before we can merge the config.
 340
 341     """
 342     for data in self.merger_data:
 343       result = self._RunCmd(data.cluster, "cat %s" %
 344                                           pathutils.CLUSTER_CONF_FILE)
 345
 346       if result.failed:
 347         raise errors.RemoteError("Unable to retrieve remote config on %s."
 348                                  " Fail reason: %s; output %s" %
 349                                  (data.cluster, result.fail_reason,
 350                                   result.output))
 351
 352       data.config_path = utils.PathJoin(self.work_dir, "%s_config.data" %
 353                                         data.cluster)
 354       utils.WriteFile(data.config_path, data=result.stdout)
 355
 356   # R0201: Method could be a function
 357   def _KillMasterDaemon(self): # pylint: disable=R0201
 358     """Kills the local master daemon.
 359
 360     @raise errors.CommandError: If unable to kill
 361
 362     """
 363     result = utils.RunCmd([pathutils.DAEMON_UTIL, "stop-master"])
 364     if result.failed:
 365       raise errors.CommandError("Unable to stop master daemons."
 366                                 " Fail reason: %s; output: %s" %
 367                                 (result.fail_reason, result.output))
 368
 369   def _MergeConfig(self):
 370     """Merges all foreign config into our own config.
 371
 372     """
 373     my_config = config.ConfigWriter(offline=True)
 374     fake_ec_id = 0 # Needs to be uniq over the whole config merge
 375
 376     for data in self.merger_data:
 377       other_config = config.ConfigWriter(data.config_path, accept_foreign=True)
 378       self._MergeClusterConfigs(my_config, other_config)
 379       self._MergeNodeGroups(my_config, other_config)
 380
 381       for node in other_config.GetNodeList():
 382         node_info = other_config.GetNodeInfo(node)
 383         # Offline the node, it will be reonlined later at node readd
 384         node_info.master_candidate = False
 385         node_info.drained = False
 386         node_info.offline = True
 387         my_config.AddNode(node_info, _CLUSTERMERGE_ECID + str(fake_ec_id))
 388         fake_ec_id += 1
 389
 390       for instance in other_config.GetInstanceList():
 391         instance_info = other_config.GetInstanceInfo(instance)
 392
 393         # Update the DRBD port assignments
 394         # This is a little bit hackish
 395         for dsk in instance_info.disks:
 396           if dsk.dev_type in constants.LDS_DRBD:
 397             port = my_config.AllocatePort()
 398
 399             logical_id = list(dsk.logical_id)
 400             logical_id[2] = port
 401             dsk.logical_id = tuple(logical_id)
 402
 403             physical_id = list(dsk.physical_id)
 404             physical_id[1] = physical_id[3] = port
 405             dsk.physical_id = tuple(physical_id)
 406
 407         my_config.AddInstance(instance_info,
 408                               _CLUSTERMERGE_ECID + str(fake_ec_id))
 409         fake_ec_id += 1
 410
 411   def _MergeClusterConfigs(self, my_config, other_config):
 412     """Checks that all relevant cluster parameters are compatible
 413
 414     """
 415     my_cluster = my_config.GetClusterInfo()
 416     other_cluster = other_config.GetClusterInfo()
 417     err_count = 0
 418
 419     #
 420     # Generic checks
 421     #
 422     check_params = [
 423       "beparams",
 424       "default_iallocator",
 425       "drbd_usermode_helper",
 426       "hidden_os",
 427       "maintain_node_health",
 428       "master_netdev",
 429       "ndparams",
 430       "nicparams",
 431       "primary_ip_family",
 432       "tags",
 433       "uid_pool",
 434       ]
 435     check_params_strict = [
 436       "volume_group_name",
 437     ]
 438     if constants.ENABLE_FILE_STORAGE:
 439       check_params_strict.append("file_storage_dir")
 440     if constants.ENABLE_SHARED_FILE_STORAGE:
 441       check_params_strict.append("shared_file_storage_dir")
 442     check_params.extend(check_params_strict)
 443
 444     if self.params == _PARAMS_STRICT:
 445       params_strict = True
 446     else:
 447       params_strict = False
 448
 449     for param_name in check_params:
 450       my_param = getattr(my_cluster, param_name)
 451       other_param = getattr(other_cluster, param_name)
 452       if my_param != other_param:
 453         logging.error("The value (%s) of the cluster parameter %s on %s"
 454                       " differs to this cluster's value (%s)",
 455                       other_param, param_name, other_cluster.cluster_name,
 456                       my_param)
 457         if params_strict or param_name in check_params_strict:
 458           err_count += 1
 459
 460     #
 461     # Custom checks
 462     #
 463
 464     # Check default hypervisor
 465     my_defhyp = my_cluster.enabled_hypervisors[0]
 466     other_defhyp = other_cluster.enabled_hypervisors[0]
 467     if my_defhyp != other_defhyp:
 468       logging.warning("The default hypervisor (%s) differs on %s, new"
 469                       " instances will be created with this cluster's"
 470                       " default hypervisor (%s)", other_defhyp,
 471                       other_cluster.cluster_name, my_defhyp)
 472
 473     if (set(my_cluster.enabled_hypervisors) !=
 474         set(other_cluster.enabled_hypervisors)):
 475       logging.error("The set of enabled hypervisors (%s) on %s differs to"
 476                     " this cluster's set (%s)",
 477                     other_cluster.enabled_hypervisors,
 478                     other_cluster.cluster_name, my_cluster.enabled_hypervisors)
 479       err_count += 1
 480
 481     # Check hypervisor params for hypervisors we care about
 482     for hyp in my_cluster.enabled_hypervisors:
 483       for param in my_cluster.hvparams[hyp]:
 484         my_value = my_cluster.hvparams[hyp][param]
 485         other_value = other_cluster.hvparams[hyp][param]
 486         if my_value != other_value:
 487           logging.error("The value (%s) of the %s parameter of the %s"
 488                         " hypervisor on %s differs to this cluster's parameter"
 489                         " (%s)",
 490                         other_value, param, hyp, other_cluster.cluster_name,
 491                         my_value)
 492           if params_strict:
 493             err_count += 1
 494
 495     # Check os hypervisor params for hypervisors we care about
 496     for os_name in set(my_cluster.os_hvp.keys() + other_cluster.os_hvp.keys()):
 497       for hyp in my_cluster.enabled_hypervisors:
 498         my_os_hvp = self._GetOsHypervisor(my_cluster, os_name, hyp)
 499         other_os_hvp = self._GetOsHypervisor(other_cluster, os_name, hyp)
 500         if my_os_hvp != other_os_hvp:
 501           logging.error("The OS parameters (%s) for the %s OS for the %s"
 502                         " hypervisor on %s differs to this cluster's parameters"
 503                         " (%s)",
 504                         other_os_hvp, os_name, hyp, other_cluster.cluster_name,
 505                         my_os_hvp)
 506           if params_strict:
 507             err_count += 1
 508
 509     #
 510     # Warnings
 511     #
 512     if my_cluster.modify_etc_hosts != other_cluster.modify_etc_hosts:
 513       logging.warning("The modify_etc_hosts value (%s) differs on %s,"
 514                       " this cluster's value (%s) will take precedence",
 515                       other_cluster.modify_etc_hosts,
 516                       other_cluster.cluster_name,
 517                       my_cluster.modify_etc_hosts)
 518
 519     if my_cluster.modify_ssh_setup != other_cluster.modify_ssh_setup:
 520       logging.warning("The modify_ssh_setup value (%s) differs on %s,"
 521                       " this cluster's value (%s) will take precedence",
 522                       other_cluster.modify_ssh_setup,
 523                       other_cluster.cluster_name,
 524                       my_cluster.modify_ssh_setup)
 525
 526     #
 527     # Actual merging
 528     #
 529     my_cluster.reserved_lvs = list(set(my_cluster.reserved_lvs +
 530                                        other_cluster.reserved_lvs))
 531
 532     if my_cluster.prealloc_wipe_disks != other_cluster.prealloc_wipe_disks:
 533       logging.warning("The prealloc_wipe_disks value (%s) on %s differs to this"
 534                       " cluster's value (%s). The least permissive value (%s)"
 535                       " will be used", other_cluster.prealloc_wipe_disks,
 536                       other_cluster.cluster_name,
 537                       my_cluster.prealloc_wipe_disks, True)
 538       my_cluster.prealloc_wipe_disks = True
 539
 540     for os_, osparams in other_cluster.osparams.items():
 541       if os_ not in my_cluster.osparams:
 542         my_cluster.osparams[os_] = osparams
 543       elif my_cluster.osparams[os_] != osparams:
 544         logging.error("The OS parameters (%s) for the %s OS on %s differs to"
 545                       " this cluster's parameters (%s)",
 546                       osparams, os_, other_cluster.cluster_name,
 547                       my_cluster.osparams[os_])
 548         if params_strict:
 549           err_count += 1
 550
 551     if err_count:
 552       raise errors.ConfigurationError("Cluster config for %s has incompatible"
 553                                       " values, please fix and re-run" %
 554                                       other_cluster.cluster_name)
 555
 556   # R0201: Method could be a function
 557   def _GetOsHypervisor(self, cluster, os_name, hyp): # pylint: disable=R0201
 558     if os_name in cluster.os_hvp:
 559       return cluster.os_hvp[os_name].get(hyp, None)
 560     else:
 561       return None
 562
 563   # R0201: Method could be a function
 564   def _MergeNodeGroups(self, my_config, other_config):
 565     """Adds foreign node groups
 566
 567     ConfigWriter.AddNodeGroup takes care of making sure there are no conflicts.
 568     """
 569     # pylint: disable=R0201
 570     logging.info("Node group conflict strategy: %s", self.groups)
 571
 572     my_grps = my_config.GetAllNodeGroupsInfo().values()
 573     other_grps = other_config.GetAllNodeGroupsInfo().values()
 574
 575     # Check for node group naming conflicts:
 576     conflicts = []
 577     for other_grp in other_grps:
 578       for my_grp in my_grps:
 579         if other_grp.name == my_grp.name:
 580           conflicts.append(other_grp)
 581
 582     if conflicts:
 583       conflict_names = utils.CommaJoin([g.name for g in conflicts])
 584       logging.info("Node groups in both local and remote cluster: %s",
 585                    conflict_names)
 586
 587       # User hasn't specified how to handle conflicts
 588       if not self.groups:
 589         raise errors.CommandError("The following node group(s) are in both"
 590                                   " clusters, and no merge strategy has been"
 591                                   " supplied (see the --groups option): %s" %
 592                                   conflict_names)
 593
 594       # User wants to rename conflicts
 595       elif self.groups == _GROUPS_RENAME:
 596         for grp in conflicts:
 597           new_name = "%s-%s" % (grp.name, other_config.GetClusterName())
 598           logging.info("Renaming remote node group from %s to %s"
 599                        " to resolve conflict", grp.name, new_name)
 600           grp.name = new_name
 601
 602       # User wants to merge conflicting groups
 603       elif self.groups == _GROUPS_MERGE:
 604         for other_grp in conflicts:
 605           logging.info("Merging local and remote '%s' groups", other_grp.name)
 606           for node_name in other_grp.members[:]:
 607             node = other_config.GetNodeInfo(node_name)
 608             # Access to a protected member of a client class
 609             # pylint: disable=W0212
 610             other_config._UnlockedRemoveNodeFromGroup(node)
 611
 612             # Access to a protected member of a client class
 613             # pylint: disable=W0212
 614             my_grp_uuid = my_config._UnlockedLookupNodeGroup(other_grp.name)
 615
 616             # Access to a protected member of a client class
 617             # pylint: disable=W0212
 618             my_config._UnlockedAddNodeToGroup(node, my_grp_uuid)
 619             node.group = my_grp_uuid
 620           # Remove from list of groups to add
 621           other_grps.remove(other_grp)
 622
 623     for grp in other_grps:
 624       #TODO: handle node group conflicts
 625       my_config.AddNodeGroup(grp, _CLUSTERMERGE_ECID)
 626
 627   # R0201: Method could be a function
 628   def _StartMasterDaemon(self, no_vote=False): # pylint: disable=R0201
 629     """Starts the local master daemon.
 630
 631     @param no_vote: Should the masterd started without voting? default: False
 632     @raise errors.CommandError: If unable to start daemon.
 633
 634     """
 635     env = {}
 636     if no_vote:
 637       env["EXTRA_MASTERD_ARGS"] = "--no-voting --yes-do-it"
 638
 639     result = utils.RunCmd([pathutils.DAEMON_UTIL, "start-master"], env=env)
 640     if result.failed:
 641       raise errors.CommandError("Couldn't start ganeti master."
 642                                 " Fail reason: %s; output: %s" %
 643                                 (result.fail_reason, result.output))
 644
 645   def _ReaddMergedNodesAndRedist(self):
 646     """Readds all merging nodes and make sure their config is up-to-date.
 647
 648     @raise errors.CommandError: If anything fails.
 649
 650     """
 651     for data in self.merger_data:
 652       for node in data.nodes:
 653         logging.info("Readding node %s", node)
 654         result = utils.RunCmd(["gnt-node", "add", "--readd",
 655                                "--no-ssh-key-check", node])
 656         if result.failed:
 657           logging.error("%s failed to be readded. Reason: %s, output: %s",
 658                          node, result.fail_reason, result.output)
 659
 660     result = utils.RunCmd(["gnt-cluster", "redist-conf"])
 661     if result.failed:
 662       raise errors.CommandError("Redistribution failed. Fail reason: %s;"
 663                                 " output: %s" % (result.fail_reason,
 664                                                  result.output))
 665
 666   # R0201: Method could be a function
 667   def _StartupAllInstances(self): # pylint: disable=R0201
 668     """Starts up all instances (locally).
 669
 670     @raise errors.CommandError: If unable to start clusters
 671
 672     """
 673     result = utils.RunCmd(["gnt-instance", "startup", "--all",
 674                            "--force-multiple"])
 675     if result.failed:
 676       raise errors.CommandError("Unable to start all instances."
 677                                 " Fail reason: %s; output: %s" %
 678                                 (result.fail_reason, result.output))
 679
 680   # R0201: Method could be a function
 681   # TODO: make this overridable, for some verify errors
 682   def _VerifyCluster(self): # pylint: disable=R0201
 683     """Runs gnt-cluster verify to verify the health.
 684
 685     @raise errors.ProgrammError: If cluster fails on verification
 686
 687     """
 688     result = utils.RunCmd(["gnt-cluster", "verify"])
 689     if result.failed:
 690       raise errors.CommandError("Verification of cluster failed."
 691                                 " Fail reason: %s; output: %s" %
 692                                 (result.fail_reason, result.output))
 693
 694   def Merge(self):
 695     """Does the actual merge.
 696
 697     It runs all the steps in the right order and updates the user about steps
 698     taken. Also it keeps track of rollback_steps to undo everything.
 699
 700     """
 701     rbsteps = []
 702     try:
 703       logging.info("Pre cluster verification")
 704       self._VerifyCluster()
 705
 706       logging.info("Prepare authorized_keys")
 707       rbsteps.append("Remove our key from authorized_keys on nodes:"
 708                      " %(nodes)s")
 709       self._PrepareAuthorizedKeys()
 710
 711       rbsteps.append("Start all instances again on the merging"
 712                      " clusters: %(clusters)s")
 713       if self.stop_instances:
 714         logging.info("Stopping merging instances (takes a while)")
 715         self._StopMergingInstances()
 716       logging.info("Checking that no instances are running on the mergees")
 717       instances_running = self._CheckRunningInstances()
 718       if instances_running:
 719         raise errors.CommandError("Some instances are still running on the"
 720                                   " mergees")
 721       logging.info("Disable watcher")
 722       self._DisableWatcher()
 723       logging.info("Merging config")
 724       self._FetchRemoteConfig()
 725       logging.info("Removing master IPs on mergee master nodes")
 726       self._RemoveMasterIps()
 727       logging.info("Stop daemons on merging nodes")
 728       self._StopDaemons()
 729
 730       logging.info("Stopping master daemon")
 731       self._KillMasterDaemon()
 732
 733       rbsteps.append("Restore %s from another master candidate"
 734                      " and restart master daemon" %
 735                      pathutils.CLUSTER_CONF_FILE)
 736       self._MergeConfig()
 737       self._StartMasterDaemon(no_vote=True)
 738
 739       # Point of no return, delete rbsteps
 740       del rbsteps[:]
 741
 742       logging.warning("We are at the point of no return. Merge can not easily"
 743                       " be undone after this point.")
 744       logging.info("Readd nodes")
 745       self._ReaddMergedNodesAndRedist()
 746
 747       logging.info("Merge done, restart master daemon normally")
 748       self._KillMasterDaemon()
 749       self._StartMasterDaemon()
 750
 751       if self.restart == _RESTART_ALL:
 752         logging.info("Starting instances again")
 753         self._StartupAllInstances()
 754       else:
 755         logging.info("Not starting instances again")
 756       logging.info("Post cluster verification")
 757       self._VerifyCluster()
 758     except errors.GenericError, e:
 759       logging.exception(e)
 760
 761       if rbsteps:
 762         nodes = Flatten([data.nodes for data in self.merger_data])
 763         info = {
 764           "clusters": self.clusters,
 765           "nodes": nodes,
 766           }
 767         logging.critical("In order to rollback do the following:")
 768         for step in rbsteps:
 769           logging.critical("  * %s", step % info)
 770       else:
 771         logging.critical("Nothing to rollback.")
 772
 773       # TODO: Keep track of steps done for a flawless resume?
 774
 775   def Cleanup(self):
 776     """Clean up our environment.
 777
 778     This cleans up remote private keys and configs and after that
 779     deletes the temporary directory.
 780
 781     """
 782     shutil.rmtree(self.work_dir)
 783
 784
 785 def main():
 786   """Main routine.
 787
 788   """
 789   program = os.path.basename(sys.argv[0])
 790
 791   parser = optparse.OptionParser(usage="%%prog [options...] <cluster...>",
 792                                  prog=program)
 793   parser.add_option(cli.DEBUG_OPT)
 794   parser.add_option(cli.VERBOSE_OPT)
 795   parser.add_option(PAUSE_PERIOD_OPT)
 796   parser.add_option(GROUPS_OPT)
 797   parser.add_option(RESTART_OPT)
 798   parser.add_option(PARAMS_OPT)
 799   parser.add_option(SKIP_STOP_INSTANCES_OPT)
 800
 801   (options, args) = parser.parse_args()
 802
 803   utils.SetupToolLogging(options.debug, options.verbose)
 804
 805   if not args:
 806     parser.error("No clusters specified")
 807
 808   cluster_merger = Merger(utils.UniqueSequence(args), options.pause_period,
 809                           options.groups, options.restart, options.params,
 810                           options.stop_instances)
 811   try:
 812     try:
 813       cluster_merger.Setup()
 814       cluster_merger.Merge()
 815     except errors.GenericError, e:
 816       logging.exception(e)
 817       return constants.EXIT_FAILURE
 818   finally:
 819     cluster_merger.Cleanup()
 820
 821   return constants.EXIT_SUCCESS
 822
 823
 824 if __name__ == "__main__":
 825   sys.exit(main())