code.grnet.gr Git - ganeti-local/blob - lib/http/client.py

   1 #
   2 #
   3
   4 # Copyright (C) 2007, 2008, 2010 Google Inc.
   5 #
   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.
  10 #
  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.
  15 #
  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
  19 # 02110-1301, USA.
  20
  21 """HTTP client module.
  22
  23 """
  24
  25 import logging
  26 import pycurl
  27 from cStringIO import StringIO
  28
  29 from ganeti import http
  30 from ganeti import compat
  31 from ganeti import netutils
  32
  33
  34 class HttpClientRequest(object):
  35   def __init__(self, host, port, method, path, headers=None, post_data=None,
  36                read_timeout=None, curl_config_fn=None):
  37     """Describes an HTTP request.
  38
  39     @type host: string
  40     @param host: Hostname
  41     @type port: int
  42     @param port: Port
  43     @type method: string
  44     @param method: Method name
  45     @type path: string
  46     @param path: Request path
  47     @type headers: list or None
  48     @param headers: Additional headers to send, list of strings
  49     @type post_data: string or None
  50     @param post_data: Additional data to send
  51     @type read_timeout: int
  52     @param read_timeout: if passed, it will be used as the read
  53         timeout while reading the response from the server
  54     @type curl_config_fn: callable
  55     @param curl_config_fn: Function to configure cURL object before request
  56                            (Note: if the function configures the connection in
  57                            a way where it wouldn't be efficient to reuse them,
  58                            a "identity" property should be defined, see
  59                            L{HttpClientRequest.identity})
  60
  61     """
  62     assert path.startswith("/"), "Path must start with slash (/)"
  63     assert curl_config_fn is None or callable(curl_config_fn)
  64
  65     # Request attributes
  66     self.host = host
  67     self.port = port
  68     self.method = method
  69     self.path = path
  70     self.read_timeout = read_timeout
  71     self.curl_config_fn = curl_config_fn
  72
  73     if post_data is None:
  74       self.post_data = ""
  75     else:
  76       self.post_data = post_data
  77
  78     if headers is None:
  79       self.headers = []
  80     elif isinstance(headers, dict):
  81       # Support for old interface
  82       self.headers = ["%s: %s" % (name, value)
  83                       for name, value in headers.items()]
  84     else:
  85       self.headers = headers
  86
  87     # Response status
  88     self.success = None
  89     self.error = None
  90
  91     # Response attributes
  92     self.resp_status_code = None
  93     self.resp_body = None
  94
  95   def __repr__(self):
  96     status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),
  97               "%s:%s" % (self.host, self.port),
  98               self.method,
  99               self.path]
 100
 101     return "<%s at %#x>" % (" ".join(status), id(self))
 102
 103   @property
 104   def url(self):
 105     """Returns the full URL for this requests.
 106
 107     """
 108     if netutils.IPAddress.IsValid(self.host):
 109       family = netutils.IPAddress.GetAddressFamily(self.host)
 110       address = netutils.FormatAddress(family, (self.host, self.port))
 111     else:
 112       address = "%s:%s" % (self.host, self.port)
 113     # TODO: Support for non-SSL requests
 114     return "https://%s%s" % (address, self.path)
 115
 116   @property
 117   def identity(self):
 118     """Returns identifier for retrieving a pooled connection for this request.
 119
 120     This allows cURL client objects to be re-used and to cache information
 121     (e.g. SSL session IDs or connections).
 122
 123     """
 124     parts = [self.host, self.port]
 125
 126     if self.curl_config_fn:
 127       try:
 128         parts.append(self.curl_config_fn.identity)
 129       except AttributeError:
 130         pass
 131
 132     return "/".join(str(i) for i in parts)
 133
 134
 135 class _HttpClient(object):
 136   def __init__(self, curl_config_fn):
 137     """Initializes this class.
 138
 139     @type curl_config_fn: callable
 140     @param curl_config_fn: Function to configure cURL object after
 141                            initialization
 142
 143     """
 144     self._req = None
 145
 146     curl = self._CreateCurlHandle()
 147     curl.setopt(pycurl.VERBOSE, False)
 148     curl.setopt(pycurl.NOSIGNAL, True)
 149     curl.setopt(pycurl.USERAGENT, http.HTTP_GANETI_VERSION)
 150     curl.setopt(pycurl.PROXY, "")
 151
 152     # Pass cURL object to external config function
 153     if curl_config_fn:
 154       curl_config_fn(curl)
 155
 156     self._curl = curl
 157
 158   @staticmethod
 159   def _CreateCurlHandle():
 160     """Returns a new cURL object.
 161
 162     """
 163     return pycurl.Curl()
 164
 165   def GetCurlHandle(self):
 166     """Returns the cURL object.
 167
 168     """
 169     return self._curl
 170
 171   def GetCurrentRequest(self):
 172     """Returns the current request.
 173
 174     @rtype: L{HttpClientRequest} or None
 175
 176     """
 177     return self._req
 178
 179   def StartRequest(self, req):
 180     """Starts a request on this client.
 181
 182     @type req: L{HttpClientRequest}
 183     @param req: HTTP request
 184
 185     """
 186     assert not self._req, "Another request is already started"
 187
 188     self._req = req
 189     self._resp_buffer = StringIO()
 190
 191     url = req.url
 192     method = req.method
 193     post_data = req.post_data
 194     headers = req.headers
 195
 196     # PycURL requires strings to be non-unicode
 197     assert isinstance(method, str)
 198     assert isinstance(url, str)
 199     assert isinstance(post_data, str)
 200     assert compat.all(isinstance(i, str) for i in headers)
 201
 202     # Configure cURL object for request
 203     curl = self._curl
 204     curl.setopt(pycurl.CUSTOMREQUEST, str(method))
 205     curl.setopt(pycurl.URL, url)
 206     curl.setopt(pycurl.POSTFIELDS, post_data)
 207     curl.setopt(pycurl.WRITEFUNCTION, self._resp_buffer.write)
 208     curl.setopt(pycurl.HTTPHEADER, headers)
 209
 210     if req.read_timeout is None:
 211       curl.setopt(pycurl.TIMEOUT, 0)
 212     else:
 213       curl.setopt(pycurl.TIMEOUT, int(req.read_timeout))
 214
 215     # Pass cURL object to external config function
 216     if req.curl_config_fn:
 217       req.curl_config_fn(curl)
 218
 219   def Done(self, errmsg):
 220     """Finishes a request.
 221
 222     @type errmsg: string or None
 223     @param errmsg: Error message if request failed
 224
 225     """
 226     req = self._req
 227     assert req, "No request"
 228
 229     logging.debug("Request %s finished, errmsg=%s", req, errmsg)
 230
 231     curl = self._curl
 232
 233     req.success = not bool(errmsg)
 234     req.error = errmsg
 235
 236     # Get HTTP response code
 237     req.resp_status_code = curl.getinfo(pycurl.RESPONSE_CODE)
 238     req.resp_body = self._resp_buffer.getvalue()
 239
 240     # Reset client object
 241     self._req = None
 242     self._resp_buffer = None
 243
 244     # Ensure no potentially large variables are referenced
 245     curl.setopt(pycurl.POSTFIELDS, "")
 246     curl.setopt(pycurl.WRITEFUNCTION, lambda _: None)
 247
 248
 249 class _PooledHttpClient:
 250   """Data structure for HTTP client pool.
 251
 252   """
 253   def __init__(self, identity, client):
 254     """Initializes this class.
 255
 256     @type identity: string
 257     @param identity: Client identifier for pool
 258     @type client: L{_HttpClient}
 259     @param client: HTTP client
 260
 261     """
 262     self.identity = identity
 263     self.client = client
 264     self.lastused = 0
 265
 266   def __repr__(self):
 267     status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),
 268               "id=%s" % self.identity,
 269               "lastuse=%s" % self.lastused,
 270               repr(self.client)]
 271
 272     return "<%s at %#x>" % (" ".join(status), id(self))
 273
 274
 275 class HttpClientPool:
 276   """A simple HTTP client pool.
 277
 278   Supports one pooled connection per identity (see
 279   L{HttpClientRequest.identity}).
 280
 281   """
 282   #: After how many generations to drop unused clients
 283   _MAX_GENERATIONS_DROP = 25
 284
 285   def __init__(self, curl_config_fn):
 286     """Initializes this class.
 287
 288     @type curl_config_fn: callable
 289     @param curl_config_fn: Function to configure cURL object after
 290                            initialization
 291
 292     """
 293     self._curl_config_fn = curl_config_fn
 294     self._generation = 0
 295     self._pool = {}
 296
 297   @staticmethod
 298   def _GetHttpClientCreator():
 299     """Returns callable to create HTTP client.
 300
 301     """
 302     return _HttpClient
 303
 304   def _Get(self, identity):
 305     """Gets an HTTP client from the pool.
 306
 307     @type identity: string
 308     @param identity: Client identifier
 309
 310     """
 311     try:
 312       pclient  = self._pool.pop(identity)
 313     except KeyError:
 314       # Need to create new client
 315       client = self._GetHttpClientCreator()(self._curl_config_fn)
 316       pclient = _PooledHttpClient(identity, client)
 317       logging.debug("Created new client %s", pclient)
 318     else:
 319       logging.debug("Reusing client %s", pclient)
 320
 321     assert pclient.identity == identity
 322
 323     return pclient
 324
 325   def _StartRequest(self, req):
 326     """Starts a request.
 327
 328     @type req: L{HttpClientRequest}
 329     @param req: HTTP request
 330
 331     """
 332     logging.debug("Starting request %r", req)
 333     pclient = self._Get(req.identity)
 334
 335     assert req.identity not in self._pool
 336
 337     pclient.client.StartRequest(req)
 338     pclient.lastused = self._generation
 339
 340     return pclient
 341
 342   def _Return(self, pclients):
 343     """Returns HTTP clients to the pool.
 344
 345     """
 346     for pc in pclients:
 347       logging.debug("Returning client %s to pool", pc)
 348       assert pc.identity not in self._pool
 349       assert pc not in self._pool.values()
 350       self._pool[pc.identity] = pc
 351
 352     # Check for unused clients
 353     for pc in self._pool.values():
 354       if (pc.lastused + self._MAX_GENERATIONS_DROP) < self._generation:
 355         logging.debug("Removing client %s which hasn't been used"
 356                       " for %s generations",
 357                       pc, self._MAX_GENERATIONS_DROP)
 358         self._pool.pop(pc.identity, None)
 359
 360     assert compat.all(pc.lastused >= (self._generation -
 361                                       self._MAX_GENERATIONS_DROP)
 362                       for pc in self._pool.values())
 363
 364   @staticmethod
 365   def _CreateCurlMultiHandle():
 366     """Creates new cURL multi handle.
 367
 368     """
 369     return pycurl.CurlMulti()
 370
 371   def ProcessRequests(self, requests):
 372     """Processes any number of HTTP client requests using pooled objects.
 373
 374     @type requests: list of L{HttpClientRequest}
 375     @param requests: List of all requests
 376
 377     """
 378     multi = self._CreateCurlMultiHandle()
 379
 380     # For client cleanup
 381     self._generation += 1
 382
 383     assert compat.all((req.error is None and
 384                        req.success is None and
 385                        req.resp_status_code is None and
 386                        req.resp_body is None)
 387                       for req in requests)
 388
 389     curl_to_pclient = {}
 390     for req in requests:
 391       pclient = self._StartRequest(req)
 392       curl = pclient.client.GetCurlHandle()
 393       curl_to_pclient[curl] = pclient
 394       multi.add_handle(curl)
 395       assert pclient.client.GetCurrentRequest() == req
 396       assert pclient.lastused >= 0
 397
 398     assert len(curl_to_pclient) == len(requests)
 399
 400     done_count = 0
 401     while True:
 402       (ret, _) = multi.perform()
 403       assert ret in (pycurl.E_MULTI_OK, pycurl.E_CALL_MULTI_PERFORM)
 404
 405       if ret == pycurl.E_CALL_MULTI_PERFORM:
 406         # cURL wants to be called again
 407         continue
 408
 409       while True:
 410         (remaining_messages, successful, failed) = multi.info_read()
 411
 412         for curl in successful:
 413           multi.remove_handle(curl)
 414           done_count += 1
 415           pclient = curl_to_pclient[curl]
 416           req = pclient.client.GetCurrentRequest()
 417           pclient.client.Done(None)
 418           assert req.success
 419           assert not pclient.client.GetCurrentRequest()
 420
 421         for curl, errnum, errmsg in failed:
 422           multi.remove_handle(curl)
 423           done_count += 1
 424           pclient = curl_to_pclient[curl]
 425           req = pclient.client.GetCurrentRequest()
 426           pclient.client.Done("Error %s: %s" % (errnum, errmsg))
 427           assert req.error
 428           assert not pclient.client.GetCurrentRequest()
 429
 430         if remaining_messages == 0:
 431           break
 432
 433       assert done_count <= len(requests)
 434
 435       if done_count == len(requests):
 436         break
 437
 438       # Wait for I/O. The I/O timeout shouldn't be too long so that HTTP
 439       # timeouts, which are only evaluated in multi.perform, aren't
 440       # unnecessarily delayed.
 441       multi.select(1.0)
 442
 443     assert compat.all(pclient.client.GetCurrentRequest() is None
 444                       for pclient in curl_to_pclient.values())
 445
 446     # Return clients to pool
 447     self._Return(curl_to_pclient.values())
 448
 449     assert done_count == len(requests)
 450     assert compat.all(req.error is not None or
 451                       (req.success and
 452                        req.resp_status_code is not None and
 453                        req.resp_body is not None)
 454                       for req in requests)