xref: /aosp_15_r20/external/angle/build/util/lib/proto/exception_recorder.py (revision 8975f5c5ed3d1c378011245431ada316dfb6f244) 
1# Copyright 2024 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4""" The module to create and manage ExceptionOccurrence records. """
5
6import os
7import json
8import traceback
9
10from abc import ABC, abstractmethod
11from typing import List
12
13from google.protobuf import any_pb2
14from google.protobuf.json_format import MessageToDict
15
16from lib.proto.exception_occurrences_pb2 import ExceptionOccurrence
17from lib.proto.exception_occurrences_pb2 import ExceptionOccurrences
18
19# This is used as the key when being uploaded to ResultDB via result_sink
20# and shouldn't be changed
21EXCEPTION_OCCURRENCES_KEY = 'exception_occurrences'
22
23# This is used as the key when being uploaded to ResultDB via rdb
24# and shouldn't be changed
25EXCEPTION_OCCURRENCES_FILENAME = f'{EXCEPTION_OCCURRENCES_KEY}.jsonpb'
26
27_records: List[ExceptionOccurrence] = []
28
29
30class Formatter(ABC):
31
32  @abstractmethod
33  def format_name(self, exc: Exception) -> str:
34    """Format the exception name."""
35
36  @abstractmethod
37  def format_stacktrace(self, exc: Exception) -> List[str]:
38    """Format the exception stacktrace."""
39
40
41class _Formatter(Formatter):
42
43  def format_name(self, exc: Exception) -> str:
44    exc_name = type(exc).__qualname__
45    exc_module = type(exc).__module__
46    if exc_module not in ('__main__', 'builtins'):
47      exc_name = '%s.%s' % (exc_module, exc_name)
48    return exc_name
49
50  def format_stacktrace(self, exc: Exception) -> List[str]:
51    return traceback.format_exception(type(exc), exc, exc.__traceback__)
52
53
54# Default formatter
55_default_formatter = _Formatter()
56
57def _record_time(exc: ExceptionOccurrence):
58  exc.occurred_time.GetCurrentTime()
59
60def register(exc: Exception,
61             formatter: Formatter = _default_formatter) -> ExceptionOccurrence:
62  """Create and register an ExceptionOccurrence record."""
63  ret = ExceptionOccurrence(name=formatter.format_name(exc),
64                            stacktrace=formatter.format_stacktrace(exc))
65  _record_time(ret)
66  _records.append(ret)
67  return ret
68
69
70def size() -> int:
71  """Get the current size of registered ExceptionOccurrence records."""
72  return len(_records)
73
74
75def clear() -> None:
76  """Clear all the registered ExceptionOccurrence records."""
77  _records.clear()
78
79
80def clear_stacktrace() -> None:
81  """Clear the stacktrace from all the records while keeping the records.
82
83  This can be called to reduce the size of the overall records and avoid
84  the size issue when uploaded to other services, e.g. RDB.
85  """
86  for record in _records:
87    record.ClearField('stacktrace')
88
89
90def to_dict() -> dict:
91  """Convert all the registered ExceptionOccurrence records to an dict.
92
93  The records are wrapped in protobuf Any message before exported as dict
94  so that an additional key "@type" is included.
95  """
96  occurrences = ExceptionOccurrences()
97  occurrences.datapoints.extend(_records)
98  any_msg = any_pb2.Any()
99  any_msg.Pack(occurrences)
100  return MessageToDict(any_msg, preserving_proto_field_name=True)
101
102
103def to_json() -> str:
104  """Convert all the registered ExceptionOccurrence records to a json str."""
105  return json.dumps(to_dict(), sort_keys=True, indent=2)
106
107
108def dump(dir_path: str) -> None:
109  """Dumps the records into |EXCEPTION_OCCURRENCES_FILENAME| in the |path|."""
110  os.makedirs(dir_path, exist_ok=True)
111  with open(os.path.join(dir_path, EXCEPTION_OCCURRENCES_FILENAME),
112            'w',
113            encoding='utf-8') as wf:
114    wf.write(to_json())
115