"""Result parser for processing raw execution responses from TCMS workflows.""" import json import re from typing import Dict, List, Optional from src.clients import tcms_client, tiworkload_client class ExecutionNode: """ Represents a single execution node in the workflow. All nodes have the same structure regardless of workload type. """ def __init__( self, node_name: str, task_id: str, status: str, phase: Optional[str] = None, raw_log: Optional[str] = None, summary: Optional[str] = None, metadata: Optional[Dict] = None, ): """ Args: node_name: Node name in the workflow task_id: Task ID status: Node status (SUCCEEDED / FAILED) phase: Execution phase (prepare/workload/cleanup), None for CH raw_log: Complete raw log content (ANSI codes cleaned) summary: Parsed summary (TPS, QPS, etc.) metadata: Node-specific metadata (exec_id, plan_id, etc.) """ self.node_name = node_name self.task_id = task_id self.status = status self.phase = phase self.raw_log = raw_log self.summary = summary self.metadata = metadata or {} def to_dict(self) -> Dict: """Convert to dictionary for JSON serialization""" return { "node_name": self.node_name, "task_id": self.task_id, "status": self.status, "phase": self.phase, "raw_log": self.raw_log, "summary": self.summary, "metadata": self.metadata, } @classmethod def from_dict(cls, d: Dict) -> "ExecutionNode": """Create ExecutionNode from dictionary""" return cls( node_name=d.get("node_name"), task_id=d.get("task_id"), status=d.get("status"), phase=d.get("phase"), raw_log=d.get("raw_log"), summary=d.get("summary"), metadata=d.get("metadata"), ) def __repr__(self): return f"ExecutionNode(name={self.node_name}, status={self.status}, phase={self.phase})" class ExecutionResult: """ Unified result structure for all execution types. This class wraps all execution results in a consistent format, making it easy to serialize and persist to database. """ def __init__( self, status: str, workload_type: str, nodes: Optional[List[ExecutionNode]] = None, failed_node_name: Optional[str] = None, error: Optional[str] = None, ): """ Args: status: Execution status (SUCCESS, FAILURE, ERROR) workload_type: Type of workload (bench-tpcc, bench-ch, etc.) nodes: List of execution nodes failed_node_name: Name of failed node (None for SUCCESS) error: Error message (None for SUCCESS) """ self.status = status self.workload_type = workload_type self.nodes = nodes or [] self.failed_node_name = failed_node_name self.error = error def to_dict(self) -> Dict: """Convert to dictionary for JSON serialization""" return { "status": self.status, "workload_type": self.workload_type, "nodes": [node.to_dict() for node in self.nodes], "failed_node_name": self.failed_node_name, "error": self.error, } def to_json(self) -> str: """Convert to JSON string""" return json.dumps(self.to_dict(), indent=2) @classmethod def from_dict(cls, d: Dict) -> "ExecutionResult": """Create ExecutionResult from dictionary""" nodes_data = d.get("nodes", []) nodes = [ExecutionNode.from_dict(n) for n in nodes_data] return cls( status=d.get("status"), workload_type=d.get("workload_type"), nodes=nodes, failed_node_name=d.get("failed_node_name"), error=d.get("error"), ) def get_successful_nodes(self) -> List[ExecutionNode]: """Get all successful nodes""" return [n for n in self.nodes if n.status == "SUCCEEDED"] def get_failed_nodes(self) -> List[ExecutionNode]: """Get all failed nodes""" return [n for n in self.nodes if n.status == "FAILED"] def __repr__(self): return f"ExecutionResult(status={self.status}, workload_type={self.workload_type}, nodes={len(self.nodes)}, error={self.error})" def remove_ansi_codes(text): """remove ANSI escape sequences""" ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") return ansi_escape.sub("", text) def parse_raw_ch_log_tcms(log_content: str) -> str: """ Extract the summary section from go-tpc (CH benchmark) raw log content. Finds the first and last lines starting with '[Summary]', then returns the content from one line before the first [Summary] to one line after the last [Summary]. For example, given a log with [Summary] lines, it extracts from "Finished: 100 OLTP workers, 1 OLAP workers" (line before first [Summary]) to "QphH: 358.1" (line after last [Summary]). Args: log_content: Raw log content as string Returns: Extracted log content containing only the summary section. Returns empty string if no [Summary] lines found. """ if not log_content: return "" # Normalize line endings: replace \r\n with \n, then split normalized_content = log_content.replace("\r\n", "\n").replace("\r", "\n") lines = normalized_content.split("\n") # Find all line indices that contain [Summary] # The log format includes timestamp and log level prefixes, so we search # for lines containing [Summary] rather than starting with it summary_indices = [] for i, line in enumerate(lines): if "[Summary]" in line: summary_indices.append(i) # If no [Summary] lines found, return empty string if not summary_indices: return "" # Get first and last [Summary] line indices first_summary_idx = summary_indices[0] last_summary_idx = summary_indices[-1] # Calculate the range: one line before first [Summary] to one line after last [Summary] start_idx = max(0, first_summary_idx - 1) end_idx = min(len(lines), last_summary_idx + 2) # +2 because slice end is exclusive # Extract and join the lines extracted_lines = lines[start_idx:end_idx] return "\n".join(extracted_lines) def parse_raw_sysbench_log_tiworkload(log_content: str) -> str: """ Extract the summary section from sysbench raw log content. Finds lines from "SQL statistics:" to "--- Command finished. ---", then extracts the text from the last [...] bracket in each line. Example input line: [2025/09/11 10:03:13.511 +00:00] [INFO] [cmd.go:43] ["SQL statistics:"] Example output: SQL statistics: queries performed: ... Args: log_content: Raw log content as string Returns: Extracted summary with cleaned text. Returns empty string if no matching section found. """ if not log_content: return "" # Normalize line endings normalized_content = log_content.replace("\r\n", "\n").replace("\r", "\n") lines = normalized_content.split("\n") # Find start and end indices start_idx = -1 end_idx = -1 for i, line in enumerate(lines): if '["SQL statistics:"]' in line or "[SQL statistics:]" in line: start_idx = i if "--- Command finished. ---" in line: end_idx = i break # If section not found, return empty if start_idx == -1 or end_idx == -1: return "" # Extract and clean lines summary_lines = [] for i in range(start_idx, end_idx + 1): line = lines[i].strip() if not line: continue # Extract text from last [...] bracket # Pattern: [...] [...] [...] ["actual content"] last_bracket_start = line.rfind("[") last_bracket_end = line.rfind("]") if last_bracket_start != -1 and last_bracket_end != -1: content = line[last_bracket_start + 1 : last_bracket_end] # Remove quotes if present content = content.strip('"') # Skip empty content or log metadata like "cmd.go:43" if content and not content.startswith("cmd.go:"): summary_lines.append(content) return "\n".join(summary_lines) def parse_raw_tpcc_log_tiworkload(log_content: str) -> str: """Extract TPCC summary block from TiWorkload go-tpc logs. Use the same block-selection strategy as parse_raw_ch_log_tcms: - Find all lines that contain "[Summary]" - Return from one line before the first [Summary] line to one line after the last [Summary] line. Args: log_content: Raw log content as string Returns: Extracted summary block as plain text. Returns empty string if no [Summary] lines are found. """ if not log_content: return "" normalized_content = log_content.replace("\r\n", "\n").replace("\r", "\n") lines = normalized_content.split("\n") # Step 1: locate the summary block by [Summary] markers summary_indices = [] for i, line in enumerate(lines): if "[Summary]" in line: summary_indices.append(i) if not summary_indices: return "" first_summary_idx = summary_indices[0] last_summary_idx = summary_indices[-1] start_idx = max(0, first_summary_idx - 1) end_idx = min(len(lines), last_summary_idx + 2) # Step 2: within that block, extract only the last ["..."] or [...] content summary_lines: list[str] = [] for i in range(start_idx, end_idx): raw_line = lines[i].strip() if not raw_line: continue content = "" # Prefer the last ["..."] pattern (e.g. ["[Summary] ..."], ["tpmC: ..."]). last_quoted_start = raw_line.rfind('["') last_quoted_end = raw_line.rfind('"]') if ( last_quoted_start != -1 and last_quoted_end != -1 and last_quoted_end > last_quoted_start + 2 ): content = raw_line[last_quoted_start + 2 : last_quoted_end].strip() else: # Fallback: take the last [...] group (e.g. [Finished]) last_bracket_start = raw_line.rfind("[") last_bracket_end = raw_line.rfind("]") if ( last_bracket_start != -1 and last_bracket_end != -1 and last_bracket_end > last_bracket_start + 1 ): content = raw_line[last_bracket_start + 1 : last_bracket_end].strip() # Skip empty content or metadata like cmd.go:43 if not content or content.startswith("cmd.go:"): continue summary_lines.append(content) return "\n".join(summary_lines) def extract_graph_nodes(raw_response: Dict) -> List[Dict]: """ Extract nodes from graph.completedTasks with key information. Each node contains: - name: node name from nodeName - kind: action kind (TEST_CASE_CALL, TI_WORKLOAD_CALL, etc.) - status: task status (SUCCEEDED, FAILED, etc.) - outputs: task outputs - state_output: stateOutput field - task_id: taskID - error_message: errMessage (for failed tasks) Args: raw_response: The raw execution response from TCMS workflow Returns: List of node dictionaries """ graph = raw_response.get("graph", {}) completed_tasks = graph.get("completedTasks", []) nodes = [] for task in completed_tasks: action = task.get("action", {}) node = { "name": task.get("nodeName"), "kind": action.get("kind"), "status": task.get("status"), "outputs": task.get("outputs", {}), "state_output": task.get("stateOutput", {}), "task_id": task.get("taskID"), "error_message": task.get("errMessage"), } nodes.append(node) return nodes def filter_nodes(nodes: List[Dict], **filters) -> List[Dict]: """ Filter nodes based on criteria. Args: nodes: List of node dictionaries **filters: Filter criteria (e.g., kind="TEST_CASE_CALL", status="FAILED") Returns: Filtered list of nodes """ filtered = nodes for key, value in filters.items(): if isinstance(value, list): # Support filtering by list of values filtered = [n for n in filtered if n.get(key) in value] else: filtered = [n for n in filtered if n.get(key) == value] return filtered def parse_execution_result( raw_response: Dict, status: str, workload_template: str ) -> ExecutionResult: """ Parse raw execution response based on status and workload template. Args: raw_response: The raw execution response from TCMS workflow status: Task status (SUCCESS, FAILURE, ERROR) workload_template: Type of workload (bench-tpcc, bench-sysbench, longterm-tpcc, bench-ch) Returns: ExecutionResult: Unified result object with nodes, failed_node_name, and error """ if status == "ERROR": _, error_msg = parse_error_result(raw_response) return ExecutionResult( status=status, workload_type=workload_template, nodes=[], failed_node_name=None, error=error_msg, ) elif status in ("SUCCESS", "FAILURE"): # Extract graph nodes once for SUCCESS and FAILURE nodes = extract_graph_nodes(raw_response) if status == "SUCCESS": execution_nodes, failed_node_name, error_msg = parse_success_result( nodes, workload_template ) return ExecutionResult( status=status, workload_type=workload_template, nodes=execution_nodes, failed_node_name=failed_node_name, error=error_msg, ) else: execution_nodes, failed_node_name, error_msg = parse_failure_result( nodes, workload_template ) return ExecutionResult( status=status, workload_type=workload_template, nodes=execution_nodes, failed_node_name=failed_node_name, error=error_msg, ) else: return ExecutionResult( status=status, workload_type=workload_template, nodes=[], failed_node_name=None, error=f"Unknown status: {status}", ) def parse_error_result(raw_response: Dict) -> tuple[None, str]: """ Parse ERROR status result. For ERROR status, extract the 'output' field as error message. Args: raw_response: The raw execution response Returns: tuple: (None, error_msg) """ error_msg = raw_response.get("output", "Unknown error") return None, error_msg def parse_success_result( nodes: List[Dict], workload_template: str ) -> tuple[List[ExecutionNode], Optional[str], None]: """ Parse SUCCESS status result. Returns list of execution nodes and failed_node_name (None for SUCCESS). Args: nodes: Extracted graph nodes workload_template: Type of workload Returns: tuple: (execution_nodes, failed_node_name, error_msg) failed_node_name is always None for SUCCESS """ if workload_template == "bench-ch": execution_nodes = _extract_ch_nodes(nodes) else: # bench-tpcc, bench-sysbench, longterm-tpcc execution_nodes = _extract_common_nodes(nodes, workload_template) return execution_nodes, None, None def _extract_ch_nodes(nodes: List[Dict]) -> List[ExecutionNode]: """ Extract execution nodes from bench-ch workload. For CH workloads, find the test_case_call task (usually named perftest_bench_dbaas_*) and extract its execution details. Also fetches the case execution log for detailed results. Args: nodes: Extracted graph nodes Returns: List of ExecutionNode objects """ execution_nodes = [] test_case_nodes = filter_nodes(nodes, kind="TEST_CASE_CALL") for node in test_case_nodes: if node["status"] != "SUCCEEDED": continue plan_execution_id = node["state_output"].get("planExecutionID") # Build metadata metadata = { "plan_execution_id": plan_execution_id, "plan_execution_url": node["state_output"].get("planExecutionURL"), "plan_execution_status": node["state_output"].get("planExecutionStatus"), "execution_duration": node["outputs"] .get("CaseExecDuration", {}) .get("string"), } raw_log = None summary = None # Fetch case execution details to get artifact URLs if plan_execution_id: case_executions = tcms_client.get_case_executions(plan_execution_id) if case_executions and case_executions.get("data"): case_data = case_executions["data"][0] artifact_urls = case_data.get("artifactURLs", {}) log_url = artifact_urls.get("main-logs") metadata["artifact_urls"] = artifact_urls metadata["log_url"] = log_url # Fetch log content if log_url: log_content = tcms_client.get_case_execution_log(log_url) if log_content: raw_log = remove_ansi_codes(log_content) # Extract summary from go-tpc log summary = parse_raw_ch_log_tcms(raw_log) execution_nodes.append( ExecutionNode( node_name=node["name"], task_id=node["task_id"], status=node["status"], phase=None, # CH doesn't have phases raw_log=raw_log, summary=summary, metadata=metadata, ) ) return execution_nodes def _extract_common_nodes( nodes: List[Dict], workload_template: str ) -> List[ExecutionNode]: """ Extract execution nodes from bench-tpcc, bench-sysbench, longterm-tpcc. These workloads use TI_WORKLOAD_CALL tasks to execute workloads and may have multiple execution phases (prepare, workload, etc.). Args: nodes: Extracted graph nodes workload_template: Type of workload Returns: List of ExecutionNode objects """ execution_nodes = [] workload_nodes = filter_nodes(nodes, kind="TI_WORKLOAD_CALL") for node in workload_nodes: if node["status"] != "SUCCEEDED": continue # Check if this node has successful workload tasks exec_tasks = node["outputs"].get("exec_tasks", {}).get("object", []) # Filter for type=workload and status=success tasks workload_tasks = [ task for task in exec_tasks if task.get("type") == "workload" and task.get("status") == "success" ] if not workload_tasks: continue node_name = node["name"] exec_id = node["outputs"].get("exec_id", {}).get("string") phase = node["state_output"].get("phase") # Build metadata metadata = { "exec_id": exec_id, "exec_status": node["outputs"].get("exec_status", {}).get("string"), "tiworkload_execution_id": node["state_output"].get("executionID"), "tiworkload_host": node["state_output"].get("tiworkloadHost"), "workload_tasks": workload_tasks, } # Fetch logs for all workload tasks and combine them all_logs = [] all_summaries = [] for task in workload_tasks: task_id = task.get("id") if exec_id and task_id: log_content = tiworkload_client.fetch_task_log(exec_id, task_id) if log_content: # Clean ANSI codes cleaned_log = remove_ansi_codes(log_content) all_logs.append(cleaned_log) # Determine which parser to use based on workload type if "sysbench" in workload_template.lower(): summary = parse_raw_sysbench_log_tiworkload(cleaned_log) else: # For tpcc and other go-tpc based workloads summary = parse_raw_tpcc_log_tiworkload(cleaned_log) if summary: all_summaries.append(summary) # Combine all logs and summaries raw_log = "\n\n=== Next Task ===\n\n".join(all_logs) if all_logs else None summary = "\n\n".join(all_summaries) if all_summaries else None execution_nodes.append( ExecutionNode( node_name=node_name, task_id=node["task_id"], status=node["status"], phase=phase, raw_log=raw_log, summary=summary, metadata=metadata, ) ) return execution_nodes def parse_failure_result( nodes: List[Dict], workload_template: str ) -> tuple[List[ExecutionNode], Optional[str], Optional[str]]: """ Parse FAILURE status result. For FAILURE, extract both: 1. Failed node information (which task failed and why) 2. All execution nodes (both successful and failed) Args: nodes: Extracted graph nodes workload_template: Type of workload Returns: tuple: (execution_nodes, failed_node_name, error_msg) """ failed_nodes = filter_nodes(nodes, status="FAILED") # Find failed node name and error message failed_node_name = None error_msg = "No failed task found in response" if failed_nodes: failed_node_name = failed_nodes[0]["name"] error_msg = failed_nodes[0].get("error_message") or "Task failed" # Extract all execution nodes (both successful and failed) # Reuse the same extraction logic as SUCCESS if workload_template == "bench-ch": execution_nodes = _extract_ch_nodes(nodes) # Also extract failed nodes for CH for node in failed_nodes: execution_nodes.append( ExecutionNode( node_name=node["name"], task_id=node["task_id"], status="FAILED", phase=None, raw_log=None, summary=None, metadata={ "plan_execution_id": node["state_output"].get( "planExecutionID" ), "plan_execution_url": node["state_output"].get( "planExecutionURL" ), "plan_execution_status": node["state_output"].get( "planExecutionStatus" ), "error_message": node.get("error_message"), }, ) ) else: execution_nodes = _extract_common_nodes(nodes, workload_template) # Also extract failed nodes for common workloads for node in failed_nodes: execution_nodes.append( ExecutionNode( node_name=node["name"], task_id=node["task_id"], status="FAILED", phase=node["state_output"].get("phase"), raw_log=None, summary=None, metadata={ "exec_id": node["outputs"].get("exec_id", {}).get("string"), "tiworkload_execution_id": node["state_output"].get( "executionID" ), "tiworkload_host": node["state_output"].get("tiworkloadHost"), "error_message": node.get("error_message"), }, ) ) return execution_nodes, failed_node_name, error_msg if __name__ == "__main__": # Test with reference files reference_files = { "ERROR": "reference/E_151656.json", "FAILURE": "reference/F_151668.json", "SUCCESS_TPCC": "reference/S_91322.json", "SUCCESS_CH": "reference/S_210065.json", } print("=" * 60) print("Testing Result Parser") print("=" * 60) # Test ERROR print("\n1. Testing ERROR parsing...") with open(reference_files["ERROR"], "r") as f: error_data = json.load(f) result = parse_execution_result(error_data, "ERROR", "bench-tpcc") print(f" Status: {result.status}") print(f" Workload type: {result.workload_type}") print(f" Error message: {result.error[:100] if result.error else None}...") # Test FAILURE print("\n2. Testing FAILURE parsing...") with open(reference_files["FAILURE"], "r") as f: failure_data = json.load(f) result = parse_execution_result(failure_data, "FAILURE", "bench-tpcc") print(f" Status: {result.status}") print(f" Workload type: {result.workload_type}") print(f" Error message: {result.error}") print(f" Failed node: {result.data.get('failed_node')}") print(f" Successful executions: {result.data.get('successful_execution_count')}") # Test SUCCESS (TPCC) print("\n3. Testing SUCCESS (bench-tpcc) parsing...") with open(reference_files["SUCCESS_TPCC"], "r") as f: success_tpcc_data = json.load(f) result = parse_execution_result(success_tpcc_data, "SUCCESS", "bench-tpcc") print(f" Status: {result.status}") print(f" Workload type: {result.data.get('workload_type')}") print(f" Execution count: {result.data.get('execution_count')}") print(" Executions:") executions = result.data.get("executions", {}) for i, (node_name, exec_info) in enumerate(executions.items()): print( f" {i+1}. {node_name}: " f"exec_id={exec_info.get('exec_id')}, " f"status={exec_info.get('exec_status')}" ) # Test SUCCESS (CH) print("\n4. Testing SUCCESS (bench-ch) parsing...") with open(reference_files["SUCCESS_CH"], "r") as f: success_ch_data = json.load(f) result = parse_execution_result(success_ch_data, "SUCCESS", "bench-ch") print(f" Status: {result.status}") print(f" Node name: {result.data.get('node_name')}") print(f" Has summary: {bool(result.data.get('summary'))}") # Test serialization print("\n5. Testing serialization...") print(f" to_dict() keys: {list(result.to_dict().keys())}") json_str = result.to_json() print(f" to_json() length: {len(json_str)} chars") # Test deserialization result_dict = result.to_dict() restored = ExecutionResult.from_dict(result_dict) print(f" from_dict() status: {restored.status}") print("\n" + "=" * 60) print("All tests completed!") print("=" * 60)