import re from io import IOBase from typing import Any, Dict, List from datetime import datetime, timezone VECTOR_LOG_PREFIX_REG = re.compile(r"[^;]+;([^;]+);[^;]+;[^;]+;[^;]+") SPLIT_BY_COLON_REG = re.compile(r"(\w+): (\S+)") # SlowLog represents a structured slow log entry. Each field corresponds # to a row in the table below. # # (tidb v8.0.0) desc INFORMATION_SCHRMA.CLUSTER_SLOW_QUERY; # +-------------------------------+---------------------+------+------+---------+-------+ # | Field | Type | Null | Key | Default | Extra | # +-------------------------------+---------------------+------+------+---------+-------+ # | INSTANCE | varchar(64) | YES | | NULL | | # | Time | timestamp(6) | NO | PRI | NULL | | # | Txn_start_ts | bigint(20) unsigned | YES | | NULL | | # | User | varchar(64) | YES | | NULL | | # | Host | varchar(64) | YES | | NULL | | # | Conn_ID | bigint(20) unsigned | YES | | NULL | | # | Session_alias | varchar(64) | YES | | NULL | | # | Exec_retry_count | bigint(20) unsigned | YES | | NULL | | # | Exec_retry_time | double | YES | | NULL | | # | Query_time | double | YES | | NULL | | # | Parse_time | double | YES | | NULL | | # | Compile_time | double | YES | | NULL | | # | Rewrite_time | double | YES | | NULL | | # | Preproc_subqueries | bigint(20) unsigned | YES | | NULL | | # | Preproc_subqueries_time | double | YES | | NULL | | # | Optimize_time | double | YES | | NULL | | # | Wait_TS | double | YES | | NULL | | # | Prewrite_time | double | YES | | NULL | | # | Wait_prewrite_binlog_time | double | YES | | NULL | | # | Commit_time | double | YES | | NULL | | # | Get_commit_ts_time | double | YES | | NULL | | # | Commit_backoff_time | double | YES | | NULL | | # | Backoff_types | varchar(64) | YES | | NULL | | # | Resolve_lock_time | double | YES | | NULL | | # | Local_latch_wait_time | double | YES | | NULL | | # | Write_keys | bigint(22) | YES | | NULL | | # | Write_size | bigint(22) | YES | | NULL | | # | Prewrite_region | bigint(22) | YES | | NULL | | # | Txn_retry | bigint(22) | YES | | NULL | | # | Cop_time | double | YES | | NULL | | # | Process_time | double | YES | | NULL | | # | Wait_time | double | YES | | NULL | | # | Backoff_time | double | YES | | NULL | | # | LockKeys_time | double | YES | | NULL | | # | Request_count | bigint(20) unsigned | YES | | NULL | | # | Total_keys | bigint(20) unsigned | YES | | NULL | | # | Process_keys | bigint(20) unsigned | YES | | NULL | | # | Rocksdb_delete_skipped_count | bigint(20) unsigned | YES | | NULL | | # | Rocksdb_key_skipped_count | bigint(20) unsigned | YES | | NULL | | # | Rocksdb_block_cache_hit_count | bigint(20) unsigned | YES | | NULL | | # | Rocksdb_block_read_count | bigint(20) unsigned | YES | | NULL | | # | Rocksdb_block_read_byte | bigint(20) unsigned | YES | | NULL | | # | DB | varchar(64) | YES | | NULL | | # | Index_names | varchar(100) | YES | | NULL | | # | Is_internal | tinyint(1) | YES | | NULL | | # | Digest | varchar(64) | YES | | NULL | | # | Stats | varchar(512) | YES | | NULL | | # | Cop_proc_avg | double | YES | | NULL | | # | Cop_proc_p90 | double | YES | | NULL | | # | Cop_proc_max | double | YES | | NULL | | # | Cop_proc_addr | varchar(64) | YES | | NULL | | # | Cop_wait_avg | double | YES | | NULL | | # | Cop_wait_p90 | double | YES | | NULL | | # | Cop_wait_max | double | YES | | NULL | | # | Cop_wait_addr | varchar(64) | YES | | NULL | | # | Mem_max | bigint(20) | YES | | NULL | | # | Disk_max | bigint(20) | YES | | NULL | | # | KV_total | double | YES | | NULL | | # | PD_total | double | YES | | NULL | | # | Backoff_total | double | YES | | NULL | | # | Write_sql_response_total | double | YES | | NULL | | # | Result_rows | bigint(22) | YES | | NULL | | # | Warnings | longtext | YES | | NULL | | # | Backoff_Detail | varchar(4096) | YES | | NULL | | # | Prepared | tinyint(1) | YES | | NULL | | # | Succ | tinyint(1) | YES | | NULL | | # | IsExplicitTxn | tinyint(1) | YES | | NULL | | # | IsWriteCacheTable | tinyint(1) | YES | | NULL | | # | Plan_from_cache | tinyint(1) | YES | | NULL | | # | Plan_from_binding | tinyint(1) | YES | | NULL | | # | Has_more_results | tinyint(1) | YES | | NULL | | # | Resource_group | varchar(64) | YES | | NULL | | # | Request_unit_read | double | YES | | NULL | | # | Request_unit_v2 | double | YES | | NULL | | # | Request_unit_write | double | YES | | NULL | | # | Time_queued_by_rc | double | YES | | NULL | | # | Plan | longtext | YES | | NULL | | # | Plan_digest | varchar(128) | YES | | NULL | | # | Binary_plan | longtext | YES | | NULL | | # | Prev_stmt | longtext | YES | | NULL | | # | Query | longtext | YES | | NULL | | # +-------------------------------+---------------------+------+------+---------+-------+ # 80 rows in set (0.09 sec) # # Serverless fields: # - Keyspace_name # - Keyspace_ID # - Serverless_tenant_ID # - Serverless_project_ID # - Serverless_cluster_ID # - Is_tiflash class SlowLog: id: int instance: str time: float date: str txn_start_ts: int user: str host: str conn_id: str session_alias: str exec_retry_count: int exec_retry_time: float query_time: float parse_time: float compile_time: float rewrite_time: float preproc_subqueries: int preproc_subqueries_time: float optimize_time: float wait_ts: float prewrite_time: float wait_prewrite_binlog_time: float commit_time: float get_commit_ts_time: float commit_backoff_time: float backoff_types: str resolve_lock_time: float local_latch_wait_time: float write_keys: int write_size: int prewrite_region: int txn_retry: int cop_time: float process_time: float wait_time: float backoff_time: float lockkeys_time: float request_count: int total_keys: int process_keys: int rocksdb_delete_skipped_count: int rocksdb_key_skipped_count: int rocksdb_block_cache_hit_count: int rocksdb_block_read_count: int rocksdb_block_read_byte: int db: str index_names: str is_internal: bool digest: str stats: str cop_proc_avg: float cop_proc_p90: float cop_proc_max: float cop_proc_addr: str cop_wait_avg: float cop_wait_p90: float cop_wait_max: float cop_wait_addr: str mem_max: int disk_max: int kv_total: float pd_total: float backoff_total: float write_sql_response_total: float result_rows: int warnings: str backoff_detail: str prepared: bool succ: bool is_explicit_txn: bool is_write_cache_table: bool plan_from_cache: bool plan_from_binding: bool has_more_results: bool resource_group: str request_unit_read: float request_unit_v2: float read_pool_task_details: str request_unit_write: float time_queued_by_rc: float plan: str plan_digest: str binary_plan: str prev_stmt: str query: str serverless_tenant_id: str serverless_project_id: str serverless_cluster_id: str worker_role: str is_tiflash: bool def __init__(self): self.instance = None self.time = None self.txn_start_ts = None self.user = None self.host = None self.conn_id = None self.session_alias = None self.exec_retry_count = None self.exec_retry_time = None self.query_time = None self.parse_time = None self.compile_time = None self.rewrite_time = None self.preproc_subqueries = None self.preproc_subqueries_time = None self.optimize_time = None self.wait_ts = None self.prewrite_time = None self.wait_prewrite_binlog_time = None self.commit_time = None self.get_commit_ts_time = None self.commit_backoff_time = None self.backoff_types = None self.resolve_lock_time = None self.local_latch_wait_time = None self.write_keys = None self.write_size = None self.prewrite_region = None self.txn_retry = None self.cop_time = None self.process_time = None self.wait_time = None self.backoff_time = None self.lockkeys_time = None self.request_count = None self.total_keys = None self.process_keys = None self.rocksdb_delete_skipped_count = None self.rocksdb_key_skipped_count = None self.rocksdb_block_cache_hit_count = None self.rocksdb_block_read_count = None self.rocksdb_block_read_byte = None self.db = None self.index_names = None self.is_internal = None self.digest = None self.stats = None self.cop_proc_avg = None self.cop_proc_p90 = None self.cop_proc_max = None self.cop_proc_addr = None self.cop_wait_avg = None self.cop_wait_p90 = None self.cop_wait_max = None self.cop_wait_addr = None self.mem_max = None self.disk_max = None self.kv_total = None self.pd_total = None self.backoff_total = None self.write_sql_response_total = None self.result_rows = None self.warnings = None self.backoff_detail = None self.prepared = None self.succ = None self.is_explicit_txn = None self.is_write_cache_table = None self.plan_from_cache = None self.plan_from_binding = None self.has_more_results = None self.resource_group = None self.request_unit_read = None self.request_unit_v2 = None self.read_pool_task_details = None self.request_unit_write = None self.time_queued_by_rc = None self.plan = None self.plan_digest = None self.binary_plan = None self.prev_stmt = None self.query = None self.serverless_tenant_id = None self.serverless_project_id = None self.serverless_cluster_id = None self.worker_role = None self.is_tiflash = None def to_dict(self) -> Dict[str, Any]: return { "INSTANCE": self.instance, "Time": self.time, "Txn_start_ts": self.txn_start_ts, "User": self.user, "Host": self.host, "Conn_ID": self.conn_id, "Session_alias": self.session_alias, "Exec_retry_count": self.exec_retry_count, "Exec_retry_time": self.exec_retry_time, "Query_time": self.query_time, "Parse_time": self.parse_time, "Compile_time": self.compile_time, "Rewrite_time": self.rewrite_time, "Preproc_subqueries": self.preproc_subqueries, "Preproc_subqueries_time": self.preproc_subqueries_time, "Optimize_time": self.optimize_time, "Wait_TS": self.wait_ts, "Prewrite_time": self.prewrite_time, "Wait_prewrite_binlog_time": self.wait_prewrite_binlog_time, "Commit_time": self.commit_time, "Get_commit_ts_time": self.get_commit_ts_time, "Commit_backoff_time": self.commit_backoff_time, "Backoff_types": self.backoff_types, "Resolve_lock_time": self.resolve_lock_time, "Local_latch_wait_time": self.local_latch_wait_time, "Write_keys": self.write_keys, "Write_size": self.write_size, "Prewrite_region": self.prewrite_region, "Txn_retry": self.txn_retry, "Cop_time": self.cop_time, "Process_time": self.process_time, "Wait_time": self.wait_time, "Backoff_time": self.backoff_time, "LockKeys_time": self.lockkeys_time, "Request_count": self.request_count, "Total_keys": self.total_keys, "Process_keys": self.process_keys, "Rocksdb_delete_skipped_count": self.rocksdb_delete_skipped_count, "Rocksdb_key_skipped_count": self.rocksdb_key_skipped_count, "Rocksdb_block_cache_hit_count": self.rocksdb_block_cache_hit_count, "Rocksdb_block_read_count": self.rocksdb_block_read_count, "Rocksdb_block_read_byte": self.rocksdb_block_read_byte, "DB": self.db, "Index_names": self.index_names, "Is_internal": self.is_internal, "Digest": self.digest, "Stats": self.stats, "Cop_proc_avg": self.cop_proc_avg, "Cop_proc_p90": self.cop_proc_p90, "Cop_proc_max": self.cop_proc_max, "Cop_proc_addr": self.cop_proc_addr, "Cop_wait_avg": self.cop_wait_avg, "Cop_wait_p90": self.cop_wait_p90, "Cop_wait_max": self.cop_wait_max, "Cop_wait_addr": self.cop_wait_addr, "Mem_max": self.mem_max, "Disk_max": self.disk_max, "KV_total": self.kv_total, "PD_total": self.pd_total, "Backoff_total": self.backoff_total, "Write_sql_response_total": self.write_sql_response_total, "Result_rows": self.result_rows, "Warnings": self.warnings, "Backoff_Detail": self.backoff_detail, "Prepared": self.prepared, "Succ": self.succ, "IsExplicitTxn": self.is_explicit_txn, "IsWriteCacheTable": self.is_write_cache_table, "Plan_from_cache": self.plan_from_cache, "Plan_from_binding": self.plan_from_binding, "Has_more_results": self.has_more_results, "Resource_group": self.resource_group, "Request_unit_read": self.request_unit_read, "Request_unit_v2": self.request_unit_v2, "Read_pool_task_details": self.read_pool_task_details, "Request_unit_write": self.request_unit_write, "Time_queued_by_rc": self.time_queued_by_rc, "Plan": self.plan, "Plan_digest": self.plan_digest, "Binary_plan": self.binary_plan, "Prev_stmt": self.prev_stmt, "Query": self.query, "Worker_role": self.worker_role, "Is_tiflash": self.is_tiflash, } # StreamParser is used to read SlowLog from io reader. class StreamParser: reader: IOBase def __init__(self, reader: IOBase): self.reader = reader def close(self): self.reader.close() def read_next(self) -> SlowLog: lines = [] while True: line = self.reader.readline() if not line: if lines: return parse(lines) return # Handle both bytes and string inputs if isinstance(line, bytes): line = line.decode().strip() else: line = line.strip() lines.append(line) if line.endswith(";"): if line.startswith("#") or line.startswith("use"): continue return parse(lines) # parse() parses structured SlowLog from List[str]. def parse(lines: List[str]) -> SlowLog: if not lines: return None sl = SlowLog() for line in lines: parse_line(sl, line) return sl # parse_line() parses a single line in a slow log entry and assigns the # corresponding field to the SlowLog. def parse_line(sl: SlowLog, line: str): if not line: return # Here is a fast path to determine the first line. # example: 2022-09-06T08:13:12.987865618Z;ip-... # Time: 2022-09-06T08:13:12.193975183Z if (line[0] == "2" and "# Time: " in line) or line.startswith("# Time: "): arr = line.split("# Time: ", 2) if len(arr) != 2: raise ValueError("invalid first line: " + line) if arr[0]: m = VECTOR_LOG_PREFIX_REG.findall(arr[0]) if len(m) == 1: sl.instance = m[0] if not arr[1].endswith("Z"): # Only support UTC timezone. raise ValueError("unsupported timezone") time_str = arr[1].removesuffix("Z") if len(time_str) > 26: time_str = time_str[:26] sl.time = datetime.strptime(time_str, "%Y-%m-%dT%H:%M:%S.%f").replace(tzinfo=timezone.utc).timestamp() elif line.startswith("# "): if line.startswith("# User@Host: "): arr = line[len("# User@Host: ") :].split(" @ ", 2) if len(arr) != 2: raise ValueError("invalid user and host line: " + line) user_arr = arr[0].split("[") if len(user_arr) < 2: raise ValueError("invalid user and host line: " + line) sl.user = user_arr[0].strip() host_arr = arr[1].split("[") if len(host_arr) < 2: raise ValueError("invalid user and host line: " + line) sl.host = host_arr[0].strip() elif line.startswith("# Session_alias: "): sl.session_alias = line[len("# Session_alias: ") :].strip() elif line.startswith("# Backoff_types: "): sl.backoff_types = line[len("# Backoff_types: ") :].strip() elif line.startswith("# DB: "): sl.db = line[len("# DB: ") :].strip() elif line.startswith("# Index_names: "): sl.index_names = line[len("# Index_names: ") :].strip() elif line.startswith("# Digest: "): sl.digest = line[len("# Digest: ") :].strip() elif line.startswith("# Stats: "): sl.stats = line[len("# Stats: ") :].strip() elif line.startswith("# Warnings: "): sl.warnings = line[len("# Warnings: ") :].strip() elif line.startswith("# Cop_backoff_"): sl.backoff_detail = line.removeprefix("# ").strip() elif line.startswith("# Plan: "): sl.plan = line[len("# Plan: ") :].strip() elif line.startswith("# Resource_group: "): sl.plan = line[len("# Resource_group: ") :].strip() elif line.startswith("# Plan_digest: "): sl.plan_digest = line[len("# Plan_digest: ") :].strip() elif line.startswith("# Binary_plan: "): sl.binary_plan = line[len("# Binary_plan: ") :].strip() elif line.startswith("# Prev_stmt: "): sl.prev_stmt = line[len("# Prev_stmt: ") :].strip() elif line.startswith("# Serverless_tenant_ID: "): sl.serverless_tenant_id = line[len("# Serverless_tenant_ID: ") :].strip() elif line.startswith("# Serverless_project_ID: "): sl.serverless_project_id = line[len("# Serverless_project_ID: ") :].strip() elif line.startswith("# Serverless_cluster_ID: "): sl.serverless_cluster_id = line[len("# Serverless_cluster_ID: ") :].strip() elif line.startswith("# Worker_role: "): sl.worker_role = line[len("# Worker_role: ") :].strip() elif line.startswith("# Read_pool_task_details: "): sl.read_pool_task_details = line[len("# Read_pool_task_details: ") :].strip() else: for k, v in SPLIT_BY_COLON_REG.findall(line.removeprefix("# ")): if k == "Txn_start_ts": sl.txn_start_ts = int(v) elif k == "Conn_ID": sl.conn_id = int(v) elif k == "Exec_retry_count": sl.exec_retry_count = int(v) elif k == "Exec_retry_time": sl.exec_retry_time = float(v) elif k == "Query_time": sl.query_time = float(v) elif k == "Parse_time": sl.parse_time = float(v) elif k == "Compile_time": sl.compile_time = float(v) elif k == "Rewrite_time": sl.rewrite_time = float(v) elif k == "Preproc_subqueries": sl.preproc_subqueries = int(v) elif k == "Preproc_subqueries_time": sl.preproc_subqueries_time = float(v) elif k == "Optimize_time": sl.optimize_time = float(v) elif k == "Wait_TS": sl.wait_ts = float(v) elif k == "Prewrite_time": sl.prewrite_time = float(v) elif k == "Wait_prewrite_binlog_time": sl.wait_prewrite_binlog_time = float(v) elif k == "Commit_time": sl.commit_time = float(v) elif k == "Get_commit_ts_time": sl.get_commit_ts_time = float(v) elif k == "Resolve_lock_time": sl.resolve_lock_time = float(v) elif k == "Local_latch_wait_time": sl.local_latch_wait_time = float(v) elif k == "Write_keys": sl.write_keys = int(v) elif k == "Write_size": sl.write_size = int(v) elif k == "Prewrite_region": sl.prewrite_region = int(v) elif k == "Txn_retry": sl.txn_retry = int(v) elif k == "Cop_time": sl.cop_time = float(v) elif k == "Process_time": sl.process_time = float(v) elif k == "Wait_time": sl.wait_time = float(v) elif k == "Backoff_time": sl.backoff_time = float(v) elif k == "LockKeys_time": sl.lockkeys_time = float(v) elif k == "Request_count": sl.request_count = int(v) elif k == "Total_keys": sl.total_keys = int(v) elif k == "Process_keys": sl.process_keys = int(v) elif k == "Rocksdb_delete_skipped_count": sl.rocksdb_delete_skipped_count = int(v) elif k == "Rocksdb_key_skipped_count": sl.rocksdb_key_skipped_count = int(v) elif k == "Rocksdb_block_cache_hit_count": sl.rocksdb_block_cache_hit_count = int(v) elif k == "Rocksdb_block_read_count": sl.rocksdb_block_read_count = int(v) elif k == "Rocksdb_block_read_byte": sl.rocksdb_block_read_byte = int(v) elif k == "Is_internal": sl.is_internal = str_to_bool(v) elif k == "Cop_proc_avg": sl.cop_proc_avg = float(v) elif k == "Cop_proc_p90": sl.cop_proc_p90 = float(v) elif k == "Cop_proc_max": sl.cop_proc_max = float(v) elif k == "Cop_proc_addr": sl.cop_proc_addr = v elif k == "Cop_wait_avg": sl.cop_wait_avg = float(v) elif k == "Cop_wait_p90": sl.cop_wait_p90 = float(v) elif k == "Cop_wait_max": sl.cop_wait_max = float(v) elif k == "Cop_wait_addr": sl.cop_wait_addr = v elif k == "Mem_max": sl.mem_max = int(v) elif k == "Disk_max": sl.disk_max = int(v) elif k == "KV_total": sl.kv_total = float(v) elif k == "PD_total": sl.pd_total = float(v) elif k == "Backoff_total": sl.backoff_total = float(v) elif k == "Write_sql_response_total": sl.write_sql_response_total = float(v) elif k == "Result_rows": sl.result_rows = int(v) elif k == "Prepared": sl.prepared = str_to_bool(v) elif k == "Succ": sl.succ = str_to_bool(v) elif k == "IsExplicitTxn": sl.is_explicit_txn = str_to_bool(v) elif k == "IsWriteCacheTable": sl.is_write_cache_table = str_to_bool(v) elif k == "Plan_from_cache": sl.plan_from_cache = str_to_bool(v) elif k == "Plan_from_binding": sl.plan_from_binding = str_to_bool(v) elif k == "Has_more_results": sl.has_more_results = str_to_bool(v) elif k == "Request_unit_read": sl.request_unit_read = float(v) elif k == "Request_unit_v2": sl.request_unit_v2 = float(v) elif k == "Request_unit_write": sl.request_unit_write = float(v) elif k == "Time_queued_by_rc": sl.time_queued_by_rc = float(v) elif k == "Is_tiflash": sl.is_tiflash = str_to_bool(v) elif line.endswith(";"): if line.startswith("use"): # `use DB` statements in the slow log is used to keep it be compatible with MySQL, # since we already get the current DB from the `# DB` field, we can ignore it here, # please see https://github.com/pingcap/tidb/issues/17846 for more details. return else: sl.query = line else: if "following end of new file" in line: # This case: # ``` # 2023-12-19T02...;slowlog tail: /var/log/auditlog/slowlog has been replaced; following end of new file # 2023-12-19T02...;slowlog # Time: 2023-12-19T02:35:33.586638997Z # # Txn_start_ts: 446418997401618680 # # Query_time: 4.86387999 # # Parse_time: 0 # # Compile_time: 0.00052436 # ... # ``` return raise ValueError("invalid line: " + line) def str_to_bool(v: str) -> bool: return v == "1" or v.lower() == "true"