4 # Copyright (C) 2006, 2007 Google Inc.
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
11 # This program is distributed in the hope that it will be useful, but
12 # WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 # General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
22 """Module implementing the master-side code."""
24 # pylint: disable-msg=W0613,W0201
34 from ganeti import rpc
35 from ganeti import ssh
36 from ganeti import logger
37 from ganeti import utils
38 from ganeti import errors
39 from ganeti import hypervisor
40 from ganeti import config
41 from ganeti import constants
42 from ganeti import objects
43 from ganeti import opcodes
44 from ganeti import ssconf
46 class LogicalUnit(object):
47 """Logical Unit base class.
49 Subclasses must follow these rules:
50 - implement CheckPrereq which also fills in the opcode instance
51 with all the fields (even if as None)
53 - implement BuildHooksEnv
54 - redefine HPATH and HTYPE
55 - optionally redefine their run requirements (REQ_CLUSTER,
56 REQ_MASTER); note that all commands require root permissions
65 def __init__(self, processor, op, cfg, sstore):
66 """Constructor for LogicalUnit.
68 This needs to be overriden in derived classes in order to check op
76 for attr_name in self._OP_REQP:
77 attr_val = getattr(op, attr_name, None)
79 raise errors.OpPrereqError("Required parameter '%s' missing" %
82 if not cfg.IsCluster():
83 raise errors.OpPrereqError("Cluster not initialized yet,"
84 " use 'gnt-cluster init' first.")
86 master = sstore.GetMasterNode()
87 if master != utils.HostInfo().name:
88 raise errors.OpPrereqError("Commands must be run on the master"
91 def CheckPrereq(self):
92 """Check prerequisites for this LU.
94 This method should check that the prerequisites for the execution
95 of this LU are fulfilled. It can do internode communication, but
96 it should be idempotent - no cluster or system changes are
99 The method should raise errors.OpPrereqError in case something is
100 not fulfilled. Its return value is ignored.
102 This method should also update all the parameters of the opcode to
103 their canonical form; e.g. a short node name must be fully
104 expanded after this method has successfully completed (so that
105 hooks, logging, etc. work correctly).
108 raise NotImplementedError
110 def Exec(self, feedback_fn):
113 This method should implement the actual work. It should raise
114 errors.OpExecError for failures that are somewhat dealt with in
118 raise NotImplementedError
120 def BuildHooksEnv(self):
121 """Build hooks environment for this LU.
123 This method should return a three-node tuple consisting of: a dict
124 containing the environment that will be used for running the
125 specific hook for this LU, a list of node names on which the hook
126 should run before the execution, and a list of node names on which
127 the hook should run after the execution.
129 The keys of the dict must not have 'GANETI_' prefixed as this will
130 be handled in the hooks runner. Also note additional keys will be
131 added by the hooks runner. If the LU doesn't define any
132 environment, an empty dict (and not None) should be returned.
134 As for the node lists, the master should not be included in the
135 them, as it will be added by the hooks runner in case this LU
136 requires a cluster to run on (otherwise we don't have a node
137 list). No nodes should be returned as an empty list (and not
140 Note that if the HPATH for a LU class is None, this function will
144 raise NotImplementedError
147 class NoHooksLU(LogicalUnit):
148 """Simple LU which runs no hooks.
150 This LU is intended as a parent for other LogicalUnits which will
151 run no hooks, in order to reduce duplicate code.
157 def BuildHooksEnv(self):
160 This is a no-op, since we don't run hooks.
166 def _RemoveHostFromEtcHosts(hostname):
167 """Wrapper around utils.RemoteEtcHostsEntry.
170 hi = utils.HostInfo(name=hostname)
171 utils.RemoveEtcHostsEntry(constants.ETC_HOSTS, hi.name)
172 utils.RemoveEtcHostsEntry(constants.ETC_HOSTS, hi.ShortName())
175 def _GetWantedNodes(lu, nodes):
176 """Returns list of checked and expanded node names.
179 nodes: List of nodes (strings) or None for all
182 if not isinstance(nodes, list):
183 raise errors.OpPrereqError("Invalid argument type 'nodes'")
189 node = lu.cfg.ExpandNodeName(name)
191 raise errors.OpPrereqError("No such node name '%s'" % name)
195 wanted = lu.cfg.GetNodeList()
196 return utils.NiceSort(wanted)
199 def _GetWantedInstances(lu, instances):
200 """Returns list of checked and expanded instance names.
203 instances: List of instances (strings) or None for all
206 if not isinstance(instances, list):
207 raise errors.OpPrereqError("Invalid argument type 'instances'")
212 for name in instances:
213 instance = lu.cfg.ExpandInstanceName(name)
215 raise errors.OpPrereqError("No such instance name '%s'" % name)
216 wanted.append(instance)
219 wanted = lu.cfg.GetInstanceList()
220 return utils.NiceSort(wanted)
223 def _CheckOutputFields(static, dynamic, selected):
224 """Checks whether all selected fields are valid.
227 static: Static fields
228 dynamic: Dynamic fields
231 static_fields = frozenset(static)
232 dynamic_fields = frozenset(dynamic)
234 all_fields = static_fields | dynamic_fields
236 if not all_fields.issuperset(selected):
237 raise errors.OpPrereqError("Unknown output fields selected: %s"
238 % ",".join(frozenset(selected).
239 difference(all_fields)))
242 def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
243 memory, vcpus, nics):
244 """Builds instance related env variables for hooks from single variables.
247 secondary_nodes: List of secondary nodes as strings
251 "INSTANCE_NAME": name,
252 "INSTANCE_PRIMARY": primary_node,
253 "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
254 "INSTANCE_OS_TYPE": os_type,
255 "INSTANCE_STATUS": status,
256 "INSTANCE_MEMORY": memory,
257 "INSTANCE_VCPUS": vcpus,
261 nic_count = len(nics)
262 for idx, (ip, bridge) in enumerate(nics):
265 env["INSTANCE_NIC%d_IP" % idx] = ip
266 env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
270 env["INSTANCE_NIC_COUNT"] = nic_count
275 def _BuildInstanceHookEnvByObject(instance, override=None):
276 """Builds instance related env variables for hooks from an object.
279 instance: objects.Instance object of instance
280 override: dict of values to override
283 'name': instance.name,
284 'primary_node': instance.primary_node,
285 'secondary_nodes': instance.secondary_nodes,
286 'os_type': instance.os,
287 'status': instance.os,
288 'memory': instance.memory,
289 'vcpus': instance.vcpus,
290 'nics': [(nic.ip, nic.bridge) for nic in instance.nics],
293 args.update(override)
294 return _BuildInstanceHookEnv(**args)
297 def _UpdateKnownHosts(fullnode, ip, pubkey):
298 """Ensure a node has a correct known_hosts entry.
301 fullnode - Fully qualified domain name of host. (str)
302 ip - IPv4 address of host (str)
303 pubkey - the public key of the cluster
306 if os.path.exists(constants.SSH_KNOWN_HOSTS_FILE):
307 f = open(constants.SSH_KNOWN_HOSTS_FILE, 'r+')
309 f = open(constants.SSH_KNOWN_HOSTS_FILE, 'w+')
318 logger.Debug('read %s' % (repr(rawline),))
320 parts = rawline.rstrip('\r\n').split()
322 # Ignore unwanted lines
323 if len(parts) >= 3 and not rawline.lstrip()[0] == '#':
324 fields = parts[0].split(',')
329 for spec in [ ip, fullnode ]:
330 if spec not in fields:
335 logger.Debug("key, pubkey = %s." % (repr((key, pubkey)),))
336 if haveall and key == pubkey:
338 save_lines.append(rawline)
339 logger.Debug("Keeping known_hosts '%s'." % (repr(rawline),))
342 if havesome and (not haveall or key != pubkey):
344 logger.Debug("Discarding known_hosts '%s'." % (repr(rawline),))
347 save_lines.append(rawline)
350 add_lines.append('%s,%s ssh-rsa %s\n' % (fullnode, ip, pubkey))
351 logger.Debug("Adding known_hosts '%s'." % (repr(add_lines[-1]),))
354 save_lines = save_lines + add_lines
356 # Write a new file and replace old.
357 fd, tmpname = tempfile.mkstemp('.tmp', 'known_hosts.',
359 newfile = os.fdopen(fd, 'w')
361 newfile.write(''.join(save_lines))
364 logger.Debug("Wrote new known_hosts.")
365 os.rename(tmpname, constants.SSH_KNOWN_HOSTS_FILE)
368 # Simply appending a new line will do the trick.
370 for add in add_lines:
376 def _HasValidVG(vglist, vgname):
377 """Checks if the volume group list is valid.
379 A non-None return value means there's an error, and the return value
380 is the error message.
383 vgsize = vglist.get(vgname, None)
385 return "volume group '%s' missing" % vgname
387 return ("volume group '%s' too small (20480MiB required, %dMib found)" %
392 def _InitSSHSetup(node):
393 """Setup the SSH configuration for the cluster.
396 This generates a dsa keypair for root, adds the pub key to the
397 permitted hosts and adds the hostkey to its own known hosts.
400 node: the name of this host as a fqdn
403 priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
405 for name in priv_key, pub_key:
406 if os.path.exists(name):
407 utils.CreateBackup(name)
408 utils.RemoveFile(name)
410 result = utils.RunCmd(["ssh-keygen", "-t", "dsa",
414 raise errors.OpExecError("Could not generate ssh keypair, error %s" %
417 f = open(pub_key, 'r')
419 utils.AddAuthorizedKey(auth_keys, f.read(8192))
424 def _InitGanetiServerSetup(ss):
425 """Setup the necessary configuration for the initial node daemon.
427 This creates the nodepass file containing the shared password for
428 the cluster and also generates the SSL certificate.
431 # Create pseudo random password
432 randpass = sha.new(os.urandom(64)).hexdigest()
433 # and write it into sstore
434 ss.SetKey(ss.SS_NODED_PASS, randpass)
436 result = utils.RunCmd(["openssl", "req", "-new", "-newkey", "rsa:1024",
437 "-days", str(365*5), "-nodes", "-x509",
438 "-keyout", constants.SSL_CERT_FILE,
439 "-out", constants.SSL_CERT_FILE, "-batch"])
441 raise errors.OpExecError("could not generate server ssl cert, command"
442 " %s had exitcode %s and error message %s" %
443 (result.cmd, result.exit_code, result.output))
445 os.chmod(constants.SSL_CERT_FILE, 0400)
447 result = utils.RunCmd([constants.NODE_INITD_SCRIPT, "restart"])
450 raise errors.OpExecError("Could not start the node daemon, command %s"
451 " had exitcode %s and error %s" %
452 (result.cmd, result.exit_code, result.output))
455 def _CheckInstanceBridgesExist(instance):
456 """Check that the brigdes needed by an instance exist.
459 # check bridges existance
460 brlist = [nic.bridge for nic in instance.nics]
461 if not rpc.call_bridges_exist(instance.primary_node, brlist):
462 raise errors.OpPrereqError("one or more target bridges %s does not"
463 " exist on destination node '%s'" %
464 (brlist, instance.primary_node))
467 class LUInitCluster(LogicalUnit):
468 """Initialise the cluster.
471 HPATH = "cluster-init"
472 HTYPE = constants.HTYPE_CLUSTER
473 _OP_REQP = ["cluster_name", "hypervisor_type", "vg_name", "mac_prefix",
474 "def_bridge", "master_netdev"]
477 def BuildHooksEnv(self):
480 Notes: Since we don't require a cluster, we must manually add
481 ourselves in the post-run node list.
484 env = {"OP_TARGET": self.op.cluster_name}
485 return env, [], [self.hostname.name]
487 def CheckPrereq(self):
488 """Verify that the passed name is a valid one.
491 if config.ConfigWriter.IsCluster():
492 raise errors.OpPrereqError("Cluster is already initialised")
494 self.hostname = hostname = utils.HostInfo()
496 if hostname.ip.startswith("127."):
497 raise errors.OpPrereqError("This host's IP resolves to the private"
498 " range (%s). Please fix DNS or /etc/hosts." %
501 self.clustername = clustername = utils.HostInfo(self.op.cluster_name)
503 if not utils.TcpPing(constants.LOCALHOST_IP_ADDRESS, hostname.ip,
504 constants.DEFAULT_NODED_PORT):
505 raise errors.OpPrereqError("Inconsistency: this host's name resolves"
506 " to %s,\nbut this ip address does not"
507 " belong to this host."
508 " Aborting." % hostname.ip)
510 secondary_ip = getattr(self.op, "secondary_ip", None)
511 if secondary_ip and not utils.IsValidIP(secondary_ip):
512 raise errors.OpPrereqError("Invalid secondary ip given")
514 secondary_ip != hostname.ip and
515 (not utils.TcpPing(constants.LOCALHOST_IP_ADDRESS, secondary_ip,
516 constants.DEFAULT_NODED_PORT))):
517 raise errors.OpPrereqError("You gave %s as secondary IP,\n"
518 "but it does not belong to this host." %
520 self.secondary_ip = secondary_ip
522 # checks presence of the volume group given
523 vgstatus = _HasValidVG(utils.ListVolumeGroups(), self.op.vg_name)
526 raise errors.OpPrereqError("Error: %s" % vgstatus)
528 if not re.match("^[0-9a-z]{2}:[0-9a-z]{2}:[0-9a-z]{2}$",
530 raise errors.OpPrereqError("Invalid mac prefix given '%s'" %
533 if self.op.hypervisor_type not in hypervisor.VALID_HTYPES:
534 raise errors.OpPrereqError("Invalid hypervisor type given '%s'" %
535 self.op.hypervisor_type)
537 result = utils.RunCmd(["ip", "link", "show", "dev", self.op.master_netdev])
539 raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
540 (self.op.master_netdev,
541 result.output.strip()))
543 if not (os.path.isfile(constants.NODE_INITD_SCRIPT) and
544 os.access(constants.NODE_INITD_SCRIPT, os.X_OK)):
545 raise errors.OpPrereqError("Init.d script '%s' missing or not "
546 "executable." % constants.NODE_INITD_SCRIPT)
548 def Exec(self, feedback_fn):
549 """Initialize the cluster.
552 clustername = self.clustername
553 hostname = self.hostname
555 # set up the simple store
556 self.sstore = ss = ssconf.SimpleStore()
557 ss.SetKey(ss.SS_HYPERVISOR, self.op.hypervisor_type)
558 ss.SetKey(ss.SS_MASTER_NODE, hostname.name)
559 ss.SetKey(ss.SS_MASTER_IP, clustername.ip)
560 ss.SetKey(ss.SS_MASTER_NETDEV, self.op.master_netdev)
561 ss.SetKey(ss.SS_CLUSTER_NAME, clustername.name)
563 # set up the inter-node password and certificate
564 _InitGanetiServerSetup(ss)
566 # start the master ip
567 rpc.call_node_start_master(hostname.name)
569 # set up ssh config and /etc/hosts
570 f = open(constants.SSH_HOST_RSA_PUB, 'r')
575 sshkey = sshline.split(" ")[1]
577 hi = utils.HostInfo(name=hostname.name)
578 utils.AddEtcHostsEntry(constants.ETC_HOSTS, hostname.name, hi.ip)
579 utils.AddEtcHostsEntry(constants.ETC_HOSTS, hi.ShortName(), hi.ip)
582 _UpdateKnownHosts(hostname.name, hostname.ip, sshkey)
584 _InitSSHSetup(hostname.name)
586 # init of cluster config file
587 self.cfg = cfgw = config.ConfigWriter()
588 cfgw.InitConfig(hostname.name, hostname.ip, self.secondary_ip,
589 sshkey, self.op.mac_prefix,
590 self.op.vg_name, self.op.def_bridge)
593 class LUDestroyCluster(NoHooksLU):
594 """Logical unit for destroying the cluster.
599 def CheckPrereq(self):
600 """Check prerequisites.
602 This checks whether the cluster is empty.
604 Any errors are signalled by raising errors.OpPrereqError.
607 master = self.sstore.GetMasterNode()
609 nodelist = self.cfg.GetNodeList()
610 if len(nodelist) != 1 or nodelist[0] != master:
611 raise errors.OpPrereqError("There are still %d node(s) in"
612 " this cluster." % (len(nodelist) - 1))
613 instancelist = self.cfg.GetInstanceList()
615 raise errors.OpPrereqError("There are still %d instance(s) in"
616 " this cluster." % len(instancelist))
618 def Exec(self, feedback_fn):
619 """Destroys the cluster.
622 master = self.sstore.GetMasterNode()
623 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
624 utils.CreateBackup(priv_key)
625 utils.CreateBackup(pub_key)
626 rpc.call_node_leave_cluster(master)
627 _RemoveHostFromEtcHosts(master)
630 class LUVerifyCluster(NoHooksLU):
631 """Verifies the cluster status.
636 def _VerifyNode(self, node, file_list, local_cksum, vglist, node_result,
637 remote_version, feedback_fn):
638 """Run multiple tests against a node.
641 - compares ganeti version
642 - checks vg existance and size > 20G
643 - checks config file checksum
644 - checks ssh to other nodes
647 node: name of the node to check
648 file_list: required list of files
649 local_cksum: dictionary of local files and their checksums
652 # compares ganeti version
653 local_version = constants.PROTOCOL_VERSION
654 if not remote_version:
655 feedback_fn(" - ERROR: connection to %s failed" % (node))
658 if local_version != remote_version:
659 feedback_fn(" - ERROR: sw version mismatch: master %s, node(%s) %s" %
660 (local_version, node, remote_version))
663 # checks vg existance and size > 20G
667 feedback_fn(" - ERROR: unable to check volume groups on node %s." %
671 vgstatus = _HasValidVG(vglist, self.cfg.GetVGName())
673 feedback_fn(" - ERROR: %s on node %s" % (vgstatus, node))
676 # checks config file checksum
679 if 'filelist' not in node_result:
681 feedback_fn(" - ERROR: node hasn't returned file checksum data")
683 remote_cksum = node_result['filelist']
684 for file_name in file_list:
685 if file_name not in remote_cksum:
687 feedback_fn(" - ERROR: file '%s' missing" % file_name)
688 elif remote_cksum[file_name] != local_cksum[file_name]:
690 feedback_fn(" - ERROR: file '%s' has wrong checksum" % file_name)
692 if 'nodelist' not in node_result:
694 feedback_fn(" - ERROR: node hasn't returned node connectivity data")
696 if node_result['nodelist']:
698 for node in node_result['nodelist']:
699 feedback_fn(" - ERROR: communication with node '%s': %s" %
700 (node, node_result['nodelist'][node]))
701 hyp_result = node_result.get('hypervisor', None)
702 if hyp_result is not None:
703 feedback_fn(" - ERROR: hypervisor verify failure: '%s'" % hyp_result)
706 def _VerifyInstance(self, instance, node_vol_is, node_instance, feedback_fn):
707 """Verify an instance.
709 This function checks to see if the required block devices are
710 available on the instance's node.
715 instancelist = self.cfg.GetInstanceList()
716 if not instance in instancelist:
717 feedback_fn(" - ERROR: instance %s not in instance list %s" %
718 (instance, instancelist))
721 instanceconfig = self.cfg.GetInstanceInfo(instance)
722 node_current = instanceconfig.primary_node
725 instanceconfig.MapLVsByNode(node_vol_should)
727 for node in node_vol_should:
728 for volume in node_vol_should[node]:
729 if node not in node_vol_is or volume not in node_vol_is[node]:
730 feedback_fn(" - ERROR: volume %s missing on node %s" %
734 if not instanceconfig.status == 'down':
735 if not instance in node_instance[node_current]:
736 feedback_fn(" - ERROR: instance %s not running on node %s" %
737 (instance, node_current))
740 for node in node_instance:
741 if (not node == node_current):
742 if instance in node_instance[node]:
743 feedback_fn(" - ERROR: instance %s should not run on node %s" %
749 def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
750 """Verify if there are any unknown volumes in the cluster.
752 The .os, .swap and backup volumes are ignored. All other volumes are
758 for node in node_vol_is:
759 for volume in node_vol_is[node]:
760 if node not in node_vol_should or volume not in node_vol_should[node]:
761 feedback_fn(" - ERROR: volume %s on node %s should not exist" %
766 def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
767 """Verify the list of running instances.
769 This checks what instances are running but unknown to the cluster.
773 for node in node_instance:
774 for runninginstance in node_instance[node]:
775 if runninginstance not in instancelist:
776 feedback_fn(" - ERROR: instance %s on node %s should not exist" %
777 (runninginstance, node))
781 def CheckPrereq(self):
782 """Check prerequisites.
784 This has no prerequisites.
789 def Exec(self, feedback_fn):
790 """Verify integrity of cluster, performing various test on nodes.
794 feedback_fn("* Verifying global settings")
795 self.cfg.VerifyConfig()
797 vg_name = self.cfg.GetVGName()
798 nodelist = utils.NiceSort(self.cfg.GetNodeList())
799 instancelist = utils.NiceSort(self.cfg.GetInstanceList())
803 # FIXME: verify OS list
805 file_names = list(self.sstore.GetFileList())
806 file_names.append(constants.SSL_CERT_FILE)
807 file_names.append(constants.CLUSTER_CONF_FILE)
808 local_checksums = utils.FingerprintFiles(file_names)
810 feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
811 all_volumeinfo = rpc.call_volume_list(nodelist, vg_name)
812 all_instanceinfo = rpc.call_instance_list(nodelist)
813 all_vglist = rpc.call_vg_list(nodelist)
814 node_verify_param = {
815 'filelist': file_names,
816 'nodelist': nodelist,
819 all_nvinfo = rpc.call_node_verify(nodelist, node_verify_param)
820 all_rversion = rpc.call_version(nodelist)
822 for node in nodelist:
823 feedback_fn("* Verifying node %s" % node)
824 result = self._VerifyNode(node, file_names, local_checksums,
825 all_vglist[node], all_nvinfo[node],
826 all_rversion[node], feedback_fn)
830 volumeinfo = all_volumeinfo[node]
832 if type(volumeinfo) != dict:
833 feedback_fn(" - ERROR: connection to %s failed" % (node,))
837 node_volume[node] = volumeinfo
840 nodeinstance = all_instanceinfo[node]
841 if type(nodeinstance) != list:
842 feedback_fn(" - ERROR: connection to %s failed" % (node,))
846 node_instance[node] = nodeinstance
850 for instance in instancelist:
851 feedback_fn("* Verifying instance %s" % instance)
852 result = self._VerifyInstance(instance, node_volume, node_instance,
856 inst_config = self.cfg.GetInstanceInfo(instance)
858 inst_config.MapLVsByNode(node_vol_should)
860 feedback_fn("* Verifying orphan volumes")
861 result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
865 feedback_fn("* Verifying remaining instances")
866 result = self._VerifyOrphanInstances(instancelist, node_instance,
873 class LURenameCluster(LogicalUnit):
874 """Rename the cluster.
877 HPATH = "cluster-rename"
878 HTYPE = constants.HTYPE_CLUSTER
881 def BuildHooksEnv(self):
886 "OP_TARGET": self.op.sstore.GetClusterName(),
887 "NEW_NAME": self.op.name,
889 mn = self.sstore.GetMasterNode()
890 return env, [mn], [mn]
892 def CheckPrereq(self):
893 """Verify that the passed name is a valid one.
896 hostname = utils.HostInfo(self.op.name)
898 new_name = hostname.name
899 self.ip = new_ip = hostname.ip
900 old_name = self.sstore.GetClusterName()
901 old_ip = self.sstore.GetMasterIP()
902 if new_name == old_name and new_ip == old_ip:
903 raise errors.OpPrereqError("Neither the name nor the IP address of the"
904 " cluster has changed")
906 result = utils.RunCmd(["fping", "-q", new_ip])
907 if not result.failed:
908 raise errors.OpPrereqError("The given cluster IP address (%s) is"
909 " reachable on the network. Aborting." %
912 self.op.name = new_name
914 def Exec(self, feedback_fn):
915 """Rename the cluster.
918 clustername = self.op.name
922 # shutdown the master IP
923 master = ss.GetMasterNode()
924 if not rpc.call_node_stop_master(master):
925 raise errors.OpExecError("Could not disable the master role")
929 ss.SetKey(ss.SS_MASTER_IP, ip)
930 ss.SetKey(ss.SS_CLUSTER_NAME, clustername)
932 # Distribute updated ss config to all nodes
933 myself = self.cfg.GetNodeInfo(master)
934 dist_nodes = self.cfg.GetNodeList()
935 if myself.name in dist_nodes:
936 dist_nodes.remove(myself.name)
938 logger.Debug("Copying updated ssconf data to all nodes")
939 for keyname in [ss.SS_CLUSTER_NAME, ss.SS_MASTER_IP]:
940 fname = ss.KeyToFilename(keyname)
941 result = rpc.call_upload_file(dist_nodes, fname)
942 for to_node in dist_nodes:
943 if not result[to_node]:
944 logger.Error("copy of file %s to node %s failed" %
947 if not rpc.call_node_start_master(master):
948 logger.Error("Could not re-enable the master role on the master,\n"
949 "please restart manually.")
952 def _WaitForSync(cfgw, instance, proc, oneshot=False, unlock=False):
953 """Sleep and poll for an instance's disk to sync.
956 if not instance.disks:
960 proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
962 node = instance.primary_node
964 for dev in instance.disks:
965 cfgw.SetDiskID(dev, node)
971 cumul_degraded = False
972 rstats = rpc.call_blockdev_getmirrorstatus(node, instance.disks)
974 proc.LogWarning("Can't get any data from node %s" % node)
977 raise errors.RemoteError("Can't contact node %s for mirror data,"
982 for i in range(len(rstats)):
985 proc.LogWarning("Can't compute data for node %s/%s" %
986 (node, instance.disks[i].iv_name))
988 # we ignore the ldisk parameter
989 perc_done, est_time, is_degraded, _ = mstat
990 cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
991 if perc_done is not None:
993 if est_time is not None:
994 rem_time = "%d estimated seconds remaining" % est_time
997 rem_time = "no time estimate"
998 proc.LogInfo("- device %s: %5.2f%% done, %s" %
999 (instance.disks[i].iv_name, perc_done, rem_time))
1006 time.sleep(min(60, max_time))
1012 proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1013 return not cumul_degraded
1016 def _CheckDiskConsistency(cfgw, dev, node, on_primary, ldisk=False):
1017 """Check that mirrors are not degraded.
1019 The ldisk parameter, if True, will change the test from the
1020 is_degraded attribute (which represents overall non-ok status for
1021 the device(s)) to the ldisk (representing the local storage status).
1024 cfgw.SetDiskID(dev, node)
1031 if on_primary or dev.AssembleOnSecondary():
1032 rstats = rpc.call_blockdev_find(node, dev)
1034 logger.ToStderr("Can't get any data from node %s" % node)
1037 result = result and (not rstats[idx])
1039 for child in dev.children:
1040 result = result and _CheckDiskConsistency(cfgw, child, node, on_primary)
1045 class LUDiagnoseOS(NoHooksLU):
1046 """Logical unit for OS diagnose/query.
1051 def CheckPrereq(self):
1052 """Check prerequisites.
1054 This always succeeds, since this is a pure query LU.
1059 def Exec(self, feedback_fn):
1060 """Compute the list of OSes.
1063 node_list = self.cfg.GetNodeList()
1064 node_data = rpc.call_os_diagnose(node_list)
1065 if node_data == False:
1066 raise errors.OpExecError("Can't gather the list of OSes")
1070 class LURemoveNode(LogicalUnit):
1071 """Logical unit for removing a node.
1074 HPATH = "node-remove"
1075 HTYPE = constants.HTYPE_NODE
1076 _OP_REQP = ["node_name"]
1078 def BuildHooksEnv(self):
1081 This doesn't run on the target node in the pre phase as a failed
1082 node would not allows itself to run.
1086 "OP_TARGET": self.op.node_name,
1087 "NODE_NAME": self.op.node_name,
1089 all_nodes = self.cfg.GetNodeList()
1090 all_nodes.remove(self.op.node_name)
1091 return env, all_nodes, all_nodes
1093 def CheckPrereq(self):
1094 """Check prerequisites.
1097 - the node exists in the configuration
1098 - it does not have primary or secondary instances
1099 - it's not the master
1101 Any errors are signalled by raising errors.OpPrereqError.
1104 node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1106 raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1108 instance_list = self.cfg.GetInstanceList()
1110 masternode = self.sstore.GetMasterNode()
1111 if node.name == masternode:
1112 raise errors.OpPrereqError("Node is the master node,"
1113 " you need to failover first.")
1115 for instance_name in instance_list:
1116 instance = self.cfg.GetInstanceInfo(instance_name)
1117 if node.name == instance.primary_node:
1118 raise errors.OpPrereqError("Instance %s still running on the node,"
1119 " please remove first." % instance_name)
1120 if node.name in instance.secondary_nodes:
1121 raise errors.OpPrereqError("Instance %s has node as a secondary,"
1122 " please remove first." % instance_name)
1123 self.op.node_name = node.name
1126 def Exec(self, feedback_fn):
1127 """Removes the node from the cluster.
1131 logger.Info("stopping the node daemon and removing configs from node %s" %
1134 rpc.call_node_leave_cluster(node.name)
1136 ssh.SSHCall(node.name, 'root', "%s stop" % constants.NODE_INITD_SCRIPT)
1138 logger.Info("Removing node %s from config" % node.name)
1140 self.cfg.RemoveNode(node.name)
1142 _RemoveHostFromEtcHosts(node.name)
1145 class LUQueryNodes(NoHooksLU):
1146 """Logical unit for querying nodes.
1149 _OP_REQP = ["output_fields", "names"]
1151 def CheckPrereq(self):
1152 """Check prerequisites.
1154 This checks that the fields required are valid output fields.
1157 self.dynamic_fields = frozenset(["dtotal", "dfree",
1158 "mtotal", "mnode", "mfree",
1161 _CheckOutputFields(static=["name", "pinst_cnt", "sinst_cnt",
1162 "pinst_list", "sinst_list",
1164 dynamic=self.dynamic_fields,
1165 selected=self.op.output_fields)
1167 self.wanted = _GetWantedNodes(self, self.op.names)
1169 def Exec(self, feedback_fn):
1170 """Computes the list of nodes and their attributes.
1173 nodenames = self.wanted
1174 nodelist = [self.cfg.GetNodeInfo(name) for name in nodenames]
1176 # begin data gathering
1178 if self.dynamic_fields.intersection(self.op.output_fields):
1180 node_data = rpc.call_node_info(nodenames, self.cfg.GetVGName())
1181 for name in nodenames:
1182 nodeinfo = node_data.get(name, None)
1185 "mtotal": utils.TryConvert(int, nodeinfo['memory_total']),
1186 "mnode": utils.TryConvert(int, nodeinfo['memory_dom0']),
1187 "mfree": utils.TryConvert(int, nodeinfo['memory_free']),
1188 "dtotal": utils.TryConvert(int, nodeinfo['vg_size']),
1189 "dfree": utils.TryConvert(int, nodeinfo['vg_free']),
1190 "bootid": nodeinfo['bootid'],
1193 live_data[name] = {}
1195 live_data = dict.fromkeys(nodenames, {})
1197 node_to_primary = dict([(name, set()) for name in nodenames])
1198 node_to_secondary = dict([(name, set()) for name in nodenames])
1200 inst_fields = frozenset(("pinst_cnt", "pinst_list",
1201 "sinst_cnt", "sinst_list"))
1202 if inst_fields & frozenset(self.op.output_fields):
1203 instancelist = self.cfg.GetInstanceList()
1205 for instance_name in instancelist:
1206 inst = self.cfg.GetInstanceInfo(instance_name)
1207 if inst.primary_node in node_to_primary:
1208 node_to_primary[inst.primary_node].add(inst.name)
1209 for secnode in inst.secondary_nodes:
1210 if secnode in node_to_secondary:
1211 node_to_secondary[secnode].add(inst.name)
1213 # end data gathering
1216 for node in nodelist:
1218 for field in self.op.output_fields:
1221 elif field == "pinst_list":
1222 val = list(node_to_primary[node.name])
1223 elif field == "sinst_list":
1224 val = list(node_to_secondary[node.name])
1225 elif field == "pinst_cnt":
1226 val = len(node_to_primary[node.name])
1227 elif field == "sinst_cnt":
1228 val = len(node_to_secondary[node.name])
1229 elif field == "pip":
1230 val = node.primary_ip
1231 elif field == "sip":
1232 val = node.secondary_ip
1233 elif field in self.dynamic_fields:
1234 val = live_data[node.name].get(field, None)
1236 raise errors.ParameterError(field)
1237 node_output.append(val)
1238 output.append(node_output)
1243 class LUQueryNodeVolumes(NoHooksLU):
1244 """Logical unit for getting volumes on node(s).
1247 _OP_REQP = ["nodes", "output_fields"]
1249 def CheckPrereq(self):
1250 """Check prerequisites.
1252 This checks that the fields required are valid output fields.
1255 self.nodes = _GetWantedNodes(self, self.op.nodes)
1257 _CheckOutputFields(static=["node"],
1258 dynamic=["phys", "vg", "name", "size", "instance"],
1259 selected=self.op.output_fields)
1262 def Exec(self, feedback_fn):
1263 """Computes the list of nodes and their attributes.
1266 nodenames = self.nodes
1267 volumes = rpc.call_node_volumes(nodenames)
1269 ilist = [self.cfg.GetInstanceInfo(iname) for iname
1270 in self.cfg.GetInstanceList()]
1272 lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1275 for node in nodenames:
1276 if node not in volumes or not volumes[node]:
1279 node_vols = volumes[node][:]
1280 node_vols.sort(key=lambda vol: vol['dev'])
1282 for vol in node_vols:
1284 for field in self.op.output_fields:
1287 elif field == "phys":
1291 elif field == "name":
1293 elif field == "size":
1294 val = int(float(vol['size']))
1295 elif field == "instance":
1297 if node not in lv_by_node[inst]:
1299 if vol['name'] in lv_by_node[inst][node]:
1305 raise errors.ParameterError(field)
1306 node_output.append(str(val))
1308 output.append(node_output)
1313 class LUAddNode(LogicalUnit):
1314 """Logical unit for adding node to the cluster.
1318 HTYPE = constants.HTYPE_NODE
1319 _OP_REQP = ["node_name"]
1321 def BuildHooksEnv(self):
1324 This will run on all nodes before, and on all nodes + the new node after.
1328 "OP_TARGET": self.op.node_name,
1329 "NODE_NAME": self.op.node_name,
1330 "NODE_PIP": self.op.primary_ip,
1331 "NODE_SIP": self.op.secondary_ip,
1333 nodes_0 = self.cfg.GetNodeList()
1334 nodes_1 = nodes_0 + [self.op.node_name, ]
1335 return env, nodes_0, nodes_1
1337 def CheckPrereq(self):
1338 """Check prerequisites.
1341 - the new node is not already in the config
1343 - its parameters (single/dual homed) matches the cluster
1345 Any errors are signalled by raising errors.OpPrereqError.
1348 node_name = self.op.node_name
1351 dns_data = utils.HostInfo(node_name)
1353 node = dns_data.name
1354 primary_ip = self.op.primary_ip = dns_data.ip
1355 secondary_ip = getattr(self.op, "secondary_ip", None)
1356 if secondary_ip is None:
1357 secondary_ip = primary_ip
1358 if not utils.IsValidIP(secondary_ip):
1359 raise errors.OpPrereqError("Invalid secondary IP given")
1360 self.op.secondary_ip = secondary_ip
1361 node_list = cfg.GetNodeList()
1362 if node in node_list:
1363 raise errors.OpPrereqError("Node %s is already in the configuration"
1366 for existing_node_name in node_list:
1367 existing_node = cfg.GetNodeInfo(existing_node_name)
1368 if (existing_node.primary_ip == primary_ip or
1369 existing_node.secondary_ip == primary_ip or
1370 existing_node.primary_ip == secondary_ip or
1371 existing_node.secondary_ip == secondary_ip):
1372 raise errors.OpPrereqError("New node ip address(es) conflict with"
1373 " existing node %s" % existing_node.name)
1375 # check that the type of the node (single versus dual homed) is the
1376 # same as for the master
1377 myself = cfg.GetNodeInfo(self.sstore.GetMasterNode())
1378 master_singlehomed = myself.secondary_ip == myself.primary_ip
1379 newbie_singlehomed = secondary_ip == primary_ip
1380 if master_singlehomed != newbie_singlehomed:
1381 if master_singlehomed:
1382 raise errors.OpPrereqError("The master has no private ip but the"
1383 " new node has one")
1385 raise errors.OpPrereqError("The master has a private ip but the"
1386 " new node doesn't have one")
1388 # checks reachablity
1389 if not utils.TcpPing(utils.HostInfo().name,
1391 constants.DEFAULT_NODED_PORT):
1392 raise errors.OpPrereqError("Node not reachable by ping")
1394 if not newbie_singlehomed:
1395 # check reachability from my secondary ip to newbie's secondary ip
1396 if not utils.TcpPing(myself.secondary_ip,
1398 constants.DEFAULT_NODED_PORT):
1399 raise errors.OpPrereqError(
1400 "Node secondary ip not reachable by TCP based ping to noded port")
1402 self.new_node = objects.Node(name=node,
1403 primary_ip=primary_ip,
1404 secondary_ip=secondary_ip)
1406 def Exec(self, feedback_fn):
1407 """Adds the new node to the cluster.
1410 new_node = self.new_node
1411 node = new_node.name
1413 # set up inter-node password and certificate and restarts the node daemon
1414 gntpass = self.sstore.GetNodeDaemonPassword()
1415 if not re.match('^[a-zA-Z0-9.]{1,64}$', gntpass):
1416 raise errors.OpExecError("ganeti password corruption detected")
1417 f = open(constants.SSL_CERT_FILE)
1419 gntpem = f.read(8192)
1422 # in the base64 pem encoding, neither '!' nor '.' are valid chars,
1423 # so we use this to detect an invalid certificate; as long as the
1424 # cert doesn't contain this, the here-document will be correctly
1425 # parsed by the shell sequence below
1426 if re.search('^!EOF\.', gntpem, re.MULTILINE):
1427 raise errors.OpExecError("invalid PEM encoding in the SSL certificate")
1428 if not gntpem.endswith("\n"):
1429 raise errors.OpExecError("PEM must end with newline")
1430 logger.Info("copy cluster pass to %s and starting the node daemon" % node)
1432 # and then connect with ssh to set password and start ganeti-noded
1433 # note that all the below variables are sanitized at this point,
1434 # either by being constants or by the checks above
1436 mycommand = ("umask 077 && "
1437 "echo '%s' > '%s' && "
1438 "cat > '%s' << '!EOF.' && \n"
1439 "%s!EOF.\n%s restart" %
1440 (gntpass, ss.KeyToFilename(ss.SS_NODED_PASS),
1441 constants.SSL_CERT_FILE, gntpem,
1442 constants.NODE_INITD_SCRIPT))
1444 result = ssh.SSHCall(node, 'root', mycommand, batch=False, ask_key=True)
1446 raise errors.OpExecError("Remote command on node %s, error: %s,"
1448 (node, result.fail_reason, result.output))
1450 # check connectivity
1453 result = rpc.call_version([node])[node]
1455 if constants.PROTOCOL_VERSION == result:
1456 logger.Info("communication to node %s fine, sw version %s match" %
1459 raise errors.OpExecError("Version mismatch master version %s,"
1460 " node version %s" %
1461 (constants.PROTOCOL_VERSION, result))
1463 raise errors.OpExecError("Cannot get version from the new node")
1466 logger.Info("copy ssh key to node %s" % node)
1467 priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
1469 keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
1470 constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
1476 keyarray.append(f.read())
1480 result = rpc.call_node_add(node, keyarray[0], keyarray[1], keyarray[2],
1481 keyarray[3], keyarray[4], keyarray[5])
1484 raise errors.OpExecError("Cannot transfer ssh keys to the new node")
1486 # Add node to our /etc/hosts, and add key to known_hosts
1487 hi = utils.HostInfo(name=new_node.name)
1488 utils.AddEtcHostsEntry(constants.ETC_HOSTS, new_node.name, hi.ip)
1489 utils.AddEtcHostsEntry(constants.ETC_HOSTS, hi.ShortName(), hi.ip)
1492 _UpdateKnownHosts(new_node.name, new_node.primary_ip,
1493 self.cfg.GetHostKey())
1495 if new_node.secondary_ip != new_node.primary_ip:
1496 if not rpc.call_node_tcp_ping(new_node.name,
1497 constants.LOCALHOST_IP_ADDRESS,
1498 new_node.secondary_ip,
1499 constants.DEFAULT_NODED_PORT,
1501 raise errors.OpExecError("Node claims it doesn't have the"
1502 " secondary ip you gave (%s).\n"
1503 "Please fix and re-run this command." %
1504 new_node.secondary_ip)
1506 success, msg = ssh.VerifyNodeHostname(node)
1508 raise errors.OpExecError("Node '%s' claims it has a different hostname"
1509 " than the one the resolver gives: %s.\n"
1510 "Please fix and re-run this command." %
1513 # Distribute updated /etc/hosts and known_hosts to all nodes,
1514 # including the node just added
1515 myself = self.cfg.GetNodeInfo(self.sstore.GetMasterNode())
1516 dist_nodes = self.cfg.GetNodeList() + [node]
1517 if myself.name in dist_nodes:
1518 dist_nodes.remove(myself.name)
1520 logger.Debug("Copying hosts and known_hosts to all nodes")
1521 for fname in ("/etc/hosts", constants.SSH_KNOWN_HOSTS_FILE):
1522 result = rpc.call_upload_file(dist_nodes, fname)
1523 for to_node in dist_nodes:
1524 if not result[to_node]:
1525 logger.Error("copy of file %s to node %s failed" %
1528 to_copy = ss.GetFileList()
1529 for fname in to_copy:
1530 if not ssh.CopyFileToNode(node, fname):
1531 logger.Error("could not copy file %s to node %s" % (fname, node))
1533 logger.Info("adding node %s to cluster.conf" % node)
1534 self.cfg.AddNode(new_node)
1537 class LUMasterFailover(LogicalUnit):
1538 """Failover the master node to the current node.
1540 This is a special LU in that it must run on a non-master node.
1543 HPATH = "master-failover"
1544 HTYPE = constants.HTYPE_CLUSTER
1548 def BuildHooksEnv(self):
1551 This will run on the new master only in the pre phase, and on all
1552 the nodes in the post phase.
1556 "OP_TARGET": self.new_master,
1557 "NEW_MASTER": self.new_master,
1558 "OLD_MASTER": self.old_master,
1560 return env, [self.new_master], self.cfg.GetNodeList()
1562 def CheckPrereq(self):
1563 """Check prerequisites.
1565 This checks that we are not already the master.
1568 self.new_master = utils.HostInfo().name
1569 self.old_master = self.sstore.GetMasterNode()
1571 if self.old_master == self.new_master:
1572 raise errors.OpPrereqError("This commands must be run on the node"
1573 " where you want the new master to be.\n"
1574 "%s is already the master" %
1577 def Exec(self, feedback_fn):
1578 """Failover the master node.
1580 This command, when run on a non-master node, will cause the current
1581 master to cease being master, and the non-master to become new
1585 #TODO: do not rely on gethostname returning the FQDN
1586 logger.Info("setting master to %s, old master: %s" %
1587 (self.new_master, self.old_master))
1589 if not rpc.call_node_stop_master(self.old_master):
1590 logger.Error("could disable the master role on the old master"
1591 " %s, please disable manually" % self.old_master)
1594 ss.SetKey(ss.SS_MASTER_NODE, self.new_master)
1595 if not rpc.call_upload_file(self.cfg.GetNodeList(),
1596 ss.KeyToFilename(ss.SS_MASTER_NODE)):
1597 logger.Error("could not distribute the new simple store master file"
1598 " to the other nodes, please check.")
1600 if not rpc.call_node_start_master(self.new_master):
1601 logger.Error("could not start the master role on the new master"
1602 " %s, please check" % self.new_master)
1603 feedback_fn("Error in activating the master IP on the new master,\n"
1604 "please fix manually.")
1608 class LUQueryClusterInfo(NoHooksLU):
1609 """Query cluster configuration.
1615 def CheckPrereq(self):
1616 """No prerequsites needed for this LU.
1621 def Exec(self, feedback_fn):
1622 """Return cluster config.
1626 "name": self.sstore.GetClusterName(),
1627 "software_version": constants.RELEASE_VERSION,
1628 "protocol_version": constants.PROTOCOL_VERSION,
1629 "config_version": constants.CONFIG_VERSION,
1630 "os_api_version": constants.OS_API_VERSION,
1631 "export_version": constants.EXPORT_VERSION,
1632 "master": self.sstore.GetMasterNode(),
1633 "architecture": (platform.architecture()[0], platform.machine()),
1639 class LUClusterCopyFile(NoHooksLU):
1640 """Copy file to cluster.
1643 _OP_REQP = ["nodes", "filename"]
1645 def CheckPrereq(self):
1646 """Check prerequisites.
1648 It should check that the named file exists and that the given list
1652 if not os.path.exists(self.op.filename):
1653 raise errors.OpPrereqError("No such filename '%s'" % self.op.filename)
1655 self.nodes = _GetWantedNodes(self, self.op.nodes)
1657 def Exec(self, feedback_fn):
1658 """Copy a file from master to some nodes.
1661 opts - class with options as members
1662 args - list containing a single element, the file name
1664 nodes - list containing the name of target nodes; if empty, all nodes
1667 filename = self.op.filename
1669 myname = utils.HostInfo().name
1671 for node in self.nodes:
1674 if not ssh.CopyFileToNode(node, filename):
1675 logger.Error("Copy of file %s to node %s failed" % (filename, node))
1678 class LUDumpClusterConfig(NoHooksLU):
1679 """Return a text-representation of the cluster-config.
1684 def CheckPrereq(self):
1685 """No prerequisites.
1690 def Exec(self, feedback_fn):
1691 """Dump a representation of the cluster config to the standard output.
1694 return self.cfg.DumpConfig()
1697 class LURunClusterCommand(NoHooksLU):
1698 """Run a command on some nodes.
1701 _OP_REQP = ["command", "nodes"]
1703 def CheckPrereq(self):
1704 """Check prerequisites.
1706 It checks that the given list of nodes is valid.
1709 self.nodes = _GetWantedNodes(self, self.op.nodes)
1711 def Exec(self, feedback_fn):
1712 """Run a command on some nodes.
1716 for node in self.nodes:
1717 result = ssh.SSHCall(node, "root", self.op.command)
1718 data.append((node, result.output, result.exit_code))
1723 class LUActivateInstanceDisks(NoHooksLU):
1724 """Bring up an instance's disks.
1727 _OP_REQP = ["instance_name"]
1729 def CheckPrereq(self):
1730 """Check prerequisites.
1732 This checks that the instance is in the cluster.
1735 instance = self.cfg.GetInstanceInfo(
1736 self.cfg.ExpandInstanceName(self.op.instance_name))
1737 if instance is None:
1738 raise errors.OpPrereqError("Instance '%s' not known" %
1739 self.op.instance_name)
1740 self.instance = instance
1743 def Exec(self, feedback_fn):
1744 """Activate the disks.
1747 disks_ok, disks_info = _AssembleInstanceDisks(self.instance, self.cfg)
1749 raise errors.OpExecError("Cannot activate block devices")
1754 def _AssembleInstanceDisks(instance, cfg, ignore_secondaries=False):
1755 """Prepare the block devices for an instance.
1757 This sets up the block devices on all nodes.
1760 instance: a ganeti.objects.Instance object
1761 ignore_secondaries: if true, errors on secondary nodes won't result
1762 in an error return from the function
1765 false if the operation failed
1766 list of (host, instance_visible_name, node_visible_name) if the operation
1767 suceeded with the mapping from node devices to instance devices
1771 for inst_disk in instance.disks:
1772 master_result = None
1773 for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
1774 cfg.SetDiskID(node_disk, node)
1775 is_primary = node == instance.primary_node
1776 result = rpc.call_blockdev_assemble(node, node_disk,
1777 instance.name, is_primary)
1779 logger.Error("could not prepare block device %s on node %s (is_pri"
1780 "mary=%s)" % (inst_disk.iv_name, node, is_primary))
1781 if is_primary or not ignore_secondaries:
1784 master_result = result
1785 device_info.append((instance.primary_node, inst_disk.iv_name,
1788 # leave the disks configured for the primary node
1789 # this is a workaround that would be fixed better by
1790 # improving the logical/physical id handling
1791 for disk in instance.disks:
1792 cfg.SetDiskID(disk, instance.primary_node)
1794 return disks_ok, device_info
1797 def _StartInstanceDisks(cfg, instance, force):
1798 """Start the disks of an instance.
1801 disks_ok, dummy = _AssembleInstanceDisks(instance, cfg,
1802 ignore_secondaries=force)
1804 _ShutdownInstanceDisks(instance, cfg)
1805 if force is not None and not force:
1806 logger.Error("If the message above refers to a secondary node,"
1807 " you can retry the operation using '--force'.")
1808 raise errors.OpExecError("Disk consistency error")
1811 class LUDeactivateInstanceDisks(NoHooksLU):
1812 """Shutdown an instance's disks.
1815 _OP_REQP = ["instance_name"]
1817 def CheckPrereq(self):
1818 """Check prerequisites.
1820 This checks that the instance is in the cluster.
1823 instance = self.cfg.GetInstanceInfo(
1824 self.cfg.ExpandInstanceName(self.op.instance_name))
1825 if instance is None:
1826 raise errors.OpPrereqError("Instance '%s' not known" %
1827 self.op.instance_name)
1828 self.instance = instance
1830 def Exec(self, feedback_fn):
1831 """Deactivate the disks
1834 instance = self.instance
1835 ins_l = rpc.call_instance_list([instance.primary_node])
1836 ins_l = ins_l[instance.primary_node]
1837 if not type(ins_l) is list:
1838 raise errors.OpExecError("Can't contact node '%s'" %
1839 instance.primary_node)
1841 if self.instance.name in ins_l:
1842 raise errors.OpExecError("Instance is running, can't shutdown"
1845 _ShutdownInstanceDisks(instance, self.cfg)
1848 def _ShutdownInstanceDisks(instance, cfg, ignore_primary=False):
1849 """Shutdown block devices of an instance.
1851 This does the shutdown on all nodes of the instance.
1853 If the ignore_primary is false, errors on the primary node are
1858 for disk in instance.disks:
1859 for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
1860 cfg.SetDiskID(top_disk, node)
1861 if not rpc.call_blockdev_shutdown(node, top_disk):
1862 logger.Error("could not shutdown block device %s on node %s" %
1863 (disk.iv_name, node))
1864 if not ignore_primary or node != instance.primary_node:
1869 class LUStartupInstance(LogicalUnit):
1870 """Starts an instance.
1873 HPATH = "instance-start"
1874 HTYPE = constants.HTYPE_INSTANCE
1875 _OP_REQP = ["instance_name", "force"]
1877 def BuildHooksEnv(self):
1880 This runs on master, primary and secondary nodes of the instance.
1884 "FORCE": self.op.force,
1886 env.update(_BuildInstanceHookEnvByObject(self.instance))
1887 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
1888 list(self.instance.secondary_nodes))
1891 def CheckPrereq(self):
1892 """Check prerequisites.
1894 This checks that the instance is in the cluster.
1897 instance = self.cfg.GetInstanceInfo(
1898 self.cfg.ExpandInstanceName(self.op.instance_name))
1899 if instance is None:
1900 raise errors.OpPrereqError("Instance '%s' not known" %
1901 self.op.instance_name)
1903 # check bridges existance
1904 _CheckInstanceBridgesExist(instance)
1906 self.instance = instance
1907 self.op.instance_name = instance.name
1909 def Exec(self, feedback_fn):
1910 """Start the instance.
1913 instance = self.instance
1914 force = self.op.force
1915 extra_args = getattr(self.op, "extra_args", "")
1917 node_current = instance.primary_node
1919 nodeinfo = rpc.call_node_info([node_current], self.cfg.GetVGName())
1921 raise errors.OpExecError("Could not contact node %s for infos" %
1924 freememory = nodeinfo[node_current]['memory_free']
1925 memory = instance.memory
1926 if memory > freememory:
1927 raise errors.OpExecError("Not enough memory to start instance"
1929 " needed %s MiB, available %s MiB" %
1930 (instance.name, node_current, memory,
1933 _StartInstanceDisks(self.cfg, instance, force)
1935 if not rpc.call_instance_start(node_current, instance, extra_args):
1936 _ShutdownInstanceDisks(instance, self.cfg)
1937 raise errors.OpExecError("Could not start instance")
1939 self.cfg.MarkInstanceUp(instance.name)
1942 class LURebootInstance(LogicalUnit):
1943 """Reboot an instance.
1946 HPATH = "instance-reboot"
1947 HTYPE = constants.HTYPE_INSTANCE
1948 _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
1950 def BuildHooksEnv(self):
1953 This runs on master, primary and secondary nodes of the instance.
1957 "IGNORE_SECONDARIES": self.op.ignore_secondaries,
1959 env.update(_BuildInstanceHookEnvByObject(self.instance))
1960 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
1961 list(self.instance.secondary_nodes))
1964 def CheckPrereq(self):
1965 """Check prerequisites.
1967 This checks that the instance is in the cluster.
1970 instance = self.cfg.GetInstanceInfo(
1971 self.cfg.ExpandInstanceName(self.op.instance_name))
1972 if instance is None:
1973 raise errors.OpPrereqError("Instance '%s' not known" %
1974 self.op.instance_name)
1976 # check bridges existance
1977 _CheckInstanceBridgesExist(instance)
1979 self.instance = instance
1980 self.op.instance_name = instance.name
1982 def Exec(self, feedback_fn):
1983 """Reboot the instance.
1986 instance = self.instance
1987 ignore_secondaries = self.op.ignore_secondaries
1988 reboot_type = self.op.reboot_type
1989 extra_args = getattr(self.op, "extra_args", "")
1991 node_current = instance.primary_node
1993 if reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
1994 constants.INSTANCE_REBOOT_HARD,
1995 constants.INSTANCE_REBOOT_FULL]:
1996 raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
1997 (constants.INSTANCE_REBOOT_SOFT,
1998 constants.INSTANCE_REBOOT_HARD,
1999 constants.INSTANCE_REBOOT_FULL))
2001 if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2002 constants.INSTANCE_REBOOT_HARD]:
2003 if not rpc.call_instance_reboot(node_current, instance,
2004 reboot_type, extra_args):
2005 raise errors.OpExecError("Could not reboot instance")
2007 if not rpc.call_instance_shutdown(node_current, instance):
2008 raise errors.OpExecError("could not shutdown instance for full reboot")
2009 _ShutdownInstanceDisks(instance, self.cfg)
2010 _StartInstanceDisks(self.cfg, instance, ignore_secondaries)
2011 if not rpc.call_instance_start(node_current, instance, extra_args):
2012 _ShutdownInstanceDisks(instance, self.cfg)
2013 raise errors.OpExecError("Could not start instance for full reboot")
2015 self.cfg.MarkInstanceUp(instance.name)
2018 class LUShutdownInstance(LogicalUnit):
2019 """Shutdown an instance.
2022 HPATH = "instance-stop"
2023 HTYPE = constants.HTYPE_INSTANCE
2024 _OP_REQP = ["instance_name"]
2026 def BuildHooksEnv(self):
2029 This runs on master, primary and secondary nodes of the instance.
2032 env = _BuildInstanceHookEnvByObject(self.instance)
2033 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2034 list(self.instance.secondary_nodes))
2037 def CheckPrereq(self):
2038 """Check prerequisites.
2040 This checks that the instance is in the cluster.
2043 instance = self.cfg.GetInstanceInfo(
2044 self.cfg.ExpandInstanceName(self.op.instance_name))
2045 if instance is None:
2046 raise errors.OpPrereqError("Instance '%s' not known" %
2047 self.op.instance_name)
2048 self.instance = instance
2050 def Exec(self, feedback_fn):
2051 """Shutdown the instance.
2054 instance = self.instance
2055 node_current = instance.primary_node
2056 if not rpc.call_instance_shutdown(node_current, instance):
2057 logger.Error("could not shutdown instance")
2059 self.cfg.MarkInstanceDown(instance.name)
2060 _ShutdownInstanceDisks(instance, self.cfg)
2063 class LUReinstallInstance(LogicalUnit):
2064 """Reinstall an instance.
2067 HPATH = "instance-reinstall"
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 and is not running.
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 if instance.disk_template == constants.DT_DISKLESS:
2094 raise errors.OpPrereqError("Instance '%s' has no disks" %
2095 self.op.instance_name)
2096 if instance.status != "down":
2097 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2098 self.op.instance_name)
2099 remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2101 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2102 (self.op.instance_name,
2103 instance.primary_node))
2105 self.op.os_type = getattr(self.op, "os_type", None)
2106 if self.op.os_type is not None:
2108 pnode = self.cfg.GetNodeInfo(
2109 self.cfg.ExpandNodeName(instance.primary_node))
2111 raise errors.OpPrereqError("Primary node '%s' is unknown" %
2113 os_obj = rpc.call_os_get(pnode.name, self.op.os_type)
2115 raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2116 " primary node" % self.op.os_type)
2118 self.instance = instance
2120 def Exec(self, feedback_fn):
2121 """Reinstall the instance.
2124 inst = self.instance
2126 if self.op.os_type is not None:
2127 feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2128 inst.os = self.op.os_type
2129 self.cfg.AddInstance(inst)
2131 _StartInstanceDisks(self.cfg, inst, None)
2133 feedback_fn("Running the instance OS create scripts...")
2134 if not rpc.call_instance_os_add(inst.primary_node, inst, "sda", "sdb"):
2135 raise errors.OpExecError("Could not install OS for instance %s "
2137 (inst.name, inst.primary_node))
2139 _ShutdownInstanceDisks(inst, self.cfg)
2142 class LURenameInstance(LogicalUnit):
2143 """Rename an instance.
2146 HPATH = "instance-rename"
2147 HTYPE = constants.HTYPE_INSTANCE
2148 _OP_REQP = ["instance_name", "new_name"]
2150 def BuildHooksEnv(self):
2153 This runs on master, primary and secondary nodes of the instance.
2156 env = _BuildInstanceHookEnvByObject(self.instance)
2157 env["INSTANCE_NEW_NAME"] = self.op.new_name
2158 nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2159 list(self.instance.secondary_nodes))
2162 def CheckPrereq(self):
2163 """Check prerequisites.
2165 This checks that the instance is in the cluster and is not running.
2168 instance = self.cfg.GetInstanceInfo(
2169 self.cfg.ExpandInstanceName(self.op.instance_name))
2170 if instance is None:
2171 raise errors.OpPrereqError("Instance '%s' not known" %
2172 self.op.instance_name)
2173 if instance.status != "down":
2174 raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2175 self.op.instance_name)
2176 remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2178 raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2179 (self.op.instance_name,
2180 instance.primary_node))
2181 self.instance = instance
2183 # new name verification
2184 name_info = utils.HostInfo(self.op.new_name)
2186 self.op.new_name = new_name = name_info.name
2187 if not getattr(self.op, "ignore_ip", False):
2188 command = ["fping", "-q", name_info.ip]
2189 result = utils.RunCmd(command)
2190 if not result.failed:
2191 raise errors.OpPrereqError("IP %s of instance %s already in use" %
2192 (name_info.ip, new_name))
2195 def Exec(self, feedback_fn):
2196 """Reinstall the instance.
2199 inst = self.instance
2200 old_name = inst.name
2202 self.cfg.RenameInstance(inst.name, self.op.new_name)
2204 # re-read the instance from the configuration after rename
2205 inst = self.cfg.GetInstanceInfo(self.op.new_name)
2207 _StartInstanceDisks(self.cfg, inst, None)
2209 if not rpc.call_instance_run_rename(inst.primary_node, inst, old_name,
2211 msg = ("Could run OS rename script for instance %s\n"
2213 "(but the instance has been renamed in Ganeti)" %
2214 (inst.name, inst.primary_node))
2217 _ShutdownInstanceDisks(inst, self.cfg)
2220 class LURemoveInstance(LogicalUnit):
2221 """Remove an instance.
2224 HPATH = "instance-remove"
2225 HTYPE = constants.HTYPE_INSTANCE
2226 _OP_REQP = ["instance_name"]
2228 def BuildHooksEnv(self):
2231 This runs on master, primary and secondary nodes of the instance.
2234 env = _BuildInstanceHookEnvByObject(self.instance)
2235 nl = [self.sstore.GetMasterNode()]
2238 def CheckPrereq(self):
2239 """Check prerequisites.
2241 This checks that the instance is in the cluster.
2244 instance = self.cfg.GetInstanceInfo(
2245 self.cfg.ExpandInstanceName(self.op.instance_name))
2246 if instance is None:
2247 raise errors.OpPrereqError("Instance '%s' not known" %
2248 self.op.instance_name)
2249 self.instance = instance
2251 def Exec(self, feedback_fn):
2252 """Remove the instance.
2255 instance = self.instance
2256 logger.Info("shutting down instance %s on node %s" %
2257 (instance.name, instance.primary_node))
2259 if not rpc.call_instance_shutdown(instance.primary_node, instance):
2260 if self.op.ignore_failures:
2261 feedback_fn("Warning: can't shutdown instance")
2263 raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2264 (instance.name, instance.primary_node))
2266 logger.Info("removing block devices for instance %s" % instance.name)
2268 if not _RemoveDisks(instance, self.cfg):
2269 if self.op.ignore_failures:
2270 feedback_fn("Warning: can't remove instance's disks")
2272 raise errors.OpExecError("Can't remove instance's disks")
2274 logger.Info("removing instance %s out of cluster config" % instance.name)
2276 self.cfg.RemoveInstance(instance.name)
2279 class LUQueryInstances(NoHooksLU):
2280 """Logical unit for querying instances.
2283 _OP_REQP = ["output_fields", "names"]
2285 def CheckPrereq(self):
2286 """Check prerequisites.
2288 This checks that the fields required are valid output fields.
2291 self.dynamic_fields = frozenset(["oper_state", "oper_ram"])
2292 _CheckOutputFields(static=["name", "os", "pnode", "snodes",
2293 "admin_state", "admin_ram",
2294 "disk_template", "ip", "mac", "bridge",
2295 "sda_size", "sdb_size"],
2296 dynamic=self.dynamic_fields,
2297 selected=self.op.output_fields)
2299 self.wanted = _GetWantedInstances(self, self.op.names)
2301 def Exec(self, feedback_fn):
2302 """Computes the list of nodes and their attributes.
2305 instance_names = self.wanted
2306 instance_list = [self.cfg.GetInstanceInfo(iname) for iname
2309 # begin data gathering
2311 nodes = frozenset([inst.primary_node for inst in instance_list])
2314 if self.dynamic_fields.intersection(self.op.output_fields):
2316 node_data = rpc.call_all_instances_info(nodes)
2318 result = node_data[name]
2320 live_data.update(result)
2321 elif result == False:
2322 bad_nodes.append(name)
2323 # else no instance is alive
2325 live_data = dict([(name, {}) for name in instance_names])
2327 # end data gathering
2330 for instance in instance_list:
2332 for field in self.op.output_fields:
2337 elif field == "pnode":
2338 val = instance.primary_node
2339 elif field == "snodes":
2340 val = list(instance.secondary_nodes)
2341 elif field == "admin_state":
2342 val = (instance.status != "down")
2343 elif field == "oper_state":
2344 if instance.primary_node in bad_nodes:
2347 val = bool(live_data.get(instance.name))
2348 elif field == "admin_ram":
2349 val = instance.memory
2350 elif field == "oper_ram":
2351 if instance.primary_node in bad_nodes:
2353 elif instance.name in live_data:
2354 val = live_data[instance.name].get("memory", "?")
2357 elif field == "disk_template":
2358 val = instance.disk_template
2360 val = instance.nics[0].ip
2361 elif field == "bridge":
2362 val = instance.nics[0].bridge
2363 elif field == "mac":
2364 val = instance.nics[0].mac
2365 elif field == "sda_size" or field == "sdb_size":
2366 disk = instance.FindDisk(field[:3])
2372 raise errors.ParameterError(field)
2379 class LUFailoverInstance(LogicalUnit):
2380 """Failover an instance.
2383 HPATH = "instance-failover"
2384 HTYPE = constants.HTYPE_INSTANCE
2385 _OP_REQP = ["instance_name", "ignore_consistency"]
2387 def BuildHooksEnv(self):
2390 This runs on master, primary and secondary nodes of the instance.
2394 "IGNORE_CONSISTENCY": self.op.ignore_consistency,
2396 env.update(_BuildInstanceHookEnvByObject(self.instance))
2397 nl = [self.sstore.GetMasterNode()] + list(self.instance.secondary_nodes)
2400 def CheckPrereq(self):
2401 """Check prerequisites.
2403 This checks that the instance is in the cluster.
2406 instance = self.cfg.GetInstanceInfo(
2407 self.cfg.ExpandInstanceName(self.op.instance_name))
2408 if instance is None:
2409 raise errors.OpPrereqError("Instance '%s' not known" %
2410 self.op.instance_name)
2412 if instance.disk_template not in constants.DTS_NET_MIRROR:
2413 raise errors.OpPrereqError("Instance's disk layout is not"
2414 " network mirrored, cannot failover.")
2416 secondary_nodes = instance.secondary_nodes
2417 if not secondary_nodes:
2418 raise errors.ProgrammerError("no secondary node but using "
2419 "DT_REMOTE_RAID1 template")
2421 # check memory requirements on the secondary node
2422 target_node = secondary_nodes[0]
2423 nodeinfo = rpc.call_node_info([target_node], self.cfg.GetVGName())
2424 info = nodeinfo.get(target_node, None)
2426 raise errors.OpPrereqError("Cannot get current information"
2427 " from node '%s'" % nodeinfo)
2428 if instance.memory > info['memory_free']:
2429 raise errors.OpPrereqError("Not enough memory on target node %s."
2430 " %d MB available, %d MB required" %
2431 (target_node, info['memory_free'],
2434 # check bridge existance
2435 brlist = [nic.bridge for nic in instance.nics]
2436 if not rpc.call_bridges_exist(target_node, brlist):
2437 raise errors.OpPrereqError("One or more target bridges %s does not"
2438 " exist on destination node '%s'" %
2439 (brlist, target_node))
2441 self.instance = instance
2443 def Exec(self, feedback_fn):
2444 """Failover an instance.
2446 The failover is done by shutting it down on its present node and
2447 starting it on the secondary.
2450 instance = self.instance
2452 source_node = instance.primary_node
2453 target_node = instance.secondary_nodes[0]
2455 feedback_fn("* checking disk consistency between source and target")
2456 for dev in instance.disks:
2457 # for remote_raid1, these are md over drbd
2458 if not _CheckDiskConsistency(self.cfg, dev, target_node, False):
2459 if not self.op.ignore_consistency:
2460 raise errors.OpExecError("Disk %s is degraded on target node,"
2461 " aborting failover." % dev.iv_name)
2463 feedback_fn("* checking target node resource availability")
2464 nodeinfo = rpc.call_node_info([target_node], self.cfg.GetVGName())
2467 raise errors.OpExecError("Could not contact target node %s." %
2470 free_memory = int(nodeinfo[target_node]['memory_free'])
2471 memory = instance.memory
2472 if memory > free_memory:
2473 raise errors.OpExecError("Not enough memory to create instance %s on"
2474 " node %s. needed %s MiB, available %s MiB" %
2475 (instance.name, target_node, memory,
2478 feedback_fn("* shutting down instance on source node")
2479 logger.Info("Shutting down instance %s on node %s" %
2480 (instance.name, source_node))
2482 if not rpc.call_instance_shutdown(source_node, instance):
2483 if self.op.ignore_consistency:
2484 logger.Error("Could not shutdown instance %s on node %s. Proceeding"
2485 " anyway. Please make sure node %s is down" %
2486 (instance.name, source_node, source_node))
2488 raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2489 (instance.name, source_node))
2491 feedback_fn("* deactivating the instance's disks on source node")
2492 if not _ShutdownInstanceDisks(instance, self.cfg, ignore_primary=True):
2493 raise errors.OpExecError("Can't shut down the instance's disks.")
2495 instance.primary_node = target_node
2496 # distribute new instance config to the other nodes
2497 self.cfg.AddInstance(instance)
2499 feedback_fn("* activating the instance's disks on target node")
2500 logger.Info("Starting instance %s on node %s" %
2501 (instance.name, target_node))
2503 disks_ok, dummy = _AssembleInstanceDisks(instance, self.cfg,
2504 ignore_secondaries=True)
2506 _ShutdownInstanceDisks(instance, self.cfg)
2507 raise errors.OpExecError("Can't activate the instance's disks")
2509 feedback_fn("* starting the instance on the target node")
2510 if not rpc.call_instance_start(target_node, instance, None):
2511 _ShutdownInstanceDisks(instance, self.cfg)
2512 raise errors.OpExecError("Could not start instance %s on node %s." %
2513 (instance.name, target_node))
2516 def _CreateBlockDevOnPrimary(cfg, node, instance, device, info):
2517 """Create a tree of block devices on the primary node.
2519 This always creates all devices.
2523 for child in device.children:
2524 if not _CreateBlockDevOnPrimary(cfg, node, instance, child, info):
2527 cfg.SetDiskID(device, node)
2528 new_id = rpc.call_blockdev_create(node, device, device.size,
2529 instance.name, True, info)
2532 if device.physical_id is None:
2533 device.physical_id = new_id
2537 def _CreateBlockDevOnSecondary(cfg, node, instance, device, force, info):
2538 """Create a tree of block devices on a secondary node.
2540 If this device type has to be created on secondaries, create it and
2543 If not, just recurse to children keeping the same 'force' value.
2546 if device.CreateOnSecondary():
2549 for child in device.children:
2550 if not _CreateBlockDevOnSecondary(cfg, node, instance,
2551 child, force, info):
2556 cfg.SetDiskID(device, node)
2557 new_id = rpc.call_blockdev_create(node, device, device.size,
2558 instance.name, False, info)
2561 if device.physical_id is None:
2562 device.physical_id = new_id
2566 def _GenerateUniqueNames(cfg, exts):
2567 """Generate a suitable LV name.
2569 This will generate a logical volume name for the given instance.
2574 new_id = cfg.GenerateUniqueID()
2575 results.append("%s%s" % (new_id, val))
2579 def _GenerateMDDRBDBranch(cfg, primary, secondary, size, names):
2580 """Generate a drbd device complete with its children.
2583 port = cfg.AllocatePort()
2584 vgname = cfg.GetVGName()
2585 dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
2586 logical_id=(vgname, names[0]))
2587 dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
2588 logical_id=(vgname, names[1]))
2589 drbd_dev = objects.Disk(dev_type=constants.LD_DRBD7, size=size,
2590 logical_id = (primary, secondary, port),
2591 children = [dev_data, dev_meta])
2595 def _GenerateDRBD8Branch(cfg, primary, secondary, size, names, iv_name):
2596 """Generate a drbd8 device complete with its children.
2599 port = cfg.AllocatePort()
2600 vgname = cfg.GetVGName()
2601 dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
2602 logical_id=(vgname, names[0]))
2603 dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
2604 logical_id=(vgname, names[1]))
2605 drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
2606 logical_id = (primary, secondary, port),
2607 children = [dev_data, dev_meta],
2611 def _GenerateDiskTemplate(cfg, template_name,
2612 instance_name, primary_node,
2613 secondary_nodes, disk_sz, swap_sz):
2614 """Generate the entire disk layout for a given template type.
2617 #TODO: compute space requirements
2619 vgname = cfg.GetVGName()
2620 if template_name == "diskless":
2622 elif template_name == "plain":
2623 if len(secondary_nodes) != 0:
2624 raise errors.ProgrammerError("Wrong template configuration")
2626 names = _GenerateUniqueNames(cfg, [".sda", ".sdb"])
2627 sda_dev = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2628 logical_id=(vgname, names[0]),
2630 sdb_dev = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2631 logical_id=(vgname, names[1]),
2633 disks = [sda_dev, sdb_dev]
2634 elif template_name == "local_raid1":
2635 if len(secondary_nodes) != 0:
2636 raise errors.ProgrammerError("Wrong template configuration")
2639 names = _GenerateUniqueNames(cfg, [".sda_m1", ".sda_m2",
2640 ".sdb_m1", ".sdb_m2"])
2641 sda_dev_m1 = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2642 logical_id=(vgname, names[0]))
2643 sda_dev_m2 = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2644 logical_id=(vgname, names[1]))
2645 md_sda_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name = "sda",
2647 children = [sda_dev_m1, sda_dev_m2])
2648 sdb_dev_m1 = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2649 logical_id=(vgname, names[2]))
2650 sdb_dev_m2 = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2651 logical_id=(vgname, names[3]))
2652 md_sdb_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name = "sdb",
2654 children = [sdb_dev_m1, sdb_dev_m2])
2655 disks = [md_sda_dev, md_sdb_dev]
2656 elif template_name == constants.DT_REMOTE_RAID1:
2657 if len(secondary_nodes) != 1:
2658 raise errors.ProgrammerError("Wrong template configuration")
2659 remote_node = secondary_nodes[0]
2660 names = _GenerateUniqueNames(cfg, [".sda_data", ".sda_meta",
2661 ".sdb_data", ".sdb_meta"])
2662 drbd_sda_dev = _GenerateMDDRBDBranch(cfg, primary_node, remote_node,
2663 disk_sz, names[0:2])
2664 md_sda_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name="sda",
2665 children = [drbd_sda_dev], size=disk_sz)
2666 drbd_sdb_dev = _GenerateMDDRBDBranch(cfg, primary_node, remote_node,
2667 swap_sz, names[2:4])
2668 md_sdb_dev = objects.Disk(dev_type=constants.LD_MD_R1, iv_name="sdb",
2669 children = [drbd_sdb_dev], size=swap_sz)
2670 disks = [md_sda_dev, md_sdb_dev]
2671 elif template_name == constants.DT_DRBD8:
2672 if len(secondary_nodes) != 1:
2673 raise errors.ProgrammerError("Wrong template configuration")
2674 remote_node = secondary_nodes[0]
2675 names = _GenerateUniqueNames(cfg, [".sda_data", ".sda_meta",
2676 ".sdb_data", ".sdb_meta"])
2677 drbd_sda_dev = _GenerateDRBD8Branch(cfg, primary_node, remote_node,
2678 disk_sz, names[0:2], "sda")
2679 drbd_sdb_dev = _GenerateDRBD8Branch(cfg, primary_node, remote_node,
2680 swap_sz, names[2:4], "sdb")
2681 disks = [drbd_sda_dev, drbd_sdb_dev]
2683 raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
2687 def _GetInstanceInfoText(instance):
2688 """Compute that text that should be added to the disk's metadata.
2691 return "originstname+%s" % instance.name
2694 def _CreateDisks(cfg, instance):
2695 """Create all disks for an instance.
2697 This abstracts away some work from AddInstance.
2700 instance: the instance object
2703 True or False showing the success of the creation process
2706 info = _GetInstanceInfoText(instance)
2708 for device in instance.disks:
2709 logger.Info("creating volume %s for instance %s" %
2710 (device.iv_name, instance.name))
2712 for secondary_node in instance.secondary_nodes:
2713 if not _CreateBlockDevOnSecondary(cfg, secondary_node, instance,
2714 device, False, info):
2715 logger.Error("failed to create volume %s (%s) on secondary node %s!" %
2716 (device.iv_name, device, secondary_node))
2719 if not _CreateBlockDevOnPrimary(cfg, instance.primary_node,
2720 instance, device, info):
2721 logger.Error("failed to create volume %s on primary!" %
2727 def _RemoveDisks(instance, cfg):
2728 """Remove all disks for an instance.
2730 This abstracts away some work from `AddInstance()` and
2731 `RemoveInstance()`. Note that in case some of the devices couldn't
2732 be removed, the removal will continue with the other ones (compare
2733 with `_CreateDisks()`).
2736 instance: the instance object
2739 True or False showing the success of the removal proces
2742 logger.Info("removing block devices for instance %s" % instance.name)
2745 for device in instance.disks:
2746 for node, disk in device.ComputeNodeTree(instance.primary_node):
2747 cfg.SetDiskID(disk, node)
2748 if not rpc.call_blockdev_remove(node, disk):
2749 logger.Error("could not remove block device %s on node %s,"
2750 " continuing anyway" %
2751 (device.iv_name, node))
2756 class LUCreateInstance(LogicalUnit):
2757 """Create an instance.
2760 HPATH = "instance-add"
2761 HTYPE = constants.HTYPE_INSTANCE
2762 _OP_REQP = ["instance_name", "mem_size", "disk_size", "pnode",
2763 "disk_template", "swap_size", "mode", "start", "vcpus",
2764 "wait_for_sync", "ip_check"]
2766 def BuildHooksEnv(self):
2769 This runs on master, primary and secondary nodes of the instance.
2773 "INSTANCE_DISK_TEMPLATE": self.op.disk_template,
2774 "INSTANCE_DISK_SIZE": self.op.disk_size,
2775 "INSTANCE_SWAP_SIZE": self.op.swap_size,
2776 "INSTANCE_ADD_MODE": self.op.mode,
2778 if self.op.mode == constants.INSTANCE_IMPORT:
2779 env["INSTANCE_SRC_NODE"] = self.op.src_node
2780 env["INSTANCE_SRC_PATH"] = self.op.src_path
2781 env["INSTANCE_SRC_IMAGE"] = self.src_image
2783 env.update(_BuildInstanceHookEnv(name=self.op.instance_name,
2784 primary_node=self.op.pnode,
2785 secondary_nodes=self.secondaries,
2786 status=self.instance_status,
2787 os_type=self.op.os_type,
2788 memory=self.op.mem_size,
2789 vcpus=self.op.vcpus,
2790 nics=[(self.inst_ip, self.op.bridge)],
2793 nl = ([self.sstore.GetMasterNode(), self.op.pnode] +
2798 def CheckPrereq(self):
2799 """Check prerequisites.
2802 if self.op.mode not in (constants.INSTANCE_CREATE,
2803 constants.INSTANCE_IMPORT):
2804 raise errors.OpPrereqError("Invalid instance creation mode '%s'" %
2807 if self.op.mode == constants.INSTANCE_IMPORT:
2808 src_node = getattr(self.op, "src_node", None)
2809 src_path = getattr(self.op, "src_path", None)
2810 if src_node is None or src_path is None:
2811 raise errors.OpPrereqError("Importing an instance requires source"
2812 " node and path options")
2813 src_node_full = self.cfg.ExpandNodeName(src_node)
2814 if src_node_full is None:
2815 raise errors.OpPrereqError("Unknown source node '%s'" % src_node)
2816 self.op.src_node = src_node = src_node_full
2818 if not os.path.isabs(src_path):
2819 raise errors.OpPrereqError("The source path must be absolute")
2821 export_info = rpc.call_export_info(src_node, src_path)
2824 raise errors.OpPrereqError("No export found in dir %s" % src_path)
2826 if not export_info.has_section(constants.INISECT_EXP):
2827 raise errors.ProgrammerError("Corrupted export config")
2829 ei_version = export_info.get(constants.INISECT_EXP, 'version')
2830 if (int(ei_version) != constants.EXPORT_VERSION):
2831 raise errors.OpPrereqError("Wrong export version %s (wanted %d)" %
2832 (ei_version, constants.EXPORT_VERSION))
2834 if int(export_info.get(constants.INISECT_INS, 'disk_count')) > 1:
2835 raise errors.OpPrereqError("Can't import instance with more than"
2838 # FIXME: are the old os-es, disk sizes, etc. useful?
2839 self.op.os_type = export_info.get(constants.INISECT_EXP, 'os')
2840 diskimage = os.path.join(src_path, export_info.get(constants.INISECT_INS,
2842 self.src_image = diskimage
2843 else: # INSTANCE_CREATE
2844 if getattr(self.op, "os_type", None) is None:
2845 raise errors.OpPrereqError("No guest OS specified")
2847 # check primary node
2848 pnode = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.pnode))
2850 raise errors.OpPrereqError("Primary node '%s' is unknown" %
2852 self.op.pnode = pnode.name
2854 self.secondaries = []
2855 # disk template and mirror node verification
2856 if self.op.disk_template not in constants.DISK_TEMPLATES:
2857 raise errors.OpPrereqError("Invalid disk template name")
2859 if self.op.disk_template in constants.DTS_NET_MIRROR:
2860 if getattr(self.op, "snode", None) is None:
2861 raise errors.OpPrereqError("The networked disk templates need"
2864 snode_name = self.cfg.ExpandNodeName(self.op.snode)
2865 if snode_name is None:
2866 raise errors.OpPrereqError("Unknown secondary node '%s'" %
2868 elif snode_name == pnode.name:
2869 raise errors.OpPrereqError("The secondary node cannot be"
2870 " the primary node.")
2871 self.secondaries.append(snode_name)
2873 # Check lv size requirements
2874 nodenames = [pnode.name] + self.secondaries
2875 nodeinfo = rpc.call_node_info(nodenames, self.cfg.GetVGName())
2877 # Required free disk space as a function of disk and swap space
2879 constants.DT_DISKLESS: 0,
2880 constants.DT_PLAIN: self.op.disk_size + self.op.swap_size,
2881 constants.DT_LOCAL_RAID1: (self.op.disk_size + self.op.swap_size) * 2,
2882 # 256 MB are added for drbd metadata, 128MB for each drbd device
2883 constants.DT_REMOTE_RAID1: self.op.disk_size + self.op.swap_size + 256,
2884 constants.DT_DRBD8: self.op.disk_size + self.op.swap_size + 256,
2887 if self.op.disk_template not in req_size_dict:
2888 raise errors.ProgrammerError("Disk template '%s' size requirement"
2889 " is unknown" % self.op.disk_template)
2891 req_size = req_size_dict[self.op.disk_template]
2893 for node in nodenames:
2894 info = nodeinfo.get(node, None)
2896 raise errors.OpPrereqError("Cannot get current information"
2897 " from node '%s'" % nodeinfo)
2898 if req_size > info['vg_free']:
2899 raise errors.OpPrereqError("Not enough disk space on target node %s."
2900 " %d MB available, %d MB required" %
2901 (node, info['vg_free'], req_size))
2904 os_obj = rpc.call_os_get(pnode.name, self.op.os_type)
2906 raise errors.OpPrereqError("OS '%s' not in supported os list for"
2907 " primary node" % self.op.os_type)
2909 # instance verification
2910 hostname1 = utils.HostInfo(self.op.instance_name)
2912 self.op.instance_name = instance_name = hostname1.name
2913 instance_list = self.cfg.GetInstanceList()
2914 if instance_name in instance_list:
2915 raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
2918 ip = getattr(self.op, "ip", None)
2919 if ip is None or ip.lower() == "none":
2921 elif ip.lower() == "auto":
2922 inst_ip = hostname1.ip
2924 if not utils.IsValidIP(ip):
2925 raise errors.OpPrereqError("given IP address '%s' doesn't look"
2926 " like a valid IP" % ip)
2928 self.inst_ip = inst_ip
2930 if self.op.start and not self.op.ip_check:
2931 raise errors.OpPrereqError("Cannot ignore IP address conflicts when"
2932 " adding an instance in start mode")
2934 if self.op.ip_check:
2935 if utils.TcpPing(utils.HostInfo().name, hostname1.ip,
2936 constants.DEFAULT_NODED_PORT):
2937 raise errors.OpPrereqError("IP %s of instance %s already in use" %
2938 (hostname1.ip, instance_name))
2940 # bridge verification
2941 bridge = getattr(self.op, "bridge", None)
2943 self.op.bridge = self.cfg.GetDefBridge()
2945 self.op.bridge = bridge
2947 if not rpc.call_bridges_exist(self.pnode.name, [self.op.bridge]):
2948 raise errors.OpPrereqError("target bridge '%s' does not exist on"
2949 " destination node '%s'" %
2950 (self.op.bridge, pnode.name))
2953 self.instance_status = 'up'
2955 self.instance_status = 'down'
2957 def Exec(self, feedback_fn):
2958 """Create and add the instance to the cluster.
2961 instance = self.op.instance_name
2962 pnode_name = self.pnode.name
2964 nic = objects.NIC(bridge=self.op.bridge, mac=self.cfg.GenerateMAC())
2965 if self.inst_ip is not None:
2966 nic.ip = self.inst_ip
2968 disks = _GenerateDiskTemplate(self.cfg,
2969 self.op.disk_template,
2970 instance, pnode_name,
2971 self.secondaries, self.op.disk_size,
2974 iobj = objects.Instance(name=instance, os=self.op.os_type,
2975 primary_node=pnode_name,
2976 memory=self.op.mem_size,
2977 vcpus=self.op.vcpus,
2978 nics=[nic], disks=disks,
2979 disk_template=self.op.disk_template,
2980 status=self.instance_status,
2983 feedback_fn("* creating instance disks...")
2984 if not _CreateDisks(self.cfg, iobj):
2985 _RemoveDisks(iobj, self.cfg)
2986 raise errors.OpExecError("Device creation failed, reverting...")
2988 feedback_fn("adding instance %s to cluster config" % instance)
2990 self.cfg.AddInstance(iobj)
2992 if self.op.wait_for_sync:
2993 disk_abort = not _WaitForSync(self.cfg, iobj, self.proc)
2994 elif iobj.disk_template in constants.DTS_NET_MIRROR:
2995 # make sure the disks are not degraded (still sync-ing is ok)
2997 feedback_fn("* checking mirrors status")
2998 disk_abort = not _WaitForSync(self.cfg, iobj, self.proc, oneshot=True)
3003 _RemoveDisks(iobj, self.cfg)
3004 self.cfg.RemoveInstance(iobj.name)
3005 raise errors.OpExecError("There are some degraded disks for"
3008 feedback_fn("creating os for instance %s on node %s" %
3009 (instance, pnode_name))
3011 if iobj.disk_template != constants.DT_DISKLESS:
3012 if self.op.mode == constants.INSTANCE_CREATE:
3013 feedback_fn("* running the instance OS create scripts...")
3014 if not rpc.call_instance_os_add(pnode_name, iobj, "sda", "sdb"):
3015 raise errors.OpExecError("could not add os for instance %s"
3017 (instance, pnode_name))
3019 elif self.op.mode == constants.INSTANCE_IMPORT:
3020 feedback_fn("* running the instance OS import scripts...")
3021 src_node = self.op.src_node
3022 src_image = self.src_image
3023 if not rpc.call_instance_os_import(pnode_name, iobj, "sda", "sdb",
3024 src_node, src_image):
3025 raise errors.OpExecError("Could not import os for instance"
3027 (instance, pnode_name))
3029 # also checked in the prereq part
3030 raise errors.ProgrammerError("Unknown OS initialization mode '%s'"
3034 logger.Info("starting instance %s on node %s" % (instance, pnode_name))
3035 feedback_fn("* starting instance...")
3036 if not rpc.call_instance_start(pnode_name, iobj, None):
3037 raise errors.OpExecError("Could not start instance")
3040 class LUConnectConsole(NoHooksLU):
3041 """Connect to an instance's console.
3043 This is somewhat special in that it returns the command line that
3044 you need to run on the master node in order to connect to the
3048 _OP_REQP = ["instance_name"]
3050 def CheckPrereq(self):
3051 """Check prerequisites.
3053 This checks that the instance is in the cluster.
3056 instance = self.cfg.GetInstanceInfo(
3057 self.cfg.ExpandInstanceName(self.op.instance_name))
3058 if instance is None:
3059 raise errors.OpPrereqError("Instance '%s' not known" %
3060 self.op.instance_name)
3061 self.instance = instance
3063 def Exec(self, feedback_fn):
3064 """Connect to the console of an instance
3067 instance = self.instance
3068 node = instance.primary_node
3070 node_insts = rpc.call_instance_list([node])[node]
3071 if node_insts is False:
3072 raise errors.OpExecError("Can't connect to node %s." % node)
3074 if instance.name not in node_insts:
3075 raise errors.OpExecError("Instance %s is not running." % instance.name)
3077 logger.Debug("connecting to console of %s on %s" % (instance.name, node))
3079 hyper = hypervisor.GetHypervisor()
3080 console_cmd = hyper.GetShellCommandForConsole(instance.name)
3082 argv = ["ssh", "-q", "-t"]
3083 argv.extend(ssh.KNOWN_HOSTS_OPTS)
3084 argv.extend(ssh.BATCH_MODE_OPTS)
3086 argv.append(console_cmd)
3090 class LUAddMDDRBDComponent(LogicalUnit):
3091 """Adda new mirror member to an instance's disk.
3094 HPATH = "mirror-add"
3095 HTYPE = constants.HTYPE_INSTANCE
3096 _OP_REQP = ["instance_name", "remote_node", "disk_name"]
3098 def BuildHooksEnv(self):
3101 This runs on the master, the primary and all the secondaries.
3105 "NEW_SECONDARY": self.op.remote_node,
3106 "DISK_NAME": self.op.disk_name,
3108 env.update(_BuildInstanceHookEnvByObject(self.instance))
3109 nl = [self.sstore.GetMasterNode(), self.instance.primary_node,
3110 self.op.remote_node,] + list(self.instance.secondary_nodes)
3113 def CheckPrereq(self):
3114 """Check prerequisites.
3116 This checks that the instance is in the cluster.
3119 instance = self.cfg.GetInstanceInfo(
3120 self.cfg.ExpandInstanceName(self.op.instance_name))
3121 if instance is None:
3122 raise errors.OpPrereqError("Instance '%s' not known" %
3123 self.op.instance_name)
3124 self.instance = instance
3126 remote_node = self.cfg.ExpandNodeName(self.op.remote_node)
3127 if remote_node is None:
3128 raise errors.OpPrereqError("Node '%s' not known" % self.op.remote_node)
3129 self.remote_node = remote_node
3131 if remote_node == instance.primary_node:
3132 raise errors.OpPrereqError("The specified node is the primary node of"
3135 if instance.disk_template != constants.DT_REMOTE_RAID1:
3136 raise errors.OpPrereqError("Instance's disk layout is not"
3138 for disk in instance.disks:
3139 if disk.iv_name == self.op.disk_name:
3142 raise errors.OpPrereqError("Can't find this device ('%s') in the"
3143 " instance." % self.op.disk_name)
3144 if len(disk.children) > 1:
3145 raise errors.OpPrereqError("The device already has two slave"
3147 "This would create a 3-disk raid1"
3148 " which we don't allow.")
3151 def Exec(self, feedback_fn):
3152 """Add the mirror component
3156 instance = self.instance
3158 remote_node = self.remote_node
3159 lv_names = [".%s_%s" % (disk.iv_name, suf) for suf in ["data", "meta"]]
3160 names = _GenerateUniqueNames(self.cfg, lv_names)
3161 new_drbd = _GenerateMDDRBDBranch(self.cfg, instance.primary_node,
3162 remote_node, disk.size, names)
3164 logger.Info("adding new mirror component on secondary")
3166 if not _CreateBlockDevOnSecondary(self.cfg, remote_node, instance,
3168 _GetInstanceInfoText(instance)):
3169 raise errors.OpExecError("Failed to create new component on secondary"
3170 " node %s" % remote_node)
3172 logger.Info("adding new mirror component on primary")
3174 if not _CreateBlockDevOnPrimary(self.cfg, instance.primary_node,
3176 _GetInstanceInfoText(instance)):
3177 # remove secondary dev
3178 self.cfg.SetDiskID(new_drbd, remote_node)
3179 rpc.call_blockdev_remove(remote_node, new_drbd)
3180 raise errors.OpExecError("Failed to create volume on primary")
3182 # the device exists now
3183 # call the primary node to add the mirror to md
3184 logger.Info("adding new mirror component to md")
3185 if not rpc.call_blockdev_addchildren(instance.primary_node,
3187 logger.Error("Can't add mirror compoment to md!")
3188 self.cfg.SetDiskID(new_drbd, remote_node)
3189 if not rpc.call_blockdev_remove(remote_node, new_drbd):
3190 logger.Error("Can't rollback on secondary")
3191 self.cfg.SetDiskID(new_drbd, instance.primary_node)
3192 if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
3193 logger.Error("Can't rollback on primary")
3194 raise errors.OpExecError("Can't add mirror component to md array")
3196 disk.children.append(new_drbd)
3198 self.cfg.AddInstance(instance)
3200 _WaitForSync(self.cfg, instance, self.proc)
3205 class LURemoveMDDRBDComponent(LogicalUnit):
3206 """Remove a component from a remote_raid1 disk.
3209 HPATH = "mirror-remove"
3210 HTYPE = constants.HTYPE_INSTANCE
3211 _OP_REQP = ["instance_name", "disk_name", "disk_id"]
3213 def BuildHooksEnv(self):
3216 This runs on the master, the primary and all the secondaries.
3220 "DISK_NAME": self.op.disk_name,
3221 "DISK_ID": self.op.disk_id,
3222 "OLD_SECONDARY": self.old_secondary,
3224 env.update(_BuildInstanceHookEnvByObject(self.instance))
3225 nl = [self.sstore.GetMasterNode(),
3226 self.instance.primary_node] + list(self.instance.secondary_nodes)
3229 def CheckPrereq(self):
3230 """Check prerequisites.
3232 This checks that the instance is in the cluster.
3235 instance = self.cfg.GetInstanceInfo(
3236 self.cfg.ExpandInstanceName(self.op.instance_name))
3237 if instance is None:
3238 raise errors.OpPrereqError("Instance '%s' not known" %
3239 self.op.instance_name)
3240 self.instance = instance
3242 if instance.disk_template != constants.DT_REMOTE_RAID1:
3243 raise errors.OpPrereqError("Instance's disk layout is not"
3245 for disk in instance.disks:
3246 if disk.iv_name == self.op.disk_name:
3249 raise errors.OpPrereqError("Can't find this device ('%s') in the"
3250 " instance." % self.op.disk_name)
3251 for child in disk.children:
3252 if (child.dev_type == constants.LD_DRBD7 and
3253 child.logical_id[2] == self.op.disk_id):
3256 raise errors.OpPrereqError("Can't find the device with this port.")
3258 if len(disk.children) < 2:
3259 raise errors.OpPrereqError("Cannot remove the last component from"
3263 if self.child.logical_id[0] == instance.primary_node:
3267 self.old_secondary = self.child.logical_id[oid]
3269 def Exec(self, feedback_fn):
3270 """Remove the mirror component
3273 instance = self.instance
3276 logger.Info("remove mirror component")
3277 self.cfg.SetDiskID(disk, instance.primary_node)
3278 if not rpc.call_blockdev_removechildren(instance.primary_node,
3280 raise errors.OpExecError("Can't remove child from mirror.")
3282 for node in child.logical_id[:2]:
3283 self.cfg.SetDiskID(child, node)
3284 if not rpc.call_blockdev_remove(node, child):
3285 logger.Error("Warning: failed to remove device from node %s,"
3286 " continuing operation." % node)
3288 disk.children.remove(child)
3289 self.cfg.AddInstance(instance)
3292 class LUReplaceDisks(LogicalUnit):
3293 """Replace the disks of an instance.
3296 HPATH = "mirrors-replace"
3297 HTYPE = constants.HTYPE_INSTANCE
3298 _OP_REQP = ["instance_name", "mode", "disks"]
3300 def BuildHooksEnv(self):
3303 This runs on the master, the primary and all the secondaries.
3307 "MODE": self.op.mode,
3308 "NEW_SECONDARY": self.op.remote_node,
3309 "OLD_SECONDARY": self.instance.secondary_nodes[0],
3311 env.update(_BuildInstanceHookEnvByObject(self.instance))
3313 self.sstore.GetMasterNode(),
3314 self.instance.primary_node,
3316 if self.op.remote_node is not None:
3317 nl.append(self.op.remote_node)
3320 def CheckPrereq(self):
3321 """Check prerequisites.
3323 This checks that the instance is in the cluster.
3326 instance = self.cfg.GetInstanceInfo(
3327 self.cfg.ExpandInstanceName(self.op.instance_name))
3328 if instance is None:
3329 raise errors.OpPrereqError("Instance '%s' not known" %
3330 self.op.instance_name)
3331 self.instance = instance
3332 self.op.instance_name = instance.name
3334 if instance.disk_template not in constants.DTS_NET_MIRROR:
3335 raise errors.OpPrereqError("Instance's disk layout is not"
3336 " network mirrored.")
3338 if len(instance.secondary_nodes) != 1:
3339 raise errors.OpPrereqError("The instance has a strange layout,"
3340 " expected one secondary but found %d" %
3341 len(instance.secondary_nodes))
3343 self.sec_node = instance.secondary_nodes[0]
3345 remote_node = getattr(self.op, "remote_node", None)
3346 if remote_node is not None:
3347 remote_node = self.cfg.ExpandNodeName(remote_node)
3348 if remote_node is None:
3349 raise errors.OpPrereqError("Node '%s' not known" %
3350 self.op.remote_node)
3351 self.remote_node_info = self.cfg.GetNodeInfo(remote_node)
3353 self.remote_node_info = None
3354 if remote_node == instance.primary_node:
3355 raise errors.OpPrereqError("The specified node is the primary node of"
3357 elif remote_node == self.sec_node:
3358 if self.op.mode == constants.REPLACE_DISK_SEC:
3359 # this is for DRBD8, where we can't execute the same mode of
3360 # replacement as for drbd7 (no different port allocated)
3361 raise errors.OpPrereqError("Same secondary given, cannot execute"
3363 # the user gave the current secondary, switch to
3364 # 'no-replace-secondary' mode for drbd7
3366 if (instance.disk_template == constants.DT_REMOTE_RAID1 and
3367 self.op.mode != constants.REPLACE_DISK_ALL):
3368 raise errors.OpPrereqError("Template 'remote_raid1' only allows all"
3369 " disks replacement, not individual ones")
3370 if instance.disk_template == constants.DT_DRBD8:
3371 if (self.op.mode == constants.REPLACE_DISK_ALL and
3372 remote_node is not None):
3373 # switch to replace secondary mode
3374 self.op.mode = constants.REPLACE_DISK_SEC
3376 if self.op.mode == constants.REPLACE_DISK_ALL:
3377 raise errors.OpPrereqError("Template 'drbd8' only allows primary or"
3378 " secondary disk replacement, not"
3380 elif self.op.mode == constants.REPLACE_DISK_PRI:
3381 if remote_node is not None:
3382 raise errors.OpPrereqError("Template 'drbd8' does not allow changing"
3383 " the secondary while doing a primary"
3384 " node disk replacement")
3385 self.tgt_node = instance.primary_node
3386 self.oth_node = instance.secondary_nodes[0]
3387 elif self.op.mode == constants.REPLACE_DISK_SEC:
3388 self.new_node = remote_node # this can be None, in which case
3389 # we don't change the secondary
3390 self.tgt_node = instance.secondary_nodes[0]
3391 self.oth_node = instance.primary_node
3393 raise errors.ProgrammerError("Unhandled disk replace mode")
3395 for name in self.op.disks:
3396 if instance.FindDisk(name) is None:
3397 raise errors.OpPrereqError("Disk '%s' not found for instance '%s'" %
3398 (name, instance.name))
3399 self.op.remote_node = remote_node
3401 def _ExecRR1(self, feedback_fn):
3402 """Replace the disks of an instance.
3405 instance = self.instance
3408 if self.op.remote_node is None:
3409 remote_node = self.sec_node
3411 remote_node = self.op.remote_node
3413 for dev in instance.disks:
3415 lv_names = [".%s_%s" % (dev.iv_name, suf) for suf in ["data", "meta"]]
3416 names = _GenerateUniqueNames(cfg, lv_names)
3417 new_drbd = _GenerateMDDRBDBranch(cfg, instance.primary_node,
3418 remote_node, size, names)
3419 iv_names[dev.iv_name] = (dev, dev.children[0], new_drbd)
3420 logger.Info("adding new mirror component on secondary for %s" %
3423 if not _CreateBlockDevOnSecondary(cfg, remote_node, instance,
3425 _GetInstanceInfoText(instance)):
3426 raise errors.OpExecError("Failed to create new component on"
3427 " secondary node %s\n"
3428 "Full abort, cleanup manually!" %
3431 logger.Info("adding new mirror component on primary")
3433 if not _CreateBlockDevOnPrimary(cfg, instance.primary_node,
3435 _GetInstanceInfoText(instance)):
3436 # remove secondary dev
3437 cfg.SetDiskID(new_drbd, remote_node)
3438 rpc.call_blockdev_remove(remote_node, new_drbd)
3439 raise errors.OpExecError("Failed to create volume on primary!\n"
3440 "Full abort, cleanup manually!!")
3442 # the device exists now
3443 # call the primary node to add the mirror to md
3444 logger.Info("adding new mirror component to md")
3445 if not rpc.call_blockdev_addchildren(instance.primary_node, dev,
3447 logger.Error("Can't add mirror compoment to md!")
3448 cfg.SetDiskID(new_drbd, remote_node)
3449 if not rpc.call_blockdev_remove(remote_node, new_drbd):
3450 logger.Error("Can't rollback on secondary")
3451 cfg.SetDiskID(new_drbd, instance.primary_node)
3452 if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
3453 logger.Error("Can't rollback on primary")
3454 raise errors.OpExecError("Full abort, cleanup manually!!")
3456 dev.children.append(new_drbd)
3457 cfg.AddInstance(instance)
3459 # this can fail as the old devices are degraded and _WaitForSync
3460 # does a combined result over all disks, so we don't check its
3462 _WaitForSync(cfg, instance, self.proc, unlock=True)
3464 # so check manually all the devices
3465 for name in iv_names:
3466 dev, child, new_drbd = iv_names[name]
3467 cfg.SetDiskID(dev, instance.primary_node)
3468 is_degr = rpc.call_blockdev_find(instance.primary_node, dev)[5]
3470 raise errors.OpExecError("MD device %s is degraded!" % name)
3471 cfg.SetDiskID(new_drbd, instance.primary_node)
3472 is_degr = rpc.call_blockdev_find(instance.primary_node, new_drbd)[5]
3474 raise errors.OpExecError("New drbd device %s is degraded!" % name)
3476 for name in iv_names:
3477 dev, child, new_drbd = iv_names[name]
3478 logger.Info("remove mirror %s component" % name)
3479 cfg.SetDiskID(dev, instance.primary_node)
3480 if not rpc.call_blockdev_removechildren(instance.primary_node,
3482 logger.Error("Can't remove child from mirror, aborting"
3483 " *this device cleanup*.\nYou need to cleanup manually!!")
3486 for node in child.logical_id[:2]:
3487 logger.Info("remove child device on %s" % node)
3488 cfg.SetDiskID(child, node)
3489 if not rpc.call_blockdev_remove(node, child):
3490 logger.Error("Warning: failed to remove device from node %s,"
3491 " continuing operation." % node)
3493 dev.children.remove(child)
3495 cfg.AddInstance(instance)
3497 def _ExecD8DiskOnly(self, feedback_fn):
3498 """Replace a disk on the primary or secondary for dbrd8.
3500 The algorithm for replace is quite complicated:
3501 - for each disk to be replaced:
3502 - create new LVs on the target node with unique names
3503 - detach old LVs from the drbd device
3504 - rename old LVs to name_replaced.<time_t>
3505 - rename new LVs to old LVs
3506 - attach the new LVs (with the old names now) to the drbd device
3507 - wait for sync across all devices
3508 - for each modified disk:
3509 - remove old LVs (which have the name name_replaces.<time_t>)
3511 Failures are not very well handled.
3515 warning, info = (self.proc.LogWarning, self.proc.LogInfo)
3516 instance = self.instance
3518 vgname = self.cfg.GetVGName()
3521 tgt_node = self.tgt_node
3522 oth_node = self.oth_node
3524 # Step: check device activation
3525 self.proc.LogStep(1, steps_total, "check device existence")
3526 info("checking volume groups")
3527 my_vg = cfg.GetVGName()
3528 results = rpc.call_vg_list([oth_node, tgt_node])
3530 raise errors.OpExecError("Can't list volume groups on the nodes")
3531 for node in oth_node, tgt_node:
3532 res = results.get(node, False)
3533 if not res or my_vg not in res:
3534 raise errors.OpExecError("Volume group '%s' not found on %s" %
3536 for dev in instance.disks:
3537 if not dev.iv_name in self.op.disks:
3539 for node in tgt_node, oth_node:
3540 info("checking %s on %s" % (dev.iv_name, node))
3541 cfg.SetDiskID(dev, node)
3542 if not rpc.call_blockdev_find(node, dev):
3543 raise errors.OpExecError("Can't find device %s on node %s" %
3544 (dev.iv_name, node))
3546 # Step: check other node consistency
3547 self.proc.LogStep(2, steps_total, "check peer consistency")
3548 for dev in instance.disks:
3549 if not dev.iv_name in self.op.disks:
3551 info("checking %s consistency on %s" % (dev.iv_name, oth_node))
3552 if not _CheckDiskConsistency(self.cfg, dev, oth_node,
3553 oth_node==instance.primary_node):
3554 raise errors.OpExecError("Peer node (%s) has degraded storage, unsafe"
3555 " to replace disks on this node (%s)" %
3556 (oth_node, tgt_node))
3558 # Step: create new storage
3559 self.proc.LogStep(3, steps_total, "allocate new storage")
3560 for dev in instance.disks:
3561 if not dev.iv_name in self.op.disks:
3564 cfg.SetDiskID(dev, tgt_node)
3565 lv_names = [".%s_%s" % (dev.iv_name, suf) for suf in ["data", "meta"]]
3566 names = _GenerateUniqueNames(cfg, lv_names)
3567 lv_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3568 logical_id=(vgname, names[0]))
3569 lv_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3570 logical_id=(vgname, names[1]))
3571 new_lvs = [lv_data, lv_meta]
3572 old_lvs = dev.children
3573 iv_names[dev.iv_name] = (dev, old_lvs, new_lvs)
3574 info("creating new local storage on %s for %s" %
3575 (tgt_node, dev.iv_name))
3576 # since we *always* want to create this LV, we use the
3577 # _Create...OnPrimary (which forces the creation), even if we
3578 # are talking about the secondary node
3579 for new_lv in new_lvs:
3580 if not _CreateBlockDevOnPrimary(cfg, tgt_node, instance, new_lv,
3581 _GetInstanceInfoText(instance)):
3582 raise errors.OpExecError("Failed to create new LV named '%s' on"
3584 (new_lv.logical_id[1], tgt_node))
3586 # Step: for each lv, detach+rename*2+attach
3587 self.proc.LogStep(4, steps_total, "change drbd configuration")
3588 for dev, old_lvs, new_lvs in iv_names.itervalues():
3589 info("detaching %s drbd from local storage" % dev.iv_name)
3590 if not rpc.call_blockdev_removechildren(tgt_node, dev, old_lvs):
3591 raise errors.OpExecError("Can't detach drbd from local storage on node"
3592 " %s for device %s" % (tgt_node, dev.iv_name))
3594 #cfg.Update(instance)
3596 # ok, we created the new LVs, so now we know we have the needed
3597 # storage; as such, we proceed on the target node to rename
3598 # old_lv to _old, and new_lv to old_lv; note that we rename LVs
3599 # using the assumption than logical_id == physical_id (which in
3600 # turn is the unique_id on that node)
3602 # FIXME(iustin): use a better name for the replaced LVs
3603 temp_suffix = int(time.time())
3604 ren_fn = lambda d, suff: (d.physical_id[0],
3605 d.physical_id[1] + "_replaced-%s" % suff)
3606 # build the rename list based on what LVs exist on the node
3608 for to_ren in old_lvs:
3609 find_res = rpc.call_blockdev_find(tgt_node, to_ren)
3610 if find_res is not None: # device exists
3611 rlist.append((to_ren, ren_fn(to_ren, temp_suffix)))
3613 info("renaming the old LVs on the target node")
3614 if not rpc.call_blockdev_rename(tgt_node, rlist):
3615 raise errors.OpExecError("Can't rename old LVs on node %s" % tgt_node)
3616 # now we rename the new LVs to the old LVs
3617 info("renaming the new LVs on the target node")
3618 rlist = [(new, old.physical_id) for old, new in zip(old_lvs, new_lvs)]
3619 if not rpc.call_blockdev_rename(tgt_node, rlist):
3620 raise errors.OpExecError("Can't rename new LVs on node %s" % tgt_node)
3622 for old, new in zip(old_lvs, new_lvs):
3623 new.logical_id = old.logical_id
3624 cfg.SetDiskID(new, tgt_node)
3626 for disk in old_lvs:
3627 disk.logical_id = ren_fn(disk, temp_suffix)
3628 cfg.SetDiskID(disk, tgt_node)
3630 # now that the new lvs have the old name, we can add them to the device
3631 info("adding new mirror component on %s" % tgt_node)
3632 if not rpc.call_blockdev_addchildren(tgt_node, dev, new_lvs):
3633 for new_lv in new_lvs:
3634 if not rpc.call_blockdev_remove(tgt_node, new_lv):
3635 warning("Can't rollback device %s", "manually cleanup unused"
3637 raise errors.OpExecError("Can't add local storage to drbd")
3639 dev.children = new_lvs
3640 cfg.Update(instance)
3642 # Step: wait for sync
3644 # this can fail as the old devices are degraded and _WaitForSync
3645 # does a combined result over all disks, so we don't check its
3647 self.proc.LogStep(5, steps_total, "sync devices")
3648 _WaitForSync(cfg, instance, self.proc, unlock=True)
3650 # so check manually all the devices
3651 for name, (dev, old_lvs, new_lvs) in iv_names.iteritems():
3652 cfg.SetDiskID(dev, instance.primary_node)
3653 is_degr = rpc.call_blockdev_find(instance.primary_node, dev)[5]
3655 raise errors.OpExecError("DRBD device %s is degraded!" % name)
3657 # Step: remove old storage
3658 self.proc.LogStep(6, steps_total, "removing old storage")
3659 for name, (dev, old_lvs, new_lvs) in iv_names.iteritems():
3660 info("remove logical volumes for %s" % name)
3662 cfg.SetDiskID(lv, tgt_node)
3663 if not rpc.call_blockdev_remove(tgt_node, lv):
3664 warning("Can't remove old LV", "manually remove unused LVs")
3667 def _ExecD8Secondary(self, feedback_fn):
3668 """Replace the secondary node for drbd8.
3670 The algorithm for replace is quite complicated:
3671 - for all disks of the instance:
3672 - create new LVs on the new node with same names
3673 - shutdown the drbd device on the old secondary
3674 - disconnect the drbd network on the primary
3675 - create the drbd device on the new secondary
3676 - network attach the drbd on the primary, using an artifice:
3677 the drbd code for Attach() will connect to the network if it
3678 finds a device which is connected to the good local disks but
3680 - wait for sync across all devices
3681 - remove all disks from the old secondary
3683 Failures are not very well handled.
3687 warning, info = (self.proc.LogWarning, self.proc.LogInfo)
3688 instance = self.instance
3690 vgname = self.cfg.GetVGName()
3693 old_node = self.tgt_node
3694 new_node = self.new_node
3695 pri_node = instance.primary_node
3697 # Step: check device activation
3698 self.proc.LogStep(1, steps_total, "check device existence")
3699 info("checking volume groups")
3700 my_vg = cfg.GetVGName()
3701 results = rpc.call_vg_list([pri_node, new_node])
3703 raise errors.OpExecError("Can't list volume groups on the nodes")
3704 for node in pri_node, new_node:
3705 res = results.get(node, False)
3706 if not res or my_vg not in res:
3707 raise errors.OpExecError("Volume group '%s' not found on %s" %
3709 for dev in instance.disks:
3710 if not dev.iv_name in self.op.disks:
3712 info("checking %s on %s" % (dev.iv_name, pri_node))
3713 cfg.SetDiskID(dev, pri_node)
3714 if not rpc.call_blockdev_find(pri_node, dev):
3715 raise errors.OpExecError("Can't find device %s on node %s" %
3716 (dev.iv_name, pri_node))
3718 # Step: check other node consistency
3719 self.proc.LogStep(2, steps_total, "check peer consistency")
3720 for dev in instance.disks:
3721 if not dev.iv_name in self.op.disks:
3723 info("checking %s consistency on %s" % (dev.iv_name, pri_node))
3724 if not _CheckDiskConsistency(self.cfg, dev, pri_node, True, ldisk=True):
3725 raise errors.OpExecError("Primary node (%s) has degraded storage,"
3726 " unsafe to replace the secondary" %
3729 # Step: create new storage
3730 self.proc.LogStep(3, steps_total, "allocate new storage")
3731 for dev in instance.disks:
3733 info("adding new local storage on %s for %s" % (new_node, dev.iv_name))
3734 # since we *always* want to create this LV, we use the
3735 # _Create...OnPrimary (which forces the creation), even if we
3736 # are talking about the secondary node
3737 for new_lv in dev.children:
3738 if not _CreateBlockDevOnPrimary(cfg, new_node, instance, new_lv,
3739 _GetInstanceInfoText(instance)):
3740 raise errors.OpExecError("Failed to create new LV named '%s' on"
3742 (new_lv.logical_id[1], new_node))
3744 iv_names[dev.iv_name] = (dev, dev.children)
3746 self.proc.LogStep(4, steps_total, "changing drbd configuration")
3747 for dev in instance.disks:
3749 info("activating a new drbd on %s for %s" % (new_node, dev.iv_name))
3750 # create new devices on new_node
3751 new_drbd = objects.Disk(dev_type=constants.LD_DRBD8,
3752 logical_id=(pri_node, new_node,
3754 children=dev.children)
3755 if not _CreateBlockDevOnSecondary(cfg, new_node, instance,
3757 _GetInstanceInfoText(instance)):
3758 raise errors.OpExecError("Failed to create new DRBD on"
3759 " node '%s'" % new_node)
3761 for dev in instance.disks:
3762 # we have new devices, shutdown the drbd on the old secondary
3763 info("shutting down drbd for %s on old node" % dev.iv_name)
3764 cfg.SetDiskID(dev, old_node)
3765 if not rpc.call_blockdev_shutdown(old_node, dev):
3766 warning("Failed to shutdown drbd for %s on old node" % dev.iv_name,
3767 "Please cleanup this device manuall as soon as possible")
3769 # we have new storage, we 'rename' the network on the primary
3770 info("switching primary drbd for %s to new secondary node" % dev.iv_name)
3771 cfg.SetDiskID(dev, pri_node)
3772 # rename to the ip of the new node
3773 new_uid = list(dev.physical_id)
3774 new_uid[2] = self.remote_node_info.secondary_ip
3775 rlist = [(dev, tuple(new_uid))]
3776 if not rpc.call_blockdev_rename(pri_node, rlist):
3777 raise errors.OpExecError("Can't detach & re-attach drbd %s on node"
3778 " %s from %s to %s" %
3779 (dev.iv_name, pri_node, old_node, new_node))
3780 dev.logical_id = (pri_node, new_node, dev.logical_id[2])
3781 cfg.SetDiskID(dev, pri_node)
3782 cfg.Update(instance)
3785 # this can fail as the old devices are degraded and _WaitForSync
3786 # does a combined result over all disks, so we don't check its
3788 self.proc.LogStep(5, steps_total, "sync devices")
3789 _WaitForSync(cfg, instance, self.proc, unlock=True)
3791 # so check manually all the devices
3792 for name, (dev, old_lvs) in iv_names.iteritems():
3793 cfg.SetDiskID(dev, pri_node)
3794 is_degr = rpc.call_blockdev_find(pri_node, dev)[5]
3796 raise errors.OpExecError("DRBD device %s is degraded!" % name)
3798 self.proc.LogStep(6, steps_total, "removing old storage")
3799 for name, (dev, old_lvs) in iv_names.iteritems():
3800 info("remove logical volumes for %s" % name)
3802 cfg.SetDiskID(lv, old_node)
3803 if not rpc.call_blockdev_remove(old_node, lv):
3804 warning("Can't remove LV on old secondary",
3805 "Cleanup stale volumes by hand")
3807 def Exec(self, feedback_fn):
3808 """Execute disk replacement.
3810 This dispatches the disk replacement to the appropriate handler.
3813 instance = self.instance
3814 if instance.disk_template == constants.DT_REMOTE_RAID1:
3816 elif instance.disk_template == constants.DT_DRBD8:
3817 if self.op.remote_node is None:
3818 fn = self._ExecD8DiskOnly
3820 fn = self._ExecD8Secondary
3822 raise errors.ProgrammerError("Unhandled disk replacement case")
3823 return fn(feedback_fn)
3826 class LUQueryInstanceData(NoHooksLU):
3827 """Query runtime instance data.
3830 _OP_REQP = ["instances"]
3832 def CheckPrereq(self):
3833 """Check prerequisites.
3835 This only checks the optional instance list against the existing names.
3838 if not isinstance(self.op.instances, list):
3839 raise errors.OpPrereqError("Invalid argument type 'instances'")
3840 if self.op.instances:
3841 self.wanted_instances = []
3842 names = self.op.instances
3844 instance = self.cfg.GetInstanceInfo(self.cfg.ExpandInstanceName(name))
3845 if instance is None:
3846 raise errors.OpPrereqError("No such instance name '%s'" % name)
3847 self.wanted_instances.append(instance)
3849 self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
3850 in self.cfg.GetInstanceList()]
3854 def _ComputeDiskStatus(self, instance, snode, dev):
3855 """Compute block device status.
3858 self.cfg.SetDiskID(dev, instance.primary_node)
3859 dev_pstatus = rpc.call_blockdev_find(instance.primary_node, dev)
3860 if dev.dev_type in constants.LDS_DRBD:
3861 # we change the snode then (otherwise we use the one passed in)
3862 if dev.logical_id[0] == instance.primary_node:
3863 snode = dev.logical_id[1]
3865 snode = dev.logical_id[0]
3868 self.cfg.SetDiskID(dev, snode)
3869 dev_sstatus = rpc.call_blockdev_find(snode, dev)
3874 dev_children = [self._ComputeDiskStatus(instance, snode, child)
3875 for child in dev.children]
3880 "iv_name": dev.iv_name,
3881 "dev_type": dev.dev_type,
3882 "logical_id": dev.logical_id,
3883 "physical_id": dev.physical_id,
3884 "pstatus": dev_pstatus,
3885 "sstatus": dev_sstatus,
3886 "children": dev_children,
3891 def Exec(self, feedback_fn):
3892 """Gather and return data"""
3894 for instance in self.wanted_instances:
3895 remote_info = rpc.call_instance_info(instance.primary_node,
3897 if remote_info and "state" in remote_info:
3900 remote_state = "down"
3901 if instance.status == "down":
3902 config_state = "down"
3906 disks = [self._ComputeDiskStatus(instance, None, device)
3907 for device in instance.disks]
3910 "name": instance.name,
3911 "config_state": config_state,
3912 "run_state": remote_state,
3913 "pnode": instance.primary_node,
3914 "snodes": instance.secondary_nodes,
3916 "memory": instance.memory,
3917 "nics": [(nic.mac, nic.ip, nic.bridge) for nic in instance.nics],
3919 "vcpus": instance.vcpus,
3922 result[instance.name] = idict
3927 class LUSetInstanceParms(LogicalUnit):
3928 """Modifies an instances's parameters.
3931 HPATH = "instance-modify"
3932 HTYPE = constants.HTYPE_INSTANCE
3933 _OP_REQP = ["instance_name"]
3935 def BuildHooksEnv(self):
3938 This runs on the master, primary and secondaries.
3943 args['memory'] = self.mem
3945 args['vcpus'] = self.vcpus
3946 if self.do_ip or self.do_bridge:
3950 ip = self.instance.nics[0].ip
3952 bridge = self.bridge
3954 bridge = self.instance.nics[0].bridge
3955 args['nics'] = [(ip, bridge)]
3956 env = _BuildInstanceHookEnvByObject(self.instance, override=args)
3957 nl = [self.sstore.GetMasterNode(),
3958 self.instance.primary_node] + list(self.instance.secondary_nodes)
3961 def CheckPrereq(self):
3962 """Check prerequisites.
3964 This only checks the instance list against the existing names.
3967 self.mem = getattr(self.op, "mem", None)
3968 self.vcpus = getattr(self.op, "vcpus", None)
3969 self.ip = getattr(self.op, "ip", None)
3970 self.bridge = getattr(self.op, "bridge", None)
3971 if [self.mem, self.vcpus, self.ip, self.bridge].count(None) == 4:
3972 raise errors.OpPrereqError("No changes submitted")
3973 if self.mem is not None:
3975 self.mem = int(self.mem)
3976 except ValueError, err:
3977 raise errors.OpPrereqError("Invalid memory size: %s" % str(err))
3978 if self.vcpus is not None:
3980 self.vcpus = int(self.vcpus)
3981 except ValueError, err:
3982 raise errors.OpPrereqError("Invalid vcpus number: %s" % str(err))
3983 if self.ip is not None:
3985 if self.ip.lower() == "none":
3988 if not utils.IsValidIP(self.ip):
3989 raise errors.OpPrereqError("Invalid IP address '%s'." % self.ip)
3992 self.do_bridge = (self.bridge is not None)
3994 instance = self.cfg.GetInstanceInfo(
3995 self.cfg.ExpandInstanceName(self.op.instance_name))
3996 if instance is None:
3997 raise errors.OpPrereqError("No such instance name '%s'" %
3998 self.op.instance_name)
3999 self.op.instance_name = instance.name
4000 self.instance = instance
4003 def Exec(self, feedback_fn):
4004 """Modifies an instance.
4006 All parameters take effect only at the next restart of the instance.
4009 instance = self.instance
4011 instance.memory = self.mem
4012 result.append(("mem", self.mem))
4014 instance.vcpus = self.vcpus
4015 result.append(("vcpus", self.vcpus))
4017 instance.nics[0].ip = self.ip
4018 result.append(("ip", self.ip))
4020 instance.nics[0].bridge = self.bridge
4021 result.append(("bridge", self.bridge))
4023 self.cfg.AddInstance(instance)
4028 class LUQueryExports(NoHooksLU):
4029 """Query the exports list
4034 def CheckPrereq(self):
4035 """Check that the nodelist contains only existing nodes.
4038 self.nodes = _GetWantedNodes(self, getattr(self.op, "nodes", None))
4040 def Exec(self, feedback_fn):
4041 """Compute the list of all the exported system images.
4044 a dictionary with the structure node->(export-list)
4045 where export-list is a list of the instances exported on
4049 return rpc.call_export_list(self.nodes)
4052 class LUExportInstance(LogicalUnit):
4053 """Export an instance to an image in the cluster.
4056 HPATH = "instance-export"
4057 HTYPE = constants.HTYPE_INSTANCE
4058 _OP_REQP = ["instance_name", "target_node", "shutdown"]
4060 def BuildHooksEnv(self):
4063 This will run on the master, primary node and target node.
4067 "EXPORT_NODE": self.op.target_node,
4068 "EXPORT_DO_SHUTDOWN": self.op.shutdown,
4070 env.update(_BuildInstanceHookEnvByObject(self.instance))
4071 nl = [self.sstore.GetMasterNode(), self.instance.primary_node,
4072 self.op.target_node]
4075 def CheckPrereq(self):
4076 """Check prerequisites.
4078 This checks that the instance name is a valid one.
4081 instance_name = self.cfg.ExpandInstanceName(self.op.instance_name)
4082 self.instance = self.cfg.GetInstanceInfo(instance_name)
4083 if self.instance is None:
4084 raise errors.OpPrereqError("Instance '%s' not found" %
4085 self.op.instance_name)
4088 dst_node_short = self.cfg.ExpandNodeName(self.op.target_node)
4089 self.dst_node = self.cfg.GetNodeInfo(dst_node_short)
4091 if self.dst_node is None:
4092 raise errors.OpPrereqError("Destination node '%s' is unknown." %
4093 self.op.target_node)
4094 self.op.target_node = self.dst_node.name
4096 def Exec(self, feedback_fn):
4097 """Export an instance to an image in the cluster.
4100 instance = self.instance
4101 dst_node = self.dst_node
4102 src_node = instance.primary_node
4103 # shutdown the instance, unless requested not to do so
4104 if self.op.shutdown:
4105 op = opcodes.OpShutdownInstance(instance_name=instance.name)
4106 self.proc.ChainOpCode(op)
4108 vgname = self.cfg.GetVGName()
4113 for disk in instance.disks:
4114 if disk.iv_name == "sda":
4115 # new_dev_name will be a snapshot of an lvm leaf of the one we passed
4116 new_dev_name = rpc.call_blockdev_snapshot(src_node, disk)
4118 if not new_dev_name:
4119 logger.Error("could not snapshot block device %s on node %s" %
4120 (disk.logical_id[1], src_node))
4122 new_dev = objects.Disk(dev_type=constants.LD_LV, size=disk.size,
4123 logical_id=(vgname, new_dev_name),
4124 physical_id=(vgname, new_dev_name),
4125 iv_name=disk.iv_name)
4126 snap_disks.append(new_dev)
4129 if self.op.shutdown:
4130 op = opcodes.OpStartupInstance(instance_name=instance.name,
4132 self.proc.ChainOpCode(op)
4134 # TODO: check for size
4136 for dev in snap_disks:
4137 if not rpc.call_snapshot_export(src_node, dev, dst_node.name,
4139 logger.Error("could not export block device %s from node"
4141 (dev.logical_id[1], src_node, dst_node.name))
4142 if not rpc.call_blockdev_remove(src_node, dev):
4143 logger.Error("could not remove snapshot block device %s from"
4144 " node %s" % (dev.logical_id[1], src_node))
4146 if not rpc.call_finalize_export(dst_node.name, instance, snap_disks):
4147 logger.Error("could not finalize export for instance %s on node %s" %
4148 (instance.name, dst_node.name))
4150 nodelist = self.cfg.GetNodeList()
4151 nodelist.remove(dst_node.name)
4153 # on one-node clusters nodelist will be empty after the removal
4154 # if we proceed the backup would be removed because OpQueryExports
4155 # substitutes an empty list with the full cluster node list.
4157 op = opcodes.OpQueryExports(nodes=nodelist)
4158 exportlist = self.proc.ChainOpCode(op)
4159 for node in exportlist:
4160 if instance.name in exportlist[node]:
4161 if not rpc.call_export_remove(node, instance.name):
4162 logger.Error("could not remove older export for instance %s"
4163 " on node %s" % (instance.name, node))
4166 class TagsLU(NoHooksLU):
4169 This is an abstract class which is the parent of all the other tags LUs.
4172 def CheckPrereq(self):
4173 """Check prerequisites.
4176 if self.op.kind == constants.TAG_CLUSTER:
4177 self.target = self.cfg.GetClusterInfo()
4178 elif self.op.kind == constants.TAG_NODE:
4179 name = self.cfg.ExpandNodeName(self.op.name)
4181 raise errors.OpPrereqError("Invalid node name (%s)" %
4184 self.target = self.cfg.GetNodeInfo(name)
4185 elif self.op.kind == constants.TAG_INSTANCE:
4186 name = self.cfg.ExpandInstanceName(self.op.name)
4188 raise errors.OpPrereqError("Invalid instance name (%s)" %
4191 self.target = self.cfg.GetInstanceInfo(name)
4193 raise errors.OpPrereqError("Wrong tag type requested (%s)" %
4197 class LUGetTags(TagsLU):
4198 """Returns the tags of a given object.
4201 _OP_REQP = ["kind", "name"]
4203 def Exec(self, feedback_fn):
4204 """Returns the tag list.
4207 return self.target.GetTags()
4210 class LUSearchTags(NoHooksLU):
4211 """Searches the tags for a given pattern.
4214 _OP_REQP = ["pattern"]
4216 def CheckPrereq(self):
4217 """Check prerequisites.
4219 This checks the pattern passed for validity by compiling it.
4223 self.re = re.compile(self.op.pattern)
4224 except re.error, err:
4225 raise errors.OpPrereqError("Invalid search pattern '%s': %s" %
4226 (self.op.pattern, err))
4228 def Exec(self, feedback_fn):
4229 """Returns the tag list.
4233 tgts = [("/cluster", cfg.GetClusterInfo())]
4234 ilist = [cfg.GetInstanceInfo(name) for name in cfg.GetInstanceList()]
4235 tgts.extend([("/instances/%s" % i.name, i) for i in ilist])
4236 nlist = [cfg.GetNodeInfo(name) for name in cfg.GetNodeList()]
4237 tgts.extend([("/nodes/%s" % n.name, n) for n in nlist])
4239 for path, target in tgts:
4240 for tag in target.GetTags():
4241 if self.re.search(tag):
4242 results.append((path, tag))
4246 class LUAddTags(TagsLU):
4247 """Sets a tag on a given object.
4250 _OP_REQP = ["kind", "name", "tags"]
4252 def CheckPrereq(self):
4253 """Check prerequisites.
4255 This checks the type and length of the tag name and value.
4258 TagsLU.CheckPrereq(self)
4259 for tag in self.op.tags:
4260 objects.TaggableObject.ValidateTag(tag)
4262 def Exec(self, feedback_fn):
4267 for tag in self.op.tags:
4268 self.target.AddTag(tag)
4269 except errors.TagError, err:
4270 raise errors.OpExecError("Error while setting tag: %s" % str(err))
4272 self.cfg.Update(self.target)
4273 except errors.ConfigurationError:
4274 raise errors.OpRetryError("There has been a modification to the"
4275 " config file and the operation has been"
4276 " aborted. Please retry.")
4279 class LUDelTags(TagsLU):
4280 """Delete a list of tags from a given object.
4283 _OP_REQP = ["kind", "name", "tags"]
4285 def CheckPrereq(self):
4286 """Check prerequisites.
4288 This checks that we have the given tag.
4291 TagsLU.CheckPrereq(self)
4292 for tag in self.op.tags:
4293 objects.TaggableObject.ValidateTag(tag)
4294 del_tags = frozenset(self.op.tags)
4295 cur_tags = self.target.GetTags()
4296 if not del_tags <= cur_tags:
4297 diff_tags = del_tags - cur_tags
4298 diff_names = ["'%s'" % tag for tag in diff_tags]
4300 raise errors.OpPrereqError("Tag(s) %s not found" %
4301 (",".join(diff_names)))
4303 def Exec(self, feedback_fn):
4304 """Remove the tag from the object.
4307 for tag in self.op.tags:
4308 self.target.RemoveTag(tag)
4310 self.cfg.Update(self.target)
4311 except errors.ConfigurationError:
4312 raise errors.OpRetryError("There has been a modification to the"
4313 " config file and the operation has been"
4314 " aborted. Please retry.")