http-backend: catch errors from HTTP calls that are not raising exceptions, but

return non 200 exception codes.

rhodecode/lib/vcs/__init__.py

@@ -1,241 +1,248 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2014-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Various version Control System version lib (vcs) management abstraction layer
 for Python. Build with server client architecture.
 """
 
 
 VERSION = (0, 5, 0, 'dev')
 
 __version__ = '.'.join((str(each) for each in VERSION[:4]))
 
 __all__ = [
     'get_version', 'get_vcs_instance', 'get_backend',
     'VCSError', 'RepositoryError', 'CommitError'
     ]
 
 import atexit
 import logging
 import subprocess32
 import time
 import urlparse
 from cStringIO import StringIO
 
 
 from rhodecode.lib.vcs.conf import settings
 from rhodecode.lib.vcs.backends import get_vcs_instance, get_backend
 from rhodecode.lib.vcs.exceptions import (
     VCSError, RepositoryError, CommitError, VCSCommunicationError)
 
 log = logging.getLogger(__name__)
 
 # The pycurl library directly accesses C API functions and is not patched by
 # gevent. This will potentially lead to deadlocks due to incompatibility to
 # gevent. Therefore we check if gevent is active and import a gevent compatible
 # wrapper in that case.
 try:
     from gevent import monkey
     if monkey.is_module_patched('__builtin__'):
         import geventcurl as pycurl
         log.debug('Using gevent comapatible pycurl: %s', pycurl)
     else:
         import pycurl
 except ImportError:
     import pycurl
 
 
 def get_version():
     """
     Returns shorter version (digit parts only) as string.
     """
     return '.'.join((str(each) for each in VERSION[:3]))
 
 
 def connect_http(server_and_port):
     from rhodecode.lib.vcs import connection, client_http
     from rhodecode.lib.middleware.utils import scm_app
 
     session_factory = client_http.ThreadlocalSessionFactory()
 
     connection.Git = client_http.RepoMaker(
         server_and_port, '/git', 'git', session_factory)
     connection.Hg = client_http.RepoMaker(
         server_and_port, '/hg', 'hg', session_factory)
     connection.Svn = client_http.RepoMaker(
         server_and_port, '/svn', 'svn', session_factory)
     connection.Service = client_http.ServiceConnection(
         server_and_port, '/_service', session_factory)
 
     scm_app.HG_REMOTE_WSGI = client_http.VcsHttpProxy(
         server_and_port, '/proxy/hg')
     scm_app.GIT_REMOTE_WSGI = client_http.VcsHttpProxy(
         server_and_port, '/proxy/git')
 
     @atexit.register
     def free_connection_resources():
         connection.Git = None
         connection.Hg = None
         connection.Svn = None
         connection.Service = None
 
 
 def connect_vcs(server_and_port, protocol):
     """
     Initializes the connection to the vcs server.
 
     :param server_and_port: str, e.g. "localhost:9900"
     :param protocol: str or "http"
     """
     if protocol == 'http':
         connect_http(server_and_port)
     else:
         raise Exception('Invalid vcs server protocol "{}"'.format(protocol))
 
 
 # TODO: johbo: This function should be moved into our test suite, there is
 # no reason to support starting the vcsserver in Enterprise itself.
 def start_vcs_server(server_and_port, protocol, log_level=None):
     """
     Starts the vcs server in a subprocess.
     """
     log.info('Starting VCSServer as a sub process with %s protocol', protocol)
     if protocol == 'http':
         return _start_http_vcs_server(server_and_port, log_level)
     else:
         raise Exception('Invalid vcs server protocol "{}"'.format(protocol))
 
 
 def _start_http_vcs_server(server_and_port, log_level=None):
     # TODO: mikhail: shutdown if an http server already runs
 
     host, port = server_and_port.rsplit(":", 1)
     args = [
         'pserve', 'rhodecode/tests/vcsserver_http.ini',
         'http_port=%s' % (port, ), 'http_host=%s' % (host, )]
     proc = subprocess32.Popen(args)
 
     def cleanup_server_process():
         proc.kill()
     atexit.register(cleanup_server_process)
 
     server = create_vcsserver_proxy(server_and_port, protocol='http')
     _wait_until_vcs_server_is_reachable(server)
 
 
 def _wait_until_vcs_server_is_reachable(server, timeout=40):
     begin = time.time()
     while (time.time() - begin) < timeout:
         try:
             server.ping()
             return
         except (VCSCommunicationError, pycurl.error):
             log.debug('VCSServer not started yet, retry to connect.')
         time.sleep(0.5)
     raise Exception(
         'Starting the VCSServer failed or took more than {} '
         'seconds.'.format(timeout))
 
 
 def _try_to_shutdown_running_server(server_and_port, protocol):
     server = create_vcsserver_proxy(server_and_port, protocol)
     try:
         server.shutdown()
     except pycurl.error:
         return
 
     # TODO: Not sure why this is important, but without it the following start
     # of the server fails.
     server = create_vcsserver_proxy(server_and_port, protocol)
     server.ping()
 
 
 def create_vcsserver_proxy(server_and_port, protocol):
     if protocol == 'http':
         return _create_vcsserver_proxy_http(server_and_port)
     else:
         raise Exception('Invalid vcs server protocol "{}"'.format(protocol))
 
 
 def _create_vcsserver_proxy_http(server_and_port):
     from rhodecode.lib.vcs import client_http
 
     session = _create_http_rpc_session()
     url = urlparse.urljoin('http://%s' % server_and_port, '/server')
     return client_http.RemoteObject(url, session)
 
 
 class CurlSession(object):
     """
     Modeled so that it provides a subset of the requests interface.
 
     This has been created so that it does only provide a minimal API for our
     needs. The parts which it provides are based on the API of the library
     `requests` which allows us to easily benchmark against it.
 
     Please have a look at the class :class:`requests.Session` when you extend
     it.
     """
 
     def __init__(self):
         curl = pycurl.Curl()
         # TODO: johbo: I did test with 7.19 of libcurl. This version has
         # trouble with 100 - continue being set in the expect header. This
         # can lead to massive performance drops, switching it off here.
         curl.setopt(curl.HTTPHEADER, ["Expect:"])
         curl.setopt(curl.TCP_NODELAY, True)
         curl.setopt(curl.PROTOCOLS, curl.PROTO_HTTP)
         self._curl = curl
 
     def post(self, url, data, allow_redirects=False):
         response_buffer = StringIO()
 
         curl = self._curl
         curl.setopt(curl.URL, url)
         curl.setopt(curl.POST, True)
         curl.setopt(curl.POSTFIELDS, data)
         curl.setopt(curl.FOLLOWLOCATION, allow_redirects)
         curl.setopt(curl.WRITEDATA, response_buffer)
         curl.perform()
 
-        return CurlResponse(response_buffer)
+        status_code = curl.getinfo(pycurl.HTTP_CODE)
+
+        return CurlResponse(response_buffer, status_code)
 
 
 class CurlResponse(object):
     """
     The response of a request, modeled after the requests API.
 
     This class provides a subset of the response interface known from the
     library `requests`. It is intentionally kept similar, so that we can use
     `requests` as a drop in replacement for benchmarking purposes.
     """
 
-    def __init__(self, response_buffer):
+    def __init__(self, response_buffer, status_code):
         self._response_buffer = response_buffer
+        self._status_code = status_code
 
     @property
     def content(self):
         return self._response_buffer.getvalue()
 
+    @property
+    def status_code(self):
+        return self._status_code
+
 
 def _create_http_rpc_session():
     session = CurlSession()
     return session

rhodecode/lib/vcs/client_http.py

@@ -1,284 +1,289 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Client for the VCSServer implemented based on HTTP.
 """
 
 import copy
 import logging
 import threading
 import urllib2
 import urlparse
 import uuid
 
 import pycurl
 import msgpack
 import requests
 
 from . import exceptions, CurlSession
 
 
 log = logging.getLogger(__name__)
 
 
 # TODO: mikhail: Keep it in sync with vcsserver's
 # HTTPApplication.ALLOWED_EXCEPTIONS
 EXCEPTIONS_MAP = {
     'KeyError': KeyError,
     'URLError': urllib2.URLError,
 }
 
 
 class RepoMaker(object):
 
     def __init__(self, server_and_port, backend_endpoint, backend_type, session_factory):
         self.url = urlparse.urljoin(
             'http://%s' % server_and_port, backend_endpoint)
         self._session_factory = session_factory
         self.backend_type = backend_type
 
     def __call__(self, path, config, with_wire=None):
         log.debug('RepoMaker call on %s', path)
         return RemoteRepo(
             path, config, self.url, self._session_factory(),
             with_wire=with_wire)
 
     def __getattr__(self, name):
         def f(*args, **kwargs):
             return self._call(name, *args, **kwargs)
         return f
 
     @exceptions.map_vcs_exceptions
     def _call(self, name, *args, **kwargs):
         payload = {
             'id': str(uuid.uuid4()),
             'method': name,
             'backend': self.backend_type,
             'params': {'args': args, 'kwargs': kwargs}
         }
         return _remote_call(
             self.url, payload, EXCEPTIONS_MAP, self._session_factory())
 
 
 class ServiceConnection(object):
     def __init__(self, server_and_port, backend_endpoint, session_factory):
         self.url = urlparse.urljoin(
             'http://%s' % server_and_port, backend_endpoint)
         self._session_factory = session_factory
 
     def __getattr__(self, name):
         def f(*args, **kwargs):
             return self._call(name, *args, **kwargs)
 
         return f
 
     @exceptions.map_vcs_exceptions
     def _call(self, name, *args, **kwargs):
         payload = {
             'id': str(uuid.uuid4()),
             'method': name,
             'params': {'args': args, 'kwargs': kwargs}
         }
         return _remote_call(
             self.url, payload, EXCEPTIONS_MAP, self._session_factory())
 
 
 class RemoteRepo(object):
 
     def __init__(self, path, config, url, session, with_wire=None):
         self.url = url
         self._session = session
         self._wire = {
             "path": path,
             "config": config,
             "context": self._create_vcs_cache_context(),
         }
         if with_wire:
             self._wire.update(with_wire)
 
         # johbo: Trading complexity for performance. Avoiding the call to
         # log.debug brings a few percent gain even if is is not active.
         if log.isEnabledFor(logging.DEBUG):
             self._call = self._call_with_logging
 
     def __getattr__(self, name):
         def f(*args, **kwargs):
             return self._call(name, *args, **kwargs)
         return f
 
     @exceptions.map_vcs_exceptions
     def _call(self, name, *args, **kwargs):
         # TODO: oliver: This is currently necessary pre-call since the
         # config object is being changed for hooking scenarios
         wire = copy.deepcopy(self._wire)
         wire["config"] = wire["config"].serialize()
         payload = {
             'id': str(uuid.uuid4()),
             'method': name,
             'params': {'wire': wire, 'args': args, 'kwargs': kwargs}
         }
         return _remote_call(self.url, payload, EXCEPTIONS_MAP, self._session)
 
     def _call_with_logging(self, name, *args, **kwargs):
         log.debug('Calling %s@%s', self.url, name)
         return RemoteRepo._call(self, name, *args, **kwargs)
 
     def __getitem__(self, key):
         return self.revision(key)
 
     def _create_vcs_cache_context(self):
         """
         Creates a unique string which is passed to the VCSServer on every
         remote call. It is used as cache key in the VCSServer.
         """
         return str(uuid.uuid4())
 
     def invalidate_vcs_cache(self):
         """
         This invalidates the context which is sent to the VCSServer on every
         call to a remote method. It forces the VCSServer to create a fresh
         repository instance on the next call to a remote method.
         """
         self._wire['context'] = self._create_vcs_cache_context()
 
 
 class RemoteObject(object):
 
     def __init__(self, url, session):
         self._url = url
         self._session = session
 
         # johbo: Trading complexity for performance. Avoiding the call to
         # log.debug brings a few percent gain even if is is not active.
         if log.isEnabledFor(logging.DEBUG):
             self._call = self._call_with_logging
 
     def __getattr__(self, name):
         def f(*args, **kwargs):
             return self._call(name, *args, **kwargs)
         return f
 
     @exceptions.map_vcs_exceptions
     def _call(self, name, *args, **kwargs):
         payload = {
             'id': str(uuid.uuid4()),
             'method': name,
             'params': {'args': args, 'kwargs': kwargs}
         }
         return _remote_call(self._url, payload, EXCEPTIONS_MAP, self._session)
 
     def _call_with_logging(self, name, *args, **kwargs):
         log.debug('Calling %s@%s', self._url, name)
         return RemoteObject._call(self, name, *args, **kwargs)
 
 
 def _remote_call(url, payload, exceptions_map, session):
     try:
         response = session.post(url, data=msgpack.packb(payload))
     except pycurl.error as e:
         raise exceptions.HttpVCSCommunicationError(e)
 
+    if response.status_code >= 400:
+        log.error('Call to %s returned non 200 HTTP code: %s',
+                  url, response.status_code)
+        raise exceptions.HttpVCSCommunicationError(repr(response.content))
+
     try:
         response = msgpack.unpackb(response.content)
     except Exception:
-        log.exception('Failed to decode repsponse %r', response.content)
+        log.exception('Failed to decode response %r', response.content)
         raise
 
     error = response.get('error')
     if error:
         type_ = error.get('type', 'Exception')
         exc = exceptions_map.get(type_, Exception)
         exc = exc(error.get('message'))
         try:
             exc._vcs_kind = error['_vcs_kind']
         except KeyError:
             pass
 
         try:
             exc._vcs_server_traceback = error['traceback']
         except KeyError:
             pass
 
         raise exc
     return response.get('result')
 
 
 class VcsHttpProxy(object):
 
     CHUNK_SIZE = 16384
 
     def __init__(self, server_and_port, backend_endpoint):
         adapter = requests.adapters.HTTPAdapter(max_retries=5)
         self.base_url = urlparse.urljoin(
             'http://%s' % server_and_port, backend_endpoint)
         self.session = requests.Session()
         self.session.mount('http://', adapter)
 
     def handle(self, environment, input_data, *args, **kwargs):
         data = {
             'environment': environment,
             'input_data': input_data,
             'args': args,
             'kwargs': kwargs
         }
         result = self.session.post(
             self.base_url, msgpack.packb(data), stream=True)
         return self._get_result(result)
 
     def _deserialize_and_raise(self, error):
         exception = Exception(error['message'])
         try:
             exception._vcs_kind = error['_vcs_kind']
         except KeyError:
             pass
         raise exception
 
     def _iterate(self, result):
         unpacker = msgpack.Unpacker()
         for line in result.iter_content(chunk_size=self.CHUNK_SIZE):
             unpacker.feed(line)
             for chunk in unpacker:
                 yield chunk
 
     def _get_result(self, result):
         iterator = self._iterate(result)
         error = iterator.next()
         if error:
             self._deserialize_and_raise(error)
 
         status = iterator.next()
         headers = iterator.next()
 
         return iterator, status, headers
 
 
 class ThreadlocalSessionFactory(object):
     """
     Creates one CurlSession per thread on demand.
     """
 
     def __init__(self):
         self._thread_local = threading.local()
 
     def __call__(self):
         if not hasattr(self._thread_local, 'curl_session'):
             self._thread_local.curl_session = CurlSession()
         return self._thread_local.curl_session

rhodecode/tests/vcs/test_client_http.py

@@ -1,96 +1,133 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
 
 import mock
 import msgpack
 import pytest
 
 from rhodecode.lib import vcs
-from rhodecode.lib.vcs import client_http
-
-
-def test_uses_persistent_http_connections(caplog, vcsbackend_hg):
-    repo = vcsbackend_hg.repo
-    remote_call = repo._remote.branches
-
-    with caplog.at_level(logging.INFO):
-        for x in range(5):
-            remote_call(normal=True, closed=False)
-
-    new_connections = [
-        r for r in caplog.record_tuples if is_new_connection(*r)]
-    assert len(new_connections) <= 1
+from rhodecode.lib.vcs import client_http, exceptions
 
 
 def is_new_connection(logger, level, message):
     return (
         logger == 'requests.packages.urllib3.connectionpool' and
         message.startswith('Starting new HTTP'))
 
 
 @pytest.fixture
 def stub_session():
     """
     Stub of `requests.Session()`.
     """
     session = mock.Mock()
-    session.post().content = msgpack.packb({})
+    post = session.post()
+    post.content = msgpack.packb({})
+    post.status_code = 200
+
+    session.reset_mock()
+    return session
+
+
[e-mail]
+def stub_fail_session():
+    """
+    Stub of `requests.Session()`.
+    """
+    session = mock.Mock()
+    post = session.post()
+    post.content = msgpack.packb({'error': '500'})
+    post.status_code = 500
+
     session.reset_mock()
     return session
 
 
 @pytest.fixture
 def stub_session_factory(stub_session):
     """
     Stub of `rhodecode.lib.vcs.client_http.ThreadlocalSessionFactory`.
     """
     session_factory = mock.Mock()
     session_factory.return_value = stub_session
     return session_factory
 
 
[e-mail]
+def stub_session_failing_factory(stub_fail_session):
+    """
+    Stub of `rhodecode.lib.vcs.client_http.ThreadlocalSessionFactory`.
+    """
+    session_factory = mock.Mock()
+    session_factory.return_value = stub_fail_session
+    return session_factory
+
+
+def test_uses_persistent_http_connections(caplog, vcsbackend_hg):
+    repo = vcsbackend_hg.repo
+    remote_call = repo._remote.branches
+
+    with caplog.at_level(logging.INFO):
+        for x in range(5):
+            remote_call(normal=True, closed=False)
+
+    new_connections = [
+        r for r in caplog.record_tuples if is_new_connection(*r)]
+    assert len(new_connections) <= 1
+
+
 def test_repo_maker_uses_session_for_classmethods(stub_session_factory):
     repo_maker = client_http.RepoMaker(
         'server_and_port', 'endpoint', 'test_dummy_scm', stub_session_factory)
     repo_maker.example_call()
     stub_session_factory().post.assert_called_with(
         'http://server_and_port/endpoint', data=mock.ANY)
 
 
 def test_repo_maker_uses_session_for_instance_methods(
         stub_session_factory, config):
     repo_maker = client_http.RepoMaker(
         'server_and_port', 'endpoint', 'test_dummy_scm', stub_session_factory)
     repo = repo_maker('stub_path', config)
     repo.example_call()
     stub_session_factory().post.assert_called_with(
         'http://server_and_port/endpoint', data=mock.ANY)
 
 
 @mock.patch('rhodecode.lib.vcs.client_http.ThreadlocalSessionFactory')
 @mock.patch('rhodecode.lib.vcs.connection')
 def test_connect_passes_in_the_same_session(
         connection, session_factory_class, stub_session):
     session_factory = session_factory_class.return_value
     session_factory.return_value = stub_session
 
     vcs.connect_http('server_and_port')
+
+
+def test_repo_maker_uses_session_that_throws_error(
+        stub_session_failing_factory, config):
+    repo_maker = client_http.RepoMaker(
+        'server_and_port', 'endpoint', 'test_dummy_scm', stub_session_failing_factory)
+    repo = repo_maker('stub_path', config)
+
+    with pytest.raises(exceptions.HttpVCSCommunicationError):
+        repo.example_call()