file-nodes: added streaming remote attributes for vcsserver.

- this will enable doing a streaming raw content or raw downloads of large files without transfering them over to enterprise for pack & repack using msgpack
- msgpack has a limit of 4gb and generally pack+repack for 2gb is very slow

rhodecode/apps/repository/views/repo_files.py

@@ -844,10 +844,9 b' class RepoFilesView(RepoAppView):'
         if disposition == 'attachment':
             disposition = self._get_attachement_headers(f_path)
 
-        def stream_node():
-            yield file_node.raw_bytes
+        stream_content = file_node.stream_bytes()
 
-        response = Response(app_iter=stream_node())
+        response = Response(app_iter=stream_content)
         response.content_disposition = disposition
         response.content_type = mimetype
 
@@ -883,10 +882,9 b' class RepoFilesView(RepoAppView):'
 
         disposition = self._get_attachement_headers(f_path)
 
-        def stream_node():
-            yield file_node.raw_bytes
+        stream_content = file_node.stream_bytes()
 
-        response = Response(app_iter=stream_node())
+        response = Response(app_iter=stream_content)
         response.content_disposition = disposition
         response.content_type = file_node.mimetype
 

rhodecode/lib/vcs/__init__.py

@@ -72,11 +72,11 b' def connect_http(server_and_port):'
 
     session_factory = client_http.ThreadlocalSessionFactory()
 
-    connection.Git = client_http.RepoMaker(
+    connection.Git = client_http.RemoteVCSMaker(
         server_and_port, '/git', 'git', session_factory)
-    connection.Hg = client_http.RepoMaker(
+    connection.Hg = client_http.RemoteVCSMaker(
         server_and_port, '/hg', 'hg', session_factory)
-    connection.Svn = client_http.RepoMaker(
+    connection.Svn = client_http.RemoteVCSMaker(
         server_and_port, '/svn', 'svn', session_factory)
     connection.Service = client_http.ServiceConnection(
         server_and_port, '/_service', session_factory)
@@ -107,21 +107,6 b' def connect_vcs(server_and_port, protoco'
         raise Exception('Invalid vcs server protocol "{}"'.format(protocol))
 
 
-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.
@@ -176,12 +161,23 b' class CurlResponse(object):'
 
     @property
     def content(self):
+        try:
         return self._response_buffer.getvalue()
+        finally:
+            self._response_buffer.close()
 
     @property
     def status_code(self):
         return self._status_code
 
+    def iter_content(self, chunk_size):
+        self._response_buffer.seek(0)
+        while 1:
+            chunk = self._response_buffer.read(chunk_size)
+            if not chunk:
+                break
+            yield chunk
+
 
 def _create_http_rpc_session():
     session = CurlSession()

rhodecode/lib/vcs/backends/base.py

@@ -1060,6 +1060,12 b' class BaseCommit(object):'
         """
         raise NotImplementedError
 
+    def get_file_content_streamed(self, path):
+        """
+        returns a streaming response from vcsserver with file content
+        """
+        raise NotImplementedError
+
     def get_file_size(self, path):
         """
         Returns size of the file at the given `path`.
@@ -1631,6 +1637,9 b' class EmptyCommit(BaseCommit):'
     def get_file_content(self, path):
         return u''
 
+    def get_file_content_streamed(self, path):
+        yield self.get_file_content()
+
     def get_file_size(self, path):
         return 0
 

rhodecode/lib/vcs/backends/git/commit.py

@@ -252,6 +252,11 b' class GitCommit(base.BaseCommit):'
         tree_id, _ = self._get_tree_id_for_path(path)
         return self._remote.blob_as_pretty_string(tree_id)
 
+    def get_file_content_streamed(self, path):
+        tree_id, _ = self._get_tree_id_for_path(path)
+        stream_method = getattr(self._remote, 'stream:blob_as_pretty_string')
+        return stream_method(tree_id)
+
     def get_file_size(self, path):
         """
         Returns size of the file at given `path`.

rhodecode/lib/vcs/backends/hg/commit.py

@@ -238,6 +238,11 b' class MercurialCommit(base.BaseCommit):'
         path = self._get_filectx(path)
         return self._remote.fctx_node_data(self.raw_id, path)
 
+    def get_file_content_streamed(self, path):
+        path = self._get_filectx(path)
+        stream_method = getattr(self._remote, 'stream:fctx_node_data')
+        return stream_method(self.raw_id, path)
+
     def get_file_size(self, path):
         """
         Returns size of the file at given ``path``.

rhodecode/lib/vcs/backends/svn/commit.py

@@ -117,6 +117,11 b' class SubversionCommit(base.BaseCommit):'
         path = self._fix_path(path)
         return self._remote.get_file_content(safe_str(path), self._svn_rev)
 
+    def get_file_content_streamed(self, path):
+        path = self._fix_path(path)
+        stream_method = getattr(self._remote, 'stream:get_file_content')
+        return stream_method(safe_str(path), self._svn_rev)
+
     def get_file_size(self, path):
         path = self._fix_path(path)
         return self._remote.get_file_size(safe_str(path), self._svn_rev)

rhodecode/lib/vcs/client_http.py

@@ -51,163 +51,6 b' EXCEPTIONS_MAP = {'
 }
 
 
-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, repo_id, config, with_wire=None):
-        log.debug('%s RepoMaker call on %s', self.backend_type.upper(), path)
-        return RemoteRepo(path, repo_id, 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, repo_id, config, url, session, with_wire=None):
-        self.url = url
-        self._session = session
-        with_wire = with_wire or {}
-
-        repo_state_uid = with_wire.get('repo_state_uid') or 'state'
-        self._wire = {
-            "path": path,  # repo path
-            "repo_id": repo_id,
-            "config": config,
-            "repo_state_uid": repo_state_uid,
-            "context": self._create_vcs_cache_context(path, repo_state_uid)
-        }
-
-        if with_wire:
-            self._wire.update(with_wire)
-
-        # NOTE(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_with_logging = True
-
-        self.cert_dir = get_cert_path(rhodecode.CONFIG.get('__file__'))
-
-    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()
-        wire["config"].append(('vcs', 'ssl_dir', self.cert_dir))
-
-        payload = {
-            'id': str(uuid.uuid4()),
-            'method': name,
-            'params': {'wire': wire, 'args': args, 'kwargs': kwargs}
-        }
-
-        if self._call_with_logging:
-            start = time.time()
-            context_uid = wire.get('context')
-            log.debug('Calling %s@%s with args:%.10240r. wire_context: %s',
-                      self.url, name, args, context_uid)
-        result = _remote_call(self.url, payload, EXCEPTIONS_MAP, self._session)
-        if self._call_with_logging:
-            log.debug('Call %s@%s took: %.4fs. wire_context: %s',
-                      self.url, name, time.time()-start, context_uid)
-        return result
-
-    def __getitem__(self, key):
-        return self.revision(key)
-
-    def _create_vcs_cache_context(self, *args):
-        """
-        Creates a unique string which is passed to the VCSServer on every
-        remote call. It is used as cache key in the VCSServer.
-        """
-        hash_key = '-'.join(map(str, args))
-        return str(uuid.uuid5(uuid.NAMESPACE_URL, hash_key))
-
-    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'] = str(uuid.uuid4())
-
-
-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))
@@ -254,6 +97,191 b' def _remote_call(url, payload, exception'
     return response.get('result')
 
 
+def _streaming_remote_call(url, payload, exceptions_map, session, chunk_size):
+    try:
+        response = session.post(url, data=msgpack.packb(payload))
+    except pycurl.error as e:
+        msg = '{}. \npycurl traceback: {}'.format(e, traceback.format_exc())
+        raise exceptions.HttpVCSCommunicationError(msg)
+    except Exception as e:
+        message = getattr(e, 'message', '')
+        if 'Failed to connect' in message:
+            # gevent doesn't return proper pycurl errors
+            raise exceptions.HttpVCSCommunicationError(e)
+        else:
+            raise
+
+    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))
+
+    return response.iter_content(chunk_size=chunk_size)
+
+
+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 RemoteVCSMaker(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.stream_url = urlparse.urljoin('http://%s' % server_and_port, backend_endpoint+'/stream')
+
+        self._session_factory = session_factory
+        self.backend_type = backend_type
+
+    def __call__(self, path, repo_id, config, with_wire=None):
+        log.debug('%s RepoMaker call on %s', self.backend_type.upper(), path)
+        return RemoteRepo(path, repo_id, config, self, with_wire=with_wire)
+
+    def __getattr__(self, name):
+        def remote_attr(*args, **kwargs):
+            return self._call(name, *args, **kwargs)
+        return remote_attr
+
+    @exceptions.map_vcs_exceptions
+    def _call(self, func_name, *args, **kwargs):
+        payload = {
+            'id': str(uuid.uuid4()),
+            'method': func_name,
+            'backend': self.backend_type,
+            'params': {'args': args, 'kwargs': kwargs}
+        }
+        url = self.url
+        return _remote_call(url, payload, EXCEPTIONS_MAP, self._session_factory())
+
+
+class RemoteRepo(object):
+    CHUNK_SIZE = 16384
+
+    def __init__(self, path, repo_id, config, remote_maker, with_wire=None):
+        self.url = remote_maker.url
+        self.stream_url = remote_maker.stream_url
+        self._session = remote_maker._session_factory()
+
+        with_wire = with_wire or {}
+
+        repo_state_uid = with_wire.get('repo_state_uid') or 'state'
+        self._wire = {
+            "path": path,  # repo path
+            "repo_id": repo_id,
+            "config": config,
+            "repo_state_uid": repo_state_uid,
+            "context": self._create_vcs_cache_context(path, repo_state_uid)
+        }
+
+        if with_wire:
+            self._wire.update(with_wire)
+
+        # NOTE(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_with_logging = True
+
+        self.cert_dir = get_cert_path(rhodecode.CONFIG.get('__file__'))
+
+    def __getattr__(self, name):
+
+        if name.startswith('stream:'):
+            def repo_remote_attr(*args, **kwargs):
+                return self._call_stream(name, *args, **kwargs)
+        else:
+            def repo_remote_attr(*args, **kwargs):
+                return self._call(name, *args, **kwargs)
+
+        return repo_remote_attr
+
+    def _base_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()
+        wire["config"].append(('vcs', 'ssl_dir', self.cert_dir))
+
+        payload = {
+            'id': str(uuid.uuid4()),
+            'method': name,
+            'params': {'wire': wire, 'args': args, 'kwargs': kwargs}
+        }
+
+        context_uid = wire.get('context')
+        return context_uid, payload
+
+    @exceptions.map_vcs_exceptions
+    def _call(self, name, *args, **kwargs):
+        context_uid, payload = self._base_call(name, *args, **kwargs)
+        url = self.url
+
+        start = time.time()
+        if self._call_with_logging:
+            log.debug('Calling %s@%s with args:%.10240r. wire_context: %s',
+                      url, name, args, context_uid)
+
+        result = _remote_call(url, payload, EXCEPTIONS_MAP, self._session)
+        if self._call_with_logging:
+            log.debug('Call %s@%s took: %.4fs. wire_context: %s',
+                      url, name, time.time()-start, context_uid)
+        return result
+
+    @exceptions.map_vcs_exceptions
+    def _call_stream(self, name, *args, **kwargs):
+        context_uid, payload = self._base_call(name, *args, **kwargs)
+        payload['chunk_size'] = self.CHUNK_SIZE
+        url = self.stream_url
+
+        start = time.time()
+        if self._call_with_logging:
+            log.debug('Calling %s@%s with args:%.10240r. wire_context: %s',
+                      url, name, args, context_uid)
+
+        result = _streaming_remote_call(url, payload, EXCEPTIONS_MAP, self._session,
+                                        self.CHUNK_SIZE)
+
+        if self._call_with_logging:
+            log.debug('Call %s@%s took: %.4fs. wire_context: %s',
+                      url, name, time.time()-start, context_uid)
+        return result
+
+    def __getitem__(self, key):
+        return self.revision(key)
+
+    def _create_vcs_cache_context(self, *args):
+        """
+        Creates a unique string which is passed to the VCSServer on every
+        remote call. It is used as cache key in the VCSServer.
+        """
+        hash_key = '-'.join(map(str, args))
+        return str(uuid.uuid5(uuid.NAMESPACE_URL, hash_key))
+
+    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'] = str(uuid.uuid4())
+
+
 class VcsHttpProxy(object):
 
     CHUNK_SIZE = 16384

rhodecode/lib/vcs/nodes.py

@@ -27,6 +27,7 b' import stat'
 
 from zope.cachedescriptors.property import Lazy as LazyProperty
 
+import rhodecode
 from rhodecode.config.conf import LANGUAGES_EXTENSIONS_MAP
 from rhodecode.lib.utils import safe_unicode, safe_str
 from rhodecode.lib.utils2 import md5
@@ -369,6 +370,17 b' class FileNode(Node):'
             content = self._content
         return content
 
+    def stream_bytes(self):
+        """
+        Returns an iterator that will stream the content of the file directly from
+        vcsserver without loading it to memory.
+        """
+        if self.commit:
+            return self.commit.get_file_content_streamed(self.path)
+        raise NodeError(
+            "Cannot retrieve message of the file without related "
+            "commit attribute")
+
     @LazyProperty
     def md5(self):
         """
@@ -848,3 +860,11 b' class LargeFileNode(FileNode):'
         Overwrites name to be the org lf path
         """
         return self.org_path
+
+    def stream_bytes(self):
+        with open(self.path, 'rb') as stream:
+            while True:
+                data = stream.read(16 * 1024)
+                if not data:
+                    break
+                yield data

rhodecode/tests/plugin.py

@@ -57,7 +57,6 b' from rhodecode.model.integration import '
 from rhodecode.integrations import integration_type_registry
 from rhodecode.integrations.types.base import IntegrationTypeBase
 from rhodecode.lib.utils import repo2db_mapper
-from rhodecode.lib.vcs import create_vcsserver_proxy
 from rhodecode.lib.vcs.backends import get_backend
 from rhodecode.lib.vcs.nodes import FileNode
 from rhodecode.tests import (
@@ -1398,82 +1397,7 b' def testrun():'
     }
 
 
[e-mail](autouse=True)
-def collect_appenlight_stats(request, testrun):
-    """
-    This fixture reports memory consumtion of single tests.
-
-    It gathers data based on `psutil` and sends them to Appenlight. The option
-    ``--ae`` has te be used to enable this fixture and the API key for your
-    application has to be provided in ``--ae-key``.
-    """
-    try:
-        # cygwin cannot have yet psutil support.
-        import psutil
-    except ImportError:
-        return
-
-    if not request.config.getoption('--appenlight'):
-        return
-    else:
-        # Only request the baseapp fixture if appenlight tracking is
-        # enabled. This will speed up a test run of unit tests by 2 to 3
-        # seconds if appenlight is not enabled.
-        baseapp = request.getfuncargvalue("baseapp")
-    url = '{}/api/logs'.format(request.config.getoption('--appenlight-url'))
-    client = AppenlightClient(
-        url=url,
-        api_key=request.config.getoption('--appenlight-api-key'),
-        namespace=request.node.nodeid,
-        request=str(testrun['uuid']),
-        testrun=testrun)
-
-    client.collect({
-        'message': "Starting",
-    })
-
-    server_and_port = baseapp.config.get_settings()['vcs.server']
-    protocol = baseapp.config.get_settings()['vcs.server.protocol']
-    server = create_vcsserver_proxy(server_and_port, protocol)
-    with server:
-        vcs_pid = server.get_pid()
-        server.run_gc()
-    vcs_process = psutil.Process(vcs_pid)
-    mem = vcs_process.memory_info()
-    client.tag_before('vcsserver.rss', mem.rss)
-    client.tag_before('vcsserver.vms', mem.vms)
-
-    test_process = psutil.Process()
-    mem = test_process.memory_info()
-    client.tag_before('test.rss', mem.rss)
-    client.tag_before('test.vms', mem.vms)
-
-    client.tag_before('time', time.time())
-
-    @request.addfinalizer
-    def send_stats():
-        client.tag_after('time', time.time())
-        with server:
-            gc_stats = server.run_gc()
-        for tag, value in gc_stats.items():
-            client.tag_after(tag, value)
-        mem = vcs_process.memory_info()
-        client.tag_after('vcsserver.rss', mem.rss)
-        client.tag_after('vcsserver.vms', mem.vms)
-
-        mem = test_process.memory_info()
-        client.tag_after('test.rss', mem.rss)
-        client.tag_after('test.vms', mem.vms)
-
-        client.collect({
-            'message': "Finished",
-        })
-        client.send_stats()
-
-    return client
-
-
-class AppenlightClient():
+class AppenlightClient(object):
 
     url_template = '{url}?protocol_version=0.5'
 

rhodecode/tests/vcs/test_client_http.py

@@ -96,7 +96,7 b' def test_uses_persistent_http_connection'
 
 
 def test_repo_maker_uses_session_for_classmethods(stub_session_factory):
-    repo_maker = client_http.RepoMaker(
+    repo_maker = client_http.RemoteVCSMaker(
         'server_and_port', 'endpoint', 'test_dummy_scm', stub_session_factory)
     repo_maker.example_call()
     stub_session_factory().post.assert_called_with(
@@ -105,7 +105,7 b' def test_repo_maker_uses_session_for_cla'
 
 def test_repo_maker_uses_session_for_instance_methods(
         stub_session_factory, config):
-    repo_maker = client_http.RepoMaker(
+    repo_maker = client_http.RemoteVCSMaker(
         'server_and_port', 'endpoint', 'test_dummy_scm', stub_session_factory)
     repo = repo_maker('stub_path', 'stub_repo_id', config)
     repo.example_call()
@@ -125,7 +125,7 b' def test_connect_passes_in_the_same_sess'
 
 def test_repo_maker_uses_session_that_throws_error(
         stub_session_failing_factory, config):
-    repo_maker = client_http.RepoMaker(
+    repo_maker = client_http.RemoteVCSMaker(
         'server_and_port', 'endpoint', 'test_dummy_scm', stub_session_failing_factory)
     repo = repo_maker('stub_path', 'stub_repo_id', config)