Improve performance of system_log traceback handling (#114992)

homeassistant/components/system_log/__init__.py

@@ -7,6 +7,7 @@ import logging
 import re
 import sys
 import traceback
+from types import FrameType
 from typing import Any, cast
 
 import voluptuous as vol
@@ -18,7 +19,7 @@ from homeassistant.core import Event, HomeAssistant, ServiceCall, callback
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.typing import ConfigType
 
-KeyType = tuple[str, tuple[str, int], str | None]
+KeyType = tuple[str, tuple[str, int], tuple[str, int, str] | None]
 
 CONF_MAX_ENTRIES = "max_entries"
 CONF_FIRE_EVENT = "fire_event"
@@ -65,16 +66,18 @@ SERVICE_WRITE_SCHEMA = vol.Schema(
 def _figure_out_source(
     record: logging.LogRecord,
     paths_re: re.Pattern[str],
-    extracted_tb: traceback.StackSummary | None = None,
+    extracted_tb: list[tuple[FrameType, int]] | None = None,
 ) -> tuple[str, int]:
     """Figure out where a log message came from."""
     # If a stack trace exists, extract file names from the entire call stack.
     # The other case is when a regular "log" is made (without an attached
     # exception). In that case, just use the file where the log was made from.
     if record.exc_info:
+        source: list[tuple[FrameType, int]] = extracted_tb or list(
+            traceback.walk_tb(record.exc_info[2])
+        )
         stack = [
-            (x[0], x[1])
+            (tb_frame.f_code.co_filename, tb_line_no) for tb_frame, tb_line_no in source
-            for x in (extracted_tb or traceback.extract_tb(record.exc_info[2]))
         ]
         for i, (filename, _) in enumerate(stack):
             # Slice the stack to the first frame that matches
@@ -176,6 +179,7 @@ class LogEntry:
         self,
         record: logging.LogRecord,
         paths_re: re.Pattern,
+        formatter: logging.Formatter | None = None,
         figure_out_source: bool = False,
     ) -> None:
         """Initialize a log entry."""
@@ -186,14 +190,21 @@ class LogEntry:
         # This must be manually tested when changing the code.
         self.message = deque([_safe_get_message(record)], maxlen=5)
         self.exception = ""
-        self.root_cause: str | None = None
+        self.root_cause: tuple[str, int, str] | None = None
-        extracted_tb: traceback.StackSummary | None = None
+        extracted_tb: list[tuple[FrameType, int]] | None = None
         if record.exc_info:
-            self.exception = "".join(traceback.format_exception(*record.exc_info))
+            if formatter and record.exc_text is None:
-            if extracted := traceback.extract_tb(record.exc_info[2]):
+                record.exc_text = formatter.formatException(record.exc_info)
+            self.exception = record.exc_text or ""
+            if extracted := list(traceback.walk_tb(record.exc_info[2])):
                 # Last line of traceback contains the root cause of the exception
                 extracted_tb = extracted
-                self.root_cause = str(extracted[-1])
+                tb_frame, tb_line_no = extracted[-1]
+                self.root_cause = (
+                    tb_frame.f_code.co_filename,
+                    tb_line_no,
+                    tb_frame.f_code.co_name,
+                )
         if figure_out_source:
             self.source = _figure_out_source(record, paths_re, extracted_tb)
         else:
@@ -273,7 +284,9 @@ class LogErrorHandler(logging.Handler):
         default upper limit is set to 50 (older entries are discarded) but can
         be changed if needed.
         """
-        entry = LogEntry(record, self.paths_re, figure_out_source=True)
+        entry = LogEntry(
+            record, self.paths_re, formatter=self.formatter, figure_out_source=True
+        )
         self.records.add_entry(entry)
         if self.fire_event:
             self.hass.bus.fire(EVENT_SYSTEM_LOG, entry.to_dict())

homeassistant/components/zha/core/gateway.py

@@ -870,7 +870,10 @@ class LogRelayHandler(logging.Handler):
     def emit(self, record: LogRecord) -> None:
         """Relay log message via dispatcher."""
         entry = LogEntry(
-            record, self.paths_re, figure_out_source=record.levelno >= logging.WARNING
+            record,
+            self.paths_re,
+            formatter=self.formatter,
+            figure_out_source=record.levelno >= logging.WARNING,
         )
         async_dispatcher_send(
             self.hass,

tests/components/system_log/test_init.py

@@ -471,10 +471,35 @@ async def test__figure_out_source(hass: HomeAssistant) -> None:
     file, line_no = system_log._figure_out_source(
         mock_record,
         paths_re,
-        traceback.extract_tb(exc_info[2]),
+        list(traceback.walk_tb(exc_info[2])),
     )
     assert file == __file__
     assert line_no != 5
 
     entry = system_log.LogEntry(mock_record, paths_re, figure_out_source=False)
     assert entry.source == ("figure_out_source is False", 5)
+
+
+async def test_formatting_exception(hass: HomeAssistant) -> None:
+    """Test that exceptions are formatted correctly."""
+    try:
+        raise ValueError("test")
+    except ValueError as ex:
+        exc_info = (type(ex), ex, ex.__traceback__)
+    mock_record = MagicMock(
+        pathname="figure_out_source is False",
+        lineno=5,
+        exc_info=exc_info,
+        exc_text=None,
+    )
+    regex_str = f"({__file__})"
+    paths_re = re.compile(regex_str)
+
+    mock_formatter = MagicMock(
+        formatException=MagicMock(return_value="formatted exception")
+    )
+    entry = system_log.LogEntry(
+        mock_record, paths_re, formatter=mock_formatter, figure_out_source=False
+    )
+    assert entry.exception == "formatted exception"
+    assert mock_record.exc_text == "formatted exception"