Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/collect/coretrace/coretrace_collect.py: 95%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ---------------------------------------------------------------------------- 

4# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

5# 

6# Licensed under the Apache License, Version 2.0 (the "License"); 

7# you may not use this file except in compliance with the License. 

8# You may obtain a copy of the License at 

9# 

10# http://www.apache.org/licenses/LICENSE-2.0 

11# 

12# Unless required by applicable law or agreed to in writing, software 

13# distributed under the License is distributed on an "AS IS" BASIS, 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

15# See the License for the specific language governing permissions and 

16# limitations under the License. 

17# ---------------------------------------------------------------------------- 

18 

19import os 

20import subprocess 

21import sys 

22from threading import Thread, Lock 

23 

24from common import FileOperate as f 

25from common import log_error, log_warning 

26from common.cmd_run import check_command, run_linux_cmd 

27from common.task_common import out_progress_bar, str_to_hex, is_hexadecimal 

28from common.const import ADDR_LEN_HEX 

29 

30 

31class ParseData: 

32 def __init__(self): 

33 self.maps = {} 

34 self.sig = 0 

35 self.pid = -1 

36 self.tgid = -1 

37 self.comm = "" 

38 

39 

40class ParseCoreTrace: 

41 missing_binary = set() 

42 lock = Lock() 

43 

44 def __init__(self, symbol, file=None): 

45 self.file = file 

46 self.symbol_path = symbol 

47 self.__addr2line = "addr2line" 

48 self.warned = False 

49 

50 def check_tool_exists(self): 

51 if not check_command(self.__addr2line): 

52 log_error("The addr2line tool does not exist, install it before using it.") 

53 return False 

54 return True 

55 

56 def warn_missing(self, binary_path): 

57 with self.lock: 

58 if binary_path not in self.missing_binary: 

59 log_warning(f"{os.path.realpath(binary_path)} is not exists.") 

60 self.missing_binary.add(binary_path) 

61 

62 def get_binary_path(self, bin_name_path): 

63 bin_name_str = bin_name_path.split("/")[-1] 

64 if self.symbol_path: 

65 for path in self.symbol_path: 

66 binary_path = os.path.join(path, bin_name_str) 

67 if os.path.exists(binary_path): 

68 return binary_path 

69 self.warn_missing(binary_path) 

70 elif not os.path.exists(bin_name_path): 

71 self.warn_missing(bin_name_path) 

72 return bin_name_path 

73 

74 def run_addr2line(self, cmd): 

75 ret = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8') 

76 return ret.stdout.readlines() 

77 

78 def parse_addr_src_line(self, fp, data, shift): 

79 parsed_line = '' 

80 for bin_name, addrs in data.maps.items(): 

81 if bin_name == "": 

82 continue 

83 bin_path = self.get_binary_path(bin_name) 

84 start_addr, end_addr = addrs 

85 if start_addr < fp < end_addr: 

86 delta = hex(fp - start_addr - shift) 

87 cmd = [self.__addr2line, "-Cifps", "-e", bin_path, "-a", delta] 

88 try: 

89 out = self.run_addr2line(cmd) 

90 except (OSError, ValueError) as e: 

91 if not self.warned: 

92 log_warning(f"Run \"{' '.join(cmd)}\" failed, error detail: {e}") 

93 self.warned = True 

94 out = [] 

95 if len(out) == 0: 

96 parsed_line += "0x%x %s %s" % (fp, delta, bin_name) + '\n' 

97 else: 

98 for func in out: 

99 parsed_line += "0x%x %s %s" % (fp, func.strip(), bin_name.strip()) + '\n' 

100 return parsed_line 

101 return parsed_line 

102 

103 def parse_line(self, line, parse_data): 

104 line_parts = line.split() 

105 this_line = line.strip() + '\n' 

106 try: 

107 if line_parts[0] == "Signal": 

108 parse_data.sig = int(line_parts[1]) 

109 return this_line 

110 elif line_parts[0] == "PID": 

111 parse_data.pid = int(line_parts[1]) 

112 parse_data.tgid = int(line_parts[3]) 

113 parse_data.comm = line_parts[5] 

114 return '\n' + this_line 

115 elif line_parts[0].startswith("#"): 

116 shift = 0 if line_parts[0] == "#0" else 4 

117 fp = int(line_parts[1].strip('\x00'), base=16) 

118 return self.parse_addr_src_line(fp, parse_data, shift) 

119 elif line_parts[0] == "[<0>]" or "(deleted)" in line: 

120 return this_line 

121 elif "uburma" in line or "davinci_manager" in line: 

122 return '' 

123 else: 

124 start_addr, end_addr = map(lambda x: int(x, base=16), line_parts[0].split("-")) 

125 bin_name = line_parts[1].strip('\x00') 

126 if bin_name in parse_data.maps: 

127 start_addr = min(parse_data.maps[bin_name][0], start_addr) 

128 end_addr = max(parse_data.maps[bin_name][1], end_addr) 

129 parse_data.maps[bin_name] = [start_addr, end_addr] 

130 return '' 

131 except (IndexError, ValueError): 

132 return this_line 

133 

134 def parse_file(self, file_lines, count): 

135 if self.file: 

136 count = len(file_lines) 

137 parse_data = ParseData() 

138 parsed_lines = '' 

139 for index, line in enumerate(file_lines): 

140 if self.file: 

141 out_progress_bar(count, index) 

142 parsed_lines += self.parse_line(line, parse_data) 

143 return parsed_lines 

144 

145 def start_parse_file(self, coretrace_file, count=0): 

146 """Parsing a single file""" 

147 coretrace_file_name = coretrace_file.split(os.sep)[-1] 

148 if not coretrace_file_name.startswith("coretrace"): 

149 log_error(f"The {coretrace_file} file is not in coretrace format.") 

150 return False 

151 # Check whether the addr2line tools exist. 

152 if not self.check_tool_exists(): 

153 return False 

154 with open(coretrace_file, "r") as fp: 

155 file_lines = fp.readlines() 

156 if not file_lines: 

157 log_error(f"The {coretrace_file_name} file is empty.") 

158 return False 

159 

160 try: 

161 parsed_lines = self.parse_file(file_lines, count) 

162 except Exception as e: 

163 log_error(f"Parse {coretrace_file} failed, error detail: {e}") 

164 return False 

165 

166 with open(coretrace_file, 'w') as fw: 

167 fw.writelines(parsed_lines) 

168 return True 

169 

170 def save_file_result(self, coretrace_file, count, num, results): 

171 ret = self.start_parse_file(coretrace_file, count) 

172 out_progress_bar(count, num) 

173 if not ret: 

174 log_error(f'Failed to analyze the "{coretrace_file}" file.') 

175 results.append(ret) 

176 

177 def run(self, coretrace_path, count=0): 

178 coretrace_dirs = f.walk_dir(coretrace_path) 

179 if not coretrace_dirs or not self.check_tool_exists: 

180 return False 

181 num = 0 

182 threads = [] 

183 results = [] 

184 for dirs, _, files in coretrace_dirs: 

185 for file in files: 

186 coretrace_file = os.path.join(dirs, file) 

187 num += 1 

188 t = Thread(target=self.save_file_result, args=(coretrace_file, count, num, results), daemon=True) 

189 t.start() 

190 threads.append(t) 

191 # wait for all threads to end. 

192 for t in threads: 

193 t.join() 

194 out_progress_bar(count, count) 

195 return any(results)