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
35 from ganeti import rpc
36 from ganeti import ssh
37 from ganeti import logger
38 from ganeti import utils
39 from ganeti import errors
40 from ganeti import hypervisor
41 from ganeti import config
42 from ganeti import constants
43 from ganeti import objects
44 from ganeti import opcodes
45 from ganeti import ssconf
47 class LogicalUnit(object):
48 """Logical Unit base class.
50 Subclasses must follow these rules:
51 - implement CheckPrereq which also fills in the opcode instance
52 with all the fields (even if as None)
54 - implement BuildHooksEnv
55 - redefine HPATH and HTYPE
56 - optionally redefine their run requirements (REQ_CLUSTER,
57 REQ_MASTER); note that all commands require root permissions
66 def __init__(self, processor, op, cfg, sstore):
67 """Constructor for LogicalUnit.
69 This needs to be overriden in derived classes in order to check op
73 self.processor = processor
77 for attr_name in self._OP_REQP:
78 attr_val = getattr(op, attr_name, None)
80 raise errors.OpPrereqError("Required parameter '%s' missing" %
83 if not cfg.IsCluster():
84 raise errors.OpPrereqError("Cluster not initialized yet,"
85 " use 'gnt-cluster init' first.")
87 master = sstore.GetMasterNode()
88 if master != utils.HostInfo().name:
89 raise errors.OpPrereqError("Commands must be run on the master"
92 def CheckPrereq(self):
93 """Check prerequisites for this LU.
95 This method should check that the prerequisites for the execution
96 of this LU are fulfilled. It can do internode communication, but
97 it should be idempotent - no cluster or system changes are
100 The method should raise errors.OpPrereqError in case something is
101 not fulfilled. Its return value is ignored.
103 This method should also update all the parameters of the opcode to
104 their canonical form; e.g. a short node name must be fully
105 expanded after this method has successfully completed (so that
106 hooks, logging, etc. work correctly).
109 raise NotImplementedError
111 def Exec(self, feedback_fn):
114 This method should implement the actual work. It should raise
115 errors.OpExecError for failures that are somewhat dealt with in
119 raise NotImplementedError
121 def BuildHooksEnv(self):
122 """Build hooks environment for this LU.
124 This method should return a three-node tuple consisting of: a dict
125 containing the environment that will be used for running the
126 specific hook for this LU, a list of node names on which the hook
127 should run before the execution, and a list of node names on which
128 the hook should run after the execution.
130 The keys of the dict must not have 'GANETI_' prefixed as this will
131 be handled in the hooks runner. Also note additional keys will be
132 added by the hooks runner. If the LU doesn't define any
133 environment, an empty dict (and not None) should be returned.
135 As for the node lists, the master should not be included in the
136 them, as it will be added by the hooks runner in case this LU
137 requires a cluster to run on (otherwise we don't have a node
138 list). No nodes should be returned as an empty list (and not
141 Note that if the HPATH for a LU class is None, this function will
145 raise NotImplementedError
148 class NoHooksLU(LogicalUnit):
149 """Simple LU which runs no hooks.
151 This LU is intended as a parent for other LogicalUnits which will
152 run no hooks, in order to reduce duplicate code.
158 def BuildHooksEnv(self):
161 This is a no-op, since we don't run hooks.
167 def _GetWantedNodes(lu, nodes):
168 """Returns list of checked and expanded node names.
171 nodes: List of nodes (strings) or None for all
174 if not isinstance(nodes, list):
175 raise errors.OpPrereqError("Invalid argument type 'nodes'")
181 node = lu.cfg.ExpandNodeName(name)
183 raise errors.OpPrereqError("No such node name '%s'" % name)
187 wanted = lu.cfg.GetNodeList()
188 return utils.NiceSort(wanted)
191 def _GetWantedInstances(lu, instances):
192 """Returns list of checked and expanded instance names.
195 instances: List of instances (strings) or None for all
198 if not isinstance(instances, list):
199 raise errors.OpPrereqError("Invalid argument type 'instances'")
204 for name in instances:
205 instance = lu.cfg.ExpandInstanceName(name)
207 raise errors.OpPrereqError("No such instance name '%s'" % name)
208 wanted.append(instance)
211 wanted = lu.cfg.GetInstanceList()
212 return utils.NiceSort(wanted)
215 def _CheckOutputFields(static, dynamic, selected):
216 """Checks whether all selected fields are valid.
219 static: Static fields
220 dynamic: Dynamic fields
223 static_fields = frozenset(static)
224 dynamic_fields = frozenset(dynamic)
226 all_fields = static_fields | dynamic_fields
228 if not all_fields.issuperset(selected):
229 raise errors.OpPrereqError("Unknown output fields selected: %s"
230 % ",".join(frozenset(selected).
231 difference(all_fields)))
234 def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
235 memory, vcpus, nics):
236 """Builds instance related env variables for hooks from single variables.
239 secondary_nodes: List of secondary nodes as strings
243 "INSTANCE_NAME": name,
244 "INSTANCE_PRIMARY": primary_node,
245 "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
246 "INSTANCE_OS_TYPE": os_type,
247 "INSTANCE_STATUS": status,
248 "INSTANCE_MEMORY": memory,
249 "INSTANCE_VCPUS": vcpus,
253 nic_count = len(nics)
254 for idx, (ip, bridge) in enumerate(nics):
257 env["INSTANCE_NIC%d_IP" % idx] = ip
258 env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
262 env["INSTANCE_NIC_COUNT"] = nic_count
267 def _BuildInstanceHookEnvByObject(instance, override=None):
268 """Builds instance related env variables for hooks from an object.
271 instance: objects.Instance object of instance
272 override: dict of values to override
275 'name': instance.name,
276 'primary_node': instance.primary_node,
277 'secondary_nodes': instance.secondary_nodes,
278 'os_type': instance.os,
279 'status': instance.os,
280 'memory': instance.memory,
281 'vcpus': instance.vcpus,
282 'nics': [(nic.ip, nic.bridge) for nic in instance.nics],
285 args.update(override)
286 return _BuildInstanceHookEnv(**args)
289 def _UpdateEtcHosts(fullnode, ip):
290 """Ensure a node has a correct entry in /etc/hosts.
293 fullnode - Fully qualified domain name of host. (str)
294 ip - IPv4 address of host (str)
297 node = fullnode.split(".", 1)[0]
299 f = open('/etc/hosts', 'r+')
308 rawline = f.readline()
314 line = rawline.split('\n')[0]
317 line = line.split('#')[0]
320 # Entire line was comment, skip
321 save_lines.append(rawline)
324 fields = line.split()
328 for spec in [ ip, fullnode, node ]:
329 if spec not in fields:
336 save_lines.append(rawline)
339 if havesome and not haveall:
340 # Line (old, or manual?) which is missing some. Remove.
344 save_lines.append(rawline)
347 add_lines.append('%s\t%s %s\n' % (ip, fullnode, node))
351 save_lines = save_lines + add_lines
353 # We removed a line, write a new file and replace old.
354 fd, tmpname = tempfile.mkstemp('tmp', 'hosts_', '/etc')
355 newfile = os.fdopen(fd, 'w')
356 newfile.write(''.join(save_lines))
358 os.rename(tmpname, '/etc/hosts')
361 # Simply appending a new line will do the trick.
363 for add in add_lines:
369 def _UpdateKnownHosts(fullnode, ip, pubkey):
370 """Ensure a node has a correct known_hosts entry.
373 fullnode - Fully qualified domain name of host. (str)
374 ip - IPv4 address of host (str)
375 pubkey - the public key of the cluster
378 if os.path.exists(constants.SSH_KNOWN_HOSTS_FILE):
379 f = open(constants.SSH_KNOWN_HOSTS_FILE, 'r+')
381 f = open(constants.SSH_KNOWN_HOSTS_FILE, 'w+')
390 rawline = f.readline()
391 logger.Debug('read %s' % (repr(rawline),))
397 line = rawline.split('\n')[0]
399 parts = line.split(' ')
400 fields = parts[0].split(',')
405 for spec in [ ip, fullnode ]:
406 if spec not in fields:
411 logger.Debug("key, pubkey = %s." % (repr((key, pubkey)),))
412 if haveall and key == pubkey:
414 save_lines.append(rawline)
415 logger.Debug("Keeping known_hosts '%s'." % (repr(rawline),))
418 if havesome and (not haveall or key != pubkey):
420 logger.Debug("Discarding known_hosts '%s'." % (repr(rawline),))
423 save_lines.append(rawline)
426 add_lines.append('%s,%s ssh-rsa %s\n' % (fullnode, ip, pubkey))
427 logger.Debug("Adding known_hosts '%s'." % (repr(add_lines[-1]),))
430 save_lines = save_lines + add_lines
432 # Write a new file and replace old.
433 fd, tmpname = tempfile.mkstemp('.tmp', 'known_hosts.',
435 newfile = os.fdopen(fd, 'w')
437 newfile.write(''.join(save_lines))
440 logger.Debug("Wrote new known_hosts.")
441 os.rename(tmpname, constants.SSH_KNOWN_HOSTS_FILE)
444 # Simply appending a new line will do the trick.
446 for add in add_lines:
452 def _HasValidVG(vglist, vgname):
453 """Checks if the volume group list is valid.
455 A non-None return value means there's an error, and the return value
456 is the error message.
459 vgsize = vglist.get(vgname, None)
461 return "volume group '%s' missing" % vgname
463 return ("volume group '%s' too small (20480MiB required, %dMib found)" %
468 def _InitSSHSetup(node):
469 """Setup the SSH configuration for the cluster.
472 This generates a dsa keypair for root, adds the pub key to the
473 permitted hosts and adds the hostkey to its own known hosts.
476 node: the name of this host as a fqdn
479 priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
481 for name in priv_key, pub_key:
482 if os.path.exists(name):
483 utils.CreateBackup(name)
484 utils.RemoveFile(name)
486 result = utils.RunCmd(["ssh-keygen", "-t", "dsa",
490 raise errors.OpExecError("Could not generate ssh keypair, error %s" %
493 f = open(pub_key, 'r')
495 utils.AddAuthorizedKey(auth_keys, f.read(8192))
500 def _InitGanetiServerSetup(ss):
501 """Setup the necessary configuration for the initial node daemon.
503 This creates the nodepass file containing the shared password for
504 the cluster and also generates the SSL certificate.
507 # Create pseudo random password
508 randpass = sha.new(os.urandom(64)).hexdigest()
509 # and write it into sstore
510 ss.SetKey(ss.SS_NODED_PASS, randpass)
512 result = utils.RunCmd(["openssl", "req", "-new", "-newkey", "rsa:1024",
513 "-days", str(365*5), "-nodes", "-x509",
514 "-keyout", constants.SSL_CERT_FILE,
515 "-out", constants.SSL_CERT_FILE, "-batch"])
517 raise errors.OpExecError("could not generate server ssl cert, command"
518 " %s had exitcode %s and error message %s" %
519 (result.cmd, result.exit_code, result.output))
521 os.chmod(constants.SSL_CERT_FILE, 0400)
523 result = utils.RunCmd([constants.NODE_INITD_SCRIPT, "restart"])
526 raise errors.OpExecError("Could not start the node daemon, command %s"
527 " had exitcode %s and error %s" %
528 (result.cmd, result.exit_code, result.output))
531 def _CheckInstanceBridgesExist(instance):
532 """Check that the brigdes needed by an instance exist.
535 # check bridges existance
536 brlist = [nic.bridge for nic in instance.nics]
537 if not rpc.call_bridges_exist(instance.primary_node, brlist):
538 raise errors.OpPrereqError("one or more target bridges %s does not"
539 " exist on destination node '%s'" %
540 (brlist, instance.primary_node))
543 class LUInitCluster(LogicalUnit):
544 """Initialise the cluster.
547 HPATH = "cluster-init"
548 HTYPE = constants.HTYPE_CLUSTER
549 _OP_REQP = ["cluster_name", "hypervisor_type", "vg_name", "mac_prefix",
550 "def_bridge", "master_netdev"]
553 def BuildHooksEnv(self):
556 Notes: Since we don't require a cluster, we must manually add
557 ourselves in the post-run node list.
560 env = {"OP_TARGET": self.op.cluster_name}
561 return env, [], [self.hostname.name]
563 def CheckPrereq(self):
564 """Verify that the passed name is a valid one.
567 if config.ConfigWriter.IsCluster():
568 raise errors.OpPrereqError("Cluster is already initialised")
570 self.hostname = hostname = utils.HostInfo()
572 if hostname.ip.startswith("127."):
573 raise errors.OpPrereqError("This host's IP resolves to the private"
574 " range (%s). Please fix DNS or /etc/hosts." %
577 self.clustername = clustername = utils.HostInfo(self.op.cluster_name)
579 if not utils.TcpPing(constants.LOCALHOST_IP_ADDRESS, hostname.ip,
580 constants.DEFAULT_NODED_PORT):
581 raise errors.OpPrereqError("Inconsistency: this host's name resolves"
582 " to %s,\nbut this ip address does not"
583 " belong to this host."
584 " Aborting." % hostname.ip)
586 secondary_ip = getattr(self.op, "secondary_ip", None)
587 if secondary_ip and not utils.IsValidIP(secondary_ip):
588 raise errors.OpPrereqError("Invalid secondary ip given")
590 secondary_ip != hostname.ip and
591 (not utils.TcpPing(constants.LOCALHOST_IP_ADDRESS, secondary_ip,
592 constants.DEFAULT_NODED_PORT))):
593 raise errors.OpPrereqError("You gave %s as secondary IP,\n"
594 "but it does not belong to this host." %
596 self.secondary_ip = secondary_ip
598 # checks presence of the volume group given
599 vgstatus = _HasValidVG(utils.ListVolumeGroups(), self.op.vg_name)
602 raise errors.OpPrereqError("Error: %s" % vgstatus)
604 if not re.match("^[0-9a-z]{2}:[0-9a-z]{2}:[0-9a-z]{2}$",
606 raise errors.OpPrereqError("Invalid mac prefix given '%s'" %
609 if self.op.hypervisor_type not in hypervisor.VALID_HTYPES:
610 raise errors.OpPrereqError("Invalid hypervisor type given '%s'" %
611 self.op.hypervisor_type)
613 result = utils.RunCmd(["ip", "link", "show", "dev", self.op.master_netdev])
615 raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
616 (self.op.master_netdev,
617 result.output.strip()))
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 master = self.sstore.GetMasterNode()
864 vg_name = self.cfg.GetVGName()
865 nodelist = utils.NiceSort(self.cfg.GetNodeList())
866 instancelist = utils.NiceSort(self.cfg.GetInstanceList())
870 # FIXME: verify OS list
872 file_names = list(self.sstore.GetFileList())
873 file_names.append(constants.SSL_CERT_FILE)
874 file_names.append(constants.CLUSTER_CONF_FILE)
875 local_checksums = utils.FingerprintFiles(file_names)
877 feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
878 all_volumeinfo = rpc.call_volume_list(nodelist, vg_name)
879 all_instanceinfo = rpc.call_instance_list(nodelist)
880 all_vglist = rpc.call_vg_list(nodelist)
881 node_verify_param = {
882 'filelist': file_names,
883 'nodelist': nodelist,
886 all_nvinfo = rpc.call_node_verify(nodelist, node_verify_param)
887 all_rversion = rpc.call_version(nodelist)
889 for node in nodelist:
890 feedback_fn("* Verifying node %s" % node)
891 result = self._VerifyNode(node, file_names, local_checksums,
892 all_vglist[node], all_nvinfo[node],
893 all_rversion[node], feedback_fn)
897 volumeinfo = all_volumeinfo[node]
899 if type(volumeinfo) != dict:
900 feedback_fn(" - ERROR: connection to %s failed" % (node,))
904 node_volume[node] = volumeinfo
907 nodeinstance = all_instanceinfo[node]
908 if type(nodeinstance) != list:
909 feedback_fn(" - ERROR: connection to %s failed" % (node,))
913 node_instance[node] = nodeinstance
917 for instance in instancelist:
918 feedback_fn("* Verifying instance %s" % instance)
919 result = self._VerifyInstance(instance, node_volume, node_instance,
923 inst_config = self.cfg.GetInstanceInfo(instance)
925 inst_config.MapLVsByNode(node_vol_should)
927 feedback_fn("* Verifying orphan volumes")
928 result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
932 feedback_fn("* Verifying remaining instances")
933 result = self._VerifyOrphanInstances(instancelist, node_instance,
940 class LURenameCluster(LogicalUnit):
941 """Rename the cluster.
944 HPATH = "cluster-rename"
945 HTYPE = constants.HTYPE_CLUSTER
948 def BuildHooksEnv(self):
953 "OP_TARGET": self.op.sstore.GetClusterName(),
954 "NEW_NAME": self.op.name,
956 mn = self.sstore.GetMasterNode()
957 return env, [mn], [mn]
959 def CheckPrereq(self):
960 """Verify that the passed name is a valid one.
963 hostname = utils.HostInfo(self.op.name)
965 new_name = hostname.name
966 self.ip = new_ip = hostname.ip
967 old_name = self.sstore.GetClusterName()
968 old_ip = self.sstore.GetMasterIP()
969 if new_name == old_name and new_ip == old_ip:
970 raise errors.OpPrereqError("Neither the name nor the IP address of the"
971 " cluster has changed")
973 result = utils.RunCmd(["fping", "-q", new_ip])
974 if not result.failed:
975 raise errors.OpPrereqError("The given cluster IP address (%s) is"
976 " reachable on the network. Aborting." %
979 self.op.name = new_name
981 def Exec(self, feedback_fn):
982 """Rename the cluster.
985 clustername = self.op.name
989 # shutdown the master IP
990 master = ss.GetMasterNode()
991 if not rpc.call_node_stop_master(master):
992 raise errors.OpExecError("Could not disable the master role")
996 ss.SetKey(ss.SS_MASTER_IP, ip)
997 ss.SetKey(ss.SS_CLUSTER_NAME, clustername)
999 # Distribute updated ss config to all nodes
1000 myself = self.cfg.GetNodeInfo(master)
1001 dist_nodes = self.cfg.GetNodeList()
1002 if myself.name in dist_nodes:
1003 dist_nodes.remove(myself.name)
1005 logger.Debug("Copying updated ssconf data to all nodes")
1006 for keyname in [ss.SS_CLUSTER_NAME, ss.SS_MASTER_IP]:
1007 fname = ss.KeyToFilename(keyname)
1008 result = rpc.call_upload_file(dist_nodes, fname)
1009 for to_node in dist_nodes:
1010 if not result[to_node]:
1011 logger.Error("copy of file %s to node %s failed" %
1014 if not rpc.call_node_start_master(master):
1015 logger.Error("Could not re-enable the master role on the master,\n"
1016 "please restart manually.")
1019 def _WaitForSync(cfgw, instance, oneshot=False, unlock=False):
1020 """Sleep and poll for an instance's disk to sync.
1023 if not instance.disks:
1027 logger.ToStdout("Waiting for instance %s to sync disks." % instance.name)
1029 node = instance.primary_node
1031 for dev in instance.disks:
1032 cfgw.SetDiskID(dev, node)
1038 cumul_degraded = False
1039 rstats = rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1041 logger.ToStderr("Can't get any data from node %s" % node)
1044 raise errors.RemoteError("Can't contact node %s for mirror data,"
1045 " aborting." % node)
1049 for i in range(len(rstats)):
1052 logger.ToStderr("Can't compute data for node %s/%s" %
1053 (node, instance.disks[i].iv_name))
1055 perc_done, est_time, is_degraded = mstat
1056 cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1057 if perc_done is not None:
1059 if est_time is not None:
1060 rem_time = "%d estimated seconds remaining" % est_time
1063 rem_time = "no time estimate"
1064 logger.ToStdout("- device %s: %5.2f%% done, %s" %
1065 (instance.disks[i].iv_name, perc_done, rem_time))
1072 time.sleep(min(60, max_time))
1078 logger.ToStdout("Instance %s's disks are in sync." % instance.name)
1079 return not cumul_degraded
1082 def _CheckDiskConsistency(cfgw, dev, node, on_primary):
1083 """Check that mirrors are not degraded.
1086 cfgw.SetDiskID(dev, node)
1089 if on_primary or dev.AssembleOnSecondary():
1090 rstats = rpc.call_blockdev_find(node, dev)
1092 logger.ToStderr("Can't get any data from node %s" % node)
1095 result = result and (not rstats[5])
1097 for child in dev.children:
1098 result = result and _CheckDiskConsistency(cfgw, child, node, on_primary)
1103 class LUDiagnoseOS(NoHooksLU):
1104 """Logical unit for OS diagnose/query.
1109 def CheckPrereq(self):
1110 """Check prerequisites.
1112 This always succeeds, since this is a pure query LU.
1117 def Exec(self, feedback_fn):
1118 """Compute the list of OSes.
1121 node_list = self.cfg.GetNodeList()
1122 node_data = rpc.call_os_diagnose(node_list)
1123 if node_data == False:
1124 raise errors.OpExecError("Can't gather the list of OSes")
1128 class LURemoveNode(LogicalUnit):
1129 """Logical unit for removing a node.
1132 HPATH = "node-remove"
1133 HTYPE = constants.HTYPE_NODE
1134 _OP_REQP = ["node_name"]
1136 def BuildHooksEnv(self):
1139 This doesn't run on the target node in the pre phase as a failed
1140 node would not allows itself to run.
1144 "OP_TARGET": self.op.node_name,
1145 "NODE_NAME": self.op.node_name,
1147 all_nodes = self.cfg.GetNodeList()
1148 all_nodes.remove(self.op.node_name)
1149 return env, all_nodes, all_nodes
1151 def CheckPrereq(self):
1152 """Check prerequisites.
1155 - the node exists in the configuration
1156 - it does not have primary or secondary instances
1157 - it's not the master
1159 Any errors are signalled by raising errors.OpPrereqError.
1162 node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1164 raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1166 instance_list = self.cfg.GetInstanceList()
1168 masternode = self.sstore.GetMasterNode()
1169 if node.name == masternode:
1170 raise errors.OpPrereqError("Node is the master node,"
1171 " you need to failover first.")
1173 for instance_name in instance_list:
1174 instance = self.cfg.GetInstanceInfo(instance_name)
1175 if node.name == instance.primary_node:
1176 raise errors.OpPrereqError("Instance %s still running on the node,"
1177 " please remove first." % instance_name)
1178 if node.name in instance.secondary_nodes:
1179 raise errors.OpPrereqError("Instance %s has node as a secondary,"
1180 " please remove first." % instance_name)
1181 self.op.node_name = node.name
1184 def Exec(self, feedback_fn):
1185 """Removes the node from the cluster.
1189 logger.Info("stopping the node daemon and removing configs from node %s" %
1192 rpc.call_node_leave_cluster(node.name)
1194 ssh.SSHCall(node.name, 'root', "%s stop" % constants.NODE_INITD_SCRIPT)
1196 logger.Info("Removing node %s from config" % node.name)
1198 self.cfg.RemoveNode(node.name)
1201 class LUQueryNodes(NoHooksLU):
1202 """Logical unit for querying nodes.
1205 _OP_REQP = ["output_fields", "names"]
1207 def CheckPrereq(self):
1208 """Check prerequisites.
1210 This checks that the fields required are valid output fields.
1213 self.dynamic_fields = frozenset(["dtotal", "dfree",
1214 "mtotal", "mnode", "mfree",
1217 _CheckOutputFields(static=["name", "pinst_cnt", "sinst_cnt",
1218 "pinst_list", "sinst_list",
1220 dynamic=self.dynamic_fields,
1221 selected=self.op.output_fields)
1223 self.wanted = _GetWantedNodes(self, self.op.names)
1225 def Exec(self, feedback_fn):
1226 """Computes the list of nodes and their attributes.
1229 nodenames = self.wanted
1230 nodelist = [self.cfg.GetNodeInfo(name) for name in nodenames]
1232 # begin data gathering
1234 if self.dynamic_fields.intersection(self.op.output_fields):
1236 node_data = rpc.call_node_info(nodenames, self.cfg.GetVGName())
1237 for name in nodenames:
1238 nodeinfo = node_data.get(name, None)
1241 "mtotal": utils.TryConvert(int, nodeinfo['memory_total']),
1242 "mnode": utils.TryConvert(int, nodeinfo['memory_dom0']),
1243 "mfree": utils.TryConvert(int, nodeinfo['memory_free']),
1244 "dtotal": utils.TryConvert(int, nodeinfo['vg_size']),
1245 "dfree": utils.TryConvert(int, nodeinfo['vg_free']),
1246 "bootid": nodeinfo['bootid'],
1249 live_data[name] = {}
1251 live_data = dict.fromkeys(nodenames, {})
1253 node_to_primary = dict([(name, set()) for name in nodenames])
1254 node_to_secondary = dict([(name, set()) for name in nodenames])
1256 inst_fields = frozenset(("pinst_cnt", "pinst_list",
1257 "sinst_cnt", "sinst_list"))
1258 if inst_fields & frozenset(self.op.output_fields):
1259 instancelist = self.cfg.GetInstanceList()
1261 for instance_name in instancelist:
1262 inst = self.cfg.GetInstanceInfo(instance_name)
1263 if inst.primary_node in node_to_primary:
1264 node_to_primary[inst.primary_node].add(inst.name)
1265 for secnode in inst.secondary_nodes:
1266 if secnode in node_to_secondary:
1267 node_to_secondary[secnode].add(inst.name)
1269 # end data gathering
1272 for node in nodelist:
1274 for field in self.op.output_fields:
1277 elif field == "pinst_list":
1278 val = list(node_to_primary[node.name])
1279 elif field == "sinst_list":
1280 val = list(node_to_secondary[node.name])
1281 elif field == "pinst_cnt":
1282 val = len(node_to_primary[node.name])
1283 elif field == "sinst_cnt":
1284 val = len(node_to_secondary[node.name])
1285 elif field == "pip":
1286 val = node.primary_ip
1287 elif field == "sip":
1288 val = node.secondary_ip
1289 elif field in self.dynamic_fields:
1290 val = live_data[node.name].get(field, None)
1292 raise errors.ParameterError(field)
1293 node_output.append(val)
1294 output.append(node_output)
1299 class LUQueryNodeVolumes(NoHooksLU):
1300 """Logical unit for getting volumes on node(s).
1303 _OP_REQP = ["nodes", "output_fields"]
1305 def CheckPrereq(self):
1306 """Check prerequisites.
1308 This checks that the fields required are valid output fields.
1311 self.nodes = _GetWantedNodes(self, self.op.nodes)
1313 _CheckOutputFields(static=["node"],
1314 dynamic=["phys", "vg", "name", "size", "instance"],
1315 selected=self.op.output_fields)
1318 def Exec(self, feedback_fn):
1319 """Computes the list of nodes and their attributes.
1322 nodenames = self.nodes
1323 volumes = rpc.call_node_volumes(nodenames)
1325 ilist = [self.cfg.GetInstanceInfo(iname) for iname
1326 in self.cfg.GetInstanceList()]
1328 lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1331 for node in nodenames:
1332 if node not in volumes or not volumes[node]:
1335 node_vols = volumes[node][:]
1336 node_vols.sort(key=lambda vol: vol['dev'])
1338 for vol in node_vols:
1340 for field in self.op.output_fields:
1343 elif field == "phys":
1347 elif field == "name":
1349 elif field == "size":
1350 val = int(float(vol['size']))
1351 elif field == "instance":
1353 if node not in lv_by_node[inst]:
1355 if vol['name'] in lv_by_node[inst][node]:
1361 raise errors.ParameterError(field)
1362 node_output.append(str(val))
1364 output.append(node_output)
1369 class LUAddNode(LogicalUnit):
1370 """Logical unit for adding node to the cluster.
1374 HTYPE = constants.HTYPE_NODE
1375 _OP_REQP = ["node_name"]
1377 def BuildHooksEnv(self):
1380 This will run on all nodes before, and on all nodes + the new node after.
1384 "OP_TARGET": self.op.node_name,
1385 "NODE_NAME": self.op.node_name,
1386 "NODE_PIP": self.op.primary_ip,
1387 "NODE_SIP": self.op.secondary_ip,
1389 nodes_0 = self.cfg.GetNodeList()
1390 nodes_1 = nodes_0 + [self.op.node_name, ]
1391 return env, nodes_0, nodes_1
1393 def CheckPrereq(self):
1394 """Check prerequisites.
1397 - the new node is not already in the config
1399 - its parameters (single/dual homed) matches the cluster
1401 Any errors are signalled by raising errors.OpPrereqError.
1404 node_name = self.op.node_name
1407 dns_data = utils.HostInfo(node_name)
1409 node = dns_data.name
1410 primary_ip = self.op.primary_ip = dns_data.ip
1411 secondary_ip = getattr(self.op, "secondary_ip", None)
1412 if secondary_ip is None:
1413 secondary_ip = primary_ip
1414 if not utils.IsValidIP(secondary_ip):
1415 raise errors.OpPrereqError("Invalid secondary IP given")
1416 self.op.secondary_ip = secondary_ip
1417 node_list = cfg.GetNodeList()
1418 if node in node_list:
1419 raise errors.OpPrereqError("Node %s is already in the configuration"
1422 for existing_node_name in node_list:
1423 existing_node = cfg.GetNodeInfo(existing_node_name)
1424 if (existing_node.primary_ip == primary_ip or
1425 existing_node.secondary_ip == primary_ip or
1426 existing_node.primary_ip == secondary_ip or
1427 existing_node.secondary_ip == secondary_ip):
1428 raise errors.OpPrereqError("New node ip address(es) conflict with"
1429 " existing node %s" % existing_node.name)
1431 # check that the type of the node (single versus dual homed) is the
1432 # same as for the master
1433 myself = cfg.GetNodeInfo(self.sstore.GetMasterNode())
1434 master_singlehomed = myself.secondary_ip == myself.primary_ip
1435 newbie_singlehomed = secondary_ip == primary_ip
1436 if master_singlehomed != newbie_singlehomed:
1437 if master_singlehomed:
1438 raise errors.OpPrereqError("The master has no private ip but the"
1439 " new node has one")
1441 raise errors.OpPrereqError("The master has a private ip but the"
1442 " new node doesn't have one")
1444 # checks reachablity
1445 if not utils.TcpPing(utils.HostInfo().name,
1447 constants.DEFAULT_NODED_PORT):
1448 raise errors.OpPrereqError("Node not reachable by ping")
1450 if not newbie_singlehomed:
1451 # check reachability from my secondary ip to newbie's secondary ip
1452 if not utils.TcpPing(myself.secondary_ip,
1454 constants.DEFAULT_NODED_PORT):
1455 raise errors.OpPrereqError(
1456 "Node secondary ip not reachable by TCP based ping to noded port")
1458 self.new_node = objects.Node(name=node,
1459 primary_ip=primary_ip,
1460 secondary_ip=secondary_ip)
1462 def Exec(self, feedback_fn):
1463 """Adds the new node to the cluster.
1466 new_node = self.new_node
1467 node = new_node.name
1469 # set up inter-node password and certificate and restarts the node daemon
1470 gntpass = self.sstore.GetNodeDaemonPassword()
1471 if not re.match('^[a-zA-Z0-9.]{1,64}$', gntpass):
1472 raise errors.OpExecError("ganeti password corruption detected")
1473 f = open(constants.SSL_CERT_FILE)
1475 gntpem = f.read(8192)
1478 # in the base64 pem encoding, neither '!' nor '.' are valid chars,
1479 # so we use this to detect an invalid certificate; as long as the
1480 # cert doesn't contain this, the here-document will be correctly
1481 # parsed by the shell sequence below
1482 if re.search('^!EOF\.', gntpem, re.MULTILINE):
1483 raise errors.OpExecError("invalid PEM encoding in the SSL certificate")
1484 if not gntpem.endswith("\n"):
1485 raise errors.OpExecError("PEM must end with newline")
1486 logger.Info("copy cluster pass to %s and starting the node daemon" % node)
1488 # and then connect with ssh to set password and start ganeti-noded
1489 # note that all the below variables are sanitized at this point,
1490 # either by being constants or by the checks above
1492 mycommand = ("umask 077 && "
1493 "echo '%s' > '%s' && "
1494 "cat > '%s' << '!EOF.' && \n"
1495 "%s!EOF.\n%s restart" %
1496 (gntpass, ss.KeyToFilename(ss.SS_NODED_PASS),
1497 constants.SSL_CERT_FILE, gntpem,
1498 constants.NODE_INITD_SCRIPT))
1500 result = ssh.SSHCall(node, 'root', mycommand, batch=False, ask_key=True)
1502 raise errors.OpExecError("Remote command on node %s, error: %s,"
1504 (node, result.fail_reason, result.output))
1506 # check connectivity
1509 result = rpc.call_version([node])[node]
1511 if constants.PROTOCOL_VERSION == result:
1512 logger.Info("communication to node %s fine, sw version %s match" %
1515 raise errors.OpExecError("Version mismatch master version %s,"
1516 " node version %s" %
1517 (constants.PROTOCOL_VERSION, result))
1519 raise errors.OpExecError("Cannot get version from the new node")
1522 logger.Info("copy ssh key to node %s" % node)
1523 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
1525 keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
1526 constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
1532 keyarray.append(f.read())
1536 result = rpc.call_node_add(node, keyarray[0], keyarray[1], keyarray[2],
1537 keyarray[3], keyarray[4], keyarray[5])
1540 raise errors.OpExecError("Cannot transfer ssh keys to the new node")
1542 # Add node to our /etc/hosts, and add key to known_hosts
1543 _UpdateEtcHosts(new_node.name, new_node.primary_ip)
1544 _UpdateKnownHosts(new_node.name, new_node.primary_ip,
1545 self.cfg.GetHostKey())
1547 if new_node.secondary_ip != new_node.primary_ip:
1548 if not rpc.call_node_tcp_ping(new_node.name,
1549 constants.LOCALHOST_IP_ADDRESS,
1550 new_node.secondary_ip,
1551 constants.DEFAULT_NODED_PORT,
1553 raise errors.OpExecError("Node claims it doesn't have the"
1554 " secondary ip you gave (%s).\n"
1555 "Please fix and re-run this command." %
1556 new_node.secondary_ip)
1558 success, msg = ssh.VerifyNodeHostname(node)
1560 raise errors.OpExecError("Node '%s' claims it has a different hostname"
1561 " than the one the resolver gives: %s.\n"
1562 "Please fix and re-run this command." %
1565 # Distribute updated /etc/hosts and known_hosts to all nodes,
1566 # including the node just added
1567 myself = self.cfg.GetNodeInfo(self.sstore.GetMasterNode())
1568 dist_nodes = self.cfg.GetNodeList() + [node]
1569 if myself.name in dist_nodes:
1570 dist_nodes.remove(myself.name)
1572 logger.Debug("Copying hosts and known_hosts to all nodes")
1573 for fname in ("/etc/hosts", constants.SSH_KNOWN_HOSTS_FILE):
1574 result = rpc.call_upload_file(dist_nodes, fname)
1575 for to_node in dist_nodes:
1576 if not result[to_node]:
1577 logger.Error("copy of file %s to node %s failed" %
1580 to_copy = ss.GetFileList()
1581 for fname in to_copy:
1582 if not ssh.CopyFileToNode(node, fname):
1583 logger.Error("could not copy file %s to node %s" % (fname, node))
1585 logger.Info("adding node %s to cluster.conf" % node)
1586 self.cfg.AddNode(new_node)
1589 class LUMasterFailover(LogicalUnit):
1590 """Failover the master node to the current node.
1592 This is a special LU in that it must run on a non-master node.
1595 HPATH = "master-failover"
1596 HTYPE = constants.HTYPE_CLUSTER
1600 def BuildHooksEnv(self):
1603 This will run on the new master only in the pre phase, and on all
1604 the nodes in the post phase.
1608 "OP_TARGET": self.new_master,
1609 "NEW_MASTER": self.new_master,
1610 "OLD_MASTER": self.old_master,
1612 return env, [self.new_master], self.cfg.GetNodeList()
1614 def CheckPrereq(self):
1615 """Check prerequisites.
1617 This checks that we are not already the master.
1620 self.new_master = utils.HostInfo().name
1621 self.old_master = self.sstore.GetMasterNode()
1623 if self.old_master == self.new_master:
1624 raise errors.OpPrereqError("This commands must be run on the node"
1625 " where you want the new master to be.\n"
1626 "%s is already the master" %
1629 def Exec(self, feedback_fn):
1630 """Failover the master node.
1632 This command, when run on a non-master node, will cause the current
1633 master to cease being master, and the non-master to become new
1637 #TODO: do not rely on gethostname returning the FQDN
1638 logger.Info("setting master to %s, old master: %s" %
1639 (self.new_master, self.old_master))
1641 if not rpc.call_node_stop_master(self.old_master):
1642 logger.Error("could disable the master role on the old master"
1643 " %s, please disable manually" % self.old_master)
1646 ss.SetKey(ss.SS_MASTER_NODE, self.new_master)
1647 if not rpc.call_upload_file(self.cfg.GetNodeList(),
1648 ss.KeyToFilename(ss.SS_MASTER_NODE)):
1649 logger.Error("could not distribute the new simple store master file"
1650 " to the other nodes, please check.")
1652 if not rpc.call_node_start_master(self.new_master):
1653 logger.Error("could not start the master role on the new master"
1654 " %s, please check" % self.new_master)
1655 feedback_fn("Error in activating the master IP on the new master,\n"
1656 "please fix manually.")
1660 class LUQueryClusterInfo(NoHooksLU):
1661 """Query cluster configuration.
1667 def CheckPrereq(self):
1668 """No prerequsites needed for this LU.
1673 def Exec(self, feedback_fn):
1674 """Return cluster config.
1678 "name": self.sstore.GetClusterName(),
1679 "software_version": constants.RELEASE_VERSION,
1680 "protocol_version": constants.PROTOCOL_VERSION,
1681 "config_version": constants.CONFIG_VERSION,
1682 "os_api_version": constants.OS_API_VERSION,
1683 "export_version": constants.EXPORT_VERSION,
1684 "master": self.sstore.GetMasterNode(),
1685 "architecture": (platform.architecture()[0], platform.machine()),
1691 class LUClusterCopyFile(NoHooksLU):
1692 """Copy file to cluster.
1695 _OP_REQP = ["nodes", "filename"]
1697 def CheckPrereq(self):
1698 """Check prerequisites.
1700 It should check that the named file exists and that the given list
1704 if not os.path.exists(self.op.filename):
1705 raise errors.OpPrereqError("No such filename '%s'" % self.op.filename)
1707 self.nodes = _GetWantedNodes(self, self.op.nodes)
1709 def Exec(self, feedback_fn):
1710 """Copy a file from master to some nodes.
1713 opts - class with options as members
1714 args - list containing a single element, the file name
1716 nodes - list containing the name of target nodes; if empty, all nodes
1719 filename = self.op.filename
1721 myname = utils.HostInfo().name
1723 for node in self.nodes:
1726 if not ssh.CopyFileToNode(node, filename):
1727 logger.Error("Copy of file %s to node %s failed" % (filename, node))
1730 class LUDumpClusterConfig(NoHooksLU):
1731 """Return a text-representation of the cluster-config.
1736 def CheckPrereq(self):
1737 """No prerequisites.
1742 def Exec(self, feedback_fn):
1743 """Dump a representation of the cluster config to the standard output.
1746 return self.cfg.DumpConfig()
1749 class LURunClusterCommand(NoHooksLU):
1750 """Run a command on some nodes.
1753 _OP_REQP = ["command", "nodes"]
1755 def CheckPrereq(self):
1756 """Check prerequisites.
1758 It checks that the given list of nodes is valid.
1761 self.nodes = _GetWantedNodes(self, self.op.nodes)
1763 def Exec(self, feedback_fn):
1764 """Run a command on some nodes.
1768 for node in self.nodes:
1769 result = ssh.SSHCall(node, "root", self.op.command)
1770 data.append((node, result.output, result.exit_code))
1775 class LUActivateInstanceDisks(NoHooksLU):
1776 """Bring up an instance's disks.
1779 _OP_REQP = ["instance_name"]
1781 def CheckPrereq(self):
1782 """Check prerequisites.
1784 This checks that the instance is in the cluster.
1787 instance = self.cfg.GetInstanceInfo(
1788 self.cfg.ExpandInstanceName(self.op.instance_name))
1789 if instance is None:
1790 raise errors.OpPrereqError("Instance '%s' not known" %
1791 self.op.instance_name)
1792 self.instance = instance
1795 def Exec(self, feedback_fn):
1796 """Activate the disks.
1799 disks_ok, disks_info = _AssembleInstanceDisks(self.instance, self.cfg)
1801 raise errors.OpExecError("Cannot activate block devices")
1806 def _AssembleInstanceDisks(instance, cfg, ignore_secondaries=False):
1807 """Prepare the block devices for an instance.
1809 This sets up the block devices on all nodes.
1812 instance: a ganeti.objects.Instance object
1813 ignore_secondaries: if true, errors on secondary nodes won't result
1814 in an error return from the function
1817 false if the operation failed
1818 list of (host, instance_visible_name, node_visible_name) if the operation
1819 suceeded with the mapping from node devices to instance devices
1823 for inst_disk in instance.disks:
1824 master_result = None
1825 for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
1826 cfg.SetDiskID(node_disk, node)
1827 is_primary = node == instance.primary_node
1828 result = rpc.call_blockdev_assemble(node, node_disk, 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 return disks_ok, device_info
1842 def _StartInstanceDisks(cfg, instance, force):
1843 """Start the disks of an instance.
1846 disks_ok, dummy = _AssembleInstanceDisks(instance, cfg,
1847 ignore_secondaries=force)
1849 _ShutdownInstanceDisks(instance, cfg)
1850 if force is not None and not force:
1851 logger.Error("If the message above refers to a secondary node,"
1852 " you can retry the operation using '--force'.")
1853 raise errors.OpExecError("Disk consistency error")
1856 class LUDeactivateInstanceDisks(NoHooksLU):
1857 """Shutdown an instance's disks.
1860 _OP_REQP = ["instance_name"]
1862 def CheckPrereq(self):
1863 """Check prerequisites.
1865 This checks that the instance is in the cluster.
1868 instance = self.cfg.GetInstanceInfo(
1869 self.cfg.ExpandInstanceName(self.op.instance_name))
1870 if instance is None:
1871 raise errors.OpPrereqError("Instance '%s' not known" %
1872 self.op.instance_name)
1873 self.instance = instance
1875 def Exec(self, feedback_fn):
1876 """Deactivate the disks
1879 instance = self.instance
1880 ins_l = rpc.call_instance_list([instance.primary_node])
1881 ins_l = ins_l[instance.primary_node]
1882 if not type(ins_l) is list:
1883 raise errors.OpExecError("Can't contact node '%s'" %
1884 instance.primary_node)
1886 if self.instance.name in ins_l:
1887 raise errors.OpExecError("Instance is running, can't shutdown"
1890 _ShutdownInstanceDisks(instance, self.cfg)
1893 def _ShutdownInstanceDisks(instance, cfg, ignore_primary=False):
1894 """Shutdown block devices of an instance.
1896 This does the shutdown on all nodes of the instance.
1898 If the ignore_primary is false, errors on the primary node are
1903 for disk in instance.disks:
1904 for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
1905 cfg.SetDiskID(top_disk, node)
1906 if not rpc.call_blockdev_shutdown(node, top_disk):
1907 logger.Error("could not shutdown block device %s on node %s" %
1908 (disk.iv_name, node))
1909 if not ignore_primary or node != instance.primary_node:
1914 class LUStartupInstance(LogicalUnit):
1915 """Starts an instance.
1918 HPATH = "instance-start"
1919 HTYPE = constants.HTYPE_INSTANCE
1920 _OP_REQP = ["instance_name", "force"]
1922 def BuildHooksEnv(self):
1925 This runs on master, primary and secondary nodes of the instance.
1929 "FORCE": self.op.force,
1931 env.update(_BuildInstanceHookEnvByObject(self.instance))
1932 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
1933 list(self.instance.secondary_nodes))
1936 def CheckPrereq(self):
1937 """Check prerequisites.
1939 This checks that the instance is in the cluster.
1942 instance = self.cfg.GetInstanceInfo(
1943 self.cfg.ExpandInstanceName(self.op.instance_name))
1944 if instance is None:
1945 raise errors.OpPrereqError("Instance '%s' not known" %
1946 self.op.instance_name)
1948 # check bridges existance
1949 _CheckInstanceBridgesExist(instance)
1951 self.instance = instance
1952 self.op.instance_name = instance.name
1954 def Exec(self, feedback_fn):
1955 """Start the instance.
1958 instance = self.instance
1959 force = self.op.force
1960 extra_args = getattr(self.op, "extra_args", "")
1962 node_current = instance.primary_node
1964 nodeinfo = rpc.call_node_info([node_current], self.cfg.GetVGName())
1966 raise errors.OpExecError("Could not contact node %s for infos" %
1969 freememory = nodeinfo[node_current]['memory_free']
1970 memory = instance.memory
1971 if memory > freememory:
1972 raise errors.OpExecError("Not enough memory to start instance"
1974 " needed %s MiB, available %s MiB" %
1975 (instance.name, node_current, memory,
1978 _StartInstanceDisks(self.cfg, instance, force)
1980 if not rpc.call_instance_start(node_current, instance, extra_args):
1981 _ShutdownInstanceDisks(instance, self.cfg)
1982 raise errors.OpExecError("Could not start instance")
1984 self.cfg.MarkInstanceUp(instance.name)
1987 class LURebootInstance(LogicalUnit):
1988 """Reboot an instance.
1991 HPATH = "instance-reboot"
1992 HTYPE = constants.HTYPE_INSTANCE
1993 _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
1995 def BuildHooksEnv(self):
1998 This runs on master, primary and secondary nodes of the instance.
2002 "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2004 env.update(_BuildInstanceHookEnvByObject(self.instance))
2005 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2006 list(self.instance.secondary_nodes))
2009 def CheckPrereq(self):
2010 """Check prerequisites.
2012 This checks that the instance is in the cluster.
2015 instance = self.cfg.GetInstanceInfo(
2016 self.cfg.ExpandInstanceName(self.op.instance_name))
2017 if instance is None:
2018 raise errors.OpPrereqError("Instance '%s' not known" %
2019 self.op.instance_name)
2021 # check bridges existance
2022 _CheckInstanceBridgesExist(instance)
2024 self.instance = instance
2025 self.op.instance_name = instance.name
2027 def Exec(self, feedback_fn):
2028 """Reboot the instance.
2031 instance = self.instance
2032 ignore_secondaries = self.op.ignore_secondaries
2033 reboot_type = self.op.reboot_type
2034 extra_args = getattr(self.op, "extra_args", "")
2036 node_current = instance.primary_node
2038 if reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2039 constants.INSTANCE_REBOOT_HARD,
2040 constants.INSTANCE_REBOOT_FULL]:
2041 raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2042 (constants.INSTANCE_REBOOT_SOFT,
2043 constants.INSTANCE_REBOOT_HARD,
2044 constants.INSTANCE_REBOOT_FULL))
2046 if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2047 constants.INSTANCE_REBOOT_HARD]:
2048 if not rpc.call_instance_reboot(node_current, instance,
2049 reboot_type, extra_args):
2050 raise errors.OpExecError("Could not reboot instance")
2052 if not rpc.call_instance_shutdown(node_current, instance):
2053 raise errors.OpExecError("could not shutdown instance for full reboot")
2054 _ShutdownInstanceDisks(instance, self.cfg)
2055 _StartInstanceDisks(self.cfg, instance, ignore_secondaries)
2056 if not rpc.call_instance_start(node_current, instance, extra_args):
2057 _ShutdownInstanceDisks(instance, self.cfg)
2058 raise errors.OpExecError("Could not start instance for full reboot")
2060 self.cfg.MarkInstanceUp(instance.name)
2063 class LUShutdownInstance(LogicalUnit):
2064 """Shutdown an instance.
2067 HPATH = "instance-stop"
2068 HTYPE = constants.HTYPE_INSTANCE
2069 _OP_REQP = ["instance_name"]
2071 def BuildHooksEnv(self):
2074 This runs on master, primary and secondary nodes of the instance.
2077 env = _BuildInstanceHookEnvByObject(self.instance)
2078 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2079 list(self.instance.secondary_nodes))
2082 def CheckPrereq(self):
2083 """Check prerequisites.
2085 This checks that the instance is in the cluster.
2088 instance = self.cfg.GetInstanceInfo(
2089 self.cfg.ExpandInstanceName(self.op.instance_name))
2090 if instance is None:
2091 raise errors.OpPrereqError("Instance '%s' not known" %
2092 self.op.instance_name)
2093 self.instance = instance
2095 def Exec(self, feedback_fn):
2096 """Shutdown the instance.
2099 instance = self.instance
2100 node_current = instance.primary_node
2101 if not rpc.call_instance_shutdown(node_current, instance):
2102 logger.Error("could not shutdown instance")
2104 self.cfg.MarkInstanceDown(instance.name)
2105 _ShutdownInstanceDisks(instance, self.cfg)
2108 class LUReinstallInstance(LogicalUnit):
2109 """Reinstall an instance.
2112 HPATH = "instance-reinstall"
2113 HTYPE = constants.HTYPE_INSTANCE
2114 _OP_REQP = ["instance_name"]
2116 def BuildHooksEnv(self):
2119 This runs on master, primary and secondary nodes of the instance.
2122 env = _BuildInstanceHookEnvByObject(self.instance)
2123 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2124 list(self.instance.secondary_nodes))
2127 def CheckPrereq(self):
2128 """Check prerequisites.
2130 This checks that the instance is in the cluster and is not running.
2133 instance = self.cfg.GetInstanceInfo(
2134 self.cfg.ExpandInstanceName(self.op.instance_name))
2135 if instance is None:
2136 raise errors.OpPrereqError("Instance '%s' not known" %
2137 self.op.instance_name)
2138 if instance.disk_template == constants.DT_DISKLESS:
2139 raise errors.OpPrereqError("Instance '%s' has no disks" %
2140 self.op.instance_name)
2141 if instance.status != "down":
2142 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2143 self.op.instance_name)
2144 remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2146 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2147 (self.op.instance_name,
2148 instance.primary_node))
2150 self.op.os_type = getattr(self.op, "os_type", None)
2151 if self.op.os_type is not None:
2153 pnode = self.cfg.GetNodeInfo(
2154 self.cfg.ExpandNodeName(instance.primary_node))
2156 raise errors.OpPrereqError("Primary node '%s' is unknown" %
2158 os_obj = rpc.call_os_get([pnode.name], self.op.os_type)[pnode.name]
2159 if not isinstance(os_obj, objects.OS):
2160 raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2161 " primary node" % self.op.os_type)
2163 self.instance = instance
2165 def Exec(self, feedback_fn):
2166 """Reinstall the instance.
2169 inst = self.instance
2171 if self.op.os_type is not None:
2172 feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2173 inst.os = self.op.os_type
2174 self.cfg.AddInstance(inst)
2176 _StartInstanceDisks(self.cfg, inst, None)
2178 feedback_fn("Running the instance OS create scripts...")
2179 if not rpc.call_instance_os_add(inst.primary_node, inst, "sda", "sdb"):
2180 raise errors.OpExecError("Could not install OS for instance %s "
2182 (inst.name, inst.primary_node))
2184 _ShutdownInstanceDisks(inst, self.cfg)
2187 class LURenameInstance(LogicalUnit):
2188 """Rename an instance.
2191 HPATH = "instance-rename"
2192 HTYPE = constants.HTYPE_INSTANCE
2193 _OP_REQP = ["instance_name", "new_name"]
2195 def BuildHooksEnv(self):
2198 This runs on master, primary and secondary nodes of the instance.
2201 env = _BuildInstanceHookEnvByObject(self.instance)
2202 env["INSTANCE_NEW_NAME"] = self.op.new_name
2203 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2204 list(self.instance.secondary_nodes))
2207 def CheckPrereq(self):
2208 """Check prerequisites.
2210 This checks that the instance is in the cluster and is not running.
2213 instance = self.cfg.GetInstanceInfo(
2214 self.cfg.ExpandInstanceName(self.op.instance_name))
2215 if instance is None:
2216 raise errors.OpPrereqError("Instance '%s' not known" %
2217 self.op.instance_name)
2218 if instance.status != "down":
2219 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2220 self.op.instance_name)
2221 remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2223 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2224 (self.op.instance_name,
2225 instance.primary_node))
2226 self.instance = instance
2228 # new name verification
2229 name_info = utils.HostInfo(self.op.new_name)
2231 self.op.new_name = new_name = name_info.name
2232 if not getattr(self.op, "ignore_ip", False):
2233 command = ["fping", "-q", name_info.ip]
2234 result = utils.RunCmd(command)
2235 if not result.failed:
2236 raise errors.OpPrereqError("IP %s of instance %s already in use" %
2237 (name_info.ip, new_name))
2240 def Exec(self, feedback_fn):
2241 """Reinstall the instance.
2244 inst = self.instance
2245 old_name = inst.name
2247 self.cfg.RenameInstance(inst.name, self.op.new_name)
2249 # re-read the instance from the configuration after rename
2250 inst = self.cfg.GetInstanceInfo(self.op.new_name)
2252 _StartInstanceDisks(self.cfg, inst, None)
2254 if not rpc.call_instance_run_rename(inst.primary_node, inst, old_name,
2256 msg = ("Could run OS rename script for instance %s\n"
2258 "(but the instance has been renamed in Ganeti)" %
2259 (inst.name, inst.primary_node))
2262 _ShutdownInstanceDisks(inst, self.cfg)
2265 class LURemoveInstance(LogicalUnit):
2266 """Remove an instance.
2269 HPATH = "instance-remove"
2270 HTYPE = constants.HTYPE_INSTANCE
2271 _OP_REQP = ["instance_name"]
2273 def BuildHooksEnv(self):
2276 This runs on master, primary and secondary nodes of the instance.
2279 env = _BuildInstanceHookEnvByObject(self.instance)
2280 nl = [self.sstore.GetMasterNode()]
2283 def CheckPrereq(self):
2284 """Check prerequisites.
2286 This checks that the instance is in the cluster.
2289 instance = self.cfg.GetInstanceInfo(
2290 self.cfg.ExpandInstanceName(self.op.instance_name))
2291 if instance is None:
2292 raise errors.OpPrereqError("Instance '%s' not known" %
2293 self.op.instance_name)
2294 self.instance = instance
2296 def Exec(self, feedback_fn):
2297 """Remove the instance.
2300 instance = self.instance
2301 logger.Info("shutting down instance %s on node %s" %
2302 (instance.name, instance.primary_node))
2304 if not rpc.call_instance_shutdown(instance.primary_node, instance):
2305 if self.op.ignore_failures:
2306 feedback_fn("Warning: can't shutdown instance")
2308 raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2309 (instance.name, instance.primary_node))
2311 logger.Info("removing block devices for instance %s" % instance.name)
2313 if not _RemoveDisks(instance, self.cfg):
2314 if self.op.ignore_failures:
2315 feedback_fn("Warning: can't remove instance's disks")
2317 raise errors.OpExecError("Can't remove instance's disks")
2319 logger.Info("removing instance %s out of cluster config" % instance.name)
2321 self.cfg.RemoveInstance(instance.name)
2324 class LUQueryInstances(NoHooksLU):
2325 """Logical unit for querying instances.
2328 _OP_REQP = ["output_fields", "names"]
2330 def CheckPrereq(self):
2331 """Check prerequisites.
2333 This checks that the fields required are valid output fields.
2336 self.dynamic_fields = frozenset(["oper_state", "oper_ram"])
2337 _CheckOutputFields(static=["name", "os", "pnode", "snodes",
2338 "admin_state", "admin_ram",
2339 "disk_template", "ip", "mac", "bridge",
2340 "sda_size", "sdb_size"],
2341 dynamic=self.dynamic_fields,
2342 selected=self.op.output_fields)
2344 self.wanted = _GetWantedInstances(self, self.op.names)
2346 def Exec(self, feedback_fn):
2347 """Computes the list of nodes and their attributes.
2350 instance_names = self.wanted
2351 instance_list = [self.cfg.GetInstanceInfo(iname) for iname
2354 # begin data gathering
2356 nodes = frozenset([inst.primary_node for inst in instance_list])
2359 if self.dynamic_fields.intersection(self.op.output_fields):
2361 node_data = rpc.call_all_instances_info(nodes)
2363 result = node_data[name]
2365 live_data.update(result)
2366 elif result == False:
2367 bad_nodes.append(name)
2368 # else no instance is alive
2370 live_data = dict([(name, {}) for name in instance_names])
2372 # end data gathering
2375 for instance in instance_list:
2377 for field in self.op.output_fields:
2382 elif field == "pnode":
2383 val = instance.primary_node
2384 elif field == "snodes":
2385 val = list(instance.secondary_nodes)
2386 elif field == "admin_state":
2387 val = (instance.status != "down")
2388 elif field == "oper_state":
2389 if instance.primary_node in bad_nodes:
2392 val = bool(live_data.get(instance.name))
2393 elif field == "admin_ram":
2394 val = instance.memory
2395 elif field == "oper_ram":
2396 if instance.primary_node in bad_nodes:
2398 elif instance.name in live_data:
2399 val = live_data[instance.name].get("memory", "?")
2402 elif field == "disk_template":
2403 val = instance.disk_template
2405 val = instance.nics[0].ip
2406 elif field == "bridge":
2407 val = instance.nics[0].bridge
2408 elif field == "mac":
2409 val = instance.nics[0].mac
2410 elif field == "sda_size" or field == "sdb_size":
2411 disk = instance.FindDisk(field[:3])
2417 raise errors.ParameterError(field)
2424 class LUFailoverInstance(LogicalUnit):
2425 """Failover an instance.
2428 HPATH = "instance-failover"
2429 HTYPE = constants.HTYPE_INSTANCE
2430 _OP_REQP = ["instance_name", "ignore_consistency"]
2432 def BuildHooksEnv(self):
2435 This runs on master, primary and secondary nodes of the instance.
2439 "IGNORE_CONSISTENCY": self.op.ignore_consistency,
2441 env.update(_BuildInstanceHookEnvByObject(self.instance))
2442 nl = [self.sstore.GetMasterNode()] + list(self.instance.secondary_nodes)
2445 def CheckPrereq(self):
2446 """Check prerequisites.
2448 This checks that the instance is in the cluster.
2451 instance = self.cfg.GetInstanceInfo(
2452 self.cfg.ExpandInstanceName(self.op.instance_name))
2453 if instance is None:
2454 raise errors.OpPrereqError("Instance '%s' not known" %
2455 self.op.instance_name)
2457 if instance.disk_template != constants.DT_REMOTE_RAID1:
2458 raise errors.OpPrereqError("Instance's disk layout is not"
2461 secondary_nodes = instance.secondary_nodes
2462 if not secondary_nodes:
2463 raise errors.ProgrammerError("no secondary node but using "
2464 "DT_REMOTE_RAID1 template")
2466 # check memory requirements on the secondary node
2467 target_node = secondary_nodes[0]
2468 nodeinfo = rpc.call_node_info([target_node], self.cfg.GetVGName())
2469 info = nodeinfo.get(target_node, None)
2471 raise errors.OpPrereqError("Cannot get current information"
2472 " from node '%s'" % nodeinfo)
2473 if instance.memory > info['memory_free']:
2474 raise errors.OpPrereqError("Not enough memory on target node %s."
2475 " %d MB available, %d MB required" %
2476 (target_node, info['memory_free'],
2479 # check bridge existance
2480 brlist = [nic.bridge for nic in instance.nics]
2481 if not rpc.call_bridges_exist(instance.primary_node, brlist):
2482 raise errors.OpPrereqError("One or more target bridges %s does not"
2483 " exist on destination node '%s'" %
2484 (brlist, instance.primary_node))
2486 self.instance = instance
2488 def Exec(self, feedback_fn):
2489 """Failover an instance.
2491 The failover is done by shutting it down on its present node and
2492 starting it on the secondary.
2495 instance = self.instance
2497 source_node = instance.primary_node
2498 target_node = instance.secondary_nodes[0]
2500 feedback_fn("* checking disk consistency between source and target")
2501 for dev in instance.disks:
2502 # for remote_raid1, these are md over drbd
2503 if not _CheckDiskConsistency(self.cfg, dev, target_node, False):
2504 if not self.op.ignore_consistency:
2505 raise errors.OpExecError("Disk %s is degraded on target node,"
2506 " aborting failover." % dev.iv_name)
2508 feedback_fn("* checking target node resource availability")
2509 nodeinfo = rpc.call_node_info([target_node], self.cfg.GetVGName())
2512 raise errors.OpExecError("Could not contact target node %s." %
2515 free_memory = int(nodeinfo[target_node]['memory_free'])
2516 memory = instance.memory
2517 if memory > free_memory:
2518 raise errors.OpExecError("Not enough memory to create instance %s on"
2519 " node %s. needed %s MiB, available %s MiB" %
2520 (instance.name, target_node, memory,
2523 feedback_fn("* shutting down instance on source node")
2524 logger.Info("Shutting down instance %s on node %s" %
2525 (instance.name, source_node))
2527 if not rpc.call_instance_shutdown(source_node, instance):
2528 logger.Error("Could not shutdown instance %s on node %s. Proceeding"
2529 " anyway. Please make sure node %s is down" %
2530 (instance.name, source_node, source_node))
2532 feedback_fn("* deactivating the instance's disks on source node")
2533 if not _ShutdownInstanceDisks(instance, self.cfg, ignore_primary=True):
2534 raise errors.OpExecError("Can't shut down the instance's disks.")
2536 instance.primary_node = target_node
2537 # distribute new instance config to the other nodes
2538 self.cfg.AddInstance(instance)
2540 feedback_fn("* activating the instance's disks on target node")
2541 logger.Info("Starting instance %s on node %s" %
2542 (instance.name, target_node))
2544 disks_ok, dummy = _AssembleInstanceDisks(instance, self.cfg,
2545 ignore_secondaries=True)
2547 _ShutdownInstanceDisks(instance, self.cfg)
2548 raise errors.OpExecError("Can't activate the instance's disks")
2550 feedback_fn("* starting the instance on the target node")
2551 if not rpc.call_instance_start(target_node, instance, None):
2552 _ShutdownInstanceDisks(instance, self.cfg)
2553 raise errors.OpExecError("Could not start instance %s on node %s." %
2554 (instance.name, target_node))
2557 def _CreateBlockDevOnPrimary(cfg, node, device, info):
2558 """Create a tree of block devices on the primary node.
2560 This always creates all devices.
2564 for child in device.children:
2565 if not _CreateBlockDevOnPrimary(cfg, node, child, info):
2568 cfg.SetDiskID(device, node)
2569 new_id = rpc.call_blockdev_create(node, device, device.size, True, info)
2572 if device.physical_id is None:
2573 device.physical_id = new_id
2577 def _CreateBlockDevOnSecondary(cfg, node, device, force, info):
2578 """Create a tree of block devices on a secondary node.
2580 If this device type has to be created on secondaries, create it and
2583 If not, just recurse to children keeping the same 'force' value.
2586 if device.CreateOnSecondary():
2589 for child in device.children:
2590 if not _CreateBlockDevOnSecondary(cfg, node, child, force, info):
2595 cfg.SetDiskID(device, node)
2596 new_id = rpc.call_blockdev_create(node, device, device.size, False, info)
2599 if device.physical_id is None:
2600 device.physical_id = new_id
2604 def _GenerateUniqueNames(cfg, exts):
2605 """Generate a suitable LV name.
2607 This will generate a logical volume name for the given instance.
2612 new_id = cfg.GenerateUniqueID()
2613 results.append("%s%s" % (new_id, val))
2617 def _GenerateMDDRBDBranch(cfg, primary, secondary, size, names):
2618 """Generate a drbd device complete with its children.
2621 port = cfg.AllocatePort()
2622 vgname = cfg.GetVGName()
2623 dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
2624 logical_id=(vgname, names[0]))
2625 dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
2626 logical_id=(vgname, names[1]))
2627 drbd_dev = objects.Disk(dev_type=constants.LD_DRBD7, size=size,
2628 logical_id = (primary, secondary, port),
2629 children = [dev_data, dev_meta])
2633 def _GenerateDiskTemplate(cfg, template_name,
2634 instance_name, primary_node,
2635 secondary_nodes, disk_sz, swap_sz):
2636 """Generate the entire disk layout for a given template type.
2639 #TODO: compute space requirements
2641 vgname = cfg.GetVGName()
2642 if template_name == "diskless":
2644 elif template_name == "plain":
2645 if len(secondary_nodes) != 0:
2646 raise errors.ProgrammerError("Wrong template configuration")
2648 names = _GenerateUniqueNames(cfg, [".sda", ".sdb"])
2649 sda_dev = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2650 logical_id=(vgname, names[0]),
2652 sdb_dev = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2653 logical_id=(vgname, names[1]),
2655 disks = [sda_dev, sdb_dev]
2656 elif template_name == "local_raid1":
2657 if len(secondary_nodes) != 0:
2658 raise errors.ProgrammerError("Wrong template configuration")
2661 names = _GenerateUniqueNames(cfg, [".sda_m1", ".sda_m2",
2662 ".sdb_m1", ".sdb_m2"])
2663 sda_dev_m1 = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2664 logical_id=(vgname, names[0]))
2665 sda_dev_m2 = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2666 logical_id=(vgname, names[1]))
2667 md_sda_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name = "sda",
2669 children = [sda_dev_m1, sda_dev_m2])
2670 sdb_dev_m1 = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2671 logical_id=(vgname, names[2]))
2672 sdb_dev_m2 = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2673 logical_id=(vgname, names[3]))
2674 md_sdb_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name = "sdb",
2676 children = [sdb_dev_m1, sdb_dev_m2])
2677 disks = [md_sda_dev, md_sdb_dev]
2678 elif template_name == constants.DT_REMOTE_RAID1:
2679 if len(secondary_nodes) != 1:
2680 raise errors.ProgrammerError("Wrong template configuration")
2681 remote_node = secondary_nodes[0]
2682 names = _GenerateUniqueNames(cfg, [".sda_data", ".sda_meta",
2683 ".sdb_data", ".sdb_meta"])
2684 drbd_sda_dev = _GenerateMDDRBDBranch(cfg, primary_node, remote_node,
2685 disk_sz, names[0:2])
2686 md_sda_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name="sda",
2687 children = [drbd_sda_dev], size=disk_sz)
2688 drbd_sdb_dev = _GenerateMDDRBDBranch(cfg, primary_node, remote_node,
2689 swap_sz, names[2:4])
2690 md_sdb_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name="sdb",
2691 children = [drbd_sdb_dev], size=swap_sz)
2692 disks = [md_sda_dev, md_sdb_dev]
2694 raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
2698 def _GetInstanceInfoText(instance):
2699 """Compute that text that should be added to the disk's metadata.
2702 return "originstname+%s" % instance.name
2705 def _CreateDisks(cfg, instance):
2706 """Create all disks for an instance.
2708 This abstracts away some work from AddInstance.
2711 instance: the instance object
2714 True or False showing the success of the creation process
2717 info = _GetInstanceInfoText(instance)
2719 for device in instance.disks:
2720 logger.Info("creating volume %s for instance %s" %
2721 (device.iv_name, instance.name))
2723 for secondary_node in instance.secondary_nodes:
2724 if not _CreateBlockDevOnSecondary(cfg, secondary_node, device, False,
2726 logger.Error("failed to create volume %s (%s) on secondary node %s!" %
2727 (device.iv_name, device, secondary_node))
2730 if not _CreateBlockDevOnPrimary(cfg, instance.primary_node, device, info):
2731 logger.Error("failed to create volume %s on primary!" %
2737 def _RemoveDisks(instance, cfg):
2738 """Remove all disks for an instance.
2740 This abstracts away some work from `AddInstance()` and
2741 `RemoveInstance()`. Note that in case some of the devices couldn't
2742 be removed, the removal will continue with the other ones (compare
2743 with `_CreateDisks()`).
2746 instance: the instance object
2749 True or False showing the success of the removal proces
2752 logger.Info("removing block devices for instance %s" % instance.name)
2755 for device in instance.disks:
2756 for node, disk in device.ComputeNodeTree(instance.primary_node):
2757 cfg.SetDiskID(disk, node)
2758 if not rpc.call_blockdev_remove(node, disk):
2759 logger.Error("could not remove block device %s on node %s,"
2760 " continuing anyway" %
2761 (device.iv_name, node))
2766 class LUCreateInstance(LogicalUnit):
2767 """Create an instance.
2770 HPATH = "instance-add"
2771 HTYPE = constants.HTYPE_INSTANCE
2772 _OP_REQP = ["instance_name", "mem_size", "disk_size", "pnode",
2773 "disk_template", "swap_size", "mode", "start", "vcpus",
2774 "wait_for_sync", "ip_check"]
2776 def BuildHooksEnv(self):
2779 This runs on master, primary and secondary nodes of the instance.
2783 "INSTANCE_DISK_TEMPLATE": self.op.disk_template,
2784 "INSTANCE_DISK_SIZE": self.op.disk_size,
2785 "INSTANCE_SWAP_SIZE": self.op.swap_size,
2786 "INSTANCE_ADD_MODE": self.op.mode,
2788 if self.op.mode == constants.INSTANCE_IMPORT:
2789 env["INSTANCE_SRC_NODE"] = self.op.src_node
2790 env["INSTANCE_SRC_PATH"] = self.op.src_path
2791 env["INSTANCE_SRC_IMAGE"] = self.src_image
2793 env.update(_BuildInstanceHookEnv(name=self.op.instance_name,
2794 primary_node=self.op.pnode,
2795 secondary_nodes=self.secondaries,
2796 status=self.instance_status,
2797 os_type=self.op.os_type,
2798 memory=self.op.mem_size,
2799 vcpus=self.op.vcpus,
2800 nics=[(self.inst_ip, self.op.bridge)],
2803 nl = ([self.sstore.GetMasterNode(), self.op.pnode] +
2808 def CheckPrereq(self):
2809 """Check prerequisites.
2812 if self.op.mode not in (constants.INSTANCE_CREATE,
2813 constants.INSTANCE_IMPORT):
2814 raise errors.OpPrereqError("Invalid instance creation mode '%s'" %
2817 if self.op.mode == constants.INSTANCE_IMPORT:
2818 src_node = getattr(self.op, "src_node", None)
2819 src_path = getattr(self.op, "src_path", None)
2820 if src_node is None or src_path is None:
2821 raise errors.OpPrereqError("Importing an instance requires source"
2822 " node and path options")
2823 src_node_full = self.cfg.ExpandNodeName(src_node)
2824 if src_node_full is None:
2825 raise errors.OpPrereqError("Unknown source node '%s'" % src_node)
2826 self.op.src_node = src_node = src_node_full
2828 if not os.path.isabs(src_path):
2829 raise errors.OpPrereqError("The source path must be absolute")
2831 export_info = rpc.call_export_info(src_node, src_path)
2834 raise errors.OpPrereqError("No export found in dir %s" % src_path)
2836 if not export_info.has_section(constants.INISECT_EXP):
2837 raise errors.ProgrammerError("Corrupted export config")
2839 ei_version = export_info.get(constants.INISECT_EXP, 'version')
2840 if (int(ei_version) != constants.EXPORT_VERSION):
2841 raise errors.OpPrereqError("Wrong export version %s (wanted %d)" %
2842 (ei_version, constants.EXPORT_VERSION))
2844 if int(export_info.get(constants.INISECT_INS, 'disk_count')) > 1:
2845 raise errors.OpPrereqError("Can't import instance with more than"
2848 # FIXME: are the old os-es, disk sizes, etc. useful?
2849 self.op.os_type = export_info.get(constants.INISECT_EXP, 'os')
2850 diskimage = os.path.join(src_path, export_info.get(constants.INISECT_INS,
2852 self.src_image = diskimage
2853 else: # INSTANCE_CREATE
2854 if getattr(self.op, "os_type", None) is None:
2855 raise errors.OpPrereqError("No guest OS specified")
2857 # check primary node
2858 pnode = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.pnode))
2860 raise errors.OpPrereqError("Primary node '%s' is unknown" %
2862 self.op.pnode = pnode.name
2864 self.secondaries = []
2865 # disk template and mirror node verification
2866 if self.op.disk_template not in constants.DISK_TEMPLATES:
2867 raise errors.OpPrereqError("Invalid disk template name")
2869 if self.op.disk_template == constants.DT_REMOTE_RAID1:
2870 if getattr(self.op, "snode", None) is None:
2871 raise errors.OpPrereqError("The 'remote_raid1' disk template needs"
2874 snode_name = self.cfg.ExpandNodeName(self.op.snode)
2875 if snode_name is None:
2876 raise errors.OpPrereqError("Unknown secondary node '%s'" %
2878 elif snode_name == pnode.name:
2879 raise errors.OpPrereqError("The secondary node cannot be"
2880 " the primary node.")
2881 self.secondaries.append(snode_name)
2883 # Check lv size requirements
2884 nodenames = [pnode.name] + self.secondaries
2885 nodeinfo = rpc.call_node_info(nodenames, self.cfg.GetVGName())
2887 # Required free disk space as a function of disk and swap space
2889 constants.DT_DISKLESS: 0,
2890 constants.DT_PLAIN: self.op.disk_size + self.op.swap_size,
2891 constants.DT_LOCAL_RAID1: (self.op.disk_size + self.op.swap_size) * 2,
2892 # 256 MB are added for drbd metadata, 128MB for each drbd device
2893 constants.DT_REMOTE_RAID1: self.op.disk_size + self.op.swap_size + 256,
2896 if self.op.disk_template not in req_size_dict:
2897 raise errors.ProgrammerError("Disk template '%s' size requirement"
2898 " is unknown" % self.op.disk_template)
2900 req_size = req_size_dict[self.op.disk_template]
2902 for node in nodenames:
2903 info = nodeinfo.get(node, None)
2905 raise errors.OpPrereqError("Cannot get current information"
2906 " from node '%s'" % nodeinfo)
2907 if req_size > info['vg_free']:
2908 raise errors.OpPrereqError("Not enough disk space on target node %s."
2909 " %d MB available, %d MB required" %
2910 (node, info['vg_free'], req_size))
2913 os_obj = rpc.call_os_get([pnode.name], self.op.os_type)[pnode.name]
2914 if not isinstance(os_obj, objects.OS):
2915 raise errors.OpPrereqError("OS '%s' not in supported os list for"
2916 " primary node" % self.op.os_type)
2918 # instance verification
2919 hostname1 = utils.HostInfo(self.op.instance_name)
2921 self.op.instance_name = instance_name = hostname1.name
2922 instance_list = self.cfg.GetInstanceList()
2923 if instance_name in instance_list:
2924 raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
2927 ip = getattr(self.op, "ip", None)
2928 if ip is None or ip.lower() == "none":
2930 elif ip.lower() == "auto":
2931 inst_ip = hostname1.ip
2933 if not utils.IsValidIP(ip):
2934 raise errors.OpPrereqError("given IP address '%s' doesn't look"
2935 " like a valid IP" % ip)
2937 self.inst_ip = inst_ip
2939 if self.op.start and not self.op.ip_check:
2940 raise errors.OpPrereqError("Cannot ignore IP address conflicts when"
2941 " adding an instance in start mode")
2943 if self.op.ip_check:
2944 if utils.TcpPing(utils.HostInfo().name, hostname1.ip,
2945 constants.DEFAULT_NODED_PORT):
2946 raise errors.OpPrereqError("IP %s of instance %s already in use" %
2947 (hostname1.ip, instance_name))
2949 # bridge verification
2950 bridge = getattr(self.op, "bridge", None)
2952 self.op.bridge = self.cfg.GetDefBridge()
2954 self.op.bridge = bridge
2956 if not rpc.call_bridges_exist(self.pnode.name, [self.op.bridge]):
2957 raise errors.OpPrereqError("target bridge '%s' does not exist on"
2958 " destination node '%s'" %
2959 (self.op.bridge, pnode.name))
2962 self.instance_status = 'up'
2964 self.instance_status = 'down'
2966 def Exec(self, feedback_fn):
2967 """Create and add the instance to the cluster.
2970 instance = self.op.instance_name
2971 pnode_name = self.pnode.name
2973 nic = objects.NIC(bridge=self.op.bridge, mac=self.cfg.GenerateMAC())
2974 if self.inst_ip is not None:
2975 nic.ip = self.inst_ip
2977 disks = _GenerateDiskTemplate(self.cfg,
2978 self.op.disk_template,
2979 instance, pnode_name,
2980 self.secondaries, self.op.disk_size,
2983 iobj = objects.Instance(name=instance, os=self.op.os_type,
2984 primary_node=pnode_name,
2985 memory=self.op.mem_size,
2986 vcpus=self.op.vcpus,
2987 nics=[nic], disks=disks,
2988 disk_template=self.op.disk_template,
2989 status=self.instance_status,
2992 feedback_fn("* creating instance disks...")
2993 if not _CreateDisks(self.cfg, iobj):
2994 _RemoveDisks(iobj, self.cfg)
2995 raise errors.OpExecError("Device creation failed, reverting...")
2997 feedback_fn("adding instance %s to cluster config" % instance)
2999 self.cfg.AddInstance(iobj)
3001 if self.op.wait_for_sync:
3002 disk_abort = not _WaitForSync(self.cfg, iobj)
3003 elif iobj.disk_template == constants.DT_REMOTE_RAID1:
3004 # make sure the disks are not degraded (still sync-ing is ok)
3006 feedback_fn("* checking mirrors status")
3007 disk_abort = not _WaitForSync(self.cfg, iobj, oneshot=True)
3012 _RemoveDisks(iobj, self.cfg)
3013 self.cfg.RemoveInstance(iobj.name)
3014 raise errors.OpExecError("There are some degraded disks for"
3017 feedback_fn("creating os for instance %s on node %s" %
3018 (instance, pnode_name))
3020 if iobj.disk_template != constants.DT_DISKLESS:
3021 if self.op.mode == constants.INSTANCE_CREATE:
3022 feedback_fn("* running the instance OS create scripts...")
3023 if not rpc.call_instance_os_add(pnode_name, iobj, "sda", "sdb"):
3024 raise errors.OpExecError("could not add os for instance %s"
3026 (instance, pnode_name))
3028 elif self.op.mode == constants.INSTANCE_IMPORT:
3029 feedback_fn("* running the instance OS import scripts...")
3030 src_node = self.op.src_node
3031 src_image = self.src_image
3032 if not rpc.call_instance_os_import(pnode_name, iobj, "sda", "sdb",
3033 src_node, src_image):
3034 raise errors.OpExecError("Could not import os for instance"
3036 (instance, pnode_name))
3038 # also checked in the prereq part
3039 raise errors.ProgrammerError("Unknown OS initialization mode '%s'"
3043 logger.Info("starting instance %s on node %s" % (instance, pnode_name))
3044 feedback_fn("* starting instance...")
3045 if not rpc.call_instance_start(pnode_name, iobj, None):
3046 raise errors.OpExecError("Could not start instance")
3049 class LUConnectConsole(NoHooksLU):
3050 """Connect to an instance's console.
3052 This is somewhat special in that it returns the command line that
3053 you need to run on the master node in order to connect to the
3057 _OP_REQP = ["instance_name"]
3059 def CheckPrereq(self):
3060 """Check prerequisites.
3062 This checks that the instance is in the cluster.
3065 instance = self.cfg.GetInstanceInfo(
3066 self.cfg.ExpandInstanceName(self.op.instance_name))
3067 if instance is None:
3068 raise errors.OpPrereqError("Instance '%s' not known" %
3069 self.op.instance_name)
3070 self.instance = instance
3072 def Exec(self, feedback_fn):
3073 """Connect to the console of an instance
3076 instance = self.instance
3077 node = instance.primary_node
3079 node_insts = rpc.call_instance_list([node])[node]
3080 if node_insts is False:
3081 raise errors.OpExecError("Can't connect to node %s." % node)
3083 if instance.name not in node_insts:
3084 raise errors.OpExecError("Instance %s is not running." % instance.name)
3086 logger.Debug("connecting to console of %s on %s" % (instance.name, node))
3088 hyper = hypervisor.GetHypervisor()
3089 console_cmd = hyper.GetShellCommandForConsole(instance.name)
3091 argv = ["ssh", "-q", "-t"]
3092 argv.extend(ssh.KNOWN_HOSTS_OPTS)
3093 argv.extend(ssh.BATCH_MODE_OPTS)
3095 argv.append(console_cmd)
3099 class LUAddMDDRBDComponent(LogicalUnit):
3100 """Adda new mirror member to an instance's disk.
3103 HPATH = "mirror-add"
3104 HTYPE = constants.HTYPE_INSTANCE
3105 _OP_REQP = ["instance_name", "remote_node", "disk_name"]
3107 def BuildHooksEnv(self):
3110 This runs on the master, the primary and all the secondaries.
3114 "NEW_SECONDARY": self.op.remote_node,
3115 "DISK_NAME": self.op.disk_name,
3117 env.update(_BuildInstanceHookEnvByObject(self.instance))
3118 nl = [self.sstore.GetMasterNode(), self.instance.primary_node,
3119 self.op.remote_node,] + list(self.instance.secondary_nodes)
3122 def CheckPrereq(self):
3123 """Check prerequisites.
3125 This checks that the instance is in the cluster.
3128 instance = self.cfg.GetInstanceInfo(
3129 self.cfg.ExpandInstanceName(self.op.instance_name))
3130 if instance is None:
3131 raise errors.OpPrereqError("Instance '%s' not known" %
3132 self.op.instance_name)
3133 self.instance = instance
3135 remote_node = self.cfg.ExpandNodeName(self.op.remote_node)
3136 if remote_node is None:
3137 raise errors.OpPrereqError("Node '%s' not known" % self.op.remote_node)
3138 self.remote_node = remote_node
3140 if remote_node == instance.primary_node:
3141 raise errors.OpPrereqError("The specified node is the primary node of"
3144 if instance.disk_template != constants.DT_REMOTE_RAID1:
3145 raise errors.OpPrereqError("Instance's disk layout is not"
3147 for disk in instance.disks:
3148 if disk.iv_name == self.op.disk_name:
3151 raise errors.OpPrereqError("Can't find this device ('%s') in the"
3152 " instance." % self.op.disk_name)
3153 if len(disk.children) > 1:
3154 raise errors.OpPrereqError("The device already has two slave"
3156 "This would create a 3-disk raid1"
3157 " which we don't allow.")
3160 def Exec(self, feedback_fn):
3161 """Add the mirror component
3165 instance = self.instance
3167 remote_node = self.remote_node
3168 lv_names = [".%s_%s" % (disk.iv_name, suf) for suf in ["data", "meta"]]
3169 names = _GenerateUniqueNames(self.cfg, lv_names)
3170 new_drbd = _GenerateMDDRBDBranch(self.cfg, instance.primary_node,
3171 remote_node, disk.size, names)
3173 logger.Info("adding new mirror component on secondary")
3175 if not _CreateBlockDevOnSecondary(self.cfg, remote_node, new_drbd, False,
3176 _GetInstanceInfoText(instance)):
3177 raise errors.OpExecError("Failed to create new component on secondary"
3178 " node %s" % remote_node)
3180 logger.Info("adding new mirror component on primary")
3182 if not _CreateBlockDevOnPrimary(self.cfg, instance.primary_node, new_drbd,
3183 _GetInstanceInfoText(instance)):
3184 # remove secondary dev
3185 self.cfg.SetDiskID(new_drbd, remote_node)
3186 rpc.call_blockdev_remove(remote_node, new_drbd)
3187 raise errors.OpExecError("Failed to create volume on primary")
3189 # the device exists now
3190 # call the primary node to add the mirror to md
3191 logger.Info("adding new mirror component to md")
3192 if not rpc.call_blockdev_addchild(instance.primary_node,
3194 logger.Error("Can't add mirror compoment to md!")
3195 self.cfg.SetDiskID(new_drbd, remote_node)
3196 if not rpc.call_blockdev_remove(remote_node, new_drbd):
3197 logger.Error("Can't rollback on secondary")
3198 self.cfg.SetDiskID(new_drbd, instance.primary_node)
3199 if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
3200 logger.Error("Can't rollback on primary")
3201 raise errors.OpExecError("Can't add mirror component to md array")
3203 disk.children.append(new_drbd)
3205 self.cfg.AddInstance(instance)
3207 _WaitForSync(self.cfg, instance)
3212 class LURemoveMDDRBDComponent(LogicalUnit):
3213 """Remove a component from a remote_raid1 disk.
3216 HPATH = "mirror-remove"
3217 HTYPE = constants.HTYPE_INSTANCE
3218 _OP_REQP = ["instance_name", "disk_name", "disk_id"]
3220 def BuildHooksEnv(self):
3223 This runs on the master, the primary and all the secondaries.
3227 "DISK_NAME": self.op.disk_name,
3228 "DISK_ID": self.op.disk_id,
3229 "OLD_SECONDARY": self.old_secondary,
3231 env.update(_BuildInstanceHookEnvByObject(self.instance))
3232 nl = [self.sstore.GetMasterNode(),
3233 self.instance.primary_node] + list(self.instance.secondary_nodes)
3236 def CheckPrereq(self):
3237 """Check prerequisites.
3239 This checks that the instance is in the cluster.
3242 instance = self.cfg.GetInstanceInfo(
3243 self.cfg.ExpandInstanceName(self.op.instance_name))
3244 if instance is None:
3245 raise errors.OpPrereqError("Instance '%s' not known" %
3246 self.op.instance_name)
3247 self.instance = instance
3249 if instance.disk_template != constants.DT_REMOTE_RAID1:
3250 raise errors.OpPrereqError("Instance's disk layout is not"
3252 for disk in instance.disks:
3253 if disk.iv_name == self.op.disk_name:
3256 raise errors.OpPrereqError("Can't find this device ('%s') in the"
3257 " instance." % self.op.disk_name)
3258 for child in disk.children:
3259 if (child.dev_type == constants.LD_DRBD7 and
3260 child.logical_id[2] == self.op.disk_id):
3263 raise errors.OpPrereqError("Can't find the device with this port.")
3265 if len(disk.children) < 2:
3266 raise errors.OpPrereqError("Cannot remove the last component from"
3270 if self.child.logical_id[0] == instance.primary_node:
3274 self.old_secondary = self.child.logical_id[oid]
3276 def Exec(self, feedback_fn):
3277 """Remove the mirror component
3280 instance = self.instance
3283 logger.Info("remove mirror component")
3284 self.cfg.SetDiskID(disk, instance.primary_node)
3285 if not rpc.call_blockdev_removechild(instance.primary_node,
3287 raise errors.OpExecError("Can't remove child from mirror.")
3289 for node in child.logical_id[:2]:
3290 self.cfg.SetDiskID(child, node)
3291 if not rpc.call_blockdev_remove(node, child):
3292 logger.Error("Warning: failed to remove device from node %s,"
3293 " continuing operation." % node)
3295 disk.children.remove(child)
3296 self.cfg.AddInstance(instance)
3299 class LUReplaceDisks(LogicalUnit):
3300 """Replace the disks of an instance.
3303 HPATH = "mirrors-replace"
3304 HTYPE = constants.HTYPE_INSTANCE
3305 _OP_REQP = ["instance_name"]
3307 def BuildHooksEnv(self):
3310 This runs on the master, the primary and all the secondaries.
3314 "NEW_SECONDARY": self.op.remote_node,
3315 "OLD_SECONDARY": self.instance.secondary_nodes[0],
3317 env.update(_BuildInstanceHookEnvByObject(self.instance))
3318 nl = [self.sstore.GetMasterNode(),
3319 self.instance.primary_node] + list(self.instance.secondary_nodes)
3322 def CheckPrereq(self):
3323 """Check prerequisites.
3325 This checks that the instance is in the cluster.
3328 instance = self.cfg.GetInstanceInfo(
3329 self.cfg.ExpandInstanceName(self.op.instance_name))
3330 if instance is None:
3331 raise errors.OpPrereqError("Instance '%s' not known" %
3332 self.op.instance_name)
3333 self.instance = instance
3335 if instance.disk_template != constants.DT_REMOTE_RAID1:
3336 raise errors.OpPrereqError("Instance's disk layout is not"
3339 if len(instance.secondary_nodes) != 1:
3340 raise errors.OpPrereqError("The instance has a strange layout,"
3341 " expected one secondary but found %d" %
3342 len(instance.secondary_nodes))
3344 remote_node = getattr(self.op, "remote_node", None)
3345 if remote_node is None:
3346 remote_node = instance.secondary_nodes[0]
3348 remote_node = self.cfg.ExpandNodeName(remote_node)
3349 if remote_node is None:
3350 raise errors.OpPrereqError("Node '%s' not known" %
3351 self.op.remote_node)
3352 if remote_node == instance.primary_node:
3353 raise errors.OpPrereqError("The specified node is the primary node of"
3355 self.op.remote_node = remote_node
3357 def Exec(self, feedback_fn):
3358 """Replace the disks of an instance.
3361 instance = self.instance
3364 remote_node = self.op.remote_node
3366 for dev in instance.disks:
3368 lv_names = [".%s_%s" % (dev.iv_name, suf) for suf in ["data", "meta"]]
3369 names = _GenerateUniqueNames(cfg, lv_names)
3370 new_drbd = _GenerateMDDRBDBranch(cfg, instance.primary_node,
3371 remote_node, size, names)
3372 iv_names[dev.iv_name] = (dev, dev.children[0], new_drbd)
3373 logger.Info("adding new mirror component on secondary for %s" %
3376 if not _CreateBlockDevOnSecondary(cfg, remote_node, new_drbd, False,
3377 _GetInstanceInfoText(instance)):
3378 raise errors.OpExecError("Failed to create new component on"
3379 " secondary node %s\n"
3380 "Full abort, cleanup manually!" %
3383 logger.Info("adding new mirror component on primary")
3385 if not _CreateBlockDevOnPrimary(cfg, instance.primary_node, new_drbd,
3386 _GetInstanceInfoText(instance)):
3387 # remove secondary dev
3388 cfg.SetDiskID(new_drbd, remote_node)
3389 rpc.call_blockdev_remove(remote_node, new_drbd)
3390 raise errors.OpExecError("Failed to create volume on primary!\n"
3391 "Full abort, cleanup manually!!")
3393 # the device exists now
3394 # call the primary node to add the mirror to md
3395 logger.Info("adding new mirror component to md")
3396 if not rpc.call_blockdev_addchild(instance.primary_node, dev,
3398 logger.Error("Can't add mirror compoment to md!")
3399 cfg.SetDiskID(new_drbd, remote_node)
3400 if not rpc.call_blockdev_remove(remote_node, new_drbd):
3401 logger.Error("Can't rollback on secondary")
3402 cfg.SetDiskID(new_drbd, instance.primary_node)
3403 if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
3404 logger.Error("Can't rollback on primary")
3405 raise errors.OpExecError("Full abort, cleanup manually!!")
3407 dev.children.append(new_drbd)
3408 cfg.AddInstance(instance)
3410 # this can fail as the old devices are degraded and _WaitForSync
3411 # does a combined result over all disks, so we don't check its
3413 _WaitForSync(cfg, instance, unlock=True)
3415 # so check manually all the devices
3416 for name in iv_names:
3417 dev, child, new_drbd = iv_names[name]
3418 cfg.SetDiskID(dev, instance.primary_node)
3419 is_degr = rpc.call_blockdev_find(instance.primary_node, dev)[5]
3421 raise errors.OpExecError("MD device %s is degraded!" % name)
3422 cfg.SetDiskID(new_drbd, instance.primary_node)
3423 is_degr = rpc.call_blockdev_find(instance.primary_node, new_drbd)[5]
3425 raise errors.OpExecError("New drbd device %s is degraded!" % name)
3427 for name in iv_names:
3428 dev, child, new_drbd = iv_names[name]
3429 logger.Info("remove mirror %s component" % name)
3430 cfg.SetDiskID(dev, instance.primary_node)
3431 if not rpc.call_blockdev_removechild(instance.primary_node,
3433 logger.Error("Can't remove child from mirror, aborting"
3434 " *this device cleanup*.\nYou need to cleanup manually!!")
3437 for node in child.logical_id[:2]:
3438 logger.Info("remove child device on %s" % node)
3439 cfg.SetDiskID(child, node)
3440 if not rpc.call_blockdev_remove(node, child):
3441 logger.Error("Warning: failed to remove device from node %s,"
3442 " continuing operation." % node)
3444 dev.children.remove(child)
3446 cfg.AddInstance(instance)
3449 class LUQueryInstanceData(NoHooksLU):
3450 """Query runtime instance data.
3453 _OP_REQP = ["instances"]
3455 def CheckPrereq(self):
3456 """Check prerequisites.
3458 This only checks the optional instance list against the existing names.
3461 if not isinstance(self.op.instances, list):
3462 raise errors.OpPrereqError("Invalid argument type 'instances'")
3463 if self.op.instances:
3464 self.wanted_instances = []
3465 names = self.op.instances
3467 instance = self.cfg.GetInstanceInfo(self.cfg.ExpandInstanceName(name))
3468 if instance is None:
3469 raise errors.OpPrereqError("No such instance name '%s'" % name)
3470 self.wanted_instances.append(instance)
3472 self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
3473 in self.cfg.GetInstanceList()]
3477 def _ComputeDiskStatus(self, instance, snode, dev):
3478 """Compute block device status.
3481 self.cfg.SetDiskID(dev, instance.primary_node)
3482 dev_pstatus = rpc.call_blockdev_find(instance.primary_node, dev)
3483 if dev.dev_type == constants.LD_DRBD7:
3484 # we change the snode then (otherwise we use the one passed in)
3485 if dev.logical_id[0] == instance.primary_node:
3486 snode = dev.logical_id[1]
3488 snode = dev.logical_id[0]
3491 self.cfg.SetDiskID(dev, snode)
3492 dev_sstatus = rpc.call_blockdev_find(snode, dev)
3497 dev_children = [self._ComputeDiskStatus(instance, snode, child)
3498 for child in dev.children]
3503 "iv_name": dev.iv_name,
3504 "dev_type": dev.dev_type,
3505 "logical_id": dev.logical_id,
3506 "physical_id": dev.physical_id,
3507 "pstatus": dev_pstatus,
3508 "sstatus": dev_sstatus,
3509 "children": dev_children,
3514 def Exec(self, feedback_fn):
3515 """Gather and return data"""
3517 for instance in self.wanted_instances:
3518 remote_info = rpc.call_instance_info(instance.primary_node,
3520 if remote_info and "state" in remote_info:
3523 remote_state = "down"
3524 if instance.status == "down":
3525 config_state = "down"
3529 disks = [self._ComputeDiskStatus(instance, None, device)
3530 for device in instance.disks]
3533 "name": instance.name,
3534 "config_state": config_state,
3535 "run_state": remote_state,
3536 "pnode": instance.primary_node,
3537 "snodes": instance.secondary_nodes,
3539 "memory": instance.memory,
3540 "nics": [(nic.mac, nic.ip, nic.bridge) for nic in instance.nics],
3542 "vcpus": instance.vcpus,
3545 result[instance.name] = idict
3550 class LUSetInstanceParms(LogicalUnit):
3551 """Modifies an instances's parameters.
3554 HPATH = "instance-modify"
3555 HTYPE = constants.HTYPE_INSTANCE
3556 _OP_REQP = ["instance_name"]
3558 def BuildHooksEnv(self):
3561 This runs on the master, primary and secondaries.
3566 args['memory'] = self.mem
3568 args['vcpus'] = self.vcpus
3569 if self.do_ip or self.do_bridge:
3573 ip = self.instance.nics[0].ip
3575 bridge = self.bridge
3577 bridge = self.instance.nics[0].bridge
3578 args['nics'] = [(ip, bridge)]
3579 env = _BuildInstanceHookEnvByObject(self.instance, override=args)
3580 nl = [self.sstore.GetMasterNode(),
3581 self.instance.primary_node] + list(self.instance.secondary_nodes)
3584 def CheckPrereq(self):
3585 """Check prerequisites.
3587 This only checks the instance list against the existing names.
3590 self.mem = getattr(self.op, "mem", None)
3591 self.vcpus = getattr(self.op, "vcpus", None)
3592 self.ip = getattr(self.op, "ip", None)
3593 self.bridge = getattr(self.op, "bridge", None)
3594 if [self.mem, self.vcpus, self.ip, self.bridge].count(None) == 4:
3595 raise errors.OpPrereqError("No changes submitted")
3596 if self.mem is not None:
3598 self.mem = int(self.mem)
3599 except ValueError, err:
3600 raise errors.OpPrereqError("Invalid memory size: %s" % str(err))
3601 if self.vcpus is not None:
3603 self.vcpus = int(self.vcpus)
3604 except ValueError, err:
3605 raise errors.OpPrereqError("Invalid vcpus number: %s" % str(err))
3606 if self.ip is not None:
3608 if self.ip.lower() == "none":
3611 if not utils.IsValidIP(self.ip):
3612 raise errors.OpPrereqError("Invalid IP address '%s'." % self.ip)
3615 self.do_bridge = (self.bridge is not None)
3617 instance = self.cfg.GetInstanceInfo(
3618 self.cfg.ExpandInstanceName(self.op.instance_name))
3619 if instance is None:
3620 raise errors.OpPrereqError("No such instance name '%s'" %
3621 self.op.instance_name)
3622 self.op.instance_name = instance.name
3623 self.instance = instance
3626 def Exec(self, feedback_fn):
3627 """Modifies an instance.
3629 All parameters take effect only at the next restart of the instance.
3632 instance = self.instance
3634 instance.memory = self.mem
3635 result.append(("mem", self.mem))
3637 instance.vcpus = self.vcpus
3638 result.append(("vcpus", self.vcpus))
3640 instance.nics[0].ip = self.ip
3641 result.append(("ip", self.ip))
3643 instance.nics[0].bridge = self.bridge
3644 result.append(("bridge", self.bridge))
3646 self.cfg.AddInstance(instance)
3651 class LUQueryExports(NoHooksLU):
3652 """Query the exports list
3657 def CheckPrereq(self):
3658 """Check that the nodelist contains only existing nodes.
3661 self.nodes = _GetWantedNodes(self, getattr(self.op, "nodes", None))
3663 def Exec(self, feedback_fn):
3664 """Compute the list of all the exported system images.
3667 a dictionary with the structure node->(export-list)
3668 where export-list is a list of the instances exported on
3672 return rpc.call_export_list(self.nodes)
3675 class LUExportInstance(LogicalUnit):
3676 """Export an instance to an image in the cluster.
3679 HPATH = "instance-export"
3680 HTYPE = constants.HTYPE_INSTANCE
3681 _OP_REQP = ["instance_name", "target_node", "shutdown"]
3683 def BuildHooksEnv(self):
3686 This will run on the master, primary node and target node.
3690 "EXPORT_NODE": self.op.target_node,
3691 "EXPORT_DO_SHUTDOWN": self.op.shutdown,
3693 env.update(_BuildInstanceHookEnvByObject(self.instance))
3694 nl = [self.sstore.GetMasterNode(), self.instance.primary_node,
3695 self.op.target_node]
3698 def CheckPrereq(self):
3699 """Check prerequisites.
3701 This checks that the instance name is a valid one.
3704 instance_name = self.cfg.ExpandInstanceName(self.op.instance_name)
3705 self.instance = self.cfg.GetInstanceInfo(instance_name)
3706 if self.instance is None:
3707 raise errors.OpPrereqError("Instance '%s' not found" %
3708 self.op.instance_name)
3711 dst_node_short = self.cfg.ExpandNodeName(self.op.target_node)
3712 self.dst_node = self.cfg.GetNodeInfo(dst_node_short)
3714 if self.dst_node is None:
3715 raise errors.OpPrereqError("Destination node '%s' is unknown." %
3716 self.op.target_node)
3717 self.op.target_node = self.dst_node.name
3719 def Exec(self, feedback_fn):
3720 """Export an instance to an image in the cluster.
3723 instance = self.instance
3724 dst_node = self.dst_node
3725 src_node = instance.primary_node
3726 # shutdown the instance, unless requested not to do so
3727 if self.op.shutdown:
3728 op = opcodes.OpShutdownInstance(instance_name=instance.name)
3729 self.processor.ChainOpCode(op, feedback_fn)
3731 vgname = self.cfg.GetVGName()
3736 for disk in instance.disks:
3737 if disk.iv_name == "sda":
3738 # new_dev_name will be a snapshot of an lvm leaf of the one we passed
3739 new_dev_name = rpc.call_blockdev_snapshot(src_node, disk)
3741 if not new_dev_name:
3742 logger.Error("could not snapshot block device %s on node %s" %
3743 (disk.logical_id[1], src_node))
3745 new_dev = objects.Disk(dev_type=constants.LD_LV, size=disk.size,
3746 logical_id=(vgname, new_dev_name),
3747 physical_id=(vgname, new_dev_name),
3748 iv_name=disk.iv_name)
3749 snap_disks.append(new_dev)
3752 if self.op.shutdown:
3753 op = opcodes.OpStartupInstance(instance_name=instance.name,
3755 self.processor.ChainOpCode(op, feedback_fn)
3757 # TODO: check for size
3759 for dev in snap_disks:
3760 if not rpc.call_snapshot_export(src_node, dev, dst_node.name,
3762 logger.Error("could not export block device %s from node"
3764 (dev.logical_id[1], src_node, dst_node.name))
3765 if not rpc.call_blockdev_remove(src_node, dev):
3766 logger.Error("could not remove snapshot block device %s from"
3767 " node %s" % (dev.logical_id[1], src_node))
3769 if not rpc.call_finalize_export(dst_node.name, instance, snap_disks):
3770 logger.Error("could not finalize export for instance %s on node %s" %
3771 (instance.name, dst_node.name))
3773 nodelist = self.cfg.GetNodeList()
3774 nodelist.remove(dst_node.name)
3776 # on one-node clusters nodelist will be empty after the removal
3777 # if we proceed the backup would be removed because OpQueryExports
3778 # substitutes an empty list with the full cluster node list.
3780 op = opcodes.OpQueryExports(nodes=nodelist)
3781 exportlist = self.processor.ChainOpCode(op, feedback_fn)
3782 for node in exportlist:
3783 if instance.name in exportlist[node]:
3784 if not rpc.call_export_remove(node, instance.name):
3785 logger.Error("could not remove older export for instance %s"
3786 " on node %s" % (instance.name, node))
3789 class TagsLU(NoHooksLU):
3792 This is an abstract class which is the parent of all the other tags LUs.
3795 def CheckPrereq(self):
3796 """Check prerequisites.
3799 if self.op.kind == constants.TAG_CLUSTER:
3800 self.target = self.cfg.GetClusterInfo()
3801 elif self.op.kind == constants.TAG_NODE:
3802 name = self.cfg.ExpandNodeName(self.op.name)
3804 raise errors.OpPrereqError("Invalid node name (%s)" %
3807 self.target = self.cfg.GetNodeInfo(name)
3808 elif self.op.kind == constants.TAG_INSTANCE:
3809 name = self.cfg.ExpandInstanceName(self.op.name)
3811 raise errors.OpPrereqError("Invalid instance name (%s)" %
3814 self.target = self.cfg.GetInstanceInfo(name)
3816 raise errors.OpPrereqError("Wrong tag type requested (%s)" %
3820 class LUGetTags(TagsLU):
3821 """Returns the tags of a given object.
3824 _OP_REQP = ["kind", "name"]
3826 def Exec(self, feedback_fn):
3827 """Returns the tag list.
3830 return self.target.GetTags()
3833 class LUAddTags(TagsLU):
3834 """Sets a tag on a given object.
3837 _OP_REQP = ["kind", "name", "tags"]
3839 def CheckPrereq(self):
3840 """Check prerequisites.
3842 This checks the type and length of the tag name and value.
3845 TagsLU.CheckPrereq(self)
3846 for tag in self.op.tags:
3847 objects.TaggableObject.ValidateTag(tag)
3849 def Exec(self, feedback_fn):
3854 for tag in self.op.tags:
3855 self.target.AddTag(tag)
3856 except errors.TagError, err:
3857 raise errors.OpExecError("Error while setting tag: %s" % str(err))
3859 self.cfg.Update(self.target)
3860 except errors.ConfigurationError:
3861 raise errors.OpRetryError("There has been a modification to the"
3862 " config file and the operation has been"
3863 " aborted. Please retry.")
3866 class LUDelTags(TagsLU):
3867 """Delete a list of tags from a given object.
3870 _OP_REQP = ["kind", "name", "tags"]
3872 def CheckPrereq(self):
3873 """Check prerequisites.
3875 This checks that we have the given tag.
3878 TagsLU.CheckPrereq(self)
3879 for tag in self.op.tags:
3880 objects.TaggableObject.ValidateTag(tag)
3881 del_tags = frozenset(self.op.tags)
3882 cur_tags = self.target.GetTags()
3883 if not del_tags <= cur_tags:
3884 diff_tags = del_tags - cur_tags
3885 diff_names = ["'%s'" % tag for tag in diff_tags]
3887 raise errors.OpPrereqError("Tag(s) %s not found" %
3888 (",".join(diff_names)))
3890 def Exec(self, feedback_fn):
3891 """Remove the tag from the object.
3894 for tag in self.op.tags:
3895 self.target.RemoveTag(tag)
3897 self.cfg.Update(self.target)
3898 except errors.ConfigurationError:
3899 raise errors.OpRetryError("There has been a modification to the"
3900 " config file and the operation has been"
3901 " aborted. Please retry.")