core: support req-id tracking
This commit is contained in:
RhodeCode Admin
parent
107b31a78f
commit
d7e732ede3
6 changed files with 208 additions and 186 deletions
|
|
@ -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",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue