怎样用脚本监控PostgreSQL性能？

wen 实用脚本 2026-07-21 182

本文目录导读：

基础查询脚本
Shell监控脚本
Python高级监控脚本
关键性能指标脚本
系统级监控脚本
Prometheus + Grafana 集成
自动化部署脚本
使用建议

我来介绍几种监控 PostgreSQL 性能的脚本方法：

基础查询脚本

活动会话监控

-- 查看当前活动连接和查询
SELECT 
    pid,
    usename,
    application_name,
    client_addr,
    state,
    query_start,
    wait_event_type,
    wait_event,
    query
FROM pg_stat_activity
WHERE state != 'idle'
ORDER BY query_start DESC;

查询性能分析

-- 慢查询分析
SELECT 
    query,
    calls,
    total_time,
    mean_time,
    rows
FROM pg_stat_statements
ORDER BY total_time DESC
LIMIT 10;

Shell监控脚本

#!/bin/bash
# postgres_monitor.sh
# 配置数据库连接
DB_HOST="localhost"
DB_PORT="5432"
DB_NAME="your_db"
DB_USER="postgres"
# 监控间隔（秒）
INTERVAL=60
# 输出日志文件
LOG_FILE="/var/log/pg_monitor.log"
while true; do
    TIMESTAMP=$(date "+%Y-%m-%d %H:%M:%S")
    # 1. 连接数统计
    CONNECTIONS=$(psql -h $DB_HOST -p $DB_PORT -d $DB_NAME -U $DB_USER -t -c "
        SELECT count(*) FROM pg_stat_activity;
    " | xargs)
    # 2. 活跃查询数
    ACTIVE_QUERIES=$(psql -h $DB_HOST -p $DB_PORT -d $DB_NAME -U $DB_USER -t -c "
        SELECT count(*) FROM pg_stat_activity 
        WHERE state = 'active';
    " | xargs)
    # 3. 数据库大小
    DB_SIZE=$(psql -h $DB_HOST -p $DB_PORT -d $DB_NAME -U $DB_USER -t -c "
        SELECT pg_size_pretty(pg_database_size('$DB_NAME'));
    " | xargs)
    # 4. 检查点统计
    CHECKPOINT_INFO=$(psql -h $DB_HOST -p $DB_PORT -d $DB_NAME -U $DB_USER -t -c "
        SELECT 
            checkpoints_timed,
            checkpoints_req,
            buffers_checkpoint,
            buffers_clean,
            buffers_backend
        FROM pg_stat_bgwriter;
    " | xargs)
    # 5. 缓存命中率
    HIT_RATIO=$(psql -h $DB_HOST -p $DB_PORT -d $DB_NAME -U $DB_USER -t -c "
        SELECT 
            ROUND((blks_hit::numeric / (blks_read + blks_hit) * 100), 2)
        FROM pg_stat_database 
        WHERE datname = '$DB_NAME';
    " | xargs)
    # 写入日志
    echo "$TIMESTAMP | Connections: $CONNECTIONS | Active: $ACTIVE_QUERIES | Size: $DB_SIZE | Hit: ${HIT_RATIO}%" >> $LOG_FILE
    # 告警条件
    if [ "$ACTIVE_QUERIES" -gt 100 ]; then
        echo "ALERT: High active queries: $ACTIVE_QUERIES" | mail -s "PG Alert" admin@example.com
    fi
    sleep $INTERVAL
done

Python高级监控脚本

#!/usr/bin/env python3
# pg_performance_monitor.py
import psycopg2
import time
import json
from datetime import datetime
import requests
class PostgreSQLMonitor:
    def __init__(self, config):
        self.config = config
        self.conn = psycopg2.connect(
            host=config['host'],
            port=config['port'],
            database=config['database'],
            user=config['user'],
            password=config['password']
        )
    def get_server_info(self):
        """获取服务器基本信息"""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    version(),
                    current_setting('server_version'),
                    pg_postmaster_start_time()
            """)
            return cur.fetchone()
    def get_connections_stats(self):
        """获取连接统计"""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    count(*) FILTER (WHERE state = 'active') as active,
                    count(*) FILTER (WHERE state = 'idle') as idle,
                    count(*) FILTER (WHERE state = 'idle in transaction') as idle_tx,
                    count(*) as total
                FROM pg_stat_activity
                WHERE backend_type = 'client backend'
            """)
            return cur.fetchone()
    def get_query_stats(self):
        """获取查询统计"""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    query,
                    calls,
                    total_time / calls as avg_time,
                    rows,
                    shared_blks_hit,
                    shared_blks_read
                FROM pg_stat_statements
                WHERE calls > 0
                ORDER BY total_time DESC
                LIMIT 10
            """)
            return cur.fetchall()
    def get_io_stats(self):
        """获取IO统计"""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    schemaname,
                    tablename,
                    seq_scan,
                    seq_tup_read,
                    idx_scan,
                    idx_tup_fetch,
                    n_tup_ins,
                    n_tup_upd,
                    n_tup_del
                FROM pg_stat_user_tables
                ORDER BY n_tup_ins + n_tup_upd + n_tup_del DESC
                LIMIT 10
            """)
            return cur.fetchall()
    def get_locks_info(self):
        """获取锁信息"""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    l.locktype,
                    l.mode,
                    l.granted,
                    a.query,
                    a.state,
                    a.pid
                FROM pg_locks l
                JOIN pg_stat_activity a ON l.pid = a.pid
                WHERE NOT l.granted
                ORDER BY a.query_start
            """)
            return cur.fetchall()
    def get_metrics(self):
        """获取所有监控指标"""
        return {
            'timestamp': datetime.now().isoformat(),
            'server_info': self.get_server_info(),
            'connections': self.get_connections_stats(),
            'top_queries': self.get_query_stats(),
            'table_stats': self.get_io_stats(),
            'blocked_locks': self.get_locks_info()
        }
    def check_alerts(self, metrics):
        """检查是否需要告警"""
        alerts = []
        # 连接数告警
        connections = metrics['connections']
        if connections[0] > 100:  # active connections
            alerts.append(f"High active connections: {connections[0]}")
        # 锁定告警
        if metrics['blocked_locks']:
            alerts.append(f"Database locks detected: {len(metrics['blocked_locks'])}")
        # 慢查询告警
        for query in metrics['top_queries']:
            if query[2] > 1000:  # average time > 1 second
                alerts.append(f"Slow query detected: {query[0][:100]}")
        return alerts
    def send_metrics(self, metrics):
        """发送指标到监控系统"""
        # 示例：发送到Prometheus Pushgateway
        if 'pushgateway' in self.config:
            requests.post(
                f"{self.config['pushgateway']}/metrics/job/postgresql",
                data=json.dumps(metrics)
            )
    def run(self, interval=60):
        """主循环"""
        while True:
            try:
                # 获取指标
                metrics = self.get_metrics()
                # 检查告警
                alerts = self.check_alerts(metrics)
                if alerts:
                    for alert in alerts:
                        print(f"ALERT: {alert}")
                        # 发送告警通知
                # 发送到监控系统
                self.send_metrics(metrics)
                # 本地输出
                print(f"[{metrics['timestamp']}] Active: {metrics['connections'][0]}, Total: {metrics['connections'][3]}")
            except Exception as e:
                print(f"Error: {e}")
                # 重连机制
                time.sleep(5)
                continue
            time.sleep(interval)
# 使用示例
if __name__ == "__main__":
    config = {
        'host': 'localhost',
        'port': 5432,
        'database': 'your_db',
        'user': 'postgres',
        'password': 'your_password',
        'pushgateway': 'http://pushgateway:9091'  # 可选
    }
    monitor = PostgreSQLMonitor(config)
    monitor.run(interval=60)

关键性能指标脚本

-- postgresql_performance_metrics.sql
-- 1. 缓存命中率
SELECT 
    'Cache Hit Ratio' as metric,
    ROUND((blks_hit::numeric / (blks_read + blks_hit + 1) * 100), 2) as value
FROM pg_stat_database 
WHERE datname = current_database()
UNION ALL
-- 2. 事务提交率
SELECT 
    'Transaction Commit Ratio',
    ROUND((xact_commit::numeric / (xact_commit + xact_rollback + 1) * 100), 2)
FROM pg_stat_database 
WHERE datname = current_database()
UNION ALL
-- 3. 索引使用率
SELECT 
    'Index Usage',
    ROUND((SUM(idx_tup_fetch)::numeric / NULLIF(SUM(seq_tup_read + idx_tup_fetch), 0) * 100), 2)
FROM pg_stat_all_tables;;
-- 4. 表膨胀率检查
SELECT
    schemaname,
    tablename,
    n_dead_tup,
    n_live_tup,
    ROUND(n_dead_tup::numeric / NULLIF(n_live_tup, 0) * 100, 2) as dead_ratio
FROM pg_stat_all_tables
WHERE n_dead_tup > 0
ORDER BY dead_ratio DESC
LIMIT 10;
-- 5. WAL生成率（需要 pg_stat_wal）
SELECT 
    wal_size / (EXTRACT(EPOCH FROM now() - min(wal_time)) / 3600) as wal_per_hour
FROM pg_stat_wal;

系统级监控脚本

#!/bin/bash
# system_pg_monitor.sh
# PostreSQL进程监控
PG_PID=$(pgrep -x postgres | head -1)
if [ -n "$PG_PID" ]; then
    # CPU使用率
    PG_CPU=$(ps -p $PG_PID -o %cpu | tail -1)
    # 内存使用
    PG_MEM=$(ps -p $PG_PID -o %mem,rss | tail -1)
    # 打开文件数
    PG_FILES=$(lsof -p $PG_PID | wc -l)
    echo "PostgreSQL Process Stats:"
    echo "  PID: $PG_PID"
    echo "  CPU: ${PG_CPU}%"
    echo "  Memory: $PG_MEM"
    echo "  Open Files: $PG_FILES"
fi
# I/O等待
iostat -x 1 1 | grep -E "sda|nvme" | awk '{print "IO Wait:", $4}'
# 系统负载
uptime | awk -F'load average:' '{print "Load:", $2}'

Prometheus + Grafana 集成

# prometheus.yml 配置
scrape_configs:
  - job_name: 'postgresql'
    static_configs:
      - targets: ['localhost:9187']  # postgres_exporter

# 使用 postgres_exporter
docker run -d \
  --name postgres_exporter \
  -e DATA_SOURCE_NAME="postgresql://user:password@localhost:5432/dbname?sslmode=disable" \
  -p 9187:9187 \
  prometheuscommunity/postgres-exporter

自动化部署脚本

#!/bin/bash
# deploy_monitor.sh
# 创建监控用户和扩展
psql -U postgres -c "CREATE USER pg_monitor WITH PASSWORD 'monitor_pass' SUPERUSER;"
psql -U postgres -c "CREATE EXTENSION IF NOT EXISTS pg_stat_statements;"
# 配置监控目录
mkdir -p /opt/pg_monitor/{scripts,logs,config}
# 安装依赖
pip3 install psycopg2-binary prometheus-client
# 设置定时任务
cat >> /etc/cron.d/pg_monitor << EOF
* * * * * root /opt/pg_monitor/scripts/collect_metrics.sh
0 * * * * root /opt/pg_monitor/scripts/analyze_queries.sh
EOF
# 启动监控服务
systemctl enable pg_monitor.service
systemctl start pg_monitor.service