Synnefo » snf-ganeti » ganeti-local

#

#

# Copyright (C) 2007, 2008, 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

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""HTTP client module.

"""

import logging

import pycurl

import threading

from cStringIO import StringIO

from ganeti import http

from ganeti import compat

from ganeti import netutils

from ganeti import locking

class HttpClientRequest(object):

  def __init__(self, host, port, method, path, headers=None, post_data=None,

               read_timeout=None, curl_config_fn=None):

    """Describes an HTTP request.

    @type host: string

    @param host: Hostname

    @type port: int

    @param port: Port

    @type method: string

    @param method: Method name

    @type path: string

    @param path: Request path

    @type headers: list or None

    @param headers: Additional headers to send, list of strings

    @type post_data: string or None

    @param post_data: Additional data to send

    @type read_timeout: int

    @param read_timeout: if passed, it will be used as the read

        timeout while reading the response from the server

    @type curl_config_fn: callable

    @param curl_config_fn: Function to configure cURL object before request

                           (Note: if the function configures the connection in

                           a way where it wouldn't be efficient to reuse them,

                           a "identity" property should be defined, see

                           L{HttpClientRequest.identity})

    """

    assert path.startswith("/"), "Path must start with slash (/)"

    assert curl_config_fn is None or callable(curl_config_fn)

    # Request attributes

    self.host = host

    self.port = port

    self.method = method

    self.path = path

    self.read_timeout = read_timeout

    self.curl_config_fn = curl_config_fn

    if post_data is None:

      self.post_data = ""

    else:

      self.post_data = post_data

    if headers is None:

      self.headers = []

    elif isinstance(headers, dict):

      # Support for old interface

      self.headers = ["%s: %s" % (name, value)

                      for name, value in headers.items()]

    else:

      self.headers = headers

    # Response status

    self.success = None

    self.error = None

    # Response attributes

    self.resp_status_code = None

    self.resp_body = None

  def __repr__(self):

    status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),

              "%s:%s" % (self.host, self.port),

              self.method,

              self.path]

    return "<%s at %#x>" % (" ".join(status), id(self))

  @property

  def url(self):

    """Returns the full URL for this requests.

    """

    if netutils.IPAddress.IsValid(self.host):

      address = netutils.FormatAddress((self.host, self.port))

    else:

      address = "%s:%s" % (self.host, self.port)

    # TODO: Support for non-SSL requests

    return "https://%s%s" % (address, self.path)

  @property

  def identity(self):

    """Returns identifier for retrieving a pooled connection for this request.

    This allows cURL client objects to be re-used and to cache information

    (e.g. SSL session IDs or connections).

    """

    parts = [self.host, self.port]

    if self.curl_config_fn:

      try:

        parts.append(self.curl_config_fn.identity)

      except AttributeError:

        pass

    return "/".join(str(i) for i in parts)

class _HttpClient(object):

  def __init__(self, curl_config_fn):

    """Initializes this class.

    @type curl_config_fn: callable

    @param curl_config_fn: Function to configure cURL object after

                           initialization

    """

    self._req = None

    curl = self._CreateCurlHandle()

    curl.setopt(pycurl.VERBOSE, False)

    curl.setopt(pycurl.NOSIGNAL, True)

    curl.setopt(pycurl.USERAGENT, http.HTTP_GANETI_VERSION)

    curl.setopt(pycurl.PROXY, "")

    # Disable SSL session ID caching (pycurl >= 7.16.0)

    if hasattr(pycurl, "SSL_SESSIONID_CACHE"):

      curl.setopt(pycurl.SSL_SESSIONID_CACHE, False)

    # Pass cURL object to external config function

    if curl_config_fn:

      curl_config_fn(curl)

    self._curl = curl

  @staticmethod

  def _CreateCurlHandle():

    """Returns a new cURL object.

    """

    return pycurl.Curl()

  def GetCurlHandle(self):

    """Returns the cURL object.

    """

    return self._curl

  def GetCurrentRequest(self):

    """Returns the current request.

    @rtype: L{HttpClientRequest} or None

    """

    return self._req

  def StartRequest(self, req):

    """Starts a request on this client.

    @type req: L{HttpClientRequest}

    @param req: HTTP request

    """

    assert not self._req, "Another request is already started"

    logging.debug("Starting request %r", req)

    self._req = req

    self._resp_buffer = StringIO()

    url = req.url

    method = req.method

    post_data = req.post_data

    headers = req.headers

    # PycURL requires strings to be non-unicode

    assert isinstance(method, str)

    assert isinstance(url, str)

    assert isinstance(post_data, str)

    assert compat.all(isinstance(i, str) for i in headers)

    # Configure cURL object for request

    curl = self._curl

    curl.setopt(pycurl.CUSTOMREQUEST, str(method))

    curl.setopt(pycurl.URL, url)

    curl.setopt(pycurl.POSTFIELDS, post_data)

    curl.setopt(pycurl.WRITEFUNCTION, self._resp_buffer.write)

    curl.setopt(pycurl.HTTPHEADER, headers)

    if req.read_timeout is None:

      curl.setopt(pycurl.TIMEOUT, 0)

    else:

      curl.setopt(pycurl.TIMEOUT, int(req.read_timeout))

    # Pass cURL object to external config function

    if req.curl_config_fn:

      req.curl_config_fn(curl)

  def Done(self, errmsg):

    """Finishes a request.

    @type errmsg: string or None

    @param errmsg: Error message if request failed

    """

    req = self._req

    assert req, "No request"

    logging.debug("Request %s finished, errmsg=%s", req, errmsg)

    curl = self._curl

    req.success = not bool(errmsg)

    req.error = errmsg

    # Get HTTP response code

    req.resp_status_code = curl.getinfo(pycurl.RESPONSE_CODE)

    req.resp_body = self._resp_buffer.getvalue()

    # Reset client object

    self._req = None

    self._resp_buffer = None

    # Ensure no potentially large variables are referenced

    curl.setopt(pycurl.POSTFIELDS, "")

    curl.setopt(pycurl.WRITEFUNCTION, lambda _: None)

class _PooledHttpClient:

  """Data structure for HTTP client pool.

  """

  def __init__(self, identity, client):

    """Initializes this class.

    @type identity: string

    @param identity: Client identifier for pool

    @type client: L{_HttpClient}

    @param client: HTTP client

    """

    self.identity = identity

    self.client = client

    self.lastused = 0

  def __repr__(self):

    status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),

              "id=%s" % self.identity,

              "lastuse=%s" % self.lastused,

              repr(self.client)]

    return "<%s at %#x>" % (" ".join(status), id(self))

class HttpClientPool:

  """A simple HTTP client pool.

  Supports one pooled connection per identity (see

  L{HttpClientRequest.identity}).

  """

  #: After how many generations to drop unused clients

  _MAX_GENERATIONS_DROP = 25

  def __init__(self, curl_config_fn):

    """Initializes this class.

    @type curl_config_fn: callable

    @param curl_config_fn: Function to configure cURL object after

                           initialization

    """

    self._curl_config_fn = curl_config_fn

    self._generation = 0

    self._pool = {}

    # Create custom logger for HTTP client pool. Change logging level to

    # C{logging.NOTSET} to get more details.

    self._logger = logging.getLogger(self.__class__.__name__)

    self._logger.setLevel(logging.INFO)

  @staticmethod

  def _GetHttpClientCreator():

    """Returns callable to create HTTP client.

    """

    return _HttpClient

  def _Get(self, identity):

    """Gets an HTTP client from the pool.

    @type identity: string

    @param identity: Client identifier

    """

    try:

      pclient = self._pool.pop(identity)

    except KeyError:

      # Need to create new client

      client = self._GetHttpClientCreator()(self._curl_config_fn)

      pclient = _PooledHttpClient(identity, client)

      self._logger.debug("Created new client %s", pclient)

    else:

      self._logger.debug("Reusing client %s", pclient)

    assert pclient.identity == identity

    return pclient

  def _StartRequest(self, req):

    """Starts a request.

    @type req: L{HttpClientRequest}

    @param req: HTTP request

    """

    pclient = self._Get(req.identity)

    assert req.identity not in self._pool

    pclient.client.StartRequest(req)

    pclient.lastused = self._generation

    return pclient

  def _Return(self, pclients):

    """Returns HTTP clients to the pool.

    """

    assert not frozenset(pclients) & frozenset(self._pool.values())

    for pc in pclients:

      self._logger.debug("Returning client %s to pool", pc)

      assert pc.identity not in self._pool

      self._pool[pc.identity] = pc

    # Check for unused clients

    for pc in self._pool.values():

      if (pc.lastused + self._MAX_GENERATIONS_DROP) < self._generation:

        self._logger.debug("Removing client %s which hasn't been used"

                           " for %s generations",

                           pc, self._MAX_GENERATIONS_DROP)

        self._pool.pop(pc.identity, None)

    assert compat.all(pc.lastused >= (self._generation -

                                      self._MAX_GENERATIONS_DROP)

                      for pc in self._pool.values())

  @staticmethod

  def _CreateCurlMultiHandle():

    """Creates new cURL multi handle.

    """

    return pycurl.CurlMulti()

  def ProcessRequests(self, requests, lock_monitor_cb=None):

    """Processes any number of HTTP client requests using pooled objects.

    @type requests: list of L{HttpClientRequest}

    @param requests: List of all requests

    @param lock_monitor_cb: Callable for registering with lock monitor

    """

    # For client cleanup

    self._generation += 1

    assert compat.all((req.error is None and

                       req.success is None and

                       req.resp_status_code is None and

                       req.resp_body is None)

                      for req in requests)

    curl_to_pclient = {}

    for req in requests:

      pclient = self._StartRequest(req)

      curl_to_pclient[pclient.client.GetCurlHandle()] = pclient

      assert pclient.client.GetCurrentRequest() == req

      assert pclient.lastused >= 0

    assert len(curl_to_pclient) == len(requests)

    if lock_monitor_cb:

      monitor = _PendingRequestMonitor(threading.currentThread(),

                                       curl_to_pclient.values)

      lock_monitor_cb(monitor)

    else:

      monitor = _NoOpRequestMonitor

    # Process all requests and act based on the returned values

    for (curl, msg) in _ProcessCurlRequests(self._CreateCurlMultiHandle(),

                                            curl_to_pclient.keys()):

      pclient = curl_to_pclient[curl]

      req = pclient.client.GetCurrentRequest()

      monitor.acquire(shared=0)

      try:

        pclient.client.Done(msg)

      finally:

        monitor.release()

      assert ((msg is None and req.success and req.error is None) ^

              (msg is not None and not req.success and req.error == msg))

    assert compat.all(pclient.client.GetCurrentRequest() is None

                      for pclient in curl_to_pclient.values())

    monitor.acquire(shared=0)

    try:

      # Don't try to read information from returned clients

      monitor.Disable()

      # Return clients to pool

      self._Return(curl_to_pclient.values())

    finally:

      monitor.release()

    assert compat.all(req.error is not None or

                      (req.success and

                       req.resp_status_code is not None and

                       req.resp_body is not None)

                      for req in requests)

class _NoOpRequestMonitor: # pylint: disable=W0232

  """No-op request monitor.

  """

  @staticmethod

  def acquire(*args, **kwargs):

    pass

  release = acquire

  Disable = acquire

class _PendingRequestMonitor:

  _LOCK = "_lock"

  def __init__(self, owner, pending_fn):

    """Initializes this class.

    """

    self._owner = owner

    self._pending_fn = pending_fn

    # The lock monitor runs in another thread, hence locking is necessary

    self._lock = locking.SharedLock("PendingHttpRequests")

    self.acquire = self._lock.acquire

    self.release = self._lock.release

  def Disable(self):

    """Disable monitor.

    """

    self._pending_fn = None

  @locking.ssynchronized(_LOCK, shared=1)

  def GetLockInfo(self, requested): # pylint: disable=W0613

    """Retrieves information about pending requests.

    @type requested: set

    @param requested: Requested information, see C{query.LQ_*}

    """

    # No need to sort here, that's being done by the lock manager and query

    # library. There are no priorities for requests, hence all show up as

    # one item under "pending".

    result = []

    if self._pending_fn:

      owner_name = self._owner.getName()

      for pclient in self._pending_fn():

        req = pclient.client.GetCurrentRequest()

        if req:

          result.append(("rpc/%s%s" % (req.host, req.path), None, None,

                         [("thread", [owner_name])]))

    return result

def _ProcessCurlRequests(multi, requests):

  """cURL request processor.

  This generator yields a tuple once for every completed request, successful or

  not. The first value in the tuple is the handle, the second an error message

  or C{None} for successful requests.

  @type multi: C{pycurl.CurlMulti}

  @param multi: cURL multi object

  @type requests: sequence

  @param requests: cURL request handles

  """

  for curl in requests:

    multi.add_handle(curl)

  while True:

    (ret, active) = multi.perform()

    assert ret in (pycurl.E_MULTI_OK, pycurl.E_CALL_MULTI_PERFORM)

    if ret == pycurl.E_CALL_MULTI_PERFORM:

      # cURL wants to be called again

      continue

    while True:

      (remaining_messages, successful, failed) = multi.info_read()

      for curl in successful:

        multi.remove_handle(curl)

        yield (curl, None)

      for curl, errnum, errmsg in failed:

        multi.remove_handle(curl)

        yield (curl, "Error %s: %s" % (errnum, errmsg))

      if remaining_messages == 0:

        break

    if active == 0:

      # No active handles anymore

      break

    # Wait for I/O. The I/O timeout shouldn't be too long so that HTTP

    # timeouts, which are only evaluated in multi.perform, aren't

    # unnecessarily delayed.

    multi.select(1.0)