feat: ASdd a traceback filter to make the error message concise

src/ai/backend/kernel/base.py

@@ -45,9 +45,11 @@
 from .jupyter_client import aexecute_interactive
 from .logging import BraceStyleAdapter, setup_logger
 from .service import ServiceParser
-from .utils import scan_proc_stats, wait_local_port_open
+from .utils import TracebackSourceFilter, scan_proc_stats, wait_local_port_open
 
-log = BraceStyleAdapter(logging.getLogger())
+logger = logging.getLogger()
+logger.addFilter(TracebackSourceFilter(str(Path(__file__).parent)))
+log = BraceStyleAdapter(logger)
 
 TReturn = TypeVar("TReturn")
 
@@ -343,10 +345,10 @@ async def _handle_exception(
         except Exception as e:
             match e:
                 case FileNotFoundError():
-                    msg = "Could not find: %r"
+                    msg = "Could not find: {!r}"
                     if help_text:
                         msg += f" ({help_text})"
-                    log.error(msg, e.filename)
+                    log.exception(msg, e.filename)
                 case _:
                     msg = "Unexpected error!"
                     if help_text:
@@ -365,7 +367,7 @@ async def _init_with_loop(self) -> None:
             if ret is FAILURE:
                 log.warning(
                     "We are skipping the runtime-specific initialization failure, "
-                    "but the container may not work as expected."
+                    "and the container may not work as expected."
                 )
             await self._handle_exception(
                 init_sshd_service(self.child_env),

src/ai/backend/kernel/utils.py

@@ -1,5 +1,7 @@
 import asyncio
+import logging
 import os
+import traceback
 from pathlib import Path
 from typing import Final
 
@@ -34,6 +36,24 @@ def find_executable(*paths):
     return None
 
 
+class TracebackSourceFilter(logging.Filter):
+    def __init__(self, path_pattern: str):
+        super().__init__()
+        self.path_pattern = path_pattern
+
+    def filter(self, record: logging.LogRecord):
+        # Filter the traceback to include only lines from the specified source file path
+        if record.exc_info:
+            exc_type, exc_value, exc_tb = record.exc_info
+            filtered_traceback = []
+            for tb in traceback.extract_tb(exc_tb):
+                if self.path_pattern in tb[0]:
+                    filtered_traceback.append(tb)
+            if filtered_traceback:
+                record.exc_text = "".join(traceback.format_list(filtered_traceback))
+        return True
+
+
 async def safe_close_task(task):
     if task is not None and not task.done():
         task.cancel()