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 """Inter-node RPC library.
26 # pylint: disable-msg=C0103,R0201,R0904
27 # C0103: Invalid name, since call_ are not valid
28 # R0201: Method could be a function, we keep all rpcs instance methods
29 # as not to change them back and forth between static/instance methods
30 # if they need to start using instance attributes
31 # R0904: Too many public methods
38 from ganeti import utils
39 from ganeti import objects
40 from ganeti import http
41 from ganeti import serializer
42 from ganeti import constants
43 from ganeti import errors
45 import ganeti.http.client
48 # Module level variable
53 """Initializes the module-global HTTP client manager.
55 Must be called before using any RPC function.
60 assert not _http_manager, "RPC module initialized more than once"
62 _http_manager = http.client.HttpClientManager()
66 """Stops the module-global HTTP client manager.
68 Must be called before quitting the program.
74 _http_manager.Shutdown()
78 class RpcResult(object):
81 This class holds an RPC result. It is needed since in multi-node
82 calls we can't raise an exception just because one one out of many
83 failed, and therefore we use this class to encapsulate the result.
85 @ivar data: the data payload, for successful results, or None
87 @ivar failed: whether the operation failed at RPC level (not
88 application level on the remote node)
89 @ivar call: the name of the RPC call
90 @ivar node: the name of the node to which we made the call
91 @ivar offline: whether the operation failed because the node was
92 offline, as opposed to actual failure; offline=True will always
93 imply failed=True, in order to allow simpler checking if
94 the user doesn't care about the exact failure mode
97 def __init__(self, data=None, failed=False, offline=False,
98 call=None, node=None):
100 self.offline = offline
105 self.error = "Node is marked offline"
106 self.data = self.payload = None
109 self.data = self.payload = None
113 if isinstance(data, (tuple, list)) and len(data) == 2:
114 self.payload = data[1]
119 """If the result has failed, raise an OpExecError.
121 This is used so that LU code doesn't have to check for each
122 result, but instead can call this function.
126 raise errors.OpExecError("Call '%s' to node '%s' has failed: %s" %
127 (self.call, self.node, self.error))
129 def RemoteFailMsg(self):
130 """Check if the remote procedure failed.
132 This is valid only for RPC calls which return result of the form
133 (status, data | error_msg).
135 @return: empty string for succcess, otherwise an error message
139 """Helper to ensure we return a 'True' value for error."""
143 return "No error information"
146 return _EnsureErr(self.error)
147 if not isinstance(self.data, (tuple, list)):
148 return "Invalid result type (%s)" % type(self.data)
149 if len(self.data) != 2:
150 return "Invalid result length (%d), expected 2" % len(self.data)
152 return _EnsureErr(self.data[1])
159 This class, given a (remote) method name, a list of parameters and a
160 list of nodes, will contact (in parallel) all nodes, and return a
161 dict of results (key: node name, value: result).
163 One current bug is that generic failure is still signaled by
164 'False' result, which is not good. This overloading of values can
168 def __init__(self, procedure, body, port):
169 self.procedure = procedure
175 http.HttpSslParams(ssl_key_path=constants.SSL_CERT_FILE,
176 ssl_cert_path=constants.SSL_CERT_FILE)
178 def ConnectList(self, node_list, address_list=None):
179 """Add a list of nodes to the target nodes.
181 @type node_list: list
182 @param node_list: the list of node names to connect
183 @type address_list: list or None
184 @keyword address_list: either None or a list with node addresses,
185 which must have the same length as the node list
188 if address_list is None:
189 address_list = [None for _ in node_list]
191 assert len(node_list) == len(address_list), \
192 "Name and address lists should have the same length"
193 for node, address in zip(node_list, address_list):
194 self.ConnectNode(node, address)
196 def ConnectNode(self, name, address=None):
197 """Add a node to the target list.
200 @param name: the node name
202 @keyword address: the node address, if known
209 http.client.HttpClientRequest(address, self.port, http.HTTP_PUT,
210 "/%s" % self.procedure,
212 ssl_params=self._ssl_params,
213 ssl_verify_peer=True)
215 def GetResults(self):
216 """Call nodes and return results.
219 @return: List of RPC results
222 assert _http_manager, "RPC module not initialized"
224 _http_manager.ExecRequests(self.nc.values())
228 for name, req in self.nc.iteritems():
229 if req.success and req.resp_status_code == http.HTTP_OK:
230 results[name] = RpcResult(data=serializer.LoadJson(req.resp_body),
231 node=name, call=self.procedure)
234 # TODO: Better error reporting
240 logging.error("RPC error in %s from node %s: %s",
241 self.procedure, name, msg)
242 results[name] = RpcResult(data=msg, failed=True, node=name,
248 class RpcRunner(object):
249 """RPC runner class"""
251 def __init__(self, cfg):
252 """Initialized the rpc runner.
254 @type cfg: C{config.ConfigWriter}
255 @param cfg: the configuration object that will be used to get data
260 self.port = utils.GetNodeDaemonPort()
262 def _InstDict(self, instance, hvp=None, bep=None):
263 """Convert the given instance to a dict.
265 This is done via the instance's ToDict() method and additionally
266 we fill the hvparams with the cluster defaults.
268 @type instance: L{objects.Instance}
269 @param instance: an Instance object
270 @type hvp: dict or None
271 @param hvp: a dictionary with overridden hypervisor parameters
272 @type bep: dict or None
273 @param bep: a dictionary with overridden backend parameters
275 @return: the instance dict, with the hvparams filled with the
279 idict = instance.ToDict()
280 cluster = self._cfg.GetClusterInfo()
281 idict["hvparams"] = cluster.FillHV(instance)
283 idict["hvparams"].update(hvp)
284 idict["beparams"] = cluster.FillBE(instance)
286 idict["beparams"].update(bep)
289 def _ConnectList(self, client, node_list, call):
290 """Helper for computing node addresses.
292 @type client: L{ganeti.rpc.Client}
293 @param client: a C{Client} instance
294 @type node_list: list
295 @param node_list: the node list we should connect
297 @param call: the name of the remote procedure call, for filling in
298 correctly any eventual offline nodes' results
301 all_nodes = self._cfg.GetAllNodesInfo()
305 for node in node_list:
306 if node in all_nodes:
307 if all_nodes[node].offline:
308 skip_dict[node] = RpcResult(node=node, offline=True, call=call)
310 val = all_nodes[node].primary_ip
313 addr_list.append(val)
314 name_list.append(node)
316 client.ConnectList(name_list, address_list=addr_list)
319 def _ConnectNode(self, client, node, call):
320 """Helper for computing one node's address.
322 @type client: L{ganeti.rpc.Client}
323 @param client: a C{Client} instance
325 @param node: the node we should connect
327 @param call: the name of the remote procedure call, for filling in
328 correctly any eventual offline nodes' results
331 node_info = self._cfg.GetNodeInfo(node)
332 if node_info is not None:
333 if node_info.offline:
334 return RpcResult(node=node, offline=True, call=call)
335 addr = node_info.primary_ip
338 client.ConnectNode(node, address=addr)
340 def _MultiNodeCall(self, node_list, procedure, args):
341 """Helper for making a multi-node call
344 body = serializer.DumpJson(args, indent=False)
345 c = Client(procedure, body, self.port)
346 skip_dict = self._ConnectList(c, node_list, procedure)
347 skip_dict.update(c.GetResults())
351 def _StaticMultiNodeCall(cls, node_list, procedure, args,
353 """Helper for making a multi-node static call
356 body = serializer.DumpJson(args, indent=False)
357 c = Client(procedure, body, utils.GetNodeDaemonPort())
358 c.ConnectList(node_list, address_list=address_list)
359 return c.GetResults()
361 def _SingleNodeCall(self, node, procedure, args):
362 """Helper for making a single-node call
365 body = serializer.DumpJson(args, indent=False)
366 c = Client(procedure, body, self.port)
367 result = self._ConnectNode(c, node, procedure)
369 # we did connect, node is not offline
370 result = c.GetResults()[node]
374 def _StaticSingleNodeCall(cls, node, procedure, args):
375 """Helper for making a single-node static call
378 body = serializer.DumpJson(args, indent=False)
379 c = Client(procedure, body, utils.GetNodeDaemonPort())
381 return c.GetResults()[node]
385 """Compresses a string for transport over RPC.
387 Small amounts of data are not compressed.
392 @return: Encoded data to send
395 # Small amounts of data are not compressed
397 return (constants.RPC_ENCODING_NONE, data)
399 # Compress with zlib and encode in base64
400 return (constants.RPC_ENCODING_ZLIB_BASE64,
401 base64.b64encode(zlib.compress(data, 3)))
407 def call_volume_list(self, node_list, vg_name):
408 """Gets the logical volumes present in a given volume group.
410 This is a multi-node call.
413 return self._MultiNodeCall(node_list, "volume_list", [vg_name])
415 def call_vg_list(self, node_list):
416 """Gets the volume group list.
418 This is a multi-node call.
421 return self._MultiNodeCall(node_list, "vg_list", [])
423 def call_bridges_exist(self, node, bridges_list):
424 """Checks if a node has all the bridges given.
426 This method checks if all bridges given in the bridges_list are
427 present on the remote node, so that an instance that uses interfaces
428 on those bridges can be started.
430 This is a single-node call.
433 return self._SingleNodeCall(node, "bridges_exist", [bridges_list])
435 def call_instance_start(self, node, instance, hvp, bep):
436 """Starts an instance.
438 This is a single-node call.
441 idict = self._InstDict(instance, hvp=hvp, bep=bep)
442 return self._SingleNodeCall(node, "instance_start", [idict])
444 def call_instance_shutdown(self, node, instance):
445 """Stops an instance.
447 This is a single-node call.
450 return self._SingleNodeCall(node, "instance_shutdown",
451 [self._InstDict(instance)])
453 def call_migration_info(self, node, instance):
454 """Gather the information necessary to prepare an instance migration.
456 This is a single-node call.
459 @param node: the node on which the instance is currently running
460 @type instance: C{objects.Instance}
461 @param instance: the instance definition
464 return self._SingleNodeCall(node, "migration_info",
465 [self._InstDict(instance)])
467 def call_accept_instance(self, node, instance, info, target):
468 """Prepare a node to accept an instance.
470 This is a single-node call.
473 @param node: the target node for the migration
474 @type instance: C{objects.Instance}
475 @param instance: the instance definition
476 @type info: opaque/hypervisor specific (string/data)
477 @param info: result for the call_migration_info call
479 @param target: target hostname (usually ip address) (on the node itself)
482 return self._SingleNodeCall(node, "accept_instance",
483 [self._InstDict(instance), info, target])
485 def call_finalize_migration(self, node, instance, info, success):
486 """Finalize any target-node migration specific operation.
488 This is called both in case of a successful migration and in case of error
489 (in which case it should abort the migration).
491 This is a single-node call.
494 @param node: the target node for the migration
495 @type instance: C{objects.Instance}
496 @param instance: the instance definition
497 @type info: opaque/hypervisor specific (string/data)
498 @param info: result for the call_migration_info call
499 @type success: boolean
500 @param success: whether the migration was a success or a failure
503 return self._SingleNodeCall(node, "finalize_migration",
504 [self._InstDict(instance), info, success])
506 def call_instance_migrate(self, node, instance, target, live):
507 """Migrate an instance.
509 This is a single-node call.
512 @param node: the node on which the instance is currently running
513 @type instance: C{objects.Instance}
514 @param instance: the instance definition
516 @param target: the target node name
518 @param live: whether the migration should be done live or not (the
519 interpretation of this parameter is left to the hypervisor)
522 return self._SingleNodeCall(node, "instance_migrate",
523 [self._InstDict(instance), target, live])
525 def call_instance_reboot(self, node, instance, reboot_type):
526 """Reboots an instance.
528 This is a single-node call.
531 return self._SingleNodeCall(node, "instance_reboot",
532 [self._InstDict(instance), reboot_type])
534 def call_instance_os_add(self, node, inst):
535 """Installs an OS on the given instance.
537 This is a single-node call.
540 return self._SingleNodeCall(node, "instance_os_add",
541 [self._InstDict(inst)])
543 def call_instance_run_rename(self, node, inst, old_name):
544 """Run the OS rename script for an instance.
546 This is a single-node call.
549 return self._SingleNodeCall(node, "instance_run_rename",
550 [self._InstDict(inst), old_name])
552 def call_instance_info(self, node, instance, hname):
553 """Returns information about a single instance.
555 This is a single-node call.
558 @param node: the list of nodes to query
559 @type instance: string
560 @param instance: the instance name
562 @param hname: the hypervisor type of the instance
565 return self._SingleNodeCall(node, "instance_info", [instance, hname])
567 def call_instance_migratable(self, node, instance):
568 """Checks whether the given instance can be migrated.
570 This is a single-node call.
572 @param node: the node to query
573 @type instance: L{objects.Instance}
574 @param instance: the instance to check
578 return self._SingleNodeCall(node, "instance_migratable",
579 [self._InstDict(instance)])
581 def call_all_instances_info(self, node_list, hypervisor_list):
582 """Returns information about all instances on the given nodes.
584 This is a multi-node call.
586 @type node_list: list
587 @param node_list: the list of nodes to query
588 @type hypervisor_list: list
589 @param hypervisor_list: the hypervisors to query for instances
592 return self._MultiNodeCall(node_list, "all_instances_info",
595 def call_instance_list(self, node_list, hypervisor_list):
596 """Returns the list of running instances on a given node.
598 This is a multi-node call.
600 @type node_list: list
601 @param node_list: the list of nodes to query
602 @type hypervisor_list: list
603 @param hypervisor_list: the hypervisors to query for instances
606 return self._MultiNodeCall(node_list, "instance_list", [hypervisor_list])
608 def call_node_tcp_ping(self, node, source, target, port, timeout,
610 """Do a TcpPing on the remote node
612 This is a single-node call.
615 return self._SingleNodeCall(node, "node_tcp_ping",
616 [source, target, port, timeout,
619 def call_node_has_ip_address(self, node, address):
620 """Checks if a node has the given IP address.
622 This is a single-node call.
625 return self._SingleNodeCall(node, "node_has_ip_address", [address])
627 def call_node_info(self, node_list, vg_name, hypervisor_type):
628 """Return node information.
630 This will return memory information and volume group size and free
633 This is a multi-node call.
635 @type node_list: list
636 @param node_list: the list of nodes to query
637 @type vg_name: C{string}
638 @param vg_name: the name of the volume group to ask for disk space
640 @type hypervisor_type: C{str}
641 @param hypervisor_type: the name of the hypervisor to ask for
645 retux = self._MultiNodeCall(node_list, "node_info",
646 [vg_name, hypervisor_type])
648 for result in retux.itervalues():
649 if result.failed or not isinstance(result.data, dict):
654 log_name = "call_node_info"
656 utils.CheckDict(result.data, {
657 'memory_total' : '-',
660 'vg_size' : 'node_unreachable',
665 def call_node_add(self, node, dsa, dsapub, rsa, rsapub, ssh, sshpub):
666 """Add a node to the cluster.
668 This is a single-node call.
671 return self._SingleNodeCall(node, "node_add",
672 [dsa, dsapub, rsa, rsapub, ssh, sshpub])
674 def call_node_verify(self, node_list, checkdict, cluster_name):
675 """Request verification of given parameters.
677 This is a multi-node call.
680 return self._MultiNodeCall(node_list, "node_verify",
681 [checkdict, cluster_name])
684 def call_node_start_master(cls, node, start_daemons, no_voting):
685 """Tells a node to activate itself as a master.
687 This is a single-node call.
690 return cls._StaticSingleNodeCall(node, "node_start_master",
691 [start_daemons, no_voting])
694 def call_node_stop_master(cls, node, stop_daemons):
695 """Tells a node to demote itself from master status.
697 This is a single-node call.
700 return cls._StaticSingleNodeCall(node, "node_stop_master", [stop_daemons])
703 def call_master_info(cls, node_list):
704 """Query master info.
706 This is a multi-node call.
709 # TODO: should this method query down nodes?
710 return cls._StaticMultiNodeCall(node_list, "master_info", [])
712 def call_version(self, node_list):
713 """Query node version.
715 This is a multi-node call.
718 return self._MultiNodeCall(node_list, "version", [])
720 def call_blockdev_create(self, node, bdev, size, owner, on_primary, info):
721 """Request creation of a given block device.
723 This is a single-node call.
726 return self._SingleNodeCall(node, "blockdev_create",
727 [bdev.ToDict(), size, owner, on_primary, info])
729 def call_blockdev_remove(self, node, bdev):
730 """Request removal of a given block device.
732 This is a single-node call.
735 return self._SingleNodeCall(node, "blockdev_remove", [bdev.ToDict()])
737 def call_blockdev_rename(self, node, devlist):
738 """Request rename of the given block devices.
740 This is a single-node call.
743 return self._SingleNodeCall(node, "blockdev_rename",
744 [(d.ToDict(), uid) for d, uid in devlist])
746 def call_blockdev_assemble(self, node, disk, owner, on_primary):
747 """Request assembling of a given block device.
749 This is a single-node call.
752 return self._SingleNodeCall(node, "blockdev_assemble",
753 [disk.ToDict(), owner, on_primary])
755 def call_blockdev_shutdown(self, node, disk):
756 """Request shutdown of a given block device.
758 This is a single-node call.
761 return self._SingleNodeCall(node, "blockdev_shutdown", [disk.ToDict()])
763 def call_blockdev_addchildren(self, node, bdev, ndevs):
764 """Request adding a list of children to a (mirroring) device.
766 This is a single-node call.
769 return self._SingleNodeCall(node, "blockdev_addchildren",
771 [disk.ToDict() for disk in ndevs]])
773 def call_blockdev_removechildren(self, node, bdev, ndevs):
774 """Request removing a list of children from a (mirroring) device.
776 This is a single-node call.
779 return self._SingleNodeCall(node, "blockdev_removechildren",
781 [disk.ToDict() for disk in ndevs]])
783 def call_blockdev_getmirrorstatus(self, node, disks):
784 """Request status of a (mirroring) device.
786 This is a single-node call.
789 return self._SingleNodeCall(node, "blockdev_getmirrorstatus",
790 [dsk.ToDict() for dsk in disks])
792 def call_blockdev_find(self, node, disk):
793 """Request identification of a given block device.
795 This is a single-node call.
798 return self._SingleNodeCall(node, "blockdev_find", [disk.ToDict()])
800 def call_blockdev_close(self, node, instance_name, disks):
801 """Closes the given block devices.
803 This is a single-node call.
806 params = [instance_name, [cf.ToDict() for cf in disks]]
807 return self._SingleNodeCall(node, "blockdev_close", params)
809 def call_blockdev_getsizes(self, node, disks):
810 """Returns the size of the given disks.
812 This is a single-node call.
815 params = [[cf.ToDict() for cf in disks]]
816 return self._SingleNodeCall(node, "blockdev_getsize", params)
818 def call_drbd_disconnect_net(self, node_list, nodes_ip, disks):
819 """Disconnects the network of the given drbd devices.
821 This is a multi-node call.
824 return self._MultiNodeCall(node_list, "drbd_disconnect_net",
825 [nodes_ip, [cf.ToDict() for cf in disks]])
827 def call_drbd_attach_net(self, node_list, nodes_ip,
828 disks, instance_name, multimaster):
829 """Disconnects the given drbd devices.
831 This is a multi-node call.
834 return self._MultiNodeCall(node_list, "drbd_attach_net",
835 [nodes_ip, [cf.ToDict() for cf in disks],
836 instance_name, multimaster])
838 def call_drbd_wait_sync(self, node_list, nodes_ip, disks):
839 """Waits for the synchronization of drbd devices is complete.
841 This is a multi-node call.
844 return self._MultiNodeCall(node_list, "drbd_wait_sync",
845 [nodes_ip, [cf.ToDict() for cf in disks]])
848 def call_upload_file(cls, node_list, file_name, address_list=None):
851 The node will refuse the operation in case the file is not on the
854 This is a multi-node call.
856 @type node_list: list
857 @param node_list: the list of node names to upload to
859 @param file_name: the filename to upload
860 @type address_list: list or None
861 @keyword address_list: an optional list of node addresses, in order
862 to optimize the RPC speed
865 file_contents = utils.ReadFile(file_name)
866 data = cls._Compress(file_contents)
867 st = os.stat(file_name)
868 params = [file_name, data, st.st_mode, st.st_uid, st.st_gid,
869 st.st_atime, st.st_mtime]
870 return cls._StaticMultiNodeCall(node_list, "upload_file", params,
871 address_list=address_list)
874 def call_write_ssconf_files(cls, node_list, values):
875 """Write ssconf files.
877 This is a multi-node call.
880 return cls._StaticMultiNodeCall(node_list, "write_ssconf_files", [values])
882 def call_os_diagnose(self, node_list):
883 """Request a diagnose of OS definitions.
885 This is a multi-node call.
888 result = self._MultiNodeCall(node_list, "os_diagnose", [])
890 for node_result in result.values():
891 if not node_result.failed and node_result.data:
892 node_result.data = [objects.OS.FromDict(oss)
893 for oss in node_result.data]
896 def call_os_get(self, node, name):
897 """Returns an OS definition.
899 This is a single-node call.
902 result = self._SingleNodeCall(node, "os_get", [name])
903 if not result.failed and isinstance(result.data, dict):
904 result.data = objects.OS.FromDict(result.data)
907 def call_hooks_runner(self, node_list, hpath, phase, env):
908 """Call the hooks runner.
911 - op: the OpCode instance
912 - env: a dictionary with the environment
914 This is a multi-node call.
917 params = [hpath, phase, env]
918 return self._MultiNodeCall(node_list, "hooks_runner", params)
920 def call_iallocator_runner(self, node, name, idata):
921 """Call an iallocator on a remote node
924 - name: the iallocator name
925 - input: the json-encoded input string
927 This is a single-node call.
930 return self._SingleNodeCall(node, "iallocator_runner", [name, idata])
932 def call_blockdev_grow(self, node, cf_bdev, amount):
933 """Request a snapshot of the given block device.
935 This is a single-node call.
938 return self._SingleNodeCall(node, "blockdev_grow",
939 [cf_bdev.ToDict(), amount])
941 def call_blockdev_snapshot(self, node, cf_bdev):
942 """Request a snapshot of the given block device.
944 This is a single-node call.
947 return self._SingleNodeCall(node, "blockdev_snapshot", [cf_bdev.ToDict()])
949 def call_snapshot_export(self, node, snap_bdev, dest_node, instance,
951 """Request the export of a given snapshot.
953 This is a single-node call.
956 return self._SingleNodeCall(node, "snapshot_export",
957 [snap_bdev.ToDict(), dest_node,
958 self._InstDict(instance), cluster_name, idx])
960 def call_finalize_export(self, node, instance, snap_disks):
961 """Request the completion of an export operation.
963 This writes the export config file, etc.
965 This is a single-node call.
969 for disk in snap_disks:
970 if isinstance(disk, bool):
971 flat_disks.append(disk)
973 flat_disks.append(disk.ToDict())
975 return self._SingleNodeCall(node, "finalize_export",
976 [self._InstDict(instance), flat_disks])
978 def call_export_info(self, node, path):
979 """Queries the export information in a given path.
981 This is a single-node call.
984 result = self._SingleNodeCall(node, "export_info", [path])
985 if not result.failed and result.data:
986 result.data = objects.SerializableConfigParser.Loads(str(result.data))
989 def call_instance_os_import(self, node, inst, src_node, src_images,
991 """Request the import of a backup into an instance.
993 This is a single-node call.
996 return self._SingleNodeCall(node, "instance_os_import",
997 [self._InstDict(inst), src_node, src_images,
1000 def call_export_list(self, node_list):
1001 """Gets the stored exports list.
1003 This is a multi-node call.
1006 return self._MultiNodeCall(node_list, "export_list", [])
1008 def call_export_remove(self, node, export):
1009 """Requests removal of a given export.
1011 This is a single-node call.
1014 return self._SingleNodeCall(node, "export_remove", [export])
1017 def call_node_leave_cluster(cls, node):
1018 """Requests a node to clean the cluster information it has.
1020 This will remove the configuration information from the ganeti data
1023 This is a single-node call.
1026 return cls._StaticSingleNodeCall(node, "node_leave_cluster", [])
1028 def call_node_volumes(self, node_list):
1029 """Gets all volumes on node(s).
1031 This is a multi-node call.
1034 return self._MultiNodeCall(node_list, "node_volumes", [])
1036 def call_node_demote_from_mc(self, node):
1037 """Demote a node from the master candidate role.
1039 This is a single-node call.
1042 return self._SingleNodeCall(node, "node_demote_from_mc", [])
1044 def call_test_delay(self, node_list, duration):
1045 """Sleep for a fixed time on given node(s).
1047 This is a multi-node call.
1050 return self._MultiNodeCall(node_list, "test_delay", [duration])
1052 def call_file_storage_dir_create(self, node, file_storage_dir):
1053 """Create the given file storage directory.
1055 This is a single-node call.
1058 return self._SingleNodeCall(node, "file_storage_dir_create",
1061 def call_file_storage_dir_remove(self, node, file_storage_dir):
1062 """Remove the given file storage directory.
1064 This is a single-node call.
1067 return self._SingleNodeCall(node, "file_storage_dir_remove",
1070 def call_file_storage_dir_rename(self, node, old_file_storage_dir,
1071 new_file_storage_dir):
1072 """Rename file storage directory.
1074 This is a single-node call.
1077 return self._SingleNodeCall(node, "file_storage_dir_rename",
1078 [old_file_storage_dir, new_file_storage_dir])
1081 def call_jobqueue_update(cls, node_list, address_list, file_name, content):
1082 """Update job queue.
1084 This is a multi-node call.
1087 return cls._StaticMultiNodeCall(node_list, "jobqueue_update",
1088 [file_name, cls._Compress(content)],
1089 address_list=address_list)
1092 def call_jobqueue_purge(cls, node):
1095 This is a single-node call.
1098 return cls._StaticSingleNodeCall(node, "jobqueue_purge", [])
1101 def call_jobqueue_rename(cls, node_list, address_list, rename):
1102 """Rename a job queue file.
1104 This is a multi-node call.
1107 return cls._StaticMultiNodeCall(node_list, "jobqueue_rename", rename,
1108 address_list=address_list)
1111 def call_jobqueue_set_drain(cls, node_list, drain_flag):
1112 """Set the drain flag on the queue.
1114 This is a multi-node call.
1116 @type node_list: list
1117 @param node_list: the list of nodes to query
1118 @type drain_flag: bool
1119 @param drain_flag: if True, will set the drain flag, otherwise reset it.
1122 return cls._StaticMultiNodeCall(node_list, "jobqueue_set_drain",
1125 def call_hypervisor_validate_params(self, node_list, hvname, hvparams):
1126 """Validate the hypervisor params.
1128 This is a multi-node call.
1130 @type node_list: list
1131 @param node_list: the list of nodes to query
1132 @type hvname: string
1133 @param hvname: the hypervisor name
1134 @type hvparams: dict
1135 @param hvparams: the hypervisor parameters to be validated
1138 cluster = self._cfg.GetClusterInfo()
1139 hv_full = cluster.FillDict(cluster.hvparams.get(hvname, {}), hvparams)
1140 return self._MultiNodeCall(node_list, "hypervisor_validate_params",