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
39 from ganeti import utils
40 from ganeti import objects
41 from ganeti import http
42 from ganeti import serializer
43 from ganeti import constants
44 from ganeti import errors
46 import ganeti.http.client
49 # Module level variable
54 """Initializes the module-global HTTP client manager.
56 Must be called before using any RPC function.
61 assert not _http_manager, "RPC module initialized more than once"
63 _http_manager = http.client.HttpClientManager()
67 """Stops the module-global HTTP client manager.
69 Must be called before quitting the program.
75 _http_manager.Shutdown()
79 class RpcResult(object):
82 This class holds an RPC result. It is needed since in multi-node
83 calls we can't raise an exception just because one one out of many
84 failed, and therefore we use this class to encapsulate the result.
86 @ivar data: the data payload, for successfull results, or None
88 @ivar failed: whether the operation failed at RPC level (not
89 application level on the remote node)
90 @ivar call: the name of the RPC call
91 @ivar node: the name of the node to which we made the call
92 @ivar offline: whether the operation failed because the node was
93 offline, as opposed to actual failure; offline=True will always
94 imply failed=True, in order to allow simpler checking if
95 the user doesn't care about the exact failure mode
98 def __init__(self, data=None, failed=False, offline=False,
99 call=None, node=None):
101 self.offline = offline
106 self.error = "Node is marked offline"
116 """If the result has failed, raise an OpExecError.
118 This is used so that LU code doesn't have to check for each
119 result, but instead can call this function.
123 raise errors.OpExecError("Call '%s' to node '%s' has failed: %s" %
124 (self.call, self.node, self.error))
126 def RemoteFailMsg(self):
127 """Check if the remote procedure failed.
129 This is valid only for RPC calls which return result of the form
130 (status, data | error_msg).
132 @return: empty string for succcess, otherwise an error message
136 """Helper to ensure we return a 'True' value for error."""
140 return "No error information"
143 return _EnsureErr(self.error)
144 if not isinstance(self.data, (tuple, list)):
145 return "Invalid result type (%s)" % type(self.data)
146 if len(self.data) != 2:
147 return "Invalid result length (%d), expected 2" % len(self.data)
149 return _EnsureErr(self.data[1])
156 This class, given a (remote) method name, a list of parameters and a
157 list of nodes, will contact (in parallel) all nodes, and return a
158 dict of results (key: node name, value: result).
160 One current bug is that generic failure is still signalled by
161 'False' result, which is not good. This overloading of values can
165 def __init__(self, procedure, body, port):
166 self.procedure = procedure
172 http.HttpSslParams(ssl_key_path=constants.SSL_CERT_FILE,
173 ssl_cert_path=constants.SSL_CERT_FILE)
175 def ConnectList(self, node_list, address_list=None):
176 """Add a list of nodes to the target nodes.
178 @type node_list: list
179 @param node_list: the list of node names to connect
180 @type address_list: list or None
181 @keyword address_list: either None or a list with node addresses,
182 which must have the same length as the node list
185 if address_list is None:
186 address_list = [None for _ in node_list]
188 assert len(node_list) == len(address_list), \
189 "Name and address lists should have the same length"
190 for node, address in zip(node_list, address_list):
191 self.ConnectNode(node, address)
193 def ConnectNode(self, name, address=None):
194 """Add a node to the target list.
197 @param name: the node name
199 @keyword address: the node address, if known
206 http.client.HttpClientRequest(address, self.port, http.HTTP_PUT,
207 "/%s" % self.procedure,
209 ssl_params=self._ssl_params,
210 ssl_verify_peer=True)
212 def GetResults(self):
213 """Call nodes and return results.
216 @returns: List of RPC results
219 assert _http_manager, "RPC module not intialized"
221 _http_manager.ExecRequests(self.nc.values())
225 for name, req in self.nc.iteritems():
226 if req.success and req.resp_status_code == http.HTTP_OK:
227 results[name] = RpcResult(data=serializer.LoadJson(req.resp_body),
228 node=name, call=self.procedure)
231 # TODO: Better error reporting
237 logging.error("RPC error from node %s: %s", name, msg)
238 results[name] = RpcResult(data=msg, failed=True, node=name,
244 class RpcRunner(object):
245 """RPC runner class"""
247 def __init__(self, cfg):
248 """Initialized the rpc runner.
250 @type cfg: C{config.ConfigWriter}
251 @param cfg: the configuration object that will be used to get data
256 self.port = utils.GetNodeDaemonPort()
258 def _InstDict(self, instance):
259 """Convert the given instance to a dict.
261 This is done via the instance's ToDict() method and additionally
262 we fill the hvparams with the cluster defaults.
264 @type instance: L{objects.Instance}
265 @param instance: an Instance object
267 @return: the instance dict, with the hvparams filled with the
271 idict = instance.ToDict()
272 cluster = self._cfg.GetClusterInfo()
273 idict["hvparams"] = cluster.FillHV(instance)
274 idict["beparams"] = cluster.FillBE(instance)
277 def _ConnectList(self, client, node_list):
278 """Helper for computing node addresses.
280 @type client: L{Client}
281 @param client: a C{Client} instance
282 @type node_list: list
283 @param node_list: the node list we should connect
286 all_nodes = self._cfg.GetAllNodesInfo()
290 for node in node_list:
291 if node in all_nodes:
292 if all_nodes[node].offline:
293 skip_dict[node] = RpcResult(node=node, offline=True)
295 val = all_nodes[node].primary_ip
298 addr_list.append(val)
299 name_list.append(node)
301 client.ConnectList(name_list, address_list=addr_list)
304 def _ConnectNode(self, client, node):
305 """Helper for computing one node's address.
307 @type client: L{Client}
308 @param client: a C{Client} instance
310 @param node: the node we should connect
313 node_info = self._cfg.GetNodeInfo(node)
314 if node_info is not None:
315 if node_info.offline:
316 return RpcResult(node=node, offline=True)
317 addr = node_info.primary_ip
320 client.ConnectNode(node, address=addr)
322 def _MultiNodeCall(self, node_list, procedure, args):
323 """Helper for making a multi-node call
326 body = serializer.DumpJson(args, indent=False)
327 c = Client(procedure, body, self.port)
328 skip_dict = self._ConnectList(c, node_list)
329 skip_dict.update(c.GetResults())
333 def _StaticMultiNodeCall(cls, node_list, procedure, args,
335 """Helper for making a multi-node static call
338 body = serializer.DumpJson(args, indent=False)
339 c = Client(procedure, body, utils.GetNodeDaemonPort())
340 c.ConnectList(node_list, address_list=address_list)
341 return c.GetResults()
343 def _SingleNodeCall(self, node, procedure, args):
344 """Helper for making a single-node call
347 body = serializer.DumpJson(args, indent=False)
348 c = Client(procedure, body, self.port)
349 result = self._ConnectNode(c, node)
351 # we did connect, node is not offline
352 result = c.GetResults()[node]
356 def _StaticSingleNodeCall(cls, node, procedure, args):
357 """Helper for making a single-node static call
360 body = serializer.DumpJson(args, indent=False)
361 c = Client(procedure, body, utils.GetNodeDaemonPort())
363 return c.GetResults()[node]
367 """Compresses a string for transport over RPC.
369 Small amounts of data are not compressed.
374 @return: Encoded data to send
377 # Small amounts of data are not compressed
379 return (constants.RPC_ENCODING_NONE, data)
381 # Compress with zlib and encode in base64
382 return (constants.RPC_ENCODING_ZLIB_BASE64,
383 base64.b64encode(zlib.compress(data, 3)))
389 def call_volume_list(self, node_list, vg_name):
390 """Gets the logical volumes present in a given volume group.
392 This is a multi-node call.
395 return self._MultiNodeCall(node_list, "volume_list", [vg_name])
397 def call_vg_list(self, node_list):
398 """Gets the volume group list.
400 This is a multi-node call.
403 return self._MultiNodeCall(node_list, "vg_list", [])
405 def call_bridges_exist(self, node, bridges_list):
406 """Checks if a node has all the bridges given.
408 This method checks if all bridges given in the bridges_list are
409 present on the remote node, so that an instance that uses interfaces
410 on those bridges can be started.
412 This is a single-node call.
415 return self._SingleNodeCall(node, "bridges_exist", [bridges_list])
417 def call_instance_start(self, node, instance, extra_args):
418 """Starts an instance.
420 This is a single-node call.
423 return self._SingleNodeCall(node, "instance_start",
424 [self._InstDict(instance), extra_args])
426 def call_instance_shutdown(self, node, instance):
427 """Stops an instance.
429 This is a single-node call.
432 return self._SingleNodeCall(node, "instance_shutdown",
433 [self._InstDict(instance)])
435 def call_instance_migrate(self, node, instance, target, live):
436 """Migrate an instance.
438 This is a single-node call.
441 @param node: the node on which the instance is currently running
442 @type instance: C{objects.Instance}
443 @param instance: the instance definition
445 @param target: the target node name
447 @param live: whether the migration should be done live or not (the
448 interpretation of this parameter is left to the hypervisor)
451 return self._SingleNodeCall(node, "instance_migrate",
452 [self._InstDict(instance), target, live])
454 def call_instance_reboot(self, node, instance, reboot_type, extra_args):
455 """Reboots an instance.
457 This is a single-node call.
460 return self._SingleNodeCall(node, "instance_reboot",
461 [self._InstDict(instance), reboot_type,
464 def call_instance_os_add(self, node, inst):
465 """Installs an OS on the given instance.
467 This is a single-node call.
470 return self._SingleNodeCall(node, "instance_os_add",
471 [self._InstDict(inst)])
473 def call_instance_run_rename(self, node, inst, old_name):
474 """Run the OS rename script for an instance.
476 This is a single-node call.
479 return self._SingleNodeCall(node, "instance_run_rename",
480 [self._InstDict(inst), old_name])
482 def call_instance_info(self, node, instance, hname):
483 """Returns information about a single instance.
485 This is a single-node call.
488 @param node: the list of nodes to query
489 @type instance: string
490 @param instance: the instance name
492 @param hname: the hypervisor type of the instance
495 return self._SingleNodeCall(node, "instance_info", [instance, hname])
497 def call_instance_migratable(self, node, instance):
498 """Checks whether the given instance can be migrated.
500 This is a single-node call.
502 @param node: the node to query
503 @type instance: L{objects.Instance}
504 @param instance: the instance to check
508 return self._SingleNodeCall(node, "instance_migratable",
509 [self._InstDict(instance)])
511 def call_all_instances_info(self, node_list, hypervisor_list):
512 """Returns information about all instances on the given nodes.
514 This is a multi-node call.
516 @type node_list: list
517 @param node_list: the list of nodes to query
518 @type hypervisor_list: list
519 @param hypervisor_list: the hypervisors to query for instances
522 return self._MultiNodeCall(node_list, "all_instances_info",
525 def call_instance_list(self, node_list, hypervisor_list):
526 """Returns the list of running instances on a given node.
528 This is a multi-node call.
530 @type node_list: list
531 @param node_list: the list of nodes to query
532 @type hypervisor_list: list
533 @param hypervisor_list: the hypervisors to query for instances
536 return self._MultiNodeCall(node_list, "instance_list", [hypervisor_list])
538 def call_node_tcp_ping(self, node, source, target, port, timeout,
540 """Do a TcpPing on the remote node
542 This is a single-node call.
545 return self._SingleNodeCall(node, "node_tcp_ping",
546 [source, target, port, timeout,
549 def call_node_has_ip_address(self, node, address):
550 """Checks if a node has the given IP address.
552 This is a single-node call.
555 return self._SingleNodeCall(node, "node_has_ip_address", [address])
557 def call_node_info(self, node_list, vg_name, hypervisor_type):
558 """Return node information.
560 This will return memory information and volume group size and free
563 This is a multi-node call.
565 @type node_list: list
566 @param node_list: the list of nodes to query
567 @type vg_name: C{string}
568 @param vg_name: the name of the volume group to ask for disk space
570 @type hypervisor_type: C{str}
571 @param hypervisor_type: the name of the hypervisor to ask for
575 retux = self._MultiNodeCall(node_list, "node_info",
576 [vg_name, hypervisor_type])
578 for result in retux.itervalues():
579 if result.failed or not isinstance(result.data, dict):
584 log_name = "call_node_info"
586 utils.CheckDict(result.data, {
587 'memory_total' : '-',
590 'vg_size' : 'node_unreachable',
595 def call_node_add(self, node, dsa, dsapub, rsa, rsapub, ssh, sshpub):
596 """Add a node to the cluster.
598 This is a single-node call.
601 return self._SingleNodeCall(node, "node_add",
602 [dsa, dsapub, rsa, rsapub, ssh, sshpub])
604 def call_node_verify(self, node_list, checkdict, cluster_name):
605 """Request verification of given parameters.
607 This is a multi-node call.
610 return self._MultiNodeCall(node_list, "node_verify",
611 [checkdict, cluster_name])
614 def call_node_start_master(cls, node, start_daemons):
615 """Tells a node to activate itself as a master.
617 This is a single-node call.
620 return cls._StaticSingleNodeCall(node, "node_start_master",
624 def call_node_stop_master(cls, node, stop_daemons):
625 """Tells a node to demote itself from master status.
627 This is a single-node call.
630 return cls._StaticSingleNodeCall(node, "node_stop_master", [stop_daemons])
633 def call_master_info(cls, node_list):
634 """Query master info.
636 This is a multi-node call.
639 # TODO: should this method query down nodes?
640 return cls._StaticMultiNodeCall(node_list, "master_info", [])
642 def call_version(self, node_list):
643 """Query node version.
645 This is a multi-node call.
648 return self._MultiNodeCall(node_list, "version", [])
650 def call_blockdev_create(self, node, bdev, size, owner, on_primary, info):
651 """Request creation of a given block device.
653 This is a single-node call.
656 return self._SingleNodeCall(node, "blockdev_create",
657 [bdev.ToDict(), size, owner, on_primary, info])
659 def call_blockdev_remove(self, node, bdev):
660 """Request removal of a given block device.
662 This is a single-node call.
665 return self._SingleNodeCall(node, "blockdev_remove", [bdev.ToDict()])
667 def call_blockdev_rename(self, node, devlist):
668 """Request rename of the given block devices.
670 This is a single-node call.
673 return self._SingleNodeCall(node, "blockdev_rename",
674 [(d.ToDict(), uid) for d, uid in devlist])
676 def call_blockdev_assemble(self, node, disk, owner, on_primary):
677 """Request assembling of a given block device.
679 This is a single-node call.
682 return self._SingleNodeCall(node, "blockdev_assemble",
683 [disk.ToDict(), owner, on_primary])
685 def call_blockdev_shutdown(self, node, disk):
686 """Request shutdown of a given block device.
688 This is a single-node call.
691 return self._SingleNodeCall(node, "blockdev_shutdown", [disk.ToDict()])
693 def call_blockdev_addchildren(self, node, bdev, ndevs):
694 """Request adding a list of children to a (mirroring) device.
696 This is a single-node call.
699 return self._SingleNodeCall(node, "blockdev_addchildren",
701 [disk.ToDict() for disk in ndevs]])
703 def call_blockdev_removechildren(self, node, bdev, ndevs):
704 """Request removing a list of children from a (mirroring) device.
706 This is a single-node call.
709 return self._SingleNodeCall(node, "blockdev_removechildren",
711 [disk.ToDict() for disk in ndevs]])
713 def call_blockdev_getmirrorstatus(self, node, disks):
714 """Request status of a (mirroring) device.
716 This is a single-node call.
719 return self._SingleNodeCall(node, "blockdev_getmirrorstatus",
720 [dsk.ToDict() for dsk in disks])
722 def call_blockdev_find(self, node, disk):
723 """Request identification of a given block device.
725 This is a single-node call.
728 return self._SingleNodeCall(node, "blockdev_find", [disk.ToDict()])
730 def call_blockdev_close(self, node, instance_name, disks):
731 """Closes the given block devices.
733 This is a single-node call.
736 params = [instance_name, [cf.ToDict() for cf in disks]]
737 return self._SingleNodeCall(node, "blockdev_close", params)
740 def call_upload_file(cls, node_list, file_name, address_list=None):
743 The node will refuse the operation in case the file is not on the
746 This is a multi-node call.
748 @type node_list: list
749 @param node_list: the list of node names to upload to
751 @param file_name: the filename to upload
752 @type address_list: list or None
753 @keyword address_list: an optional list of node addresses, in order
754 to optimize the RPC speed
757 file_contents = utils.ReadFile(file_name)
758 data = cls._Compress(file_contents)
759 st = os.stat(file_name)
760 params = [file_name, data, st.st_mode, st.st_uid, st.st_gid,
761 st.st_atime, st.st_mtime]
762 return cls._StaticMultiNodeCall(node_list, "upload_file", params,
763 address_list=address_list)
766 def call_write_ssconf_files(cls, node_list, values):
767 """Write ssconf files.
769 This is a multi-node call.
772 return cls._StaticMultiNodeCall(node_list, "write_ssconf_files", [values])
774 def call_os_diagnose(self, node_list):
775 """Request a diagnose of OS definitions.
777 This is a multi-node call.
780 result = self._MultiNodeCall(node_list, "os_diagnose", [])
782 for node_result in result.values():
783 if not node_result.failed and node_result.data:
784 node_result.data = [objects.OS.FromDict(oss)
785 for oss in node_result.data]
788 def call_os_get(self, node, name):
789 """Returns an OS definition.
791 This is a single-node call.
794 result = self._SingleNodeCall(node, "os_get", [name])
795 if not result.failed and isinstance(result.data, dict):
796 result.data = objects.OS.FromDict(result.data)
799 def call_hooks_runner(self, node_list, hpath, phase, env):
800 """Call the hooks runner.
803 - op: the OpCode instance
804 - env: a dictionary with the environment
806 This is a multi-node call.
809 params = [hpath, phase, env]
810 return self._MultiNodeCall(node_list, "hooks_runner", params)
812 def call_iallocator_runner(self, node, name, idata):
813 """Call an iallocator on a remote node
816 - name: the iallocator name
817 - input: the json-encoded input string
819 This is a single-node call.
822 return self._SingleNodeCall(node, "iallocator_runner", [name, idata])
824 def call_blockdev_grow(self, node, cf_bdev, amount):
825 """Request a snapshot of the given block device.
827 This is a single-node call.
830 return self._SingleNodeCall(node, "blockdev_grow",
831 [cf_bdev.ToDict(), amount])
833 def call_blockdev_snapshot(self, node, cf_bdev):
834 """Request a snapshot of the given block device.
836 This is a single-node call.
839 return self._SingleNodeCall(node, "blockdev_snapshot", [cf_bdev.ToDict()])
841 def call_snapshot_export(self, node, snap_bdev, dest_node, instance,
843 """Request the export of a given snapshot.
845 This is a single-node call.
848 return self._SingleNodeCall(node, "snapshot_export",
849 [snap_bdev.ToDict(), dest_node,
850 self._InstDict(instance), cluster_name, idx])
852 def call_finalize_export(self, node, instance, snap_disks):
853 """Request the completion of an export operation.
855 This writes the export config file, etc.
857 This is a single-node call.
861 for disk in snap_disks:
862 flat_disks.append(disk.ToDict())
864 return self._SingleNodeCall(node, "finalize_export",
865 [self._InstDict(instance), flat_disks])
867 def call_export_info(self, node, path):
868 """Queries the export information in a given path.
870 This is a single-node call.
873 result = self._SingleNodeCall(node, "export_info", [path])
874 if not result.failed and result.data:
875 result.data = objects.SerializableConfigParser.Loads(str(result.data))
878 def call_instance_os_import(self, node, inst, src_node, src_images,
880 """Request the import of a backup into an instance.
882 This is a single-node call.
885 return self._SingleNodeCall(node, "instance_os_import",
886 [self._InstDict(inst), src_node, src_images,
889 def call_export_list(self, node_list):
890 """Gets the stored exports list.
892 This is a multi-node call.
895 return self._MultiNodeCall(node_list, "export_list", [])
897 def call_export_remove(self, node, export):
898 """Requests removal of a given export.
900 This is a single-node call.
903 return self._SingleNodeCall(node, "export_remove", [export])
906 def call_node_leave_cluster(cls, node):
907 """Requests a node to clean the cluster information it has.
909 This will remove the configuration information from the ganeti data
912 This is a single-node call.
915 return cls._StaticSingleNodeCall(node, "node_leave_cluster", [])
917 def call_node_volumes(self, node_list):
918 """Gets all volumes on node(s).
920 This is a multi-node call.
923 return self._MultiNodeCall(node_list, "node_volumes", [])
925 def call_node_demote_from_mc(self, node):
926 """Demote a node from the master candidate role.
928 This is a single-node call.
931 return self._SingleNodeCall(node, "node_demote_from_mc", [])
933 def call_test_delay(self, node_list, duration):
934 """Sleep for a fixed time on given node(s).
936 This is a multi-node call.
939 return self._MultiNodeCall(node_list, "test_delay", [duration])
941 def call_file_storage_dir_create(self, node, file_storage_dir):
942 """Create the given file storage directory.
944 This is a single-node call.
947 return self._SingleNodeCall(node, "file_storage_dir_create",
950 def call_file_storage_dir_remove(self, node, file_storage_dir):
951 """Remove the given file storage directory.
953 This is a single-node call.
956 return self._SingleNodeCall(node, "file_storage_dir_remove",
959 def call_file_storage_dir_rename(self, node, old_file_storage_dir,
960 new_file_storage_dir):
961 """Rename file storage directory.
963 This is a single-node call.
966 return self._SingleNodeCall(node, "file_storage_dir_rename",
967 [old_file_storage_dir, new_file_storage_dir])
970 def call_jobqueue_update(cls, node_list, address_list, file_name, content):
973 This is a multi-node call.
976 return cls._StaticMultiNodeCall(node_list, "jobqueue_update",
977 [file_name, cls._Compress(content)],
978 address_list=address_list)
981 def call_jobqueue_purge(cls, node):
984 This is a single-node call.
987 return cls._StaticSingleNodeCall(node, "jobqueue_purge", [])
990 def call_jobqueue_rename(cls, node_list, address_list, rename):
991 """Rename a job queue file.
993 This is a multi-node call.
996 return cls._StaticMultiNodeCall(node_list, "jobqueue_rename", rename,
997 address_list=address_list)
1000 def call_jobqueue_set_drain(cls, node_list, drain_flag):
1001 """Set the drain flag on the queue.
1003 This is a multi-node call.
1005 @type node_list: list
1006 @param node_list: the list of nodes to query
1007 @type drain_flag: bool
1008 @param drain_flag: if True, will set the drain flag, otherwise reset it.
1011 return cls._StaticMultiNodeCall(node_list, "jobqueue_set_drain",
1014 def call_hypervisor_validate_params(self, node_list, hvname, hvparams):
1015 """Validate the hypervisor params.
1017 This is a multi-node call.
1019 @type node_list: list
1020 @param node_list: the list of nodes to query
1021 @type hvname: string
1022 @param hvname: the hypervisor name
1023 @type hvparams: dict
1024 @param hvparams: the hypervisor parameters to be validated
1027 cluster = self._cfg.GetClusterInfo()
1028 hv_full = cluster.FillDict(cluster.hvparams.get(hvname, {}), hvparams)
1029 return self._MultiNodeCall(node_list, "hypervisor_validate_params",