MySQL AST改写排障之保存指纹、结构和执行计划代码示例

import time
import json
import logging
import traceback
from typing import dict, any, optional

logging.basicconfig(level=logging.info, format='%(asctime)s - %(levelname)s - %(message)s')

class parserdiagnosticcontext:
    def __init__(self, trace_id: str, client_ip: str):
        self.trace_id = trace_id
        self.client_ip = client_ip
        self.start_time = 0.0
        self.spans = []

    def start_span(self, name: str) -> dict[str, any]:
        span = {
            "name": name,
            "start_time": time.perf_counter_ns(),
            "end_time": 0,
            "status": "ok",
            "attributes": {}
        }
        self.spans.append(span)
        return span

    def finish_span(self, span: dict[str, any], status: str = "ok", err_msg: str = ""):
        span["end_time"] = time.perf_counter_ns()
        span["status"] = status
        if err_msg:
            span["attributes"]["error.message"] = err_msg

class custommysqlparser:
    def __init__(self, slow_parse_threshold_ms: float = 5.0):
        self.slow_parse_threshold_ms = slow_parse_threshold_ms

    def parse_and_rewrite(self, raw_sql: str, diag_ctx: parserdiagnosticcontext) -> optional[str]:
        # 1. 词法与语法分析 span
        lexer_span = diag_ctx.start_span("lexer_yacc_parse")
        try:
            time.sleep(0.001) # 模拟解析耗时
            ast = self._build_ast(raw_sql)
            diag_ctx.finish_span(lexer_span)
        except exception as e:
            diag_ctx.finish_span(lexer_span, status="error", err_msg=str(e))
            self._dump_evidence(raw_sql, diag_ctx, error_detail=str(e))
            return none

        # 2. ast 重写 span
        rewrite_span = diag_ctx.start_span("ast_rewrite_optimization")
        try:
            rewritten_sql = self._apply_ai_rewrite(ast, raw_sql)
            rewrite_span["attributes"]["pre_rewrite_len"] = len(raw_sql)
            rewrite_span["attributes"]["post_rewrite_len"] = len(rewritten_sql)
            diag_ctx.finish_span(rewrite_span)
        except exception as e:
            diag_ctx.finish_span(rewrite_span, status="error", err_msg=str(e))
            self._dump_evidence(raw_sql, diag_ctx, error_detail=str(e))
            return none

        # 3. 检查总耗时，超过阈值留存慢解析证据
        total_duration_ms = (time.perf_counter_ns() - diag_ctx.spans[0]["start_time"]) / 1e6
        if total_duration_ms > self.slow_parse_threshold_ms:
            self._dump_evidence(raw_sql, diag_ctx, slow_parse=true, duration_ms=total_duration_ms)

        return rewritten_sql

    def _build_ast(self, sql: str) -> dict[str, any]:
        if "syntax_error" in sql:
            raise valueerror("1064 (42000): you have an error in your sql syntax near 'syntax_error'")
        return {"type": "select", "tables": ["users"], "where": "id = 1"}

    def _apply_ai_rewrite(self, ast: dict[str, any], raw_sql: str) -> str:
        return f"/*+ max_execution_time(1000) */ {raw_sql}"

    def _dump_evidence(self, raw_sql: str, diag_ctx: parserdiagnosticcontext, 
                       error_detail: str = "", slow_parse: bool = false, duration_ms: float = 0.0):
        """将关键现场排障证据落盘为标准结构化 json 日志"""
        evidence_packet = {
            "timestamp": time.strftime("%y-%m-%dt%h:%m:%sz", time.gmtime()),
            "trace_id": diag_ctx.trace_id,
            "client_ip": diag_ctx.client_ip,
            "raw_sql_masked": self._mask_sql(raw_sql),
            "is_slow_parse": slow_parse,
            "duration_ms": duration_ms,
            "error_detail": error_detail,
            "spans_trace": diag_ctx.spans
        }
        # 实际生产中输出到 stderr 或 dedicated diagnostics log file
        logging.error(f"[parser evidence dump]\n{json.dumps(evidence_packet, indent=2, ensure_ascii=false)}")

    def _mask_sql(self, sql: str) -> str:
        # 简单的脱敏逻辑
        return sql.strip()

if __name__ == "__main__":
    parser = custommysqlparser(slow_parse_threshold_ms=0.5)
    
    # 模拟正常请求
    ctx1 = parserdiagnosticcontext(trace_id="tx-90123-abc", client_ip="192.168.1.50")
    parser.parse_and_rewrite("select * from users where status = 1", ctx1)
    
    # 模拟语法错误故障现场
    ctx2 = parserdiagnosticcontext(trace_id="tx-90124-err", client_ip="192.168.1.51")
    parser.parse_and_rewrite("select * from syntax_error users", ctx2)

import unittest
import logging
from typing import dict, any

logging.basicconfig(level=logging.info, format='%(asctime)s - %(levelname)s - %(message)s')

class testparserregressionsuite(unittest.testcase):
    def setup(self):
        self.parser = custommysqlparser(slow_parse_threshold_ms=10.0)

    def test_normal_parse_and_rewrite(self):
        """验证正常 sql 解析与 hint 注入断言"""
        sql = "select id, name from users where age > 18"
        ctx = parserdiagnosticcontext(trace_id="test-001", client_ip="127.0.0.1")
        
        rewritten_sql = self.parser.parse_and_rewrite(sql, ctx)
        
        self.assertisnotnone(rewritten_sql, "正常 sql 解析结果不应为 none")
        self.assertin("max_execution_time", rewritten_sql, "改写后的 sql 必须注入超时 hint")
        self.assertequal(ctx.spans[0]["status"], "ok", "词法解析 span 状态必须为 ok")

    def test_syntax_error_handling(self):
        """验证非法 sql 的崩溃现场捕获与断言"""
        invalid_sql = "select * from syntax_error users"
        ctx = parserdiagnosticcontext(trace_id="test-002", client_ip="127.0.0.1")
        
        rewritten_sql = self.parser.parse_and_rewrite(invalid_sql, ctx)
        
        self.assertisnone(rewritten_sql, "语法错误的 sql 应返回 none 并安全降级")
        self.assertequal(ctx.spans[0]["status"], "error", "词法解析 span 状态应标记为 error")
        self.assertin("error.message", ctx.spans[0]["attributes"], "span 必须附带详细错误信息")

    def test_slow_parse_boundary(self):
        """验证慢解析边界触发与阈值断言"""
        strict_parser = custommysqlparser(slow_parse_threshold_ms=0.0001) # 极低阈值强行触发
        sql = "select * from orders where status = 'pending'"
        ctx = parserdiagnosticcontext(trace_id="test-003", client_ip="127.0.0.1")
        
        result = strict_parser.parse_and_rewrite(sql, ctx)
        self.assertisnotnone(result)

if __name__ == "__main__":
    unittest.main()