Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/collect/ops/ops_collect.py: 93%
180 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-08-21 15:37 +0800
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# ----------------------------------------------------------------------------
19import json
20import os.path
21import re
23from params import ParamDict
24from common import consts
25from common import log_debug, log_warning
26from common import FileOperate as f
27from common.file_operate import COPY_MODE, MOVE_MODE
28from drv import EnvVarName
30__all__ = ["collect_ops"]
33def get_fault_kernel_name(output_root_path):
34 """
35 Obtains fault_kernel_name from plog.
36 """
37 error_msg = ["Aicore kernel execute failed|AI Core kernel execution failed", "fftsplus task execute failed"]
38 run_plog = os.path.join(output_root_path, "dfx", "log", "host", "cann", "run", "plog")
39 debug_plog = os.path.join(output_root_path, "dfx", "log", "host", "cann", "debug", "plog")
40 plog_files = [run_plog, debug_plog]
41 for plog in plog_files:
42 if not f.check_dir(plog):
43 continue
44 plog_lines = []
45 for msg in error_msg:
46 cmd_ret = os.popen(f"grep '{msg}' -inrE {plog}")
47 plog_lines += cmd_ret.readlines()
48 cmd_ret.close()
49 if len(plog_lines) == 0:
50 continue
52 static_regexp = r" stream_id=\d+,.*?task_id=\d+,.*?fault kernel_name=.*?,.*?" \
53 r" fault kernel info ext=(.*?),"
54 dynamic_regexp = r" stream_id=\d+,.*?task_id=\d+,.*?fault kernel_name=(.*?),"
56 for regexp in [static_regexp, dynamic_regexp]:
57 kernel_name_ret = re.findall(regexp, plog_lines[0])
58 if not kernel_name_ret or kernel_name_ret[0] == 'none':
59 continue
60 kernel_name = kernel_name_ret[0]
61 return kernel_name.replace("_mix_aic", "").replace("_mix_aiv", "")
62 return None
65def get_all_kernel_name_from_file(file_path):
66 """
67 read JSON file and check whether they contain kernel_name.
68 """
69 all_kernel_name = []
70 try:
71 with open(file_path, 'r') as json_file:
72 dict_obj = json.load(json_file)
73 all_kernel_name.append(dict_obj.get("binFileName"))
74 all_kernel_name.append(dict_obj.get("kernelName"))
75 all_kernel_name += [kernel.get("kernelName") for kernel in dict_obj.get("kernelList", [])]
76 except Exception as e:
77 log_debug(f"Failed to load the '{file_path}', {e}")
78 return all_kernel_name
80 # remove its 'None' elements
81 return list(filter(None, all_kernel_name))
84def get_fault_kernel_name_files(collect_path, kernel_name):
85 """
86 Obtain the .o.json file corresponding to fault_kernel_name.
87 """
88 collect_files = []
89 if not kernel_name:
90 return collect_files
91 opp_path = EnvVarName().opp_path
92 for path, _, files in os.walk(collect_path):
93 for file in files:
94 file_path = os.path.join(path, file)
95 # ASCEND_OPP_PATH only needs to read files in the '/kernel/'.
96 if not file.endswith(".json") or (collect_path == opp_path and "kernel" not in path.split("/")):
97 continue
98 # read all JSON files and check whether they contain kernel_name.
99 if kernel_name not in get_all_kernel_name_from_file(file_path):
100 continue
101 # collect the .json file and .o file that contain kernel_name.
102 collect_files.append(file_path)
103 o_file_path = os.path.join(path, file.replace(".json", ".o"))
104 if os.path.isfile(o_file_path):
105 collect_files.append(o_file_path)
107 return collect_files
110def collect_op_files(ops_res, target_dir, mode=COPY_MODE):
111 """
112 Collect the .o.json file corresponding to fault_kernel_name.
113 """
114 ret = True
115 for file_path in ops_res:
116 op_file_ret = f.collect_file_to_dir(file_path, target_dir, mode)
117 ret = ret and op_file_ret
118 return ret
121def collect_ops_from_dump(output_root_path):
122 """
123 Collect ops files from the dump directory.
124 """
125 target_dir = os.path.join(output_root_path, "dfx", "ops")
126 dump_path = os.path.join(output_root_path, "dfx", "data-dump")
127 if not f.check_dir(dump_path):
128 return False
129 ops_files = []
130 for path, _, files in os.walk(dump_path):
131 for file in files:
132 if file.endswith(".o") or file.endswith(".json"):
133 ops_files.append(os.path.join(path, file))
134 if len(ops_files) >= 3: # two .o, one .json
135 return collect_op_files(ops_files, target_dir, MOVE_MODE)
137 return False
140def collect_ops_files_env_var(output_root_path, ops_target_dir):
142 collect_path_list = []
143 task_dir = ParamDict().get_arg("task_dir")
144 # ops files priority: NPU_COLLECT_PATH > ASCEND_CACHE_PATH > ASCEND_WORK_PATH > $HOME/atc_data >
145 # ASCEND_CUSTOM_OPP_PATH > ASCEND_OPP_PATH > ./
146 env_var = EnvVarName()
147 for collect_path in [task_dir, env_var.npu_collect_path, env_var.cache_path, env_var.work_path,
148 os.path.join(env_var.home_path, "atc_data"), env_var.custom_opp_path, env_var.opp_path,
149 env_var.current_path]:
150 if collect_path and f.check_dir(collect_path):
151 collect_path_list.append(collect_path)
153 kernel_name = get_fault_kernel_name(output_root_path)
154 for path in collect_path_list:
155 collect_files = get_fault_kernel_name_files(path, kernel_name)
156 if collect_files:
157 return collect_op_files(collect_files, ops_target_dir, COPY_MODE)
159 log_warning("The JSON file of the fault kernel_name is not found.")
160 return False
163def check_launch_ops():
164 if (ParamDict().get_command() == consts.launch_cmd) and (not ParamDict().get_ini("ops") == "1"):
165 log_debug("ops is not set on, not collect ops files")
166 return False
167 return True
170def collect_debug_kernel(output_root_path):
171 ops_target_dir = os.path.join(output_root_path, "dfx", "ops")
172 opp_path = EnvVarName().opp_path
173 if opp_path is None:
174 log_debug("ASCEND_OPP_PATH not set")
175 return
176 debug_kernel_path = os.path.join(opp_path, "debug_kernel")
177 if debug_kernel_path and f.check_access(debug_kernel_path) and f.check_dir(debug_kernel_path):
178 if f.list_dir(debug_kernel_path):
179 debug_kernel_target_path = os.path.join(ops_target_dir, os.path.basename(debug_kernel_path))
180 if debug_kernel_target_path.startswith(debug_kernel_path):
181 log_debug("Cannot copy debug_kernel to %s" % debug_kernel_target_path)
182 else:
183 f.copy_dir(debug_kernel_path, debug_kernel_target_path)
186def collect_file(output_root_path):
188 ops_target_dir = os.path.join(output_root_path, "dfx", "ops")
189 ret = False
190 if ParamDict().get_command() == consts.launch_cmd:
191 ops_source_dir = os.path.join(
192 ParamDict().asys_output_timestamp_dir, "npu_collect_intermediates", "extra-info", "ops")
193 if f.check_dir(ops_source_dir):
194 ret = f.collect_dir(ops_source_dir, ops_target_dir, MOVE_MODE)
195 else:
196 ret = collect_ops_files_env_var(output_root_path, ops_target_dir)
197 if not ret:
198 log_warning("Ops collect failed.")
201def collect_cfg_json(output_root_path, cfg_dir, json_dir, config):
202 if not os.path.exists(json_dir):
203 return False
205 ret = True
206 for path, _, files in os.walk(json_dir):
207 if "/config/" not in path:
208 continue
209 for file in files:
210 if not file.endswith(".json"):
211 continue
212 dst_dir = os.path.join(output_root_path, "dfx", "ops", config, cfg_dir, os.path.relpath(path, json_dir))
213 if not os.path.exists(dst_dir):
214 os.makedirs(dst_dir)
215 ret = ret and f.copy_file_to_dir(os.path.join(path, file), dst_dir)
216 return ret
219def collect_opp_config(output_root_path):
220 opp_path = EnvVarName().opp_path
221 if opp_path is None:
222 log_debug("ASCEND_OPP_PATH is not set.")
223 return False
225 config_path = os.path.join(opp_path, "vendors", "config.ini")
226 if not os.path.isfile(config_path):
227 log_debug(f"The {config_path} is not a file.")
228 return False
229 try:
230 with open(config_path, 'r') as cfg:
231 cfg_content = cfg.read()
232 except PermissionError:
233 log_warning(f"The {config_path} file does not have the read permission.")
234 return False
236 load_priority = re.search("load_priority=(.+?)\n", cfg_content)
237 if not load_priority:
238 log_warning(f"The {config_path} file does not contain the load_priority field.")
239 return False
240 load_priority = load_priority.group(1).split(",")
241 ret = True
242 for cfg_dir in load_priority:
243 # remove front and back spaces
244 _cfg_dir = cfg_dir.strip()
245 json_dir = os.path.join(opp_path, "vendors", _cfg_dir)
246 ret = ret and collect_cfg_json(output_root_path, _cfg_dir, json_dir, "vendor_config")
248 ret = ret and f.copy_file_to_dir(config_path, os.path.join(output_root_path, "dfx", "ops", "vendor_config"))
249 return ret
252def collect_custom_opp_config(output_root_path):
253 custom_opp_path = EnvVarName().custom_opp_path
254 if custom_opp_path is None:
255 log_debug("ASCEND_CUSTOM_OPP_PATH is not set.")
256 return False
257 return collect_cfg_json(output_root_path, "", custom_opp_path, "custom_config")
260def collect_ops(output_root_path):
261 if not check_launch_ops():
262 return
264 if not collect_ops_from_dump(output_root_path):
265 collect_file(output_root_path)
266 collect_debug_kernel(output_root_path)
267 collect_opp_config(output_root_path)
268 collect_custom_opp_config(output_root_path)