X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/92fd2250e8a41cd48a319942f81052180d36c58b..69750d444a602b7643858086fd3f80e1fc604948:/lib/rpc.py diff --git a/lib/rpc.py b/lib/rpc.py index 8131211..f64848b 100644 --- a/lib/rpc.py +++ b/lib/rpc.py @@ -1,7 +1,7 @@ # # -# Copyright (C) 2006, 2007 Google Inc. +# Copyright (C) 2006, 2007, 2008, 2009, 2010 Google Inc. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by @@ -34,6 +34,8 @@ import os import logging import zlib import base64 +import pycurl +import threading from ganeti import utils from ganeti import objects @@ -41,13 +43,20 @@ from ganeti import http from ganeti import serializer from ganeti import constants from ganeti import errors +from ganeti import netutils +from ganeti import ssconf # pylint has a bug here, doesn't see this import import ganeti.http.client # pylint: disable-msg=W0611 -# Module level variable -_http_manager = None +# Timeout for connecting to nodes (seconds) +_RPC_CONNECT_TIMEOUT = 5 + +_RPC_CLIENT_HEADERS = [ + "Content-type: %s" % http.HTTP_APP_JSON, + "Expect:", + ] # Various time constants for the timeout table _TMO_URGENT = 60 # one minute @@ -71,29 +80,62 @@ _TIMEOUTS = { def Init(): """Initializes the module-global HTTP client manager. - Must be called before using any RPC function. + Must be called before using any RPC function and while exactly one thread is + running. """ - global _http_manager # pylint: disable-msg=W0603 - - assert not _http_manager, "RPC module initialized more than once" + # curl_global_init(3) and curl_global_cleanup(3) must be called with only + # one thread running. This check is just a safety measure -- it doesn't + # cover all cases. + assert threading.activeCount() == 1, \ + "Found more than one active thread when initializing pycURL" - http.InitSsl() + logging.info("Using PycURL %s", pycurl.version) - _http_manager = http.client.HttpClientManager() + pycurl.global_init(pycurl.GLOBAL_ALL) def Shutdown(): """Stops the module-global HTTP client manager. - Must be called before quitting the program. + Must be called before quitting the program and while exactly one thread is + running. """ - global _http_manager # pylint: disable-msg=W0603 + pycurl.global_cleanup() + + +def _ConfigRpcCurl(curl): + noded_cert = str(constants.NODED_CERT_FILE) - if _http_manager: - _http_manager.Shutdown() - _http_manager = None + curl.setopt(pycurl.FOLLOWLOCATION, False) + curl.setopt(pycurl.CAINFO, noded_cert) + curl.setopt(pycurl.SSL_VERIFYHOST, 0) + curl.setopt(pycurl.SSL_VERIFYPEER, True) + curl.setopt(pycurl.SSLCERTTYPE, "PEM") + curl.setopt(pycurl.SSLCERT, noded_cert) + curl.setopt(pycurl.SSLKEYTYPE, "PEM") + curl.setopt(pycurl.SSLKEY, noded_cert) + curl.setopt(pycurl.CONNECTTIMEOUT, _RPC_CONNECT_TIMEOUT) + + +class _RpcThreadLocal(threading.local): + def GetHttpClientPool(self): + """Returns a per-thread HTTP client pool. + + @rtype: L{http.client.HttpClientPool} + + """ + try: + pool = self.hcp + except AttributeError: + pool = http.client.HttpClientPool(_ConfigRpcCurl) + self.hcp = pool + + return pool + + +_thread_local = _RpcThreadLocal() def _RpcTimeout(secs): @@ -111,6 +153,23 @@ def _RpcTimeout(secs): return decorator +def RunWithRPC(fn): + """RPC-wrapper decorator. + + When applied to a function, it runs it with the RPC system + initialized, and it shutsdown the system afterwards. This means the + function must be called without RPC being initialized. + + """ + def wrapper(*args, **kwargs): + Init() + try: + return fn(*args, **kwargs) + finally: + Shutdown() + return wrapper + + class RpcResult(object): """RPC Result class. @@ -193,12 +252,41 @@ class RpcResult(object): else: ec = errors.OpExecError if ecode is not None: - args = (msg, prereq) + args = (msg, ecode) else: args = (msg, ) raise ec(*args) # pylint: disable-msg=W0142 +def _AddressLookup(node_list, + ssc=ssconf.SimpleStore, + nslookup_fn=netutils.Hostname.GetIP): + """Return addresses for given node names. + + @type node_list: list + @param node_list: List of node names + @type ssc: class + @param ssc: SimpleStore class that is used to obtain node->ip mappings + @type nslookup_fn: callable + @param nslookup_fn: function use to do NS lookup + @rtype: list of addresses and/or None's + @returns: List of corresponding addresses, if found + + """ + ss = ssc() + iplist = ss.GetNodePrimaryIPList() + family = ss.GetPrimaryIPFamily() + addresses = [] + ipmap = dict(entry.split() for entry in iplist) + for node in node_list: + address = ipmap.get(node) + if address is None: + address = nslookup_fn(node, family=family) + addresses.append(address) + + return addresses + + class Client: """RPC Client class. @@ -211,17 +299,14 @@ class Client: cause bugs. """ - def __init__(self, procedure, body, port): + def __init__(self, procedure, body, port, address_lookup_fn=_AddressLookup): assert procedure in _TIMEOUTS, ("New RPC call not declared in the" " timeouts table") self.procedure = procedure self.body = body self.port = port - self.nc = {} - - self._ssl_params = \ - http.HttpSslParams(ssl_key_path=constants.NODED_CERT_FILE, - ssl_cert_path=constants.NODED_CERT_FILE) + self._request = {} + self._address_lookup_fn = address_lookup_fn def ConnectList(self, node_list, address_list=None, read_timeout=None): """Add a list of nodes to the target nodes. @@ -232,15 +317,16 @@ class Client: @keyword address_list: either None or a list with node addresses, which must have the same length as the node list @type read_timeout: int - @param read_timeout: overwrites the default read timeout for the - given operation + @param read_timeout: overwrites default timeout for operation """ if address_list is None: - address_list = [None for _ in node_list] - else: - assert len(node_list) == len(address_list), \ - "Name and address lists should have the same length" + # Always use IP address instead of node name + address_list = self._address_lookup_fn(node_list) + + assert len(node_list) == len(address_list), \ + "Name and address lists must have the same length" + for node, address in zip(node_list, address_list): self.ConnectNode(node, address, read_timeout=read_timeout) @@ -250,37 +336,42 @@ class Client: @type name: str @param name: the node name @type address: str - @keyword address: the node address, if known + @param address: the node address, if known + @type read_timeout: int + @param read_timeout: overwrites default timeout for operation """ if address is None: - address = name + # Always use IP address instead of node name + address = self._address_lookup_fn([name])[0] + + assert(address is not None) if read_timeout is None: read_timeout = _TIMEOUTS[self.procedure] - self.nc[name] = \ - http.client.HttpClientRequest(address, self.port, http.HTTP_PUT, - "/%s" % self.procedure, - post_data=self.body, - ssl_params=self._ssl_params, - ssl_verify_peer=True, + self._request[name] = \ + http.client.HttpClientRequest(str(address), self.port, + http.HTTP_PUT, str("/%s" % self.procedure), + headers=_RPC_CLIENT_HEADERS, + post_data=str(self.body), read_timeout=read_timeout) - def GetResults(self): + def GetResults(self, http_pool=None): """Call nodes and return results. @rtype: list @return: List of RPC results """ - assert _http_manager, "RPC module not initialized" + if not http_pool: + http_pool = _thread_local.GetHttpClientPool() - _http_manager.ExecRequests(self.nc.values()) + http_pool.ProcessRequests(self._request.values()) results = {} - for name, req in self.nc.iteritems(): + for name, req in self._request.iteritems(): if req.success and req.resp_status_code == http.HTTP_OK: results[name] = RpcResult(data=serializer.LoadJson(req.resp_body), node=name, call=self.procedure) @@ -327,9 +418,9 @@ class RpcRunner(object): """ self._cfg = cfg - self.port = utils.GetDaemonPort(constants.NODED) + self.port = netutils.GetDaemonPort(constants.NODED) - def _InstDict(self, instance, hvp=None, bep=None): + def _InstDict(self, instance, hvp=None, bep=None, osp=None): """Convert the given instance to a dict. This is done via the instance's ToDict() method and additionally @@ -341,6 +432,8 @@ class RpcRunner(object): @param hvp: a dictionary with overridden hypervisor parameters @type bep: dict or None @param bep: a dictionary with overridden backend parameters + @type osp: dict or None + @param osp: a dictionary with overriden os parameters @rtype: dict @return: the instance dict, with the hvparams filled with the cluster defaults @@ -354,6 +447,9 @@ class RpcRunner(object): idict["beparams"] = cluster.FillBE(instance) if bep is not None: idict["beparams"].update(bep) + idict["osparams"] = cluster.SimpleFillOS(instance.os, instance.osparams) + if osp is not None: + idict["osparams"].update(osp) for nic in idict["nics"]: nic['nicparams'] = objects.FillDict( cluster.nicparams[constants.PP_DEFAULT], @@ -436,7 +532,7 @@ class RpcRunner(object): """ body = serializer.DumpJson(args, indent=False) - c = Client(procedure, body, utils.GetDaemonPort(constants.NODED)) + c = Client(procedure, body, netutils.GetDaemonPort(constants.NODED)) c.ConnectList(node_list, address_list=address_list, read_timeout=read_timeout) return c.GetResults() @@ -459,7 +555,7 @@ class RpcRunner(object): """ body = serializer.DumpJson(args, indent=False) - c = Client(procedure, body, utils.GetDaemonPort(constants.NODED)) + c = Client(procedure, body, netutils.GetDaemonPort(constants.NODED)) c.ConnectNode(node, read_timeout=read_timeout) return c.GetResults()[node] @@ -779,14 +875,20 @@ class RpcRunner(object): [vg_name, hypervisor_type]) @_RpcTimeout(_TMO_NORMAL) - def call_node_add(self, node, dsa, dsapub, rsa, rsapub, ssh, sshpub): - """Add a node to the cluster. + def call_etc_hosts_modify(self, node, mode, name, ip): + """Modify hosts file with name - This is a single-node call. + @type node: string + @param node: The node to call + @type mode: string + @param mode: The mode to operate. Currently "add" or "remove" + @type name: string + @param name: The host name to be modified + @type ip: string + @param ip: The ip of the entry (just valid if mode is "add") """ - return self._SingleNodeCall(node, "node_add", - [dsa, dsapub, rsa, rsapub, ssh, sshpub]) + return self._SingleNodeCall(node, "etc_hosts_modify", [mode, name, ip]) @_RpcTimeout(_TMO_NORMAL) def call_node_verify(self, node_list, checkdict, cluster_name): @@ -988,6 +1090,15 @@ class RpcRunner(object): return self._MultiNodeCall(node_list, "drbd_wait_sync", [nodes_ip, [cf.ToDict() for cf in disks]]) + @_RpcTimeout(_TMO_URGENT) + def call_drbd_helper(self, node_list): + """Gets drbd helper. + + This is a multi-node call. + + """ + return self._MultiNodeCall(node_list, "drbd_helper", []) + @classmethod @_RpcTimeout(_TMO_NORMAL) def call_upload_file(cls, node_list, file_name, address_list=None): @@ -1046,6 +1157,16 @@ class RpcRunner(object): result.payload = objects.OS.FromDict(result.payload) return result + @_RpcTimeout(_TMO_FAST) + def call_os_validate(self, required, nodes, name, checks, params): + """Run a validation routine for a given OS. + + This is a multi-node call. + + """ + return self._MultiNodeCall(nodes, "os_validate", + [required, name, checks, params]) + @_RpcTimeout(_TMO_NORMAL) def call_hooks_runner(self, node_list, hpath, phase, env): """Call the hooks runner. @@ -1265,22 +1386,6 @@ class RpcRunner(object): return cls._StaticMultiNodeCall(node_list, "jobqueue_rename", rename, address_list=address_list) - @classmethod - @_RpcTimeout(_TMO_FAST) - def call_jobqueue_set_drain(cls, node_list, drain_flag): - """Set the drain flag on the queue. - - This is a multi-node call. - - @type node_list: list - @param node_list: the list of nodes to query - @type drain_flag: bool - @param drain_flag: if True, will set the drain flag, otherwise reset it. - - """ - return cls._StaticMultiNodeCall(node_list, "jobqueue_set_drain", - [drain_flag]) - @_RpcTimeout(_TMO_NORMAL) def call_hypervisor_validate_params(self, node_list, hvname, hvparams): """Validate the hypervisor params.