交付物：pg_stat_statements 慢查询 Top-N 诊断脚本（Python）

交付：pg_stat_statements 慢查询 Top-N 诊断脚本（Python）

结论

基于 pg_stat_statements 视图，实现了一个可直接 python3 运行的慢查询 Top-N 诊断工具。核心能力：

1. 按 mean_exec_time DESC 取 Top-N，支持 min_calls / min_avg_ms / 指定 db 过滤；
2. 自动计算磁盘读占比（shared_blks_read / (hit + read)）、临时文件溢出量、方差比，并给出经验诊断标签（缺索引 / 参数敏感 / 大排序溢出等）；
3. 可选对 Top-3 跑 EXPLAIN (COSTS, BUFFERS)，直接看到执行计划；
4. 支持 --format json 输出，方便接 ELK / Loki 日志链路。

使用方式

# 基本用法：Top-20 慢查询
python3 pg_statements_slow.py --dsn "dbname=mydb user=app host=10.0.1.5"

# 只看均值 > 100ms 且至少 10 次调用
python3 pg_statements_slow.py --min-avg-ms 100 --min-calls 10

# JSON 输出（接日志系统）
python3 pg_statements_slow.py --format json

# 附带 Top-3 的 EXPLAIN 执行计划
python3 pg_statements_slow.py --explain

# 只看某个库
python3 pg_statements_slow.py --db myapp

完整源码

#!/usr/bin/env python3
"""
pg_statements_slow.py - PostgreSQL pg_stat_statements 慢查询 Top-N 诊断工具
用法: python3 pg_statements_slow.py --dsn "dbname=mydb user=app host=10.0.1.5" --top-n 20
依赖: pip install psycopg2-binary
"""
import argparse, json, sys
from dataclasses import dataclass, asdict
from typing import Optional
try:
    import psycopg2, psycopg2.extras
except ImportError:
    print("ERROR: pip install psycopg2-binary", file=sys.stderr); sys.exit(1)

DEFAULT_DSN = "dbname=postgres user=postgres host=localhost port=5432"

@dataclass
class SlowQuery:
    rank: int; queryid: int; query: str; calls: int
    total_ms: float; mean_ms: float; min_ms: float; max_ms: float; stddev_ms: float
    rows: int; blk_hit: int; blk_read: int; blk_written: int
    temp_read: int; temp_written: int; blk_read_pct: float
    db: str; user: str

def build_sql(top_n, min_calls, min_avg_ms, db=None):
    db_clause = f"AND d.datname = '{db}'" if db else ""
    return f"""
SELECT s.queryid, s.query, s.calls, s.total_exec_time, s.mean_exec_time,
       s.min_exec_time, s.max_exec_time, s.stddev_exec_time, s.rows,
       s.shared_blks_hit, s.shared_blks_read, s.shared_blks_written,
       s.temp_blks_read, s.temp_blks_written, d.datname, u.usename
FROM pg_stat_statements s
JOIN pg_database d ON d.oid = s.dbid
JOIN pg_roles u ON u.oid = s.userid
WHERE s.calls >= {min_calls} AND s.mean_exec_time >= {min_avg_ms}
  {db_clause}
ORDER BY s.mean_exec_time DESC LIMIT {top_n}"""

def fetch(dsn, top_n, min_calls, min_avg_ms, db=None):
    conn = psycopg2.connect(dsn, connect_timeout=10); conn.autocommit = True
    cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
    cur.execute(build_sql(top_n, min_calls, min_avg_ms, db))
    rows = cur.fetchall(); cur.close(); conn.close()
    out = []
    for i, r in enumerate(rows, 1):
        hit, rd = r["shared_blks_hit"] or 0, r["shared_blks_read"] or 0
        pct = (rd / (hit + rd) * 100) if (hit + rd) > 0 else 0.0
        out.append(SlowQuery(i, r["queryid"], r["query"][:2000], r["calls"],
            round(r["total_exec_time"],2), round(r["mean_exec_time"],2),
            round(r["min_exec_time"],2), round(r["max_exec_time"],2),
            round(r["stddev_exec_time"],2) if r["stddev_exec_time"] else 0.0,
            r["rows"], hit, rd, r["shared_blks_written"] or 0,
            r["temp_blks_read"] or 0, r["temp_blks_written"] or 0,
            round(pct,2), r["datname"], r["usename"]))
    return out

def diagnose(q: SlowQuery) -> str:
    tags = []
    if q.blk_read_pct > 10:
        tags.append(f"⚠️ 高磁盘IO(read {q.blk_read_pct:.1f}%)→疑似缺索引/全表扫")
    if q.temp_written > 0:
        tags.append(f"⚠️ 临时文件({q.temp_written//1024}MB)→大排序/哈希溢出")
    if q.mean_ms > 0 and q.stddev_ms / q.mean_ms > 2.0:
        tags.append(f"⚠️ 高方差(stddev/mean={q.stddev_ms/q.mean_ms:.1f})→参数敏感")
    if q.calls > 0 and q.rows/q.calls < 1 and q.blk_read > 1000:
        tags.append("⚠️ 平均<1行但读盘>1MB→扫描效率低")
    if q.blk_written > 100000:
        tags.append(f"⚠️ 高写放大({q.blk_written//1024}MB)→检查批量操作")
    return "; ".join(tags) if tags else "✅ 各项指标正常"

def print_table(queries, verbose=False):
    h = f"{'#':>3}|{'mean_ms':>10}|{'total_ms':>12}|{'calls':>8}|{'rows':>8}|{'blk%':>7}|{'db':<12}|query"
    print(h); print("-"*len(h))
    for q in queries:
        print(f"{q.rank:>3}|{q.mean_ms:>10.2f}|{q.total_ms:>12.2f}|{q.calls:>8}|{q.rows:>8}|{q.blk_read_pct:>6.1f}%|{q.db:<12}|{q.query[:55].replace(chr(10),' ')}")
        if verbose:
            print(f"     └─ {diagnose(q)}")
            print(f"     └─ queryid={q.queryid} max={q.max_ms}ms stddev={q.stddev_ms}ms user={q.user}")

def print_json(queries):
    out = []
    for q in queries:
        d = asdict(q); d["diagnosis"] = diagnose(q); out.append(d)
    print(json.dumps(out, ensure_ascii=False, indent=2))

def print_explain(queries, dsn, top=3):
    conn = psycopg2.connect(dsn, connect_timeout=15); conn.autocommit = True
    cur = conn.cursor()
    for q in queries[:top]:
        print(f"\n{'='*80}\nEXPLAIN queryid={q.queryid} (mean={q.mean_ms}ms)\n  {q.query[:300]}...\n{'='*80}")
        try:
            cur.execute("EXPLAIN (COSTS, BUFFERS) " + q.query[:2000])
            for row in cur.fetchall(): print("  " + row[0])
        except Exception as e:
            print(f"  [EXPLAIN failed: {e}]")
    cur.close(); conn.close()

def main():
    p = argparse.ArgumentParser(description="pg_stat_statements 慢查询 Top-N 诊断")
    p.add_argument("--dsn", default=DEFAULT_DSN)
    p.add_argument("--top-n", type=int, default=20)
    p.add_argument("--min-calls", type=int, default=1)
    p.add_argument("--min-avg-ms", type=float, default=10.0)
    p.add_argument("--db", default=None)
    p.add_argument("--format", choices=["table","json"], default="table")
    p.add_argument("--verbose", action="store_true")
    p.add_argument("--explain", action="store_true")
    a = p.parse_args()
    try:
        qs = fetch(a.dsn, a.top_n, a.min_calls, a.min_avg_ms, a.db)
    except psycopg2.OperationalError as e:
        print(f"ERROR: 连接失败: {e}", file=sys.stderr); sys.exit(1)
    except psycopg2.ProgrammingError as e:
        if "pg_stat_statements" in str(e) or "does not exist" in str(e):
            print("ERROR: pg_stat_statements 未安装。执行: CREATE EXTENSION IF NOT EXISTS pg_stat_statements;", file=sys.stderr)
            print("并确认 shared_preload_libraries 包含 pg_stat_statements", file=sys.stderr)
        else: print(f"ERROR: {e}", file=sys.stderr)
        sys.exit(1)
    if not qs:
        print("未找到慢查询，试试降低 --min-avg-ms"); sys.exit(0)
    if a.format == "json": print_json(qs)
    else: print_table(qs, a.verbose)
    if a.explain: print_explain(qs, a.dsn)
    print(f"\n--- 汇总: Top-{len(qs)} 总调用={sum(q.calls for q in qs)}, 最高均值={qs[0].mean_ms}ms ---")
    warn = [q for q in qs if q.blk_read_pct > 10 or q.temp_written > 0]
    if warn: print(f"⚠️ {len(warn)} 条存在 IO/临时文件异常，建议优先排查")

if __name__ == "__main__":
    main()

设计取舍说明

| 决策 | 理由 | |------|------| | 用 mean_exec_time 而非 total_exec_time 排序 | mean 能抓到"每次都很慢"的查询，total 容易被高频但单次快的查询淹没 | | 计算 blk_read_pct 而非只看 blk_read 绝对值 | 绝对值没有上下文；占比 >10% 意味着缓存命中率低，大概率缺索引 | | temp_blks_written > 0 作为独立告警 | 任何 temp file 写入都意味着内存不够、排序/哈希溢出了磁盘 | | stddev/mean > 2.0 标记参数敏感 | 同一 queryid 不同参数走了不同计划，是慢查询最常见的"隐藏杀手" | | EXPLAIN 用 COSTS, BUFFERS 而非 ANALYZE | 避免二次执行副作用（尤其写操作类查询），BUFFERS 已够判断磁盘 IO | | 支持 --format json | 方便 cron + curl 定期跑，输出直接推 Loki/ES，配合 Grafana 做趋势 |

验收要点对照

• ✅ Python 实现，单文件可运行，依赖仅 psycopg2-binary
• ✅ 基于 pg_stat_statements 视图，取 Top-N 慢查询
• ✅ 含诊断标签（IO / 临时文件 / 参数敏感 / 写放大）
• ✅ 可选 EXPLAIN 执行计划
• ✅ 支持 JSON 输出、多库过滤、可调阈值
• ✅ 错误处理（扩展未安装 / 连接失败 / 空结果）均有明确提示