Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/collect/ops/ops_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 json 

20import os.path 

21import re 

22 

23from params import ParamDict 

24from common import consts 

25from common import log_debug, log_info, log_warning 

26from common import FileOperate as f 

27from common.file_operate import COPY_MODE, MOVE_MODE 

28from drv import EnvVarName 

29 

30__all__ = ["collect_ops"] 

31 

32 

33def get_sk_kernel_name(plog): 

34 """ 

35 Obtains kernelName from the SuperKernel scenario marker print in plog. 

36 """ 

37 sk_marker = "Begin to dump callback exception" 

38 cmd_ret = os.popen(f"grep '{sk_marker}' -inrE {plog}") 

39 sk_lines = cmd_ret.readlines() 

40 cmd_ret.close() 

41 if not sk_lines: 

42 return None 

43 sk_regexp = r"kernelName=([^\n]*?)\.\s*$" 

44 for line in sk_lines: 

45 sk_ret = re.findall(sk_regexp, line.strip()) 

46 if sk_ret: 

47 return sk_ret[0] 

48 return None 

49 

50 

51def is_sk_scenario(output_root_path): 

52 """ 

53 Determine whether the current scenario is SK (super kernel) by the marker 

54 print in plog. SK is judged by the marker only, independent of whether the 

55 device-side files are generated. 

56 """ 

57 sk_marker = "Begin to dump callback exception" 

58 run_plog = os.path.join(output_root_path, "dfx", "log", "host", "cann", "run", "plog") 

59 debug_plog = os.path.join(output_root_path, "dfx", "log", "host", "cann", "debug", "plog") 

60 for plog in [run_plog, debug_plog]: 

61 if not f.check_dir(plog): 

62 continue 

63 cmd_ret = os.popen(f"grep '{sk_marker}' -inrE {plog}") 

64 sk_lines = cmd_ret.readlines() 

65 cmd_ret.close() 

66 if sk_lines: 

67 return True 

68 return False 

69 

70 

71def get_fault_kernel_name(output_root_path): 

72 """ 

73 Obtains fault_kernel_name from plog. 

74 """ 

75 error_msg = ["Aicore kernel execute failed|AI Core kernel execution failed", "fftsplus task execute failed"] 

76 run_plog = os.path.join(output_root_path, "dfx", "log", "host", "cann", "run", "plog") 

77 debug_plog = os.path.join(output_root_path, "dfx", "log", "host", "cann", "debug", "plog") 

78 plog_files = [run_plog, debug_plog] 

79 for plog in plog_files: 

80 if not f.check_dir(plog): 

81 continue 

82 # 优先查找SK场景标志性打印，命中则该打印中的kernelName才是正确的算子名 

83 sk_kernel_name = get_sk_kernel_name(plog) 

84 if sk_kernel_name: 

85 return sk_kernel_name 

86 plog_lines = [] 

87 for msg in error_msg: 

88 cmd_ret = os.popen(f"grep '{msg}' -inrE {plog}") 

89 plog_lines += cmd_ret.readlines() 

90 cmd_ret.close() 

91 if len(plog_lines) == 0: 

92 continue 

93 

94 static_regexp = r" stream_id=\d+,.*?task_id=\d+,.*?fault kernel_name=.*?,.*?" \ 

95 r" fault kernel info ext=(.*?)," 

96 dynamic_regexp = r" stream_id=\d+,.*?task_id=\d+,.*?fault kernel_name=(.*?)," 

97 

98 for regexp in [static_regexp, dynamic_regexp]: 

99 kernel_name_ret = re.findall(regexp, plog_lines[0]) 

100 if not kernel_name_ret or kernel_name_ret[0] == 'none': 

101 continue 

102 kernel_name = kernel_name_ret[0] 

103 return kernel_name.replace("_mix_aic", "").replace("_mix_aiv", "") 

104 return None 

105 

106 

107def get_all_kernel_name_from_file(file_path): 

108 """ 

109 read JSON file and check whether they contain kernel_name. 

110 """ 

111 all_kernel_name = [] 

112 try: 

113 with open(file_path, 'r') as json_file: 

114 dict_obj = json.load(json_file) 

115 all_kernel_name.append(dict_obj.get("binFileName")) 

116 all_kernel_name.append(dict_obj.get("kernelName")) 

117 all_kernel_name += [kernel.get("kernelName") for kernel in dict_obj.get("kernelList", [])] 

118 except Exception as e: 

119 log_debug(f"Failed to load the '{file_path}', {e}") 

120 return all_kernel_name 

121 

122 # remove its 'None' elements 

123 return list(filter(None, all_kernel_name)) 

124 

125 

126def get_fault_kernel_name_files(collect_path, kernel_name): 

127 """ 

128 Obtain the .o.json file corresponding to fault_kernel_name. 

129 """ 

130 collect_files = [] 

131 if not kernel_name: 

132 return collect_files 

133 opp_path = EnvVarName().opp_path 

134 for path, _, files in os.walk(collect_path): 

135 for file in files: 

136 file_path = os.path.join(path, file) 

137 # ASCEND_OPP_PATH only needs to read files in the '/kernel/'. 

138 if not file.endswith(".json") or (collect_path == opp_path and "kernel" not in path.split("/")): 

139 continue 

140 # read all JSON files and check whether they contain kernel_name. 

141 if kernel_name not in get_all_kernel_name_from_file(file_path): 

142 continue 

143 # collect the .json file and .o file that contain kernel_name. 

144 collect_files.append(file_path) 

145 o_file_path = os.path.join(path, file.replace(".json", ".o")) 

146 if os.path.isfile(o_file_path): 

147 collect_files.append(o_file_path) 

148 

149 return collect_files 

150 

151 

152def collect_op_files(ops_res, target_dir, mode=COPY_MODE): 

153 """ 

154 Collect the .o.json file corresponding to fault_kernel_name. 

155 """ 

156 ret = True 

157 for file_path in ops_res: 

158 op_file_ret = f.collect_file_to_dir(file_path, target_dir, mode) 

159 ret = ret and op_file_ret 

160 return ret 

161 

162 

163def collect_ops_from_dump(output_root_path): 

164 """ 

165 Collect ops files from the dump directory. 

166 """ 

167 target_dir = os.path.join(output_root_path, "dfx", "ops") 

168 dump_path = os.path.join(output_root_path, "dfx", "data-dump") 

169 if not f.check_dir(dump_path): 

170 return False 

171 ops_files = [] 

172 for path, _, files in os.walk(dump_path): 

173 for file in files: 

174 if file.endswith(".o") or file.endswith(".json"): 

175 ops_files.append(os.path.join(path, file)) 

176 if ops_files: 

177 return collect_op_files(ops_files, target_dir, MOVE_MODE) 

178 

179 return False 

180 

181 

182def collect_ops_files_env_var(output_root_path, ops_target_dir): 

183 

184 collect_path_list = [] 

185 task_dir = ParamDict().get_arg("task_dir") 

186 # ops files priority: NPU_COLLECT_PATH > ASCEND_CACHE_PATH > ASCEND_WORK_PATH > $HOME/atc_data > 

187 # ASCEND_CUSTOM_OPP_PATH > ASCEND_OPP_PATH > ./ 

188 env_var = EnvVarName() 

189 for collect_path in [task_dir, env_var.npu_collect_path, env_var.cache_path, env_var.work_path, 

190 os.path.join(env_var.home_path, "atc_data"), env_var.custom_opp_path, env_var.opp_path, 

191 env_var.current_path]: 

192 if collect_path and f.check_dir(collect_path): 

193 collect_path_list.append(collect_path) 

194 

195 kernel_name = get_fault_kernel_name(output_root_path) 

196 for path in collect_path_list: 

197 collect_files = get_fault_kernel_name_files(path, kernel_name) 

198 if collect_files: 

199 return collect_op_files(collect_files, ops_target_dir, COPY_MODE) 

200 

201 log_warning("The JSON file of the fault kernel_name is not found.") 

202 return False 

203 

204 

205def check_launch_ops(): 

206 if (ParamDict().get_command() == consts.launch_cmd) and (not ParamDict().get_ini("ops") == "1"): 

207 log_debug("ops is not set on, not collect ops files") 

208 return False 

209 return True 

210 

211 

212def collect_debug_kernel(output_root_path): 

213 ops_target_dir = os.path.join(output_root_path, "dfx", "ops") 

214 opp_path = EnvVarName().opp_path 

215 if opp_path is None: 

216 log_debug("ASCEND_OPP_PATH not set") 

217 return 

218 debug_kernel_path = os.path.join(opp_path, "debug_kernel") 

219 if debug_kernel_path and f.check_access(debug_kernel_path) and f.check_dir(debug_kernel_path): 

220 if f.list_dir(debug_kernel_path): 

221 debug_kernel_target_path = os.path.join(ops_target_dir, os.path.basename(debug_kernel_path)) 

222 if debug_kernel_target_path.startswith(debug_kernel_path): 

223 log_debug("Cannot copy debug_kernel to %s" % debug_kernel_target_path) 

224 else: 

225 f.copy_dir(debug_kernel_path, debug_kernel_target_path) 

226 

227 

228def collect_file(output_root_path): 

229 

230 ops_target_dir = os.path.join(output_root_path, "dfx", "ops") 

231 ret = False 

232 if ParamDict().get_command() == consts.launch_cmd: 

233 ops_source_dir = os.path.join( 

234 ParamDict().asys_output_timestamp_dir, "npu_collect_intermediates", "extra-info", "ops") 

235 if f.check_dir(ops_source_dir): 

236 ret = f.collect_dir(ops_source_dir, ops_target_dir, MOVE_MODE) 

237 else: 

238 ret = collect_ops_files_env_var(output_root_path, ops_target_dir) 

239 if not ret: 

240 log_warning("Ops collect failed.") 

241 

242 

243def collect_cfg_json(output_root_path, cfg_dir, json_dir, config): 

244 if not os.path.exists(json_dir): 

245 return False 

246 

247 ret = True 

248 for path, _, files in os.walk(json_dir): 

249 if "/config/" not in path: 

250 continue 

251 for file in files: 

252 if not file.endswith(".json"): 

253 continue 

254 dst_dir = os.path.join(output_root_path, "dfx", "ops", config, cfg_dir, os.path.relpath(path, json_dir)) 

255 if not os.path.exists(dst_dir): 

256 os.makedirs(dst_dir) 

257 ret = ret and f.copy_file_to_dir(os.path.join(path, file), dst_dir) 

258 return ret 

259 

260 

261def collect_opp_config(output_root_path): 

262 opp_path = EnvVarName().opp_path 

263 if opp_path is None: 

264 log_debug("ASCEND_OPP_PATH is not set.") 

265 return False 

266 

267 config_path = os.path.join(opp_path, "vendors", "config.ini") 

268 if not os.path.isfile(config_path): 

269 log_debug(f"The {config_path} is not a file.") 

270 return False 

271 try: 

272 with open(config_path, 'r') as cfg: 

273 cfg_content = cfg.read() 

274 except PermissionError: 

275 log_warning(f"The {config_path} file does not have the read permission.") 

276 return False 

277 

278 load_priority = re.search("load_priority=(.+?)\n", cfg_content) 

279 if not load_priority: 

280 log_warning(f"The {config_path} file does not contain the load_priority field.") 

281 return False 

282 load_priority = load_priority.group(1).split(",") 

283 ret = True 

284 for cfg_dir in load_priority: 

285 # remove front and back spaces 

286 _cfg_dir = cfg_dir.strip() 

287 json_dir = os.path.join(opp_path, "vendors", _cfg_dir) 

288 ret = ret and collect_cfg_json(output_root_path, _cfg_dir, json_dir, "vendor_config") 

289 

290 ret = ret and f.copy_file_to_dir(config_path, os.path.join(output_root_path, "dfx", "ops", "vendor_config")) 

291 return ret 

292 

293 

294def collect_custom_opp_config(output_root_path): 

295 custom_opp_path = EnvVarName().custom_opp_path 

296 if custom_opp_path is None: 

297 log_debug("ASCEND_CUSTOM_OPP_PATH is not set.") 

298 return False 

299 return collect_cfg_json(output_root_path, "", custom_opp_path, "custom_config") 

300 

301 

302def collect_ops(output_root_path): 

303 if not check_launch_ops(): 

304 return 

305 

306 # SK场景下只生成host.o，没有device .o/.json，跳过算子文件收集，其余配置类收集保持不变 

307 if is_sk_scenario(output_root_path): 

308 log_info("SuperKernel scenario detected, skip collecting operator files.") 

309 elif not collect_ops_from_dump(output_root_path): 

310 collect_file(output_root_path) 

311 collect_debug_kernel(output_root_path) 

312 collect_opp_config(output_root_path) 

313 collect_custom_opp_config(output_root_path)