From d7e732ede3bff9725687c2ac795cffb21e9d9fd6 Mon Sep 17 00:00:00 2001 From: RhodeCode Admin Date: Mon, 30 Dec 2024 17:07:53 +0100 Subject: [PATCH] core: support req-id tracking --- rhodecode/lib/middleware/request_wrapper.py | 57 +++-- .../lib/middleware/utils/scm_app_http.py | 103 ++++----- rhodecode/lib/vcs/backends/svn/repository.py | 3 +- rhodecode/lib/vcs/client_http.py | 218 +++++++++--------- rhodecode/lib/vcs/connection.py | 11 +- rhodecode/model/db.py | 2 +- 6 files changed, 208 insertions(+), 186 deletions(-) diff --git a/rhodecode/lib/middleware/request_wrapper.py b/rhodecode/lib/middleware/request_wrapper.py index d87511b1..e87a4852 100644 --- a/rhodecode/lib/middleware/request_wrapper.py +++ b/rhodecode/lib/middleware/request_wrapper.py @@ -29,7 +29,7 @@ from rhodecode.lib.utils2 import safe_str, get_current_rhodecode_user log = logging.getLogger(__name__) -class RequestWrapperTween(object): +class RequestWrapperTween: def __init__(self, handler, registry): self.handler = handler self.registry = registry @@ -44,51 +44,66 @@ class RequestWrapperTween(object): def __call__(self, request): start = time.time() - log.debug('Starting request processing') + _path = get_path_info(request.environ) + _method = request.environ.get("REQUEST_METHOD") + + log.debug("Starting request processing: %s Request to %s", _method, _path) response = None request.req_wrapper_start = start try: response = self.handler(request) finally: + ua = get_user_agent(request.environ) count = request.request_count() _ver_ = rhodecode.__version__ - _path = get_path_info(request.environ) - _auth_user = self._get_user_info(request) ip = get_ip_addr(request.environ) match_route = request.matched_route.name if request.matched_route else "NOT_FOUND" - resp_code = getattr(response, 'status_code', 'UNDEFINED') + resp_code = getattr(response, "status_code", "UNDEFINED") + + req_id = getattr(request, 'req_id', None) + _auth_user = self._get_user_info(request) total = time.time() - start log.info( - 'Finished request processing: req[%4s] %s %s Request to %s time: %.4fs [%s], RhodeCode %s', - count, _auth_user, request.environ.get('REQUEST_METHOD'), - _path, total, get_user_agent(request. environ), _ver_, - extra={"time": total, "ver": _ver_, "ip": ip, - "path": _path, "view_name": match_route, "code": resp_code} + "Finished request processing: %s Request to %s req[%4s] %s time: %.4fs [%s], RhodeCode %s req_id=%s", + _method, + _path, + count, + _auth_user, + total, + ua, + _ver_, + req_id, + extra={ + "time": total, + "ver": _ver_, + "code": resp_code, + "path": _path, + "view_name": match_route, + "user_agent": ua, + "ip": ip, + }, ) + response.headers.setdefault("X-Rc-Req-Id", req_id) statsd = request.registry.statsd if statsd: elapsed_time_ms = round(1000.0 * total) # use ms only statsd.timing( - "rhodecode_req_timing.histogram", elapsed_time_ms, - tags=[ - "view_name:{}".format(match_route), - "code:{}".format(resp_code) - ], - use_decimals=False + "rhodecode_req_timing.histogram", + elapsed_time_ms, + tags=[f"view_name:{match_route}", f"code:{resp_code}"], + use_decimals=False, ) statsd.incr( - 'rhodecode_req_total', tags=[ - "view_name:{}".format(match_route), - "code:{}".format(resp_code) - ]) + "rhodecode_req_total", tags=[f"view_name:{match_route}", f"code:{resp_code}"] + ) return response def includeme(config): config.add_tween( - 'rhodecode.lib.middleware.request_wrapper.RequestWrapperTween', + "rhodecode.lib.middleware.request_wrapper.RequestWrapperTween", ) diff --git a/rhodecode/lib/middleware/utils/scm_app_http.py b/rhodecode/lib/middleware/utils/scm_app_http.py index da49d2a8..f9551ba9 100644 --- a/rhodecode/lib/middleware/utils/scm_app_http.py +++ b/rhodecode/lib/middleware/utils/scm_app_http.py @@ -36,28 +36,29 @@ log = logging.getLogger(__name__) def create_git_wsgi_app(repo_path, repo_name, config): - url = _vcs_streaming_url() + 'git/' + url = _vcs_streaming_url() + "git/" return VcsHttpProxy(url, repo_path, repo_name, config) def create_hg_wsgi_app(repo_path, repo_name, config): - url = _vcs_streaming_url() + 'hg/' + url = _vcs_streaming_url() + "hg/" return VcsHttpProxy(url, repo_path, repo_name, config) def _vcs_streaming_url(): - template = 'http://{}/stream/' - return template.format(rhodecode.CONFIG['vcs.server']) + vcs_server = rhodecode.CONFIG["vcs.server"] + vcs_url = f"http://{vcs_server}/stream/" + return vcs_url # TODO: johbo: Avoid the global. session = requests.Session() # Requests speedup, avoid reading .netrc and similar session.trust_env = False +session.verify = False # prevent urllib3 spawning our logs. -logging.getLogger("requests.packages.urllib3.connectionpool").setLevel( - logging.WARNING) +logging.getLogger("urllib3.connectionpool").setLevel(logging.WARNING) class VcsHttpProxy(object): @@ -78,9 +79,7 @@ class VcsHttpProxy(object): self._repo_path = repo_path self._config = config self.rc_extras = {} - log.debug( - "Creating VcsHttpProxy for repo %s, url %s", - repo_name, url) + log.debug("Creating VcsHttpProxy for repo %s, url %s", repo_name, url) def __call__(self, environ, start_response): config = self._config @@ -89,57 +88,55 @@ class VcsHttpProxy(object): call_context = { # TODO: johbo: Remove this, rely on URL path only - 'repo_name': self._repo_name, - 'repo_path': self._repo_path, - 'path_info': get_path_info(environ), - - 'repo_store': self.rc_extras.get('repo_store'), - 'server_config_file': self.rc_extras.get('config'), - - 'auth_user': self.rc_extras.get('username'), - 'auth_user_id': str(self.rc_extras.get('user_id')), - 'auth_user_ip': self.rc_extras.get('ip'), - - 'repo_config': config, - 'locked_status_code': rhodecode.CONFIG.get('lock_ret_code'), + "repo_name": self._repo_name, + "repo_path": self._repo_path, + "path_info": get_path_info(environ), + "repo_store": self.rc_extras.get("repo_store"), + "server_config_file": self.rc_extras.get("config"), + "auth_user": self.rc_extras.get("username"), + "auth_user_id": str(self.rc_extras.get("user_id")), + "auth_user_ip": self.rc_extras.get("ip"), + "repo_config": config, + "locked_status_code": rhodecode.CONFIG.get("lock_ret_code"), } - request_headers.update({ - # TODO: johbo: Avoid encoding and put this into payload? - 'X_RC_VCS_STREAM_CALL_CONTEXT': base64.b64encode(msgpack.packb(call_context)) - }) + request_headers.update( + { + # TODO: johbo: Avoid encoding and put this into payload? + "X_RC_VCS_STREAM_CALL_CONTEXT": base64.b64encode(msgpack.packb(call_context)) + } + ) - method = environ['REQUEST_METHOD'] + method = environ["REQUEST_METHOD"] # Preserve the query string url = self._url url = urllib.parse.urljoin(url, self._repo_name) - if environ.get('QUERY_STRING'): - url += '?' + environ['QUERY_STRING'] + if environ.get("QUERY_STRING"): + url += "?" + environ["QUERY_STRING"] + + log.debug("http-app: preparing %s request to: %s", method, url) + + import ipdb; + import rich + from rich import print as pprint, inspect + pprint("** STARTING DEBUG **") + ipdb.set_trace() - log.debug('http-app: preparing request to: %s', url) response = session.request( - method, - url, - data=_maybe_stream_request(environ), - headers=request_headers, - stream=True) + method, url, data=_maybe_stream_request(environ), headers=request_headers, stream=True + ) - log.debug('http-app: got vcsserver response: %s', response) + log.debug("http-app: got vcsserver response: %s", response) if response.status_code >= 500: - log.error('Exception returned by vcsserver at: %s %s, %s', - url, response.status_code, response.content) + log.error("Exception returned by vcsserver at: %s %s, %s", url, response.status_code, response.content) + response_headers = [] # Preserve the headers of the response, except hop_by_hop ones - response_headers = [ - (h, v) for h, v in response.headers.items() - if not wsgiref.util.is_hop_by_hop(h) - ] + response_headers = [(h, v) for h, v in response.headers.items() if not wsgiref.util.is_hop_by_hop(h)] # Build status argument for start_response callable. - status = '{status_code} {reason_phrase}'.format( - status_code=response.status_code, - reason_phrase=response.reason) + status = f"{response.status_code} {response.reason}" start_response(status, response_headers) return _maybe_stream_response(response) @@ -158,21 +155,22 @@ def read_in_chunks(stream_obj, block_size=1024, chunks=-1): def _is_request_chunked(environ): - stream = environ.get('HTTP_TRANSFER_ENCODING', '') == 'chunked' + stream = environ.get("HTTP_TRANSFER_ENCODING", "") == "chunked" return stream def _maybe_stream_request(environ): path = get_path_info(environ) stream = _is_request_chunked(environ) - req_method = environ['REQUEST_METHOD'] - log.debug('handling scm request: %s `%s` with stream support: %s', req_method, path, stream) + + req_method = environ["REQUEST_METHOD"] + log.debug("handling scm request: %s `%s` with stream support: %s", req_method, path, stream) if stream: # set stream by 256k - return read_in_chunks(environ['wsgi.input'], block_size=1024 * 256) + return read_in_chunks(environ["wsgi.input"], block_size=1024 * 256) else: - return environ['wsgi.input'].read() + return environ["wsgi.input"].read() def _maybe_stream_response(response): @@ -180,7 +178,8 @@ def _maybe_stream_response(response): Try to generate chunks from the response if it is chunked. """ stream = _is_chunked(response) - log.debug('returning response with stream: %s', stream) + log.debug("returning response with stream: %s", stream) + if stream: # read in 256k Chunks return response.raw.read_chunked(amt=1024 * 256) @@ -189,4 +188,4 @@ def _maybe_stream_response(response): def _is_chunked(response): - return response.headers.get('Transfer-Encoding', '') == 'chunked' + return response.headers.get("Transfer-Encoding", "") == "chunked" diff --git a/rhodecode/lib/vcs/backends/svn/repository.py b/rhodecode/lib/vcs/backends/svn/repository.py index 97e91066..4581807a 100644 --- a/rhodecode/lib/vcs/backends/svn/repository.py +++ b/rhodecode/lib/vcs/backends/svn/repository.py @@ -69,8 +69,7 @@ class SubversionRepository(base.BaseRepository): contact = base.BaseRepository.DEFAULT_CONTACT description = base.BaseRepository.DEFAULT_DESCRIPTION - def __init__(self, repo_path, config=None, create=False, src_url=None, with_wire=None, - bare=False, **kwargs): + def __init__(self, repo_path, config=None, create=False, src_url=None, with_wire=None, bare=False, **kwargs): self.path = safe_str(os.path.abspath(repo_path)) self.config = config if config else self.get_default_config() self.with_wire = with_wire or {"cache": False} # default should not use cache diff --git a/rhodecode/lib/vcs/client_http.py b/rhodecode/lib/vcs/client_http.py index 59ed0110..3025fe4c 100644 --- a/rhodecode/lib/vcs/client_http.py +++ b/rhodecode/lib/vcs/client_http.py @@ -48,13 +48,12 @@ log = logging.getLogger(__name__) # TODO: mikhail: Keep it in sync with vcsserver's # HTTPApplication.ALLOWED_EXCEPTIONS EXCEPTIONS_MAP = { - 'KeyError': KeyError, - 'URLError': urllib.error.URLError, + "KeyError": KeyError, + "URLError": urllib.error.URLError, } def _remote_call(url, payload, exceptions_map, session, retries=3): - for attempt in range(retries): try: response = session.post(url, data=msgpack.packb(payload)) @@ -62,14 +61,13 @@ def _remote_call(url, payload, exceptions_map, session, retries=3): except pycurl.error as e: error_code, error_message = e.args if error_code == pycurl.E_RECV_ERROR: - log.warning(f'Received a "Connection reset by peer" error. ' - f'Retrying... ({attempt + 1}/{retries})') + log.warning(f'Received a "Connection reset by peer" error. ' f"Retrying... ({attempt + 1}/{retries})") continue # Retry if connection reset error. - msg = f'{e}. \npycurl traceback: {traceback.format_exc()}' + msg = f"{e}. \npycurl traceback: {traceback.format_exc()}" raise exceptions.HttpVCSCommunicationError(msg) except Exception as e: - message = getattr(e, 'message', '') - if 'Failed to connect' in message: + message = getattr(e, "message", "") + if "Failed to connect" in message: # gevent doesn't return proper pycurl errors raise exceptions.HttpVCSCommunicationError(e) else: @@ -77,69 +75,64 @@ def _remote_call(url, payload, exceptions_map, session, retries=3): if response.status_code >= 400: content_type = response.content_type - log.error('Call to %s returned non 200 HTTP code: %s [%s]', - url, response.status_code, content_type) + log.error("Call to %s returned non 200 HTTP code: %s [%s]", url, response.status_code, content_type) raise exceptions.HttpVCSCommunicationError(repr(response.content)) try: response = msgpack.unpackb(response.content) except Exception: - log.exception('Failed to decode response from msgpack') + log.exception("Failed to decode response from msgpack") raise - error = response.get('error') + error = response.get("error") if error: - type_ = error.get('type', 'Exception') + type_ = error.get("type", "Exception") exc = exceptions_map.get(type_, Exception) - exc = exc(error.get('message')) + exc = exc(error.get("message")) try: - exc._vcs_kind = error['_vcs_kind'] + exc._vcs_kind = error["_vcs_kind"] except KeyError: pass try: - exc._vcs_server_traceback = error['traceback'] - exc._vcs_server_org_exc_name = error['org_exc'] - exc._vcs_server_org_exc_tb = error['org_exc_tb'] + exc._vcs_server_traceback = error["traceback"] + exc._vcs_server_org_exc_name = error["org_exc"] + exc._vcs_server_org_exc_tb = error["org_exc_tb"] except KeyError: pass exc.add_note(attach_exc_details(error)) raise exc # raising the org exception from vcsserver - return response.get('result') + return response.get("result") def attach_exc_details(error): - note = '-- EXC NOTE -- :\n' + note = "-- EXC NOTE -- :\n" note += f'vcs_kind: {error.get("_vcs_kind")}\n' note += f'org_exc: {error.get("_vcs_kind")}\n' note += f'tb: {error.get("traceback")}\n' - note += '-- END EXC NOTE --' + note += "-- END EXC NOTE --" return note def _streaming_remote_call(url, payload, exceptions_map, session, chunk_size): try: - headers = { - 'X-RC-Method': payload.get('method'), - 'X-RC-Repo-Name': payload.get('_repo_name') - } + headers = {"X-RC-Method": payload.get("method"), "X-RC-Repo-Name": payload.get("_repo_name")} response = session.post(url, data=msgpack.packb(payload), headers=headers) except pycurl.error as e: error_code, error_message = e.args - msg = f'{e}. \npycurl traceback: {traceback.format_exc()}' + msg = f"{e}. \npycurl traceback: {traceback.format_exc()}" raise exceptions.HttpVCSCommunicationError(msg) except Exception as e: - message = getattr(e, 'message', '') - if 'Failed to connect' in message: + 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) + 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) @@ -147,56 +140,56 @@ def _streaming_remote_call(url, payload, exceptions_map, session, chunk_size): class ServiceConnection(object): def __init__(self, server_and_port, backend_endpoint, session_factory): - self.url = urllib.parse.urljoin(f'http://{server_and_port}', backend_endpoint) + self.url = urllib.parse.urljoin(f"http://{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()) + 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 = urllib.parse.urljoin(f'http://{server_and_port}', backend_endpoint) - self.stream_url = urllib.parse.urljoin(f'http://{server_and_port}', backend_endpoint+'/stream') + self.url = urllib.parse.urljoin(f"http://{server_and_port}", backend_endpoint) + self.stream_url = urllib.parse.urljoin(f"http://{server_and_port}", backend_endpoint + "/stream") self._session_factory = session_factory self.backend_type = backend_type @classmethod def init_cache_region(cls, repo_id): - cache_namespace_uid = f'repo.{repo_id}' - region = rc_cache.get_or_create_region('cache_repo', cache_namespace_uid) + cache_namespace_uid = f"repo.{repo_id}" + region = rc_cache.get_or_create_region("cache_repo", cache_namespace_uid) return region, cache_namespace_uid def __call__(self, path, repo_id, config, with_wire=None): - log.debug('%s RepoMaker call on %s', self.backend_type.upper(), path) + 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): + from rhodecode.lib.pyramid_utils import get_current_request + + req_id = getattr(get_current_request(), "req_id", None) payload = { - 'id': str(uuid.uuid4()), - 'method': func_name, - 'backend': self.backend_type, - 'params': {'args': args, 'kwargs': kwargs} + "id": str(uuid.uuid4()), + "method": func_name, + "backend": self.backend_type, + "req_id": req_id, + "params": {"args": args, "kwargs": kwargs}, } url = self.url return _remote_call(url, payload, EXCEPTIONS_MAP, self._session_factory()) @@ -212,21 +205,22 @@ class RemoteRepo(object): cache_repo_id = self._repo_id_sanitizer(repo_id) _repo_name = self._get_repo_name(config, path) - self._cache_region, self._cache_namespace = \ - remote_maker.init_cache_region(cache_repo_id) + self._cache_region, self._cache_namespace = remote_maker.init_cache_region(cache_repo_id) with_wire = with_wire or {"cache": False} - repo_state_uid = with_wire.get('repo_state_uid') or 'state' + repo_state_uid = with_wire.get("repo_state_uid") or "state" + req_id = with_wire.get("req_id") self._wire = { "_repo_name": _repo_name, + "_req_id": req_id, "path": path, # repo path "repo_id": repo_id, "cache_repo_id": cache_repo_id, "config": config, "repo_state_uid": repo_state_uid, - "context": self._create_vcs_cache_context(path, repo_state_uid) + "context": self._create_vcs_cache_context(path, repo_state_uid), } if with_wire: @@ -237,22 +231,23 @@ class RemoteRepo(object): if log.isEnabledFor(logging.DEBUG): self._call_with_logging = True - self.cert_dir = get_cert_path(rhodecode.CONFIG.get('__file__')) + self.cert_dir = get_cert_path(rhodecode.CONFIG.get("__file__")) def _get_repo_name(self, config, path): - repo_store = config.get('paths', '/') - return path.split(repo_store)[-1].lstrip('/') + repo_store = config.get("paths", "/") + return path.split(repo_store)[-1].lstrip("/") def _repo_id_sanitizer(self, repo_id): - pathless = repo_id.replace('/', '__').replace('-', '_') - return ''.join(char if ord(char) < 128 else '_{}_'.format(ord(char)) for char in pathless) + pathless = repo_id.replace("/", "__").replace("-", "_") + return "".join(char if ord(char) < 128 else "_{}_".format(ord(char)) for char in pathless) def __getattr__(self, name): + if name.startswith("stream:"): - 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) @@ -263,40 +258,48 @@ class RemoteRepo(object): # 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)) + wire["config"].append(("vcs", "ssl_dir", self.cert_dir)) payload = { - 'id': str(uuid.uuid4()), - 'method': name, - "_repo_name": wire['_repo_name'], - 'params': {'wire': wire, 'args': args, 'kwargs': kwargs} + "id": str(uuid.uuid4()), + "method": name, + "_repo_name": wire["_repo_name"], + "_req_id": wire["_req_id"], + "params": {"wire": wire, "args": args, "kwargs": kwargs}, } - context_uid = wire.get('context') + context_uid = wire.get("context") return context_uid, payload def get_local_cache(self, name, args): cache_on = False - cache_key = '' - local_cache_on = rhodecode.ConfigGet().get_bool('vcs.methods.cache') + cache_key = "" + local_cache_on = rhodecode.ConfigGet().get_bool("vcs.methods.cache") cache_methods = [ - 'branches', 'tags', 'bookmarks', - 'is_large_file', 'is_binary', - 'fctx_size', 'stream:fctx_node_data', 'blob_raw_length', - 'node_history', - 'revision', 'tree_items', - 'ctx_branch', 'ctx_description', - 'bulk_request', - 'assert_correct_path', - 'is_path_valid_repository', + "branches", + "tags", + "bookmarks", + "is_large_file", + "is_binary", + "fctx_size", + "stream:fctx_node_data", + "blob_raw_length", + "node_history", + "revision", + "tree_items", + "ctx_branch", + "ctx_description", + "bulk_request", + "assert_correct_path", + "is_path_valid_repository", ] - wire_cache = self._wire['cache'] + wire_cache = self._wire["cache"] if local_cache_on and wire_cache and name in cache_methods: cache_on = True - repo_state_uid = self._wire['repo_state_uid'] + repo_state_uid = self._wire["repo_state_uid"] call_args = [a for a in args] cache_key = compute_key_from_params(repo_state_uid, name, *call_args) @@ -304,7 +307,6 @@ class RemoteRepo(object): @exceptions.map_vcs_exceptions def _call(self, name, *args, **kwargs): - context_uid, payload = self._base_call(name, *args, **kwargs) url = self.url @@ -312,24 +314,30 @@ class RemoteRepo(object): cache_on, cache_key = self.get_local_cache(name, args) @self._cache_region.conditional_cache_on_arguments( - namespace=self._cache_namespace, condition=cache_on and cache_key) + namespace=self._cache_namespace, condition=cache_on and cache_key + ) def remote_call(_cache_key): if self._call_with_logging: - args_repr = f'ARG: {str(args):.512}|KW: {str(kwargs):.512}' - log.debug('Calling %s@%s with args:%r. wire_context: %s cache_on: %s', - url, name, args_repr, context_uid, cache_on) + args_repr = f"ARG: {str(args):.512}|KW: {str(kwargs):.512}" + log.debug( + "Calling %s@%s with args:%r. wire_context: %s cache_on: %s", + url, + name, + args_repr, + context_uid, + cache_on, + ) return _remote_call(url, payload, EXCEPTIONS_MAP, self._session) result = remote_call(cache_key) if self._call_with_logging: - log.debug('Call %s@%s took: %.4fs. wire_context: %s', - url, name, time.time()-start, context_uid) + 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 + payload["chunk_size"] = self.CHUNK_SIZE url = self.stream_url start = time.time() @@ -338,15 +346,20 @@ class RemoteRepo(object): # Cache is a problem because this is a stream def streaming_remote_call(_cache_key): if self._call_with_logging: - args_repr = f'ARG: {str(args):.512}|KW: {str(kwargs):.512}' - log.debug('Calling %s@%s with args:%r. wire_context: %s cache_on: %s', - url, name, args_repr, context_uid, cache_on) + args_repr = f"ARG: {str(args):.512}|KW: {str(kwargs):.512}" + log.debug( + "Calling %s@%s with args:%r. wire_context: %s cache_on: %s", + url, + name, + args_repr, + context_uid, + cache_on, + ) return _streaming_remote_call(url, payload, EXCEPTIONS_MAP, self._session, self.CHUNK_SIZE) result = streaming_remote_call(cache_key) if self._call_with_logging: - log.debug('Call %s@%s took: %.4fs. wire_context: %s', - url, name, time.time()-start, context_uid) + log.debug("Call %s@%s took: %.4fs. wire_context: %s", url, name, time.time() - start, context_uid) return result def __getitem__(self, key): @@ -357,7 +370,7 @@ class RemoteRepo(object): 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)) + hash_key = "-".join(map(str, args)) return str(uuid.uuid5(uuid.NAMESPACE_URL, hash_key)) def invalidate_vcs_cache(self): @@ -366,36 +379,29 @@ class RemoteRepo(object): 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()) + self._wire["context"] = str(uuid.uuid4()) class VcsHttpProxy(object): - CHUNK_SIZE = 16384 def __init__(self, server_and_port, backend_endpoint): retries = Retry(total=5, connect=None, read=None, redirect=None) adapter = requests.adapters.HTTPAdapter(max_retries=retries) - self.base_url = urllib.parse.urljoin('http://%s' % server_and_port, backend_endpoint) + self.base_url = urllib.parse.urljoin("http://%s" % server_and_port, backend_endpoint) self.session = requests.Session() - self.session.mount('http://', adapter) + 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) + 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']) + exception = Exception(error["message"]) try: - exception._vcs_kind = error['_vcs_kind'] + exception._vcs_kind = error["_vcs_kind"] except KeyError: pass raise exception @@ -427,6 +433,6 @@ class ThreadlocalSessionFactory(object): self._thread_local = threading.local() def __call__(self): - if not hasattr(self._thread_local, 'curl_session'): + if not hasattr(self._thread_local, "curl_session"): self._thread_local.curl_session = CurlSession() return self._thread_local.curl_session diff --git a/rhodecode/lib/vcs/connection.py b/rhodecode/lib/vcs/connection.py index a919ea50..98573255 100644 --- a/rhodecode/lib/vcs/connection.py +++ b/rhodecode/lib/vcs/connection.py @@ -21,13 +21,16 @@ Holds connection for remote server. """ -class NotInitializedConnection(object): +class NotInitializedConnection: """Placeholder for objects which have to be initialized first.""" + def __init__(self, *args, **kwargs): + pass + def _raise_exc(self): raise Exception( - "rhodecode.lib.vcs is not yet initialized. " - "Make sure `vcs.server` is enabled in your configuration.") + "rhodecode.lib.vcs is not yet initialized. " "Make sure `vcs.server` is enabled in your configuration." + ) def __getattr__(self, item): self._raise_exc() @@ -35,7 +38,7 @@ class NotInitializedConnection(object): def __call__(self, *args, **kwargs): self._raise_exc() -# TODO: figure out a nice default value for these things + Service = NotInitializedConnection() Git = NotInitializedConnection() diff --git a/rhodecode/model/db.py b/rhodecode/model/db.py index 85e7bf22..5e6e0db5 100644 --- a/rhodecode/model/db.py +++ b/rhodecode/model/db.py @@ -2822,7 +2822,7 @@ class Repository(Base, BaseModel): config = config or self._config custom_wire = { 'cache': cache, # controls the vcs.remote cache - 'repo_state_uid': repo_state_uid + 'repo_state_uid': repo_state_uid, } repo = get_vcs_instance(