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
72 self.processor = processor
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 _GetWantedNodes(lu, nodes):
167 """Returns list of checked and expanded node names.
170 nodes: List of nodes (strings) or None for all
173 if not isinstance(nodes, list):
174 raise errors.OpPrereqError("Invalid argument type 'nodes'")
180 node = lu.cfg.ExpandNodeName(name)
182 raise errors.OpPrereqError("No such node name '%s'" % name)
186 wanted = lu.cfg.GetNodeList()
187 return utils.NiceSort(wanted)
190 def _GetWantedInstances(lu, instances):
191 """Returns list of checked and expanded instance names.
194 instances: List of instances (strings) or None for all
197 if not isinstance(instances, list):
198 raise errors.OpPrereqError("Invalid argument type 'instances'")
203 for name in instances:
204 instance = lu.cfg.ExpandInstanceName(name)
206 raise errors.OpPrereqError("No such instance name '%s'" % name)
207 wanted.append(instance)
210 wanted = lu.cfg.GetInstanceList()
211 return utils.NiceSort(wanted)
214 def _CheckOutputFields(static, dynamic, selected):
215 """Checks whether all selected fields are valid.
218 static: Static fields
219 dynamic: Dynamic fields
222 static_fields = frozenset(static)
223 dynamic_fields = frozenset(dynamic)
225 all_fields = static_fields | dynamic_fields
227 if not all_fields.issuperset(selected):
228 raise errors.OpPrereqError("Unknown output fields selected: %s"
229 % ",".join(frozenset(selected).
230 difference(all_fields)))
233 def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
234 memory, vcpus, nics):
235 """Builds instance related env variables for hooks from single variables.
238 secondary_nodes: List of secondary nodes as strings
242 "INSTANCE_NAME": name,
243 "INSTANCE_PRIMARY": primary_node,
244 "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
245 "INSTANCE_OS_TYPE": os_type,
246 "INSTANCE_STATUS": status,
247 "INSTANCE_MEMORY": memory,
248 "INSTANCE_VCPUS": vcpus,
252 nic_count = len(nics)
253 for idx, (ip, bridge) in enumerate(nics):
256 env["INSTANCE_NIC%d_IP" % idx] = ip
257 env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
261 env["INSTANCE_NIC_COUNT"] = nic_count
266 def _BuildInstanceHookEnvByObject(instance, override=None):
267 """Builds instance related env variables for hooks from an object.
270 instance: objects.Instance object of instance
271 override: dict of values to override
274 'name': instance.name,
275 'primary_node': instance.primary_node,
276 'secondary_nodes': instance.secondary_nodes,
277 'os_type': instance.os,
278 'status': instance.os,
279 'memory': instance.memory,
280 'vcpus': instance.vcpus,
281 'nics': [(nic.ip, nic.bridge) for nic in instance.nics],
284 args.update(override)
285 return _BuildInstanceHookEnv(**args)
288 def _UpdateEtcHosts(fullnode, ip):
289 """Ensure a node has a correct entry in /etc/hosts.
292 fullnode - Fully qualified domain name of host. (str)
293 ip - IPv4 address of host (str)
296 node = fullnode.split(".", 1)[0]
298 f = open('/etc/hosts', 'r+')
307 rawline = f.readline()
313 line = rawline.split('\n')[0]
316 line = line.split('#')[0]
319 # Entire line was comment, skip
320 save_lines.append(rawline)
323 fields = line.split()
327 for spec in [ ip, fullnode, node ]:
328 if spec not in fields:
335 save_lines.append(rawline)
338 if havesome and not haveall:
339 # Line (old, or manual?) which is missing some. Remove.
343 save_lines.append(rawline)
346 add_lines.append('%s\t%s %s\n' % (ip, fullnode, node))
350 save_lines = save_lines + add_lines
352 # We removed a line, write a new file and replace old.
353 fd, tmpname = tempfile.mkstemp('tmp', 'hosts_', '/etc')
354 newfile = os.fdopen(fd, 'w')
355 newfile.write(''.join(save_lines))
357 os.rename(tmpname, '/etc/hosts')
360 # Simply appending a new line will do the trick.
362 for add in add_lines:
368 def _UpdateKnownHosts(fullnode, ip, pubkey):
369 """Ensure a node has a correct known_hosts entry.
372 fullnode - Fully qualified domain name of host. (str)
373 ip - IPv4 address of host (str)
374 pubkey - the public key of the cluster
377 if os.path.exists(constants.SSH_KNOWN_HOSTS_FILE):
378 f = open(constants.SSH_KNOWN_HOSTS_FILE, 'r+')
380 f = open(constants.SSH_KNOWN_HOSTS_FILE, 'w+')
389 logger.Debug('read %s' % (repr(rawline),))
391 parts = rawline.rstrip('\r\n').split()
393 # Ignore unwanted lines
394 if len(parts) >= 3 and not rawline.lstrip()[0] == '#':
395 fields = parts[0].split(',')
400 for spec in [ ip, fullnode ]:
401 if spec not in fields:
406 logger.Debug("key, pubkey = %s." % (repr((key, pubkey)),))
407 if haveall and key == pubkey:
409 save_lines.append(rawline)
410 logger.Debug("Keeping known_hosts '%s'." % (repr(rawline),))
413 if havesome and (not haveall or key != pubkey):
415 logger.Debug("Discarding known_hosts '%s'." % (repr(rawline),))
418 save_lines.append(rawline)
421 add_lines.append('%s,%s ssh-rsa %s\n' % (fullnode, ip, pubkey))
422 logger.Debug("Adding known_hosts '%s'." % (repr(add_lines[-1]),))
425 save_lines = save_lines + add_lines
427 # Write a new file and replace old.
428 fd, tmpname = tempfile.mkstemp('.tmp', 'known_hosts.',
430 newfile = os.fdopen(fd, 'w')
432 newfile.write(''.join(save_lines))
435 logger.Debug("Wrote new known_hosts.")
436 os.rename(tmpname, constants.SSH_KNOWN_HOSTS_FILE)
439 # Simply appending a new line will do the trick.
441 for add in add_lines:
447 def _HasValidVG(vglist, vgname):
448 """Checks if the volume group list is valid.
450 A non-None return value means there's an error, and the return value
451 is the error message.
454 vgsize = vglist.get(vgname, None)
456 return "volume group '%s' missing" % vgname
458 return ("volume group '%s' too small (20480MiB required, %dMib found)" %
463 def _InitSSHSetup(node):
464 """Setup the SSH configuration for the cluster.
467 This generates a dsa keypair for root, adds the pub key to the
468 permitted hosts and adds the hostkey to its own known hosts.
471 node: the name of this host as a fqdn
474 priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
476 for name in priv_key, pub_key:
477 if os.path.exists(name):
478 utils.CreateBackup(name)
479 utils.RemoveFile(name)
481 result = utils.RunCmd(["ssh-keygen", "-t", "dsa",
485 raise errors.OpExecError("Could not generate ssh keypair, error %s" %
488 f = open(pub_key, 'r')
490 utils.AddAuthorizedKey(auth_keys, f.read(8192))
495 def _InitGanetiServerSetup(ss):
496 """Setup the necessary configuration for the initial node daemon.
498 This creates the nodepass file containing the shared password for
499 the cluster and also generates the SSL certificate.
502 # Create pseudo random password
503 randpass = sha.new(os.urandom(64)).hexdigest()
504 # and write it into sstore
505 ss.SetKey(ss.SS_NODED_PASS, randpass)
507 result = utils.RunCmd(["openssl", "req", "-new", "-newkey", "rsa:1024",
508 "-days", str(365*5), "-nodes", "-x509",
509 "-keyout", constants.SSL_CERT_FILE,
510 "-out", constants.SSL_CERT_FILE, "-batch"])
512 raise errors.OpExecError("could not generate server ssl cert, command"
513 " %s had exitcode %s and error message %s" %
514 (result.cmd, result.exit_code, result.output))
516 os.chmod(constants.SSL_CERT_FILE, 0400)
518 result = utils.RunCmd([constants.NODE_INITD_SCRIPT, "restart"])
521 raise errors.OpExecError("Could not start the node daemon, command %s"
522 " had exitcode %s and error %s" %
523 (result.cmd, result.exit_code, result.output))
526 def _CheckInstanceBridgesExist(instance):
527 """Check that the brigdes needed by an instance exist.
530 # check bridges existance
531 brlist = [nic.bridge for nic in instance.nics]
532 if not rpc.call_bridges_exist(instance.primary_node, brlist):
533 raise errors.OpPrereqError("one or more target bridges %s does not"
534 " exist on destination node '%s'" %
535 (brlist, instance.primary_node))
538 class LUInitCluster(LogicalUnit):
539 """Initialise the cluster.
542 HPATH = "cluster-init"
543 HTYPE = constants.HTYPE_CLUSTER
544 _OP_REQP = ["cluster_name", "hypervisor_type", "vg_name", "mac_prefix",
545 "def_bridge", "master_netdev"]
548 def BuildHooksEnv(self):
551 Notes: Since we don't require a cluster, we must manually add
552 ourselves in the post-run node list.
555 env = {"OP_TARGET": self.op.cluster_name}
556 return env, [], [self.hostname.name]
558 def CheckPrereq(self):
559 """Verify that the passed name is a valid one.
562 if config.ConfigWriter.IsCluster():
563 raise errors.OpPrereqError("Cluster is already initialised")
565 self.hostname = hostname = utils.HostInfo()
567 if hostname.ip.startswith("127."):
568 raise errors.OpPrereqError("This host's IP resolves to the private"
569 " range (%s). Please fix DNS or /etc/hosts." %
572 self.clustername = clustername = utils.HostInfo(self.op.cluster_name)
574 if not utils.TcpPing(constants.LOCALHOST_IP_ADDRESS, hostname.ip,
575 constants.DEFAULT_NODED_PORT):
576 raise errors.OpPrereqError("Inconsistency: this host's name resolves"
577 " to %s,\nbut this ip address does not"
578 " belong to this host."
579 " Aborting." % hostname.ip)
581 secondary_ip = getattr(self.op, "secondary_ip", None)
582 if secondary_ip and not utils.IsValidIP(secondary_ip):
583 raise errors.OpPrereqError("Invalid secondary ip given")
585 secondary_ip != hostname.ip and
586 (not utils.TcpPing(constants.LOCALHOST_IP_ADDRESS, secondary_ip,
587 constants.DEFAULT_NODED_PORT))):
588 raise errors.OpPrereqError("You gave %s as secondary IP,\n"
589 "but it does not belong to this host." %
591 self.secondary_ip = secondary_ip
593 # checks presence of the volume group given
594 vgstatus = _HasValidVG(utils.ListVolumeGroups(), self.op.vg_name)
597 raise errors.OpPrereqError("Error: %s" % vgstatus)
599 if not re.match("^[0-9a-z]{2}:[0-9a-z]{2}:[0-9a-z]{2}$",
601 raise errors.OpPrereqError("Invalid mac prefix given '%s'" %
604 if self.op.hypervisor_type not in hypervisor.VALID_HTYPES:
605 raise errors.OpPrereqError("Invalid hypervisor type given '%s'" %
606 self.op.hypervisor_type)
608 result = utils.RunCmd(["ip", "link", "show", "dev", self.op.master_netdev])
610 raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
611 (self.op.master_netdev,
612 result.output.strip()))
614 if not os.path.exists(constants.NODE_INITD_SCRIPT):
615 raise errors.OpPrereqError("Missing init.d script '%s'. Please reinstall"
616 " or install the script manually." %
617 constants.NODE_INITD_SCRIPT)
619 def Exec(self, feedback_fn):
620 """Initialize the cluster.
623 clustername = self.clustername
624 hostname = self.hostname
626 # set up the simple store
627 self.sstore = ss = ssconf.SimpleStore()
628 ss.SetKey(ss.SS_HYPERVISOR, self.op.hypervisor_type)
629 ss.SetKey(ss.SS_MASTER_NODE, hostname.name)
630 ss.SetKey(ss.SS_MASTER_IP, clustername.ip)
631 ss.SetKey(ss.SS_MASTER_NETDEV, self.op.master_netdev)
632 ss.SetKey(ss.SS_CLUSTER_NAME, clustername.name)
634 # set up the inter-node password and certificate
635 _InitGanetiServerSetup(ss)
637 # start the master ip
638 rpc.call_node_start_master(hostname.name)
640 # set up ssh config and /etc/hosts
641 f = open(constants.SSH_HOST_RSA_PUB, 'r')
646 sshkey = sshline.split(" ")[1]
648 _UpdateEtcHosts(hostname.name, hostname.ip)
650 _UpdateKnownHosts(hostname.name, hostname.ip, sshkey)
652 _InitSSHSetup(hostname.name)
654 # init of cluster config file
655 self.cfg = cfgw = config.ConfigWriter()
656 cfgw.InitConfig(hostname.name, hostname.ip, self.secondary_ip,
657 sshkey, self.op.mac_prefix,
658 self.op.vg_name, self.op.def_bridge)
661 class LUDestroyCluster(NoHooksLU):
662 """Logical unit for destroying the cluster.
667 def CheckPrereq(self):
668 """Check prerequisites.
670 This checks whether the cluster is empty.
672 Any errors are signalled by raising errors.OpPrereqError.
675 master = self.sstore.GetMasterNode()
677 nodelist = self.cfg.GetNodeList()
678 if len(nodelist) != 1 or nodelist[0] != master:
679 raise errors.OpPrereqError("There are still %d node(s) in"
680 " this cluster." % (len(nodelist) - 1))
681 instancelist = self.cfg.GetInstanceList()
683 raise errors.OpPrereqError("There are still %d instance(s) in"
684 " this cluster." % len(instancelist))
686 def Exec(self, feedback_fn):
687 """Destroys the cluster.
690 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
691 utils.CreateBackup(priv_key)
692 utils.CreateBackup(pub_key)
693 rpc.call_node_leave_cluster(self.sstore.GetMasterNode())
696 class LUVerifyCluster(NoHooksLU):
697 """Verifies the cluster status.
702 def _VerifyNode(self, node, file_list, local_cksum, vglist, node_result,
703 remote_version, feedback_fn):
704 """Run multiple tests against a node.
707 - compares ganeti version
708 - checks vg existance and size > 20G
709 - checks config file checksum
710 - checks ssh to other nodes
713 node: name of the node to check
714 file_list: required list of files
715 local_cksum: dictionary of local files and their checksums
718 # compares ganeti version
719 local_version = constants.PROTOCOL_VERSION
720 if not remote_version:
721 feedback_fn(" - ERROR: connection to %s failed" % (node))
724 if local_version != remote_version:
725 feedback_fn(" - ERROR: sw version mismatch: master %s, node(%s) %s" %
726 (local_version, node, remote_version))
729 # checks vg existance and size > 20G
733 feedback_fn(" - ERROR: unable to check volume groups on node %s." %
737 vgstatus = _HasValidVG(vglist, self.cfg.GetVGName())
739 feedback_fn(" - ERROR: %s on node %s" % (vgstatus, node))
742 # checks config file checksum
745 if 'filelist' not in node_result:
747 feedback_fn(" - ERROR: node hasn't returned file checksum data")
749 remote_cksum = node_result['filelist']
750 for file_name in file_list:
751 if file_name not in remote_cksum:
753 feedback_fn(" - ERROR: file '%s' missing" % file_name)
754 elif remote_cksum[file_name] != local_cksum[file_name]:
756 feedback_fn(" - ERROR: file '%s' has wrong checksum" % file_name)
758 if 'nodelist' not in node_result:
760 feedback_fn(" - ERROR: node hasn't returned node connectivity data")
762 if node_result['nodelist']:
764 for node in node_result['nodelist']:
765 feedback_fn(" - ERROR: communication with node '%s': %s" %
766 (node, node_result['nodelist'][node]))
767 hyp_result = node_result.get('hypervisor', None)
768 if hyp_result is not None:
769 feedback_fn(" - ERROR: hypervisor verify failure: '%s'" % hyp_result)
772 def _VerifyInstance(self, instance, node_vol_is, node_instance, feedback_fn):
773 """Verify an instance.
775 This function checks to see if the required block devices are
776 available on the instance's node.
781 instancelist = self.cfg.GetInstanceList()
782 if not instance in instancelist:
783 feedback_fn(" - ERROR: instance %s not in instance list %s" %
784 (instance, instancelist))
787 instanceconfig = self.cfg.GetInstanceInfo(instance)
788 node_current = instanceconfig.primary_node
791 instanceconfig.MapLVsByNode(node_vol_should)
793 for node in node_vol_should:
794 for volume in node_vol_should[node]:
795 if node not in node_vol_is or volume not in node_vol_is[node]:
796 feedback_fn(" - ERROR: volume %s missing on node %s" %
800 if not instanceconfig.status == 'down':
801 if not instance in node_instance[node_current]:
802 feedback_fn(" - ERROR: instance %s not running on node %s" %
803 (instance, node_current))
806 for node in node_instance:
807 if (not node == node_current):
808 if instance in node_instance[node]:
809 feedback_fn(" - ERROR: instance %s should not run on node %s" %
815 def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
816 """Verify if there are any unknown volumes in the cluster.
818 The .os, .swap and backup volumes are ignored. All other volumes are
824 for node in node_vol_is:
825 for volume in node_vol_is[node]:
826 if node not in node_vol_should or volume not in node_vol_should[node]:
827 feedback_fn(" - ERROR: volume %s on node %s should not exist" %
832 def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
833 """Verify the list of running instances.
835 This checks what instances are running but unknown to the cluster.
839 for node in node_instance:
840 for runninginstance in node_instance[node]:
841 if runninginstance not in instancelist:
842 feedback_fn(" - ERROR: instance %s on node %s should not exist" %
843 (runninginstance, node))
847 def CheckPrereq(self):
848 """Check prerequisites.
850 This has no prerequisites.
855 def Exec(self, feedback_fn):
856 """Verify integrity of cluster, performing various test on nodes.
860 feedback_fn("* Verifying global settings")
861 self.cfg.VerifyConfig()
863 vg_name = self.cfg.GetVGName()
864 nodelist = utils.NiceSort(self.cfg.GetNodeList())
865 instancelist = utils.NiceSort(self.cfg.GetInstanceList())
869 # FIXME: verify OS list
871 file_names = list(self.sstore.GetFileList())
872 file_names.append(constants.SSL_CERT_FILE)
873 file_names.append(constants.CLUSTER_CONF_FILE)
874 local_checksums = utils.FingerprintFiles(file_names)
876 feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
877 all_volumeinfo = rpc.call_volume_list(nodelist, vg_name)
878 all_instanceinfo = rpc.call_instance_list(nodelist)
879 all_vglist = rpc.call_vg_list(nodelist)
880 node_verify_param = {
881 'filelist': file_names,
882 'nodelist': nodelist,
885 all_nvinfo = rpc.call_node_verify(nodelist, node_verify_param)
886 all_rversion = rpc.call_version(nodelist)
888 for node in nodelist:
889 feedback_fn("* Verifying node %s" % node)
890 result = self._VerifyNode(node, file_names, local_checksums,
891 all_vglist[node], all_nvinfo[node],
892 all_rversion[node], feedback_fn)
896 volumeinfo = all_volumeinfo[node]
898 if type(volumeinfo) != dict:
899 feedback_fn(" - ERROR: connection to %s failed" % (node,))
903 node_volume[node] = volumeinfo
906 nodeinstance = all_instanceinfo[node]
907 if type(nodeinstance) != list:
908 feedback_fn(" - ERROR: connection to %s failed" % (node,))
912 node_instance[node] = nodeinstance
916 for instance in instancelist:
917 feedback_fn("* Verifying instance %s" % instance)
918 result = self._VerifyInstance(instance, node_volume, node_instance,
922 inst_config = self.cfg.GetInstanceInfo(instance)
924 inst_config.MapLVsByNode(node_vol_should)
926 feedback_fn("* Verifying orphan volumes")
927 result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
931 feedback_fn("* Verifying remaining instances")
932 result = self._VerifyOrphanInstances(instancelist, node_instance,
939 class LURenameCluster(LogicalUnit):
940 """Rename the cluster.
943 HPATH = "cluster-rename"
944 HTYPE = constants.HTYPE_CLUSTER
947 def BuildHooksEnv(self):
952 "OP_TARGET": self.op.sstore.GetClusterName(),
953 "NEW_NAME": self.op.name,
955 mn = self.sstore.GetMasterNode()
956 return env, [mn], [mn]
958 def CheckPrereq(self):
959 """Verify that the passed name is a valid one.
962 hostname = utils.HostInfo(self.op.name)
964 new_name = hostname.name
965 self.ip = new_ip = hostname.ip
966 old_name = self.sstore.GetClusterName()
967 old_ip = self.sstore.GetMasterIP()
968 if new_name == old_name and new_ip == old_ip:
969 raise errors.OpPrereqError("Neither the name nor the IP address of the"
970 " cluster has changed")
972 result = utils.RunCmd(["fping", "-q", new_ip])
973 if not result.failed:
974 raise errors.OpPrereqError("The given cluster IP address (%s) is"
975 " reachable on the network. Aborting." %
978 self.op.name = new_name
980 def Exec(self, feedback_fn):
981 """Rename the cluster.
984 clustername = self.op.name
988 # shutdown the master IP
989 master = ss.GetMasterNode()
990 if not rpc.call_node_stop_master(master):
991 raise errors.OpExecError("Could not disable the master role")
995 ss.SetKey(ss.SS_MASTER_IP, ip)
996 ss.SetKey(ss.SS_CLUSTER_NAME, clustername)
998 # Distribute updated ss config to all nodes
999 myself = self.cfg.GetNodeInfo(master)
1000 dist_nodes = self.cfg.GetNodeList()
1001 if myself.name in dist_nodes:
1002 dist_nodes.remove(myself.name)
1004 logger.Debug("Copying updated ssconf data to all nodes")
1005 for keyname in [ss.SS_CLUSTER_NAME, ss.SS_MASTER_IP]:
1006 fname = ss.KeyToFilename(keyname)
1007 result = rpc.call_upload_file(dist_nodes, fname)
1008 for to_node in dist_nodes:
1009 if not result[to_node]:
1010 logger.Error("copy of file %s to node %s failed" %
1013 if not rpc.call_node_start_master(master):
1014 logger.Error("Could not re-enable the master role on the master,\n"
1015 "please restart manually.")
1018 def _WaitForSync(cfgw, instance, oneshot=False, unlock=False):
1019 """Sleep and poll for an instance's disk to sync.
1022 if not instance.disks:
1026 logger.ToStdout("Waiting for instance %s to sync disks." % instance.name)
1028 node = instance.primary_node
1030 for dev in instance.disks:
1031 cfgw.SetDiskID(dev, node)
1037 cumul_degraded = False
1038 rstats = rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1040 logger.ToStderr("Can't get any data from node %s" % node)
1043 raise errors.RemoteError("Can't contact node %s for mirror data,"
1044 " aborting." % node)
1048 for i in range(len(rstats)):
1051 logger.ToStderr("Can't compute data for node %s/%s" %
1052 (node, instance.disks[i].iv_name))
1054 perc_done, est_time, is_degraded = mstat
1055 cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1056 if perc_done is not None:
1058 if est_time is not None:
1059 rem_time = "%d estimated seconds remaining" % est_time
1062 rem_time = "no time estimate"
1063 logger.ToStdout("- device %s: %5.2f%% done, %s" %
1064 (instance.disks[i].iv_name, perc_done, rem_time))
1071 time.sleep(min(60, max_time))
1077 logger.ToStdout("Instance %s's disks are in sync." % instance.name)
1078 return not cumul_degraded
1081 def _CheckDiskConsistency(cfgw, dev, node, on_primary):
1082 """Check that mirrors are not degraded.
1085 cfgw.SetDiskID(dev, node)
1088 if on_primary or dev.AssembleOnSecondary():
1089 rstats = rpc.call_blockdev_find(node, dev)
1091 logger.ToStderr("Can't get any data from node %s" % node)
1094 result = result and (not rstats[5])
1096 for child in dev.children:
1097 result = result and _CheckDiskConsistency(cfgw, child, node, on_primary)
1102 class LUDiagnoseOS(NoHooksLU):
1103 """Logical unit for OS diagnose/query.
1108 def CheckPrereq(self):
1109 """Check prerequisites.
1111 This always succeeds, since this is a pure query LU.
1116 def Exec(self, feedback_fn):
1117 """Compute the list of OSes.
1120 node_list = self.cfg.GetNodeList()
1121 node_data = rpc.call_os_diagnose(node_list)
1122 if node_data == False:
1123 raise errors.OpExecError("Can't gather the list of OSes")
1127 class LURemoveNode(LogicalUnit):
1128 """Logical unit for removing a node.
1131 HPATH = "node-remove"
1132 HTYPE = constants.HTYPE_NODE
1133 _OP_REQP = ["node_name"]
1135 def BuildHooksEnv(self):
1138 This doesn't run on the target node in the pre phase as a failed
1139 node would not allows itself to run.
1143 "OP_TARGET": self.op.node_name,
1144 "NODE_NAME": self.op.node_name,
1146 all_nodes = self.cfg.GetNodeList()
1147 all_nodes.remove(self.op.node_name)
1148 return env, all_nodes, all_nodes
1150 def CheckPrereq(self):
1151 """Check prerequisites.
1154 - the node exists in the configuration
1155 - it does not have primary or secondary instances
1156 - it's not the master
1158 Any errors are signalled by raising errors.OpPrereqError.
1161 node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1163 raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1165 instance_list = self.cfg.GetInstanceList()
1167 masternode = self.sstore.GetMasterNode()
1168 if node.name == masternode:
1169 raise errors.OpPrereqError("Node is the master node,"
1170 " you need to failover first.")
1172 for instance_name in instance_list:
1173 instance = self.cfg.GetInstanceInfo(instance_name)
1174 if node.name == instance.primary_node:
1175 raise errors.OpPrereqError("Instance %s still running on the node,"
1176 " please remove first." % instance_name)
1177 if node.name in instance.secondary_nodes:
1178 raise errors.OpPrereqError("Instance %s has node as a secondary,"
1179 " please remove first." % instance_name)
1180 self.op.node_name = node.name
1183 def Exec(self, feedback_fn):
1184 """Removes the node from the cluster.
1188 logger.Info("stopping the node daemon and removing configs from node %s" %
1191 rpc.call_node_leave_cluster(node.name)
1193 ssh.SSHCall(node.name, 'root', "%s stop" % constants.NODE_INITD_SCRIPT)
1195 logger.Info("Removing node %s from config" % node.name)
1197 self.cfg.RemoveNode(node.name)
1200 class LUQueryNodes(NoHooksLU):
1201 """Logical unit for querying nodes.
1204 _OP_REQP = ["output_fields", "names"]
1206 def CheckPrereq(self):
1207 """Check prerequisites.
1209 This checks that the fields required are valid output fields.
1212 self.dynamic_fields = frozenset(["dtotal", "dfree",
1213 "mtotal", "mnode", "mfree",
1216 _CheckOutputFields(static=["name", "pinst_cnt", "sinst_cnt",
1217 "pinst_list", "sinst_list",
1219 dynamic=self.dynamic_fields,
1220 selected=self.op.output_fields)
1222 self.wanted = _GetWantedNodes(self, self.op.names)
1224 def Exec(self, feedback_fn):
1225 """Computes the list of nodes and their attributes.
1228 nodenames = self.wanted
1229 nodelist = [self.cfg.GetNodeInfo(name) for name in nodenames]
1231 # begin data gathering
1233 if self.dynamic_fields.intersection(self.op.output_fields):
1235 node_data = rpc.call_node_info(nodenames, self.cfg.GetVGName())
1236 for name in nodenames:
1237 nodeinfo = node_data.get(name, None)
1240 "mtotal": utils.TryConvert(int, nodeinfo['memory_total']),
1241 "mnode": utils.TryConvert(int, nodeinfo['memory_dom0']),
1242 "mfree": utils.TryConvert(int, nodeinfo['memory_free']),
1243 "dtotal": utils.TryConvert(int, nodeinfo['vg_size']),
1244 "dfree": utils.TryConvert(int, nodeinfo['vg_free']),
1245 "bootid": nodeinfo['bootid'],
1248 live_data[name] = {}
1250 live_data = dict.fromkeys(nodenames, {})
1252 node_to_primary = dict([(name, set()) for name in nodenames])
1253 node_to_secondary = dict([(name, set()) for name in nodenames])
1255 inst_fields = frozenset(("pinst_cnt", "pinst_list",
1256 "sinst_cnt", "sinst_list"))
1257 if inst_fields & frozenset(self.op.output_fields):
1258 instancelist = self.cfg.GetInstanceList()
1260 for instance_name in instancelist:
1261 inst = self.cfg.GetInstanceInfo(instance_name)
1262 if inst.primary_node in node_to_primary:
1263 node_to_primary[inst.primary_node].add(inst.name)
1264 for secnode in inst.secondary_nodes:
1265 if secnode in node_to_secondary:
1266 node_to_secondary[secnode].add(inst.name)
1268 # end data gathering
1271 for node in nodelist:
1273 for field in self.op.output_fields:
1276 elif field == "pinst_list":
1277 val = list(node_to_primary[node.name])
1278 elif field == "sinst_list":
1279 val = list(node_to_secondary[node.name])
1280 elif field == "pinst_cnt":
1281 val = len(node_to_primary[node.name])
1282 elif field == "sinst_cnt":
1283 val = len(node_to_secondary[node.name])
1284 elif field == "pip":
1285 val = node.primary_ip
1286 elif field == "sip":
1287 val = node.secondary_ip
1288 elif field in self.dynamic_fields:
1289 val = live_data[node.name].get(field, None)
1291 raise errors.ParameterError(field)
1292 node_output.append(val)
1293 output.append(node_output)
1298 class LUQueryNodeVolumes(NoHooksLU):
1299 """Logical unit for getting volumes on node(s).
1302 _OP_REQP = ["nodes", "output_fields"]
1304 def CheckPrereq(self):
1305 """Check prerequisites.
1307 This checks that the fields required are valid output fields.
1310 self.nodes = _GetWantedNodes(self, self.op.nodes)
1312 _CheckOutputFields(static=["node"],
1313 dynamic=["phys", "vg", "name", "size", "instance"],
1314 selected=self.op.output_fields)
1317 def Exec(self, feedback_fn):
1318 """Computes the list of nodes and their attributes.
1321 nodenames = self.nodes
1322 volumes = rpc.call_node_volumes(nodenames)
1324 ilist = [self.cfg.GetInstanceInfo(iname) for iname
1325 in self.cfg.GetInstanceList()]
1327 lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1330 for node in nodenames:
1331 if node not in volumes or not volumes[node]:
1334 node_vols = volumes[node][:]
1335 node_vols.sort(key=lambda vol: vol['dev'])
1337 for vol in node_vols:
1339 for field in self.op.output_fields:
1342 elif field == "phys":
1346 elif field == "name":
1348 elif field == "size":
1349 val = int(float(vol['size']))
1350 elif field == "instance":
1352 if node not in lv_by_node[inst]:
1354 if vol['name'] in lv_by_node[inst][node]:
1360 raise errors.ParameterError(field)
1361 node_output.append(str(val))
1363 output.append(node_output)
1368 class LUAddNode(LogicalUnit):
1369 """Logical unit for adding node to the cluster.
1373 HTYPE = constants.HTYPE_NODE
1374 _OP_REQP = ["node_name"]
1376 def BuildHooksEnv(self):
1379 This will run on all nodes before, and on all nodes + the new node after.
1383 "OP_TARGET": self.op.node_name,
1384 "NODE_NAME": self.op.node_name,
1385 "NODE_PIP": self.op.primary_ip,
1386 "NODE_SIP": self.op.secondary_ip,
1388 nodes_0 = self.cfg.GetNodeList()
1389 nodes_1 = nodes_0 + [self.op.node_name, ]
1390 return env, nodes_0, nodes_1
1392 def CheckPrereq(self):
1393 """Check prerequisites.
1396 - the new node is not already in the config
1398 - its parameters (single/dual homed) matches the cluster
1400 Any errors are signalled by raising errors.OpPrereqError.
1403 node_name = self.op.node_name
1406 dns_data = utils.HostInfo(node_name)
1408 node = dns_data.name
1409 primary_ip = self.op.primary_ip = dns_data.ip
1410 secondary_ip = getattr(self.op, "secondary_ip", None)
1411 if secondary_ip is None:
1412 secondary_ip = primary_ip
1413 if not utils.IsValidIP(secondary_ip):
1414 raise errors.OpPrereqError("Invalid secondary IP given")
1415 self.op.secondary_ip = secondary_ip
1416 node_list = cfg.GetNodeList()
1417 if node in node_list:
1418 raise errors.OpPrereqError("Node %s is already in the configuration"
1421 for existing_node_name in node_list:
1422 existing_node = cfg.GetNodeInfo(existing_node_name)
1423 if (existing_node.primary_ip == primary_ip or
1424 existing_node.secondary_ip == primary_ip or
1425 existing_node.primary_ip == secondary_ip or
1426 existing_node.secondary_ip == secondary_ip):
1427 raise errors.OpPrereqError("New node ip address(es) conflict with"
1428 " existing node %s" % existing_node.name)
1430 # check that the type of the node (single versus dual homed) is the
1431 # same as for the master
1432 myself = cfg.GetNodeInfo(self.sstore.GetMasterNode())
1433 master_singlehomed = myself.secondary_ip == myself.primary_ip
1434 newbie_singlehomed = secondary_ip == primary_ip
1435 if master_singlehomed != newbie_singlehomed:
1436 if master_singlehomed:
1437 raise errors.OpPrereqError("The master has no private ip but the"
1438 " new node has one")
1440 raise errors.OpPrereqError("The master has a private ip but the"
1441 " new node doesn't have one")
1443 # checks reachablity
1444 if not utils.TcpPing(utils.HostInfo().name,
1446 constants.DEFAULT_NODED_PORT):
1447 raise errors.OpPrereqError("Node not reachable by ping")
1449 if not newbie_singlehomed:
1450 # check reachability from my secondary ip to newbie's secondary ip
1451 if not utils.TcpPing(myself.secondary_ip,
1453 constants.DEFAULT_NODED_PORT):
1454 raise errors.OpPrereqError(
1455 "Node secondary ip not reachable by TCP based ping to noded port")
1457 self.new_node = objects.Node(name=node,
1458 primary_ip=primary_ip,
1459 secondary_ip=secondary_ip)
1461 def Exec(self, feedback_fn):
1462 """Adds the new node to the cluster.
1465 new_node = self.new_node
1466 node = new_node.name
1468 # set up inter-node password and certificate and restarts the node daemon
1469 gntpass = self.sstore.GetNodeDaemonPassword()
1470 if not re.match('^[a-zA-Z0-9.]{1,64}$', gntpass):
1471 raise errors.OpExecError("ganeti password corruption detected")
1472 f = open(constants.SSL_CERT_FILE)
1474 gntpem = f.read(8192)
1477 # in the base64 pem encoding, neither '!' nor '.' are valid chars,
1478 # so we use this to detect an invalid certificate; as long as the
1479 # cert doesn't contain this, the here-document will be correctly
1480 # parsed by the shell sequence below
1481 if re.search('^!EOF\.', gntpem, re.MULTILINE):
1482 raise errors.OpExecError("invalid PEM encoding in the SSL certificate")
1483 if not gntpem.endswith("\n"):
1484 raise errors.OpExecError("PEM must end with newline")
1485 logger.Info("copy cluster pass to %s and starting the node daemon" % node)
1487 # and then connect with ssh to set password and start ganeti-noded
1488 # note that all the below variables are sanitized at this point,
1489 # either by being constants or by the checks above
1491 mycommand = ("umask 077 && "
1492 "echo '%s' > '%s' && "
1493 "cat > '%s' << '!EOF.' && \n"
1494 "%s!EOF.\n%s restart" %
1495 (gntpass, ss.KeyToFilename(ss.SS_NODED_PASS),
1496 constants.SSL_CERT_FILE, gntpem,
1497 constants.NODE_INITD_SCRIPT))
1499 result = ssh.SSHCall(node, 'root', mycommand, batch=False, ask_key=True)
1501 raise errors.OpExecError("Remote command on node %s, error: %s,"
1503 (node, result.fail_reason, result.output))
1505 # check connectivity
1508 result = rpc.call_version([node])[node]
1510 if constants.PROTOCOL_VERSION == result:
1511 logger.Info("communication to node %s fine, sw version %s match" %
1514 raise errors.OpExecError("Version mismatch master version %s,"
1515 " node version %s" %
1516 (constants.PROTOCOL_VERSION, result))
1518 raise errors.OpExecError("Cannot get version from the new node")
1521 logger.Info("copy ssh key to node %s" % node)
1522 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
1524 keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
1525 constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
1531 keyarray.append(f.read())
1535 result = rpc.call_node_add(node, keyarray[0], keyarray[1], keyarray[2],
1536 keyarray[3], keyarray[4], keyarray[5])
1539 raise errors.OpExecError("Cannot transfer ssh keys to the new node")
1541 # Add node to our /etc/hosts, and add key to known_hosts
1542 _UpdateEtcHosts(new_node.name, new_node.primary_ip)
1543 _UpdateKnownHosts(new_node.name, new_node.primary_ip,
1544 self.cfg.GetHostKey())
1546 if new_node.secondary_ip != new_node.primary_ip:
1547 if not rpc.call_node_tcp_ping(new_node.name,
1548 constants.LOCALHOST_IP_ADDRESS,
1549 new_node.secondary_ip,
1550 constants.DEFAULT_NODED_PORT,
1552 raise errors.OpExecError("Node claims it doesn't have the"
1553 " secondary ip you gave (%s).\n"
1554 "Please fix and re-run this command." %
1555 new_node.secondary_ip)
1557 success, msg = ssh.VerifyNodeHostname(node)
1559 raise errors.OpExecError("Node '%s' claims it has a different hostname"
1560 " than the one the resolver gives: %s.\n"
1561 "Please fix and re-run this command." %
1564 # Distribute updated /etc/hosts and known_hosts to all nodes,
1565 # including the node just added
1566 myself = self.cfg.GetNodeInfo(self.sstore.GetMasterNode())
1567 dist_nodes = self.cfg.GetNodeList() + [node]
1568 if myself.name in dist_nodes:
1569 dist_nodes.remove(myself.name)
1571 logger.Debug("Copying hosts and known_hosts to all nodes")
1572 for fname in ("/etc/hosts", constants.SSH_KNOWN_HOSTS_FILE):
1573 result = rpc.call_upload_file(dist_nodes, fname)
1574 for to_node in dist_nodes:
1575 if not result[to_node]:
1576 logger.Error("copy of file %s to node %s failed" %
1579 to_copy = ss.GetFileList()
1580 for fname in to_copy:
1581 if not ssh.CopyFileToNode(node, fname):
1582 logger.Error("could not copy file %s to node %s" % (fname, node))
1584 logger.Info("adding node %s to cluster.conf" % node)
1585 self.cfg.AddNode(new_node)
1588 class LUMasterFailover(LogicalUnit):
1589 """Failover the master node to the current node.
1591 This is a special LU in that it must run on a non-master node.
1594 HPATH = "master-failover"
1595 HTYPE = constants.HTYPE_CLUSTER
1599 def BuildHooksEnv(self):
1602 This will run on the new master only in the pre phase, and on all
1603 the nodes in the post phase.
1607 "OP_TARGET": self.new_master,
1608 "NEW_MASTER": self.new_master,
1609 "OLD_MASTER": self.old_master,
1611 return env, [self.new_master], self.cfg.GetNodeList()
1613 def CheckPrereq(self):
1614 """Check prerequisites.
1616 This checks that we are not already the master.
1619 self.new_master = utils.HostInfo().name
1620 self.old_master = self.sstore.GetMasterNode()
1622 if self.old_master == self.new_master:
1623 raise errors.OpPrereqError("This commands must be run on the node"
1624 " where you want the new master to be.\n"
1625 "%s is already the master" %
1628 def Exec(self, feedback_fn):
1629 """Failover the master node.
1631 This command, when run on a non-master node, will cause the current
1632 master to cease being master, and the non-master to become new
1636 #TODO: do not rely on gethostname returning the FQDN
1637 logger.Info("setting master to %s, old master: %s" %
1638 (self.new_master, self.old_master))
1640 if not rpc.call_node_stop_master(self.old_master):
1641 logger.Error("could disable the master role on the old master"
1642 " %s, please disable manually" % self.old_master)
1645 ss.SetKey(ss.SS_MASTER_NODE, self.new_master)
1646 if not rpc.call_upload_file(self.cfg.GetNodeList(),
1647 ss.KeyToFilename(ss.SS_MASTER_NODE)):
1648 logger.Error("could not distribute the new simple store master file"
1649 " to the other nodes, please check.")
1651 if not rpc.call_node_start_master(self.new_master):
1652 logger.Error("could not start the master role on the new master"
1653 " %s, please check" % self.new_master)
1654 feedback_fn("Error in activating the master IP on the new master,\n"
1655 "please fix manually.")
1659 class LUQueryClusterInfo(NoHooksLU):
1660 """Query cluster configuration.
1666 def CheckPrereq(self):
1667 """No prerequsites needed for this LU.
1672 def Exec(self, feedback_fn):
1673 """Return cluster config.
1677 "name": self.sstore.GetClusterName(),
1678 "software_version": constants.RELEASE_VERSION,
1679 "protocol_version": constants.PROTOCOL_VERSION,
1680 "config_version": constants.CONFIG_VERSION,
1681 "os_api_version": constants.OS_API_VERSION,
1682 "export_version": constants.EXPORT_VERSION,
1683 "master": self.sstore.GetMasterNode(),
1684 "architecture": (platform.architecture()[0], platform.machine()),
1690 class LUClusterCopyFile(NoHooksLU):
1691 """Copy file to cluster.
1694 _OP_REQP = ["nodes", "filename"]
1696 def CheckPrereq(self):
1697 """Check prerequisites.
1699 It should check that the named file exists and that the given list
1703 if not os.path.exists(self.op.filename):
1704 raise errors.OpPrereqError("No such filename '%s'" % self.op.filename)
1706 self.nodes = _GetWantedNodes(self, self.op.nodes)
1708 def Exec(self, feedback_fn):
1709 """Copy a file from master to some nodes.
1712 opts - class with options as members
1713 args - list containing a single element, the file name
1715 nodes - list containing the name of target nodes; if empty, all nodes
1718 filename = self.op.filename
1720 myname = utils.HostInfo().name
1722 for node in self.nodes:
1725 if not ssh.CopyFileToNode(node, filename):
1726 logger.Error("Copy of file %s to node %s failed" % (filename, node))
1729 class LUDumpClusterConfig(NoHooksLU):
1730 """Return a text-representation of the cluster-config.
1735 def CheckPrereq(self):
1736 """No prerequisites.
1741 def Exec(self, feedback_fn):
1742 """Dump a representation of the cluster config to the standard output.
1745 return self.cfg.DumpConfig()
1748 class LURunClusterCommand(NoHooksLU):
1749 """Run a command on some nodes.
1752 _OP_REQP = ["command", "nodes"]
1754 def CheckPrereq(self):
1755 """Check prerequisites.
1757 It checks that the given list of nodes is valid.
1760 self.nodes = _GetWantedNodes(self, self.op.nodes)
1762 def Exec(self, feedback_fn):
1763 """Run a command on some nodes.
1767 for node in self.nodes:
1768 result = ssh.SSHCall(node, "root", self.op.command)
1769 data.append((node, result.output, result.exit_code))
1774 class LUActivateInstanceDisks(NoHooksLU):
1775 """Bring up an instance's disks.
1778 _OP_REQP = ["instance_name"]
1780 def CheckPrereq(self):
1781 """Check prerequisites.
1783 This checks that the instance is in the cluster.
1786 instance = self.cfg.GetInstanceInfo(
1787 self.cfg.ExpandInstanceName(self.op.instance_name))
1788 if instance is None:
1789 raise errors.OpPrereqError("Instance '%s' not known" %
1790 self.op.instance_name)
1791 self.instance = instance
1794 def Exec(self, feedback_fn):
1795 """Activate the disks.
1798 disks_ok, disks_info = _AssembleInstanceDisks(self.instance, self.cfg)
1800 raise errors.OpExecError("Cannot activate block devices")
1805 def _AssembleInstanceDisks(instance, cfg, ignore_secondaries=False):
1806 """Prepare the block devices for an instance.
1808 This sets up the block devices on all nodes.
1811 instance: a ganeti.objects.Instance object
1812 ignore_secondaries: if true, errors on secondary nodes won't result
1813 in an error return from the function
1816 false if the operation failed
1817 list of (host, instance_visible_name, node_visible_name) if the operation
1818 suceeded with the mapping from node devices to instance devices
1822 for inst_disk in instance.disks:
1823 master_result = None
1824 for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
1825 cfg.SetDiskID(node_disk, node)
1826 is_primary = node == instance.primary_node
1827 result = rpc.call_blockdev_assemble(node, node_disk,
1828 instance.name, is_primary)
1830 logger.Error("could not prepare block device %s on node %s (is_pri"
1831 "mary=%s)" % (inst_disk.iv_name, node, is_primary))
1832 if is_primary or not ignore_secondaries:
1835 master_result = result
1836 device_info.append((instance.primary_node, inst_disk.iv_name,
1839 # leave the disks configured for the primary node
1840 # this is a workaround that would be fixed better by
1841 # improving the logical/physical id handling
1842 for disk in instance.disks:
1843 cfg.SetDiskID(disk, instance.primary_node)
1845 return disks_ok, device_info
1848 def _StartInstanceDisks(cfg, instance, force):
1849 """Start the disks of an instance.
1852 disks_ok, dummy = _AssembleInstanceDisks(instance, cfg,
1853 ignore_secondaries=force)
1855 _ShutdownInstanceDisks(instance, cfg)
1856 if force is not None and not force:
1857 logger.Error("If the message above refers to a secondary node,"
1858 " you can retry the operation using '--force'.")
1859 raise errors.OpExecError("Disk consistency error")
1862 class LUDeactivateInstanceDisks(NoHooksLU):
1863 """Shutdown an instance's disks.
1866 _OP_REQP = ["instance_name"]
1868 def CheckPrereq(self):
1869 """Check prerequisites.
1871 This checks that the instance is in the cluster.
1874 instance = self.cfg.GetInstanceInfo(
1875 self.cfg.ExpandInstanceName(self.op.instance_name))
1876 if instance is None:
1877 raise errors.OpPrereqError("Instance '%s' not known" %
1878 self.op.instance_name)
1879 self.instance = instance
1881 def Exec(self, feedback_fn):
1882 """Deactivate the disks
1885 instance = self.instance
1886 ins_l = rpc.call_instance_list([instance.primary_node])
1887 ins_l = ins_l[instance.primary_node]
1888 if not type(ins_l) is list:
1889 raise errors.OpExecError("Can't contact node '%s'" %
1890 instance.primary_node)
1892 if self.instance.name in ins_l:
1893 raise errors.OpExecError("Instance is running, can't shutdown"
1896 _ShutdownInstanceDisks(instance, self.cfg)
1899 def _ShutdownInstanceDisks(instance, cfg, ignore_primary=False):
1900 """Shutdown block devices of an instance.
1902 This does the shutdown on all nodes of the instance.
1904 If the ignore_primary is false, errors on the primary node are
1909 for disk in instance.disks:
1910 for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
1911 cfg.SetDiskID(top_disk, node)
1912 if not rpc.call_blockdev_shutdown(node, top_disk):
1913 logger.Error("could not shutdown block device %s on node %s" %
1914 (disk.iv_name, node))
1915 if not ignore_primary or node != instance.primary_node:
1920 class LUStartupInstance(LogicalUnit):
1921 """Starts an instance.
1924 HPATH = "instance-start"
1925 HTYPE = constants.HTYPE_INSTANCE
1926 _OP_REQP = ["instance_name", "force"]
1928 def BuildHooksEnv(self):
1931 This runs on master, primary and secondary nodes of the instance.
1935 "FORCE": self.op.force,
1937 env.update(_BuildInstanceHookEnvByObject(self.instance))
1938 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
1939 list(self.instance.secondary_nodes))
1942 def CheckPrereq(self):
1943 """Check prerequisites.
1945 This checks that the instance is in the cluster.
1948 instance = self.cfg.GetInstanceInfo(
1949 self.cfg.ExpandInstanceName(self.op.instance_name))
1950 if instance is None:
1951 raise errors.OpPrereqError("Instance '%s' not known" %
1952 self.op.instance_name)
1954 # check bridges existance
1955 _CheckInstanceBridgesExist(instance)
1957 self.instance = instance
1958 self.op.instance_name = instance.name
1960 def Exec(self, feedback_fn):
1961 """Start the instance.
1964 instance = self.instance
1965 force = self.op.force
1966 extra_args = getattr(self.op, "extra_args", "")
1968 node_current = instance.primary_node
1970 nodeinfo = rpc.call_node_info([node_current], self.cfg.GetVGName())
1972 raise errors.OpExecError("Could not contact node %s for infos" %
1975 freememory = nodeinfo[node_current]['memory_free']
1976 memory = instance.memory
1977 if memory > freememory:
1978 raise errors.OpExecError("Not enough memory to start instance"
1980 " needed %s MiB, available %s MiB" %
1981 (instance.name, node_current, memory,
1984 _StartInstanceDisks(self.cfg, instance, force)
1986 if not rpc.call_instance_start(node_current, instance, extra_args):
1987 _ShutdownInstanceDisks(instance, self.cfg)
1988 raise errors.OpExecError("Could not start instance")
1990 self.cfg.MarkInstanceUp(instance.name)
1993 class LURebootInstance(LogicalUnit):
1994 """Reboot an instance.
1997 HPATH = "instance-reboot"
1998 HTYPE = constants.HTYPE_INSTANCE
1999 _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2001 def BuildHooksEnv(self):
2004 This runs on master, primary and secondary nodes of the instance.
2008 "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2010 env.update(_BuildInstanceHookEnvByObject(self.instance))
2011 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2012 list(self.instance.secondary_nodes))
2015 def CheckPrereq(self):
2016 """Check prerequisites.
2018 This checks that the instance is in the cluster.
2021 instance = self.cfg.GetInstanceInfo(
2022 self.cfg.ExpandInstanceName(self.op.instance_name))
2023 if instance is None:
2024 raise errors.OpPrereqError("Instance '%s' not known" %
2025 self.op.instance_name)
2027 # check bridges existance
2028 _CheckInstanceBridgesExist(instance)
2030 self.instance = instance
2031 self.op.instance_name = instance.name
2033 def Exec(self, feedback_fn):
2034 """Reboot the instance.
2037 instance = self.instance
2038 ignore_secondaries = self.op.ignore_secondaries
2039 reboot_type = self.op.reboot_type
2040 extra_args = getattr(self.op, "extra_args", "")
2042 node_current = instance.primary_node
2044 if reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2045 constants.INSTANCE_REBOOT_HARD,
2046 constants.INSTANCE_REBOOT_FULL]:
2047 raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2048 (constants.INSTANCE_REBOOT_SOFT,
2049 constants.INSTANCE_REBOOT_HARD,
2050 constants.INSTANCE_REBOOT_FULL))
2052 if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2053 constants.INSTANCE_REBOOT_HARD]:
2054 if not rpc.call_instance_reboot(node_current, instance,
2055 reboot_type, extra_args):
2056 raise errors.OpExecError("Could not reboot instance")
2058 if not rpc.call_instance_shutdown(node_current, instance):
2059 raise errors.OpExecError("could not shutdown instance for full reboot")
2060 _ShutdownInstanceDisks(instance, self.cfg)
2061 _StartInstanceDisks(self.cfg, instance, ignore_secondaries)
2062 if not rpc.call_instance_start(node_current, instance, extra_args):
2063 _ShutdownInstanceDisks(instance, self.cfg)
2064 raise errors.OpExecError("Could not start instance for full reboot")
2066 self.cfg.MarkInstanceUp(instance.name)
2069 class LUShutdownInstance(LogicalUnit):
2070 """Shutdown an instance.
2073 HPATH = "instance-stop"
2074 HTYPE = constants.HTYPE_INSTANCE
2075 _OP_REQP = ["instance_name"]
2077 def BuildHooksEnv(self):
2080 This runs on master, primary and secondary nodes of the instance.
2083 env = _BuildInstanceHookEnvByObject(self.instance)
2084 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2085 list(self.instance.secondary_nodes))
2088 def CheckPrereq(self):
2089 """Check prerequisites.
2091 This checks that the instance is in the cluster.
2094 instance = self.cfg.GetInstanceInfo(
2095 self.cfg.ExpandInstanceName(self.op.instance_name))
2096 if instance is None:
2097 raise errors.OpPrereqError("Instance '%s' not known" %
2098 self.op.instance_name)
2099 self.instance = instance
2101 def Exec(self, feedback_fn):
2102 """Shutdown the instance.
2105 instance = self.instance
2106 node_current = instance.primary_node
2107 if not rpc.call_instance_shutdown(node_current, instance):
2108 logger.Error("could not shutdown instance")
2110 self.cfg.MarkInstanceDown(instance.name)
2111 _ShutdownInstanceDisks(instance, self.cfg)
2114 class LUReinstallInstance(LogicalUnit):
2115 """Reinstall an instance.
2118 HPATH = "instance-reinstall"
2119 HTYPE = constants.HTYPE_INSTANCE
2120 _OP_REQP = ["instance_name"]
2122 def BuildHooksEnv(self):
2125 This runs on master, primary and secondary nodes of the instance.
2128 env = _BuildInstanceHookEnvByObject(self.instance)
2129 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2130 list(self.instance.secondary_nodes))
2133 def CheckPrereq(self):
2134 """Check prerequisites.
2136 This checks that the instance is in the cluster and is not running.
2139 instance = self.cfg.GetInstanceInfo(
2140 self.cfg.ExpandInstanceName(self.op.instance_name))
2141 if instance is None:
2142 raise errors.OpPrereqError("Instance '%s' not known" %
2143 self.op.instance_name)
2144 if instance.disk_template == constants.DT_DISKLESS:
2145 raise errors.OpPrereqError("Instance '%s' has no disks" %
2146 self.op.instance_name)
2147 if instance.status != "down":
2148 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2149 self.op.instance_name)
2150 remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2152 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2153 (self.op.instance_name,
2154 instance.primary_node))
2156 self.op.os_type = getattr(self.op, "os_type", None)
2157 if self.op.os_type is not None:
2159 pnode = self.cfg.GetNodeInfo(
2160 self.cfg.ExpandNodeName(instance.primary_node))
2162 raise errors.OpPrereqError("Primary node '%s' is unknown" %
2164 os_obj = rpc.call_os_get(pnode.name, self.op.os_type)
2166 raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2167 " primary node" % self.op.os_type)
2169 self.instance = instance
2171 def Exec(self, feedback_fn):
2172 """Reinstall the instance.
2175 inst = self.instance
2177 if self.op.os_type is not None:
2178 feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2179 inst.os = self.op.os_type
2180 self.cfg.AddInstance(inst)
2182 _StartInstanceDisks(self.cfg, inst, None)
2184 feedback_fn("Running the instance OS create scripts...")
2185 if not rpc.call_instance_os_add(inst.primary_node, inst, "sda", "sdb"):
2186 raise errors.OpExecError("Could not install OS for instance %s "
2188 (inst.name, inst.primary_node))
2190 _ShutdownInstanceDisks(inst, self.cfg)
2193 class LURenameInstance(LogicalUnit):
2194 """Rename an instance.
2197 HPATH = "instance-rename"
2198 HTYPE = constants.HTYPE_INSTANCE
2199 _OP_REQP = ["instance_name", "new_name"]
2201 def BuildHooksEnv(self):
2204 This runs on master, primary and secondary nodes of the instance.
2207 env = _BuildInstanceHookEnvByObject(self.instance)
2208 env["INSTANCE_NEW_NAME"] = self.op.new_name
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.status != "down":
2225 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2226 self.op.instance_name)
2227 remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2229 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2230 (self.op.instance_name,
2231 instance.primary_node))
2232 self.instance = instance
2234 # new name verification
2235 name_info = utils.HostInfo(self.op.new_name)
2237 self.op.new_name = new_name = name_info.name
2238 if not getattr(self.op, "ignore_ip", False):
2239 command = ["fping", "-q", name_info.ip]
2240 result = utils.RunCmd(command)
2241 if not result.failed:
2242 raise errors.OpPrereqError("IP %s of instance %s already in use" %
2243 (name_info.ip, new_name))
2246 def Exec(self, feedback_fn):
2247 """Reinstall the instance.
2250 inst = self.instance
2251 old_name = inst.name
2253 self.cfg.RenameInstance(inst.name, self.op.new_name)
2255 # re-read the instance from the configuration after rename
2256 inst = self.cfg.GetInstanceInfo(self.op.new_name)
2258 _StartInstanceDisks(self.cfg, inst, None)
2260 if not rpc.call_instance_run_rename(inst.primary_node, inst, old_name,
2262 msg = ("Could run OS rename script for instance %s\n"
2264 "(but the instance has been renamed in Ganeti)" %
2265 (inst.name, inst.primary_node))
2268 _ShutdownInstanceDisks(inst, self.cfg)
2271 class LURemoveInstance(LogicalUnit):
2272 """Remove an instance.
2275 HPATH = "instance-remove"
2276 HTYPE = constants.HTYPE_INSTANCE
2277 _OP_REQP = ["instance_name"]
2279 def BuildHooksEnv(self):
2282 This runs on master, primary and secondary nodes of the instance.
2285 env = _BuildInstanceHookEnvByObject(self.instance)
2286 nl = [self.sstore.GetMasterNode()]
2289 def CheckPrereq(self):
2290 """Check prerequisites.
2292 This checks that the instance is in the cluster.
2295 instance = self.cfg.GetInstanceInfo(
2296 self.cfg.ExpandInstanceName(self.op.instance_name))
2297 if instance is None:
2298 raise errors.OpPrereqError("Instance '%s' not known" %
2299 self.op.instance_name)
2300 self.instance = instance
2302 def Exec(self, feedback_fn):
2303 """Remove the instance.
2306 instance = self.instance
2307 logger.Info("shutting down instance %s on node %s" %
2308 (instance.name, instance.primary_node))
2310 if not rpc.call_instance_shutdown(instance.primary_node, instance):
2311 if self.op.ignore_failures:
2312 feedback_fn("Warning: can't shutdown instance")
2314 raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2315 (instance.name, instance.primary_node))
2317 logger.Info("removing block devices for instance %s" % instance.name)
2319 if not _RemoveDisks(instance, self.cfg):
2320 if self.op.ignore_failures:
2321 feedback_fn("Warning: can't remove instance's disks")
2323 raise errors.OpExecError("Can't remove instance's disks")
2325 logger.Info("removing instance %s out of cluster config" % instance.name)
2327 self.cfg.RemoveInstance(instance.name)
2330 class LUQueryInstances(NoHooksLU):
2331 """Logical unit for querying instances.
2334 _OP_REQP = ["output_fields", "names"]
2336 def CheckPrereq(self):
2337 """Check prerequisites.
2339 This checks that the fields required are valid output fields.
2342 self.dynamic_fields = frozenset(["oper_state", "oper_ram"])
2343 _CheckOutputFields(static=["name", "os", "pnode", "snodes",
2344 "admin_state", "admin_ram",
2345 "disk_template", "ip", "mac", "bridge",
2346 "sda_size", "sdb_size"],
2347 dynamic=self.dynamic_fields,
2348 selected=self.op.output_fields)
2350 self.wanted = _GetWantedInstances(self, self.op.names)
2352 def Exec(self, feedback_fn):
2353 """Computes the list of nodes and their attributes.
2356 instance_names = self.wanted
2357 instance_list = [self.cfg.GetInstanceInfo(iname) for iname
2360 # begin data gathering
2362 nodes = frozenset([inst.primary_node for inst in instance_list])
2365 if self.dynamic_fields.intersection(self.op.output_fields):
2367 node_data = rpc.call_all_instances_info(nodes)
2369 result = node_data[name]
2371 live_data.update(result)
2372 elif result == False:
2373 bad_nodes.append(name)
2374 # else no instance is alive
2376 live_data = dict([(name, {}) for name in instance_names])
2378 # end data gathering
2381 for instance in instance_list:
2383 for field in self.op.output_fields:
2388 elif field == "pnode":
2389 val = instance.primary_node
2390 elif field == "snodes":
2391 val = list(instance.secondary_nodes)
2392 elif field == "admin_state":
2393 val = (instance.status != "down")
2394 elif field == "oper_state":
2395 if instance.primary_node in bad_nodes:
2398 val = bool(live_data.get(instance.name))
2399 elif field == "admin_ram":
2400 val = instance.memory
2401 elif field == "oper_ram":
2402 if instance.primary_node in bad_nodes:
2404 elif instance.name in live_data:
2405 val = live_data[instance.name].get("memory", "?")
2408 elif field == "disk_template":
2409 val = instance.disk_template
2411 val = instance.nics[0].ip
2412 elif field == "bridge":
2413 val = instance.nics[0].bridge
2414 elif field == "mac":
2415 val = instance.nics[0].mac
2416 elif field == "sda_size" or field == "sdb_size":
2417 disk = instance.FindDisk(field[:3])
2423 raise errors.ParameterError(field)
2430 class LUFailoverInstance(LogicalUnit):
2431 """Failover an instance.
2434 HPATH = "instance-failover"
2435 HTYPE = constants.HTYPE_INSTANCE
2436 _OP_REQP = ["instance_name", "ignore_consistency"]
2438 def BuildHooksEnv(self):
2441 This runs on master, primary and secondary nodes of the instance.
2445 "IGNORE_CONSISTENCY": self.op.ignore_consistency,
2447 env.update(_BuildInstanceHookEnvByObject(self.instance))
2448 nl = [self.sstore.GetMasterNode()] + list(self.instance.secondary_nodes)
2451 def CheckPrereq(self):
2452 """Check prerequisites.
2454 This checks that the instance is in the cluster.
2457 instance = self.cfg.GetInstanceInfo(
2458 self.cfg.ExpandInstanceName(self.op.instance_name))
2459 if instance is None:
2460 raise errors.OpPrereqError("Instance '%s' not known" %
2461 self.op.instance_name)
2463 if instance.disk_template not in constants.DTS_NET_MIRROR:
2464 raise errors.OpPrereqError("Instance's disk layout is not"
2465 " network mirrored, cannot failover.")
2467 secondary_nodes = instance.secondary_nodes
2468 if not secondary_nodes:
2469 raise errors.ProgrammerError("no secondary node but using "
2470 "DT_REMOTE_RAID1 template")
2472 # check memory requirements on the secondary node
2473 target_node = secondary_nodes[0]
2474 nodeinfo = rpc.call_node_info([target_node], self.cfg.GetVGName())
2475 info = nodeinfo.get(target_node, None)
2477 raise errors.OpPrereqError("Cannot get current information"
2478 " from node '%s'" % nodeinfo)
2479 if instance.memory > info['memory_free']:
2480 raise errors.OpPrereqError("Not enough memory on target node %s."
2481 " %d MB available, %d MB required" %
2482 (target_node, info['memory_free'],
2485 # check bridge existance
2486 brlist = [nic.bridge for nic in instance.nics]
2487 if not rpc.call_bridges_exist(target_node, brlist):
2488 raise errors.OpPrereqError("One or more target bridges %s does not"
2489 " exist on destination node '%s'" %
2490 (brlist, target_node))
2492 self.instance = instance
2494 def Exec(self, feedback_fn):
2495 """Failover an instance.
2497 The failover is done by shutting it down on its present node and
2498 starting it on the secondary.
2501 instance = self.instance
2503 source_node = instance.primary_node
2504 target_node = instance.secondary_nodes[0]
2506 feedback_fn("* checking disk consistency between source and target")
2507 for dev in instance.disks:
2508 # for remote_raid1, these are md over drbd
2509 if not _CheckDiskConsistency(self.cfg, dev, target_node, False):
2510 if not self.op.ignore_consistency:
2511 raise errors.OpExecError("Disk %s is degraded on target node,"
2512 " aborting failover." % dev.iv_name)
2514 feedback_fn("* checking target node resource availability")
2515 nodeinfo = rpc.call_node_info([target_node], self.cfg.GetVGName())
2518 raise errors.OpExecError("Could not contact target node %s." %
2521 free_memory = int(nodeinfo[target_node]['memory_free'])
2522 memory = instance.memory
2523 if memory > free_memory:
2524 raise errors.OpExecError("Not enough memory to create instance %s on"
2525 " node %s. needed %s MiB, available %s MiB" %
2526 (instance.name, target_node, memory,
2529 feedback_fn("* shutting down instance on source node")
2530 logger.Info("Shutting down instance %s on node %s" %
2531 (instance.name, source_node))
2533 if not rpc.call_instance_shutdown(source_node, instance):
2534 if self.op.ignore_consistency:
2535 logger.Error("Could not shutdown instance %s on node %s. Proceeding"
2536 " anyway. Please make sure node %s is down" %
2537 (instance.name, source_node, source_node))
2539 raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2540 (instance.name, source_node))
2542 feedback_fn("* deactivating the instance's disks on source node")
2543 if not _ShutdownInstanceDisks(instance, self.cfg, ignore_primary=True):
2544 raise errors.OpExecError("Can't shut down the instance's disks.")
2546 instance.primary_node = target_node
2547 # distribute new instance config to the other nodes
2548 self.cfg.AddInstance(instance)
2550 feedback_fn("* activating the instance's disks on target node")
2551 logger.Info("Starting instance %s on node %s" %
2552 (instance.name, target_node))
2554 disks_ok, dummy = _AssembleInstanceDisks(instance, self.cfg,
2555 ignore_secondaries=True)
2557 _ShutdownInstanceDisks(instance, self.cfg)
2558 raise errors.OpExecError("Can't activate the instance's disks")
2560 feedback_fn("* starting the instance on the target node")
2561 if not rpc.call_instance_start(target_node, instance, None):
2562 _ShutdownInstanceDisks(instance, self.cfg)
2563 raise errors.OpExecError("Could not start instance %s on node %s." %
2564 (instance.name, target_node))
2567 def _CreateBlockDevOnPrimary(cfg, node, instance, device, info):
2568 """Create a tree of block devices on the primary node.
2570 This always creates all devices.
2574 for child in device.children:
2575 if not _CreateBlockDevOnPrimary(cfg, node, instance, child, info):
2578 cfg.SetDiskID(device, node)
2579 new_id = rpc.call_blockdev_create(node, device, device.size,
2580 instance.name, True, info)
2583 if device.physical_id is None:
2584 device.physical_id = new_id
2588 def _CreateBlockDevOnSecondary(cfg, node, instance, device, force, info):
2589 """Create a tree of block devices on a secondary node.
2591 If this device type has to be created on secondaries, create it and
2594 If not, just recurse to children keeping the same 'force' value.
2597 if device.CreateOnSecondary():
2600 for child in device.children:
2601 if not _CreateBlockDevOnSecondary(cfg, node, instance,
2602 child, force, info):
2607 cfg.SetDiskID(device, node)
2608 new_id = rpc.call_blockdev_create(node, device, device.size,
2609 instance.name, False, info)
2612 if device.physical_id is None:
2613 device.physical_id = new_id
2617 def _GenerateUniqueNames(cfg, exts):
2618 """Generate a suitable LV name.
2620 This will generate a logical volume name for the given instance.
2625 new_id = cfg.GenerateUniqueID()
2626 results.append("%s%s" % (new_id, val))
2630 def _GenerateMDDRBDBranch(cfg, primary, secondary, size, names):
2631 """Generate a drbd device complete with its children.
2634 port = cfg.AllocatePort()
2635 vgname = cfg.GetVGName()
2636 dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
2637 logical_id=(vgname, names[0]))
2638 dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
2639 logical_id=(vgname, names[1]))
2640 drbd_dev = objects.Disk(dev_type=constants.LD_DRBD7, size=size,
2641 logical_id = (primary, secondary, port),
2642 children = [dev_data, dev_meta])
2646 def _GenerateDRBD8Branch(cfg, primary, secondary, size, names, iv_name):
2647 """Generate a drbd8 device complete with its children.
2650 port = cfg.AllocatePort()
2651 vgname = cfg.GetVGName()
2652 dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
2653 logical_id=(vgname, names[0]))
2654 dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
2655 logical_id=(vgname, names[1]))
2656 drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
2657 logical_id = (primary, secondary, port),
2658 children = [dev_data, dev_meta],
2662 def _GenerateDiskTemplate(cfg, template_name,
2663 instance_name, primary_node,
2664 secondary_nodes, disk_sz, swap_sz):
2665 """Generate the entire disk layout for a given template type.
2668 #TODO: compute space requirements
2670 vgname = cfg.GetVGName()
2671 if template_name == "diskless":
2673 elif template_name == "plain":
2674 if len(secondary_nodes) != 0:
2675 raise errors.ProgrammerError("Wrong template configuration")
2677 names = _GenerateUniqueNames(cfg, [".sda", ".sdb"])
2678 sda_dev = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2679 logical_id=(vgname, names[0]),
2681 sdb_dev = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2682 logical_id=(vgname, names[1]),
2684 disks = [sda_dev, sdb_dev]
2685 elif template_name == "local_raid1":
2686 if len(secondary_nodes) != 0:
2687 raise errors.ProgrammerError("Wrong template configuration")
2690 names = _GenerateUniqueNames(cfg, [".sda_m1", ".sda_m2",
2691 ".sdb_m1", ".sdb_m2"])
2692 sda_dev_m1 = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2693 logical_id=(vgname, names[0]))
2694 sda_dev_m2 = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2695 logical_id=(vgname, names[1]))
2696 md_sda_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name = "sda",
2698 children = [sda_dev_m1, sda_dev_m2])
2699 sdb_dev_m1 = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2700 logical_id=(vgname, names[2]))
2701 sdb_dev_m2 = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2702 logical_id=(vgname, names[3]))
2703 md_sdb_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name = "sdb",
2705 children = [sdb_dev_m1, sdb_dev_m2])
2706 disks = [md_sda_dev, md_sdb_dev]
2707 elif template_name == constants.DT_REMOTE_RAID1:
2708 if len(secondary_nodes) != 1:
2709 raise errors.ProgrammerError("Wrong template configuration")
2710 remote_node = secondary_nodes[0]
2711 names = _GenerateUniqueNames(cfg, [".sda_data", ".sda_meta",
2712 ".sdb_data", ".sdb_meta"])
2713 drbd_sda_dev = _GenerateMDDRBDBranch(cfg, primary_node, remote_node,
2714 disk_sz, names[0:2])
2715 md_sda_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name="sda",
2716 children = [drbd_sda_dev], size=disk_sz)
2717 drbd_sdb_dev = _GenerateMDDRBDBranch(cfg, primary_node, remote_node,
2718 swap_sz, names[2:4])
2719 md_sdb_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name="sdb",
2720 children = [drbd_sdb_dev], size=swap_sz)
2721 disks = [md_sda_dev, md_sdb_dev]
2722 elif template_name == constants.DT_DRBD8:
2723 if len(secondary_nodes) != 1:
2724 raise errors.ProgrammerError("Wrong template configuration")
2725 remote_node = secondary_nodes[0]
2726 names = _GenerateUniqueNames(cfg, [".sda_data", ".sda_meta",
2727 ".sdb_data", ".sdb_meta"])
2728 drbd_sda_dev = _GenerateDRBD8Branch(cfg, primary_node, remote_node,
2729 disk_sz, names[0:2], "sda")
2730 drbd_sdb_dev = _GenerateDRBD8Branch(cfg, primary_node, remote_node,
2731 swap_sz, names[2:4], "sdb")
2732 disks = [drbd_sda_dev, drbd_sdb_dev]
2734 raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
2738 def _GetInstanceInfoText(instance):
2739 """Compute that text that should be added to the disk's metadata.
2742 return "originstname+%s" % instance.name
2745 def _CreateDisks(cfg, instance):
2746 """Create all disks for an instance.
2748 This abstracts away some work from AddInstance.
2751 instance: the instance object
2754 True or False showing the success of the creation process
2757 info = _GetInstanceInfoText(instance)
2759 for device in instance.disks:
2760 logger.Info("creating volume %s for instance %s" %
2761 (device.iv_name, instance.name))
2763 for secondary_node in instance.secondary_nodes:
2764 if not _CreateBlockDevOnSecondary(cfg, secondary_node, instance,
2765 device, False, info):
2766 logger.Error("failed to create volume %s (%s) on secondary node %s!" %
2767 (device.iv_name, device, secondary_node))
2770 if not _CreateBlockDevOnPrimary(cfg, instance.primary_node,
2771 instance, device, info):
2772 logger.Error("failed to create volume %s on primary!" %
2778 def _RemoveDisks(instance, cfg):
2779 """Remove all disks for an instance.
2781 This abstracts away some work from `AddInstance()` and
2782 `RemoveInstance()`. Note that in case some of the devices couldn't
2783 be removed, the removal will continue with the other ones (compare
2784 with `_CreateDisks()`).
2787 instance: the instance object
2790 True or False showing the success of the removal proces
2793 logger.Info("removing block devices for instance %s" % instance.name)
2796 for device in instance.disks:
2797 for node, disk in device.ComputeNodeTree(instance.primary_node):
2798 cfg.SetDiskID(disk, node)
2799 if not rpc.call_blockdev_remove(node, disk):
2800 logger.Error("could not remove block device %s on node %s,"
2801 " continuing anyway" %
2802 (device.iv_name, node))
2807 class LUCreateInstance(LogicalUnit):
2808 """Create an instance.
2811 HPATH = "instance-add"
2812 HTYPE = constants.HTYPE_INSTANCE
2813 _OP_REQP = ["instance_name", "mem_size", "disk_size", "pnode",
2814 "disk_template", "swap_size", "mode", "start", "vcpus",
2815 "wait_for_sync", "ip_check"]
2817 def BuildHooksEnv(self):
2820 This runs on master, primary and secondary nodes of the instance.
2824 "INSTANCE_DISK_TEMPLATE": self.op.disk_template,
2825 "INSTANCE_DISK_SIZE": self.op.disk_size,
2826 "INSTANCE_SWAP_SIZE": self.op.swap_size,
2827 "INSTANCE_ADD_MODE": self.op.mode,
2829 if self.op.mode == constants.INSTANCE_IMPORT:
2830 env["INSTANCE_SRC_NODE"] = self.op.src_node
2831 env["INSTANCE_SRC_PATH"] = self.op.src_path
2832 env["INSTANCE_SRC_IMAGE"] = self.src_image
2834 env.update(_BuildInstanceHookEnv(name=self.op.instance_name,
2835 primary_node=self.op.pnode,
2836 secondary_nodes=self.secondaries,
2837 status=self.instance_status,
2838 os_type=self.op.os_type,
2839 memory=self.op.mem_size,
2840 vcpus=self.op.vcpus,
2841 nics=[(self.inst_ip, self.op.bridge)],
2844 nl = ([self.sstore.GetMasterNode(), self.op.pnode] +
2849 def CheckPrereq(self):
2850 """Check prerequisites.
2853 if self.op.mode not in (constants.INSTANCE_CREATE,
2854 constants.INSTANCE_IMPORT):
2855 raise errors.OpPrereqError("Invalid instance creation mode '%s'" %
2858 if self.op.mode == constants.INSTANCE_IMPORT:
2859 src_node = getattr(self.op, "src_node", None)
2860 src_path = getattr(self.op, "src_path", None)
2861 if src_node is None or src_path is None:
2862 raise errors.OpPrereqError("Importing an instance requires source"
2863 " node and path options")
2864 src_node_full = self.cfg.ExpandNodeName(src_node)
2865 if src_node_full is None:
2866 raise errors.OpPrereqError("Unknown source node '%s'" % src_node)
2867 self.op.src_node = src_node = src_node_full
2869 if not os.path.isabs(src_path):
2870 raise errors.OpPrereqError("The source path must be absolute")
2872 export_info = rpc.call_export_info(src_node, src_path)
2875 raise errors.OpPrereqError("No export found in dir %s" % src_path)
2877 if not export_info.has_section(constants.INISECT_EXP):
2878 raise errors.ProgrammerError("Corrupted export config")
2880 ei_version = export_info.get(constants.INISECT_EXP, 'version')
2881 if (int(ei_version) != constants.EXPORT_VERSION):
2882 raise errors.OpPrereqError("Wrong export version %s (wanted %d)" %
2883 (ei_version, constants.EXPORT_VERSION))
2885 if int(export_info.get(constants.INISECT_INS, 'disk_count')) > 1:
2886 raise errors.OpPrereqError("Can't import instance with more than"
2889 # FIXME: are the old os-es, disk sizes, etc. useful?
2890 self.op.os_type = export_info.get(constants.INISECT_EXP, 'os')
2891 diskimage = os.path.join(src_path, export_info.get(constants.INISECT_INS,
2893 self.src_image = diskimage
2894 else: # INSTANCE_CREATE
2895 if getattr(self.op, "os_type", None) is None:
2896 raise errors.OpPrereqError("No guest OS specified")
2898 # check primary node
2899 pnode = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.pnode))
2901 raise errors.OpPrereqError("Primary node '%s' is unknown" %
2903 self.op.pnode = pnode.name
2905 self.secondaries = []
2906 # disk template and mirror node verification
2907 if self.op.disk_template not in constants.DISK_TEMPLATES:
2908 raise errors.OpPrereqError("Invalid disk template name")
2910 if self.op.disk_template in constants.DTS_NET_MIRROR:
2911 if getattr(self.op, "snode", None) is None:
2912 raise errors.OpPrereqError("The networked disk templates need"
2915 snode_name = self.cfg.ExpandNodeName(self.op.snode)
2916 if snode_name is None:
2917 raise errors.OpPrereqError("Unknown secondary node '%s'" %
2919 elif snode_name == pnode.name:
2920 raise errors.OpPrereqError("The secondary node cannot be"
2921 " the primary node.")
2922 self.secondaries.append(snode_name)
2924 # Check lv size requirements
2925 nodenames = [pnode.name] + self.secondaries
2926 nodeinfo = rpc.call_node_info(nodenames, self.cfg.GetVGName())
2928 # Required free disk space as a function of disk and swap space
2930 constants.DT_DISKLESS: 0,
2931 constants.DT_PLAIN: self.op.disk_size + self.op.swap_size,
2932 constants.DT_LOCAL_RAID1: (self.op.disk_size + self.op.swap_size) * 2,
2933 # 256 MB are added for drbd metadata, 128MB for each drbd device
2934 constants.DT_REMOTE_RAID1: self.op.disk_size + self.op.swap_size + 256,
2935 constants.DT_DRBD8: self.op.disk_size + self.op.swap_size + 256,
2938 if self.op.disk_template not in req_size_dict:
2939 raise errors.ProgrammerError("Disk template '%s' size requirement"
2940 " is unknown" % self.op.disk_template)
2942 req_size = req_size_dict[self.op.disk_template]
2944 for node in nodenames:
2945 info = nodeinfo.get(node, None)
2947 raise errors.OpPrereqError("Cannot get current information"
2948 " from node '%s'" % nodeinfo)
2949 if req_size > info['vg_free']:
2950 raise errors.OpPrereqError("Not enough disk space on target node %s."
2951 " %d MB available, %d MB required" %
2952 (node, info['vg_free'], req_size))
2955 os_obj = rpc.call_os_get(pnode.name, self.op.os_type)
2957 raise errors.OpPrereqError("OS '%s' not in supported os list for"
2958 " primary node" % self.op.os_type)
2960 # instance verification
2961 hostname1 = utils.HostInfo(self.op.instance_name)
2963 self.op.instance_name = instance_name = hostname1.name
2964 instance_list = self.cfg.GetInstanceList()
2965 if instance_name in instance_list:
2966 raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
2969 ip = getattr(self.op, "ip", None)
2970 if ip is None or ip.lower() == "none":
2972 elif ip.lower() == "auto":
2973 inst_ip = hostname1.ip
2975 if not utils.IsValidIP(ip):
2976 raise errors.OpPrereqError("given IP address '%s' doesn't look"
2977 " like a valid IP" % ip)
2979 self.inst_ip = inst_ip
2981 if self.op.start and not self.op.ip_check:
2982 raise errors.OpPrereqError("Cannot ignore IP address conflicts when"
2983 " adding an instance in start mode")
2985 if self.op.ip_check:
2986 if utils.TcpPing(utils.HostInfo().name, hostname1.ip,
2987 constants.DEFAULT_NODED_PORT):
2988 raise errors.OpPrereqError("IP %s of instance %s already in use" %
2989 (hostname1.ip, instance_name))
2991 # bridge verification
2992 bridge = getattr(self.op, "bridge", None)
2994 self.op.bridge = self.cfg.GetDefBridge()
2996 self.op.bridge = bridge
2998 if not rpc.call_bridges_exist(self.pnode.name, [self.op.bridge]):
2999 raise errors.OpPrereqError("target bridge '%s' does not exist on"
3000 " destination node '%s'" %
3001 (self.op.bridge, pnode.name))
3004 self.instance_status = 'up'
3006 self.instance_status = 'down'
3008 def Exec(self, feedback_fn):
3009 """Create and add the instance to the cluster.
3012 instance = self.op.instance_name
3013 pnode_name = self.pnode.name
3015 nic = objects.NIC(bridge=self.op.bridge, mac=self.cfg.GenerateMAC())
3016 if self.inst_ip is not None:
3017 nic.ip = self.inst_ip
3019 disks = _GenerateDiskTemplate(self.cfg,
3020 self.op.disk_template,
3021 instance, pnode_name,
3022 self.secondaries, self.op.disk_size,
3025 iobj = objects.Instance(name=instance, os=self.op.os_type,
3026 primary_node=pnode_name,
3027 memory=self.op.mem_size,
3028 vcpus=self.op.vcpus,
3029 nics=[nic], disks=disks,
3030 disk_template=self.op.disk_template,
3031 status=self.instance_status,
3034 feedback_fn("* creating instance disks...")
3035 if not _CreateDisks(self.cfg, iobj):
3036 _RemoveDisks(iobj, self.cfg)
3037 raise errors.OpExecError("Device creation failed, reverting...")
3039 feedback_fn("adding instance %s to cluster config" % instance)
3041 self.cfg.AddInstance(iobj)
3043 if self.op.wait_for_sync:
3044 disk_abort = not _WaitForSync(self.cfg, iobj)
3045 elif iobj.disk_template in constants.DTS_NET_MIRROR:
3046 # make sure the disks are not degraded (still sync-ing is ok)
3048 feedback_fn("* checking mirrors status")
3049 disk_abort = not _WaitForSync(self.cfg, iobj, oneshot=True)
3054 _RemoveDisks(iobj, self.cfg)
3055 self.cfg.RemoveInstance(iobj.name)
3056 raise errors.OpExecError("There are some degraded disks for"
3059 feedback_fn("creating os for instance %s on node %s" %
3060 (instance, pnode_name))
3062 if iobj.disk_template != constants.DT_DISKLESS:
3063 if self.op.mode == constants.INSTANCE_CREATE:
3064 feedback_fn("* running the instance OS create scripts...")
3065 if not rpc.call_instance_os_add(pnode_name, iobj, "sda", "sdb"):
3066 raise errors.OpExecError("could not add os for instance %s"
3068 (instance, pnode_name))
3070 elif self.op.mode == constants.INSTANCE_IMPORT:
3071 feedback_fn("* running the instance OS import scripts...")
3072 src_node = self.op.src_node
3073 src_image = self.src_image
3074 if not rpc.call_instance_os_import(pnode_name, iobj, "sda", "sdb",
3075 src_node, src_image):
3076 raise errors.OpExecError("Could not import os for instance"
3078 (instance, pnode_name))
3080 # also checked in the prereq part
3081 raise errors.ProgrammerError("Unknown OS initialization mode '%s'"
3085 logger.Info("starting instance %s on node %s" % (instance, pnode_name))
3086 feedback_fn("* starting instance...")
3087 if not rpc.call_instance_start(pnode_name, iobj, None):
3088 raise errors.OpExecError("Could not start instance")
3091 class LUConnectConsole(NoHooksLU):
3092 """Connect to an instance's console.
3094 This is somewhat special in that it returns the command line that
3095 you need to run on the master node in order to connect to the
3099 _OP_REQP = ["instance_name"]
3101 def CheckPrereq(self):
3102 """Check prerequisites.
3104 This checks that the instance is in the cluster.
3107 instance = self.cfg.GetInstanceInfo(
3108 self.cfg.ExpandInstanceName(self.op.instance_name))
3109 if instance is None:
3110 raise errors.OpPrereqError("Instance '%s' not known" %
3111 self.op.instance_name)
3112 self.instance = instance
3114 def Exec(self, feedback_fn):
3115 """Connect to the console of an instance
3118 instance = self.instance
3119 node = instance.primary_node
3121 node_insts = rpc.call_instance_list([node])[node]
3122 if node_insts is False:
3123 raise errors.OpExecError("Can't connect to node %s." % node)
3125 if instance.name not in node_insts:
3126 raise errors.OpExecError("Instance %s is not running." % instance.name)
3128 logger.Debug("connecting to console of %s on %s" % (instance.name, node))
3130 hyper = hypervisor.GetHypervisor()
3131 console_cmd = hyper.GetShellCommandForConsole(instance.name)
3133 argv = ["ssh", "-q", "-t"]
3134 argv.extend(ssh.KNOWN_HOSTS_OPTS)
3135 argv.extend(ssh.BATCH_MODE_OPTS)
3137 argv.append(console_cmd)
3141 class LUAddMDDRBDComponent(LogicalUnit):
3142 """Adda new mirror member to an instance's disk.
3145 HPATH = "mirror-add"
3146 HTYPE = constants.HTYPE_INSTANCE
3147 _OP_REQP = ["instance_name", "remote_node", "disk_name"]
3149 def BuildHooksEnv(self):
3152 This runs on the master, the primary and all the secondaries.
3156 "NEW_SECONDARY": self.op.remote_node,
3157 "DISK_NAME": self.op.disk_name,
3159 env.update(_BuildInstanceHookEnvByObject(self.instance))
3160 nl = [self.sstore.GetMasterNode(), self.instance.primary_node,
3161 self.op.remote_node,] + list(self.instance.secondary_nodes)
3164 def CheckPrereq(self):
3165 """Check prerequisites.
3167 This checks that the instance is in the cluster.
3170 instance = self.cfg.GetInstanceInfo(
3171 self.cfg.ExpandInstanceName(self.op.instance_name))
3172 if instance is None:
3173 raise errors.OpPrereqError("Instance '%s' not known" %
3174 self.op.instance_name)
3175 self.instance = instance
3177 remote_node = self.cfg.ExpandNodeName(self.op.remote_node)
3178 if remote_node is None:
3179 raise errors.OpPrereqError("Node '%s' not known" % self.op.remote_node)
3180 self.remote_node = remote_node
3182 if remote_node == instance.primary_node:
3183 raise errors.OpPrereqError("The specified node is the primary node of"
3186 if instance.disk_template != constants.DT_REMOTE_RAID1:
3187 raise errors.OpPrereqError("Instance's disk layout is not"
3189 for disk in instance.disks:
3190 if disk.iv_name == self.op.disk_name:
3193 raise errors.OpPrereqError("Can't find this device ('%s') in the"
3194 " instance." % self.op.disk_name)
3195 if len(disk.children) > 1:
3196 raise errors.OpPrereqError("The device already has two slave"
3198 "This would create a 3-disk raid1"
3199 " which we don't allow.")
3202 def Exec(self, feedback_fn):
3203 """Add the mirror component
3207 instance = self.instance
3209 remote_node = self.remote_node
3210 lv_names = [".%s_%s" % (disk.iv_name, suf) for suf in ["data", "meta"]]
3211 names = _GenerateUniqueNames(self.cfg, lv_names)
3212 new_drbd = _GenerateMDDRBDBranch(self.cfg, instance.primary_node,
3213 remote_node, disk.size, names)
3215 logger.Info("adding new mirror component on secondary")
3217 if not _CreateBlockDevOnSecondary(self.cfg, remote_node, instance,
3219 _GetInstanceInfoText(instance)):
3220 raise errors.OpExecError("Failed to create new component on secondary"
3221 " node %s" % remote_node)
3223 logger.Info("adding new mirror component on primary")
3225 if not _CreateBlockDevOnPrimary(self.cfg, instance.primary_node,
3227 _GetInstanceInfoText(instance)):
3228 # remove secondary dev
3229 self.cfg.SetDiskID(new_drbd, remote_node)
3230 rpc.call_blockdev_remove(remote_node, new_drbd)
3231 raise errors.OpExecError("Failed to create volume on primary")
3233 # the device exists now
3234 # call the primary node to add the mirror to md
3235 logger.Info("adding new mirror component to md")
3236 if not rpc.call_blockdev_addchildren(instance.primary_node,
3238 logger.Error("Can't add mirror compoment to md!")
3239 self.cfg.SetDiskID(new_drbd, remote_node)
3240 if not rpc.call_blockdev_remove(remote_node, new_drbd):
3241 logger.Error("Can't rollback on secondary")
3242 self.cfg.SetDiskID(new_drbd, instance.primary_node)
3243 if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
3244 logger.Error("Can't rollback on primary")
3245 raise errors.OpExecError("Can't add mirror component to md array")
3247 disk.children.append(new_drbd)
3249 self.cfg.AddInstance(instance)
3251 _WaitForSync(self.cfg, instance)
3256 class LURemoveMDDRBDComponent(LogicalUnit):
3257 """Remove a component from a remote_raid1 disk.
3260 HPATH = "mirror-remove"
3261 HTYPE = constants.HTYPE_INSTANCE
3262 _OP_REQP = ["instance_name", "disk_name", "disk_id"]
3264 def BuildHooksEnv(self):
3267 This runs on the master, the primary and all the secondaries.
3271 "DISK_NAME": self.op.disk_name,
3272 "DISK_ID": self.op.disk_id,
3273 "OLD_SECONDARY": self.old_secondary,
3275 env.update(_BuildInstanceHookEnvByObject(self.instance))
3276 nl = [self.sstore.GetMasterNode(),
3277 self.instance.primary_node] + list(self.instance.secondary_nodes)
3280 def CheckPrereq(self):
3281 """Check prerequisites.
3283 This checks that the instance is in the cluster.
3286 instance = self.cfg.GetInstanceInfo(
3287 self.cfg.ExpandInstanceName(self.op.instance_name))
3288 if instance is None:
3289 raise errors.OpPrereqError("Instance '%s' not known" %
3290 self.op.instance_name)
3291 self.instance = instance
3293 if instance.disk_template != constants.DT_REMOTE_RAID1:
3294 raise errors.OpPrereqError("Instance's disk layout is not"
3296 for disk in instance.disks:
3297 if disk.iv_name == self.op.disk_name:
3300 raise errors.OpPrereqError("Can't find this device ('%s') in the"
3301 " instance." % self.op.disk_name)
3302 for child in disk.children:
3303 if (child.dev_type == constants.LD_DRBD7 and
3304 child.logical_id[2] == self.op.disk_id):
3307 raise errors.OpPrereqError("Can't find the device with this port.")
3309 if len(disk.children) < 2:
3310 raise errors.OpPrereqError("Cannot remove the last component from"
3314 if self.child.logical_id[0] == instance.primary_node:
3318 self.old_secondary = self.child.logical_id[oid]
3320 def Exec(self, feedback_fn):
3321 """Remove the mirror component
3324 instance = self.instance
3327 logger.Info("remove mirror component")
3328 self.cfg.SetDiskID(disk, instance.primary_node)
3329 if not rpc.call_blockdev_removechildren(instance.primary_node,
3331 raise errors.OpExecError("Can't remove child from mirror.")
3333 for node in child.logical_id[:2]:
3334 self.cfg.SetDiskID(child, node)
3335 if not rpc.call_blockdev_remove(node, child):
3336 logger.Error("Warning: failed to remove device from node %s,"
3337 " continuing operation." % node)
3339 disk.children.remove(child)
3340 self.cfg.AddInstance(instance)
3343 class LUReplaceDisks(LogicalUnit):
3344 """Replace the disks of an instance.
3347 HPATH = "mirrors-replace"
3348 HTYPE = constants.HTYPE_INSTANCE
3349 _OP_REQP = ["instance_name", "mode", "disks"]
3351 def BuildHooksEnv(self):
3354 This runs on the master, the primary and all the secondaries.
3358 "MODE": self.op.mode,
3359 "NEW_SECONDARY": self.op.remote_node,
3360 "OLD_SECONDARY": self.instance.secondary_nodes[0],
3362 env.update(_BuildInstanceHookEnvByObject(self.instance))
3363 nl = [self.sstore.GetMasterNode(),
3364 self.instance.primary_node] + list(self.instance.secondary_nodes)
3367 def CheckPrereq(self):
3368 """Check prerequisites.
3370 This checks that the instance is in the cluster.
3373 instance = self.cfg.GetInstanceInfo(
3374 self.cfg.ExpandInstanceName(self.op.instance_name))
3375 if instance is None:
3376 raise errors.OpPrereqError("Instance '%s' not known" %
3377 self.op.instance_name)
3378 self.instance = instance
3380 if instance.disk_template not in constants.DTS_NET_MIRROR:
3381 raise errors.OpPrereqError("Instance's disk layout is not"
3382 " network mirrored.")
3384 if len(instance.secondary_nodes) != 1:
3385 raise errors.OpPrereqError("The instance has a strange layout,"
3386 " expected one secondary but found %d" %
3387 len(instance.secondary_nodes))
3389 self.sec_node = instance.secondary_nodes[0]
3391 remote_node = getattr(self.op, "remote_node", None)
3392 if remote_node is not None:
3393 remote_node = self.cfg.ExpandNodeName(remote_node)
3394 if remote_node is None:
3395 raise errors.OpPrereqError("Node '%s' not known" %
3396 self.op.remote_node)
3397 self.remote_node_info = self.cfg.GetNodeInfo(remote_node)
3399 self.remote_node_info = None
3400 if remote_node == instance.primary_node:
3401 raise errors.OpPrereqError("The specified node is the primary node of"
3403 elif remote_node == self.sec_node:
3404 # the user gave the current secondary, switch to
3405 # 'no-replace-secondary' mode
3407 if (instance.disk_template == constants.DT_REMOTE_RAID1 and
3408 self.op.mode != constants.REPLACE_DISK_ALL):
3409 raise errors.OpPrereqError("Template 'remote_raid1' only allows all"
3410 " disks replacement, not individual ones")
3411 if instance.disk_template == constants.DT_DRBD8:
3412 if self.op.mode == constants.REPLACE_DISK_ALL:
3413 raise errors.OpPrereqError("Template 'drbd8' only allows primary or"
3414 " secondary disk replacement, not"
3416 elif self.op.mode == constants.REPLACE_DISK_PRI:
3417 if remote_node is not None:
3418 raise errors.OpPrereqError("Template 'drbd8' does not allow changing"
3419 " the secondary while doing a primary"
3420 " node disk replacement")
3421 self.tgt_node = instance.primary_node
3422 self.oth_node = instance.secondary_nodes[0]
3423 elif self.op.mode == constants.REPLACE_DISK_SEC:
3424 self.new_node = remote_node # this can be None, in which case
3425 # we don't change the secondary
3426 self.tgt_node = instance.secondary_nodes[0]
3427 self.oth_node = instance.primary_node
3429 raise errors.ProgrammerError("Unhandled disk replace mode")
3431 for name in self.op.disks:
3432 if instance.FindDisk(name) is None:
3433 raise errors.OpPrereqError("Disk '%s' not found for instance '%s'" %
3434 (name, instance.name))
3435 self.op.remote_node = remote_node
3437 def _ExecRR1(self, feedback_fn):
3438 """Replace the disks of an instance.
3441 instance = self.instance
3444 if self.op.remote_node is None:
3445 remote_node = self.sec_node
3447 remote_node = self.op.remote_node
3449 for dev in instance.disks:
3451 lv_names = [".%s_%s" % (dev.iv_name, suf) for suf in ["data", "meta"]]
3452 names = _GenerateUniqueNames(cfg, lv_names)
3453 new_drbd = _GenerateMDDRBDBranch(cfg, instance.primary_node,
3454 remote_node, size, names)
3455 iv_names[dev.iv_name] = (dev, dev.children[0], new_drbd)
3456 logger.Info("adding new mirror component on secondary for %s" %
3459 if not _CreateBlockDevOnSecondary(cfg, remote_node, instance,
3461 _GetInstanceInfoText(instance)):
3462 raise errors.OpExecError("Failed to create new component on"
3463 " secondary node %s\n"
3464 "Full abort, cleanup manually!" %
3467 logger.Info("adding new mirror component on primary")
3469 if not _CreateBlockDevOnPrimary(cfg, instance.primary_node,
3471 _GetInstanceInfoText(instance)):
3472 # remove secondary dev
3473 cfg.SetDiskID(new_drbd, remote_node)
3474 rpc.call_blockdev_remove(remote_node, new_drbd)
3475 raise errors.OpExecError("Failed to create volume on primary!\n"
3476 "Full abort, cleanup manually!!")
3478 # the device exists now
3479 # call the primary node to add the mirror to md
3480 logger.Info("adding new mirror component to md")
3481 if not rpc.call_blockdev_addchildren(instance.primary_node, dev,
3483 logger.Error("Can't add mirror compoment to md!")
3484 cfg.SetDiskID(new_drbd, remote_node)
3485 if not rpc.call_blockdev_remove(remote_node, new_drbd):
3486 logger.Error("Can't rollback on secondary")
3487 cfg.SetDiskID(new_drbd, instance.primary_node)
3488 if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
3489 logger.Error("Can't rollback on primary")
3490 raise errors.OpExecError("Full abort, cleanup manually!!")
3492 dev.children.append(new_drbd)
3493 cfg.AddInstance(instance)
3495 # this can fail as the old devices are degraded and _WaitForSync
3496 # does a combined result over all disks, so we don't check its
3498 _WaitForSync(cfg, instance, unlock=True)
3500 # so check manually all the devices
3501 for name in iv_names:
3502 dev, child, new_drbd = iv_names[name]
3503 cfg.SetDiskID(dev, instance.primary_node)
3504 is_degr = rpc.call_blockdev_find(instance.primary_node, dev)[5]
3506 raise errors.OpExecError("MD device %s is degraded!" % name)
3507 cfg.SetDiskID(new_drbd, instance.primary_node)
3508 is_degr = rpc.call_blockdev_find(instance.primary_node, new_drbd)[5]
3510 raise errors.OpExecError("New drbd device %s is degraded!" % name)
3512 for name in iv_names:
3513 dev, child, new_drbd = iv_names[name]
3514 logger.Info("remove mirror %s component" % name)
3515 cfg.SetDiskID(dev, instance.primary_node)
3516 if not rpc.call_blockdev_removechildren(instance.primary_node,
3518 logger.Error("Can't remove child from mirror, aborting"
3519 " *this device cleanup*.\nYou need to cleanup manually!!")
3522 for node in child.logical_id[:2]:
3523 logger.Info("remove child device on %s" % node)
3524 cfg.SetDiskID(child, node)
3525 if not rpc.call_blockdev_remove(node, child):
3526 logger.Error("Warning: failed to remove device from node %s,"
3527 " continuing operation." % node)
3529 dev.children.remove(child)
3531 cfg.AddInstance(instance)
3533 def _ExecD8DiskOnly(self, feedback_fn):
3534 """Replace a disk on the primary or secondary for dbrd8.
3536 The algorithm for replace is quite complicated:
3537 - for each disk to be replaced:
3538 - create new LVs on the target node with unique names
3539 - detach old LVs from the drbd device
3540 - rename old LVs to name_replaced.<time_t>
3541 - rename new LVs to old LVs
3542 - attach the new LVs (with the old names now) to the drbd device
3543 - wait for sync across all devices
3544 - for each modified disk:
3545 - remove old LVs (which have the name name_replaces.<time_t>)
3547 Failures are not very well handled.
3551 warning, info = (self.processor.LogWarning, self.processor.LogInfo)
3552 instance = self.instance
3554 vgname = self.cfg.GetVGName()
3557 tgt_node = self.tgt_node
3558 oth_node = self.oth_node
3560 # Step: check device activation
3561 self.processor.LogStep(1, steps_total, "check device existence")
3562 info("checking volume groups")
3563 my_vg = cfg.GetVGName()
3564 results = rpc.call_vg_list([oth_node, tgt_node])
3566 raise errors.OpExecError("Can't list volume groups on the nodes")
3567 for node in oth_node, tgt_node:
3568 res = results.get(node, False)
3569 if not res or my_vg not in res:
3570 raise errors.OpExecError("Volume group '%s' not found on %s" %
3572 for dev in instance.disks:
3573 if not dev.iv_name in self.op.disks:
3575 for node in tgt_node, oth_node:
3576 info("checking %s on %s" % (dev.iv_name, node))
3577 cfg.SetDiskID(dev, node)
3578 if not rpc.call_blockdev_find(node, dev):
3579 raise errors.OpExecError("Can't find device %s on node %s" %
3580 (dev.iv_name, node))
3582 # Step: check other node consistency
3583 self.processor.LogStep(2, steps_total, "check peer consistency")
3584 for dev in instance.disks:
3585 if not dev.iv_name in self.op.disks:
3587 info("checking %s consistency on %s" % (dev.iv_name, oth_node))
3588 if not _CheckDiskConsistency(self.cfg, dev, oth_node,
3589 oth_node==instance.primary_node):
3590 raise errors.OpExecError("Peer node (%s) has degraded storage, unsafe"
3591 " to replace disks on this node (%s)" %
3592 (oth_node, tgt_node))
3594 # Step: create new storage
3595 self.processor.LogStep(3, steps_total, "allocate new storage")
3596 for dev in instance.disks:
3597 if not dev.iv_name in self.op.disks:
3600 cfg.SetDiskID(dev, tgt_node)
3601 lv_names = [".%s_%s" % (dev.iv_name, suf) for suf in ["data", "meta"]]
3602 names = _GenerateUniqueNames(cfg, lv_names)
3603 lv_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3604 logical_id=(vgname, names[0]))
3605 lv_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3606 logical_id=(vgname, names[1]))
3607 new_lvs = [lv_data, lv_meta]
3608 old_lvs = dev.children
3609 iv_names[dev.iv_name] = (dev, old_lvs, new_lvs)
3610 info("creating new local storage on %s for %s" %
3611 (tgt_node, dev.iv_name))
3612 # since we *always* want to create this LV, we use the
3613 # _Create...OnPrimary (which forces the creation), even if we
3614 # are talking about the secondary node
3615 for new_lv in new_lvs:
3616 if not _CreateBlockDevOnPrimary(cfg, tgt_node, instance, new_lv,
3617 _GetInstanceInfoText(instance)):
3618 raise errors.OpExecError("Failed to create new LV named '%s' on"
3620 (new_lv.logical_id[1], tgt_node))
3622 # Step: for each lv, detach+rename*2+attach
3623 self.processor.LogStep(4, steps_total, "change drbd configuration")
3624 for dev, old_lvs, new_lvs in iv_names.itervalues():
3625 info("detaching %s drbd from local storage" % dev.iv_name)
3626 if not rpc.call_blockdev_removechildren(tgt_node, dev, old_lvs):
3627 raise errors.OpExecError("Can't detach drbd from local storage on node"
3628 " %s for device %s" % (tgt_node, dev.iv_name))
3630 #cfg.Update(instance)
3632 # ok, we created the new LVs, so now we know we have the needed
3633 # storage; as such, we proceed on the target node to rename
3634 # old_lv to _old, and new_lv to old_lv; note that we rename LVs
3635 # using the assumption than logical_id == physical_id (which in
3636 # turn is the unique_id on that node)
3638 # FIXME(iustin): use a better name for the replaced LVs
3639 temp_suffix = int(time.time())
3640 ren_fn = lambda d, suff: (d.physical_id[0],
3641 d.physical_id[1] + "_replaced-%s" % suff)
3642 # build the rename list based on what LVs exist on the node
3644 for to_ren in old_lvs:
3645 find_res = rpc.call_blockdev_find(tgt_node, to_ren)
3646 if find_res is not None: # device exists
3647 rlist.append((to_ren, ren_fn(to_ren, temp_suffix)))
3649 info("renaming the old LVs on the target node")
3650 if not rpc.call_blockdev_rename(tgt_node, rlist):
3651 raise errors.OpExecError("Can't rename old LVs on node %s" % tgt_node)
3652 # now we rename the new LVs to the old LVs
3653 info("renaming the new LVs on the target node")
3654 rlist = [(new, old.physical_id) for old, new in zip(old_lvs, new_lvs)]
3655 if not rpc.call_blockdev_rename(tgt_node, rlist):
3656 raise errors.OpExecError("Can't rename new LVs on node %s" % tgt_node)
3658 for old, new in zip(old_lvs, new_lvs):
3659 new.logical_id = old.logical_id
3660 cfg.SetDiskID(new, tgt_node)
3662 for disk in old_lvs:
3663 disk.logical_id = ren_fn(disk, temp_suffix)
3664 cfg.SetDiskID(disk, tgt_node)
3666 # now that the new lvs have the old name, we can add them to the device
3667 info("adding new mirror component on %s" % tgt_node)
3668 if not rpc.call_blockdev_addchildren(tgt_node, dev, new_lvs):
3669 for new_lv in new_lvs:
3670 if not rpc.call_blockdev_remove(tgt_node, new_lv):
3671 warning("Can't rollback device %s", "manually cleanup unused"
3673 raise errors.OpExecError("Can't add local storage to drbd")
3675 dev.children = new_lvs
3676 cfg.Update(instance)
3678 # Step: wait for sync
3680 # this can fail as the old devices are degraded and _WaitForSync
3681 # does a combined result over all disks, so we don't check its
3683 self.processor.LogStep(5, steps_total, "sync devices")
3684 _WaitForSync(cfg, instance, unlock=True)
3686 # so check manually all the devices
3687 for name, (dev, old_lvs, new_lvs) in iv_names.iteritems():
3688 cfg.SetDiskID(dev, instance.primary_node)
3689 is_degr = rpc.call_blockdev_find(instance.primary_node, dev)[5]
3691 raise errors.OpExecError("DRBD device %s is degraded!" % name)
3693 # Step: remove old storage
3694 self.processor.LogStep(6, steps_total, "removing old storage")
3695 for name, (dev, old_lvs, new_lvs) in iv_names.iteritems():
3696 info("remove logical volumes for %s" % name)
3698 cfg.SetDiskID(lv, tgt_node)
3699 if not rpc.call_blockdev_remove(tgt_node, lv):
3700 warning("Can't remove old LV", "manually remove unused LVs")
3703 def _ExecD8Secondary(self, feedback_fn):
3704 """Replace the secondary node for drbd8.
3706 The algorithm for replace is quite complicated:
3707 - for all disks of the instance:
3708 - create new LVs on the new node with same names
3709 - shutdown the drbd device on the old secondary
3710 - disconnect the drbd network on the primary
3711 - create the drbd device on the new secondary
3712 - network attach the drbd on the primary, using an artifice:
3713 the drbd code for Attach() will connect to the network if it
3714 finds a device which is connected to the good local disks but
3716 - wait for sync across all devices
3717 - remove all disks from the old secondary
3719 Failures are not very well handled.
3721 instance = self.instance
3723 vgname = self.cfg.GetVGName()
3726 old_node = self.tgt_node
3727 new_node = self.new_node
3728 pri_node = instance.primary_node
3729 for dev in instance.disks:
3731 logger.Info("adding new local storage on %s for %s" %
3732 (new_node, dev.iv_name))
3733 # since we *always* want to create this LV, we use the
3734 # _Create...OnPrimary (which forces the creation), even if we
3735 # are talking about the secondary node
3736 for new_lv in dev.children:
3737 if not _CreateBlockDevOnPrimary(cfg, new_node, instance, new_lv,
3738 _GetInstanceInfoText(instance)):
3739 raise errors.OpExecError("Failed to create new LV named '%s' on"
3741 (new_lv.logical_id[1], new_node))
3743 # create new devices on new_node
3744 new_drbd = objects.Disk(dev_type=constants.LD_DRBD8,
3745 logical_id=(pri_node, new_node,
3747 children=dev.children)
3748 if not _CreateBlockDevOnSecondary(cfg, new_node, instance,
3750 _GetInstanceInfoText(instance)):
3751 raise errors.OpExecError("Failed to create new DRBD on"
3752 " node '%s'" % new_node)
3754 # we have new devices, shutdown the drbd on the old secondary
3755 cfg.SetDiskID(dev, old_node)
3756 if not rpc.call_blockdev_shutdown(old_node, dev):
3757 raise errors.OpExecError("Failed to shutdown DRBD on old node")
3759 # we have new storage, we 'rename' the network on the primary
3760 cfg.SetDiskID(dev, pri_node)
3761 # rename to the ip of the new node
3762 new_uid = list(dev.physical_id)
3763 new_uid[2] = self.remote_node_info.secondary_ip
3764 rlist = [(dev, tuple(new_uid))]
3765 if not rpc.call_blockdev_rename(pri_node, rlist):
3766 raise errors.OpExecError("Can't detach re-attach drbd %s on node"
3767 " %s from %s to %s" %
3768 (dev.iv_name, pri_node, old_node, new_node))
3769 dev.logical_id = (pri_node, new_node, dev.logical_id[2])
3770 cfg.SetDiskID(dev, pri_node)
3771 cfg.Update(instance)
3773 iv_names[dev.iv_name] = (dev, dev.children)
3775 # this can fail as the old devices are degraded and _WaitForSync
3776 # does a combined result over all disks, so we don't check its
3778 logger.Info("Done changing drbd configs, waiting for sync")
3779 _WaitForSync(cfg, instance, unlock=True)
3781 # so check manually all the devices
3782 for name, (dev, old_lvs) in iv_names.iteritems():
3783 cfg.SetDiskID(dev, pri_node)
3784 is_degr = rpc.call_blockdev_find(pri_node, dev)[5]
3786 raise errors.OpExecError("DRBD device %s is degraded!" % name)
3788 for name, (dev, old_lvs) in iv_names.iteritems():
3789 logger.Info("remove logical volumes for %s" % name)
3791 cfg.SetDiskID(lv, old_node)
3792 if not rpc.call_blockdev_remove(old_node, lv):
3793 logger.Error("Can't cleanup child device, skipping. You need to"
3797 def Exec(self, feedback_fn):
3798 """Execute disk replacement.
3800 This dispatches the disk replacement to the appropriate handler.
3803 instance = self.instance
3804 if instance.disk_template == constants.DT_REMOTE_RAID1:
3806 elif instance.disk_template == constants.DT_DRBD8:
3807 if self.op.remote_node is None:
3808 fn = self._ExecD8DiskOnly
3810 fn = self._ExecD8Secondary
3812 raise errors.ProgrammerError("Unhandled disk replacement case")
3813 return fn(feedback_fn)
3816 class LUQueryInstanceData(NoHooksLU):
3817 """Query runtime instance data.
3820 _OP_REQP = ["instances"]
3822 def CheckPrereq(self):
3823 """Check prerequisites.
3825 This only checks the optional instance list against the existing names.
3828 if not isinstance(self.op.instances, list):
3829 raise errors.OpPrereqError("Invalid argument type 'instances'")
3830 if self.op.instances:
3831 self.wanted_instances = []
3832 names = self.op.instances
3834 instance = self.cfg.GetInstanceInfo(self.cfg.ExpandInstanceName(name))
3835 if instance is None:
3836 raise errors.OpPrereqError("No such instance name '%s'" % name)
3837 self.wanted_instances.append(instance)
3839 self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
3840 in self.cfg.GetInstanceList()]
3844 def _ComputeDiskStatus(self, instance, snode, dev):
3845 """Compute block device status.
3848 self.cfg.SetDiskID(dev, instance.primary_node)
3849 dev_pstatus = rpc.call_blockdev_find(instance.primary_node, dev)
3850 if dev.dev_type in constants.LDS_DRBD:
3851 # we change the snode then (otherwise we use the one passed in)
3852 if dev.logical_id[0] == instance.primary_node:
3853 snode = dev.logical_id[1]
3855 snode = dev.logical_id[0]
3858 self.cfg.SetDiskID(dev, snode)
3859 dev_sstatus = rpc.call_blockdev_find(snode, dev)
3864 dev_children = [self._ComputeDiskStatus(instance, snode, child)
3865 for child in dev.children]
3870 "iv_name": dev.iv_name,
3871 "dev_type": dev.dev_type,
3872 "logical_id": dev.logical_id,
3873 "physical_id": dev.physical_id,
3874 "pstatus": dev_pstatus,
3875 "sstatus": dev_sstatus,
3876 "children": dev_children,
3881 def Exec(self, feedback_fn):
3882 """Gather and return data"""
3884 for instance in self.wanted_instances:
3885 remote_info = rpc.call_instance_info(instance.primary_node,
3887 if remote_info and "state" in remote_info:
3890 remote_state = "down"
3891 if instance.status == "down":
3892 config_state = "down"
3896 disks = [self._ComputeDiskStatus(instance, None, device)
3897 for device in instance.disks]
3900 "name": instance.name,
3901 "config_state": config_state,
3902 "run_state": remote_state,
3903 "pnode": instance.primary_node,
3904 "snodes": instance.secondary_nodes,
3906 "memory": instance.memory,
3907 "nics": [(nic.mac, nic.ip, nic.bridge) for nic in instance.nics],
3909 "vcpus": instance.vcpus,
3912 result[instance.name] = idict
3917 class LUSetInstanceParms(LogicalUnit):
3918 """Modifies an instances's parameters.
3921 HPATH = "instance-modify"
3922 HTYPE = constants.HTYPE_INSTANCE
3923 _OP_REQP = ["instance_name"]
3925 def BuildHooksEnv(self):
3928 This runs on the master, primary and secondaries.
3933 args['memory'] = self.mem
3935 args['vcpus'] = self.vcpus
3936 if self.do_ip or self.do_bridge:
3940 ip = self.instance.nics[0].ip
3942 bridge = self.bridge
3944 bridge = self.instance.nics[0].bridge
3945 args['nics'] = [(ip, bridge)]
3946 env = _BuildInstanceHookEnvByObject(self.instance, override=args)
3947 nl = [self.sstore.GetMasterNode(),
3948 self.instance.primary_node] + list(self.instance.secondary_nodes)
3951 def CheckPrereq(self):
3952 """Check prerequisites.
3954 This only checks the instance list against the existing names.
3957 self.mem = getattr(self.op, "mem", None)
3958 self.vcpus = getattr(self.op, "vcpus", None)
3959 self.ip = getattr(self.op, "ip", None)
3960 self.bridge = getattr(self.op, "bridge", None)
3961 if [self.mem, self.vcpus, self.ip, self.bridge].count(None) == 4:
3962 raise errors.OpPrereqError("No changes submitted")
3963 if self.mem is not None:
3965 self.mem = int(self.mem)
3966 except ValueError, err:
3967 raise errors.OpPrereqError("Invalid memory size: %s" % str(err))
3968 if self.vcpus is not None:
3970 self.vcpus = int(self.vcpus)
3971 except ValueError, err:
3972 raise errors.OpPrereqError("Invalid vcpus number: %s" % str(err))
3973 if self.ip is not None:
3975 if self.ip.lower() == "none":
3978 if not utils.IsValidIP(self.ip):
3979 raise errors.OpPrereqError("Invalid IP address '%s'." % self.ip)
3982 self.do_bridge = (self.bridge is not None)
3984 instance = self.cfg.GetInstanceInfo(
3985 self.cfg.ExpandInstanceName(self.op.instance_name))
3986 if instance is None:
3987 raise errors.OpPrereqError("No such instance name '%s'" %
3988 self.op.instance_name)
3989 self.op.instance_name = instance.name
3990 self.instance = instance
3993 def Exec(self, feedback_fn):
3994 """Modifies an instance.
3996 All parameters take effect only at the next restart of the instance.
3999 instance = self.instance
4001 instance.memory = self.mem
4002 result.append(("mem", self.mem))
4004 instance.vcpus = self.vcpus
4005 result.append(("vcpus", self.vcpus))
4007 instance.nics[0].ip = self.ip
4008 result.append(("ip", self.ip))
4010 instance.nics[0].bridge = self.bridge
4011 result.append(("bridge", self.bridge))
4013 self.cfg.AddInstance(instance)
4018 class LUQueryExports(NoHooksLU):
4019 """Query the exports list
4024 def CheckPrereq(self):
4025 """Check that the nodelist contains only existing nodes.
4028 self.nodes = _GetWantedNodes(self, getattr(self.op, "nodes", None))
4030 def Exec(self, feedback_fn):
4031 """Compute the list of all the exported system images.
4034 a dictionary with the structure node->(export-list)
4035 where export-list is a list of the instances exported on
4039 return rpc.call_export_list(self.nodes)
4042 class LUExportInstance(LogicalUnit):
4043 """Export an instance to an image in the cluster.
4046 HPATH = "instance-export"
4047 HTYPE = constants.HTYPE_INSTANCE
4048 _OP_REQP = ["instance_name", "target_node", "shutdown"]
4050 def BuildHooksEnv(self):
4053 This will run on the master, primary node and target node.
4057 "EXPORT_NODE": self.op.target_node,
4058 "EXPORT_DO_SHUTDOWN": self.op.shutdown,
4060 env.update(_BuildInstanceHookEnvByObject(self.instance))
4061 nl = [self.sstore.GetMasterNode(), self.instance.primary_node,
4062 self.op.target_node]
4065 def CheckPrereq(self):
4066 """Check prerequisites.
4068 This checks that the instance name is a valid one.
4071 instance_name = self.cfg.ExpandInstanceName(self.op.instance_name)
4072 self.instance = self.cfg.GetInstanceInfo(instance_name)
4073 if self.instance is None:
4074 raise errors.OpPrereqError("Instance '%s' not found" %
4075 self.op.instance_name)
4078 dst_node_short = self.cfg.ExpandNodeName(self.op.target_node)
4079 self.dst_node = self.cfg.GetNodeInfo(dst_node_short)
4081 if self.dst_node is None:
4082 raise errors.OpPrereqError("Destination node '%s' is unknown." %
4083 self.op.target_node)
4084 self.op.target_node = self.dst_node.name
4086 def Exec(self, feedback_fn):
4087 """Export an instance to an image in the cluster.
4090 instance = self.instance
4091 dst_node = self.dst_node
4092 src_node = instance.primary_node
4093 # shutdown the instance, unless requested not to do so
4094 if self.op.shutdown:
4095 op = opcodes.OpShutdownInstance(instance_name=instance.name)
4096 self.processor.ChainOpCode(op)
4098 vgname = self.cfg.GetVGName()
4103 for disk in instance.disks:
4104 if disk.iv_name == "sda":
4105 # new_dev_name will be a snapshot of an lvm leaf of the one we passed
4106 new_dev_name = rpc.call_blockdev_snapshot(src_node, disk)
4108 if not new_dev_name:
4109 logger.Error("could not snapshot block device %s on node %s" %
4110 (disk.logical_id[1], src_node))
4112 new_dev = objects.Disk(dev_type=constants.LD_LV, size=disk.size,
4113 logical_id=(vgname, new_dev_name),
4114 physical_id=(vgname, new_dev_name),
4115 iv_name=disk.iv_name)
4116 snap_disks.append(new_dev)
4119 if self.op.shutdown:
4120 op = opcodes.OpStartupInstance(instance_name=instance.name,
4122 self.processor.ChainOpCode(op)
4124 # TODO: check for size
4126 for dev in snap_disks:
4127 if not rpc.call_snapshot_export(src_node, dev, dst_node.name,
4129 logger.Error("could not export block device %s from node"
4131 (dev.logical_id[1], src_node, dst_node.name))
4132 if not rpc.call_blockdev_remove(src_node, dev):
4133 logger.Error("could not remove snapshot block device %s from"
4134 " node %s" % (dev.logical_id[1], src_node))
4136 if not rpc.call_finalize_export(dst_node.name, instance, snap_disks):
4137 logger.Error("could not finalize export for instance %s on node %s" %
4138 (instance.name, dst_node.name))
4140 nodelist = self.cfg.GetNodeList()
4141 nodelist.remove(dst_node.name)
4143 # on one-node clusters nodelist will be empty after the removal
4144 # if we proceed the backup would be removed because OpQueryExports
4145 # substitutes an empty list with the full cluster node list.
4147 op = opcodes.OpQueryExports(nodes=nodelist)
4148 exportlist = self.processor.ChainOpCode(op)
4149 for node in exportlist:
4150 if instance.name in exportlist[node]:
4151 if not rpc.call_export_remove(node, instance.name):
4152 logger.Error("could not remove older export for instance %s"
4153 " on node %s" % (instance.name, node))
4156 class TagsLU(NoHooksLU):
4159 This is an abstract class which is the parent of all the other tags LUs.
4162 def CheckPrereq(self):
4163 """Check prerequisites.
4166 if self.op.kind == constants.TAG_CLUSTER:
4167 self.target = self.cfg.GetClusterInfo()
4168 elif self.op.kind == constants.TAG_NODE:
4169 name = self.cfg.ExpandNodeName(self.op.name)
4171 raise errors.OpPrereqError("Invalid node name (%s)" %
4174 self.target = self.cfg.GetNodeInfo(name)
4175 elif self.op.kind == constants.TAG_INSTANCE:
4176 name = self.cfg.ExpandInstanceName(self.op.name)
4178 raise errors.OpPrereqError("Invalid instance name (%s)" %
4181 self.target = self.cfg.GetInstanceInfo(name)
4183 raise errors.OpPrereqError("Wrong tag type requested (%s)" %
4187 class LUGetTags(TagsLU):
4188 """Returns the tags of a given object.
4191 _OP_REQP = ["kind", "name"]
4193 def Exec(self, feedback_fn):
4194 """Returns the tag list.
4197 return self.target.GetTags()
4200 class LUSearchTags(NoHooksLU):
4201 """Searches the tags for a given pattern.
4204 _OP_REQP = ["pattern"]
4206 def CheckPrereq(self):
4207 """Check prerequisites.
4209 This checks the pattern passed for validity by compiling it.
4213 self.re = re.compile(self.op.pattern)
4214 except re.error, err:
4215 raise errors.OpPrereqError("Invalid search pattern '%s': %s" %
4216 (self.op.pattern, err))
4218 def Exec(self, feedback_fn):
4219 """Returns the tag list.
4223 tgts = [("/cluster", cfg.GetClusterInfo())]
4224 ilist = [cfg.GetInstanceInfo(name) for name in cfg.GetInstanceList()]
4225 tgts.extend([("/instances/%s" % i.name, i) for i in ilist])
4226 nlist = [cfg.GetNodeInfo(name) for name in cfg.GetNodeList()]
4227 tgts.extend([("/nodes/%s" % n.name, n) for n in nlist])
4229 for path, target in tgts:
4230 for tag in target.GetTags():
4231 if self.re.search(tag):
4232 results.append((path, tag))
4236 class LUAddTags(TagsLU):
4237 """Sets a tag on a given object.
4240 _OP_REQP = ["kind", "name", "tags"]
4242 def CheckPrereq(self):
4243 """Check prerequisites.
4245 This checks the type and length of the tag name and value.
4248 TagsLU.CheckPrereq(self)
4249 for tag in self.op.tags:
4250 objects.TaggableObject.ValidateTag(tag)
4252 def Exec(self, feedback_fn):
4257 for tag in self.op.tags:
4258 self.target.AddTag(tag)
4259 except errors.TagError, err:
4260 raise errors.OpExecError("Error while setting tag: %s" % str(err))
4262 self.cfg.Update(self.target)
4263 except errors.ConfigurationError:
4264 raise errors.OpRetryError("There has been a modification to the"
4265 " config file and the operation has been"
4266 " aborted. Please retry.")
4269 class LUDelTags(TagsLU):
4270 """Delete a list of tags from a given object.
4273 _OP_REQP = ["kind", "name", "tags"]
4275 def CheckPrereq(self):
4276 """Check prerequisites.
4278 This checks that we have the given tag.
4281 TagsLU.CheckPrereq(self)
4282 for tag in self.op.tags:
4283 objects.TaggableObject.ValidateTag(tag)
4284 del_tags = frozenset(self.op.tags)
4285 cur_tags = self.target.GetTags()
4286 if not del_tags <= cur_tags:
4287 diff_tags = del_tags - cur_tags
4288 diff_names = ["'%s'" % tag for tag in diff_tags]
4290 raise errors.OpPrereqError("Tag(s) %s not found" %
4291 (",".join(diff_names)))
4293 def Exec(self, feedback_fn):
4294 """Remove the tag from the object.
4297 for tag in self.op.tags:
4298 self.target.RemoveTag(tag)
4300 self.cfg.Update(self.target)
4301 except errors.ConfigurationError:
4302 raise errors.OpRetryError("There has been a modification to the"
4303 " config file and the operation has been"
4304 " aborted. Please retry.")