已合并
mstx data support combine multi data of same mark_id #346
mei-feiyao创建于 7月8日
mstx data support combine multi data of same mark_id #346
已合并
共 3 个文件变更+69-39
| @@ -39,11 +39,12 @@ class MsprofTxParser(IParser, MsMultiProcess): | |||
| 39 | """ | 39 | """ |
| 40 | parsing MsprofTx data class | 40 | parsing MsprofTx data class |
| 41 | """ | 41 | """ |
| 42 | + | ||
| 42 | EVENT_DICT = { | 43 | EVENT_DICT = { |
| 43 | NumberConstant.MARKER: 'marker', | 44 | NumberConstant.MARKER: 'marker', |
| 44 | NumberConstant.PUSH_AND_POP: 'push/pop', | 45 | NumberConstant.PUSH_AND_POP: 'push/pop', |
| 45 | NumberConstant.START_AND_END: 'start/end', | 46 | NumberConstant.START_AND_END: 'start/end', |
| 46 | - NumberConstant.MARKER_EX: 'marker_ex' | 47 | + NumberConstant.MARKER_EX: 'marker_ex', |
| 47 | } | 48 | } |
| 48 | TX_INFO_TYPE = 0 | 49 | TX_INFO_TYPE = 0 |
| 49 | TX_EX_INFO_TYPE = 1 | 50 | TX_EX_INFO_TYPE = 1 |
| @@ -54,7 +55,7 @@ class MsprofTxParser(IParser, MsMultiProcess): | |||
| 54 | self._project_path = sample_config.get(StrConstant.SAMPLE_CONFIG_PROJECT_PATH, '') | 55 | self._project_path = sample_config.get(StrConstant.SAMPLE_CONFIG_PROJECT_PATH, '') |
| 55 | self._cur_file_list = [] | 56 | self._cur_file_list = [] |
| 56 | self._msproftx_data = [] | 57 | self._msproftx_data = [] |
| 57 | - self._msproftx_ex_data = [] | 58 | + self._msproftx_ex_data_dict = {} |
| 58 | 59 | ||
| 59 | def parse(self: any) -> None: | 60 | def parse(self: any) -> None: |
| 60 | """ | 61 | """ |
| @@ -75,13 +76,46 @@ class MsprofTxParser(IParser, MsMultiProcess): | |||
| 75 | :return: None | 76 | :return: None |
| 76 | """ | 77 | """ |
| 77 | if self._msproftx_data: | 78 | if self._msproftx_data: |
| 78 | - with MsprofTxModel(self._project_path, DBNameConstant.DB_MSPROFTX, | 79 | + with MsprofTxModel( |
| 79 | - [DBNameConstant.TABLE_MSPROFTX]) as tx_model: | 80 | + self._project_path, DBNameConstant.DB_MSPROFTX, [DBNameConstant.TABLE_MSPROFTX] |
| 81 | + ) as tx_model: | ||
| 80 | tx_model.flush(self._msproftx_data) | 82 | tx_model.flush(self._msproftx_data) |
| 81 | - if self._msproftx_ex_data: | 83 | + msproftx_ex_data = [] |
| 82 | - with MsprofTxExModel(self._project_path, DBNameConstant.DB_MSPROFTX, | 84 | + hash_data = HashDictData(self._project_path).get_ge_hash_dict() |
| 83 | - [DBNameConstant.TABLE_MSPROFTX_EX]) as tx_ex_model: | 85 | + if self._msproftx_ex_data_dict: |
| 84 | - tx_ex_model.flush(self._msproftx_ex_data) | 86 | + for mark_id, obj_list in self._msproftx_ex_data_dict.items(): |
| 87 | + obj_list.sort(key=lambda x: x.seg_idx) # sort by seg_idx | ||
| 88 | + seg_all = [obj.seg_idx for obj in obj_list] | ||
| 89 | + | ||
| 90 | + # seg_idx should be continuous and start from 0, if not, log a warning | ||
| 91 | + missing_segs = set(range(0, obj_list[-1].seg_idx + 1)) - set(seg_all) | ||
| 92 | + if missing_segs: | ||
| 93 | + logging.warning( | ||
| 94 | + "mark_id=%s segment lost, missing seg index: %s, received seg indices: %s", | ||
| 95 | + mark_id, | ||
| 96 | + missing_segs, | ||
| 97 | + seg_all, | ||
| 98 | + ) | ||
| 99 | + | ||
| 100 | + full_msg = "".join(obj.message for obj in obj_list) # concatenate message by seg_idx | ||
| 101 | + first_obj = obj_list[0] | ||
| 102 | + domain_str = str(first_obj.domain) | ||
| 103 | + final_row = ( | ||
| 104 | + first_obj.process_id, | ||
| 105 | + first_obj.thread_id, | ||
| 106 | + self.EVENT_DICT.get(first_obj.event_type, ''), | ||
| 107 | + first_obj.start_time, | ||
| 108 | + first_obj.end_time, | ||
| 109 | + mark_id, | ||
| 110 | + hash_data.get(domain_str, 'invalid'), | ||
| 111 | + full_msg, | ||
| 112 | + ) | ||
| 113 | + msproftx_ex_data.append(final_row) | ||
| 114 | + if msproftx_ex_data: | ||
| 115 | + with MsprofTxExModel( | ||
| 116 | + self._project_path, DBNameConstant.DB_MSPROFTX, [DBNameConstant.TABLE_MSPROFTX_EX] | ||
| 117 | + ) as tx_ex_model: | ||
| 118 | + tx_ex_model.flush(msproftx_ex_data) | ||
| 85 | 119 | ||
| 86 | def ms_run(self: any) -> None: | 120 | def ms_run(self: any) -> None: |
| 87 | """ | 121 | """ |
| @@ -100,27 +134,31 @@ class MsprofTxParser(IParser, MsMultiProcess): | |||
| 100 | """ | 134 | """ |
| 101 | parsing msproftx data and insert into msproftx.db | 135 | parsing msproftx data and insert into msproftx.db |
| 102 | """ | 136 | """ |
| 103 | - hash_data = HashDictData(self._project_path).get_ge_hash_dict() | 137 | + calculate = OffsetCalculator(self._cur_file_list, StructFmt.MSPROFTX_FMT_SIZE, self._project_path) |
| 104 | - calculate = OffsetCalculator(self._cur_file_list, StructFmt.MSPROFTX_FMT_SIZE, | ||
| 105 | - self._project_path) | ||
| 106 | msproftx_file = PathManager.get_data_file_path(self._project_path, file_name) | 138 | msproftx_file = PathManager.get_data_file_path(self._project_path, file_name) |
| 107 | with FileOpen(msproftx_file, 'rb') as msproftx_f: | 139 | with FileOpen(msproftx_file, 'rb') as msproftx_f: |
| 108 | msproftx_data = calculate.pre_process(msproftx_f.file_reader, os.path.getsize(msproftx_file)) | 140 | msproftx_data = calculate.pre_process(msproftx_f.file_reader, os.path.getsize(msproftx_file)) |
| 109 | for chunk in Utils.chunks(msproftx_data, StructFmt.MSPROFTX_FMT_SIZE): | 141 | for chunk in Utils.chunks(msproftx_data, StructFmt.MSPROFTX_FMT_SIZE): |
| 110 | data_object = MsprofTxDecoder.decode(chunk) | 142 | data_object = MsprofTxDecoder.decode(chunk) |
| 111 | if data_object.info_type == MsprofTxParser.TX_INFO_TYPE: | 143 | if data_object.info_type == MsprofTxParser.TX_INFO_TYPE: |
| 112 | - self._msproftx_data.append((data_object.process_id, data_object.thread_id, | 144 | + self._msproftx_data.append( |
| 113 | - data_object.category, self.EVENT_DICT.get(data_object.event_type, ''), | 145 | + ( |
| 114 | - data_object.payload_type, data_object.payload_value, | 146 | + data_object.process_id, |
| 115 | - data_object.start_time, data_object.end_time, | 147 | + data_object.thread_id, |
| 116 | - data_object.message_type, | 148 | + data_object.category, |
| 117 | - data_object.message)) | 149 | + self.EVENT_DICT.get(data_object.event_type, ''), |
| 150 | + data_object.payload_type, | ||
| 151 | + data_object.payload_value, | ||
| 152 | + data_object.start_time, | ||
| 153 | + data_object.end_time, | ||
| 154 | + data_object.message_type, | ||
| 155 | + data_object.message, | ||
| 156 | + ) | ||
| 157 | + ) | ||
| 118 | elif data_object.info_type == MsprofTxParser.TX_EX_INFO_TYPE: | 158 | elif data_object.info_type == MsprofTxParser.TX_EX_INFO_TYPE: |
| 119 | - domain_str = str(data_object.domain) | 159 | + mark_id = data_object.mark_id |
| 120 | - self._msproftx_ex_data.append((data_object.process_id, data_object.thread_id, | 160 | + if mark_id not in self._msproftx_ex_data_dict: |
| 121 | - self.EVENT_DICT.get(data_object.event_type, ''), | 161 | + self._msproftx_ex_data_dict[mark_id] = [] |
| 122 | - data_object.start_time, data_object.end_time, | 162 | + self._msproftx_ex_data_dict[mark_id].append(data_object) |
| 123 | - data_object.mark_id, hash_data.get(domain_str, 'invalid'), | ||
| 124 | - data_object.message)) | ||
| 125 | else: | 163 | else: |
| 126 | logging.error("Invalid info_type: %d", data_object.info_type) | 164 | logging.error("Invalid info_type: %d", data_object.info_type) |
| @@ -14,7 +14,6 @@ | |||
| 14 | # See the Mulan PSL v2 for more details. | 14 | # See the Mulan PSL v2 for more details. |
| 15 | # ------------------------------------------------------------------------- | 15 | # ------------------------------------------------------------------------- |
| 16 | 16 | ||
| 17 | -from common_func.constant import Constant | ||
| 18 | 17 | ||
| 19 | from profiling_bean.struct_info.struct_decoder import StructDecoder | 18 | from profiling_bean.struct_info.struct_decoder import StructDecoder |
| 20 | 19 | ||
| @@ -28,6 +27,7 @@ class MsprofTxDecoder(StructDecoder): | |||
| 28 | filed = args[0] | 27 | filed = args[0] |
| 29 | self._magic = filed[0] | 28 | self._magic = filed[0] |
| 30 | self._info_type = filed[6] | 29 | self._info_type = filed[6] |
| 30 | + self._seg_idx = filed[7] | ||
| 31 | self._data_tag = filed[10] | 31 | self._data_tag = filed[10] |
| 32 | self._process_id = filed[11] | 32 | self._process_id = filed[11] |
| 33 | self._thread_id = filed[12] | 33 | self._thread_id = filed[12] |
| @@ -50,6 +50,10 @@ class MsprofTxDecoder(StructDecoder): | |||
| 50 | def info_type(self: any) -> int: | 50 | def info_type(self: any) -> int: |
| 51 | return self._info_type | 51 | return self._info_type |
| 52 | 52 | ||
| 53 | + | ||
| 54 | + def seg_idx(self: any) -> int: | ||
| 55 | + return self._seg_idx | ||
| 56 | + | ||
| 53 | 57 | ||
| 54 | def data_tag(self: any) -> int: | 58 | def data_tag(self: any) -> int: |
| 55 | return self._data_tag | 59 | return self._data_tag |
| @@ -86,28 +86,16 @@ class TestMsprofTxParser(unittest.TestCase): | |||
| 86 | check = MsprofTxParser(self.file_list, self.CONFIG) | 86 | check = MsprofTxParser(self.file_list, self.CONFIG) |
| 87 | check.parse() | 87 | check.parse() |
| 88 | self.assertFalse(check._msproftx_data) | 88 | self.assertFalse(check._msproftx_data) |
| 89 | - self.assertFalse(check._msproftx_ex_data) | 89 | + self.assertFalse(check._msproftx_ex_data_dict) |
| 90 | 90 | ||
| 91 | - def test_parse_should_parse_success_when_open_file_success(self): | 91 | + def test_parse_should_run_success_when_open_file_success(self): |
| 92 | with mock.patch(NAMESPACE + '.logging.error'): | 92 | with mock.patch(NAMESPACE + '.logging.error'): |
| 93 | InfoConfReader()._info_json = {'devices': '0'} | 93 | InfoConfReader()._info_json = {'devices': '0'} |
| 94 | check = MsprofTxParser(self.file_list, self.CONFIG) | 94 | check = MsprofTxParser(self.file_list, self.CONFIG) |
| 95 | check.parse() | 95 | check.parse() |
| 96 | self.assertTrue(check._msproftx_data) | 96 | self.assertTrue(check._msproftx_data) |
| 97 | - self.assertTrue(check._msproftx_ex_data) | 97 | + self.assertTrue(check._msproftx_ex_data_dict) |
| 98 | - | ||
| 99 | - def test_save_with_msproftx_data(self): | ||
| 100 | - with mock.patch('msmodel.msproftx.msproftx_model.MsprofTxModel.flush'): | ||
| 101 | - InfoConfReader()._info_json = {"devices": '0'} | ||
| 102 | - check = MsprofTxParser(self.file_list, self.CONFIG) | ||
| 103 | - check._msproftx_data = [123] | ||
| 104 | check.save() | 98 | check.save() |
| 105 | - | ||
| 106 | - def test_save_with_msproftx_data(self): | ||
| 107 | - with mock.patch('msmodel.msproftx.msproftx_model.MsprofTxExModel.flush'): | ||
| 108 | - InfoConfReader()._info_json = {"devices": '0'} | ||
| 109 | - check = MsprofTxParser(self.file_list, self.CONFIG) | ||
| 110 | - check._msproftx_ex_data = [123] | ||
| 111 | check.save() | 99 | check.save() |
 | |||
| 112 | 100 | ||
| 113 | def test_ms_run(self): | 101 | def test_ms_run(self): |
🟡 Medium Priority
变更后的
test_parse_should_run_success_when_open_file_success（第 91-99 行）调用了check.save()，但未 mockHashDictData、MsprofTxModel和MsprofTxExModel。旧代码中原本有独立的test_save_with_msproftx_data测试方法，通过 mockMsprofTxModel.flush和MsprofTxExModel.flush来隔离 DB 操作。现在save()直接执行真实的 DB 写入逻辑，其行为依赖运行时文件系统状态（host/sqlite/目录是否存在、sqlite3 是否可用等），导致测试脆弱，可能因环境差异而失败。建议：在
test_parse_should_run_success_when_open_file_success中 mockHashDictData、MsprofTxModel.flush和MsprofTxExModel.flush，保持测试对 DB 层的隔离。或者将 save 调用恢复为独立的、带 mock 的测试方法。