Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/collect/log/rc_log_collect.py: 88%

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 

20 

21from params import ParamDict 

22from common import consts 

23from common import log_warning 

24from common import FileOperate as f 

25from common.file_operate import COPY_MODE 

26from common.path import get_log_conf_path 

27 

28__all__ = ["collect_rc_logs"] 

29 

30 

31def collect_messages(output_root_dir): 

32 message_root_dir = os.path.join(os.sep, "var", "log") 

33 message_file_name = "syslog" if f.check_file(os.path.join(message_root_dir, "syslog")) else "messages" 

34 message_source = os.path.join(message_root_dir, message_file_name) 

35 message_target = os.path.join(output_root_dir, "dfx", "log", "message") 

36 return f.collect_file_to_dir(message_source, message_target, COPY_MODE) 

37 

38 

39def collect_stackcore(output_root_dir): 

40 coredump_path = os.path.join(os.sep, "var", "log", "npu", "coredump") 

41 stackcore_target = os.path.join(output_root_dir, "dfx", "stackcore") 

42 ret = True 

43 if not f.list_dir(coredump_path): 

44 log_warning("No files or dirs in {0}".format(coredump_path)) 

45 return False 

46 stackcore_source_dirs = os.walk(coredump_path) 

47 for stackcore_path, _, stackcore_files in stackcore_source_dirs: 

48 for stackcore_file in stackcore_files: 

49 stackcore_source_name = os.path.join(stackcore_path, stackcore_file) 

50 if ParamDict().get_command() == consts.launch_cmd and \ 

51 f".{str(ParamDict().get_task_pid())}." not in stackcore_file: 

52 continue 

53 ret = ret and f.collect_file_to_dir(stackcore_source_name, 

54 stackcore_path.replace(coredump_path, stackcore_target), COPY_MODE) 

55 return ret 

56 

57 

58def collect_install(output_root_dir): 

59 install_source = os.path.join(os.sep, "var", "log", "ascend_seclog", "ascend_install.log") 

60 install_target = os.path.join(output_root_dir, "dfx", "log", "install") 

61 return f.collect_file_to_dir(install_source, install_target, COPY_MODE) 

62 

63 

64def collect_bbox(output_root_dir): 

65 bbox_source = get_log_conf_path("bbox") 

66 bbox_target = os.path.join(output_root_dir, "dfx", "bbox") 

67 return f.collect_dir(bbox_source, bbox_target, COPY_MODE) 

68 

69 

70def collect_device_os(slog_path, output_root_dir): 

71 ret = True 

72 log_types = ["debug", "run", "security"] 

73 task_pid = ParamDict().get_task_pid() 

74 for log_type in log_types: 

75 dev_os_path = os.path.join(slog_path, log_type) 

76 device_os_dirs = f.list_dir(dev_os_path) 

77 if not device_os_dirs: 

78 log_warning("No files or dirs in {0}".format(dev_os_path)) 

79 return False 

80 for device_os_dir_name in device_os_dirs: 

81 if device_os_dir_name == "device-os": 

82 device_os_log_source = os.path.join(slog_path, log_type, device_os_dir_name) 

83 device_os_log_target = os.path.join(output_root_dir, "dfx", "log", "system", 

84 log_type, device_os_dir_name) 

85 ret = ret and f.collect_dir(device_os_log_source, device_os_log_target, COPY_MODE) 

86 # collect: task_pid is None; launch: task_pid is subprocess pid. 

87 elif (task_pid is None and device_os_dir_name.startswith("device-app")) or \ 

88 device_os_dir_name.endswith(str(task_pid)): 

89 device_os_log_source = os.path.join(slog_path, log_type, device_os_dir_name) 

90 device_os_log_target = os.path.join(output_root_dir, "dfx", "log", "cann", 

91 log_type, device_os_dir_name) 

92 ret = ret and f.collect_dir(device_os_log_source, device_os_log_target, COPY_MODE) 

93 return ret 

94 

95 

96def collect_device_id(slog_path, output_root_dir): 

97 dev_os_dirs = f.list_dir(slog_path) 

98 if not dev_os_dirs: 

99 log_warning("No files or dirs in {0}".format(slog_path)) 

100 return False 

101 ret = True 

102 for dev_os_id in dev_os_dirs: 

103 if "device-" not in dev_os_id: 

104 continue 

105 device_id_log_source = os.path.join(slog_path, dev_os_id) 

106 device_id_log_target = os.path.join(output_root_dir, "dfx", "log", "firmware", dev_os_id) 

107 ret = ret and f.collect_dir(device_id_log_source, device_id_log_target, COPY_MODE) 

108 return ret 

109 

110 

111def collect_rc_logs(output_root_dir): 

112 all_func = [collect_messages, collect_install, collect_bbox, collect_stackcore] 

113 for func in all_func: 

114 ret = func(output_root_dir) 

115 if not ret: 

116 msg_name = func.__name__.split('_')[-1] 

117 log_warning("Collect {} logs failed.".format(msg_name)) 

118 

119 slog_path = get_log_conf_path("slog") 

120 device_os_ret = collect_device_os(slog_path, output_root_dir) 

121 if not device_os_ret: 

122 log_warning("Collect device device_os logs failed.") 

123 device_id_ret = collect_device_id(slog_path, output_root_dir) 

124 if not device_id_ret: 

125 log_warning("Collect device device_id logs failed.")