from __future__ import annotations import re from lib.contracts.evaluate_tikv_slow import ALLOWED_REFERENCE_FILES from lib.contracts.missing_evidence import ( REASON_CLINIC_DATA_NOT_READY, REASON_GRAFANA_RELATIVE_WINDOW, REASON_LOWER_LAYER_EVIDENCE_MISSING, REASON_PER_HOP_EVIDENCE_MISSING, REASON_PER_STORE_EVIDENCE_MISSING, ) from lib.models.diagnosis_result import DiagnosisResult def _joined_text(*parts) -> str: return '\n'.join(str(part or '') for part in parts if str(part or '').strip()) def _collect_clinic_text(clinic_results: list[dict]) -> str: bits = [] for item in clinic_results or []: if item.get('error'): continue deep_rca = item.get('tikv_write_path_evidence') or {} metrics = item.get('metrics_evidence') or {} summary = str(deep_rca.get('summary') or '').strip() if summary: bits.append(summary) for entry in (((metrics.get('performance_overview_summary') or {}).get('items')) or [])[:6]: judgement = str((entry or {}).get('judgement') or '').strip() if judgement: bits.append(judgement) incident_window = deep_rca.get('incident_window') or {} if incident_window: bits.append(_joined_text(incident_window.get('from'), incident_window.get('to'))) return '\n'.join(bits) def _collect_attachment_text(attachment_results: list[dict]) -> str: bits = [] for item in attachment_results or []: if item.get('error'): continue analysis = item.get('analysis') or {} slow_log_summary = analysis.get('slow_log_summary') or {} bits.extend(str(line) for line in (slow_log_summary.get('summary_lines') or []) if str(line).strip()) for stage in (slow_log_summary.get('top_stages') or [])[:5]: bits.append( f'{stage.get("name")}: max={stage.get("max_seconds")} p95={stage.get("p95_seconds")}' ) bits.extend(str(line) for line in (analysis.get('suspicious_lines') or [])[:5] if str(line).strip()) return '\n'.join(bits) def _contains_any(text: str, tokens: tuple[str, ...]) -> bool: return any(token in text for token in tokens) def _infer_read_path_queueing_signal(text: str) -> dict[str, object]: lowered = text.lower() read_pool_tokens = ( 'unified read pool', 'unifed read pool', 'read pool 满', 'read pool满', 'thread pool 已经耗尽', 'thread pool耗尽', ) queueing_tokens = ( 'running task', 'running tasks', 'cop wait', 'schedule wait', 'time used by level', 'time used level', 'queueing', '堆积', ) flat_handle_tokens = ( 'handle duration 并没有变慢', 'handle duration没有变慢', 'handle duration not slow', 'handle duration did not slow', 'handle flat', 'task exec stay flat', 'task exec flat', ) localized_cpu_tokens = ( 'cpu usage by name', 'ti kv 进程 cpu 消耗并不高', 'tikv 进程 cpu 消耗并不高', 'process cpu stayed below quota', 'process cpu below quota', 'cpu 并不高', 'cpu quota', ) has_pool_pressure = _contains_any(lowered, read_pool_tokens) has_queueing = _contains_any(lowered, queueing_tokens) has_flat_handle = _contains_any(lowered, flat_handle_tokens) has_localized_cpu_hint = _contains_any(lowered, localized_cpu_tokens) has_store_local_scope = bool(re.search(r'\b\d{1,3}(?:\.\d{1,3}){3}:\d+\b', text)) or _contains_any( lowered, ('这个 tikv', 'this tikv', 'affected tikv', '慢在 tikv'), ) if not has_pool_pressure: return {} bottleneck = 'read_pool_pressure' key_evidence: list[str] = [] if has_queueing: bottleneck = 'unified_read_pool_queueing' key_evidence.append( 'Ticket/comment evidence points to unified read pool pressure together with running-task or wait buildup before execution.' ) else: key_evidence.append( 'Ticket/comment evidence points to unified read pool pressure on the affected TiKV instance.' ) if has_flat_handle: bottleneck = 'unified_read_pool_queueing' key_evidence.append( 'Comment text explicitly says handle duration stayed flat, so the first bottleneck is more likely queueing before execution than handler work.' ) if has_localized_cpu_hint: if bottleneck == 'unified_read_pool_queueing': bottleneck = 'localized_read_pool_saturation' key_evidence.append( 'Comment text suggests TiKV process CPU stayed below quota, so prefer localized read-pool saturation over instance-wide CPU saturation.' ) return { 'bottleneck': bottleneck, 'key_evidence': key_evidence, 'scope': 'store_subset_or_hotspot' if has_store_local_scope else None, } def json_item_summary(value) -> str: try: return str(value or '') except Exception: return '' def _collect_evidence_text( *, clinic_results: list[dict], grafana_results: list[dict], attachment_results: list[dict], ) -> str: clinic_text = _collect_clinic_text(clinic_results) grafana_text = _joined_text( *[ _joined_text( ((item.get('cluster_summary') or {}).get('cluster_name')), json_item_summary(item.get('dashboard_discovery') or {}), ) for item in grafana_results or [] if not item.get('error') ] ) attachment_text = _collect_attachment_text(attachment_results) return _joined_text(clinic_text, grafana_text, attachment_text) def _route_reference_files(entry_mode: str, lowered_text: str) -> tuple[list[str], str]: if entry_mode == 'leader_drop': return ['references/cross-cutting.md', 'references/leader-drop.md'], 'write_path' if entry_mode == 'flow_control': return ['references/cross-cutting.md', 'references/flow-control.md'], 'read_or_write_path' if entry_mode == 'io_jitter': return ['references/cross-cutting.md', 'references/io-jitter.md'], 'store_local_path' if entry_mode == 'network_issues': return ['references/cross-cutting.md', 'references/network-issues.md'], 'network_path' if entry_mode == 'snapshot_backlog': return ['references/cross-cutting.md', 'references/snapshot-backlog.md'], 'snapshot_path' refs = ['references/cross-cutting.md'] path = 'unknown' if any(token in lowered_text for token in ('prewrite', 'commit', 'write hotspot', 'rocksdb', 'apply')): path = 'write_path' refs.append('references/write-path.md') elif any(token in lowered_text for token in ('cop', 'batchget', 'read', 'scan')): path = 'read_path' refs.append('references/read-path.md') return refs, path def _route_evidence_assessment( entry_mode: str, *, clinic_results: list[dict], grafana_results: list[dict], attachment_results: list[dict], ) -> dict[str, object]: evidence_text = _collect_evidence_text( clinic_results=clinic_results, grafana_results=grafana_results, attachment_results=attachment_results, ) lowered = evidence_text.lower() route_profiles = { 'flow_control': { 'support_tokens': ( 'serverisbusy', 'server is busy', 'deadline exceeded', 'throttle duration', 'scheduler discard ratio', 'discard ratio', 'pending write', 'compaction pending', 'memtable', 'l0', 'flow control', 'throttle', 'admission', 'reject', 'rejection', '限流', '流控', ), 'fallback_checks': ( ( 'network_issues', ( 'pdrpc', 'pd rpc', 'get leader failed', "leader address doesn't exist", 'cannot update member', 'requested pd is not leader', 'errclientgetleader', 'errclientgettso', 'get tso failed', 'tso rpc', 'pd leader', '连不上 pd', 'pd client', ), ), ), }, 'network_issues': { 'support_tokens': ( 'tikvrpc', 'batch receive', 'wait connection establish', 'rpc layer latency', 'network jitter', 'network latency', 'packet loss', 'transport latency', 'unreachable', 'tso rpc', 'pdrpc', 'pd client', ), 'fallback_checks': ( ( 'io_jitter', ('sync_log', 'persist_log', 'write_wal', 'fdatasync', 'disk latency', 'wal sync', 'store-local', 'node-local'), ), ), }, 'io_jitter': { 'support_tokens': ( 'io jitter', 'disk spike', 'disk latency', 'fdatasync', 'sync_log', 'persist_log', 'write_wal', 'rocksdb kv write', 'rocksdb background compaction', 'store-local', 'node-local', 'wal sync', ), 'fallback_checks': ( ( 'network_issues', ('tikvrpc', 'tso rpc', 'pdrpc', 'transport latency', 'network jitter', 'wait connection establish'), ), ), }, 'leader_drop': { 'support_tokens': ( 'leader drop', 'leader missing', 're-election', 'split vote', 'heartbeat delay', 'evict-leader', 'evict-slow-store', 'not leader', ), 'fallback_checks': ( ( 'snapshot_backlog', ('snapshot', 'scatter', 'rebalance', 'delete-range', 'hot leader'), ), ( 'network_issues', ('tso rpc', 'pdrpc', 'pd client', 'transport latency', 'network jitter'), ), ), }, 'snapshot_backlog': { 'support_tokens': ( 'snapshot', 'scatter', 'rebalance', 'hot leader', 'delete-range', 'generate snapshot', 'snapshot generation', ), 'fallback_checks': ( ( 'io_jitter', ('sync_log', 'persist_log', 'write_wal', 'fdatasync', 'disk latency'), ), ( 'network_issues', ('transport latency', 'packet loss', 'network jitter', 'tso rpc', 'pdrpc'), ), ), }, } profile = route_profiles.get(entry_mode) if not profile: return {'supported': True, 'support_hits': [], 'contradiction_hits': [], 'fallback_entry_mode': ''} support_hits = [token for token in profile.get('support_tokens', ()) if token in lowered] if support_hits: return { 'supported': True, 'support_hits': support_hits, 'contradiction_hits': [], 'fallback_entry_mode': '', } contradiction_hits: list[str] = [] fallback_entry_mode = 'generic_tikv_slow' for fallback_mode, tokens in profile.get('fallback_checks', ()): hits = [token for token in tokens if token in lowered] if hits: fallback_entry_mode = fallback_mode contradiction_hits = hits break return { 'supported': False, 'support_hits': [], 'contradiction_hits': contradiction_hits, 'fallback_entry_mode': fallback_entry_mode, } def _route_scores( context: dict, *, clinic_results: list[dict], attachment_results: list[dict], context_is_leader_drop, context_is_network_issue, context_is_flow_control, ) -> dict[str, int]: context_text = _joined_text( context.get('title'), context.get('problem_statement'), context.get('phenomenon'), context.get('routing_context_text'), ).lower() clinic_text = _collect_clinic_text(clinic_results).lower() attachment_text = _collect_attachment_text(attachment_results).lower() merged = '\n'.join([context_text, clinic_text, attachment_text]) scores = { 'leader_drop': 0, 'network_issues': 0, 'flow_control': 0, 'snapshot_backlog': 0, 'io_jitter': 0, } if context_is_leader_drop(context): scores['leader_drop'] += 6 if context_is_flow_control(context): scores['flow_control'] += 5 if context_is_network_issue(context): scores['network_issues'] += 3 token_groups = { 'leader_drop': ( 'leader drop', 'leader missing', 're-election', 'split vote', 'heartbeat', ), 'network_issues': ( 'tikvrpc', 'batch receive', 'wait connection establish', 'rpc layer latency', 'network jitter', 'network latency', 'packet loss', 'transport latency', 'unreachable', ), 'flow_control': ( 'serverisbusy', 'server is busy', 'flow control', 'throttle', 'deadline exceeded', 'discard ratio', 'pending write', 'compaction pending', 'memtable', 'l0', ), 'snapshot_backlog': ( 'snapshot', 'scatter', 'rebalance', 'hot leader', 'delete-range', ), 'io_jitter': ( 'io jitter', 'disk spike', 'disk latency', 'fdatasync', 'sync_log', 'persist_log', 'write_wal', 'rocksdb kv write', 'rocksdb background compaction', 'store-local', 'node-local', 'wal sync', ), } for route, tokens in token_groups.items(): for token in tokens: if token in merged: scores[route] += 2 if 'commit_log' in attachment_text: scores['io_jitter'] += 2 if 'store_batch_wait' in attachment_text or 'prewrite:' in attachment_text: scores['io_jitter'] += 1 if 'rpc layer latency' in clinic_text and 'sync_log' not in attachment_text and 'persist_log' not in attachment_text: scores['network_issues'] += 1 if 'localized node stall' in clinic_text or 'local io' in clinic_text: scores['io_jitter'] += 3 return scores def build_diagnosis_result( context: dict, *, clinic_results: list[dict], grafana_results: list[dict], bundle_results: list[dict], attachment_results: list[dict] | None = None, missing_evidence_reasons: list[str], context_is_leader_drop, context_is_network_issue, context_is_flow_control, ): text = '\n'.join( str(context.get(key) or '') for key in ('title', 'problem_statement', 'phenomenon', 'routing_context_text') ) lowered = text.lower() entry_mode = 'generic_tikv_slow' activated = ['generic_tikv_slow'] reference_files = ['references/cross-cutting.md'] path = 'unknown' attachment_results = attachment_results or [] route_scores = _route_scores( context, clinic_results=clinic_results, attachment_results=attachment_results, context_is_leader_drop=context_is_leader_drop, context_is_network_issue=context_is_network_issue, context_is_flow_control=context_is_flow_control, ) strongest_route = max(route_scores.items(), key=lambda item: item[1])[0] read_path_runtime_hint = _infer_read_path_queueing_signal(text) if strongest_route == 'leader_drop' and route_scores['leader_drop'] > 0: entry_mode = 'leader_drop' activated = ['leader_drop'] elif strongest_route == 'flow_control' and route_scores['flow_control'] > 0: entry_mode = 'flow_control' activated = ['flow_control'] elif strongest_route == 'io_jitter' and route_scores['io_jitter'] > 0: entry_mode = 'io_jitter' activated = ['io_jitter'] elif strongest_route == 'network_issues' and route_scores['network_issues'] > 0: entry_mode = 'network_issues' activated = ['network_issues'] elif strongest_route == 'snapshot_backlog' or any(token in lowered for token in ('scatter', 'hot leader', 'snapshot', 'delete-range', 'rebalance')): entry_mode = 'snapshot_backlog' activated = ['snapshot_backlog'] reference_files, path = _route_reference_files(entry_mode, lowered) reference_files = [item for item in reference_files if item in ALLOWED_REFERENCE_FILES] is_tikv_side = bool( 'tikv' in lowered or entry_mode != 'generic_tikv_slow' or any(result for result in clinic_results if result.get('tikv_write_path_evidence')) ) successful_clinic = [item for item in clinic_results if not item.get('error')] successful_grafana = [item for item in grafana_results if not item.get('error')] successful_bundles = [item for item in bundle_results if not item.get('error')] route_assessment = _route_evidence_assessment( entry_mode, clinic_results=clinic_results, grafana_results=grafana_results, attachment_results=attachment_results, ) if entry_mode != 'generic_tikv_slow' and (successful_clinic or successful_grafana or attachment_results): if not bool(route_assessment.get('supported')): entry_mode = str(route_assessment.get('fallback_entry_mode') or 'generic_tikv_slow') activated = [entry_mode] reference_files, path = _route_reference_files(entry_mode, lowered) scope = 'cluster_wide_or_unknown' earliest_bottleneck = 'unproven' key_evidence: list[str] = [] attachment_summaries = [] for item in attachment_results: if item.get('error'): continue slow_log_summary = ((item.get('analysis') or {}).get('slow_log_summary') or {}) attachment_summaries.extend(str(line) for line in (slow_log_summary.get('summary_lines') or []) if str(line).strip()) for hint in (slow_log_summary.get('focus_hints') or [])[:2]: if str(hint).strip(): attachment_summaries.append(str(hint).strip()) if successful_clinic: first = successful_clinic[0] cluster_summary = first.get('cluster_summary') or {} stores = cluster_summary.get('stores_used_by_instance') or [] if stores: scope = 'store_subset_or_hotspot' deep_rca = first.get('tikv_write_path_evidence') or {} summary = str(deep_rca.get('summary') or '').strip() if summary: key_evidence.append(summary) earliest_bottleneck = summary specialized = {} if entry_mode == 'network_issues': specialized = { 'detail_lines': ['transport latency evidence on TiDB <-> TiKV or TiKV <-> TiKV hop'], } elif entry_mode == 'flow_control': specialized = { 'detail_lines': ['flow-control or admission-protection signals were active in the same window'], } elif ( strongest_route == 'flow_control' and not bool(route_assessment.get('supported')) and route_assessment.get('contradiction_hits') ): specialized = { 'detail_lines': [ 'Initial text-only flow-control route was rejected after evidence review because no direct TiKV throttling or rejection signal was found.', ], } if specialized.get('detail_lines'): key_evidence.extend(str(item) for item in specialized.get('detail_lines')[:2] if str(item).strip()) if not key_evidence: findings = (((first.get('metrics_evidence') or {}).get('performance_overview_summary') or {}).get('items')) or [] for item in findings[:2]: judgement = str((item or {}).get('judgement') or '').strip() if judgement: key_evidence.append(judgement) if entry_mode == 'io_jitter' and attachment_summaries: for item in attachment_summaries[:2]: if item not in key_evidence: key_evidence.append(item) elif successful_grafana: scope = 'cluster_wide_or_unknown' first = successful_grafana[0] cluster_summary = first.get('cluster_summary') or {} cluster_name = str(cluster_summary.get('cluster_name') or '').strip() if cluster_name: key_evidence.append(f'Grafana cluster summary was collected for {cluster_name}.') earliest_bottleneck = 'monitoring context collected from Grafana, but TiKV bottleneck not yet proven' elif successful_bundles: earliest_bottleneck = 'bundle metadata collected, but metric-stage bottleneck not yet proven' first = successful_bundles[0] tidb_version = str(first.get('tidb_version') or '').strip() if tidb_version: key_evidence.append(f'Clinic bundle metadata confirmed cluster version {tidb_version}.') if not key_evidence and attachment_summaries: key_evidence.extend(attachment_summaries[:3]) earliest_bottleneck = attachment_summaries[0] if ( strongest_route == 'flow_control' and entry_mode != 'flow_control' and not bool(route_assessment.get('supported')) ): contradiction_hits = [str(item).strip() for item in route_assessment.get('contradiction_hits') or [] if str(item).strip()] key_evidence.insert( 0, 'Initial text-only route pointed to `flow_control`, but post-evidence review did not confirm TiKV throttling / rejection signals, so the diagnosis was re-opened.' ) if contradiction_hits: key_evidence.insert( 1, f'Evidence instead points toward upstream PD-client / leader reachability signals: {", ".join(contradiction_hits[:4])}.' ) if earliest_bottleneck == 'unproven': if entry_mode == 'network_issues': earliest_bottleneck = 'upstream_pd_connectivity_or_leader_reachability' else: earliest_bottleneck = 'flow_control_not_confirmed_reopen_diagnosis' if ( strongest_route != entry_mode and strongest_route != 'flow_control' and entry_mode != strongest_route and not bool(route_assessment.get('supported')) and route_scores.get(strongest_route, 0) > 0 ): contradiction_hits = [str(item).strip() for item in route_assessment.get('contradiction_hits') or [] if str(item).strip()] key_evidence.insert( 0, f'Initial text-only route pointed to `{strongest_route}`, but evidence review did not confirm that path, so the diagnosis was re-opened.' ) if contradiction_hits: key_evidence.insert( 1, f'Evidence instead aligns better with `{entry_mode}` signals: {", ".join(contradiction_hits[:4])}.' ) if earliest_bottleneck == 'unproven': if entry_mode == 'network_issues': earliest_bottleneck = 'reopened_to_network_path_after_evidence_review' elif entry_mode == 'io_jitter': earliest_bottleneck = 'reopened_to_store_local_io_path_after_evidence_review' elif entry_mode == 'snapshot_backlog': earliest_bottleneck = 'reopened_to_snapshot_path_after_evidence_review' else: earliest_bottleneck = 'initial_route_not_confirmed_reopen_diagnosis' if ( entry_mode == 'generic_tikv_slow' and path == 'read_path' and read_path_runtime_hint and earliest_bottleneck in ('unproven', 'monitoring context collected from Grafana, but TiKV bottleneck not yet proven') ): earliest_bottleneck = str(read_path_runtime_hint.get('bottleneck') or earliest_bottleneck) if read_path_runtime_hint.get('scope'): scope = str(read_path_runtime_hint['scope']) for item in read_path_runtime_hint.get('key_evidence') or []: evidence = str(item or '').strip() if evidence and evidence not in key_evidence: key_evidence.append(evidence) if entry_mode == 'io_jitter' and path != 'store_local_path': path = 'store_local_path' if entry_mode == 'network_issues' and attachment_summaries and any( re.search(r'sync_log|persist_log|write_wal|store-local|node-local', item, re.I) for item in attachment_summaries ): entry_mode = 'io_jitter' activated = ['io_jitter'] reference_files = ['references/cross-cutting.md', 'references/io-jitter.md'] path = 'store_local_path' if attachment_summaries: key_evidence = attachment_summaries[:2] + key_evidence[:2] missing_evidence = [] if REASON_CLINIC_DATA_NOT_READY in missing_evidence_reasons: missing_evidence.append('Clinic link exists but data_status != 100 / not ready yet.') if REASON_GRAFANA_RELATIVE_WINDOW in missing_evidence_reasons: missing_evidence.append('Grafana evidence still uses a relative time window and is not replay-stable.') if REASON_PER_STORE_EVIDENCE_MISSING in missing_evidence_reasons: missing_evidence.append('Per-store evidence is still absent.') if REASON_PER_HOP_EVIDENCE_MISSING in missing_evidence_reasons: missing_evidence.append('Per-hop evidence is still absent.') if REASON_LOWER_LAYER_EVIDENCE_MISSING in missing_evidence_reasons: missing_evidence.append('Lower-layer evidence is still absent before naming a durable root cause.') if not missing_evidence and not successful_clinic and not successful_grafana: missing_evidence.append('Replay-stable metric evidence is still absent.') if not key_evidence: key_evidence.append('Ticket-visible symptom and owner comments were parsed, but metric proof is still limited.') if missing_evidence: uncertainty_statement = 'Current conclusion remains provisional because key proof thresholds are not fully met.' else: uncertainty_statement = 'Current conclusion is limited to the evidence collected in this workflow and should still be cross-checked against the full diagnosis skill.' if successful_clinic: provisional_conclusion = f'Current evidence most strongly routes this case into `{entry_mode}` with a provisional `{path}` judgement.' elif successful_grafana: provisional_conclusion = f'Current evidence suggests `{entry_mode}` may be relevant, but Grafana-only context is not enough to prove the TiKV bottleneck.' else: provisional_conclusion = f'Current output only establishes a first-pass route of `{entry_mode}`; TiKV-side root cause is still unproven.' if ( strongest_route == 'flow_control' and entry_mode != 'flow_control' and not bool(route_assessment.get('supported')) ): if entry_mode == 'network_issues': provisional_conclusion = ( 'The case first matched a text-only `flow_control` route, but evidence review did not confirm TiKV throttling or rejection; ' 'the diagnosis was reopened and now points more strongly to upstream PD connectivity / leader reachability.' ) else: provisional_conclusion = ( 'The case first matched a text-only `flow_control` route, but evidence review did not confirm TiKV throttling or rejection; ' 'the diagnosis was reopened instead of keeping the original flow-control conclusion.' ) elif ( strongest_route != entry_mode and route_scores.get(strongest_route, 0) > 0 and not bool(route_assessment.get('supported')) ): provisional_conclusion = ( f'The case first matched a text-only `{strongest_route}` route, but evidence review did not confirm that path; ' f'the diagnosis was reopened and now points more strongly to `{entry_mode}`.' ) if ( entry_mode == 'generic_tikv_slow' and path == 'read_path' and earliest_bottleneck == 'unified_read_pool_queueing' ): provisional_conclusion = ( 'Current evidence most strongly supports `read_path` queueing before execution, ' 'with `unified_read_pool_queueing` more likely than handler execution as the earliest bottleneck.' ) elif ( entry_mode == 'generic_tikv_slow' and path == 'read_path' and earliest_bottleneck == 'localized_read_pool_saturation' ): provisional_conclusion = ( 'Current evidence most strongly supports store-local `read_path` pressure, ' 'with `localized_read_pool_saturation` explaining the duration rise earlier than handler execution.' ) return DiagnosisResult( entry_mode=entry_mode, activated_subcapabilities=activated, is_tikv_side=is_tikv_side, scope=scope, path=path, earliest_proven_bottleneck=earliest_bottleneck, key_evidence=key_evidence, missing_evidence=missing_evidence, uncertainty_statement=uncertainty_statement, provisional_conclusion=provisional_conclusion, reference_files_consulted=reference_files, )