4 # Copyright (C) 2006, 2007 Google Inc.
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.
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.
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
22 """Module implementing the master-side code."""
24 # pylint: disable-msg=W0613,W0201
34 from ganeti import rpc
35 from ganeti import ssh
36 from ganeti import logger
37 from ganeti import utils
38 from ganeti import errors
39 from ganeti import hypervisor
40 from ganeti import config
41 from ganeti import constants
42 from ganeti import objects
43 from ganeti import opcodes
44 from ganeti import ssconf
46 class LogicalUnit(object):
47 """Logical Unit base class.
49 Subclasses must follow these rules:
50 - implement CheckPrereq which also fills in the opcode instance
51 with all the fields (even if as None)
53 - implement BuildHooksEnv
54 - redefine HPATH and HTYPE
55 - optionally redefine their run requirements (REQ_CLUSTER,
56 REQ_MASTER); note that all commands require root permissions
65 def __init__(self, processor, op, cfg, sstore):
66 """Constructor for LogicalUnit.
68 This needs to be overriden in derived classes in order to check op
76 for attr_name in self._OP_REQP:
77 attr_val = getattr(op, attr_name, None)
79 raise errors.OpPrereqError("Required parameter '%s' missing" %
82 if not cfg.IsCluster():
83 raise errors.OpPrereqError("Cluster not initialized yet,"
84 " use 'gnt-cluster init' first.")
86 master = sstore.GetMasterNode()
87 if master != utils.HostInfo().name:
88 raise errors.OpPrereqError("Commands must be run on the master"
91 def CheckPrereq(self):
92 """Check prerequisites for this LU.
94 This method should check that the prerequisites for the execution
95 of this LU are fulfilled. It can do internode communication, but
96 it should be idempotent - no cluster or system changes are
99 The method should raise errors.OpPrereqError in case something is
100 not fulfilled. Its return value is ignored.
102 This method should also update all the parameters of the opcode to
103 their canonical form; e.g. a short node name must be fully
104 expanded after this method has successfully completed (so that
105 hooks, logging, etc. work correctly).
108 raise NotImplementedError
110 def Exec(self, feedback_fn):
113 This method should implement the actual work. It should raise
114 errors.OpExecError for failures that are somewhat dealt with in
118 raise NotImplementedError
120 def BuildHooksEnv(self):
121 """Build hooks environment for this LU.
123 This method should return a three-node tuple consisting of: a dict
124 containing the environment that will be used for running the
125 specific hook for this LU, a list of node names on which the hook
126 should run before the execution, and a list of node names on which
127 the hook should run after the execution.
129 The keys of the dict must not have 'GANETI_' prefixed as this will
130 be handled in the hooks runner. Also note additional keys will be
131 added by the hooks runner. If the LU doesn't define any
132 environment, an empty dict (and not None) should be returned.
134 As for the node lists, the master should not be included in the
135 them, as it will be added by the hooks runner in case this LU
136 requires a cluster to run on (otherwise we don't have a node
137 list). No nodes should be returned as an empty list (and not
140 Note that if the HPATH for a LU class is None, this function will
144 raise NotImplementedError
147 class NoHooksLU(LogicalUnit):
148 """Simple LU which runs no hooks.
150 This LU is intended as a parent for other LogicalUnits which will
151 run no hooks, in order to reduce duplicate code.
157 def BuildHooksEnv(self):
160 This is a no-op, since we don't run hooks.
166 def _AddHostToEtcHosts(hostname):
167 """Wrapper around utils.SetEtcHostsEntry.
170 hi = utils.HostInfo(name=hostname)
171 utils.SetEtcHostsEntry(constants.ETC_HOSTS, hi.ip, hi.name, [hi.ShortName()])
174 def _RemoveHostFromEtcHosts(hostname):
175 """Wrapper around utils.RemoveEtcHostsEntry.
178 hi = utils.HostInfo(name=hostname)
179 utils.RemoveEtcHostsEntry(constants.ETC_HOSTS, hi.name)
180 utils.RemoveEtcHostsEntry(constants.ETC_HOSTS, hi.ShortName())
183 def _GetWantedNodes(lu, nodes):
184 """Returns list of checked and expanded node names.
187 nodes: List of nodes (strings) or None for all
190 if not isinstance(nodes, list):
191 raise errors.OpPrereqError("Invalid argument type 'nodes'")
197 node = lu.cfg.ExpandNodeName(name)
199 raise errors.OpPrereqError("No such node name '%s'" % name)
203 wanted = lu.cfg.GetNodeList()
204 return utils.NiceSort(wanted)
207 def _GetWantedInstances(lu, instances):
208 """Returns list of checked and expanded instance names.
211 instances: List of instances (strings) or None for all
214 if not isinstance(instances, list):
215 raise errors.OpPrereqError("Invalid argument type 'instances'")
220 for name in instances:
221 instance = lu.cfg.ExpandInstanceName(name)
223 raise errors.OpPrereqError("No such instance name '%s'" % name)
224 wanted.append(instance)
227 wanted = lu.cfg.GetInstanceList()
228 return utils.NiceSort(wanted)
231 def _CheckOutputFields(static, dynamic, selected):
232 """Checks whether all selected fields are valid.
235 static: Static fields
236 dynamic: Dynamic fields
239 static_fields = frozenset(static)
240 dynamic_fields = frozenset(dynamic)
242 all_fields = static_fields | dynamic_fields
244 if not all_fields.issuperset(selected):
245 raise errors.OpPrereqError("Unknown output fields selected: %s"
246 % ",".join(frozenset(selected).
247 difference(all_fields)))
250 def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
251 memory, vcpus, nics):
252 """Builds instance related env variables for hooks from single variables.
255 secondary_nodes: List of secondary nodes as strings
259 "INSTANCE_NAME": name,
260 "INSTANCE_PRIMARY": primary_node,
261 "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
262 "INSTANCE_OS_TYPE": os_type,
263 "INSTANCE_STATUS": status,
264 "INSTANCE_MEMORY": memory,
265 "INSTANCE_VCPUS": vcpus,
269 nic_count = len(nics)
270 for idx, (ip, bridge, mac) in enumerate(nics):
273 env["INSTANCE_NIC%d_IP" % idx] = ip
274 env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
275 env["INSTANCE_NIC%d_HWADDR" % idx] = mac
279 env["INSTANCE_NIC_COUNT"] = nic_count
284 def _BuildInstanceHookEnvByObject(instance, override=None):
285 """Builds instance related env variables for hooks from an object.
288 instance: objects.Instance object of instance
289 override: dict of values to override
292 'name': instance.name,
293 'primary_node': instance.primary_node,
294 'secondary_nodes': instance.secondary_nodes,
295 'os_type': instance.os,
296 'status': instance.os,
297 'memory': instance.memory,
298 'vcpus': instance.vcpus,
299 'nics': [(nic.ip, nic.bridge, nic.mac) for nic in instance.nics],
302 args.update(override)
303 return _BuildInstanceHookEnv(**args)
306 def _UpdateKnownHosts(fullnode, ip, pubkey):
307 """Ensure a node has a correct known_hosts entry.
310 fullnode - Fully qualified domain name of host. (str)
311 ip - IPv4 address of host (str)
312 pubkey - the public key of the cluster
315 if os.path.exists(constants.SSH_KNOWN_HOSTS_FILE):
316 f = open(constants.SSH_KNOWN_HOSTS_FILE, 'r+')
318 f = open(constants.SSH_KNOWN_HOSTS_FILE, 'w+')
327 logger.Debug('read %s' % (repr(rawline),))
329 parts = rawline.rstrip('\r\n').split()
331 # Ignore unwanted lines
332 if len(parts) >= 3 and not rawline.lstrip()[0] == '#':
333 fields = parts[0].split(',')
338 for spec in [ ip, fullnode ]:
339 if spec not in fields:
344 logger.Debug("key, pubkey = %s." % (repr((key, pubkey)),))
345 if haveall and key == pubkey:
347 save_lines.append(rawline)
348 logger.Debug("Keeping known_hosts '%s'." % (repr(rawline),))
351 if havesome and (not haveall or key != pubkey):
353 logger.Debug("Discarding known_hosts '%s'." % (repr(rawline),))
356 save_lines.append(rawline)
359 add_lines.append('%s,%s ssh-rsa %s\n' % (fullnode, ip, pubkey))
360 logger.Debug("Adding known_hosts '%s'." % (repr(add_lines[-1]),))
363 save_lines = save_lines + add_lines
365 # Write a new file and replace old.
366 fd, tmpname = tempfile.mkstemp('.tmp', 'known_hosts.',
368 newfile = os.fdopen(fd, 'w')
370 newfile.write(''.join(save_lines))
373 logger.Debug("Wrote new known_hosts.")
374 os.rename(tmpname, constants.SSH_KNOWN_HOSTS_FILE)
377 # Simply appending a new line will do the trick.
379 for add in add_lines:
385 def _HasValidVG(vglist, vgname):
386 """Checks if the volume group list is valid.
388 A non-None return value means there's an error, and the return value
389 is the error message.
392 vgsize = vglist.get(vgname, None)
394 return "volume group '%s' missing" % vgname
396 return ("volume group '%s' too small (20480MiB required, %dMib found)" %
401 def _InitSSHSetup(node):
402 """Setup the SSH configuration for the cluster.
405 This generates a dsa keypair for root, adds the pub key to the
406 permitted hosts and adds the hostkey to its own known hosts.
409 node: the name of this host as a fqdn
412 priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
414 for name in priv_key, pub_key:
415 if os.path.exists(name):
416 utils.CreateBackup(name)
417 utils.RemoveFile(name)
419 result = utils.RunCmd(["ssh-keygen", "-t", "dsa",
423 raise errors.OpExecError("Could not generate ssh keypair, error %s" %
426 f = open(pub_key, 'r')
428 utils.AddAuthorizedKey(auth_keys, f.read(8192))
433 def _InitGanetiServerSetup(ss):
434 """Setup the necessary configuration for the initial node daemon.
436 This creates the nodepass file containing the shared password for
437 the cluster and also generates the SSL certificate.
440 # Create pseudo random password
441 randpass = sha.new(os.urandom(64)).hexdigest()
442 # and write it into sstore
443 ss.SetKey(ss.SS_NODED_PASS, randpass)
445 result = utils.RunCmd(["openssl", "req", "-new", "-newkey", "rsa:1024",
446 "-days", str(365*5), "-nodes", "-x509",
447 "-keyout", constants.SSL_CERT_FILE,
448 "-out", constants.SSL_CERT_FILE, "-batch"])
450 raise errors.OpExecError("could not generate server ssl cert, command"
451 " %s had exitcode %s and error message %s" %
452 (result.cmd, result.exit_code, result.output))
454 os.chmod(constants.SSL_CERT_FILE, 0400)
456 result = utils.RunCmd([constants.NODE_INITD_SCRIPT, "restart"])
459 raise errors.OpExecError("Could not start the node daemon, command %s"
460 " had exitcode %s and error %s" %
461 (result.cmd, result.exit_code, result.output))
464 def _CheckInstanceBridgesExist(instance):
465 """Check that the brigdes needed by an instance exist.
468 # check bridges existance
469 brlist = [nic.bridge for nic in instance.nics]
470 if not rpc.call_bridges_exist(instance.primary_node, brlist):
471 raise errors.OpPrereqError("one or more target bridges %s does not"
472 " exist on destination node '%s'" %
473 (brlist, instance.primary_node))
476 class LUInitCluster(LogicalUnit):
477 """Initialise the cluster.
480 HPATH = "cluster-init"
481 HTYPE = constants.HTYPE_CLUSTER
482 _OP_REQP = ["cluster_name", "hypervisor_type", "vg_name", "mac_prefix",
483 "def_bridge", "master_netdev"]
486 def BuildHooksEnv(self):
489 Notes: Since we don't require a cluster, we must manually add
490 ourselves in the post-run node list.
493 env = {"OP_TARGET": self.op.cluster_name}
494 return env, [], [self.hostname.name]
496 def CheckPrereq(self):
497 """Verify that the passed name is a valid one.
500 if config.ConfigWriter.IsCluster():
501 raise errors.OpPrereqError("Cluster is already initialised")
503 if self.op.hypervisor_type == constants.HT_XEN_HVM31:
504 if not os.path.exists(constants.VNC_PASSWORD_FILE):
505 raise errors.OpPrereqError("Please prepare the cluster VNC"
507 constants.VNC_PASSWORD_FILE)
509 self.hostname = hostname = utils.HostInfo()
511 if hostname.ip.startswith("127."):
512 raise errors.OpPrereqError("This host's IP resolves to the private"
513 " range (%s). Please fix DNS or /etc/hosts." %
516 self.clustername = clustername = utils.HostInfo(self.op.cluster_name)
518 if not utils.TcpPing(constants.LOCALHOST_IP_ADDRESS, hostname.ip,
519 constants.DEFAULT_NODED_PORT):
520 raise errors.OpPrereqError("Inconsistency: this host's name resolves"
521 " to %s,\nbut this ip address does not"
522 " belong to this host."
523 " Aborting." % hostname.ip)
525 secondary_ip = getattr(self.op, "secondary_ip", None)
526 if secondary_ip and not utils.IsValidIP(secondary_ip):
527 raise errors.OpPrereqError("Invalid secondary ip given")
529 secondary_ip != hostname.ip and
530 (not utils.TcpPing(constants.LOCALHOST_IP_ADDRESS, secondary_ip,
531 constants.DEFAULT_NODED_PORT))):
532 raise errors.OpPrereqError("You gave %s as secondary IP,"
533 " but it does not belong to this host." %
535 self.secondary_ip = secondary_ip
537 # checks presence of the volume group given
538 vgstatus = _HasValidVG(utils.ListVolumeGroups(), self.op.vg_name)
541 raise errors.OpPrereqError("Error: %s" % vgstatus)
543 if not re.match("^[0-9a-z]{2}:[0-9a-z]{2}:[0-9a-z]{2}$",
545 raise errors.OpPrereqError("Invalid mac prefix given '%s'" %
548 if self.op.hypervisor_type not in constants.HYPER_TYPES:
549 raise errors.OpPrereqError("Invalid hypervisor type given '%s'" %
550 self.op.hypervisor_type)
552 result = utils.RunCmd(["ip", "link", "show", "dev", self.op.master_netdev])
554 raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
555 (self.op.master_netdev,
556 result.output.strip()))
558 if not (os.path.isfile(constants.NODE_INITD_SCRIPT) and
559 os.access(constants.NODE_INITD_SCRIPT, os.X_OK)):
560 raise errors.OpPrereqError("Init.d script '%s' missing or not"
561 " executable." % constants.NODE_INITD_SCRIPT)
563 def Exec(self, feedback_fn):
564 """Initialize the cluster.
567 clustername = self.clustername
568 hostname = self.hostname
570 # set up the simple store
571 self.sstore = ss = ssconf.SimpleStore()
572 ss.SetKey(ss.SS_HYPERVISOR, self.op.hypervisor_type)
573 ss.SetKey(ss.SS_MASTER_NODE, hostname.name)
574 ss.SetKey(ss.SS_MASTER_IP, clustername.ip)
575 ss.SetKey(ss.SS_MASTER_NETDEV, self.op.master_netdev)
576 ss.SetKey(ss.SS_CLUSTER_NAME, clustername.name)
578 # set up the inter-node password and certificate
579 _InitGanetiServerSetup(ss)
581 # start the master ip
582 rpc.call_node_start_master(hostname.name)
584 # set up ssh config and /etc/hosts
585 f = open(constants.SSH_HOST_RSA_PUB, 'r')
590 sshkey = sshline.split(" ")[1]
592 _AddHostToEtcHosts(hostname.name)
594 _UpdateKnownHosts(hostname.name, hostname.ip, sshkey)
596 _InitSSHSetup(hostname.name)
598 # init of cluster config file
599 self.cfg = cfgw = config.ConfigWriter()
600 cfgw.InitConfig(hostname.name, hostname.ip, self.secondary_ip,
601 sshkey, self.op.mac_prefix,
602 self.op.vg_name, self.op.def_bridge)
605 class LUDestroyCluster(NoHooksLU):
606 """Logical unit for destroying the cluster.
611 def CheckPrereq(self):
612 """Check prerequisites.
614 This checks whether the cluster is empty.
616 Any errors are signalled by raising errors.OpPrereqError.
619 master = self.sstore.GetMasterNode()
621 nodelist = self.cfg.GetNodeList()
622 if len(nodelist) != 1 or nodelist[0] != master:
623 raise errors.OpPrereqError("There are still %d node(s) in"
624 " this cluster." % (len(nodelist) - 1))
625 instancelist = self.cfg.GetInstanceList()
627 raise errors.OpPrereqError("There are still %d instance(s) in"
628 " this cluster." % len(instancelist))
630 def Exec(self, feedback_fn):
631 """Destroys the cluster.
634 master = self.sstore.GetMasterNode()
635 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
636 utils.CreateBackup(priv_key)
637 utils.CreateBackup(pub_key)
638 rpc.call_node_leave_cluster(master)
641 class LUVerifyCluster(NoHooksLU):
642 """Verifies the cluster status.
647 def _VerifyNode(self, node, file_list, local_cksum, vglist, node_result,
648 remote_version, feedback_fn):
649 """Run multiple tests against a node.
652 - compares ganeti version
653 - checks vg existance and size > 20G
654 - checks config file checksum
655 - checks ssh to other nodes
658 node: name of the node to check
659 file_list: required list of files
660 local_cksum: dictionary of local files and their checksums
663 # compares ganeti version
664 local_version = constants.PROTOCOL_VERSION
665 if not remote_version:
666 feedback_fn(" - ERROR: connection to %s failed" % (node))
669 if local_version != remote_version:
670 feedback_fn(" - ERROR: sw version mismatch: master %s, node(%s) %s" %
671 (local_version, node, remote_version))
674 # checks vg existance and size > 20G
678 feedback_fn(" - ERROR: unable to check volume groups on node %s." %
682 vgstatus = _HasValidVG(vglist, self.cfg.GetVGName())
684 feedback_fn(" - ERROR: %s on node %s" % (vgstatus, node))
687 # checks config file checksum
690 if 'filelist' not in node_result:
692 feedback_fn(" - ERROR: node hasn't returned file checksum data")
694 remote_cksum = node_result['filelist']
695 for file_name in file_list:
696 if file_name not in remote_cksum:
698 feedback_fn(" - ERROR: file '%s' missing" % file_name)
699 elif remote_cksum[file_name] != local_cksum[file_name]:
701 feedback_fn(" - ERROR: file '%s' has wrong checksum" % file_name)
703 if 'nodelist' not in node_result:
705 feedback_fn(" - ERROR: node hasn't returned node connectivity data")
707 if node_result['nodelist']:
709 for node in node_result['nodelist']:
710 feedback_fn(" - ERROR: communication with node '%s': %s" %
711 (node, node_result['nodelist'][node]))
712 hyp_result = node_result.get('hypervisor', None)
713 if hyp_result is not None:
714 feedback_fn(" - ERROR: hypervisor verify failure: '%s'" % hyp_result)
717 def _VerifyInstance(self, instance, node_vol_is, node_instance, feedback_fn):
718 """Verify an instance.
720 This function checks to see if the required block devices are
721 available on the instance's node.
726 instancelist = self.cfg.GetInstanceList()
727 if not instance in instancelist:
728 feedback_fn(" - ERROR: instance %s not in instance list %s" %
729 (instance, instancelist))
732 instanceconfig = self.cfg.GetInstanceInfo(instance)
733 node_current = instanceconfig.primary_node
736 instanceconfig.MapLVsByNode(node_vol_should)
738 for node in node_vol_should:
739 for volume in node_vol_should[node]:
740 if node not in node_vol_is or volume not in node_vol_is[node]:
741 feedback_fn(" - ERROR: volume %s missing on node %s" %
745 if not instanceconfig.status == 'down':
746 if not instance in node_instance[node_current]:
747 feedback_fn(" - ERROR: instance %s not running on node %s" %
748 (instance, node_current))
751 for node in node_instance:
752 if (not node == node_current):
753 if instance in node_instance[node]:
754 feedback_fn(" - ERROR: instance %s should not run on node %s" %
760 def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
761 """Verify if there are any unknown volumes in the cluster.
763 The .os, .swap and backup volumes are ignored. All other volumes are
769 for node in node_vol_is:
770 for volume in node_vol_is[node]:
771 if node not in node_vol_should or volume not in node_vol_should[node]:
772 feedback_fn(" - ERROR: volume %s on node %s should not exist" %
777 def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
778 """Verify the list of running instances.
780 This checks what instances are running but unknown to the cluster.
784 for node in node_instance:
785 for runninginstance in node_instance[node]:
786 if runninginstance not in instancelist:
787 feedback_fn(" - ERROR: instance %s on node %s should not exist" %
788 (runninginstance, node))
792 def CheckPrereq(self):
793 """Check prerequisites.
795 This has no prerequisites.
800 def Exec(self, feedback_fn):
801 """Verify integrity of cluster, performing various test on nodes.
805 feedback_fn("* Verifying global settings")
806 for msg in self.cfg.VerifyConfig():
807 feedback_fn(" - ERROR: %s" % msg)
809 vg_name = self.cfg.GetVGName()
810 nodelist = utils.NiceSort(self.cfg.GetNodeList())
811 instancelist = utils.NiceSort(self.cfg.GetInstanceList())
815 # FIXME: verify OS list
817 file_names = list(self.sstore.GetFileList())
818 file_names.append(constants.SSL_CERT_FILE)
819 file_names.append(constants.CLUSTER_CONF_FILE)
820 local_checksums = utils.FingerprintFiles(file_names)
822 feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
823 all_volumeinfo = rpc.call_volume_list(nodelist, vg_name)
824 all_instanceinfo = rpc.call_instance_list(nodelist)
825 all_vglist = rpc.call_vg_list(nodelist)
826 node_verify_param = {
827 'filelist': file_names,
828 'nodelist': nodelist,
831 all_nvinfo = rpc.call_node_verify(nodelist, node_verify_param)
832 all_rversion = rpc.call_version(nodelist)
834 for node in nodelist:
835 feedback_fn("* Verifying node %s" % node)
836 result = self._VerifyNode(node, file_names, local_checksums,
837 all_vglist[node], all_nvinfo[node],
838 all_rversion[node], feedback_fn)
842 volumeinfo = all_volumeinfo[node]
844 if isinstance(volumeinfo, basestring):
845 feedback_fn(" - ERROR: LVM problem on node %s: %s" %
846 (node, volumeinfo[-400:].encode('string_escape')))
848 node_volume[node] = {}
849 elif not isinstance(volumeinfo, dict):
850 feedback_fn(" - ERROR: connection to %s failed" % (node,))
854 node_volume[node] = volumeinfo
857 nodeinstance = all_instanceinfo[node]
858 if type(nodeinstance) != list:
859 feedback_fn(" - ERROR: connection to %s failed" % (node,))
863 node_instance[node] = nodeinstance
867 for instance in instancelist:
868 feedback_fn("* Verifying instance %s" % instance)
869 result = self._VerifyInstance(instance, node_volume, node_instance,
873 inst_config = self.cfg.GetInstanceInfo(instance)
875 inst_config.MapLVsByNode(node_vol_should)
877 feedback_fn("* Verifying orphan volumes")
878 result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
882 feedback_fn("* Verifying remaining instances")
883 result = self._VerifyOrphanInstances(instancelist, node_instance,
890 class LUVerifyDisks(NoHooksLU):
891 """Verifies the cluster disks status.
896 def CheckPrereq(self):
897 """Check prerequisites.
899 This has no prerequisites.
904 def Exec(self, feedback_fn):
905 """Verify integrity of cluster disks.
908 result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {}
910 vg_name = self.cfg.GetVGName()
911 nodes = utils.NiceSort(self.cfg.GetNodeList())
912 instances = [self.cfg.GetInstanceInfo(name)
913 for name in self.cfg.GetInstanceList()]
916 for inst in instances:
918 if (inst.status != "up" or
919 inst.disk_template not in constants.DTS_NET_MIRROR):
921 inst.MapLVsByNode(inst_lvs)
922 # transform { iname: {node: [vol,],},} to {(node, vol): iname}
923 for node, vol_list in inst_lvs.iteritems():
925 nv_dict[(node, vol)] = inst
930 node_lvs = rpc.call_volume_list(nodes, vg_name)
937 if isinstance(lvs, basestring):
938 logger.Info("error enumerating LVs on node %s: %s" % (node, lvs))
940 elif not isinstance(lvs, dict):
941 logger.Info("connection to node %s failed or invalid data returned" %
943 res_nodes.append(node)
946 for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems():
947 inst = nv_dict.pop((node, lv_name), None)
948 if (not lv_online and inst is not None
949 and inst.name not in res_instances):
950 res_instances.append(inst.name)
952 # any leftover items in nv_dict are missing LVs, let's arrange the
954 for key, inst in nv_dict.iteritems():
955 if inst.name not in res_missing:
956 res_missing[inst.name] = []
957 res_missing[inst.name].append(key)
962 class LURenameCluster(LogicalUnit):
963 """Rename the cluster.
966 HPATH = "cluster-rename"
967 HTYPE = constants.HTYPE_CLUSTER
970 def BuildHooksEnv(self):
975 "OP_TARGET": self.op.sstore.GetClusterName(),
976 "NEW_NAME": self.op.name,
978 mn = self.sstore.GetMasterNode()
979 return env, [mn], [mn]
981 def CheckPrereq(self):
982 """Verify that the passed name is a valid one.
985 hostname = utils.HostInfo(self.op.name)
987 new_name = hostname.name
988 self.ip = new_ip = hostname.ip
989 old_name = self.sstore.GetClusterName()
990 old_ip = self.sstore.GetMasterIP()
991 if new_name == old_name and new_ip == old_ip:
992 raise errors.OpPrereqError("Neither the name nor the IP address of the"
993 " cluster has changed")
995 result = utils.RunCmd(["fping", "-q", new_ip])
996 if not result.failed:
997 raise errors.OpPrereqError("The given cluster IP address (%s) is"
998 " reachable on the network. Aborting." %
1001 self.op.name = new_name
1003 def Exec(self, feedback_fn):
1004 """Rename the cluster.
1007 clustername = self.op.name
1011 # shutdown the master IP
1012 master = ss.GetMasterNode()
1013 if not rpc.call_node_stop_master(master):
1014 raise errors.OpExecError("Could not disable the master role")
1018 ss.SetKey(ss.SS_MASTER_IP, ip)
1019 ss.SetKey(ss.SS_CLUSTER_NAME, clustername)
1021 # Distribute updated ss config to all nodes
1022 myself = self.cfg.GetNodeInfo(master)
1023 dist_nodes = self.cfg.GetNodeList()
1024 if myself.name in dist_nodes:
1025 dist_nodes.remove(myself.name)
1027 logger.Debug("Copying updated ssconf data to all nodes")
1028 for keyname in [ss.SS_CLUSTER_NAME, ss.SS_MASTER_IP]:
1029 fname = ss.KeyToFilename(keyname)
1030 result = rpc.call_upload_file(dist_nodes, fname)
1031 for to_node in dist_nodes:
1032 if not result[to_node]:
1033 logger.Error("copy of file %s to node %s failed" %
1036 if not rpc.call_node_start_master(master):
1037 logger.Error("Could not re-enable the master role on the master,"
1038 " please restart manually.")
1041 def _WaitForSync(cfgw, instance, proc, oneshot=False, unlock=False):
1042 """Sleep and poll for an instance's disk to sync.
1045 if not instance.disks:
1049 proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1051 node = instance.primary_node
1053 for dev in instance.disks:
1054 cfgw.SetDiskID(dev, node)
1060 cumul_degraded = False
1061 rstats = rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1063 proc.LogWarning("Can't get any data from node %s" % node)
1066 raise errors.RemoteError("Can't contact node %s for mirror data,"
1067 " aborting." % node)
1071 for i in range(len(rstats)):
1074 proc.LogWarning("Can't compute data for node %s/%s" %
1075 (node, instance.disks[i].iv_name))
1077 # we ignore the ldisk parameter
1078 perc_done, est_time, is_degraded, _ = mstat
1079 cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1080 if perc_done is not None:
1082 if est_time is not None:
1083 rem_time = "%d estimated seconds remaining" % est_time
1086 rem_time = "no time estimate"
1087 proc.LogInfo("- device %s: %5.2f%% done, %s" %
1088 (instance.disks[i].iv_name, perc_done, rem_time))
1095 time.sleep(min(60, max_time))
1101 proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1102 return not cumul_degraded
1105 def _CheckDiskConsistency(cfgw, dev, node, on_primary, ldisk=False):
1106 """Check that mirrors are not degraded.
1108 The ldisk parameter, if True, will change the test from the
1109 is_degraded attribute (which represents overall non-ok status for
1110 the device(s)) to the ldisk (representing the local storage status).
1113 cfgw.SetDiskID(dev, node)
1120 if on_primary or dev.AssembleOnSecondary():
1121 rstats = rpc.call_blockdev_find(node, dev)
1123 logger.ToStderr("Can't get any data from node %s" % node)
1126 result = result and (not rstats[idx])
1128 for child in dev.children:
1129 result = result and _CheckDiskConsistency(cfgw, child, node, on_primary)
1134 class LUDiagnoseOS(NoHooksLU):
1135 """Logical unit for OS diagnose/query.
1140 def CheckPrereq(self):
1141 """Check prerequisites.
1143 This always succeeds, since this is a pure query LU.
1148 def Exec(self, feedback_fn):
1149 """Compute the list of OSes.
1152 node_list = self.cfg.GetNodeList()
1153 node_data = rpc.call_os_diagnose(node_list)
1154 if node_data == False:
1155 raise errors.OpExecError("Can't gather the list of OSes")
1159 class LURemoveNode(LogicalUnit):
1160 """Logical unit for removing a node.
1163 HPATH = "node-remove"
1164 HTYPE = constants.HTYPE_NODE
1165 _OP_REQP = ["node_name"]
1167 def BuildHooksEnv(self):
1170 This doesn't run on the target node in the pre phase as a failed
1171 node would not allows itself to run.
1175 "OP_TARGET": self.op.node_name,
1176 "NODE_NAME": self.op.node_name,
1178 all_nodes = self.cfg.GetNodeList()
1179 all_nodes.remove(self.op.node_name)
1180 return env, all_nodes, all_nodes
1182 def CheckPrereq(self):
1183 """Check prerequisites.
1186 - the node exists in the configuration
1187 - it does not have primary or secondary instances
1188 - it's not the master
1190 Any errors are signalled by raising errors.OpPrereqError.
1193 node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1195 raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1197 instance_list = self.cfg.GetInstanceList()
1199 masternode = self.sstore.GetMasterNode()
1200 if node.name == masternode:
1201 raise errors.OpPrereqError("Node is the master node,"
1202 " you need to failover first.")
1204 for instance_name in instance_list:
1205 instance = self.cfg.GetInstanceInfo(instance_name)
1206 if node.name == instance.primary_node:
1207 raise errors.OpPrereqError("Instance %s still running on the node,"
1208 " please remove first." % instance_name)
1209 if node.name in instance.secondary_nodes:
1210 raise errors.OpPrereqError("Instance %s has node as a secondary,"
1211 " please remove first." % instance_name)
1212 self.op.node_name = node.name
1215 def Exec(self, feedback_fn):
1216 """Removes the node from the cluster.
1220 logger.Info("stopping the node daemon and removing configs from node %s" %
1223 rpc.call_node_leave_cluster(node.name)
1225 ssh.SSHCall(node.name, 'root', "%s stop" % constants.NODE_INITD_SCRIPT)
1227 logger.Info("Removing node %s from config" % node.name)
1229 self.cfg.RemoveNode(node.name)
1231 _RemoveHostFromEtcHosts(node.name)
1234 class LUQueryNodes(NoHooksLU):
1235 """Logical unit for querying nodes.
1238 _OP_REQP = ["output_fields", "names"]
1240 def CheckPrereq(self):
1241 """Check prerequisites.
1243 This checks that the fields required are valid output fields.
1246 self.dynamic_fields = frozenset(["dtotal", "dfree",
1247 "mtotal", "mnode", "mfree",
1250 _CheckOutputFields(static=["name", "pinst_cnt", "sinst_cnt",
1251 "pinst_list", "sinst_list",
1253 dynamic=self.dynamic_fields,
1254 selected=self.op.output_fields)
1256 self.wanted = _GetWantedNodes(self, self.op.names)
1258 def Exec(self, feedback_fn):
1259 """Computes the list of nodes and their attributes.
1262 nodenames = self.wanted
1263 nodelist = [self.cfg.GetNodeInfo(name) for name in nodenames]
1265 # begin data gathering
1267 if self.dynamic_fields.intersection(self.op.output_fields):
1269 node_data = rpc.call_node_info(nodenames, self.cfg.GetVGName())
1270 for name in nodenames:
1271 nodeinfo = node_data.get(name, None)
1274 "mtotal": utils.TryConvert(int, nodeinfo['memory_total']),
1275 "mnode": utils.TryConvert(int, nodeinfo['memory_dom0']),
1276 "mfree": utils.TryConvert(int, nodeinfo['memory_free']),
1277 "dtotal": utils.TryConvert(int, nodeinfo['vg_size']),
1278 "dfree": utils.TryConvert(int, nodeinfo['vg_free']),
1279 "bootid": nodeinfo['bootid'],
1282 live_data[name] = {}
1284 live_data = dict.fromkeys(nodenames, {})
1286 node_to_primary = dict([(name, set()) for name in nodenames])
1287 node_to_secondary = dict([(name, set()) for name in nodenames])
1289 inst_fields = frozenset(("pinst_cnt", "pinst_list",
1290 "sinst_cnt", "sinst_list"))
1291 if inst_fields & frozenset(self.op.output_fields):
1292 instancelist = self.cfg.GetInstanceList()
1294 for instance_name in instancelist:
1295 inst = self.cfg.GetInstanceInfo(instance_name)
1296 if inst.primary_node in node_to_primary:
1297 node_to_primary[inst.primary_node].add(inst.name)
1298 for secnode in inst.secondary_nodes:
1299 if secnode in node_to_secondary:
1300 node_to_secondary[secnode].add(inst.name)
1302 # end data gathering
1305 for node in nodelist:
1307 for field in self.op.output_fields:
1310 elif field == "pinst_list":
1311 val = list(node_to_primary[node.name])
1312 elif field == "sinst_list":
1313 val = list(node_to_secondary[node.name])
1314 elif field == "pinst_cnt":
1315 val = len(node_to_primary[node.name])
1316 elif field == "sinst_cnt":
1317 val = len(node_to_secondary[node.name])
1318 elif field == "pip":
1319 val = node.primary_ip
1320 elif field == "sip":
1321 val = node.secondary_ip
1322 elif field in self.dynamic_fields:
1323 val = live_data[node.name].get(field, None)
1325 raise errors.ParameterError(field)
1326 node_output.append(val)
1327 output.append(node_output)
1332 class LUQueryNodeVolumes(NoHooksLU):
1333 """Logical unit for getting volumes on node(s).
1336 _OP_REQP = ["nodes", "output_fields"]
1338 def CheckPrereq(self):
1339 """Check prerequisites.
1341 This checks that the fields required are valid output fields.
1344 self.nodes = _GetWantedNodes(self, self.op.nodes)
1346 _CheckOutputFields(static=["node"],
1347 dynamic=["phys", "vg", "name", "size", "instance"],
1348 selected=self.op.output_fields)
1351 def Exec(self, feedback_fn):
1352 """Computes the list of nodes and their attributes.
1355 nodenames = self.nodes
1356 volumes = rpc.call_node_volumes(nodenames)
1358 ilist = [self.cfg.GetInstanceInfo(iname) for iname
1359 in self.cfg.GetInstanceList()]
1361 lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1364 for node in nodenames:
1365 if node not in volumes or not volumes[node]:
1368 node_vols = volumes[node][:]
1369 node_vols.sort(key=lambda vol: vol['dev'])
1371 for vol in node_vols:
1373 for field in self.op.output_fields:
1376 elif field == "phys":
1380 elif field == "name":
1382 elif field == "size":
1383 val = int(float(vol['size']))
1384 elif field == "instance":
1386 if node not in lv_by_node[inst]:
1388 if vol['name'] in lv_by_node[inst][node]:
1394 raise errors.ParameterError(field)
1395 node_output.append(str(val))
1397 output.append(node_output)
1402 class LUAddNode(LogicalUnit):
1403 """Logical unit for adding node to the cluster.
1407 HTYPE = constants.HTYPE_NODE
1408 _OP_REQP = ["node_name"]
1410 def BuildHooksEnv(self):
1413 This will run on all nodes before, and on all nodes + the new node after.
1417 "OP_TARGET": self.op.node_name,
1418 "NODE_NAME": self.op.node_name,
1419 "NODE_PIP": self.op.primary_ip,
1420 "NODE_SIP": self.op.secondary_ip,
1422 nodes_0 = self.cfg.GetNodeList()
1423 nodes_1 = nodes_0 + [self.op.node_name, ]
1424 return env, nodes_0, nodes_1
1426 def CheckPrereq(self):
1427 """Check prerequisites.
1430 - the new node is not already in the config
1432 - its parameters (single/dual homed) matches the cluster
1434 Any errors are signalled by raising errors.OpPrereqError.
1437 node_name = self.op.node_name
1440 dns_data = utils.HostInfo(node_name)
1442 node = dns_data.name
1443 primary_ip = self.op.primary_ip = dns_data.ip
1444 secondary_ip = getattr(self.op, "secondary_ip", None)
1445 if secondary_ip is None:
1446 secondary_ip = primary_ip
1447 if not utils.IsValidIP(secondary_ip):
1448 raise errors.OpPrereqError("Invalid secondary IP given")
1449 self.op.secondary_ip = secondary_ip
1450 node_list = cfg.GetNodeList()
1451 if node in node_list:
1452 raise errors.OpPrereqError("Node %s is already in the configuration"
1455 for existing_node_name in node_list:
1456 existing_node = cfg.GetNodeInfo(existing_node_name)
1457 if (existing_node.primary_ip == primary_ip or
1458 existing_node.secondary_ip == primary_ip or
1459 existing_node.primary_ip == secondary_ip or
1460 existing_node.secondary_ip == secondary_ip):
1461 raise errors.OpPrereqError("New node ip address(es) conflict with"
1462 " existing node %s" % existing_node.name)
1464 # check that the type of the node (single versus dual homed) is the
1465 # same as for the master
1466 myself = cfg.GetNodeInfo(self.sstore.GetMasterNode())
1467 master_singlehomed = myself.secondary_ip == myself.primary_ip
1468 newbie_singlehomed = secondary_ip == primary_ip
1469 if master_singlehomed != newbie_singlehomed:
1470 if master_singlehomed:
1471 raise errors.OpPrereqError("The master has no private ip but the"
1472 " new node has one")
1474 raise errors.OpPrereqError("The master has a private ip but the"
1475 " new node doesn't have one")
1477 # checks reachablity
1478 if not utils.TcpPing(utils.HostInfo().name,
1480 constants.DEFAULT_NODED_PORT):
1481 raise errors.OpPrereqError("Node not reachable by ping")
1483 if not newbie_singlehomed:
1484 # check reachability from my secondary ip to newbie's secondary ip
1485 if not utils.TcpPing(myself.secondary_ip,
1487 constants.DEFAULT_NODED_PORT):
1488 raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
1489 " based ping to noded port")
1491 self.new_node = objects.Node(name=node,
1492 primary_ip=primary_ip,
1493 secondary_ip=secondary_ip)
1495 if self.sstore.GetHypervisorType() == constants.HT_XEN_HVM31:
1496 if not os.path.exists(constants.VNC_PASSWORD_FILE):
1497 raise errors.OpPrereqError("Cluster VNC password file %s missing" %
1498 constants.VNC_PASSWORD_FILE)
1500 def Exec(self, feedback_fn):
1501 """Adds the new node to the cluster.
1504 new_node = self.new_node
1505 node = new_node.name
1507 # set up inter-node password and certificate and restarts the node daemon
1508 gntpass = self.sstore.GetNodeDaemonPassword()
1509 if not re.match('^[a-zA-Z0-9.]{1,64}$', gntpass):
1510 raise errors.OpExecError("ganeti password corruption detected")
1511 f = open(constants.SSL_CERT_FILE)
1513 gntpem = f.read(8192)
1516 # in the base64 pem encoding, neither '!' nor '.' are valid chars,
1517 # so we use this to detect an invalid certificate; as long as the
1518 # cert doesn't contain this, the here-document will be correctly
1519 # parsed by the shell sequence below
1520 if re.search('^!EOF\.', gntpem, re.MULTILINE):
1521 raise errors.OpExecError("invalid PEM encoding in the SSL certificate")
1522 if not gntpem.endswith("\n"):
1523 raise errors.OpExecError("PEM must end with newline")
1524 logger.Info("copy cluster pass to %s and starting the node daemon" % node)
1526 # and then connect with ssh to set password and start ganeti-noded
1527 # note that all the below variables are sanitized at this point,
1528 # either by being constants or by the checks above
1530 mycommand = ("umask 077 && "
1531 "echo '%s' > '%s' && "
1532 "cat > '%s' << '!EOF.' && \n"
1533 "%s!EOF.\n%s restart" %
1534 (gntpass, ss.KeyToFilename(ss.SS_NODED_PASS),
1535 constants.SSL_CERT_FILE, gntpem,
1536 constants.NODE_INITD_SCRIPT))
1538 result = ssh.SSHCall(node, 'root', mycommand, batch=False, ask_key=True)
1540 raise errors.OpExecError("Remote command on node %s, error: %s,"
1542 (node, result.fail_reason, result.output))
1544 # check connectivity
1547 result = rpc.call_version([node])[node]
1549 if constants.PROTOCOL_VERSION == result:
1550 logger.Info("communication to node %s fine, sw version %s match" %
1553 raise errors.OpExecError("Version mismatch master version %s,"
1554 " node version %s" %
1555 (constants.PROTOCOL_VERSION, result))
1557 raise errors.OpExecError("Cannot get version from the new node")
1560 logger.Info("copy ssh key to node %s" % node)
1561 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
1563 keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
1564 constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
1570 keyarray.append(f.read())
1574 result = rpc.call_node_add(node, keyarray[0], keyarray[1], keyarray[2],
1575 keyarray[3], keyarray[4], keyarray[5])
1578 raise errors.OpExecError("Cannot transfer ssh keys to the new node")
1580 # Add node to our /etc/hosts, and add key to known_hosts
1581 _AddHostToEtcHosts(new_node.name)
1583 _UpdateKnownHosts(new_node.name, new_node.primary_ip,
1584 self.cfg.GetHostKey())
1586 if new_node.secondary_ip != new_node.primary_ip:
1587 if not rpc.call_node_tcp_ping(new_node.name,
1588 constants.LOCALHOST_IP_ADDRESS,
1589 new_node.secondary_ip,
1590 constants.DEFAULT_NODED_PORT,
1592 raise errors.OpExecError("Node claims it doesn't have the secondary ip"
1593 " you gave (%s). Please fix and re-run this"
1594 " command." % new_node.secondary_ip)
1596 success, msg = ssh.VerifyNodeHostname(node)
1598 raise errors.OpExecError("Node '%s' claims it has a different hostname"
1599 " than the one the resolver gives: %s."
1600 " Please fix and re-run this command." %
1603 # Distribute updated /etc/hosts and known_hosts to all nodes,
1604 # including the node just added
1605 myself = self.cfg.GetNodeInfo(self.sstore.GetMasterNode())
1606 dist_nodes = self.cfg.GetNodeList() + [node]
1607 if myself.name in dist_nodes:
1608 dist_nodes.remove(myself.name)
1610 logger.Debug("Copying hosts and known_hosts to all nodes")
1611 for fname in ("/etc/hosts", constants.SSH_KNOWN_HOSTS_FILE):
1612 result = rpc.call_upload_file(dist_nodes, fname)
1613 for to_node in dist_nodes:
1614 if not result[to_node]:
1615 logger.Error("copy of file %s to node %s failed" %
1618 to_copy = ss.GetFileList()
1619 if self.sstore.GetHypervisorType() == constants.HT_XEN_HVM31:
1620 to_copy.append(constants.VNC_PASSWORD_FILE)
1621 for fname in to_copy:
1622 if not ssh.CopyFileToNode(node, fname):
1623 logger.Error("could not copy file %s to node %s" % (fname, node))
1625 logger.Info("adding node %s to cluster.conf" % node)
1626 self.cfg.AddNode(new_node)
1629 class LUMasterFailover(LogicalUnit):
1630 """Failover the master node to the current node.
1632 This is a special LU in that it must run on a non-master node.
1635 HPATH = "master-failover"
1636 HTYPE = constants.HTYPE_CLUSTER
1640 def BuildHooksEnv(self):
1643 This will run on the new master only in the pre phase, and on all
1644 the nodes in the post phase.
1648 "OP_TARGET": self.new_master,
1649 "NEW_MASTER": self.new_master,
1650 "OLD_MASTER": self.old_master,
1652 return env, [self.new_master], self.cfg.GetNodeList()
1654 def CheckPrereq(self):
1655 """Check prerequisites.
1657 This checks that we are not already the master.
1660 self.new_master = utils.HostInfo().name
1661 self.old_master = self.sstore.GetMasterNode()
1663 if self.old_master == self.new_master:
1664 raise errors.OpPrereqError("This commands must be run on the node"
1665 " where you want the new master to be."
1666 " %s is already the master" %
1669 def Exec(self, feedback_fn):
1670 """Failover the master node.
1672 This command, when run on a non-master node, will cause the current
1673 master to cease being master, and the non-master to become new
1677 #TODO: do not rely on gethostname returning the FQDN
1678 logger.Info("setting master to %s, old master: %s" %
1679 (self.new_master, self.old_master))
1681 if not rpc.call_node_stop_master(self.old_master):
1682 logger.Error("could disable the master role on the old master"
1683 " %s, please disable manually" % self.old_master)
1686 ss.SetKey(ss.SS_MASTER_NODE, self.new_master)
1687 if not rpc.call_upload_file(self.cfg.GetNodeList(),
1688 ss.KeyToFilename(ss.SS_MASTER_NODE)):
1689 logger.Error("could not distribute the new simple store master file"
1690 " to the other nodes, please check.")
1692 if not rpc.call_node_start_master(self.new_master):
1693 logger.Error("could not start the master role on the new master"
1694 " %s, please check" % self.new_master)
1695 feedback_fn("Error in activating the master IP on the new master,"
1696 " please fix manually.")
1700 class LUQueryClusterInfo(NoHooksLU):
1701 """Query cluster configuration.
1707 def CheckPrereq(self):
1708 """No prerequsites needed for this LU.
1713 def Exec(self, feedback_fn):
1714 """Return cluster config.
1718 "name": self.sstore.GetClusterName(),
1719 "software_version": constants.RELEASE_VERSION,
1720 "protocol_version": constants.PROTOCOL_VERSION,
1721 "config_version": constants.CONFIG_VERSION,
1722 "os_api_version": constants.OS_API_VERSION,
1723 "export_version": constants.EXPORT_VERSION,
1724 "master": self.sstore.GetMasterNode(),
1725 "architecture": (platform.architecture()[0], platform.machine()),
1731 class LUClusterCopyFile(NoHooksLU):
1732 """Copy file to cluster.
1735 _OP_REQP = ["nodes", "filename"]
1737 def CheckPrereq(self):
1738 """Check prerequisites.
1740 It should check that the named file exists and that the given list
1744 if not os.path.exists(self.op.filename):
1745 raise errors.OpPrereqError("No such filename '%s'" % self.op.filename)
1747 self.nodes = _GetWantedNodes(self, self.op.nodes)
1749 def Exec(self, feedback_fn):
1750 """Copy a file from master to some nodes.
1753 opts - class with options as members
1754 args - list containing a single element, the file name
1756 nodes - list containing the name of target nodes; if empty, all nodes
1759 filename = self.op.filename
1761 myname = utils.HostInfo().name
1763 for node in self.nodes:
1766 if not ssh.CopyFileToNode(node, filename):
1767 logger.Error("Copy of file %s to node %s failed" % (filename, node))
1770 class LUDumpClusterConfig(NoHooksLU):
1771 """Return a text-representation of the cluster-config.
1776 def CheckPrereq(self):
1777 """No prerequisites.
1782 def Exec(self, feedback_fn):
1783 """Dump a representation of the cluster config to the standard output.
1786 return self.cfg.DumpConfig()
1789 class LURunClusterCommand(NoHooksLU):
1790 """Run a command on some nodes.
1793 _OP_REQP = ["command", "nodes"]
1795 def CheckPrereq(self):
1796 """Check prerequisites.
1798 It checks that the given list of nodes is valid.
1801 self.nodes = _GetWantedNodes(self, self.op.nodes)
1803 def Exec(self, feedback_fn):
1804 """Run a command on some nodes.
1808 for node in self.nodes:
1809 result = ssh.SSHCall(node, "root", self.op.command)
1810 data.append((node, result.output, result.exit_code))
1815 class LUActivateInstanceDisks(NoHooksLU):
1816 """Bring up an instance's disks.
1819 _OP_REQP = ["instance_name"]
1821 def CheckPrereq(self):
1822 """Check prerequisites.
1824 This checks that the instance is in the cluster.
1827 instance = self.cfg.GetInstanceInfo(
1828 self.cfg.ExpandInstanceName(self.op.instance_name))
1829 if instance is None:
1830 raise errors.OpPrereqError("Instance '%s' not known" %
1831 self.op.instance_name)
1832 self.instance = instance
1835 def Exec(self, feedback_fn):
1836 """Activate the disks.
1839 disks_ok, disks_info = _AssembleInstanceDisks(self.instance, self.cfg)
1841 raise errors.OpExecError("Cannot activate block devices")
1846 def _AssembleInstanceDisks(instance, cfg, ignore_secondaries=False):
1847 """Prepare the block devices for an instance.
1849 This sets up the block devices on all nodes.
1852 instance: a ganeti.objects.Instance object
1853 ignore_secondaries: if true, errors on secondary nodes won't result
1854 in an error return from the function
1857 false if the operation failed
1858 list of (host, instance_visible_name, node_visible_name) if the operation
1859 suceeded with the mapping from node devices to instance devices
1863 iname = instance.name
1864 # With the two passes mechanism we try to reduce the window of
1865 # opportunity for the race condition of switching DRBD to primary
1866 # before handshaking occured, but we do not eliminate it
1868 # The proper fix would be to wait (with some limits) until the
1869 # connection has been made and drbd transitions from WFConnection
1870 # into any other network-connected state (Connected, SyncTarget,
1873 # 1st pass, assemble on all nodes in secondary mode
1874 for inst_disk in instance.disks:
1875 for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
1876 cfg.SetDiskID(node_disk, node)
1877 result = rpc.call_blockdev_assemble(node, node_disk, iname, False)
1879 logger.Error("could not prepare block device %s on node %s"
1880 " (is_primary=False, pass=1)" % (inst_disk.iv_name, node))
1881 if not ignore_secondaries:
1884 # FIXME: race condition on drbd migration to primary
1886 # 2nd pass, do only the primary node
1887 for inst_disk in instance.disks:
1888 for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
1889 if node != instance.primary_node:
1891 cfg.SetDiskID(node_disk, node)
1892 result = rpc.call_blockdev_assemble(node, node_disk, iname, True)
1894 logger.Error("could not prepare block device %s on node %s"
1895 " (is_primary=True, pass=2)" % (inst_disk.iv_name, node))
1897 device_info.append((instance.primary_node, inst_disk.iv_name, result))
1899 # leave the disks configured for the primary node
1900 # this is a workaround that would be fixed better by
1901 # improving the logical/physical id handling
1902 for disk in instance.disks:
1903 cfg.SetDiskID(disk, instance.primary_node)
1905 return disks_ok, device_info
1908 def _StartInstanceDisks(cfg, instance, force):
1909 """Start the disks of an instance.
1912 disks_ok, dummy = _AssembleInstanceDisks(instance, cfg,
1913 ignore_secondaries=force)
1915 _ShutdownInstanceDisks(instance, cfg)
1916 if force is not None and not force:
1917 logger.Error("If the message above refers to a secondary node,"
1918 " you can retry the operation using '--force'.")
1919 raise errors.OpExecError("Disk consistency error")
1922 class LUDeactivateInstanceDisks(NoHooksLU):
1923 """Shutdown an instance's disks.
1926 _OP_REQP = ["instance_name"]
1928 def CheckPrereq(self):
1929 """Check prerequisites.
1931 This checks that the instance is in the cluster.
1934 instance = self.cfg.GetInstanceInfo(
1935 self.cfg.ExpandInstanceName(self.op.instance_name))
1936 if instance is None:
1937 raise errors.OpPrereqError("Instance '%s' not known" %
1938 self.op.instance_name)
1939 self.instance = instance
1941 def Exec(self, feedback_fn):
1942 """Deactivate the disks
1945 instance = self.instance
1946 ins_l = rpc.call_instance_list([instance.primary_node])
1947 ins_l = ins_l[instance.primary_node]
1948 if not type(ins_l) is list:
1949 raise errors.OpExecError("Can't contact node '%s'" %
1950 instance.primary_node)
1952 if self.instance.name in ins_l:
1953 raise errors.OpExecError("Instance is running, can't shutdown"
1956 _ShutdownInstanceDisks(instance, self.cfg)
1959 def _ShutdownInstanceDisks(instance, cfg, ignore_primary=False):
1960 """Shutdown block devices of an instance.
1962 This does the shutdown on all nodes of the instance.
1964 If the ignore_primary is false, errors on the primary node are
1969 for disk in instance.disks:
1970 for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
1971 cfg.SetDiskID(top_disk, node)
1972 if not rpc.call_blockdev_shutdown(node, top_disk):
1973 logger.Error("could not shutdown block device %s on node %s" %
1974 (disk.iv_name, node))
1975 if not ignore_primary or node != instance.primary_node:
1980 def _CheckNodeFreeMemory(cfg, node, reason, requested):
1981 """Checks if a node has enough free memory.
1983 This function check if a given node has the needed amount of free
1984 memory. In case the node has less memory or we cannot get the
1985 information from the node, this function raise an OpPrereqError
1989 - cfg: a ConfigWriter instance
1990 - node: the node name
1991 - reason: string to use in the error message
1992 - requested: the amount of memory in MiB
1995 nodeinfo = rpc.call_node_info([node], cfg.GetVGName())
1996 if not nodeinfo or not isinstance(nodeinfo, dict):
1997 raise errors.OpPrereqError("Could not contact node %s for resource"
1998 " information" % (node,))
2000 free_mem = nodeinfo[node].get('memory_free')
2001 if not isinstance(free_mem, int):
2002 raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2003 " was '%s'" % (node, free_mem))
2004 if requested > free_mem:
2005 raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2006 " needed %s MiB, available %s MiB" %
2007 (node, reason, requested, free_mem))
2010 class LUStartupInstance(LogicalUnit):
2011 """Starts an instance.
2014 HPATH = "instance-start"
2015 HTYPE = constants.HTYPE_INSTANCE
2016 _OP_REQP = ["instance_name", "force"]
2018 def BuildHooksEnv(self):
2021 This runs on master, primary and secondary nodes of the instance.
2025 "FORCE": self.op.force,
2027 env.update(_BuildInstanceHookEnvByObject(self.instance))
2028 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2029 list(self.instance.secondary_nodes))
2032 def CheckPrereq(self):
2033 """Check prerequisites.
2035 This checks that the instance is in the cluster.
2038 instance = self.cfg.GetInstanceInfo(
2039 self.cfg.ExpandInstanceName(self.op.instance_name))
2040 if instance is None:
2041 raise errors.OpPrereqError("Instance '%s' not known" %
2042 self.op.instance_name)
2044 # check bridges existance
2045 _CheckInstanceBridgesExist(instance)
2047 _CheckNodeFreeMemory(self.cfg, instance.primary_node,
2048 "starting instance %s" % instance.name,
2051 self.instance = instance
2052 self.op.instance_name = instance.name
2054 def Exec(self, feedback_fn):
2055 """Start the instance.
2058 instance = self.instance
2059 force = self.op.force
2060 extra_args = getattr(self.op, "extra_args", "")
2062 node_current = instance.primary_node
2064 _StartInstanceDisks(self.cfg, instance, force)
2066 if not rpc.call_instance_start(node_current, instance, extra_args):
2067 _ShutdownInstanceDisks(instance, self.cfg)
2068 raise errors.OpExecError("Could not start instance")
2070 self.cfg.MarkInstanceUp(instance.name)
2073 class LURebootInstance(LogicalUnit):
2074 """Reboot an instance.
2077 HPATH = "instance-reboot"
2078 HTYPE = constants.HTYPE_INSTANCE
2079 _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2081 def BuildHooksEnv(self):
2084 This runs on master, primary and secondary nodes of the instance.
2088 "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2090 env.update(_BuildInstanceHookEnvByObject(self.instance))
2091 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2092 list(self.instance.secondary_nodes))
2095 def CheckPrereq(self):
2096 """Check prerequisites.
2098 This checks that the instance is in the cluster.
2101 instance = self.cfg.GetInstanceInfo(
2102 self.cfg.ExpandInstanceName(self.op.instance_name))
2103 if instance is None:
2104 raise errors.OpPrereqError("Instance '%s' not known" %
2105 self.op.instance_name)
2107 # check bridges existance
2108 _CheckInstanceBridgesExist(instance)
2110 self.instance = instance
2111 self.op.instance_name = instance.name
2113 def Exec(self, feedback_fn):
2114 """Reboot the instance.
2117 instance = self.instance
2118 ignore_secondaries = self.op.ignore_secondaries
2119 reboot_type = self.op.reboot_type
2120 extra_args = getattr(self.op, "extra_args", "")
2122 node_current = instance.primary_node
2124 if reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2125 constants.INSTANCE_REBOOT_HARD,
2126 constants.INSTANCE_REBOOT_FULL]:
2127 raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2128 (constants.INSTANCE_REBOOT_SOFT,
2129 constants.INSTANCE_REBOOT_HARD,
2130 constants.INSTANCE_REBOOT_FULL))
2132 if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2133 constants.INSTANCE_REBOOT_HARD]:
2134 if not rpc.call_instance_reboot(node_current, instance,
2135 reboot_type, extra_args):
2136 raise errors.OpExecError("Could not reboot instance")
2138 if not rpc.call_instance_shutdown(node_current, instance):
2139 raise errors.OpExecError("could not shutdown instance for full reboot")
2140 _ShutdownInstanceDisks(instance, self.cfg)
2141 _StartInstanceDisks(self.cfg, instance, ignore_secondaries)
2142 if not rpc.call_instance_start(node_current, instance, extra_args):
2143 _ShutdownInstanceDisks(instance, self.cfg)
2144 raise errors.OpExecError("Could not start instance for full reboot")
2146 self.cfg.MarkInstanceUp(instance.name)
2149 class LUShutdownInstance(LogicalUnit):
2150 """Shutdown an instance.
2153 HPATH = "instance-stop"
2154 HTYPE = constants.HTYPE_INSTANCE
2155 _OP_REQP = ["instance_name"]
2157 def BuildHooksEnv(self):
2160 This runs on master, primary and secondary nodes of the instance.
2163 env = _BuildInstanceHookEnvByObject(self.instance)
2164 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2165 list(self.instance.secondary_nodes))
2168 def CheckPrereq(self):
2169 """Check prerequisites.
2171 This checks that the instance is in the cluster.
2174 instance = self.cfg.GetInstanceInfo(
2175 self.cfg.ExpandInstanceName(self.op.instance_name))
2176 if instance is None:
2177 raise errors.OpPrereqError("Instance '%s' not known" %
2178 self.op.instance_name)
2179 self.instance = instance
2181 def Exec(self, feedback_fn):
2182 """Shutdown the instance.
2185 instance = self.instance
2186 node_current = instance.primary_node
2187 if not rpc.call_instance_shutdown(node_current, instance):
2188 logger.Error("could not shutdown instance")
2190 self.cfg.MarkInstanceDown(instance.name)
2191 _ShutdownInstanceDisks(instance, self.cfg)
2194 class LUReinstallInstance(LogicalUnit):
2195 """Reinstall an instance.
2198 HPATH = "instance-reinstall"
2199 HTYPE = constants.HTYPE_INSTANCE
2200 _OP_REQP = ["instance_name"]
2202 def BuildHooksEnv(self):
2205 This runs on master, primary and secondary nodes of the instance.
2208 env = _BuildInstanceHookEnvByObject(self.instance)
2209 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2210 list(self.instance.secondary_nodes))
2213 def CheckPrereq(self):
2214 """Check prerequisites.
2216 This checks that the instance is in the cluster and is not running.
2219 instance = self.cfg.GetInstanceInfo(
2220 self.cfg.ExpandInstanceName(self.op.instance_name))
2221 if instance is None:
2222 raise errors.OpPrereqError("Instance '%s' not known" %
2223 self.op.instance_name)
2224 if instance.disk_template == constants.DT_DISKLESS:
2225 raise errors.OpPrereqError("Instance '%s' has no disks" %
2226 self.op.instance_name)
2227 if instance.status != "down":
2228 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2229 self.op.instance_name)
2230 remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2232 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2233 (self.op.instance_name,
2234 instance.primary_node))
2236 self.op.os_type = getattr(self.op, "os_type", None)
2237 if self.op.os_type is not None:
2239 pnode = self.cfg.GetNodeInfo(
2240 self.cfg.ExpandNodeName(instance.primary_node))
2242 raise errors.OpPrereqError("Primary node '%s' is unknown" %
2244 os_obj = rpc.call_os_get(pnode.name, self.op.os_type)
2246 raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2247 " primary node" % self.op.os_type)
2249 self.instance = instance
2251 def Exec(self, feedback_fn):
2252 """Reinstall the instance.
2255 inst = self.instance
2257 if self.op.os_type is not None:
2258 feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2259 inst.os = self.op.os_type
2260 self.cfg.AddInstance(inst)
2262 _StartInstanceDisks(self.cfg, inst, None)
2264 feedback_fn("Running the instance OS create scripts...")
2265 if not rpc.call_instance_os_add(inst.primary_node, inst, "sda", "sdb"):
2266 raise errors.OpExecError("Could not install OS for instance %s"
2268 (inst.name, inst.primary_node))
2270 _ShutdownInstanceDisks(inst, self.cfg)
2273 class LURenameInstance(LogicalUnit):
2274 """Rename an instance.
2277 HPATH = "instance-rename"
2278 HTYPE = constants.HTYPE_INSTANCE
2279 _OP_REQP = ["instance_name", "new_name"]
2281 def BuildHooksEnv(self):
2284 This runs on master, primary and secondary nodes of the instance.
2287 env = _BuildInstanceHookEnvByObject(self.instance)
2288 env["INSTANCE_NEW_NAME"] = self.op.new_name
2289 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2290 list(self.instance.secondary_nodes))
2293 def CheckPrereq(self):
2294 """Check prerequisites.
2296 This checks that the instance is in the cluster and is not running.
2299 instance = self.cfg.GetInstanceInfo(
2300 self.cfg.ExpandInstanceName(self.op.instance_name))
2301 if instance is None:
2302 raise errors.OpPrereqError("Instance '%s' not known" %
2303 self.op.instance_name)
2304 if instance.status != "down":
2305 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2306 self.op.instance_name)
2307 remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2309 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2310 (self.op.instance_name,
2311 instance.primary_node))
2312 self.instance = instance
2314 # new name verification
2315 name_info = utils.HostInfo(self.op.new_name)
2317 self.op.new_name = new_name = name_info.name
2318 if not getattr(self.op, "ignore_ip", False):
2319 command = ["fping", "-q", name_info.ip]
2320 result = utils.RunCmd(command)
2321 if not result.failed:
2322 raise errors.OpPrereqError("IP %s of instance %s already in use" %
2323 (name_info.ip, new_name))
2326 def Exec(self, feedback_fn):
2327 """Reinstall the instance.
2330 inst = self.instance
2331 old_name = inst.name
2333 self.cfg.RenameInstance(inst.name, self.op.new_name)
2335 # re-read the instance from the configuration after rename
2336 inst = self.cfg.GetInstanceInfo(self.op.new_name)
2338 _StartInstanceDisks(self.cfg, inst, None)
2340 if not rpc.call_instance_run_rename(inst.primary_node, inst, old_name,
2342 msg = ("Could run OS rename script for instance %s on node %s (but the"
2343 " instance has been renamed in Ganeti)" %
2344 (inst.name, inst.primary_node))
2347 _ShutdownInstanceDisks(inst, self.cfg)
2350 class LURemoveInstance(LogicalUnit):
2351 """Remove an instance.
2354 HPATH = "instance-remove"
2355 HTYPE = constants.HTYPE_INSTANCE
2356 _OP_REQP = ["instance_name"]
2358 def BuildHooksEnv(self):
2361 This runs on master, primary and secondary nodes of the instance.
2364 env = _BuildInstanceHookEnvByObject(self.instance)
2365 nl = [self.sstore.GetMasterNode()]
2368 def CheckPrereq(self):
2369 """Check prerequisites.
2371 This checks that the instance is in the cluster.
2374 instance = self.cfg.GetInstanceInfo(
2375 self.cfg.ExpandInstanceName(self.op.instance_name))
2376 if instance is None:
2377 raise errors.OpPrereqError("Instance '%s' not known" %
2378 self.op.instance_name)
2379 self.instance = instance
2381 def Exec(self, feedback_fn):
2382 """Remove the instance.
2385 instance = self.instance
2386 logger.Info("shutting down instance %s on node %s" %
2387 (instance.name, instance.primary_node))
2389 if not rpc.call_instance_shutdown(instance.primary_node, instance):
2390 if self.op.ignore_failures:
2391 feedback_fn("Warning: can't shutdown instance")
2393 raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2394 (instance.name, instance.primary_node))
2396 logger.Info("removing block devices for instance %s" % instance.name)
2398 if not _RemoveDisks(instance, self.cfg):
2399 if self.op.ignore_failures:
2400 feedback_fn("Warning: can't remove instance's disks")
2402 raise errors.OpExecError("Can't remove instance's disks")
2404 logger.Info("removing instance %s out of cluster config" % instance.name)
2406 self.cfg.RemoveInstance(instance.name)
2409 class LUQueryInstances(NoHooksLU):
2410 """Logical unit for querying instances.
2413 _OP_REQP = ["output_fields", "names"]
2415 def CheckPrereq(self):
2416 """Check prerequisites.
2418 This checks that the fields required are valid output fields.
2421 self.dynamic_fields = frozenset(["oper_state", "oper_ram"])
2422 _CheckOutputFields(static=["name", "os", "pnode", "snodes",
2423 "admin_state", "admin_ram",
2424 "disk_template", "ip", "mac", "bridge",
2425 "sda_size", "sdb_size", "vcpus"],
2426 dynamic=self.dynamic_fields,
2427 selected=self.op.output_fields)
2429 self.wanted = _GetWantedInstances(self, self.op.names)
2431 def Exec(self, feedback_fn):
2432 """Computes the list of nodes and their attributes.
2435 instance_names = self.wanted
2436 instance_list = [self.cfg.GetInstanceInfo(iname) for iname
2439 # begin data gathering
2441 nodes = frozenset([inst.primary_node for inst in instance_list])
2444 if self.dynamic_fields.intersection(self.op.output_fields):
2446 node_data = rpc.call_all_instances_info(nodes)
2448 result = node_data[name]
2450 live_data.update(result)
2451 elif result == False:
2452 bad_nodes.append(name)
2453 # else no instance is alive
2455 live_data = dict([(name, {}) for name in instance_names])
2457 # end data gathering
2460 for instance in instance_list:
2462 for field in self.op.output_fields:
2467 elif field == "pnode":
2468 val = instance.primary_node
2469 elif field == "snodes":
2470 val = list(instance.secondary_nodes)
2471 elif field == "admin_state":
2472 val = (instance.status != "down")
2473 elif field == "oper_state":
2474 if instance.primary_node in bad_nodes:
2477 val = bool(live_data.get(instance.name))
2478 elif field == "admin_ram":
2479 val = instance.memory
2480 elif field == "oper_ram":
2481 if instance.primary_node in bad_nodes:
2483 elif instance.name in live_data:
2484 val = live_data[instance.name].get("memory", "?")
2487 elif field == "disk_template":
2488 val = instance.disk_template
2490 val = instance.nics[0].ip
2491 elif field == "bridge":
2492 val = instance.nics[0].bridge
2493 elif field == "mac":
2494 val = instance.nics[0].mac
2495 elif field == "sda_size" or field == "sdb_size":
2496 disk = instance.FindDisk(field[:3])
2501 elif field == "vcpus":
2502 val = instance.vcpus
2504 raise errors.ParameterError(field)
2511 class LUFailoverInstance(LogicalUnit):
2512 """Failover an instance.
2515 HPATH = "instance-failover"
2516 HTYPE = constants.HTYPE_INSTANCE
2517 _OP_REQP = ["instance_name", "ignore_consistency"]
2519 def BuildHooksEnv(self):
2522 This runs on master, primary and secondary nodes of the instance.
2526 "IGNORE_CONSISTENCY": self.op.ignore_consistency,
2528 env.update(_BuildInstanceHookEnvByObject(self.instance))
2529 nl = [self.sstore.GetMasterNode()] + list(self.instance.secondary_nodes)
2532 def CheckPrereq(self):
2533 """Check prerequisites.
2535 This checks that the instance is in the cluster.
2538 instance = self.cfg.GetInstanceInfo(
2539 self.cfg.ExpandInstanceName(self.op.instance_name))
2540 if instance is None:
2541 raise errors.OpPrereqError("Instance '%s' not known" %
2542 self.op.instance_name)
2544 if instance.disk_template not in constants.DTS_NET_MIRROR:
2545 raise errors.OpPrereqError("Instance's disk layout is not"
2546 " network mirrored, cannot failover.")
2548 secondary_nodes = instance.secondary_nodes
2549 if not secondary_nodes:
2550 raise errors.ProgrammerError("no secondary node but using "
2551 "DT_REMOTE_RAID1 template")
2553 target_node = secondary_nodes[0]
2554 # check memory requirements on the secondary node
2555 _CheckNodeFreeMemory(self.cfg, target_node, "failing over instance %s" %
2556 instance.name, instance.memory)
2558 # check bridge existance
2559 brlist = [nic.bridge for nic in instance.nics]
2560 if not rpc.call_bridges_exist(target_node, brlist):
2561 raise errors.OpPrereqError("One or more target bridges %s does not"
2562 " exist on destination node '%s'" %
2563 (brlist, target_node))
2565 self.instance = instance
2567 def Exec(self, feedback_fn):
2568 """Failover an instance.
2570 The failover is done by shutting it down on its present node and
2571 starting it on the secondary.
2574 instance = self.instance
2576 source_node = instance.primary_node
2577 target_node = instance.secondary_nodes[0]
2579 feedback_fn("* checking disk consistency between source and target")
2580 for dev in instance.disks:
2581 # for remote_raid1, these are md over drbd
2582 if not _CheckDiskConsistency(self.cfg, dev, target_node, False):
2583 if not self.op.ignore_consistency:
2584 raise errors.OpExecError("Disk %s is degraded on target node,"
2585 " aborting failover." % dev.iv_name)
2587 feedback_fn("* shutting down instance on source node")
2588 logger.Info("Shutting down instance %s on node %s" %
2589 (instance.name, source_node))
2591 if not rpc.call_instance_shutdown(source_node, instance):
2592 if self.op.ignore_consistency:
2593 logger.Error("Could not shutdown instance %s on node %s. Proceeding"
2594 " anyway. Please make sure node %s is down" %
2595 (instance.name, source_node, source_node))
2597 raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2598 (instance.name, source_node))
2600 feedback_fn("* deactivating the instance's disks on source node")
2601 if not _ShutdownInstanceDisks(instance, self.cfg, ignore_primary=True):
2602 raise errors.OpExecError("Can't shut down the instance's disks.")
2604 instance.primary_node = target_node
2605 # distribute new instance config to the other nodes
2606 self.cfg.AddInstance(instance)
2608 feedback_fn("* activating the instance's disks on target node")
2609 logger.Info("Starting instance %s on node %s" %
2610 (instance.name, target_node))
2612 disks_ok, dummy = _AssembleInstanceDisks(instance, self.cfg,
2613 ignore_secondaries=True)
2615 _ShutdownInstanceDisks(instance, self.cfg)
2616 raise errors.OpExecError("Can't activate the instance's disks")
2618 feedback_fn("* starting the instance on the target node")
2619 if not rpc.call_instance_start(target_node, instance, None):
2620 _ShutdownInstanceDisks(instance, self.cfg)
2621 raise errors.OpExecError("Could not start instance %s on node %s." %
2622 (instance.name, target_node))
2625 def _CreateBlockDevOnPrimary(cfg, node, instance, device, info):
2626 """Create a tree of block devices on the primary node.
2628 This always creates all devices.
2632 for child in device.children:
2633 if not _CreateBlockDevOnPrimary(cfg, node, instance, child, info):
2636 cfg.SetDiskID(device, node)
2637 new_id = rpc.call_blockdev_create(node, device, device.size,
2638 instance.name, True, info)
2641 if device.physical_id is None:
2642 device.physical_id = new_id
2646 def _CreateBlockDevOnSecondary(cfg, node, instance, device, force, info):
2647 """Create a tree of block devices on a secondary node.
2649 If this device type has to be created on secondaries, create it and
2652 If not, just recurse to children keeping the same 'force' value.
2655 if device.CreateOnSecondary():
2658 for child in device.children:
2659 if not _CreateBlockDevOnSecondary(cfg, node, instance,
2660 child, force, info):
2665 cfg.SetDiskID(device, node)
2666 new_id = rpc.call_blockdev_create(node, device, device.size,
2667 instance.name, False, info)
2670 if device.physical_id is None:
2671 device.physical_id = new_id
2675 def _GenerateUniqueNames(cfg, exts):
2676 """Generate a suitable LV name.
2678 This will generate a logical volume name for the given instance.
2683 new_id = cfg.GenerateUniqueID()
2684 results.append("%s%s" % (new_id, val))
2688 def _GenerateMDDRBDBranch(cfg, primary, secondary, size, names):
2689 """Generate a drbd device complete with its children.
2692 port = cfg.AllocatePort()
2693 vgname = cfg.GetVGName()
2694 dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
2695 logical_id=(vgname, names[0]))
2696 dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
2697 logical_id=(vgname, names[1]))
2698 drbd_dev = objects.Disk(dev_type=constants.LD_DRBD7, size=size,
2699 logical_id = (primary, secondary, port),
2700 children = [dev_data, dev_meta])
2704 def _GenerateDRBD8Branch(cfg, primary, secondary, size, names, iv_name):
2705 """Generate a drbd8 device complete with its children.
2708 port = cfg.AllocatePort()
2709 vgname = cfg.GetVGName()
2710 dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
2711 logical_id=(vgname, names[0]))
2712 dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
2713 logical_id=(vgname, names[1]))
2714 drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
2715 logical_id = (primary, secondary, port),
2716 children = [dev_data, dev_meta],
2720 def _GenerateDiskTemplate(cfg, template_name,
2721 instance_name, primary_node,
2722 secondary_nodes, disk_sz, swap_sz):
2723 """Generate the entire disk layout for a given template type.
2726 #TODO: compute space requirements
2728 vgname = cfg.GetVGName()
2729 if template_name == "diskless":
2731 elif template_name == "plain":
2732 if len(secondary_nodes) != 0:
2733 raise errors.ProgrammerError("Wrong template configuration")
2735 names = _GenerateUniqueNames(cfg, [".sda", ".sdb"])
2736 sda_dev = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2737 logical_id=(vgname, names[0]),
2739 sdb_dev = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2740 logical_id=(vgname, names[1]),
2742 disks = [sda_dev, sdb_dev]
2743 elif template_name == "local_raid1":
2744 if len(secondary_nodes) != 0:
2745 raise errors.ProgrammerError("Wrong template configuration")
2748 names = _GenerateUniqueNames(cfg, [".sda_m1", ".sda_m2",
2749 ".sdb_m1", ".sdb_m2"])
2750 sda_dev_m1 = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2751 logical_id=(vgname, names[0]))
2752 sda_dev_m2 = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2753 logical_id=(vgname, names[1]))
2754 md_sda_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name = "sda",
2756 children = [sda_dev_m1, sda_dev_m2])
2757 sdb_dev_m1 = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2758 logical_id=(vgname, names[2]))
2759 sdb_dev_m2 = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2760 logical_id=(vgname, names[3]))
2761 md_sdb_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name = "sdb",
2763 children = [sdb_dev_m1, sdb_dev_m2])
2764 disks = [md_sda_dev, md_sdb_dev]
2765 elif template_name == constants.DT_REMOTE_RAID1:
2766 if len(secondary_nodes) != 1:
2767 raise errors.ProgrammerError("Wrong template configuration")
2768 remote_node = secondary_nodes[0]
2769 names = _GenerateUniqueNames(cfg, [".sda_data", ".sda_meta",
2770 ".sdb_data", ".sdb_meta"])
2771 drbd_sda_dev = _GenerateMDDRBDBranch(cfg, primary_node, remote_node,
2772 disk_sz, names[0:2])
2773 md_sda_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name="sda",
2774 children = [drbd_sda_dev], size=disk_sz)
2775 drbd_sdb_dev = _GenerateMDDRBDBranch(cfg, primary_node, remote_node,
2776 swap_sz, names[2:4])
2777 md_sdb_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name="sdb",
2778 children = [drbd_sdb_dev], size=swap_sz)
2779 disks = [md_sda_dev, md_sdb_dev]
2780 elif template_name == constants.DT_DRBD8:
2781 if len(secondary_nodes) != 1:
2782 raise errors.ProgrammerError("Wrong template configuration")
2783 remote_node = secondary_nodes[0]
2784 names = _GenerateUniqueNames(cfg, [".sda_data", ".sda_meta",
2785 ".sdb_data", ".sdb_meta"])
2786 drbd_sda_dev = _GenerateDRBD8Branch(cfg, primary_node, remote_node,
2787 disk_sz, names[0:2], "sda")
2788 drbd_sdb_dev = _GenerateDRBD8Branch(cfg, primary_node, remote_node,
2789 swap_sz, names[2:4], "sdb")
2790 disks = [drbd_sda_dev, drbd_sdb_dev]
2792 raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
2796 def _GetInstanceInfoText(instance):
2797 """Compute that text that should be added to the disk's metadata.
2800 return "originstname+%s" % instance.name
2803 def _CreateDisks(cfg, instance):
2804 """Create all disks for an instance.
2806 This abstracts away some work from AddInstance.
2809 instance: the instance object
2812 True or False showing the success of the creation process
2815 info = _GetInstanceInfoText(instance)
2817 for device in instance.disks:
2818 logger.Info("creating volume %s for instance %s" %
2819 (device.iv_name, instance.name))
2821 for secondary_node in instance.secondary_nodes:
2822 if not _CreateBlockDevOnSecondary(cfg, secondary_node, instance,
2823 device, False, info):
2824 logger.Error("failed to create volume %s (%s) on secondary node %s!" %
2825 (device.iv_name, device, secondary_node))
2828 if not _CreateBlockDevOnPrimary(cfg, instance.primary_node,
2829 instance, device, info):
2830 logger.Error("failed to create volume %s on primary!" %
2836 def _RemoveDisks(instance, cfg):
2837 """Remove all disks for an instance.
2839 This abstracts away some work from `AddInstance()` and
2840 `RemoveInstance()`. Note that in case some of the devices couldn't
2841 be removed, the removal will continue with the other ones (compare
2842 with `_CreateDisks()`).
2845 instance: the instance object
2848 True or False showing the success of the removal proces
2851 logger.Info("removing block devices for instance %s" % instance.name)
2854 for device in instance.disks:
2855 for node, disk in device.ComputeNodeTree(instance.primary_node):
2856 cfg.SetDiskID(disk, node)
2857 if not rpc.call_blockdev_remove(node, disk):
2858 logger.Error("could not remove block device %s on node %s,"
2859 " continuing anyway" %
2860 (device.iv_name, node))
2865 class LUCreateInstance(LogicalUnit):
2866 """Create an instance.
2869 HPATH = "instance-add"
2870 HTYPE = constants.HTYPE_INSTANCE
2871 _OP_REQP = ["instance_name", "mem_size", "disk_size", "pnode",
2872 "disk_template", "swap_size", "mode", "start", "vcpus",
2873 "wait_for_sync", "ip_check", "mac"]
2875 def BuildHooksEnv(self):
2878 This runs on master, primary and secondary nodes of the instance.
2882 "INSTANCE_DISK_TEMPLATE": self.op.disk_template,
2883 "INSTANCE_DISK_SIZE": self.op.disk_size,
2884 "INSTANCE_SWAP_SIZE": self.op.swap_size,
2885 "INSTANCE_ADD_MODE": self.op.mode,
2887 if self.op.mode == constants.INSTANCE_IMPORT:
2888 env["INSTANCE_SRC_NODE"] = self.op.src_node
2889 env["INSTANCE_SRC_PATH"] = self.op.src_path
2890 env["INSTANCE_SRC_IMAGE"] = self.src_image
2892 env.update(_BuildInstanceHookEnv(name=self.op.instance_name,
2893 primary_node=self.op.pnode,
2894 secondary_nodes=self.secondaries,
2895 status=self.instance_status,
2896 os_type=self.op.os_type,
2897 memory=self.op.mem_size,
2898 vcpus=self.op.vcpus,
2899 nics=[(self.inst_ip, self.op.bridge, self.op.mac)],
2902 nl = ([self.sstore.GetMasterNode(), self.op.pnode] +
2907 def CheckPrereq(self):
2908 """Check prerequisites.
2911 for attr in ["kernel_path", "initrd_path", "hvm_boot_order"]:
2912 if not hasattr(self.op, attr):
2913 setattr(self.op, attr, None)
2915 if self.op.mode not in (constants.INSTANCE_CREATE,
2916 constants.INSTANCE_IMPORT):
2917 raise errors.OpPrereqError("Invalid instance creation mode '%s'" %
2920 if self.op.mode == constants.INSTANCE_IMPORT:
2921 src_node = getattr(self.op, "src_node", None)
2922 src_path = getattr(self.op, "src_path", None)
2923 if src_node is None or src_path is None:
2924 raise errors.OpPrereqError("Importing an instance requires source"
2925 " node and path options")
2926 src_node_full = self.cfg.ExpandNodeName(src_node)
2927 if src_node_full is None:
2928 raise errors.OpPrereqError("Unknown source node '%s'" % src_node)
2929 self.op.src_node = src_node = src_node_full
2931 if not os.path.isabs(src_path):
2932 raise errors.OpPrereqError("The source path must be absolute")
2934 export_info = rpc.call_export_info(src_node, src_path)
2937 raise errors.OpPrereqError("No export found in dir %s" % src_path)
2939 if not export_info.has_section(constants.INISECT_EXP):
2940 raise errors.ProgrammerError("Corrupted export config")
2942 ei_version = export_info.get(constants.INISECT_EXP, 'version')
2943 if (int(ei_version) != constants.EXPORT_VERSION):
2944 raise errors.OpPrereqError("Wrong export version %s (wanted %d)" %
2945 (ei_version, constants.EXPORT_VERSION))
2947 if int(export_info.get(constants.INISECT_INS, 'disk_count')) > 1:
2948 raise errors.OpPrereqError("Can't import instance with more than"
2951 # FIXME: are the old os-es, disk sizes, etc. useful?
2952 self.op.os_type = export_info.get(constants.INISECT_EXP, 'os')
2953 diskimage = os.path.join(src_path, export_info.get(constants.INISECT_INS,
2955 self.src_image = diskimage
2956 else: # INSTANCE_CREATE
2957 if getattr(self.op, "os_type", None) is None:
2958 raise errors.OpPrereqError("No guest OS specified")
2960 # check primary node
2961 pnode = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.pnode))
2963 raise errors.OpPrereqError("Primary node '%s' is unknown" %
2965 self.op.pnode = pnode.name
2967 self.secondaries = []
2968 # disk template and mirror node verification
2969 if self.op.disk_template not in constants.DISK_TEMPLATES:
2970 raise errors.OpPrereqError("Invalid disk template name")
2972 if self.op.disk_template in constants.DTS_NET_MIRROR:
2973 if getattr(self.op, "snode", None) is None:
2974 raise errors.OpPrereqError("The networked disk templates need"
2977 snode_name = self.cfg.ExpandNodeName(self.op.snode)
2978 if snode_name is None:
2979 raise errors.OpPrereqError("Unknown secondary node '%s'" %
2981 elif snode_name == pnode.name:
2982 raise errors.OpPrereqError("The secondary node cannot be"
2983 " the primary node.")
2984 self.secondaries.append(snode_name)
2986 # Required free disk space as a function of disk and swap space
2988 constants.DT_DISKLESS: None,
2989 constants.DT_PLAIN: self.op.disk_size + self.op.swap_size,
2990 constants.DT_LOCAL_RAID1: (self.op.disk_size + self.op.swap_size) * 2,
2991 # 256 MB are added for drbd metadata, 128MB for each drbd device
2992 constants.DT_REMOTE_RAID1: self.op.disk_size + self.op.swap_size + 256,
2993 constants.DT_DRBD8: self.op.disk_size + self.op.swap_size + 256,
2996 if self.op.disk_template not in req_size_dict:
2997 raise errors.ProgrammerError("Disk template '%s' size requirement"
2998 " is unknown" % self.op.disk_template)
3000 req_size = req_size_dict[self.op.disk_template]
3002 # Check lv size requirements
3003 if req_size is not None:
3004 nodenames = [pnode.name] + self.secondaries
3005 nodeinfo = rpc.call_node_info(nodenames, self.cfg.GetVGName())
3006 for node in nodenames:
3007 info = nodeinfo.get(node, None)
3009 raise errors.OpPrereqError("Cannot get current information"
3010 " from node '%s'" % nodeinfo)
3011 vg_free = info.get('vg_free', None)
3012 if not isinstance(vg_free, int):
3013 raise errors.OpPrereqError("Can't compute free disk space on"
3015 if req_size > info['vg_free']:
3016 raise errors.OpPrereqError("Not enough disk space on target node %s."
3017 " %d MB available, %d MB required" %
3018 (node, info['vg_free'], req_size))
3021 os_obj = rpc.call_os_get(pnode.name, self.op.os_type)
3023 raise errors.OpPrereqError("OS '%s' not in supported os list for"
3024 " primary node" % self.op.os_type)
3026 if self.op.kernel_path == constants.VALUE_NONE:
3027 raise errors.OpPrereqError("Can't set instance kernel to none")
3029 # instance verification
3030 hostname1 = utils.HostInfo(self.op.instance_name)
3032 self.op.instance_name = instance_name = hostname1.name
3033 instance_list = self.cfg.GetInstanceList()
3034 if instance_name in instance_list:
3035 raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3038 ip = getattr(self.op, "ip", None)
3039 if ip is None or ip.lower() == "none":
3041 elif ip.lower() == "auto":
3042 inst_ip = hostname1.ip
3044 if not utils.IsValidIP(ip):
3045 raise errors.OpPrereqError("given IP address '%s' doesn't look"
3046 " like a valid IP" % ip)
3048 self.inst_ip = inst_ip
3050 if self.op.start and not self.op.ip_check:
3051 raise errors.OpPrereqError("Cannot ignore IP address conflicts when"
3052 " adding an instance in start mode")
3054 if self.op.ip_check:
3055 if utils.TcpPing(utils.HostInfo().name, hostname1.ip,
3056 constants.DEFAULT_NODED_PORT):
3057 raise errors.OpPrereqError("IP %s of instance %s already in use" %
3058 (hostname1.ip, instance_name))
3060 # MAC address verification
3061 if self.op.mac != "auto":
3062 if not utils.IsValidMac(self.op.mac.lower()):
3063 raise errors.OpPrereqError("invalid MAC address specified: %s" %
3066 # bridge verification
3067 bridge = getattr(self.op, "bridge", None)
3069 self.op.bridge = self.cfg.GetDefBridge()
3071 self.op.bridge = bridge
3073 if not rpc.call_bridges_exist(self.pnode.name, [self.op.bridge]):
3074 raise errors.OpPrereqError("target bridge '%s' does not exist on"
3075 " destination node '%s'" %
3076 (self.op.bridge, pnode.name))
3078 # boot order verification
3079 if self.op.hvm_boot_order is not None:
3080 if len(self.op.hvm_boot_order.strip("acdn")) != 0:
3081 raise errors.OpPrereqError("invalid boot order specified,"
3082 " must be one or more of [acdn]")
3085 self.instance_status = 'up'
3087 self.instance_status = 'down'
3089 def Exec(self, feedback_fn):
3090 """Create and add the instance to the cluster.
3093 instance = self.op.instance_name
3094 pnode_name = self.pnode.name
3096 if self.op.mac == "auto":
3097 mac_address = self.cfg.GenerateMAC()
3099 mac_address = self.op.mac
3101 nic = objects.NIC(bridge=self.op.bridge, mac=mac_address)
3102 if self.inst_ip is not None:
3103 nic.ip = self.inst_ip
3105 ht_kind = self.sstore.GetHypervisorType()
3106 if ht_kind in constants.HTS_REQ_PORT:
3107 network_port = self.cfg.AllocatePort()
3111 disks = _GenerateDiskTemplate(self.cfg,
3112 self.op.disk_template,
3113 instance, pnode_name,
3114 self.secondaries, self.op.disk_size,
3117 iobj = objects.Instance(name=instance, os=self.op.os_type,
3118 primary_node=pnode_name,
3119 memory=self.op.mem_size,
3120 vcpus=self.op.vcpus,
3121 nics=[nic], disks=disks,
3122 disk_template=self.op.disk_template,
3123 status=self.instance_status,
3124 network_port=network_port,
3125 kernel_path=self.op.kernel_path,
3126 initrd_path=self.op.initrd_path,
3127 hvm_boot_order=self.op.hvm_boot_order,
3130 feedback_fn("* creating instance disks...")
3131 if not _CreateDisks(self.cfg, iobj):
3132 _RemoveDisks(iobj, self.cfg)
3133 raise errors.OpExecError("Device creation failed, reverting...")
3135 feedback_fn("adding instance %s to cluster config" % instance)
3137 self.cfg.AddInstance(iobj)
3139 if self.op.wait_for_sync:
3140 disk_abort = not _WaitForSync(self.cfg, iobj, self.proc)
3141 elif iobj.disk_template in constants.DTS_NET_MIRROR:
3142 # make sure the disks are not degraded (still sync-ing is ok)
3144 feedback_fn("* checking mirrors status")
3145 disk_abort = not _WaitForSync(self.cfg, iobj, self.proc, oneshot=True)
3150 _RemoveDisks(iobj, self.cfg)
3151 self.cfg.RemoveInstance(iobj.name)
3152 raise errors.OpExecError("There are some degraded disks for"
3155 feedback_fn("creating os for instance %s on node %s" %
3156 (instance, pnode_name))
3158 if iobj.disk_template != constants.DT_DISKLESS:
3159 if self.op.mode == constants.INSTANCE_CREATE:
3160 feedback_fn("* running the instance OS create scripts...")
3161 if not rpc.call_instance_os_add(pnode_name, iobj, "sda", "sdb"):
3162 raise errors.OpExecError("could not add os for instance %s"
3164 (instance, pnode_name))
3166 elif self.op.mode == constants.INSTANCE_IMPORT:
3167 feedback_fn("* running the instance OS import scripts...")
3168 src_node = self.op.src_node
3169 src_image = self.src_image
3170 if not rpc.call_instance_os_import(pnode_name, iobj, "sda", "sdb",
3171 src_node, src_image):
3172 raise errors.OpExecError("Could not import os for instance"
3174 (instance, pnode_name))
3176 # also checked in the prereq part
3177 raise errors.ProgrammerError("Unknown OS initialization mode '%s'"
3181 logger.Info("starting instance %s on node %s" % (instance, pnode_name))
3182 feedback_fn("* starting instance...")
3183 if not rpc.call_instance_start(pnode_name, iobj, None):
3184 raise errors.OpExecError("Could not start instance")
3187 class LUConnectConsole(NoHooksLU):
3188 """Connect to an instance's console.
3190 This is somewhat special in that it returns the command line that
3191 you need to run on the master node in order to connect to the
3195 _OP_REQP = ["instance_name"]
3197 def CheckPrereq(self):
3198 """Check prerequisites.
3200 This checks that the instance is in the cluster.
3203 instance = self.cfg.GetInstanceInfo(
3204 self.cfg.ExpandInstanceName(self.op.instance_name))
3205 if instance is None:
3206 raise errors.OpPrereqError("Instance '%s' not known" %
3207 self.op.instance_name)
3208 self.instance = instance
3210 def Exec(self, feedback_fn):
3211 """Connect to the console of an instance
3214 instance = self.instance
3215 node = instance.primary_node
3217 node_insts = rpc.call_instance_list([node])[node]
3218 if node_insts is False:
3219 raise errors.OpExecError("Can't connect to node %s." % node)
3221 if instance.name not in node_insts:
3222 raise errors.OpExecError("Instance %s is not running." % instance.name)
3224 logger.Debug("connecting to console of %s on %s" % (instance.name, node))
3226 hyper = hypervisor.GetHypervisor()
3227 console_cmd = hyper.GetShellCommandForConsole(instance)
3229 argv = ["ssh", "-q", "-t"]
3230 argv.extend(ssh.KNOWN_HOSTS_OPTS)
3231 argv.extend(ssh.BATCH_MODE_OPTS)
3233 argv.append(console_cmd)
3237 class LUAddMDDRBDComponent(LogicalUnit):
3238 """Adda new mirror member to an instance's disk.
3241 HPATH = "mirror-add"
3242 HTYPE = constants.HTYPE_INSTANCE
3243 _OP_REQP = ["instance_name", "remote_node", "disk_name"]
3245 def BuildHooksEnv(self):
3248 This runs on the master, the primary and all the secondaries.
3252 "NEW_SECONDARY": self.op.remote_node,
3253 "DISK_NAME": self.op.disk_name,
3255 env.update(_BuildInstanceHookEnvByObject(self.instance))
3256 nl = [self.sstore.GetMasterNode(), self.instance.primary_node,
3257 self.op.remote_node,] + list(self.instance.secondary_nodes)
3260 def CheckPrereq(self):
3261 """Check prerequisites.
3263 This checks that the instance is in the cluster.
3266 instance = self.cfg.GetInstanceInfo(
3267 self.cfg.ExpandInstanceName(self.op.instance_name))
3268 if instance is None:
3269 raise errors.OpPrereqError("Instance '%s' not known" %
3270 self.op.instance_name)
3271 self.instance = instance
3273 remote_node = self.cfg.ExpandNodeName(self.op.remote_node)
3274 if remote_node is None:
3275 raise errors.OpPrereqError("Node '%s' not known" % self.op.remote_node)
3276 self.remote_node = remote_node
3278 if remote_node == instance.primary_node:
3279 raise errors.OpPrereqError("The specified node is the primary node of"
3282 if instance.disk_template != constants.DT_REMOTE_RAID1:
3283 raise errors.OpPrereqError("Instance's disk layout is not"
3285 for disk in instance.disks:
3286 if disk.iv_name == self.op.disk_name:
3289 raise errors.OpPrereqError("Can't find this device ('%s') in the"
3290 " instance." % self.op.disk_name)
3291 if len(disk.children) > 1:
3292 raise errors.OpPrereqError("The device already has two slave devices."
3293 " This would create a 3-disk raid1 which we"
3297 def Exec(self, feedback_fn):
3298 """Add the mirror component
3302 instance = self.instance
3304 remote_node = self.remote_node
3305 lv_names = [".%s_%s" % (disk.iv_name, suf) for suf in ["data", "meta"]]
3306 names = _GenerateUniqueNames(self.cfg, lv_names)
3307 new_drbd = _GenerateMDDRBDBranch(self.cfg, instance.primary_node,
3308 remote_node, disk.size, names)
3310 logger.Info("adding new mirror component on secondary")
3312 if not _CreateBlockDevOnSecondary(self.cfg, remote_node, instance,
3314 _GetInstanceInfoText(instance)):
3315 raise errors.OpExecError("Failed to create new component on secondary"
3316 " node %s" % remote_node)
3318 logger.Info("adding new mirror component on primary")
3320 if not _CreateBlockDevOnPrimary(self.cfg, instance.primary_node,
3322 _GetInstanceInfoText(instance)):
3323 # remove secondary dev
3324 self.cfg.SetDiskID(new_drbd, remote_node)
3325 rpc.call_blockdev_remove(remote_node, new_drbd)
3326 raise errors.OpExecError("Failed to create volume on primary")
3328 # the device exists now
3329 # call the primary node to add the mirror to md
3330 logger.Info("adding new mirror component to md")
3331 if not rpc.call_blockdev_addchildren(instance.primary_node,
3333 logger.Error("Can't add mirror compoment to md!")
3334 self.cfg.SetDiskID(new_drbd, remote_node)
3335 if not rpc.call_blockdev_remove(remote_node, new_drbd):
3336 logger.Error("Can't rollback on secondary")
3337 self.cfg.SetDiskID(new_drbd, instance.primary_node)
3338 if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
3339 logger.Error("Can't rollback on primary")
3340 raise errors.OpExecError("Can't add mirror component to md array")
3342 disk.children.append(new_drbd)
3344 self.cfg.AddInstance(instance)
3346 _WaitForSync(self.cfg, instance, self.proc)
3351 class LURemoveMDDRBDComponent(LogicalUnit):
3352 """Remove a component from a remote_raid1 disk.
3355 HPATH = "mirror-remove"
3356 HTYPE = constants.HTYPE_INSTANCE
3357 _OP_REQP = ["instance_name", "disk_name", "disk_id"]
3359 def BuildHooksEnv(self):
3362 This runs on the master, the primary and all the secondaries.
3366 "DISK_NAME": self.op.disk_name,
3367 "DISK_ID": self.op.disk_id,
3368 "OLD_SECONDARY": self.old_secondary,
3370 env.update(_BuildInstanceHookEnvByObject(self.instance))
3371 nl = [self.sstore.GetMasterNode(),
3372 self.instance.primary_node] + list(self.instance.secondary_nodes)
3375 def CheckPrereq(self):
3376 """Check prerequisites.
3378 This checks that the instance is in the cluster.
3381 instance = self.cfg.GetInstanceInfo(
3382 self.cfg.ExpandInstanceName(self.op.instance_name))
3383 if instance is None:
3384 raise errors.OpPrereqError("Instance '%s' not known" %
3385 self.op.instance_name)
3386 self.instance = instance
3388 if instance.disk_template != constants.DT_REMOTE_RAID1:
3389 raise errors.OpPrereqError("Instance's disk layout is not"
3391 for disk in instance.disks:
3392 if disk.iv_name == self.op.disk_name:
3395 raise errors.OpPrereqError("Can't find this device ('%s') in the"
3396 " instance." % self.op.disk_name)
3397 for child in disk.children:
3398 if (child.dev_type == constants.LD_DRBD7 and
3399 child.logical_id[2] == self.op.disk_id):
3402 raise errors.OpPrereqError("Can't find the device with this port.")
3404 if len(disk.children) < 2:
3405 raise errors.OpPrereqError("Cannot remove the last component from"
3409 if self.child.logical_id[0] == instance.primary_node:
3413 self.old_secondary = self.child.logical_id[oid]
3415 def Exec(self, feedback_fn):
3416 """Remove the mirror component
3419 instance = self.instance
3422 logger.Info("remove mirror component")
3423 self.cfg.SetDiskID(disk, instance.primary_node)
3424 if not rpc.call_blockdev_removechildren(instance.primary_node,
3426 raise errors.OpExecError("Can't remove child from mirror.")
3428 for node in child.logical_id[:2]:
3429 self.cfg.SetDiskID(child, node)
3430 if not rpc.call_blockdev_remove(node, child):
3431 logger.Error("Warning: failed to remove device from node %s,"
3432 " continuing operation." % node)
3434 disk.children.remove(child)
3435 self.cfg.AddInstance(instance)
3438 class LUReplaceDisks(LogicalUnit):
3439 """Replace the disks of an instance.
3442 HPATH = "mirrors-replace"
3443 HTYPE = constants.HTYPE_INSTANCE
3444 _OP_REQP = ["instance_name", "mode", "disks"]
3446 def BuildHooksEnv(self):
3449 This runs on the master, the primary and all the secondaries.
3453 "MODE": self.op.mode,
3454 "NEW_SECONDARY": self.op.remote_node,
3455 "OLD_SECONDARY": self.instance.secondary_nodes[0],
3457 env.update(_BuildInstanceHookEnvByObject(self.instance))
3459 self.sstore.GetMasterNode(),
3460 self.instance.primary_node,
3462 if self.op.remote_node is not None:
3463 nl.append(self.op.remote_node)
3466 def CheckPrereq(self):
3467 """Check prerequisites.
3469 This checks that the instance is in the cluster.
3472 instance = self.cfg.GetInstanceInfo(
3473 self.cfg.ExpandInstanceName(self.op.instance_name))
3474 if instance is None:
3475 raise errors.OpPrereqError("Instance '%s' not known" %
3476 self.op.instance_name)
3477 self.instance = instance
3478 self.op.instance_name = instance.name
3480 if instance.disk_template not in constants.DTS_NET_MIRROR:
3481 raise errors.OpPrereqError("Instance's disk layout is not"
3482 " network mirrored.")
3484 if len(instance.secondary_nodes) != 1:
3485 raise errors.OpPrereqError("The instance has a strange layout,"
3486 " expected one secondary but found %d" %
3487 len(instance.secondary_nodes))
3489 self.sec_node = instance.secondary_nodes[0]
3491 remote_node = getattr(self.op, "remote_node", None)
3492 if remote_node is not None:
3493 remote_node = self.cfg.ExpandNodeName(remote_node)
3494 if remote_node is None:
3495 raise errors.OpPrereqError("Node '%s' not known" %
3496 self.op.remote_node)
3497 self.remote_node_info = self.cfg.GetNodeInfo(remote_node)
3499 self.remote_node_info = None
3500 if remote_node == instance.primary_node:
3501 raise errors.OpPrereqError("The specified node is the primary node of"
3503 elif remote_node == self.sec_node:
3504 if self.op.mode == constants.REPLACE_DISK_SEC:
3505 # this is for DRBD8, where we can't execute the same mode of
3506 # replacement as for drbd7 (no different port allocated)
3507 raise errors.OpPrereqError("Same secondary given, cannot execute"
3509 # the user gave the current secondary, switch to
3510 # 'no-replace-secondary' mode for drbd7
3512 if (instance.disk_template == constants.DT_REMOTE_RAID1 and
3513 self.op.mode != constants.REPLACE_DISK_ALL):
3514 raise errors.OpPrereqError("Template 'remote_raid1' only allows all"
3515 " disks replacement, not individual ones")
3516 if instance.disk_template == constants.DT_DRBD8:
3517 if (self.op.mode == constants.REPLACE_DISK_ALL and
3518 remote_node is not None):
3519 # switch to replace secondary mode
3520 self.op.mode = constants.REPLACE_DISK_SEC
3522 if self.op.mode == constants.REPLACE_DISK_ALL:
3523 raise errors.OpPrereqError("Template 'drbd' only allows primary or"
3524 " secondary disk replacement, not"
3526 elif self.op.mode == constants.REPLACE_DISK_PRI:
3527 if remote_node is not None:
3528 raise errors.OpPrereqError("Template 'drbd' does not allow changing"
3529 " the secondary while doing a primary"
3530 " node disk replacement")
3531 self.tgt_node = instance.primary_node
3532 self.oth_node = instance.secondary_nodes[0]
3533 elif self.op.mode == constants.REPLACE_DISK_SEC:
3534 self.new_node = remote_node # this can be None, in which case
3535 # we don't change the secondary
3536 self.tgt_node = instance.secondary_nodes[0]
3537 self.oth_node = instance.primary_node
3539 raise errors.ProgrammerError("Unhandled disk replace mode")
3541 for name in self.op.disks:
3542 if instance.FindDisk(name) is None:
3543 raise errors.OpPrereqError("Disk '%s' not found for instance '%s'" %
3544 (name, instance.name))
3545 self.op.remote_node = remote_node
3547 def _ExecRR1(self, feedback_fn):
3548 """Replace the disks of an instance.
3551 instance = self.instance
3554 if self.op.remote_node is None:
3555 remote_node = self.sec_node
3557 remote_node = self.op.remote_node
3559 for dev in instance.disks:
3561 lv_names = [".%s_%s" % (dev.iv_name, suf) for suf in ["data", "meta"]]
3562 names = _GenerateUniqueNames(cfg, lv_names)
3563 new_drbd = _GenerateMDDRBDBranch(cfg, instance.primary_node,
3564 remote_node, size, names)
3565 iv_names[dev.iv_name] = (dev, dev.children[0], new_drbd)
3566 logger.Info("adding new mirror component on secondary for %s" %
3569 if not _CreateBlockDevOnSecondary(cfg, remote_node, instance,
3571 _GetInstanceInfoText(instance)):
3572 raise errors.OpExecError("Failed to create new component on secondary"
3573 " node %s. Full abort, cleanup manually!" %
3576 logger.Info("adding new mirror component on primary")
3578 if not _CreateBlockDevOnPrimary(cfg, instance.primary_node,
3580 _GetInstanceInfoText(instance)):
3581 # remove secondary dev
3582 cfg.SetDiskID(new_drbd, remote_node)
3583 rpc.call_blockdev_remove(remote_node, new_drbd)
3584 raise errors.OpExecError("Failed to create volume on primary!"
3585 " Full abort, cleanup manually!!")
3587 # the device exists now
3588 # call the primary node to add the mirror to md
3589 logger.Info("adding new mirror component to md")
3590 if not rpc.call_blockdev_addchildren(instance.primary_node, dev,
3592 logger.Error("Can't add mirror compoment to md!")
3593 cfg.SetDiskID(new_drbd, remote_node)
3594 if not rpc.call_blockdev_remove(remote_node, new_drbd):
3595 logger.Error("Can't rollback on secondary")
3596 cfg.SetDiskID(new_drbd, instance.primary_node)
3597 if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
3598 logger.Error("Can't rollback on primary")
3599 raise errors.OpExecError("Full abort, cleanup manually!!")
3601 dev.children.append(new_drbd)
3602 cfg.AddInstance(instance)
3604 # this can fail as the old devices are degraded and _WaitForSync
3605 # does a combined result over all disks, so we don't check its
3607 _WaitForSync(cfg, instance, self.proc, unlock=True)
3609 # so check manually all the devices
3610 for name in iv_names:
3611 dev, child, new_drbd = iv_names[name]
3612 cfg.SetDiskID(dev, instance.primary_node)
3613 is_degr = rpc.call_blockdev_find(instance.primary_node, dev)[5]
3615 raise errors.OpExecError("MD device %s is degraded!" % name)
3616 cfg.SetDiskID(new_drbd, instance.primary_node)
3617 is_degr = rpc.call_blockdev_find(instance.primary_node, new_drbd)[5]
3619 raise errors.OpExecError("New drbd device %s is degraded!" % name)
3621 for name in iv_names:
3622 dev, child, new_drbd = iv_names[name]
3623 logger.Info("remove mirror %s component" % name)
3624 cfg.SetDiskID(dev, instance.primary_node)
3625 if not rpc.call_blockdev_removechildren(instance.primary_node,
3627 logger.Error("Can't remove child from mirror, aborting"
3628 " *this device cleanup*.\nYou need to cleanup manually!!")
3631 for node in child.logical_id[:2]:
3632 logger.Info("remove child device on %s" % node)
3633 cfg.SetDiskID(child, node)
3634 if not rpc.call_blockdev_remove(node, child):
3635 logger.Error("Warning: failed to remove device from node %s,"
3636 " continuing operation." % node)
3638 dev.children.remove(child)
3640 cfg.AddInstance(instance)
3642 def _ExecD8DiskOnly(self, feedback_fn):
3643 """Replace a disk on the primary or secondary for dbrd8.
3645 The algorithm for replace is quite complicated:
3646 - for each disk to be replaced:
3647 - create new LVs on the target node with unique names
3648 - detach old LVs from the drbd device
3649 - rename old LVs to name_replaced.<time_t>
3650 - rename new LVs to old LVs
3651 - attach the new LVs (with the old names now) to the drbd device
3652 - wait for sync across all devices
3653 - for each modified disk:
3654 - remove old LVs (which have the name name_replaces.<time_t>)
3656 Failures are not very well handled.
3660 warning, info = (self.proc.LogWarning, self.proc.LogInfo)
3661 instance = self.instance
3663 vgname = self.cfg.GetVGName()
3666 tgt_node = self.tgt_node
3667 oth_node = self.oth_node
3669 # Step: check device activation
3670 self.proc.LogStep(1, steps_total, "check device existence")
3671 info("checking volume groups")
3672 my_vg = cfg.GetVGName()
3673 results = rpc.call_vg_list([oth_node, tgt_node])
3675 raise errors.OpExecError("Can't list volume groups on the nodes")
3676 for node in oth_node, tgt_node:
3677 res = results.get(node, False)
3678 if not res or my_vg not in res:
3679 raise errors.OpExecError("Volume group '%s' not found on %s" %
3681 for dev in instance.disks:
3682 if not dev.iv_name in self.op.disks:
3684 for node in tgt_node, oth_node:
3685 info("checking %s on %s" % (dev.iv_name, node))
3686 cfg.SetDiskID(dev, node)
3687 if not rpc.call_blockdev_find(node, dev):
3688 raise errors.OpExecError("Can't find device %s on node %s" %
3689 (dev.iv_name, node))
3691 # Step: check other node consistency
3692 self.proc.LogStep(2, steps_total, "check peer consistency")
3693 for dev in instance.disks:
3694 if not dev.iv_name in self.op.disks:
3696 info("checking %s consistency on %s" % (dev.iv_name, oth_node))
3697 if not _CheckDiskConsistency(self.cfg, dev, oth_node,
3698 oth_node==instance.primary_node):
3699 raise errors.OpExecError("Peer node (%s) has degraded storage, unsafe"
3700 " to replace disks on this node (%s)" %
3701 (oth_node, tgt_node))
3703 # Step: create new storage
3704 self.proc.LogStep(3, steps_total, "allocate new storage")
3705 for dev in instance.disks:
3706 if not dev.iv_name in self.op.disks:
3709 cfg.SetDiskID(dev, tgt_node)
3710 lv_names = [".%s_%s" % (dev.iv_name, suf) for suf in ["data", "meta"]]
3711 names = _GenerateUniqueNames(cfg, lv_names)
3712 lv_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3713 logical_id=(vgname, names[0]))
3714 lv_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3715 logical_id=(vgname, names[1]))
3716 new_lvs = [lv_data, lv_meta]
3717 old_lvs = dev.children
3718 iv_names[dev.iv_name] = (dev, old_lvs, new_lvs)
3719 info("creating new local storage on %s for %s" %
3720 (tgt_node, dev.iv_name))
3721 # since we *always* want to create this LV, we use the
3722 # _Create...OnPrimary (which forces the creation), even if we
3723 # are talking about the secondary node
3724 for new_lv in new_lvs:
3725 if not _CreateBlockDevOnPrimary(cfg, tgt_node, instance, new_lv,
3726 _GetInstanceInfoText(instance)):
3727 raise errors.OpExecError("Failed to create new LV named '%s' on"
3729 (new_lv.logical_id[1], tgt_node))
3731 # Step: for each lv, detach+rename*2+attach
3732 self.proc.LogStep(4, steps_total, "change drbd configuration")
3733 for dev, old_lvs, new_lvs in iv_names.itervalues():
3734 info("detaching %s drbd from local storage" % dev.iv_name)
3735 if not rpc.call_blockdev_removechildren(tgt_node, dev, old_lvs):
3736 raise errors.OpExecError("Can't detach drbd from local storage on node"
3737 " %s for device %s" % (tgt_node, dev.iv_name))
3739 #cfg.Update(instance)
3741 # ok, we created the new LVs, so now we know we have the needed
3742 # storage; as such, we proceed on the target node to rename
3743 # old_lv to _old, and new_lv to old_lv; note that we rename LVs
3744 # using the assumption than logical_id == physical_id (which in
3745 # turn is the unique_id on that node)
3747 # FIXME(iustin): use a better name for the replaced LVs
3748 temp_suffix = int(time.time())
3749 ren_fn = lambda d, suff: (d.physical_id[0],
3750 d.physical_id[1] + "_replaced-%s" % suff)
3751 # build the rename list based on what LVs exist on the node
3753 for to_ren in old_lvs:
3754 find_res = rpc.call_blockdev_find(tgt_node, to_ren)
3755 if find_res is not None: # device exists
3756 rlist.append((to_ren, ren_fn(to_ren, temp_suffix)))
3758 info("renaming the old LVs on the target node")
3759 if not rpc.call_blockdev_rename(tgt_node, rlist):
3760 raise errors.OpExecError("Can't rename old LVs on node %s" % tgt_node)
3761 # now we rename the new LVs to the old LVs
3762 info("renaming the new LVs on the target node")
3763 rlist = [(new, old.physical_id) for old, new in zip(old_lvs, new_lvs)]
3764 if not rpc.call_blockdev_rename(tgt_node, rlist):
3765 raise errors.OpExecError("Can't rename new LVs on node %s" % tgt_node)
3767 for old, new in zip(old_lvs, new_lvs):
3768 new.logical_id = old.logical_id
3769 cfg.SetDiskID(new, tgt_node)
3771 for disk in old_lvs:
3772 disk.logical_id = ren_fn(disk, temp_suffix)
3773 cfg.SetDiskID(disk, tgt_node)
3775 # now that the new lvs have the old name, we can add them to the device
3776 info("adding new mirror component on %s" % tgt_node)
3777 if not rpc.call_blockdev_addchildren(tgt_node, dev, new_lvs):
3778 for new_lv in new_lvs:
3779 if not rpc.call_blockdev_remove(tgt_node, new_lv):
3780 warning("Can't rollback device %s", hint="manually cleanup unused"
3782 raise errors.OpExecError("Can't add local storage to drbd")
3784 dev.children = new_lvs
3785 cfg.Update(instance)
3787 # Step: wait for sync
3789 # this can fail as the old devices are degraded and _WaitForSync
3790 # does a combined result over all disks, so we don't check its
3792 self.proc.LogStep(5, steps_total, "sync devices")
3793 _WaitForSync(cfg, instance, self.proc, unlock=True)
3795 # so check manually all the devices
3796 for name, (dev, old_lvs, new_lvs) in iv_names.iteritems():
3797 cfg.SetDiskID(dev, instance.primary_node)
3798 is_degr = rpc.call_blockdev_find(instance.primary_node, dev)[5]
3800 raise errors.OpExecError("DRBD device %s is degraded!" % name)
3802 # Step: remove old storage
3803 self.proc.LogStep(6, steps_total, "removing old storage")
3804 for name, (dev, old_lvs, new_lvs) in iv_names.iteritems():
3805 info("remove logical volumes for %s" % name)
3807 cfg.SetDiskID(lv, tgt_node)
3808 if not rpc.call_blockdev_remove(tgt_node, lv):
3809 warning("Can't remove old LV", hint="manually remove unused LVs")
3812 def _ExecD8Secondary(self, feedback_fn):
3813 """Replace the secondary node for drbd8.
3815 The algorithm for replace is quite complicated:
3816 - for all disks of the instance:
3817 - create new LVs on the new node with same names
3818 - shutdown the drbd device on the old secondary
3819 - disconnect the drbd network on the primary
3820 - create the drbd device on the new secondary
3821 - network attach the drbd on the primary, using an artifice:
3822 the drbd code for Attach() will connect to the network if it
3823 finds a device which is connected to the good local disks but
3825 - wait for sync across all devices
3826 - remove all disks from the old secondary
3828 Failures are not very well handled.
3832 warning, info = (self.proc.LogWarning, self.proc.LogInfo)
3833 instance = self.instance
3835 vgname = self.cfg.GetVGName()
3838 old_node = self.tgt_node
3839 new_node = self.new_node
3840 pri_node = instance.primary_node
3842 # Step: check device activation
3843 self.proc.LogStep(1, steps_total, "check device existence")
3844 info("checking volume groups")
3845 my_vg = cfg.GetVGName()
3846 results = rpc.call_vg_list([pri_node, new_node])
3848 raise errors.OpExecError("Can't list volume groups on the nodes")
3849 for node in pri_node, new_node:
3850 res = results.get(node, False)
3851 if not res or my_vg not in res:
3852 raise errors.OpExecError("Volume group '%s' not found on %s" %
3854 for dev in instance.disks:
3855 if not dev.iv_name in self.op.disks:
3857 info("checking %s on %s" % (dev.iv_name, pri_node))
3858 cfg.SetDiskID(dev, pri_node)
3859 if not rpc.call_blockdev_find(pri_node, dev):
3860 raise errors.OpExecError("Can't find device %s on node %s" %
3861 (dev.iv_name, pri_node))
3863 # Step: check other node consistency
3864 self.proc.LogStep(2, steps_total, "check peer consistency")
3865 for dev in instance.disks:
3866 if not dev.iv_name in self.op.disks:
3868 info("checking %s consistency on %s" % (dev.iv_name, pri_node))
3869 if not _CheckDiskConsistency(self.cfg, dev, pri_node, True, ldisk=True):
3870 raise errors.OpExecError("Primary node (%s) has degraded storage,"
3871 " unsafe to replace the secondary" %
3874 # Step: create new storage
3875 self.proc.LogStep(3, steps_total, "allocate new storage")
3876 for dev in instance.disks:
3878 info("adding new local storage on %s for %s" % (new_node, dev.iv_name))
3879 # since we *always* want to create this LV, we use the
3880 # _Create...OnPrimary (which forces the creation), even if we
3881 # are talking about the secondary node
3882 for new_lv in dev.children:
3883 if not _CreateBlockDevOnPrimary(cfg, new_node, instance, new_lv,
3884 _GetInstanceInfoText(instance)):
3885 raise errors.OpExecError("Failed to create new LV named '%s' on"
3887 (new_lv.logical_id[1], new_node))
3889 iv_names[dev.iv_name] = (dev, dev.children)
3891 self.proc.LogStep(4, steps_total, "changing drbd configuration")
3892 for dev in instance.disks:
3894 info("activating a new drbd on %s for %s" % (new_node, dev.iv_name))
3895 # create new devices on new_node
3896 new_drbd = objects.Disk(dev_type=constants.LD_DRBD8,
3897 logical_id=(pri_node, new_node,
3899 children=dev.children)
3900 if not _CreateBlockDevOnSecondary(cfg, new_node, instance,
3902 _GetInstanceInfoText(instance)):
3903 raise errors.OpExecError("Failed to create new DRBD on"
3904 " node '%s'" % new_node)
3906 for dev in instance.disks:
3907 # we have new devices, shutdown the drbd on the old secondary
3908 info("shutting down drbd for %s on old node" % dev.iv_name)
3909 cfg.SetDiskID(dev, old_node)
3910 if not rpc.call_blockdev_shutdown(old_node, dev):
3911 warning("Failed to shutdown drbd for %s on old node" % dev.iv_name,
3912 hint="Please cleanup this device manually as soon as possible")
3914 info("detaching primary drbds from the network (=> standalone)")
3916 for dev in instance.disks:
3917 cfg.SetDiskID(dev, pri_node)
3918 # set the physical (unique in bdev terms) id to None, meaning
3919 # detach from network
3920 dev.physical_id = (None,) * len(dev.physical_id)
3921 # and 'find' the device, which will 'fix' it to match the
3923 if rpc.call_blockdev_find(pri_node, dev):
3926 warning("Failed to detach drbd %s from network, unusual case" %
3930 # no detaches succeeded (very unlikely)
3931 raise errors.OpExecError("Can't detach at least one DRBD from old node")
3933 # if we managed to detach at least one, we update all the disks of
3934 # the instance to point to the new secondary
3935 info("updating instance configuration")
3936 for dev in instance.disks:
3937 dev.logical_id = (pri_node, new_node) + dev.logical_id[2:]
3938 cfg.SetDiskID(dev, pri_node)
3939 cfg.Update(instance)
3941 # and now perform the drbd attach
3942 info("attaching primary drbds to new secondary (standalone => connected)")
3944 for dev in instance.disks:
3945 info("attaching primary drbd for %s to new secondary node" % dev.iv_name)
3946 # since the attach is smart, it's enough to 'find' the device,
3947 # it will automatically activate the network, if the physical_id
3949 cfg.SetDiskID(dev, pri_node)
3950 if not rpc.call_blockdev_find(pri_node, dev):
3951 warning("can't attach drbd %s to new secondary!" % dev.iv_name,
3952 "please do a gnt-instance info to see the status of disks")
3954 # this can fail as the old devices are degraded and _WaitForSync
3955 # does a combined result over all disks, so we don't check its
3957 self.proc.LogStep(5, steps_total, "sync devices")
3958 _WaitForSync(cfg, instance, self.proc, unlock=True)
3960 # so check manually all the devices
3961 for name, (dev, old_lvs) in iv_names.iteritems():
3962 cfg.SetDiskID(dev, pri_node)
3963 is_degr = rpc.call_blockdev_find(pri_node, dev)[5]
3965 raise errors.OpExecError("DRBD device %s is degraded!" % name)
3967 self.proc.LogStep(6, steps_total, "removing old storage")
3968 for name, (dev, old_lvs) in iv_names.iteritems():
3969 info("remove logical volumes for %s" % name)
3971 cfg.SetDiskID(lv, old_node)
3972 if not rpc.call_blockdev_remove(old_node, lv):
3973 warning("Can't remove LV on old secondary",
3974 hint="Cleanup stale volumes by hand")
3976 def Exec(self, feedback_fn):
3977 """Execute disk replacement.
3979 This dispatches the disk replacement to the appropriate handler.
3982 instance = self.instance
3983 if instance.disk_template == constants.DT_REMOTE_RAID1:
3985 elif instance.disk_template == constants.DT_DRBD8:
3986 if self.op.remote_node is None:
3987 fn = self._ExecD8DiskOnly
3989 fn = self._ExecD8Secondary
3991 raise errors.ProgrammerError("Unhandled disk replacement case")
3992 return fn(feedback_fn)
3995 class LUQueryInstanceData(NoHooksLU):
3996 """Query runtime instance data.
3999 _OP_REQP = ["instances"]
4001 def CheckPrereq(self):
4002 """Check prerequisites.
4004 This only checks the optional instance list against the existing names.
4007 if not isinstance(self.op.instances, list):
4008 raise errors.OpPrereqError("Invalid argument type 'instances'")
4009 if self.op.instances:
4010 self.wanted_instances = []
4011 names = self.op.instances
4013 instance = self.cfg.GetInstanceInfo(self.cfg.ExpandInstanceName(name))
4014 if instance is None:
4015 raise errors.OpPrereqError("No such instance name '%s'" % name)
4016 self.wanted_instances.append(instance)
4018 self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
4019 in self.cfg.GetInstanceList()]
4023 def _ComputeDiskStatus(self, instance, snode, dev):
4024 """Compute block device status.
4027 self.cfg.SetDiskID(dev, instance.primary_node)
4028 dev_pstatus = rpc.call_blockdev_find(instance.primary_node, dev)
4029 if dev.dev_type in constants.LDS_DRBD:
4030 # we change the snode then (otherwise we use the one passed in)
4031 if dev.logical_id[0] == instance.primary_node:
4032 snode = dev.logical_id[1]
4034 snode = dev.logical_id[0]
4037 self.cfg.SetDiskID(dev, snode)
4038 dev_sstatus = rpc.call_blockdev_find(snode, dev)
4043 dev_children = [self._ComputeDiskStatus(instance, snode, child)
4044 for child in dev.children]
4049 "iv_name": dev.iv_name,
4050 "dev_type": dev.dev_type,
4051 "logical_id": dev.logical_id,
4052 "physical_id": dev.physical_id,
4053 "pstatus": dev_pstatus,
4054 "sstatus": dev_sstatus,
4055 "children": dev_children,
4060 def Exec(self, feedback_fn):
4061 """Gather and return data"""
4063 for instance in self.wanted_instances:
4064 remote_info = rpc.call_instance_info(instance.primary_node,
4066 if remote_info and "state" in remote_info:
4069 remote_state = "down"
4070 if instance.status == "down":
4071 config_state = "down"
4075 disks = [self._ComputeDiskStatus(instance, None, device)
4076 for device in instance.disks]
4079 "name": instance.name,
4080 "config_state": config_state,
4081 "run_state": remote_state,
4082 "pnode": instance.primary_node,
4083 "snodes": instance.secondary_nodes,
4085 "memory": instance.memory,
4086 "nics": [(nic.mac, nic.ip, nic.bridge) for nic in instance.nics],
4088 "network_port": instance.network_port,
4089 "vcpus": instance.vcpus,
4090 "kernel_path": instance.kernel_path,
4091 "initrd_path": instance.initrd_path,
4092 "hvm_boot_order": instance.hvm_boot_order,
4095 result[instance.name] = idict
4100 class LUSetInstanceParms(LogicalUnit):
4101 """Modifies an instances's parameters.
4104 HPATH = "instance-modify"
4105 HTYPE = constants.HTYPE_INSTANCE
4106 _OP_REQP = ["instance_name"]
4108 def BuildHooksEnv(self):
4111 This runs on the master, primary and secondaries.
4116 args['memory'] = self.mem
4118 args['vcpus'] = self.vcpus
4119 if self.do_ip or self.do_bridge:
4123 ip = self.instance.nics[0].ip
4125 bridge = self.bridge
4127 bridge = self.instance.nics[0].bridge
4128 args['nics'] = [(ip, bridge)]
4129 env = _BuildInstanceHookEnvByObject(self.instance, override=args)
4130 nl = [self.sstore.GetMasterNode(),
4131 self.instance.primary_node] + list(self.instance.secondary_nodes)
4134 def CheckPrereq(self):
4135 """Check prerequisites.
4137 This only checks the instance list against the existing names.
4140 self.mem = getattr(self.op, "mem", None)
4141 self.vcpus = getattr(self.op, "vcpus", None)
4142 self.ip = getattr(self.op, "ip", None)
4143 self.mac = getattr(self.op, "mac", None)
4144 self.bridge = getattr(self.op, "bridge", None)
4145 self.kernel_path = getattr(self.op, "kernel_path", None)
4146 self.initrd_path = getattr(self.op, "initrd_path", None)
4147 self.hvm_boot_order = getattr(self.op, "hvm_boot_order", None)
4148 all_parms = [self.mem, self.vcpus, self.ip, self.bridge, self.mac,
4149 self.kernel_path, self.initrd_path, self.hvm_boot_order]
4150 if all_parms.count(None) == len(all_parms):
4151 raise errors.OpPrereqError("No changes submitted")
4152 if self.mem is not None:
4154 self.mem = int(self.mem)
4155 except ValueError, err:
4156 raise errors.OpPrereqError("Invalid memory size: %s" % str(err))
4157 if self.vcpus is not None:
4159 self.vcpus = int(self.vcpus)
4160 except ValueError, err:
4161 raise errors.OpPrereqError("Invalid vcpus number: %s" % str(err))
4162 if self.ip is not None:
4164 if self.ip.lower() == "none":
4167 if not utils.IsValidIP(self.ip):
4168 raise errors.OpPrereqError("Invalid IP address '%s'." % self.ip)
4171 self.do_bridge = (self.bridge is not None)
4172 if self.mac is not None:
4173 if self.cfg.IsMacInUse(self.mac):
4174 raise errors.OpPrereqError('MAC address %s already in use in cluster' %
4176 if not utils.IsValidMac(self.mac):
4177 raise errors.OpPrereqError('Invalid MAC address %s' % self.mac)
4179 if self.kernel_path is not None:
4180 self.do_kernel_path = True
4181 if self.kernel_path == constants.VALUE_NONE:
4182 raise errors.OpPrereqError("Can't set instance to no kernel")
4184 if self.kernel_path != constants.VALUE_DEFAULT:
4185 if not os.path.isabs(self.kernel_path):
4186 raise errors.OpPrereqError("The kernel path must be an absolute"
4189 self.do_kernel_path = False
4191 if self.initrd_path is not None:
4192 self.do_initrd_path = True
4193 if self.initrd_path not in (constants.VALUE_NONE,
4194 constants.VALUE_DEFAULT):
4195 if not os.path.isabs(self.initrd_path):
4196 raise errors.OpPrereqError("The initrd path must be an absolute"
4199 self.do_initrd_path = False
4201 # boot order verification
4202 if self.hvm_boot_order is not None:
4203 if self.hvm_boot_order != constants.VALUE_DEFAULT:
4204 if len(self.hvm_boot_order.strip("acdn")) != 0:
4205 raise errors.OpPrereqError("invalid boot order specified,"
4206 " must be one or more of [acdn]"
4209 instance = self.cfg.GetInstanceInfo(
4210 self.cfg.ExpandInstanceName(self.op.instance_name))
4211 if instance is None:
4212 raise errors.OpPrereqError("No such instance name '%s'" %
4213 self.op.instance_name)
4214 self.op.instance_name = instance.name
4215 self.instance = instance
4218 def Exec(self, feedback_fn):
4219 """Modifies an instance.
4221 All parameters take effect only at the next restart of the instance.
4224 instance = self.instance
4226 instance.memory = self.mem
4227 result.append(("mem", self.mem))
4229 instance.vcpus = self.vcpus
4230 result.append(("vcpus", self.vcpus))
4232 instance.nics[0].ip = self.ip
4233 result.append(("ip", self.ip))
4235 instance.nics[0].bridge = self.bridge
4236 result.append(("bridge", self.bridge))
4238 instance.nics[0].mac = self.mac
4239 result.append(("mac", self.mac))
4240 if self.do_kernel_path:
4241 instance.kernel_path = self.kernel_path
4242 result.append(("kernel_path", self.kernel_path))
4243 if self.do_initrd_path:
4244 instance.initrd_path = self.initrd_path
4245 result.append(("initrd_path", self.initrd_path))
4246 if self.hvm_boot_order:
4247 if self.hvm_boot_order == constants.VALUE_DEFAULT:
4248 instance.hvm_boot_order = None
4250 instance.hvm_boot_order = self.hvm_boot_order
4251 result.append(("hvm_boot_order", self.hvm_boot_order))
4253 self.cfg.AddInstance(instance)
4258 class LUQueryExports(NoHooksLU):
4259 """Query the exports list
4264 def CheckPrereq(self):
4265 """Check that the nodelist contains only existing nodes.
4268 self.nodes = _GetWantedNodes(self, getattr(self.op, "nodes", None))
4270 def Exec(self, feedback_fn):
4271 """Compute the list of all the exported system images.
4274 a dictionary with the structure node->(export-list)
4275 where export-list is a list of the instances exported on
4279 return rpc.call_export_list(self.nodes)
4282 class LUExportInstance(LogicalUnit):
4283 """Export an instance to an image in the cluster.
4286 HPATH = "instance-export"
4287 HTYPE = constants.HTYPE_INSTANCE
4288 _OP_REQP = ["instance_name", "target_node", "shutdown"]
4290 def BuildHooksEnv(self):
4293 This will run on the master, primary node and target node.
4297 "EXPORT_NODE": self.op.target_node,
4298 "EXPORT_DO_SHUTDOWN": self.op.shutdown,
4300 env.update(_BuildInstanceHookEnvByObject(self.instance))
4301 nl = [self.sstore.GetMasterNode(), self.instance.primary_node,
4302 self.op.target_node]
4305 def CheckPrereq(self):
4306 """Check prerequisites.
4308 This checks that the instance name is a valid one.
4311 instance_name = self.cfg.ExpandInstanceName(self.op.instance_name)
4312 self.instance = self.cfg.GetInstanceInfo(instance_name)
4313 if self.instance is None:
4314 raise errors.OpPrereqError("Instance '%s' not found" %
4315 self.op.instance_name)
4318 dst_node_short = self.cfg.ExpandNodeName(self.op.target_node)
4319 self.dst_node = self.cfg.GetNodeInfo(dst_node_short)
4321 if self.dst_node is None:
4322 raise errors.OpPrereqError("Destination node '%s' is unknown." %
4323 self.op.target_node)
4324 self.op.target_node = self.dst_node.name
4326 def Exec(self, feedback_fn):
4327 """Export an instance to an image in the cluster.
4330 instance = self.instance
4331 dst_node = self.dst_node
4332 src_node = instance.primary_node
4333 # shutdown the instance, unless requested not to do so
4334 if self.op.shutdown:
4335 op = opcodes.OpShutdownInstance(instance_name=instance.name)
4336 self.proc.ChainOpCode(op)
4338 vgname = self.cfg.GetVGName()
4343 for disk in instance.disks:
4344 if disk.iv_name == "sda":
4345 # new_dev_name will be a snapshot of an lvm leaf of the one we passed
4346 new_dev_name = rpc.call_blockdev_snapshot(src_node, disk)
4348 if not new_dev_name:
4349 logger.Error("could not snapshot block device %s on node %s" %
4350 (disk.logical_id[1], src_node))
4352 new_dev = objects.Disk(dev_type=constants.LD_LV, size=disk.size,
4353 logical_id=(vgname, new_dev_name),
4354 physical_id=(vgname, new_dev_name),
4355 iv_name=disk.iv_name)
4356 snap_disks.append(new_dev)
4359 if self.op.shutdown:
4360 op = opcodes.OpStartupInstance(instance_name=instance.name,
4362 self.proc.ChainOpCode(op)
4364 # TODO: check for size
4366 for dev in snap_disks:
4367 if not rpc.call_snapshot_export(src_node, dev, dst_node.name,
4369 logger.Error("could not export block device %s from node"
4371 (dev.logical_id[1], src_node, dst_node.name))
4372 if not rpc.call_blockdev_remove(src_node, dev):
4373 logger.Error("could not remove snapshot block device %s from"
4374 " node %s" % (dev.logical_id[1], src_node))
4376 if not rpc.call_finalize_export(dst_node.name, instance, snap_disks):
4377 logger.Error("could not finalize export for instance %s on node %s" %
4378 (instance.name, dst_node.name))
4380 nodelist = self.cfg.GetNodeList()
4381 nodelist.remove(dst_node.name)
4383 # on one-node clusters nodelist will be empty after the removal
4384 # if we proceed the backup would be removed because OpQueryExports
4385 # substitutes an empty list with the full cluster node list.
4387 op = opcodes.OpQueryExports(nodes=nodelist)
4388 exportlist = self.proc.ChainOpCode(op)
4389 for node in exportlist:
4390 if instance.name in exportlist[node]:
4391 if not rpc.call_export_remove(node, instance.name):
4392 logger.Error("could not remove older export for instance %s"
4393 " on node %s" % (instance.name, node))
4396 class TagsLU(NoHooksLU):
4399 This is an abstract class which is the parent of all the other tags LUs.
4402 def CheckPrereq(self):
4403 """Check prerequisites.
4406 if self.op.kind == constants.TAG_CLUSTER:
4407 self.target = self.cfg.GetClusterInfo()
4408 elif self.op.kind == constants.TAG_NODE:
4409 name = self.cfg.ExpandNodeName(self.op.name)
4411 raise errors.OpPrereqError("Invalid node name (%s)" %
4414 self.target = self.cfg.GetNodeInfo(name)
4415 elif self.op.kind == constants.TAG_INSTANCE:
4416 name = self.cfg.ExpandInstanceName(self.op.name)
4418 raise errors.OpPrereqError("Invalid instance name (%s)" %
4421 self.target = self.cfg.GetInstanceInfo(name)
4423 raise errors.OpPrereqError("Wrong tag type requested (%s)" %
4427 class LUGetTags(TagsLU):
4428 """Returns the tags of a given object.
4431 _OP_REQP = ["kind", "name"]
4433 def Exec(self, feedback_fn):
4434 """Returns the tag list.
4437 return self.target.GetTags()
4440 class LUSearchTags(NoHooksLU):
4441 """Searches the tags for a given pattern.
4444 _OP_REQP = ["pattern"]
4446 def CheckPrereq(self):
4447 """Check prerequisites.
4449 This checks the pattern passed for validity by compiling it.
4453 self.re = re.compile(self.op.pattern)
4454 except re.error, err:
4455 raise errors.OpPrereqError("Invalid search pattern '%s': %s" %
4456 (self.op.pattern, err))
4458 def Exec(self, feedback_fn):
4459 """Returns the tag list.
4463 tgts = [("/cluster", cfg.GetClusterInfo())]
4464 ilist = [cfg.GetInstanceInfo(name) for name in cfg.GetInstanceList()]
4465 tgts.extend([("/instances/%s" % i.name, i) for i in ilist])
4466 nlist = [cfg.GetNodeInfo(name) for name in cfg.GetNodeList()]
4467 tgts.extend([("/nodes/%s" % n.name, n) for n in nlist])
4469 for path, target in tgts:
4470 for tag in target.GetTags():
4471 if self.re.search(tag):
4472 results.append((path, tag))
4476 class LUAddTags(TagsLU):
4477 """Sets a tag on a given object.
4480 _OP_REQP = ["kind", "name", "tags"]
4482 def CheckPrereq(self):
4483 """Check prerequisites.
4485 This checks the type and length of the tag name and value.
4488 TagsLU.CheckPrereq(self)
4489 for tag in self.op.tags:
4490 objects.TaggableObject.ValidateTag(tag)
4492 def Exec(self, feedback_fn):
4497 for tag in self.op.tags:
4498 self.target.AddTag(tag)
4499 except errors.TagError, err:
4500 raise errors.OpExecError("Error while setting tag: %s" % str(err))
4502 self.cfg.Update(self.target)
4503 except errors.ConfigurationError:
4504 raise errors.OpRetryError("There has been a modification to the"
4505 " config file and the operation has been"
4506 " aborted. Please retry.")
4509 class LUDelTags(TagsLU):
4510 """Delete a list of tags from a given object.
4513 _OP_REQP = ["kind", "name", "tags"]
4515 def CheckPrereq(self):
4516 """Check prerequisites.
4518 This checks that we have the given tag.
4521 TagsLU.CheckPrereq(self)
4522 for tag in self.op.tags:
4523 objects.TaggableObject.ValidateTag(tag)
4524 del_tags = frozenset(self.op.tags)
4525 cur_tags = self.target.GetTags()
4526 if not del_tags <= cur_tags:
4527 diff_tags = del_tags - cur_tags
4528 diff_names = ["'%s'" % tag for tag in diff_tags]
4530 raise errors.OpPrereqError("Tag(s) %s not found" %
4531 (",".join(diff_names)))
4533 def Exec(self, feedback_fn):
4534 """Remove the tag from the object.
4537 for tag in self.op.tags:
4538 self.target.RemoveTag(tag)
4540 self.cfg.Update(self.target)
4541 except errors.ConfigurationError:
4542 raise errors.OpRetryError("There has been a modification to the"
4543 " config file and the operation has been"
4544 " aborted. Please retry.")
4546 class LUTestDelay(NoHooksLU):
4547 """Sleep for a specified amount of time.
4549 This LU sleeps on the master and/or nodes for a specified amoutn of
4553 _OP_REQP = ["duration", "on_master", "on_nodes"]
4555 def CheckPrereq(self):
4556 """Check prerequisites.
4558 This checks that we have a good list of nodes and/or the duration
4563 if self.op.on_nodes:
4564 self.op.on_nodes = _GetWantedNodes(self, self.op.on_nodes)
4566 def Exec(self, feedback_fn):
4567 """Do the actual sleep.
4570 if self.op.on_master:
4571 if not utils.TestDelay(self.op.duration):
4572 raise errors.OpExecError("Error during master delay test")
4573 if self.op.on_nodes:
4574 result = rpc.call_test_delay(self.op.on_nodes, self.op.duration)
4576 raise errors.OpExecError("Complete failure from rpc call")
4577 for node, node_result in result.items():
4579 raise errors.OpExecError("Failure during rpc call to node %s,"
4580 " result: %s" % (node, node_result))