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       address = netutils.FormatAddress((self.host, self.port))
 110     else:
 111       address = "%s:%s" % (self.host, self.port)
 112     # TODO: Support for non-SSL requests
 113     return "https://%s%s" % (address, self.path)
 114
 115   @property
 116   def identity(self):
 117     """Returns identifier for retrieving a pooled connection for this request.
 118
 119     This allows cURL client objects to be re-used and to cache information
 120     (e.g. SSL session IDs or connections).
 121
 122     """
 123     parts = [self.host, self.port]
 124
 125     if self.curl_config_fn:
 126       try:
 127         parts.append(self.curl_config_fn.identity)
 128       except AttributeError:
 129         pass
 130
 131     return "/".join(str(i) for i in parts)
 132
 133
 134 class _HttpClient(object):
 135   def __init__(self, curl_config_fn):
 136     """Initializes this class.
 137
 138     @type curl_config_fn: callable
 139     @param curl_config_fn: Function to configure cURL object after
 140                            initialization
 141
 142     """
 143     self._req = None
 144
 145     curl = self._CreateCurlHandle()
 146     curl.setopt(pycurl.VERBOSE, False)
 147     curl.setopt(pycurl.NOSIGNAL, True)
 148     curl.setopt(pycurl.USERAGENT, http.HTTP_GANETI_VERSION)
 149     curl.setopt(pycurl.PROXY, "")
 150
 151     # Disable SSL session ID caching (pycurl >= 7.16.0)
 152     if hasattr(pycurl, "SSL_SESSIONID_CACHE"):
 153       curl.setopt(pycurl.SSL_SESSIONID_CACHE, False)
 154
 155     # Pass cURL object to external config function
 156     if curl_config_fn:
 157       curl_config_fn(curl)
 158
 159     self._curl = curl
 160
 161   @staticmethod
 162   def _CreateCurlHandle():
 163     """Returns a new cURL object.
 164
 165     """
 166     return pycurl.Curl()
 167
 168   def GetCurlHandle(self):
 169     """Returns the cURL object.
 170
 171     """
 172     return self._curl
 173
 174   def GetCurrentRequest(self):
 175     """Returns the current request.
 176
 177     @rtype: L{HttpClientRequest} or None
 178
 179     """
 180     return self._req
 181
 182   def StartRequest(self, req):
 183     """Starts a request on this client.
 184
 185     @type req: L{HttpClientRequest}
 186     @param req: HTTP request
 187
 188     """
 189     assert not self._req, "Another request is already started"
 190
 191     logging.debug("Starting request %r", req)
 192
 193     self._req = req
 194     self._resp_buffer = StringIO()
 195
 196     url = req.url
 197     method = req.method
 198     post_data = req.post_data
 199     headers = req.headers
 200
 201     # PycURL requires strings to be non-unicode
 202     assert isinstance(method, str)
 203     assert isinstance(url, str)
 204     assert isinstance(post_data, str)
 205     assert compat.all(isinstance(i, str) for i in headers)
 206
 207     # Configure cURL object for request
 208     curl = self._curl
 209     curl.setopt(pycurl.CUSTOMREQUEST, str(method))
 210     curl.setopt(pycurl.URL, url)
 211     curl.setopt(pycurl.POSTFIELDS, post_data)
 212     curl.setopt(pycurl.WRITEFUNCTION, self._resp_buffer.write)
 213     curl.setopt(pycurl.HTTPHEADER, headers)
 214
 215     if req.read_timeout is None:
 216       curl.setopt(pycurl.TIMEOUT, 0)
 217     else:
 218       curl.setopt(pycurl.TIMEOUT, int(req.read_timeout))
 219
 220     # Pass cURL object to external config function
 221     if req.curl_config_fn:
 222       req.curl_config_fn(curl)
 223
 224   def Done(self, errmsg):
 225     """Finishes a request.
 226
 227     @type errmsg: string or None
 228     @param errmsg: Error message if request failed
 229
 230     """
 231     req = self._req
 232     assert req, "No request"
 233
 234     logging.debug("Request %s finished, errmsg=%s", req, errmsg)
 235
 236     curl = self._curl
 237
 238     req.success = not bool(errmsg)
 239     req.error = errmsg
 240
 241     # Get HTTP response code
 242     req.resp_status_code = curl.getinfo(pycurl.RESPONSE_CODE)
 243     req.resp_body = self._resp_buffer.getvalue()
 244
 245     # Reset client object
 246     self._req = None
 247     self._resp_buffer = None
 248
 249     # Ensure no potentially large variables are referenced
 250     curl.setopt(pycurl.POSTFIELDS, "")
 251     curl.setopt(pycurl.WRITEFUNCTION, lambda _: None)
 252
 253
 254 class _PooledHttpClient:
 255   """Data structure for HTTP client pool.
 256
 257   """
 258   def __init__(self, identity, client):
 259     """Initializes this class.
 260
 261     @type identity: string
 262     @param identity: Client identifier for pool
 263     @type client: L{_HttpClient}
 264     @param client: HTTP client
 265
 266     """
 267     self.identity = identity
 268     self.client = client
 269     self.lastused = 0
 270
 271   def __repr__(self):
 272     status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),
 273               "id=%s" % self.identity,
 274               "lastuse=%s" % self.lastused,
 275               repr(self.client)]
 276
 277     return "<%s at %#x>" % (" ".join(status), id(self))
 278
 279
 280 class HttpClientPool:
 281   """A simple HTTP client pool.
 282
 283   Supports one pooled connection per identity (see
 284   L{HttpClientRequest.identity}).
 285
 286   """
 287   #: After how many generations to drop unused clients
 288   _MAX_GENERATIONS_DROP = 25
 289
 290   def __init__(self, curl_config_fn):
 291     """Initializes this class.
 292
 293     @type curl_config_fn: callable
 294     @param curl_config_fn: Function to configure cURL object after
 295                            initialization
 296
 297     """
 298     self._curl_config_fn = curl_config_fn
 299     self._generation = 0
 300     self._pool = {}
 301
 302     # Create custom logger for HTTP client pool. Change logging level to
 303     # C{logging.NOTSET} to get more details.
 304     self._logger = logging.getLogger(self.__class__.__name__)
 305     self._logger.setLevel(logging.INFO)
 306
 307   @staticmethod
 308   def _GetHttpClientCreator():
 309     """Returns callable to create HTTP client.
 310
 311     """
 312     return _HttpClient
 313
 314   def _Get(self, identity):
 315     """Gets an HTTP client from the pool.
 316
 317     @type identity: string
 318     @param identity: Client identifier
 319
 320     """
 321     try:
 322       pclient = self._pool.pop(identity)
 323     except KeyError:
 324       # Need to create new client
 325       client = self._GetHttpClientCreator()(self._curl_config_fn)
 326       pclient = _PooledHttpClient(identity, client)
 327       self._logger.debug("Created new client %s", pclient)
 328     else:
 329       self._logger.debug("Reusing client %s", pclient)
 330
 331     assert pclient.identity == identity
 332
 333     return pclient
 334
 335   def _StartRequest(self, req):
 336     """Starts a request.
 337
 338     @type req: L{HttpClientRequest}
 339     @param req: HTTP request
 340
 341     """
 342     pclient = self._Get(req.identity)
 343
 344     assert req.identity not in self._pool
 345
 346     pclient.client.StartRequest(req)
 347     pclient.lastused = self._generation
 348
 349     return pclient
 350
 351   def _Return(self, pclients):
 352     """Returns HTTP clients to the pool.
 353
 354     """
 355     for pc in pclients:
 356       self._logger.debug("Returning client %s to pool", pc)
 357       assert pc.identity not in self._pool
 358       assert pc not in self._pool.values()
 359       self._pool[pc.identity] = pc
 360
 361     # Check for unused clients
 362     for pc in self._pool.values():
 363       if (pc.lastused + self._MAX_GENERATIONS_DROP) < self._generation:
 364         self._logger.debug("Removing client %s which hasn't been used"
 365                            " for %s generations",
 366                            pc, self._MAX_GENERATIONS_DROP)
 367         self._pool.pop(pc.identity, None)
 368
 369     assert compat.all(pc.lastused >= (self._generation -
 370                                       self._MAX_GENERATIONS_DROP)
 371                       for pc in self._pool.values())
 372
 373   @staticmethod
 374   def _CreateCurlMultiHandle():
 375     """Creates new cURL multi handle.
 376
 377     """
 378     return pycurl.CurlMulti()
 379
 380   def ProcessRequests(self, requests):
 381     """Processes any number of HTTP client requests using pooled objects.
 382
 383     @type requests: list of L{HttpClientRequest}
 384     @param requests: List of all requests
 385
 386     """
 387     multi = self._CreateCurlMultiHandle()
 388
 389     # For client cleanup
 390     self._generation += 1
 391
 392     assert compat.all((req.error is None and
 393                        req.success is None and
 394                        req.resp_status_code is None and
 395                        req.resp_body is None)
 396                       for req in requests)
 397
 398     curl_to_pclient = {}
 399     for req in requests:
 400       pclient = self._StartRequest(req)
 401       curl = pclient.client.GetCurlHandle()
 402       curl_to_pclient[curl] = pclient
 403       multi.add_handle(curl)
 404       assert pclient.client.GetCurrentRequest() == req
 405       assert pclient.lastused >= 0
 406
 407     assert len(curl_to_pclient) == len(requests)
 408
 409     done_count = 0
 410     while True:
 411       (ret, _) = multi.perform()
 412       assert ret in (pycurl.E_MULTI_OK, pycurl.E_CALL_MULTI_PERFORM)
 413
 414       if ret == pycurl.E_CALL_MULTI_PERFORM:
 415         # cURL wants to be called again
 416         continue
 417
 418       while True:
 419         (remaining_messages, successful, failed) = multi.info_read()
 420
 421         for curl in successful:
 422           multi.remove_handle(curl)
 423           done_count += 1
 424           pclient = curl_to_pclient[curl]
 425           req = pclient.client.GetCurrentRequest()
 426           pclient.client.Done(None)
 427           assert req.success
 428           assert not pclient.client.GetCurrentRequest()
 429
 430         for curl, errnum, errmsg in failed:
 431           multi.remove_handle(curl)
 432           done_count += 1
 433           pclient = curl_to_pclient[curl]
 434           req = pclient.client.GetCurrentRequest()
 435           pclient.client.Done("Error %s: %s" % (errnum, errmsg))
 436           assert req.error
 437           assert not pclient.client.GetCurrentRequest()
 438
 439         if remaining_messages == 0:
 440           break
 441
 442       assert done_count <= len(requests)
 443
 444       if done_count == len(requests):
 445         break
 446
 447       # Wait for I/O. The I/O timeout shouldn't be too long so that HTTP
 448       # timeouts, which are only evaluated in multi.perform, aren't
 449       # unnecessarily delayed.
 450       multi.select(1.0)
 451
 452     assert compat.all(pclient.client.GetCurrentRequest() is None
 453                       for pclient in curl_to_pclient.values())
 454
 455     # Return clients to pool
 456     self._Return(curl_to_pclient.values())
 457
 458     assert done_count == len(requests)
 459     assert compat.all(req.error is not None or
 460                       (req.success and
 461                        req.resp_status_code is not None and
 462                        req.resp_body is not None)
 463                       for req in requests)