[PostgreSQL]模型训练数据库连接池耗尽：PostgreSQL调优实录

# 典型错误堆栈
sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) FATAL:  remaining connection slots are reserved for non-replication superuser connections
(Background on this error at: https://sqlalche.me/e/14/e3q8)

-- 查询当前连接状态分布
SELECT 
  state,
  application_name,
  COUNT(*) as connection_count,
  AVG(EXTRACT(EPOCH FROM (now() - query_start))) as avg_duration
FROM pg_stat_activity 
WHERE datname = 'ml_platform'
GROUP BY state, application_name
ORDER BY connection_count DESC;

-- 结果示例
┌─────────┬────────────────────┬──────────────────┬──────────────┐
│  state  │ application_name   │ connection_count │ avg_duration │
├─────────┼────────────────────┼──────────────────┼──────────────┤
│ idle    │ data_loader_worker │        127       │    NULL      │
│ active  │ feature_service    │         58       │    45.2      │
│ idle in │ training_manager   │         12       │    120.8     │
│ transaction│                  │                  │              │
└─────────┴────────────────────┴──────────────────┴──────────────┘

// PostgreSQL backend process创建开销示例
// 每个连接消耗约1.5-3MB内存
typedef struct PGPROC {
    // ShmemLock 相关
    slock_t     sem;              // 信号量
    int         waitStatus;       // 等待状态
    
    // 事务相关
    TransactionIdxid;            // 事务ID
    int         pid;              // 进程ID
    
    // 内存上下文
    MemoryContext topMemoryContext; // 约~100KB
    MemoryContext curTransactionContext; // 动态增长
    
    // 其他资源...
} PGPROC;

// 连接建立流程（简化版）
void BackendStartup(Port *port) {
    PGPROC *proc = InitProcess(); // 初始化进程结构
    InitPostgres();               // 加载数据库状态
    BackendLoop();                // 进入查询处理循环
}

# SQLAlchemy QueuePool核心实现
class QueuePool(Pool):
    def __init__(self, creator, pool_size=5, max_overflow=10, 
                 timeout=30, **kw):
        # 初始化连接队列
        self._pool = util.LazyQueue(pool_size)
        self._overflow = 0 - max_overflow
        self._max_overflow = max_overflow
        self._timeout = timeout
        
    def _do_get(self):
        # 尝试获取连接，支持阻塞等待
        try:
            return self._pool.get(block=False)
        except sqla_queue.Empty:
            if self._overflow >= self._max_overflow:
                # 达到最大限制，阻塞等待
                return self._pool.get(block=True, 
                                   timeout=self._timeout)
            else:
                # 创建新连接
                self._overflow += 1
                return self._create_connection()
                
    def _do_return_conn(self, conn):
        # 连接归还池子
        self._pool.put(conn, block=False)

-- 1. 连接状态详细监控视图
CREATE OR REPLACE VIEW monitoring.connection_stats AS
SELECT 
  (SELECT COUNT(*) FROM pg_stat_activity WHERE state = 'active') as active_connections,
  (SELECT COUNT(*) FROM pg_stat_activity WHERE state = 'idle') as idle_connections,
  (SELECT COUNT(*) FROM pg_stat_activity WHERE state = 'idle in transaction') as idle_in_tx,
  (SELECT COUNT(*) FROM pg_stat_activity WHERE wait_event_type IS NOT NULL) as waiting_connections,
  (SELECT MAX(EXTRACT(EPOCH FROM (now() - query_start))) 
   FROM pg_stat_activity WHERE state = 'active') as max_query_time,
  (SELECT COUNT(*) FROM pg_stat_activity WHERE backend_type = 'client backend') as total_clients;

-- 2. 连接来源分析
SELECT 
  application_name,
  client_addr,
  usename,
  COUNT(*) as conn_count,
  COUNT(CASE WHEN state = 'active' THEN 1 END) as active_count,
  COUNT(CASE WHEN state = 'idle' THEN 1 END) as idle_count
FROM pg_stat_activity
GROUP BY application_name, client_addr, usename
HAVING COUNT(*) > 5
ORDER BY conn_count DESC;

-- 3. 锁等待深度分析
SELECT 
  blocked_locks.pid AS blocked_pid,
  blocked_activity.application_name AS blocked_app,
  blocked_activity.query AS blocked_query,
  blocking_locks.pid AS blocking_pid,
  blocking_activity.query AS blocking_query,
  blocked_activity.wait_event_type AS wait_type
FROM pg_catalog.pg_locks blocked_locks
JOIN pg_catalog.pg_stat_activity blocked_activity 
  ON blocked_activity.pid = blocked_locks.pid
JOIN pg_catalog.pg_locks blocking_locks 
  ON blocking_locks.locktype = blocked_locks.locktype
  AND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.database
  AND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relation
JOIN pg_catalog.pg_stat_activity blocking_activity 
  ON blocking_activity.pid = blocking_locks.pid
WHERE NOT blocked_locks.granted;

# 连接池指标收集器
import psutil
from sqlalchemy import event
from prometheus_client import Counter, Histogram, Gauge

class ConnectionPoolMonitor:
    # 定义Prometheus指标
    pool_size_gauge = Gauge('db_pool_size', 'Current pool size')
    active_connections_gauge = Gauge('db_active_connections', 'Active connections')
    waiting_connections_gauge = Gauge('db_waiting_connections', 'Waiting for connection')
    connection_duration_hist = Histogram('db_connection_duration_seconds', 
                                       'Connection usage duration')
    
    @staticmethod
    def monitor_pool(pool):
        """监控SQLAlchemy连接池"""
        
        @event.listens_for(pool, 'checkout')
        def on_checkout(dbapi_conn, connection_record, connection_proxy):
            # 记录连接被取出时间
            connection_record.info['checkout_time'] = time.time()
            
        @event.listens_for(pool, 'checkin')
        def on_checkin(dbapi_conn, connection_record, connection_proxy):
            # 计算连接使用时长
            if 'checkout_time' in connection_record.info:
                duration = time.time() - connection_record.info['checkout_time']
                ConnectionPoolMonitor.connection_duration_hist.observe(duration)
                
        @event.listens_for(pool, 'connect')
        def on_connect(dbapi_conn, connection_record):
            # 更新池大小指标
            size = pool.size()
            overflow = pool.overflow()
            ConnectionPoolMonitor.pool_size_gauge.set(size + overflow)
            
        # 定期采集状态
        def collect_metrics():
            stats = pool.logger.get_stats()
            ConnectionPoolMonitor.active_connections_gauge.set(stats['checked_out'])
            ConnectionPoolMonitor.waiting_connections_gauge.set(stats.get('waiting', 0))
            
        # 每5秒采集一次
        from threading import Timer
        def schedule():
            collect_metrics()
            Timer(5.0, schedule).start()
        schedule()

# 在应用初始化时挂载监控
from sqlalchemy import create_engine
engine = create_engine(
    "postgresql://user:pass@host/ml_platform",
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True
)
ConnectionPoolMonitor.monitor_pool(engine.pool)

# 1. 修改postgresql.conf核心参数
# 内存相关
shared_buffers = 32GB                    # 25%系统内存
effective_cache_size = 96GB              # 75%系统内存
work_mem = 64MB                          # 每连接排序内存
maintenance_work_mem = 2GB               # 维护操作内存

# 连接相关
max_connections = 500                    # 根据连接池需求调整
superuser_reserved_connections = 10      # 保留给超级用户
max_prepared_transactions = 100          # 分布式事务支持

# 并发控制
max_worker_processes = 64                # 等于CPU核数
max_parallel_workers_per_gather = 8      # 单查询并行度
max_parallel_workers = 48                # 总并行工作数

# WAL和日志
wal_buffers = 256MB                      # WAL缓存
checkpoint_timeout = 15min               # 检查点间隔
max_wal_size = 8GB

# 应用修改
pg_ctl -D /var/lib/postgresql/14/main reload

# 2. 内核参数优化（/etc/sysctl.conf）
# 网络相关
net.core.somaxconn = 4096
net.ipv4.ip_local_port_range = 10240 65535

# 内存相关
vm.dirty_background_ratio = 5
vm.dirty_ratio = 10
vm.overcommit_memory = 2
vm.overcommit_ratio = 75

# /etc/pgbouncer/pgbouncer.ini
[databases]
ml_platform = host=localhost port=5432 dbname=ml_platform

[pgbouncer]
listen_addr = 0.0.0.0
listen_port = 6432
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt

# 连接池核心配置
pool_mode = transaction          # 事务级复用
default_pool_size = 25           # 每个数据库的最大连接数
max_client_conn = 10000          # 最大客户端连接数
reserve_pool_size = 5            # 预留连接
reserve_pool_timeout = 3         # 预留超时

# 连接生命周期
server_idle_timeout = 600        # 服务端空闲超时
server_lifetime = 3600           # 连接最大生命周期
server_connect_timeout = 15      # 连接超时
query_timeout = 300              # 查询超时

# 日志和监控
log_connections = 1
log_disconnections = 1
log_pooler_errors = 1
admin_users = postgres
stats_users = monitoring

# config/database_config.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
import os

def get_ml_engine():
    """训练专用连接池配置"""
    return create_engine(
        # 连接PgBouncer而非直连PostgreSQL
        "postgresql://user:pass@pgbouncer-host:6432/ml_platform",
        
        # 核心连接池参数
        pool_size=5,                    # 每个进程保持5个连接
        max_overflow=15,                # 允许突发增长到20
        pool_timeout=30,                # 获取连接超时30秒
        
        # 连接健康检查
        pool_pre_ping=True,             # 预先检测连接有效性
        pool_recycle=3600,               # 1小时后回收连接
        
        # 连接断开后处理
        echo=False,                      # 关闭SQL日志
        echo_pool=True,                  # 开启连接池日志
        
        # 执行选项
        execution_options={
            "isolation_level": "READ COMMITTED",
            "statement_timeout": 60000,  # 语句超时60秒
            "idle_in_transaction_session_timeout": 30000,  # 事务空闲超时
        }
    )

def get_feature_engine():
    """特征服务专用配置 - 更激进的复用"""
    return create_engine(
        "postgresql://user:pass@pgbouncer-host:6432/ml_platform",
        pool_size=3,                    # 特征查询较轻量
        max_overflow=7,
        pool_timeout=10,
        
        # 使用NullPool避免连接池占用
        # 适合短链接密集型场景
        poolclass=NullPool,
    )

# 为不同负载创建独立Session
MLSession = scoped_session(sessionmaker(bind=get_ml_engine()))
FeatureSession = scoped_session(sessionmaker(bind=get_feature_engine()))

-- 查询pg_stat_statements获取TOP SQL
SELECT 
  query,
  calls,
  total_exec_time,
  mean_exec_time,
  max_exec_time,
  stddev_exec_time,
  rows/calls as avg_rows
FROM pg_stat_statements
WHERE query LIKE '%sample_metadata%'
ORDER BY total_exec_time DESC
LIMIT 10;

-- 结果示例
┌────────────────────────────────┬────────┬────────────────┬──────────────┬──────────────┐
│ query                          │ calls  │ total_exec_time│ mean_exec_time│ max_exec_time│
├────────────────────────────────┼────────┼────────────────┼──────────────┼──────────────┤
│ SELECT * FROM sample_metadata  │ 127845 │ 125432.5 ms    │ 0.98 ms      │ 4523.1 ms    │
│ WHERE dataset_id = $1 AND      │        │                │              │              │
│ feature_hash = $2              │        │                │              │              │
├────────────────────────────────┼────────┼────────────────┼──────────────┼──────────────┤
│ UPDATE training_jobs SET       │ 8923   │ 89421.3 ms     │ 10.02 ms     │ 8921.4 ms    │
│ status = $1, updated_at = $2  │        │                │              │              │
│ WHERE id = $3                  │        │                │              │              │
└────────────────────────────────┴────────┴────────────────┴──────────────┴──────────────┘

-- 1. 为高频查询创建复合索引
-- 原查询：SELECT * FROM sample_metadata WHERE dataset_id = $1 AND feature_hash = $2
CREATE INDEX CONCURRENTLY idx_sample_metadata_lookup 
ON sample_metadata (dataset_id, feature_hash) 
INCLUDE (sample_id, label, created_at);  -- 覆盖索引

-- 查看索引使用情况
SELECT 
  schemaname,
  tablename,
  indexname,
  idx_scan,
  idx_tup_read,
  idx_tup_fetch
FROM pg_stat_user_indexes
WHERE tablename = 'sample_metadata';

-- 2. 部分索引优化状态查询
-- 原查询：UPDATE training_jobs SET status = 'running' WHERE id = $1 AND status = 'pending'
CREATE INDEX CONCURRENTLY idx_training_jobs_pending
ON training_jobs (id, status)
WHERE status = 'pending';

-- 3. BRIN索引优化时间序列查询
-- 适合按created_at范围查询的大表
CREATE INDEX CONCURRENTLY idx_sample_metadata_created_brin
ON sample_metadata USING BRIN (created_at) WITH (pages_per_range = 128);

-- 4. 创建索引后分析执行计划
EXPLAIN (ANALYZE, BUFFERS) 
SELECT * FROM sample_metadata 
WHERE dataset_id = 'ds_12345' 
  AND feature_hash = 'a1b2c3d4e5';

-- 优化后执行计划
┌─────────────────────────────────────────────────────────┐
│ Index Scan using idx_sample_metadata_lookup             │
│   Index Cond: ((dataset_id = 'ds_12345'::text) AND      │
│                (feature_hash = 'a1b2c3d4e5'::text))      │
│   Buffers: shared hit=4 read=0                          │
│   Execution Time: 0.123 ms                              │
└─────────────────────────────────────────────────────────┘

# 优化前：循环查询（产生大量短连接）
def load_samples_bad(dataset_id, sample_ids):
    results = []
    for sample_id in sample_ids:  # N+1查询
        result = session.query(SampleMetadata).filter(
            SampleMetadata.dataset_id == dataset_id,
            SampleMetadata.sample_id == sample_id
        ).first()
        results.append(result)
    return results
# 1000个样本产生1000次查询，消耗大量连接

# 优化后：批量查询+索引利用
def load_samples_good(dataset_id, sample_ids):
    # 单次查询，利用索引
    results = session.query(SampleMetadata).filter(
        SampleMetadata.dataset_id == dataset_id,
        SampleMetadata.sample_id.in_(sample_ids)  # 批量IN查询
    ).all()
    
    # 转换为字典快速查找
    return {r.sample_id: r for r in results}

# 优化效果对比
# 连接消耗：1000 → 1
# 查询时间：450ms → 12ms
# 网络RTT：减少999次

# db/context_managers.py
from contextlib import contextmanager
from sqlalchemy.orm import Session
import logging

logger = logging.getLogger(__name__)

class SafeDBConnection:
    """安全连接管理器"""
    
    @staticmethod
    @contextmanager
    def get_session(session_factory):
        """自动管理会话生命周期"""
        session = session_factory()
        try:
            yield session
            session.commit()
        except Exception as e:
            session.rollback()
            logger.error(f"Database error: {e}", exc_info=True)
            raise
        finally:
            # 确保连接归还
            session.close()
    
    @staticmethod
    @contextmanager
    def get_connection(engine):
        """原始连接管理"""
        conn = engine.connect()
        try:
            yield conn
        finally:
            conn.close()

# 使用示例
def process_training_batch(batch_id: int):
    """批量处理训练数据"""
    from db.context_managers import SafeDBConnection
    
    # 安全的连接使用模式
    with SafeDBConnection.get_session MLSession) as session:
        # 查询操作
        samples = session.query(SampleMetadata)\
            .filter_by(batch_id=batch_id)\
            .yield_per(1000)  # 流式查询，减少内存占用
        
        # 处理逻辑
        for sample in samples:
            process_sample(sample)
            
    # 连接自动归还，即使发生异常
    
    # 对比：不安全的写法
    # session = MLSession()
    # samples = session.query(...)  # 如果这里异常，连接永不释放！

# async/db_pool.py
import asyncio
import asyncpg
from typing import Optional

class AsyncPGConnectionPool:
    """异步PostgreSQL连接池"""
    
    def __init__(self, dsn: str, min_size: int = 5, max_size: int = 20):
        self.dsn = dsn
        self.min_size = min_size
        self.max_size = max_size
        self._pool: Optional[asyncpg.Pool] = None
    
    async def init(self):
        """初始化连接池"""
        self._pool = await asyncpg.create_pool(
            self.dsn,
            min_size=self.min_size,
            max_size=self.max_size,
            
            # 连接建立回调
            setup=self._setup_connection,
            
            # 健康检查
            health_check_interval=30,
            
            # 连接超时
            command_timeout=180,
            
            # 连接参数
            server_settings={
                'application_name': 'async_ml_service',
                'statement_timeout': '180000',  # 180秒
            }
        )
    
    async def _setup_connection(self, conn: asyncpg.Connection):
        """配置新连接"""
        await conn.set_type_codec(
            'jsonb',
            encoder=str,
            decoder=lambda x: x,
            schema='pg_catalog'
        )
    
    async def get_connection(self):
        """获取连接（上下文管理器）"""
        return await self._pool.acquire()
    
    async def release_connection(self, conn: asyncpg.Connection):
        """释放连接"""
        await self._pool.release(conn)
    
    async def close(self):
        """关闭连接池"""
        await self._pool.close()

# 异步特征查询服务
class AsyncFeatureService:
    def __init__(self, pool: AsyncPGConnectionPool):
        self.pool = pool
    
    async def get_features_batch(self, sample_ids: list) -> dict:
        """批量异步查询特征"""
        # 直接 await 会阻塞所有查询，应该使用 asyncio.gather
        # 错误示范：
        # results = []
        # for sid in sample_ids:
        #     result = await self.query_one(sid)  # 串行执行！
        #     results.append(result)
        
        # 正确做法：并行查询但复用连接
        async with self.pool.get_connection() as conn:
            # 使用单个连接执行多个查询
            tasks = [
                self.query_one(conn, sid) 
                for sid in sample_ids
            ]
            results = await asyncio.gather(*tasks)
            return dict(zip(sample_ids, results))
    
    async def query_one(self, conn: asyncpg.Connection, sample_id: str):
        """使用已获取的连接查询"""
        return await conn.fetchrow(
            "SELECT * FROM sample_features WHERE sample_id = $1",
            sample_id
        )

# FastAPI集成示例
from fastapi import FastAPI, Depends

app = FastAPI()
pool = AsyncPGConnectionPool("postgresql://...", min_size=10, max_size=30)

@app.on_event("startup")
async def startup():
    await pool.init()

@app.on_event("shutdown")
async def shutdown():
    await pool.close()

async def get_db():
    """依赖注入获取连接"""
    async with pool.get_connection() as conn:
        yield conn

@app.get("/features/{sample_id}")
async def get_features(sample_id: str, conn: asyncpg.Connection = Depends(get_db)):
    return await conn.fetchrow(
        "SELECT * FROM sample_features WHERE sample_id = $1", 
        sample_id
    )

# alert_rules.yml
groups:
- name: pg_connection_alerts
  interval: 15s
  rules:
  
  # 连接池使用率告警
  - alert: HighConnectionPoolUsage
    expr: |
      (pg_stat_activity_count{datname="ml_platform"} 
       / pg_settings_max_connections) > 0.8
    for: 5m
    labels:
      severity: warning
      team: ml-platform
    annotations:
      summary: "PostgreSQL连接池使用率超过80%"
      description: |
        当前实例 {{ $labels.instance }} 连接数: {{ $value | humanize }}
        已持续5分钟，请检查应用连接配置
  
  # 连接等待时间告警
  - alert: ConnectionWaitTimeout
    expr: |
      rate(pgbouncer_pools_client_waiting{database="ml_platform"}[5m]) > 5
    for: 3m
    labels:
      severity: critical
      team: ml-platform
    annotations:
      summary: "PgBouncer连接等待数持续增长"
      description: "每秒超过5个连接在等待，当前值: {{ $value }}"
  
  # 慢查询消耗连接告警
  - alert: LongRunningQueryBlockingConnections
    expr: |
      pg_stat_activity_max_tx_duration{datname="ml_platform"} > 300
    for: 2m
    labels:
      severity: warning
      team: ml-platform
    annotations:
      summary: "存在运行超过5分钟的查询"
      description: "查询PID {{ $labels.pid }} 已运行 {{ $value }}秒"
  
  # 连接泄漏检测
  - alert: ConnectionLeakDetected
    expr: |
      (pg_stat_activity_idle{datname="ml_platform"} 
       > 100) and 
      (pg_stat_activity_idle{datname="ml_platform"} 
       / pg_stat_activity_total{datname="ml_platform"} > 0.6)
    for: 10m
    labels:
      severity: critical
      team: ml-platform
    annotations:
      summary: "检测到大量空闲连接，可能存在连接泄漏"
      description: "空闲连接占比: {{ $value | humanizePercentage }}"

# automation/connection_scaler.py
import subprocess
import psycopg2
from prometheus_api_client import PrometheusConnect

class AutoConnectionScaler:
    """自动连接池扩缩容"""
    
    def __init__(self, prometheus_url: str, db_config: dict):
        self.prom = PrometheusConnect(url=prometheus_url)
        self.db_config = db_config
    
    def get_current_load(self) -> dict:
        """获取当前负载"""
        queries = {
            'connection_usage': '''
                pg_stat_activity_count{job="postgresql"} 
                / pg_settings_max_connections{job="postgresql"}
            ''',
            'waiting_connections': '''
                pgbouncer_pools_client_waiting{database="ml_platform"}
            ''',
            'active_queries': '''
                pg_stat_activity_count{state="active"}
            '''
        }
        
        return {
            k: self.prom.custom_query(v)[0]['value'][1]
            for k, v in queries.items()
        }
    
    def adjust_pgbouncer_pool(self, target_pool_size: int):
        """动态调整PgBouncer连接池"""
        # 连接到PgBouncer管理接口
        conn = psycopg2.connect(
            dbname='pgbouncer',
            user='admin',
            password=self.db_config['admin_pass'],
            host='pgbouncer-host',
            port=6432
        )
        
        # 执行在线配置修改
        cur = conn.cursor()
        cur.execute(f"SET DEFAULT_POOL_SIZE = {target_pool_size}")
        cur.execute("RELOAD")
        cur.close()
        conn.close()
        
        print(f"PgBouncer pool_size adjusted to {target_pool_size}")
    
    def scale_decision(self):
        """扩缩容决策逻辑"""
        load = self.get_current_load()
        
        # 扩容策略
        if float(load['connection_usage']) > 0.85:
            # 增加20%连接池容量
            self.adjust_pgbouncer_pool(target_pool_size=30)
            
        # 缩容策略  
        elif float(load['connection_usage']) < 0.3:
            # 减少连接数
            self.adjust_pgbouncer_pool(target_pool_size=15)
            
        # 写入时序数据库记录
        self.record_scaling_event(load)

# Kubernetes CronJob配置
"""
apiVersion: batch/v1
kind: CronJob
metadata:
  name: connection-scaler
spec:
  schedule: "*/5 * * * *"  # 每5分钟检查一次
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: scaler
            image: ml-platform/connection-scaler:latest
            command: ["python", "automation/connection_scaler.py"]
            env:
            - name: PROMETHEUS_URL
              value: "http://prometheus:9090"
            - name: DB_CONFIG
              valueFrom:
                secretKeyRef:
                  name: db-admin-secret
                  key: config
          restartPolicy: OnFailure
"""

{
  "dashboard": {
    "title": "ML Platform - PostgreSQL Connection Pool",
    "panels": [
      {
        "id": 1,
        "title": "连接池状态",
        "type": "graph",
        "targets": [
          {
            "expr": "db_pool_size{job=\"ml-training\"}",
            "legendFormat": "Pool Size"
          },
          {
            "expr": "db_active_connections{job=\"ml-training\"}",
            "legendFormat": "Active"
          },
          {
            "expr": "db_waiting_connections{job=\"ml-training\"}",
            "legendFormat": "Waiting"
          }
        ],
        "yAxes": [
          {
            "label": "连接数",
            "min": 0,
            "max": 50
          }
        ]
      },
      {
        "id": 2,
        "title": "连接使用时长分布",
        "type": "heatmap",
        "targets": [
          {
            "expr": "rate(db_connection_duration_seconds_bucket[5m])",
            "format": "heatmap"
          }
        ]
      },
      {
        "id": 3,
        "title": "连接泄漏检测",
        "type": "singlestat",
        "targets": [
          {
            "expr": "increase(db_connection_leaked_total[1h])",
            "legendFormat": "Leaked Connections"
          }
        ],
        "thresholds": "0,5,10",
        "colors": ["green", "yellow", "red"]
      }
    ],
    "annotations": [
      {
        "name": "Deployments",
        "datasource": "Prometheus",
        "expr": "changes(kube_pod_container_status_restarts_total{pod=~\"ml-training-.*\"}[1m]) > 0",
        "tagKeys": "pod",
        "textFormat": "{{ pod }} 重启"
      }
    ]
  }
}

# 测试配置
test_config = {
    "baseline": {
        "max_connections": 200,
        "direct_connect": True,
        "pool_size": 50
    },
    "optimized": {
        "max_connections": 500,
        "pgbouncer_enabled": True,
        "pool_mode": "transaction",
        "pool_size": 5
    }
}

# 测试结果数据（100次训练任务平均）
results_table = """
| 阶段 | 连接建立时间 | 查询P50 | 查询P99 | 任务完成时间 | 连接峰值 |
| :--- | :--- | :--- | :--- | :--- | :--- |
| Baseline | 45.2ms | 12ms | 450ms | 1h 23min | 198 |
| Optimized | 0.8ms | 8ms | 15ms | 58min | 48 |
| 提升 | 56.5x | 1.5x | 30x | 1.43x | 4.1x |
"""

# 成本效益分析
cost_benefit = """
硬件成本节约：
- 避免购买4台32核64GB数据库服务器：约 4 × $5000 = $20,000
- 年维护成本节约：约 $5,000

人力成本：
- 调优投入：2人 × 3周 = 120人时
- 按$100/小时计算：$12,000

ROI计算：
- 首年ROI = ($25,000 - $12,000) / $12,000 = 108%
- 后续年份ROI = $25,000 / $2,000 = 1150% (维护成本)
"""

# 混沌实验配置
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: pgbouncer-delay
spec:
  action: delay
  mode: one
  selector:
    labelSelectors:
      app: pgbouncer
  delay:
    latency: "100ms"
  duration: "10m"
---
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: kill-postgres-pod
spec:
  action: pod-kill
  mode: fixed-percent
  value: "33"
  selector:
    labelSelectors:
      app: postgresql
  duration: "5m"