from __future__ import annotations import importlib.util import re from pathlib import Path from lib.contracts.missing_evidence import analyze_missing_evidence_reasons from lib.models.evaluator import AnalysisWindow, CaseReview, FinalHumanConclusion, ReplayReadyCase, SkillConclusion from lib.workflows.diagnosis_result import build_diagnosis_result _COMPARE_PATH = Path(__file__).resolve().with_name('compare.py') _COMPARE_SPEC = importlib.util.spec_from_file_location('tikv_slow_evaluation_compare_local', _COMPARE_PATH) if _COMPARE_SPEC is None or _COMPARE_SPEC.loader is None: raise RuntimeError(f'failed to load local compare module from {_COMPARE_PATH}') _COMPARE_MODULE = importlib.util.module_from_spec(_COMPARE_SPEC) _COMPARE_SPEC.loader.exec_module(_COMPARE_MODULE) compare_case_review = _COMPARE_MODULE.compare_case_review WINDOW_RANGE_RE = re.compile( r'(?P\d{4}[/-]\d{2}[/-]\d{2})\s+(?P\d{2}:\d{2})(?:[:]\d{2})?\s*(?:-|~|to|至|到)?\s*(?P\d{2}:\d{2})(?:[:]\d{2})?', re.I, ) def skill_conclusion_from_diagnosis_result(diagnosis_result: dict) -> SkillConclusion: return SkillConclusion( entry_mode=str(diagnosis_result.get('entry_mode') or ''), activated_subcapabilities=list(diagnosis_result.get('activated_subcapabilities') or []), reference_files_consulted=list(diagnosis_result.get('reference_files_consulted') or []), is_tikv_side=str(diagnosis_result.get('is_tikv_side')), scope=str(diagnosis_result.get('scope') or ''), path=str(diagnosis_result.get('path') or ''), earliest_proven_bottleneck=str(diagnosis_result.get('earliest_proven_bottleneck') or ''), key_evidence=list(diagnosis_result.get('key_evidence') or []), missing_evidence=list(diagnosis_result.get('missing_evidence') or []), uncertainty_statement=str(diagnosis_result.get('uncertainty_statement') or ''), provisional_conclusion=str(diagnosis_result.get('provisional_conclusion') or ''), ) def _derive_structured_bits_from_owner_comment(owner_final_comment: str) -> list[str]: text = re.sub(r'\s+', ' ', str(owner_final_comment or '')).strip() lowered = text.lower() if not text: return [] if any(token in lowered for token in ('disk spike', 'io jitter', 'raft engine wal sync', 'fdatasync', 'sync_log')): return [ 'Root cause: Owner-side evidence points to short-lived disk-side IO jitter, supported by Raft Engine WAL sync / sync_log fdatasync spikes.' ] if any(token in lowered for token in ('packet loss', 'network abnormal', 'network abnormalities', 'tikvrpc', 'wait connection establish', 'rpc layer latency')): return [ 'Root cause: Owner-side evidence points to TiKV communication instability, likely caused by network abnormalities or transport-layer delay.' ] if any(token in lowered for token in ('serverisbusy', 'server is busy', 'flow control', 'throttle', 'pending write', 'compaction pending', 'discard ratio', 'l0')): return [ 'Root cause: Owner-side evidence points to TiKV-side flow control or admission protection becoming active.' ] if any(token in lowered for token in ('snapshot', 'scatter', 'rebalance', 'hot leader', 'delete-range')): return [ 'Root cause: Owner-side evidence points to snapshot generation, scatter, or rebalance backlog slowing TiKV-side progress.' ] if any(token in lowered for token in ('leader drop', 'leader missing', 're-election', 'split vote', 'heartbeat')): return [ 'Root cause: Owner-side evidence points to a TiKV leader-drop style incident with delayed re-election or heartbeat progress.' ] if any( token in lowered for token in ( 'unified read pool', 'read pool 满', 'read pool满', 'thread pool 已经耗尽', 'thread pool耗尽', 'running task', 'running tasks', 'cop duration', 'coprocessor 慢', 'coprocessor slow', ) ): bits = [] if any(token in lowered for token in ('unified read pool', 'read pool 满', 'read pool满', 'thread pool 已经耗尽', 'thread pool耗尽')): bits.append('Root cause: Owner-side evidence points to unified read pool saturation or queueing on the affected TiKV instance.') if any(token in lowered for token in ('running task', 'running tasks', '堆积')): bits.append('Evidence: Running-task buildup was used as supporting proof of read-pool queueing pressure.') if any(token in lowered for token in ('handle duration 并没有变慢', 'handle duration没有变慢', 'handle duration did not slow', 'handle duration not slow')): bits.append('Evidence: Flat handle duration was used to rule out handler execution as the first bottleneck.') if bits: return bits return [] def _extract_analysis_windows( owner_final_comment: str, *, all_comment_summary: str, analysis_progression: list[str], supplementary_evidence: list[str], ) -> list[AnalysisWindow]: text_parts = [ str(owner_final_comment or '').strip(), str(all_comment_summary or '').strip(), *[str(item or '').strip() for item in analysis_progression], *[str(item or '').strip() for item in supplementary_evidence], ] text = '\n'.join(part for part in text_parts if part) lowered = text.lower() windows: list[AnalysisWindow] = [] if not text: return windows range_matches = list(WINDOW_RANGE_RE.finditer(text)) for idx, match in enumerate(range_matches, start=1): date_text = match.group('date').replace('/', '-') start_time = f'{date_text} {match.group("start")}' end_time = f'{date_text} {match.group("end")}' windows.append( AnalysisWindow( label=f'window-{idx}', start_time=start_time, end_time=end_time, summary='Maintainer-side evidence references a distinct latency window inside the same incident.', evidence=[match.group(0)], ) ) if len(windows) >= 2: if any(token in lowered for token in ('unified read pool', 'running task', 'cop duration', 'coprocessor', 'read pool 满', 'read pool满')): for window in windows: window.inferred_path = 'read_path' window.inferred_bottleneck = 'unified_read_pool_queueing' window.summary = 'This window is described as a read-path wait/queueing rise rather than a handle-first slowdown.' return windows if any(token in lowered for token in ('两次', '2次', 'two spikes', 'two rises', 'multiple windows', '多个窗口', '重复出现', 'duration 上升两次', 'duration rise twice')): summary = 'The human-side analysis explicitly treats this incident as repeated spikes rather than one blended window.' inferred_path = '' inferred_bottleneck = '' if any(token in lowered for token in ('unified read pool', 'running task', 'cop duration', 'coprocessor', 'read pool 满', 'read pool满')): inferred_path = 'read_path' inferred_bottleneck = 'unified_read_pool_queueing' summary = 'The human-side analysis treats this incident as repeated read-path queueing spikes with one shared bottleneck family.' return [ AnalysisWindow( label='window-1', summary=summary, inferred_path=inferred_path, inferred_bottleneck=inferred_bottleneck, evidence=[owner_final_comment] if owner_final_comment else [], ), AnalysisWindow( label='window-2', summary=summary, inferred_path=inferred_path, inferred_bottleneck=inferred_bottleneck, evidence=[owner_final_comment] if owner_final_comment else [], ), ] return windows def _summarize_comment_corpus(entries: list[dict], *, owner_names: set[str]) -> tuple[str, list[str], list[str], list[str]]: summary_bits: list[str] = [] progression: list[str] = [] supplementary: list[str] = [] derived_structured_bits: list[str] = [] def add_unique(target: list[str], value: str) -> None: text = str(value or '').strip() if text and text not in target: target.append(text) for entry in entries: author = str(entry.get('author') or '').strip() created = str(entry.get('created') or '').strip() body = re.sub(r'\s+', ' ', str(entry.get('body') or '')).strip() lowered = body.lower() if not body: continue who = f'{created} | {author}' if created or author else 'comment' is_owner = author.lower() in owner_names if 'slow query' in lowered or 'slow queries' in lowered or 'slow query file' in lowered: add_unique(summary_bits, 'Slow-query evidence was introduced and used as a targeted diagnosis pivot.') add_unique(progression, f'{who}: slow-query evidence was added to move diagnosis beyond coarse TiKV CPU / IO averages.') if any(token in lowered for token in ('prewrite time', 'prewrite:', 'store_batch_wait', 'persist_log', 'commit_log')): add_unique(summary_bits, 'Slow-log transaction-stage fields exposed a write-path long-tail pattern.') add_unique(progression, f'{who}: write-path stage fields such as prewrite/store_batch_wait/persist_log were used as direct evidence.') if any(token in lowered for token in ('p99.999', '0.99999', 'wal duration breakdown', 'raft engine', 'wal sync')): add_unique(summary_bits, 'Ultra-high-quantile WAL metrics were used to detect short-lived spikes that average metrics miss.') add_unique(progression, f'{who}: analysis escalated from average metrics to ultra-high-quantile WAL evidence.') if any(token in lowered for token in ('sync_log', 'fdatasync')): add_unique(summary_bits, 'Slow-log sync_log / fdatasync tails were treated as evidence of extreme storage-side latency spikes.') add_unique(progression, f'{who}: sync_log/fdatasync tails were used to confirm extreme-tail latency rather than broad p99 slowness.') add_unique(derived_structured_bits, 'Root cause: Full comment analysis points to extreme-tail store-local IO jitter, supported by sync_log / fdatasync evidence from slow logs and WAL metrics.') if any(token in lowered for token in ('store_batch_wait ≈ 0', 'tidb-layer', 'separate investigation', 'different root cause')): add_unique(summary_bits, 'The comment corpus explicitly separates TiKV-side and TiDB-side slowness windows instead of treating them as one incident.') add_unique(progression, f'{who}: analysis identified multiple windows with different root-cause families and separated TiDB-side spikes from TiKV-side stalls.') if any(token in lowered for token in ('customer provide', 'here is some updates from customer', 'from customer', 'disk health status', 'network metrics')): add_unique(supplementary, f'{who}: reporter or customer-side follow-up added host, disk, or network context.') if not is_owner and any(token in lowered for token in ('consider', 'recommended action', 'recommended actions', 'suggesting', 'warrants a separate investigation')): add_unique(supplementary, f'{who}: non-owner analytical comment added actionable follow-up hypotheses.') summary = ' '.join(summary_bits) return summary, progression, supplementary, derived_structured_bits def build_skill_conclusion_from_replay_ready_case( case: ReplayReadyCase, *, context_is_leader_drop, context_is_network_issue, context_is_flow_control, ) -> SkillConclusion: canonical = case.canonical_incident context = { 'title': canonical.title, 'problem_statement': canonical.description_text, 'phenomenon': case.initial_symptom, 'routing_context_text': '\n'.join( [ canonical.title, canonical.description_text, case.initial_symptom, ] ), 'clinic_references': [{'link': link} for link in canonical.clinic_links], 'grafana_references': [{'link': link} for link in canonical.grafana_links], } missing_evidence_reasons = analyze_missing_evidence_reasons(context, []) diagnosis_result = build_diagnosis_result( context, clinic_results=[], grafana_results=[], bundle_results=[], missing_evidence_reasons=missing_evidence_reasons, context_is_leader_drop=context_is_leader_drop, context_is_network_issue=context_is_network_issue, context_is_flow_control=context_is_flow_control, ).to_dict() return skill_conclusion_from_diagnosis_result(diagnosis_result) def apply_live_diagnosis_result_to_case_review( review: CaseReview, diagnosis_result: dict, *, source_note: str, ) -> CaseReview: review.skill_conclusion = skill_conclusion_from_diagnosis_result(diagnosis_result) review.match_level = '' review.mismatch_type = [] review.improvement_suggestions = [] review.comparison_notes = [ note for note in (review.comparison_notes or []) if not ( note.startswith('Skill conclusion was generated from replay-boundary inputs only.') or note.startswith('Inferred human ') or note.startswith('Structured compare found') ) ] if source_note not in review.comparison_notes: review.comparison_notes.insert(0, source_note) return compare_case_review(review) def apply_maintainer_evidence_hints_to_case_review( review: CaseReview, *, grafana_links: list[str], window_specs: list[dict], source_note: str, ) -> CaseReview: final_human = review.final_human_conclusion for link in grafana_links: text = str(link or '').strip() if not text: continue note = f'Maintainer evidence link: {text}' if note not in final_human.supplementary_evidence: final_human.supplementary_evidence.append(note) existing_windows = list(final_human.analysis_windows or []) seen_keys = { ( str(item.label or '').strip(), str(item.start_time or '').strip(), str(item.end_time or '').strip(), str(item.inferred_bottleneck or '').strip(), ) for item in existing_windows } for item in window_specs: label = str(item.get('label') or '').strip() start_time = str(item.get('start_time') or '').strip() end_time = str(item.get('end_time') or '').strip() inferred_path = str(item.get('inferred_path') or '').strip() inferred_bottleneck = str(item.get('inferred_bottleneck') or '').strip() summary = str(item.get('summary') or '').strip() evidence = [str(x or '').strip() for x in (item.get('evidence') or []) if str(x or '').strip()] key = (label, start_time, end_time, inferred_bottleneck) if key in seen_keys: continue seen_keys.add(key) existing_windows.append( AnalysisWindow( label=label, start_time=start_time, end_time=end_time, summary=summary, inferred_path=inferred_path, inferred_bottleneck=inferred_bottleneck, evidence=evidence, ) ) final_human.analysis_windows = existing_windows if source_note and source_note not in review.comparison_notes: review.comparison_notes.insert(0, source_note) return compare_case_review(review) def build_final_human_conclusion_for_jira( raw: dict, *, extract_comment_entries_with_time, author_set, is_owner_comment, clean_line, summarize_owner_comments, ) -> FinalHumanConclusion: fields = (raw or {}).get('fields') or {} owners = author_set(fields) owner_entries = [] all_entries = extract_comment_entries_with_time(raw) for entry in all_entries: if is_owner_comment(entry.get('author', ''), owners): body = clean_line(entry.get('body', '')) if not body: continue owner_entries.append( { 'author': str(entry.get('author') or '').strip(), 'created': str(entry.get('created') or '').strip(), 'body': body, } ) all_comment_summary, progression, supplementary, derived_structured_bits = _summarize_comment_corpus( all_entries, owner_names=owners, ) if not owner_entries: return FinalHumanConclusion( status='human_conclusion_unclear', owner_final_comment='', structured_conclusion=' '.join(derived_structured_bits), evidence_references=[], all_comment_summary=all_comment_summary, analysis_progression=progression, supplementary_evidence=supplementary, analysis_windows=[], ) latest_entry = owner_entries[-1] owner_final_comment = latest_entry['body'] root_cause, workaround, todo = summarize_owner_comments(owner_final_comment) structured_bits = [] if root_cause and root_cause != 'TBD': structured_bits.append(f'Root cause: {root_cause}') if workaround and workaround != 'None yet': structured_bits.append(f'Workaround: {workaround}') if todo and todo != 'TBD': structured_bits.append(f'TODO: {todo}') if not structured_bits: structured_bits.extend(_derive_structured_bits_from_owner_comment(owner_final_comment)) for bit in derived_structured_bits: if bit not in structured_bits: structured_bits.append(bit) status = 'clear' if structured_bits else 'human_conclusion_unclear' evidence_refs = [ f'{latest_entry["created"]} | {latest_entry["author"]}' if latest_entry.get('created') or latest_entry.get('author') else '' ] evidence_refs = [item for item in evidence_refs if item] analysis_windows = _extract_analysis_windows( owner_final_comment, all_comment_summary=all_comment_summary, analysis_progression=progression, supplementary_evidence=supplementary, ) return FinalHumanConclusion( status=status, owner_final_comment=owner_final_comment, structured_conclusion=' '.join(structured_bits), evidence_references=evidence_refs, all_comment_summary=all_comment_summary, analysis_progression=progression, supplementary_evidence=supplementary, analysis_windows=analysis_windows, ) def build_final_human_conclusion_for_pingkai( ticket: dict, *, pingkai_owner_comment_entries, clean_line, extract_pingkai_owner_conclusions, ) -> FinalHumanConclusion: owner_entries = [] corpus_entries = list(pingkai_owner_comment_entries(ticket)) for item in corpus_entries: body = clean_line(item.get('body', '')) if not body: continue owner_entries.append( { 'author': str(item.get('author_name') or '').strip(), 'created': str(item.get('created') or '').strip(), 'body': body, } ) normalized_owner_names = {str(item.get('author') or '').strip().lower() for item in owner_entries} all_comment_summary, progression, supplementary, derived_structured_bits = _summarize_comment_corpus( [ { 'author': item.get('author_name') or '', 'created': item.get('created') or '', 'body': clean_line(item.get('body', '')), } for item in corpus_entries ], owner_names=normalized_owner_names, ) if not owner_entries: return FinalHumanConclusion( status='human_conclusion_unclear', owner_final_comment='', structured_conclusion=' '.join(derived_structured_bits), evidence_references=[], all_comment_summary=all_comment_summary, analysis_progression=progression, supplementary_evidence=supplementary, analysis_windows=[], ) latest_entry = owner_entries[-1] owner_final_comment = latest_entry['body'] root_cause, workaround, todo = extract_pingkai_owner_conclusions([latest_entry]) structured_bits = [] if root_cause and root_cause != 'TBD': structured_bits.append(f'Root cause: {root_cause}') if workaround and workaround != 'None yet': structured_bits.append(f'Workaround: {workaround}') if todo and todo != 'TBD': structured_bits.append(f'TODO: {todo}') if not structured_bits: structured_bits.extend(_derive_structured_bits_from_owner_comment(owner_final_comment)) for bit in derived_structured_bits: if bit not in structured_bits: structured_bits.append(bit) status = 'clear' if structured_bits else 'human_conclusion_unclear' evidence_refs = [ f'{latest_entry["created"]} | {latest_entry["author"]}' if latest_entry.get('created') or latest_entry.get('author') else '' ] evidence_refs = [item for item in evidence_refs if item] analysis_windows = _extract_analysis_windows( owner_final_comment, all_comment_summary=all_comment_summary, analysis_progression=progression, supplementary_evidence=supplementary, ) return FinalHumanConclusion( status=status, owner_final_comment=owner_final_comment, structured_conclusion=' '.join(structured_bits), evidence_references=evidence_refs, all_comment_summary=all_comment_summary, analysis_progression=progression, supplementary_evidence=supplementary, analysis_windows=analysis_windows, ) def build_case_review_from_jira_case( case: ReplayReadyCase, raw: dict, *, extract_comment_entries_with_time, author_set, is_owner_comment, clean_line, summarize_owner_comments, context_is_leader_drop, context_is_network_issue, context_is_flow_control, ) -> CaseReview: skill_conclusion = build_skill_conclusion_from_replay_ready_case( case, context_is_leader_drop=context_is_leader_drop, context_is_network_issue=context_is_network_issue, context_is_flow_control=context_is_flow_control, ) final_human = build_final_human_conclusion_for_jira( raw, extract_comment_entries_with_time=extract_comment_entries_with_time, author_set=author_set, is_owner_comment=is_owner_comment, clean_line=clean_line, summarize_owner_comments=summarize_owner_comments, ) comparison_notes = [ 'Skill conclusion was generated from replay-boundary inputs only.', 'Final human conclusion was extracted from owner-side comments.', ] match_level = '' if final_human.status == 'human_conclusion_unclear': match_level = 'human_conclusion_unclear' comparison_notes.append('Comparison is blocked because the final owner conclusion is not explicit enough yet.') review = CaseReview( incident_id=case.incident_id, source=case.source, initial_symptom=case.initial_symptom, canonical_incident=case.canonical_incident, replay_inputs=case.replay_inputs, skill_conclusion=skill_conclusion, final_human_conclusion=final_human, match_level=match_level, mismatch_type=[], comparison_notes=comparison_notes, improvement_suggestions=[], ) return compare_case_review(review) def build_case_review_from_pingkai_case( case: ReplayReadyCase, ticket: dict, *, pingkai_owner_comment_entries, clean_line, extract_pingkai_owner_conclusions, context_is_leader_drop, context_is_network_issue, context_is_flow_control, ) -> CaseReview: skill_conclusion = build_skill_conclusion_from_replay_ready_case( case, context_is_leader_drop=context_is_leader_drop, context_is_network_issue=context_is_network_issue, context_is_flow_control=context_is_flow_control, ) final_human = build_final_human_conclusion_for_pingkai( ticket, pingkai_owner_comment_entries=pingkai_owner_comment_entries, clean_line=clean_line, extract_pingkai_owner_conclusions=extract_pingkai_owner_conclusions, ) comparison_notes = [ 'Skill conclusion was generated from replay-boundary inputs only.', 'Final human conclusion was extracted from owner-side Pingkai comments.', ] match_level = '' if final_human.status == 'human_conclusion_unclear': match_level = 'human_conclusion_unclear' comparison_notes.append('Comparison is blocked because the final owner conclusion is not explicit enough yet.') review = CaseReview( incident_id=case.incident_id, source=case.source, initial_symptom=case.initial_symptom, canonical_incident=case.canonical_incident, replay_inputs=case.replay_inputs, skill_conclusion=skill_conclusion, final_human_conclusion=final_human, match_level=match_level, mismatch_type=[], comparison_notes=comparison_notes, improvement_suggestions=[], ) return compare_case_review(review)