通过 Pulumi 管理数据库迁移任务：在编排层处理 schema 变更与连接池耗尽风险的联动问题

数据库迁移这件事，说白了就是把旧表结构变成新表结构。听起来很简单，跑个 SQL 就行了。但当你把迁移任务交给编排工具去管的时候，事情就变得有趣了——尤其是当你的应用连接池正好卡在那个间隙里。

本文聚焦一个很实际的问题：当你用 Pulumi 去管理数据库迁移任务时，如何在编排层同时处理好 schema 变更和连接池耗尽之间的联动风险。这不是纸上谈兵，而是一个在真实生产环境中很容易踩坑的坑。

一、为什么数据库迁移不能只管 SQL

传统做法是写个迁移脚本，然后在 CI/CD 流水线里跑一下。SQL 跑完了，应用更新了，完事。但这里有个隐藏的大坑：

假设你的应用跑在一个微服务里，数据库连接池默认配了 20 个连接。现在你执行一条 ALTER TABLE 语句，这个操作需要拿行锁或者表锁。如果应用正好在读写这个表，那些连接就会被阻塞。20 个连接很快就全堵住了，应用侧的线程全部挂起等待数据库响应。

这时候，连接池彻底耗尽。新的请求进不来，服务直接挂了。最要命的是，如果你用的是 Flyway 或者 Liquibase 这类迁移工具，它也是通过数据库连接来执行的。连接池一满，迁移脚本自己也拿不到连接，整个流水线卡死。

这就是 schema 变更和连接池耗尽之间的联动关系。它们不是两个独立的问题，而是互相影响的。

二、Pulumi 编排层的核心思路

Pulumi 是一个基础设施即代码工具，用编程语言（TypeScript、Python、Go 等）来定义和管理基础设施资源。它的强项在于把基础设施的声明变成可编程的——你可以写条件判断、循环、错误处理，这些在传统基础设施工具里是做不到的。

把数据库迁移放进 Pulumi 管理，好处是你可以把 schema 变更和其他步骤串起来，形成一个有状态的编排流程。具体到本文的主题，核心思路是这样的：

把 schema 变更拆成多个阶段，而不是一个大 SQL。在每个阶段之间插入连接池健康检查。如果连接池出现异常，暂停后续阶段。连接池恢复正常后，再继续执行。

说白了，就是在 schema 变更和连接池之间加一个"缓冲带"，让两个操作之间有一个安全的过渡期。

2.1 Pulumi 编程模型简介

在开始写代码之前，先简单说一下 Pulumi 的几个核心概念，后面示例里会用到。

Pulumi 里的一切都是 Resource（资源），每个资源都有类型和属性。资源之间可以建立依赖关系，比如一个数据库表依赖数据库实例先创建好。Pulumi 用 Output 来包裹异步的值，因为资源创建是异步的，你不能直接访问属性，得通过 Output.apply 来操作。

// Pulumi 基础示例：理解 Output 的用法
import * as pulumi from "@pulumi/pulumi";

// 创建一个配置项
const config = new pulumi.Config();
const dbHost = config.require("dbHost");

// 这里 dbHost 是一个 Output<string>，不是直接的字符串
// 要操作它的值，需要用 apply 方法
const displayHost = dbHost.apply((host) => {
  // 这个回调函数在 host 值真正解析出来后执行
  console.log(`数据库地址是: ${host}`);
  return host;
});

// 导出资源，Pulumi 会自动跟踪依赖关系
export const myDbHost = displayHost;

Output 这个概念是关键。因为 Pulumi 的资源创建是异步的，你写代码的时候资源可能还没创建好，所以所有资源的属性都通过 Output 来访问。后面的迁移编排逻辑会大量用到 Output.apply 和 Output.all 这些方法。

三、示例：用 Pulumi 编排一次完整的迁移流程

下面用 TypeScript 写一个完整的 Pulumi 程序，演示如何编排数据库迁移，并在过程中监控连接池状态。

技术栈：Pulumi (TypeScript) + PostgreSQL

3.1 项目初始化

# 创建项目目录并进入
mkdir db-migration-orchestrator
cd db-migration-orchestrator

# 用 Pulumi 初始化 TypeScript 项目
pulumi new typescript

# 安装需要的依赖包
npm install @pulumi/pulumi @pulumi/postgresql

初始化之后，项目结构会包含 Pulumi.yaml、index.ts 等文件。

// Pulumi.yaml - 项目配置文件
{
  "name": "db-migration-orchestrator",
  "runtime": "nodejs"
}

3.2 数据库连接池健康检查模块

这是整个编排流程的核心模块，负责检查连接池状态。

import * as pulumi from "@pulumi/pulumi";
import { execSync } from "child_process";

// ========== 连接池健康检查结果类型 ==========
// 定义了健康检查返回的数据结构
interface PoolHealthResult {
  healthy: boolean;          // 连接池是否健康
  activeConnections: number; // 当前活跃连接数
  maxConnections: number;    // 最大连接数
  checkTime: string;         // 检查时间戳
}

// ========== 基础连接池健康检查函数 ==========
// 这个函数通过查询 PostgreSQL 的 pg_stat_activity 表来获取连接池状态
// pg_stat_activity 是 PostgreSQL 内置的视图，记录了所有活跃的后端进程信息
async function checkConnectionPoolHealth(
  host: string,
  port: number,
  user: string,
  password: string,
  db: string
): Promise<PoolHealthResult> {
  // 查询当前数据库的活跃连接数
  // 排除 idle in transaction 状态的连接（这些连接已经空闲了，不算活跃）
  const query = `
    SELECT count(*) as active_count 
    FROM pg_stat_activity 
    WHERE datname = '${db}' 
      AND state != 'idle'
      AND state != 'idle in transaction';
  `;

  try {
    // 使用 psql 命令行工具执行查询
    // 这里假设运行环境安装了 PostgreSQL 客户端
    const result = execSync(
      `PGPASSWORD=${password} psql -h ${host} -p ${port} -U ${user} -d ${db} -t -c "${query}"`,
      { encoding: "utf-8", timeout: 15000 } // 设置 15 秒超时
    ).trim();

    const activeConnections = parseInt(result);
    
    // 查询数据库配置的最大连接数
    const maxQuery = `SHOW max_connections;`;
    const maxResult = execSync(
      `PGPASSWORD=${password} psql -h ${host} -p ${port} -U ${user} -d ${db} -t -c "${maxQuery}"`,
      { encoding: "utf-8", timeout: 10000 }
    ).trim();
    const maxConnections = parseInt(maxResult);

    // 健康判定标准：活跃连接数不超过最大连接数的 80%
    // 留出 20% 的余量给迁移操作和其他紧急操作
    const healthy = activeConnections < maxConnections * 0.8;

    return {
      healthy,
      activeConnections,
      maxConnections,
      checkTime: new Date().toISOString(),
    };
  } catch (error) {
    // 查询失败说明可能连不上数据库，或者连接池已经耗尽
    return {
      healthy: false,
      activeConnections: -1, // -1 表示获取失败
      maxConnections: 0,
      checkTime: new Date().toISOString(),
    };
  }
}

这个函数通过查询 PostgreSQL 系统视图 pg_stat_activity 来获取当前数据库的活跃连接数。健康判定的标准是活跃连接数不超过最大连接数的 80%，留出余量给迁移操作。

3.3 完整迁移编排流程

import * as pulumi from "@pulumi/pulumi";
import * as postgres from "@pulumi/postgresql";

// ========== 读取 Pulumi 配置 ==========
const config = new pulumi.Config();

const dbHost = config.require("dbHost");
const dbPort = config.requireNumber("dbPort", 5432);
const dbUser = config.require("dbUser");
const dbPassword = config.requireSecret("dbPassword");
const dbName = config.require("dbName");

// ========== 创建数据库实例（如果不存在） ==========
const database = new postgres.Database("orchestration_db", {
  name: dbName,
}, {
  // 使用远程数据库，不创建新的数据库实例
  // 如果是本地数据库，可以指定 host、port、username 等
  provider: undefined, 
});

// ========== 步骤一：迁移前连接池检查 ==========
// 在执行任何 schema 变更前，先确认连接池状态正常
const preMigrationCheck = database.name.apply(async (dbNameValue) => {
  const [host, user, password] = await pulumi.output([
    dbHost,
    dbUser,
    dbPassword,
  ]).promise();

  return checkConnectionPoolHealth(host, dbPort, user, password, dbNameValue);
});

// 根据检查结果决定是否继续迁移
const shouldProceed = preMigrationCheck.apply((result) => {
  if (!result.healthy) {
    pulumi.log.warn(
      `迁移前连接池状态异常：活跃连接 ${result.activeConnections}/${result.maxConnections}，暂停迁移`
    );
  }
  return result.healthy;
});

// ========== 步骤二：执行 Schema 变更 ==========
// 使用 postgres.Query 资源来执行 SQL 迁移
// 注意：每个迁移步骤独立定义，而不是一个大全量 SQL
const schemaMigration = new postgres.Query("add_email_verified_column", {
  queries: [
    // 先添加新列，使用 IF NOT EXISTS 避免重复添加
    // 设置 DEFAULT FALSE，不影响现有数据行
    "ALTER TABLE users ADD COLUMN IF NOT EXISTS email_verified BOOLEAN DEFAULT FALSE;",
    
    // 为新增列添加注释，方便后续维护
    "COMMENT ON COLUMN users.email_verified IS '用户邮箱是否已验证';",
  ],
  // autoApply 设置为 false，让 Pulumi 自动管理执行时机
  autoApply: false,
}, {
  // 确保数据库资源创建好之后再执行迁移
  dependsOn: [database],
});

// ========== 步骤三：迁移后连接池检查 ==========
// 等待 schema 变更执行完毕，再检查连接池是否恢复正常
const postMigrationCheck = schemaMigration.id.apply(async () => {
  const dbNameValue = await database.name.promise();
  const [host, user, password] = await pulumi.output([
    dbHost,
    dbUser,
    dbPassword,
  ]).promise();

  // 等待 30 秒让数据库连接池有时间恢复
  // 这个等待时间可以根据实际情况调整
  // 如果迁移涉及大量数据操作，可能需要更长时间
  const waitTime = 30000;
  pulumi.log.info(`等待 ${waitTime / 1000} 秒让连接池稳定...`);
  await new Promise((resolve) => setTimeout(resolve, waitTime));

  // 检查连接池状态
  return checkConnectionPoolHealth(host, dbPort, user, password, dbNameValue);
});

// ========== 步骤四：数据回填（条件执行） ==========
// 只有连接池恢复正常，才执行耗时的数据操作
const dataBackfill = postMigrationCheck.apply(async (health) => {
  if (!health.healthy) {
    // 连接池未恢复，抛出异常终止迁移流程
    throw new Error(
      `连接池未恢复，活跃连接: ${health.activeConnections}/${health.maxConnections}，暂停数据回填`
    );
  }

  // 连接池正常，执行数据回填
  // 这里更新所有用户的 email_verified 字段
  return new postgres.Query("backfill_email_verified", {
    queries: [
      // 将 NULL 值更新为 FALSE（理论上不应该有 NULL，但做兜底）
      "UPDATE users SET email_verified = FALSE WHERE email_verified IS NULL;",
    ],
    autoApply: false,
  });
});

// ========== 导出关键状态到 Pulumi 控制台 ==========
// 这些导出的值可以在 pulumi state 命令中查看
export const migrationPhase1Result = preMigrationCheck;
export const migrationPhase2Result = postMigrationCheck;
export const migrationDecision = shouldProceed;

这个编排流程把一次数据库迁移拆成了四个步骤：迁移前检查、执行 schema 变更、迁移后检查、数据回填。每个步骤之间都有连接池状态检查，防止 schema 变更把连接池堵死。

3.4 带重试机制的连接池监控

在实际生产环境中，连接池的恢复不会那么顺利。一个临时的网络抖动或者事务还没提交，都可能导致检查失败。所以需要重试机制。

import * as pulumi from "@pulumi/pulumi";

// ========== 带指数退避的连接池检查 ==========
// 最多重试指定次数，每次间隔递增（指数退避）
// 这种策略的好处是：如果是临时的连接池拥堵，几秒后就能恢复；
// 如果是持续性的问题，不会一直重试浪费时间
async function checkConnectionPoolWithRetry(
  host: string,
  port: number,
  user: string,
  password: string,
  db: string,
  maxRetries: number = 5,
  initialWaitMs: number = 2000
): Promise<PoolHealthResult> {
  let lastResult: PoolHealthResult | null = null;

  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    // 执行一次健康检查
    const result = await checkConnectionPoolHealth(host, port, user, password, db);
    lastResult = result;

    // 连接池健康，直接返回结果
    if (result.healthy) {
      if (attempt > 1) {
        pulumi.log.info(`连接池在第 ${attempt} 次重试后恢复健康`);
      }
      return result;
    }

    // 连接池不健康，但不是最后一次尝试
    if (attempt < maxRetries) {
      // 计算等待时间：指数退避
      // 第1次: 2秒, 第2次: 4秒, 第3次: 8秒, 第4次: 16秒, 第5次: 32秒
      const waitTime = initialWaitMs * Math.pow(2, attempt - 1);
      pulumi.log.info(
        `连接池未恢复（活跃连接: ${result.activeConnections}/${result.maxConnections}），` +
        `第 ${attempt}/${maxRetries} 次重试，等待 ${waitTime / 1000} 秒...`
      );
      await new Promise((resolve) => setTimeout(resolve, waitTime));
    }
  }

  // 所有重试都失败了
  pulumi.log.error(
    `连接池检查失败，已重试 ${maxRetries} 次，最后状态: ` +
    `活跃连接 ${lastResult?.activeConnections}/${lastResult?.maxConnections}`
  );
  return lastResult!;
}

这个重试函数用了指数退避策略。第一次等 2 秒，第二次等 4 秒，第三次等 8 秒，以此类推。指数退避的好处是，临时的拥堵可以快速恢复，持续性问题不会浪费太多时间。

3.5 熔断器模式：防止雪崩

当连接池反复检查失败时，需要快速失败，而不是让所有操作都卡住。这就是熔断器（Circuit Breaker）模式。

import * as pulumi from "@pulumi/pulumi";

// ========== 连接池熔断器实现 ==========
// 熔断器有三个状态：
// CLOSED（正常）：所有请求都通过
// OPEN（熔断）：所有请求都快速失败
// HALF_OPEN（半开）：允许少量试探性请求，成功则恢复到 CLOSED
class ConnectionPoolCircuitBreaker {
  private failures: number = 0;
  private successes: number = 0;
  private state: "CLOSED" | "OPEN" | "HALF_OPEN" = "CLOSED";
  private lastFailureTime: number = 0;
  private readonly failureThreshold: number;   // 失败阈值
  private readonly successThreshold: number;   // 成功阈值
  private readonly recoveryTimeout: number;    // 恢复超时时间

  constructor(
    failureThreshold: number = 3,    // 连续失败 3 次触发熔断
    successThreshold: number = 2,    // 连续成功 2 次恢复
    recoveryTimeout: number = 60000  // 熔断 60 秒后尝试恢复
  ) {
    this.failureThreshold = failureThreshold;
    this.successThreshold = successThreshold;
    this.recoveryTimeout = recoveryTimeout;
  }

  // 检查是否允许继续操作
  async shouldAllowRequest(): Promise<{ allowed: boolean; reason: string }> {
    switch (this.state) {
      case "CLOSED":
        // 正常状态，允许所有请求
        return { allowed: true, reason: "熔断器关闭，一切正常" };

      case "OPEN":
        // 熔断状态，检查是否到了恢复时间
        if (Date.now() - this.lastFailureTime > this.recoveryTimeout) {
          // 到了恢复时间，切换到半开状态
          this.state = "HALF_OPEN";
          pulumi.log.warn("熔断器从 OPEN 切换到 HALF_OPEN，尝试恢复");
          return { allowed: true, reason: "熔断器半开，允许试探性请求" };
        }
        // 还没到恢复时间，快速失败
        return { allowed: false, reason: "熔断器打开，快速失败" };

      case "HALF_OPEN":
        // 半开状态，允许少量请求
        return { allowed: true, reason: "熔断器半开，允许试探性请求" };
    }
  }

  // 记录操作成功
  recordSuccess(): void {
    this.failures = 0;
    this.successes++;
    pulumi.log.debug(`连接池操作成功，连续成功: ${this.successes}`);

    if (this.state === "HALF_OPEN" && this.successes >= this.successThreshold) {
      // 半开状态下连续成功，恢复到正常状态
      this.state = "CLOSED";
      this.successes = 0;
      pulumi.log.info("熔断器恢复到 CLOSED 状态，连接池恢复正常");
    }
  }

  // 记录操作失败
  recordFailure(): void {
    this.failures++;
    this.lastFailureTime = Date.now();
    this.successes = 0;
    pulumi.log.debug(`连接池操作失败，连续失败: ${this.failures}`);

    if (this.state === "HALF_OPEN") {
      // 半开状态下失败，立即回到熔断状态
      this.state = "OPEN";
      pulumi.log.warn("半开状态下失败，熔断器回到 OPEN 状态");
    } else if (this.failures >= this.failureThreshold) {
      // 达到失败阈值，触发熔断
      this.state = "OPEN";
      pulumi.log.warn(
        `熔断器打开，连续失败 ${this.failures} 次，` +
        `${this.recoveryTimeout / 1000} 秒后尝试恢复`
      );
    }
  }

  // 获取当前状态（用于日志和监控）
  getState(): string {
    return `${this.state} (失败: ${this.failures}, 成功: ${this.successes})`;
  }
}

// ========== 使用熔断器包装连接池检查 ==========
const circuitBreaker = new ConnectionPoolCircuitBreaker(
  3,   // 连续失败 3 次触发熔断
  2,   // 连续成功 2 次恢复
  60000 // 熔断 60 秒后尝试恢复
);

async function safeConnectionPoolCheck(
  host: string,
  port: number,
  user: string,
  password: string,
  db: string
): Promise<PoolHealthResult> {
  // 先检查熔断器状态
  const { allowed, reason } = await circuitBreaker.shouldAllowRequest();

  if (!allowed) {
    // 熔断器打开，快速返回失败结果，不执行实际检查
    pulumi.log.warn(`熔断器阻止检查: ${reason}`);
    return {
      healthy: false,
      activeConnections: -1,
      maxConnections: 0,
      checkTime: new Date().toISOString(),
    };
  }

  try {
    // 执行实际的健康检查（带重试）
    const result = await checkConnectionPoolWithRetry(
      host, port, user, password, db, 3
    );
    // 记录成功
    circuitBreaker.recordSuccess();
    return result;
  } catch (error) {
    // 记录失败
    circuitBreaker.recordFailure();
    pulumi.log.error(
      `连接池检查异常: ${error}`, 
      `熔断器状态: ${circuitBreaker.getState()}`
    );
    throw error;
  }
}

熔断器模式的核心思想是：当连接池反复出问题的时候，不要每次都去重试，而是快速失败，让上层逻辑决定是回滚还是跳过。等过了一段时间（恢复超时），再允许少量试探性请求，确认连接池确实恢复了才恢复正常操作。

四、Schema 变更与连接池耗尽的联动风险详解

理解了编排流程之后，让我们深入分析一下为什么这两个问题会联动，以及背后的技术原理。

4.1 锁机制是连接的罪魁祸首

数据库的锁机制是导致联动风险的根本原因。当你对一张表执行 DDL（数据定义语言）操作时，比如 ALTER TABLE，数据库需要获取排他锁。这意味着在这把锁释放之前，所有试图访问这张表的操作都会被阻塞。

如果你用的是 PostgreSQL，ALTER TABLE 通常会获取 ACCESS EXCLUSIVE 锁，这是最严格的锁级别。它阻塞所有其他操作，包括读操作。MySQL 的 ALTER TABLE 也会获取类似的排他锁，虽然 MySQL 在 8.0 之后引入了一些在线 DDL 优化，但大表的 ALTER TABLE 仍然会持有长时间的锁。

关键在于：这些锁不会立即获取。如果表上已经有事务持有了其他级别的锁，ALTER TABLE 会先等待。而等待的过程中，数据库连接就一直被占用着。你的应用连接池里的连接，一个接一个地被锁住，最终全部耗尽。

4.2 连接池的雪崩效应

假设你的应用有 20 个数据库连接，连接池设置为最大 20。现在发生了这样一幕：

应用正常工作时，连接池里有 15 个活跃连接在处理请求。这时候你执行 ALTER TABLE，需要排他锁。排他锁要等现有事务完成，所以 ALTER TABLE 占住 1 个连接在等待。新的请求进来，拿到第 16 个连接，试图读表，发现排他锁没释放，也开始等待。请求越来越多，连接被逐渐耗尽。当第 20 个连接也被占用后，连接池满了，新请求直接报错。

这就是雪崩效应。一个 ALTER TABLE 操作，把整个连接池拖垮了。而且这个过程往往在毫秒级别就完成了，等你的监控告警出来，服务已经挂了。

4.3 迁移工具自己也变成受害者

更讽刺的是，如果你用 Flyway 或 Liquibase 来执行迁移，它们也是通过数据库连接来跑 SQL 的。当连接池满了的时候，迁移工具自己也拿不到连接，整个迁移流程就卡住了。

这就是为什么在编排层处理这个问题很重要——你需要在 schema 变更之前先评估连接池状态，在变更之后验证连接池是否恢复，而且要有一个兜底方案来处理异常情况。编排层是连接池和迁移任务之间的唯一桥梁，也是最合适的干预点。

五、如何避免联动风险：策略与实践

5.1 分批迁移而非一次性大改

不要把所有 schema 变更塞在一个事务里。把它拆成多个小步骤，每个步骤之间做连接池检查。

// 将大迁移拆分为小步骤，每步之间检查连接池
const migrationSteps: Array<{
  description: string;
  sql: string;
  needsPoolCheck: boolean; // 执行前是否需要检查连接池
  needsPoolCheckAfter: boolean; // 执行后是否需要检查连接池
}> = [
  {
    description: "添加新列（兼容旧代码）",
    sql: "ALTER TABLE users ADD COLUMN IF NOT EXISTS email_verified BOOLEAN DEFAULT FALSE;",
    needsPoolCheck: true,
    needsPoolCheckAfter: true,
  },
  {
    description: "添加索引（低优先级，可以在空闲时间执行）",
    sql: "CREATE INDEX IF NOT EXISTS idx_users_email_verified ON users(email_verified);",
    needsPoolCheck: true,
    needsPoolCheckAfter: true,
  },
  {
    description: "数据回填（耗时的 UPDATE 操作）",
    sql: "UPDATE users SET email_verified = TRUE WHERE verified_date IS NOT NULL;",
    needsPoolCheck: true,
    needsPoolCheckAfter: true,
  },
  {
    description: "移除旧列（确认应用已切换后再执行）",
    sql: "ALTER TABLE users DROP COLUMN IF EXISTS verified_date;",
    needsPoolCheck: true,
    needsPoolCheckAfter: false,
  },
];

// 按顺序执行每个步骤
for (let i = 0; i < migrationSteps.length; i++) {
  const step = migrationSteps[i];
  pulumi.log.info(`执行步骤 ${i + 1}/${migrationSteps.length}: ${step.description}`);

  // 执行前检查连接池
  if (step.needsPoolCheck) {
    const preCheck = await safeConnectionPoolCheck(
      dbHost, dbPort, dbUser, dbPassword.get(), dbName
    );
    if (!preCheck.healthy) {
      throw new Error(
        `步骤 ${i + 1} 执行前连接池不健康，暂停迁移: ` +
        `活跃连接 ${preCheck.activeConnections}/${preCheck.maxConnections}`
      );
    }
  }

  // 执行当前步骤的 SQL
  // 使用 postgres.Query 资源，设置 autoApply 让 Pulumi 自动执行
  new postgres.Query(`migration_step_${i + 1}`, {
    queries: [step.sql],
    autoApply: true,
  });

  // 执行后检查连接池
  if (step.needsPoolCheckAfter) {
    // 等待连接池稳定
    await new Promise((resolve) => setTimeout(resolve, 5000));
    const postCheck = await safeConnectionPoolCheck(
      dbHost, dbPort, dbUser, dbPassword.get(), dbName
    );
    if (!postCheck.healthy) {
      throw new Error(
        `步骤 ${i + 1} 执行后连接池未恢复，暂停迁移: ` +
        `活跃连接 ${postCheck.activeConnections}/${postCheck.maxConnections}`
      );
    }
    pulumi.log.info(`步骤 ${i + 1} 完成，连接池健康: ` +
      `${postCheck.activeConnections}/${postCheck.maxConnections}`);
  }
}

5.2 使用 Online Schema Change 工具

如果你的数据库支持 Online Schema Change，可以大大减少锁的持有时间。PostgreSQL 可以用 pg_partman 或者 gh-ost（MySQL 专用），这些工具的核心原理是创建一个新表，把旧数据分批复制到新表，然后快速切换表名。

// 示例：使用 gh-ost 进行在线迁移
// gh-ost 会创建一个新表，复制数据，然后切换表名
// 这个过程对旧表的锁持有时间极短（只有最终的 rename 操作需要排他锁）
import { execSync } from "child_process";

function executeOnlineMigrationWithGhOst(
  table: string,
  alterSql: string,
  host: string,
  port: number,
  user: string,
  password: string,
  db: string
): void {
  // gh-ost 命令示例
  // 关键参数说明：
  // --chunk-size=1000: 每次复制 1000 行，避免大批量操作锁表
  // --max-lag-millis=1500: 如果主从延迟超过 1.5 秒，暂停复制，防止影响线上
  // --execute: 确认执行（不加这个参数只是预检查）
  const ghOstCommand = `
    gh-ost \\
      --host="${host}" \\
      --port="${port}" \\
      --user="${user}" \\
      --password="${password}" \\
      --database="${db}" \\
      --table="${table}" \\
      --alter="${alterSql}" \\
      --chunk-size=1000 \\
      --max-lag-millis=1500 \\
      --assume-rbr \\
      --execute
  `;

  pulumi.log.info(`开始在线迁移表 ${table}: ${alterSql}`);
  execSync(ghOstCommand, {
    stdio: "inherit",
    timeout: 600000, // 最长 10 分钟超时
  });
  pulumi.log.info(`在线迁移完成: ${table}`);
}

gh-ost 的巧妙之处在于，它不是在原表上做 ALTER TABLE，而是创建一个影子表，分批复制数据，最后用 rename 操作来切换。这个 rename 操作虽然也需要排他锁，但持有时间极短（毫秒级别），对连接池的影响微乎其微。

5.3 动态调整连接池大小

在 schema 变更期间，可以临时调大连接池上限，给迁移操作留出足够的连接。

// 在 Pulumi 中管理连接池配置
// 根据迁移阶段动态调整连接池大小
const config = new pulumi.Config();

// 迁移前的连接池配置
const normalPoolSize = config.getNumber("normalPoolSize") || 20;
// 迁移时的连接池配置（增大以容纳迁移操作）
const migrationPoolSize = config.getNumber("migrationPoolSize") || 50;

// 使用 Kubernetes ConfigMap 注入连接池配置
// 假设应用部署在 K8s 上
const poolConfig = new pulumi.Output.tuple(normalPoolSize, migrationPoolSize)
  .apply(([normal, migration]) => ({
    // 正常状态的连接池配置
    normal: {
      maxPoolSize: normal,
      minIdle: 5,
      connectionTimeout: 10000,   // 10秒获取连接超时
      idleTimeout: 60000,         // 空闲连接 60秒后回收
    },
    // 迁移期间的连接池配置
    duringMigration: {
      maxPoolSize: migration,
      minIdle: 10,
      connectionTimeout: 30000,   // 30秒超时，给慢操作更多时间
      idleTimeout: 120000,        // 空闲连接 120秒后回收
      // 增加等待队列长度，允许更多请求排队
      maxWaitQueueSize: 100,
    },
  }));

// 导出连接池配置
export const connectionPoolConfig = poolConfig;

调大连接池不是万能药，它只是买时间。如果 schema 变更本身就有问题（比如锁等待时间过长），连接池再大也会被耗尽。所以连接池调整要和分批迁移、在线迁移工具配合使用。

六、应用场景

这套方法适用于以下几种场景：

6.1 微服务架构中的数据库迁移

微服务各自管理自己的数据库，但共享基础设施。当某个微服务需要 schema 变更时，如果连接池被占满，不仅影响自己，还可能影响依赖它连接池的其他服务。Pulumi 编排可以确保每次迁移前做健康检查，迁移后验证恢复。

6.2 大规模数据迁移

当你的表里有上亿行数据，ALTER TABLE 操作可能持续几分钟甚至更久。这种情况下，连接池被长时间占用的风险非常高。分批执行、在线迁移工具、连接池大小动态调整这些策略组合使用，可以大幅降低风险。

6.3 多环境部署

生产环境、预发布环境、测试环境各有不同的数据库。用 Pulumi 统一管理，可以在每个环境部署时自动执行对应的迁移，同时检查连接池状态。环境之间的差异也可以自动适配，比如生产环境连接池更小，检查阈值就需要相应调整。

七、技术优缺点

优点

第一，声明式管理。Pulumi 用代码描述基础设施状态，schema 迁移也是代码的一部分，方便版本控制和代码审查。团队里每个人都能看到迁移逻辑的变化历史。

第二，可编程的编排能力。相比 Terraform，Pulumi 支持完整的编程能力，可以写条件判断、循环、重试逻辑，这些都是处理联动风险所需要的。Terraform 的资源依赖是声明式的，没法写复杂的条件逻辑。

第三，自动回滚。Pulumi 内置状态管理，如果某一步失败，可以自动回滚到之前的状态。不需要手动编写回滚脚本。

第四，跨平台。不管你的数据库是 AWS RDS、阿里云 RDS、还是自建服务器，Pulumi 都能统一管理。连接池检查的逻辑可以复用，只需要调整参数。

缺点

第一，学习曲线。Pulumi 有自己的编程模型，尤其是 Output、Resource、Dependency 这些概念，新手需要花时间理解。Output 的异步模型尤其容易让人困惑。

第二，复杂度高。把数据库迁移放进基础设施编排工具里，增加了整个系统的复杂度。简单的场景用 Flyway 或 Liquibase 就够了，没必要上 Pulumi。

第三，调试困难。Pulumi 的状态管理是自动的，但当出问题的时候，调试会比传统方式复杂。你需要理解 Pulumi 的状态存储机制和部署模型。

第四，连接池检查依赖外部工具。示例中用了 psql 命令行工具来检查连接池，这要求运行环境必须安装 PostgreSQL 客户端。在生产环境中，可能需要用其他方式替代，比如用数据库 SDK 直接查询。

八、注意事项

8.1 不要在生产环境直接测试

先把编排流程在测试环境跑通，确认每个阶段的连接池检查都能正常工作，再迁移到生产环境。特别是熔断器和重试的参数，需要在测试环境中调优。

8.2 设置合理的超时时间

所有的数据库操作和连接池检查都应该设置超时。

// 连接池检查超时设置示例
// 超时时间不要设得太长，否则迁移流程会一直卡着
const CHECK_TIMEOUT = 30000; // 30秒超时

async function checkWithTimeout(
  host: string,
  port: number,
  user: string,
  password: string,
  db: string,
  maxConnections: number
): Promise<PoolHealthResult> {
  return Promise.race([
    // 正常检查流程
    checkConnectionPoolWithRetry(
      host, port, user, password, db, 3
    ),
    // 超时控制：30秒后强制中断
    new Promise<never>((_, reject) =>
      setTimeout(
        () => reject(new Error(`连接池检查超时（${CHECK_TIMEOUT}ms）`)),
        CHECK_TIMEOUT
      )
    ),
  ]);
}

8.3 监控和告警要跟上

编排层的每个检查点都应该有日志输出，关键指标（活跃连接数、锁等待时间、迁移步骤耗时）应该上报到监控系统中。否则出了问题你根本不知道是在哪个环节。

8.4 注意数据库版本的兼容性

不同的数据库版本，锁的行为可能不同。PostgreSQL 13 和 16 在 ALTER TABLE 的实现上就有差异。做 schema 变更之前，先确认数据库版本支持你需要的操作。

8.5 备份永远不要省

不管编排流程设计得多好，都不要忘了在迁移前做数据库备份。这是最后一道防线。Pulumi 也可以管理备份策略，在迁移前自动触发一次全量备份。

九、文章总结

通过 Pulumi 管理数据库迁移任务，本质上是在编排层引入了一层智能调度。schema 变更和连接池耗尽不是两个独立的问题，它们通过锁机制和连接管理紧密联动。Pulumi 的编程能力让我们可以在编排流程中嵌入健康检查、重试逻辑、熔断机制，从而把这种联动风险降到最低。

核心要点回顾：

第一，拆步骤。把大迁移拆成多个小步骤，每步之间检查连接池。

第二，加缓冲。使用指数退避、熔断器来应对临时的连接池拥堵。

第三，用工具。gh-ost 等在线迁移工具可以减少锁持有时间。

第四，调配置。迁移期间动态调整连接池大小，给迁移操作留出空间。

第五，重监控。每个检查点都要有日志和监控，出了问题要能快速定位。

数据库迁移从来不是跑个 SQL 就完事那么简单。当你开始用编排工具管理它的时候，你其实在解决一个系统工程问题。Pulumi 的编排能力，加上对连接池状态的持续监控，可以让这个过程变得更可控、更安全。

记住：最复杂的 schema 变更，往往需要最简单的保护策略——先检查，再执行，出了问题能回滚。