Merge lp:~cjwatson/launchpad/timeout-with-requests into lp:launchpad

[Colin Watson]

Convert lp.services.timeout to use requests rather than urllib2.

Convert lp.services.timeout to use requests rather than urllib2.

This is far more difficult than it ought to be: ideally none of this Cleanable* edifice would be needed in the first place, and requests/urllib3 makes us dig through several layers to get to the actual socket. There seems to be no reasonable way to implement this without reimplementing a couple of methods from requests, so this has to be checked when we upgrade to new versions.

[William Grant (wgrant)]

You disable trust_env, but AFAIK we rely on squid.internal for accessing Google and blog.launchpad.net.

review: Needs Fixing (code)

[William Grant (wgrant)]

review: Approve (code)

=== modified file 'lib/lp/app/browser/root.py'
--- lib/lp/app/browser/root.py	2015-04-29 00:36:59 +0000
+++ lib/lp/app/browser/root.py	2016-04-19 14:24:39 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2013 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 """Browser code for the Launchpad root page."""
 
@@ -14,6 +14,7 @@
 
 import feedparser
 from lazr.batchnavigator.z3batching import batch
+import requests
 from zope.component import getUtility
 from zope.formlib.interfaces import ConversionError
 from zope.interface import Interface
@@ -172,10 +173,10 @@
             return cached_data
         try:
             # Use urlfetch which supports timeout
-            data = urlfetch(config.launchpad.homepage_recent_posts_feed)
-        except IOError:
+            response = urlfetch(config.launchpad.homepage_recent_posts_feed)
+        except requests.RequestException:
             return []
-        feed = feedparser.parse(data)
+        feed = feedparser.parse(response.content)
         posts = []
         max_count = config.launchpad.homepage_recent_posts_count
         # FeedParser takes care of HTML sanitisation.

=== modified file 'lib/lp/services/googlesearch/__init__.py'
--- lib/lp/services/googlesearch/__init__.py	2015-07-08 16:05:11 +0000
+++ lib/lp/services/googlesearch/__init__.py	2016-04-19 14:24:39 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Interfaces for searching and working with results."""
@@ -12,7 +12,6 @@
     ]
 
 import urllib
-import urllib2
 from urlparse import (
     parse_qsl,
     urlunparse,
@@ -21,6 +20,7 @@
 
 from lazr.restful.utils import get_current_browser_request
 from lazr.uri import URI
+import requests
 from zope.interface import implementer
 
 from lp.services.config import config
@@ -32,7 +32,10 @@
     ISearchService,
     )
 from lp.services.timeline.requesttimeline import get_request_timeline
-from lp.services.timeout import TimeoutError
+from lp.services.timeout import (
+    TimeoutError,
+    urlfetch,
+    )
 from lp.services.webapp import urlparse
 
 
@@ -203,20 +206,19 @@
         :raise: `GoogleWrongGSPVersion` if the xml cannot be parsed.
         """
         search_url = self.create_search_url(terms, start=start)
-        from lp.services.timeout import urlfetch
         request = get_current_browser_request()
         timeline = get_request_timeline(request)
         action = timeline.start("google-search-api", search_url)
         try:
-            gsp_xml = urlfetch(search_url)
-        except (TimeoutError, urllib2.HTTPError, urllib2.URLError) as error:
+            response = urlfetch(search_url)
+        except (TimeoutError, requests.RequestException) as error:
             # Google search service errors are not code errors. Let the
             # call site choose to handle the unavailable service.
             raise GoogleResponseError(
                 "The response errored: %s" % str(error))
         finally:
             action.finish()
-        page_matches = self._parse_google_search_protocol(gsp_xml)
+        page_matches = self._parse_google_search_protocol(response.content)
         return page_matches
 
     def _checkParameter(self, name, value, is_int=False):

=== modified file 'lib/lp/services/googlesearch/doc/google-searchservice.txt'
--- lib/lp/services/googlesearch/doc/google-searchservice.txt	2011-12-30 08:13:14 +0000
+++ lib/lp/services/googlesearch/doc/google-searchservice.txt	2016-04-19 14:24:39 +0000
@@ -6,6 +6,12 @@
 (cs-be) client. Given one or more terms, it will retrieve an XML
 summary of the matching launchpad.net pages.
 
+We silence logging of new HTTP connections from requests throughout.
+
+    >>> from fixtures import FakeLogger
+    >>> logger = FakeLogger()
+    >>> logger.setUp()
+
 
 GoogleSearchService
 ===================
@@ -616,3 +622,5 @@
     # Restore the configuration and the timeout state.
     >>> timeout_data = config.pop('timeout_data')
     >>> set_default_timeout_function(old_timeout_function)
+
+    >>> logger.cleanUp()

=== modified file 'lib/lp/services/googlesearch/tests/test_google.py'
--- lib/lp/services/googlesearch/tests/test_google.py	2011-12-30 08:13:14 +0000
+++ lib/lp/services/googlesearch/tests/test_google.py	2016-04-19 14:24:39 +0000
@@ -1,4 +1,4 @@
-# Copyright 2011 Canonical Ltd.  This software is licensed under the
+# Copyright 2011-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Test the google search service."""
@@ -6,16 +6,17 @@
 __metaclass__ = type
 
 from contextlib import contextmanager
-from urllib2 import (
+
+from requests.exceptions import (
+    ConnectionError,
     HTTPError,
-    URLError,
     )
 
 from lp.services.googlesearch import GoogleSearchService
 from lp.services.googlesearch.interfaces import GoogleResponseError
 from lp.services.timeout import TimeoutError
 from lp.testing import TestCase
-from lp.testing.layers import FunctionalLayer
+from lp.testing.layers import LaunchpadFunctionalLayer
 
 
 @contextmanager
@@ -41,7 +42,7 @@
 class TestGoogleSearchService(TestCase):
     """Test GoogleSearchService."""
 
-    layer = FunctionalLayer
+    layer = LaunchpadFunctionalLayer
 
     def setUp(self):
         super(TestGoogleSearchService, self).setUp()
@@ -54,9 +55,9 @@
             self.assertRaises(
                 GoogleResponseError, self.search_service.search, 'fnord')
 
-    def test_search_converts_URLError(self):
-        # The method converts URLError to GoogleResponseError.
-        with urlfetch_exception(URLError, 'oops'):
+    def test_search_converts_ConnectionError(self):
+        # The method converts ConnectionError to GoogleResponseError.
+        with urlfetch_exception(ConnectionError, 'oops'):
             self.assertRaises(
                 GoogleResponseError, self.search_service.search, 'fnord')
 

=== modified file 'lib/lp/services/gpg/handler.py'
--- lib/lp/services/gpg/handler.py	2016-03-30 03:14:02 +0000
+++ lib/lp/services/gpg/handler.py	2016-04-19 14:24:39 +0000
@@ -20,11 +20,11 @@
 import sys
 import tempfile
 import urllib
-import urllib2
 
 import gpgme
 from gpgservice_client import GPGClient
 from lazr.restful.utils import get_current_browser_request
+import requests
 from zope.interface import implementer
 from zope.security.proxy import removeSecurityProxy
 
@@ -490,21 +490,20 @@
         # minutes." The details of the error do not matter for users
         # (and for the code in callsites), but we should be able to see
         # if this problem occurs too often.
-        except urllib2.HTTPError as exc:
+        except requests.HTTPError as exc:
             # Old versions of SKS return a 500 error when queried for a
             # non-existent key. Production was upgraded in 2013/01, but
             # let's leave this here for a while.
             #
             # We can extract the fact that the key is unknown by looking
             # into the response's content.
-            if exc.code in (404, 500) and exc.fp is not None:
-                content = exc.fp.read()
+            if exc.response.status_code in (404, 500):
                 no_key_message = 'No results found: No keys found'
-                if content.find(no_key_message) >= 0:
+                if exc.response.content.find(no_key_message) >= 0:
                     raise GPGKeyDoesNotExistOnServer(fingerprint)
                 errorlog.globalErrorUtility.raising(sys.exc_info(), request)
                 raise GPGKeyTemporarilyNotFoundError(fingerprint)
-        except (TimeoutError, urllib2.URLError) as exc:
+        except (TimeoutError, requests.RequestException) as exc:
             errorlog.globalErrorUtility.raising(sys.exc_info(), request)
             raise GPGKeyTemporarilyNotFoundError(fingerprint)
         finally:
@@ -513,7 +512,7 @@
     def _grabPage(self, action, fingerprint):
         """Wrapper to collect KeyServer Pages."""
         url = self.getURLForKeyInServer(fingerprint, action)
-        return urlfetch(url)
+        return urlfetch(url).content
 
 
 @implementer(IPymeSignature)

=== modified file 'lib/lp/services/tests/test_timeout.py'
--- lib/lp/services/tests/test_timeout.py	2015-10-14 15:22:01 +0000
+++ lib/lp/services/tests/test_timeout.py	2016-04-19 14:24:39 +0000
@@ -1,4 +1,4 @@
-# Copyright 2012 Canonical Ltd.  This software is licensed under the
+# Copyright 2012-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """timeout.py tests.
@@ -13,9 +13,14 @@
 import socket
 from textwrap import dedent
 import threading
-import urllib2
 import xmlrpclib
 
+from requests.exceptions import (
+    ConnectionError,
+    InvalidSchema,
+    )
+from testtools.matchers import MatchesStructure
+
 from lp.services.timeout import (
     get_default_timeout_function,
     set_default_timeout_function,
@@ -197,10 +202,10 @@
         self.assertEqual([True], using_default)
 
     def make_test_socket(self):
-        """One common use case for timing out is when making an HTTP request to
-        an external site to fetch content. To this end, the timeout module has
-        a urlfetch() function that retrieve a URL using custom urllib2 handlers
-        that will timeout using the default timeout function and clean-up the
+        """One common use case for timing out is when making an HTTP request
+        to an external site to fetch content. To this end, the timeout
+        module has a urlfetch() function that retrieves a URL in such a way
+        as to timeout using the default timeout function and clean-up the
         socket properly.
         """
         sock = socket.socket()
@@ -213,11 +218,11 @@
         http_server_url = 'http://%s:%d/' % sock.getsockname()
         return sock, http_server_url
 
-    def test_urlfetch_raises_urllib2_exceptions(self):
-        """Normal urllib2 exceptions are raised."""
+    def test_urlfetch_raises_requests_exceptions(self):
+        """Normal requests exceptions are raised."""
         sock, http_server_url = self.make_test_socket()
 
-        e = self.assertRaises(urllib2.URLError, urlfetch, http_server_url)
+        e = self.assertRaises(ConnectionError, urlfetch, http_server_url)
         self.assertIn('Connection refused', str(e))
 
     def test_urlfetch_timeout_after_listen(self):
@@ -290,15 +295,19 @@
 
         t = threading.Thread(target=success_result)
         t.start()
-        self.assertEqual('Success.', urlfetch(http_server_url))
+        self.assertThat(
+            urlfetch(http_server_url),
+            MatchesStructure.byEquality(status_code=200, content='Success.'))
         t.join()
 
-    def test_urlfetch_only_supports_http_urls(self):
-        """urlfetch() only supports http urls:"""
+    def test_urlfetch_does_not_support_ftp_urls(self):
+        """urlfetch() does not support ftp urls."""
         set_default_timeout_function(lambda: 1)
         self.addCleanup(set_default_timeout_function, None)
-        e = self.assertRaises(AssertionError, urlfetch, 'ftp://localhost')
-        self.assertEqual('only http is supported.', str(e))
+        url = 'ftp://localhost/'
+        e = self.assertRaises(InvalidSchema, urlfetch, url)
+        self.assertEqual(
+            "No connection adapters were found for '%s'" % url, str(e))
 
     def test_xmlrpc_transport(self):
         """ Another use case for timeouts is communicating with external

=== modified file 'lib/lp/services/timeout.py'
--- lib/lp/services/timeout.py	2014-08-29 01:41:14 +0000
+++ lib/lp/services/timeout.py	2016-04-19 14:24:39 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2016 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Helpers to time out external operations."""
@@ -14,16 +14,29 @@
     "with_timeout",
     ]
 
-import httplib
 import socket
 import sys
-from threading import Thread
-import urllib2
+from threading import (
+    Lock,
+    Thread,
+    )
 from xmlrpclib import (
     SafeTransport,
     Transport,
     )
 
+from requests import Session
+from requests.adapters import (
+    DEFAULT_POOLBLOCK,
+    HTTPAdapter,
+    )
+from requests.packages.urllib3.connectionpool import (
+    HTTPConnectionPool,
+    HTTPSConnectionPool,
+    )
+from requests.packages.urllib3.exceptions import ClosedPoolError
+from requests.packages.urllib3.poolmanager import PoolManager
+
 
 default_timeout_function = None
 
@@ -145,47 +158,113 @@
         return call_with_timeout
 
 
-class CleanableHTTPHandler(urllib2.HTTPHandler):
-    """Subclass of `urllib2.HTTPHandler` that can be cleaned-up."""
-
-    def http_open(self, req):
-        """See `urllib2.HTTPHandler`."""
-        def connection_factory(*args, **kwargs):
-            """Save the created connection so that we can clean it up."""
-            self.__conn = httplib.HTTPConnection(*args, **kwargs)
-            return self.__conn
-        return self.do_open(connection_factory, req)
-
-    def reset_connection(self):
-        """Reset the underlying HTTP connection."""
-        try:
-            self.__conn.sock.shutdown(socket.SHUT_RDWR)
-        except AttributeError:
-            # It's possible that the other thread closed the socket
-            # beforehand.
-            pass
-        self.__conn.close()
+class CleanableConnectionPoolMixin:
+    """Enhance urllib3's connection pools to support forced socket cleanup."""
+
+    def __init__(self, *args, **kwargs):
+        super(CleanableConnectionPoolMixin, self).__init__(*args, **kwargs)
+        self._all_connections = []
+        self._all_connections_mutex = Lock()
+
+    def _new_conn(self):
+        with self._all_connections_mutex:
+            if self._all_connections is None:
+                raise ClosedPoolError(self, "Pool is closed.")
+            conn = super(CleanableConnectionPoolMixin, self)._new_conn()
+            self._all_connections.append(conn)
+            return conn
+
+    def close(self):
+        with self._all_connections_mutex:
+            if self._all_connections is None:
+                return
+            for conn in self._all_connections:
+                sock = getattr(conn, "sock", None)
+                if sock is not None:
+                    sock.shutdown(socket.SHUT_RDWR)
+                    sock.close()
+                    conn.sock = None
+            self._all_connections = None
+        super(CleanableConnectionPoolMixin, self).close()
+
+
+class CleanableHTTPConnectionPool(
+    CleanableConnectionPoolMixin, HTTPConnectionPool):
+    pass
+
+
+class CleanableHTTPSConnectionPool(
+    CleanableConnectionPoolMixin, HTTPSConnectionPool):
+    pass
+
+
+cleanable_pool_classes_by_scheme = {
+    "http": CleanableHTTPConnectionPool,
+    "https": CleanableHTTPSConnectionPool,
+    }
+
+
+class CleanablePoolManager(PoolManager):
+    """A version of urllib3's PoolManager supporting forced socket cleanup."""
+
+    # XXX cjwatson 2015-03-11: Reimplements PoolManager._new_pool; check
+    # this when upgrading requests.
+    def _new_pool(self, scheme, host, port):
+        if scheme not in cleanable_pool_classes_by_scheme:
+            raise ValueError("Unhandled scheme: %s" % scheme)
+        pool_cls = cleanable_pool_classes_by_scheme[scheme]
+        kwargs = self.connection_pool_kw
+        if scheme == 'http':
+            kwargs = self.connection_pool_kw.copy()
+            for kw in ('key_file', 'cert_file', 'cert_reqs', 'ca_certs',
+                       'ssl_version'):
+                kwargs.pop(kw, None)
+
+        return pool_cls(host, port, **kwargs)
+
+
+class CleanableHTTPAdapter(HTTPAdapter):
+    """Enhance HTTPAdapter to use CleanablePoolManager."""
+
+    # XXX cjwatson 2015-03-11: Reimplements HTTPAdapter.init_poolmanager;
+    # check this when upgrading requests.
+    def init_poolmanager(self, connections, maxsize, block=DEFAULT_POOLBLOCK,
+                         **pool_kwargs):
+        # save these values for pickling
+        self._pool_connections = connections
+        self._pool_maxsize = maxsize
+        self._pool_block = block
+
+        self.poolmanager = CleanablePoolManager(
+            num_pools=connections, maxsize=maxsize, block=block, strict=True,
+            **pool_kwargs)
 
 
 class URLFetcher:
     """Object fetching remote URLs with a time out."""
 
     @with_timeout(cleanup='cleanup')
-    def fetch(self, url, data=None):
+    def fetch(self, url, **request_kwargs):
         """Fetch the URL using a custom HTTP handler supporting timeout."""
-        assert url.startswith('http://'), "only http is supported."
-        self.handler = CleanableHTTPHandler()
-        opener = urllib2.build_opener(self.handler)
-        return opener.open(url, data).read()
+        request_kwargs.setdefault("method", "GET")
+        self.session = Session()
+        # Mount our custom adapters.
+        self.session.mount("https://", CleanableHTTPAdapter())
+        self.session.mount("http://", CleanableHTTPAdapter())
+        response = self.session.request(url=url, **request_kwargs)
+        response.raise_for_status()
+        # Make sure the content has been consumed before returning.
+        response.content
+        return response
 
     def cleanup(self):
         """Reset the connection when the operation timed out."""
-        self.handler.reset_connection()
-
-
-def urlfetch(url, data=None):
-    """Wrapper for `urllib2.urlopen()` that times out."""
-    return URLFetcher().fetch(url, data)
+        self.session.close()
+
+
+def urlfetch(url, **request_kwargs):
+    """Wrapper for `requests.get()` that times out."""
+    return URLFetcher().fetch(url, **request_kwargs)
 
 
 class TransportWithTimeout(Transport):