Source Code for Module ganeti.rpc.client

  1  # 
  2  # 
  3   
  4  # Copyright (C) 2013 Google Inc. 
  5  # All rights reserved. 
  6  # 
  7  # Redistribution and use in source and binary forms, with or without 
  8  # modification, are permitted provided that the following conditions are 
  9  # met: 
 10  # 
 11  # 1. Redistributions of source code must retain the above copyright notice, 
 12  # this list of conditions and the following disclaimer. 
 13  # 
 14  # 2. Redistributions in binary form must reproduce the above copyright 
 15  # notice, this list of conditions and the following disclaimer in the 
 16  # documentation and/or other materials provided with the distribution. 
 17  # 
 18  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS 
 19  # IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED 
 20  # TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR 
 21  # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR 
 22  # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
 23  # EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
 24  # PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
 25  # PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
 26  # LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
 27  # NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
 28  # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 
 29   
 30   
 31  """Module for generic RPC clients. 
 32   
 33  """ 
 34   
 35  import logging 
 36  import time 
 37   
 38  import ganeti.rpc.transport as t 
 39   
 40  from ganeti import constants 
 41  from ganeti import errors 
 42  from ganeti.rpc.errors import (ProtocolError, RequestError, LuxiError) 
 43  from ganeti import serializer 
 44   
 45  KEY_METHOD = constants.LUXI_KEY_METHOD 
 46  KEY_ARGS = constants.LUXI_KEY_ARGS 
 47  KEY_SUCCESS = constants.LUXI_KEY_SUCCESS 
 48  KEY_RESULT = constants.LUXI_KEY_RESULT 
 49  KEY_VERSION = constants.LUXI_KEY_VERSION 
 50   
 51   
 52 -def ParseRequest(msg): 
 53    """Parses a request message. 
 54   
 55    """ 
 56    try: 
 57      request = serializer.LoadJson(msg) 
 58    except ValueError, err: 
 59      raise ProtocolError("Invalid RPC request (parsing error): %s" % err) 
 60   
 61    logging.debug("RPC request: %s", request) 
 62   
 63    if not isinstance(request, dict): 
 64      logging.error("RPC request not a dict: %r", msg) 
 65      raise ProtocolError("Invalid RPC request (not a dict)") 
 66   
 67    method = request.get(KEY_METHOD, None) # pylint: disable=E1103 
 68    args = request.get(KEY_ARGS, None) # pylint: disable=E1103 
 69    version = request.get(KEY_VERSION, None) # pylint: disable=E1103 
 70   
 71    if method is None or args is None: 
 72      logging.error("RPC request missing method or arguments: %r", msg) 
 73      raise ProtocolError(("Invalid RPC request (no method or arguments" 
 74                           " in request): %r") % msg) 
 75   
 76    return (method, args, version) 
 77   
 78   
 79 -def ParseResponse(msg): 
 80    """Parses a response message. 
 81   
 82    """ 
 83    # Parse the result 
 84    try: 
 85      data = serializer.LoadJson(msg) 
 86    except KeyboardInterrupt: 
 87      raise 
 88    except Exception, err: 
 89      raise ProtocolError("Error while deserializing response: %s" % str(err)) 
 90   
 91    # Validate response 
 92    if not (isinstance(data, dict) and 
 93            KEY_SUCCESS in data and 
 94            KEY_RESULT in data): 
 95      raise ProtocolError("Invalid response from server: %r" % data) 
 96   
 97    return (data[KEY_SUCCESS], data[KEY_RESULT], 
 98            data.get(KEY_VERSION, None)) # pylint: disable=E1103 
 99   
100   
101 -def FormatResponse(success, result, version=None): 
102    """Formats a response message. 
103   
104    """ 
105    response = { 
106      KEY_SUCCESS: success, 
107      KEY_RESULT: result, 
108      } 
109   
110    if version is not None: 
111      response[KEY_VERSION] = version 
112   
113    logging.debug("RPC response: %s", response) 
114   
115    return serializer.DumpJson(response) 
116   
117   
118 -def FormatRequest(method, args, version=None): 
119    """Formats a request message. 
120   
121    """ 
122    # Build request 
123    request = { 
124      KEY_METHOD: method, 
125      KEY_ARGS: args, 
126      } 
127   
128    if version is not None: 
129      request[KEY_VERSION] = version 
130   
131    # Serialize the request 
132    return serializer.DumpJson(request, 
133                               private_encoder=serializer.EncodeWithPrivateFields) 
134   
135   
136 -def CallRPCMethod(transport_cb, method, args, version=None): 
137    """Send a RPC request via a transport and return the response. 
138   
139    """ 
140    assert callable(transport_cb) 
141    t1 = time.time()*1000 
142    request_msg = FormatRequest(method, args, version=version) 
143    t2 = time.time()*1000 
144    # Send request and wait for response 
145    response_msg = transport_cb(request_msg) 
146    t3 = time.time()*1000 
147    (success, result, resp_version) = ParseResponse(response_msg) 
148    t4 = time.time()*1000 
149    logging.info("CallRPCMethod %s: format: %dms, sock: %dms, " 
150        "parse: %dms", method, int(t2-t1), int(t3-t2), int(t4-t3)) 
151    # Verify version if there was one in the response 
152    if resp_version is not None and resp_version != version: 
153      raise LuxiError("RPC version mismatch, client %s, response %s" % 
154                      (version, resp_version)) 
155   
156    if success: 
157      return result 
158   
159    errors.MaybeRaise(result) 
160    raise RequestError(result) 
161   
162   
163 -class AbstractClient(object): 
164    """High-level client abstraction. 
165   
166    This uses a backing Transport-like class on top of which it 
167    implements data serialization/deserialization. 
168   
169    """ 
170   
171 -  def __init__(self, timeouts=None, transport=t.Transport, 
172                 allow_non_master=False): 
173      """Constructor for the Client class. 
174   
175      If timeout is not passed, the default timeouts of the transport 
176      class are used. 
177   
178      @type timeouts: list of ints 
179      @param timeouts: timeouts to be used on connect and read/write 
180      @type transport: L{Transport} or another compatible class 
181      @param transport: the underlying transport to use for the RPC calls 
182      @type allow_non_master: bool 
183      @param allow_non_master: skip checks for the master node on errors 
184   
185      """ 
186      self.timeouts = timeouts 
187      self.transport_class = transport 
188      self.allow_non_master = allow_non_master 
189      self.transport = None 
190      # The version used in RPC communication, by default unused: 
191      self.version = None 
192   
193 -  def _GetAddress(self): 
194      """Returns the socket address 
195   
196      """ 
197      raise NotImplementedError 
198   
199 -  def _InitTransport(self): 
200      """(Re)initialize the transport if needed. 
201   
202      """ 
203      if self.transport is None: 
204        self.transport = \ 
205          self.transport_class(self._GetAddress(), 
206                               timeouts=self.timeouts, 
207                               allow_non_master=self.allow_non_master) 
208   
209 -  def _CloseTransport(self): 
210      """Close the transport, ignoring errors. 
211   
212      """ 
213      if self.transport is None: 
214        return 
215      try: 
216        old_transp = self.transport 
217        self.transport = None 
218        old_transp.Close() 
219      except Exception: # pylint: disable=W0703 
220        pass 
221   
222 -  def _SendMethodCall(self, data): 
223      # Send request and wait for response 
224      def send(try_no): 
225        if try_no: 
226          logging.debug("RPC peer disconnected, retrying") 
227        self._InitTransport() 
228        return self.transport.Call(data) 
229      return t.Transport.RetryOnNetworkError(send, 
230                                             lambda _: self._CloseTransport()) 
231   
232 -  def Close(self): 
233      """Close the underlying connection. 
234   
235      """ 
236      self._CloseTransport() 
237   
238 -  def close(self): 
239      """Same as L{Close}, to be used with contextlib.closing(...). 
240   
241      """ 
242      self.Close() 
243   
244 -  def CallMethod(self, method, args): 
245      """Send a generic request and return the response. 
246   
247      """ 
248      if not isinstance(args, (list, tuple)): 
249        raise errors.ProgrammerError("Invalid parameter passed to CallMethod:" 
250                                     " expected list, got %s" % type(args)) 
251      return CallRPCMethod(self._SendMethodCall, method, args, 
252                           version=self.version) 
253   
254   
255 -class AbstractStubClient(AbstractClient): 
256    """An abstract Client that connects a generated stub client to a L{Transport}. 
257   
258    Subclasses should inherit from this class (first) as well and a designated 
259    stub (second). 
260    """ 
261   
262 -  def __init__(self, timeouts=None, transport=t.Transport, 
263                 allow_non_master=None): 
264      """Constructor for the class. 
265   
266      Arguments are the same as for L{AbstractClient}. Checks that SOCKET_PATH 
267      attribute is defined (in the stub class). 
268   
269      @type timeouts: list of ints 
270      @param timeouts: timeouts to be used on connect and read/write 
271      @type transport: L{Transport} or another compatible class 
272      @param transport: the underlying transport to use for the RPC calls 
273      @type allow_non_master: bool 
274      @param allow_non_master: skip checks for the master node on errors 
275      """ 
276   
277      super(AbstractStubClient, self).__init__(timeouts=timeouts, 
278                                               transport=transport, 
279                                               allow_non_master=allow_non_master) 
280   
281 -  def _GenericInvoke(self, method, *args): 
282      return self.CallMethod(method, args) 
283   
284 -  def _GetAddress(self): 
285      return self._GetSocketPath() # pylint: disable=E1101 
286