exc-tracking: synced with ce

vcsserver/lib/exc_tracking.py

@@ -1,267 +1,272 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # 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 General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
-
+import io
 import os
 import time
 import sys
 import datetime
 import msgpack
 import logging
 import traceback
 import tempfile
 import glob
 
 log = logging.getLogger(__name__)
 
 # NOTE: Any changes should be synced with exc_tracking at rhodecode.lib.exc_tracking
 global_prefix = 'vcsserver'
 exc_store_dir_name = 'rc_exception_store_v1'
 
 
 def exc_serialize(exc_id, tb, exc_type, extra_data=None):
-
     data = {
-        'version': 'v1',
-        'exc_id': exc_id,
-        'exc_utc_date': datetime.datetime.utcnow().isoformat(),
-        'exc_timestamp': repr(time.time()),
-        'exc_message': tb,
-        'exc_type': exc_type,
+        "version": "v1",
+        "exc_id": exc_id,
+        "exc_utc_date": datetime.datetime.utcnow().isoformat(),
+        "exc_timestamp": repr(time.time()),
+        "exc_message": tb,
+        "exc_type": exc_type,
     }
     if extra_data:
         data.update(extra_data)
     return msgpack.packb(data), data
 
 
 def exc_unserialize(tb):
     return msgpack.unpackb(tb)
 
 
 _exc_store = None
 
 
 def get_exc_store():
     """
     Get and create exception store if it's not existing
     """
     global _exc_store
 
     if _exc_store is not None:
         # quick global cache
         return _exc_store
 
     import vcsserver as app
 
-    exc_store_dir = app.CONFIG.get('exception_tracker.store_path', '') or tempfile.gettempdir()
+    exc_store_dir = (
+        app.CONFIG.get("exception_tracker.store_path", "") or tempfile.gettempdir()
+    )
     _exc_store_path = os.path.join(exc_store_dir, exc_store_dir_name)
 
     _exc_store_path = os.path.abspath(_exc_store_path)
     if not os.path.isdir(_exc_store_path):
         os.makedirs(_exc_store_path)
-        log.debug('Initializing exceptions store at %s', _exc_store_path)
+        log.debug("Initializing exceptions store at %s", _exc_store_path)
         _exc_store = _exc_store_path
 
     return _exc_store_path
 
 
 def get_detailed_tb(exc_info):
-    from io import StringIO
-
     try:
-        from pip._vendor.rich import traceback as rich_tb, scope as rich_scope, console as rich_console
+        from pip._vendor.rich import (
+            traceback as rich_tb,
+            scope as rich_scope,
+            console as rich_console,
+        )
     except ImportError:
         try:
-            from rich import traceback as rich_tb, scope as rich_scope, console as rich_console
+            from rich import (
+                traceback as rich_tb,
+                scope as rich_scope,
+                console as rich_console,
+            )
         except ImportError:
             return None
 
-    console = rich_console.Console(
-        width=160,
-        file=StringIO()
-    )
+    console = rich_console.Console(width=160, file=io.StringIO())
 
     exc = rich_tb.Traceback.extract(*exc_info, show_locals=True)
 
     tb_rich = rich_tb.Traceback(
         trace=exc,
         width=160,
         extra_lines=3,
         theme=None,
         word_wrap=False,
         show_locals=False,
-        max_frames=100
+        max_frames=100,
     )
 
-    formatted_locals = ""
-
     # last_stack = exc.stacks[-1]
     # last_frame = last_stack.frames[-1]
     # if last_frame and last_frame.locals:
     #     console.print(
     #         rich_scope.render_scope(
     #             last_frame.locals,
     #             title=f'{last_frame.filename}:{last_frame.lineno}'))
 
     console.print(tb_rich)
     formatted_locals = console.file.getvalue()
 
     return formatted_locals
 
 
 def get_request_metadata(request=None) -> dict:
     request_metadata = {}
     if not request:
         from pyramid.threadlocal import get_current_request
+
         request = get_current_request()
 
     # NOTE(marcink): store request information into exc_data
     if request:
-        request_metadata['client_address'] = getattr(request, 'client_addr', '')
-        request_metadata['user_agent'] = getattr(request, 'user_agent', '')
-        request_metadata['method'] = getattr(request, 'method', '')
-        request_metadata['url'] = getattr(request, 'url', '')
+        request_metadata["client_address"] = getattr(request, "client_addr", "")
+        request_metadata["user_agent"] = getattr(request, "user_agent", "")
+        request_metadata["method"] = getattr(request, "method", "")
+        request_metadata["url"] = getattr(request, "url", "")
     return request_metadata
 
 
 def format_exc(exc_info):
     exc_type, exc_value, exc_traceback = exc_info
     tb = "++ TRACEBACK ++\n\n"
     tb += "".join(traceback.format_exception(exc_type, exc_value, exc_traceback, None))
     
     detailed_tb = getattr(exc_value, "_org_exc_tb", None)
     
     if detailed_tb:
         remote_tb = detailed_tb
         if isinstance(detailed_tb, str):
             remote_tb = [detailed_tb]
     
         tb += (
             "\n+++ BEG SOURCE EXCEPTION +++\n\n"
             "{}\n"
             "+++ END SOURCE EXCEPTION +++\n"
             "".format("\n".join(remote_tb))
         )
     
         # Avoid that remote_tb also appears in the frame
         del remote_tb
     
     locals_tb = get_detailed_tb(exc_info)
     if locals_tb:
         tb += f"\n+++ DETAILS +++\n\n{locals_tb}\n" ""
     return tb
 
 
 def _store_exception(exc_id, exc_info, prefix, request_path=''):
     """
     Low level function to store exception in the exception tracker
     """
 
     extra_data = {}
     extra_data.update(get_request_metadata())
 
     exc_type, exc_value, exc_traceback = exc_info
     tb = format_exc(exc_info)
 
     exc_type_name = exc_type.__name__
     exc_data, org_data = exc_serialize(exc_id, tb, exc_type_name, extra_data=extra_data)
 
-    exc_pref_id = '{}_{}_{}'.format(exc_id, prefix, org_data['exc_timestamp'])
+    exc_pref_id = f"{exc_id}_{prefix}_{org_data['exc_timestamp']}"
     exc_store_path = get_exc_store()
     if not os.path.isdir(exc_store_path):
         os.makedirs(exc_store_path)
     stored_exc_path = os.path.join(exc_store_path, exc_pref_id)
-    with open(stored_exc_path, 'wb') as f:
+    with open(stored_exc_path, "wb") as f:
         f.write(exc_data)
-    log.debug('Stored generated exception %s as: %s', exc_id, stored_exc_path)
+    log.debug("Stored generated exception %s as: %s", exc_id, stored_exc_path)
 
-    log.error(
-        'error occurred handling this request.\n'
-        'Path: `%s`, %s',
-        request_path, tb)
+    if request_path:
+        log.error(
+            'error occurred handling this request.\n'
+            'Path: `%s`, %s',
+            request_path, tb)
 
 
 def store_exception(exc_id, exc_info, prefix=global_prefix, request_path=''):
     """
     Example usage::
 
         exc_info = sys.exc_info()
         store_exception(id(exc_info), exc_info)
     """
 
     try:
         exc_type = exc_info[0]
         exc_type_name = exc_type.__name__
 
-        _store_exception(exc_id=exc_id, exc_info=exc_info, prefix=prefix,
-                         request_path=request_path)
+        _store_exception(
+            exc_id=exc_id, exc_info=exc_info, prefix=prefix, request_path=request_path,
+        )
         return exc_id, exc_type_name
     except Exception:
-        log.exception('Failed to store exception `%s` information', exc_id)
+        log.exception("Failed to store exception `%s` information", exc_id)
         # there's no way this can fail, it will crash server badly if it does.
         pass
 
 
 def _find_exc_file(exc_id, prefix=global_prefix):
     exc_store_path = get_exc_store()
     if prefix:
-        exc_id = f'{exc_id}_{prefix}'
+        exc_id = f"{exc_id}_{prefix}"
     else:
         # search without a prefix
-        exc_id = f'{exc_id}'
+        exc_id = f"{exc_id}"
 
     found_exc_id = None
-    matches = glob.glob(os.path.join(exc_store_path, exc_id) + '*')
+    matches = glob.glob(os.path.join(exc_store_path, exc_id) + "*")
     if matches:
         found_exc_id = matches[0]
 
     return found_exc_id
 
 
 def _read_exception(exc_id, prefix):
     exc_id_file_path = _find_exc_file(exc_id=exc_id, prefix=prefix)
     if exc_id_file_path:
-        with open(exc_id_file_path, 'rb') as f:
+        with open(exc_id_file_path, "rb") as f:
             return exc_unserialize(f.read())
     else:
-        log.debug('Exception File `%s` not found', exc_id_file_path)
+        log.debug("Exception File `%s` not found", exc_id_file_path)
     return None
 
 
 def read_exception(exc_id, prefix=global_prefix):
     try:
         return _read_exception(exc_id=exc_id, prefix=prefix)
     except Exception:
-        log.exception('Failed to read exception `%s` information', exc_id)
+        log.exception("Failed to read exception `%s` information", exc_id)
         # there's no way this can fail, it will crash server badly if it does.
     return None
 
 
 def delete_exception(exc_id, prefix=global_prefix):
     try:
         exc_id_file_path = _find_exc_file(exc_id, prefix=prefix)
         if exc_id_file_path:
             os.remove(exc_id_file_path)
 
     except Exception:
-        log.exception('Failed to remove exception `%s` information', exc_id)
+        log.exception("Failed to remove exception `%s` information", exc_id)
         # there's no way this can fail, it will crash server badly if it does.
         pass
 
 
 def generate_id():
     return id(object())