AWS CDK自定义资源调用Lambda初始化RDS MySQL遇连接丢失错误求助

问题描述 #

我通过AWS CDK的database-stack创建MySQL类型的RDS实例，同时配置AWSCustomResource，希望在数据库首次部署时自动执行table-creation.sql创建表。但每次实例启动后，表都创建失败，报错信息如下：

2023-06-11T14:23:19.813Z    8155569d-cd7e-4fc4-833c-db7da6afbf3d    INFO    Error: Connection lost: The server closed the connection.
    at Protocol.end (/var/task/index.js:9350:17)
    at Socket.<anonymous> (/var/task/index.js:9907:32)
    at Socket.<anonymous> (/var/task/index.js:10233:14)
    at Socket.emit (node:events:525:35)
    at Socket.emit (node:domain:489:12)
    at endReadableNT (node:internal/streams/readable:1358:12)
    at processTicksAndRejections (node:internal/process/task_queues:83:21)
    --------------------
    at Protocol._enqueue (/var/task/index.js:9374:52)
    at Protocol.handshake (/var/task/index.js:9304:27)
    at PoolConnection.connect (/var/task/index.js:9925:22)
    at Pool.getConnection (/var/task/index.js:10330:20)
    at Pool.query (/var/task/index.js:10447:12)
    at /var/task/index.js:10997:22
    at new Promise (<anonymous>)
    at Runtime.handler (/var/task/index.js:10996:18)
    at processTicksAndRejections (node:internal/process/task_queues:96:5) {
  fatal: true,
  code: 'PROTOCOL_CONNECTION_LOST'
}

可能原因 #

• RDS实例未完全就绪：CustomResource在RDS实例创建完成后立即触发Lambda，但此时RDS可能仍在后台初始化（如数据库引擎启动、网络配置生效），导致连接超时断开。
• 网络连通性延迟：虽然配置了VPC和安全组，但规则可能未及时同步，导致Lambda与RDS建立连接的过程中被中断。
• 数据库客户端配置缺失：Lambda中的MySQL客户端未设置超时、重试机制，遇到短暂不可用时直接失败。
• RDS代理未完成关联：使用dbProxy.endpoint作为连接地址时，代理可能还未完成与RDS实例的绑定，导致连接请求被拒绝。

适配Lambda环境的解决方案 #

1. 添加RDS就绪等待机制 #

在Lambda中加入指数退避式的重试逻辑，先通过AWS SDK检查RDS实例状态为available后，再尝试建立数据库连接：

async function waitForRdsReady(dbInstanceIdentifier) {
  const rds = new AWS.RDS();
  let attempts = 0;
  const maxAttempts = 10;
  
  while (attempts < maxAttempts) {
    try {
      const response = await rds.describeDBInstances({ DBInstanceIdentifier }).promise();
      const instance = response.DBInstances[0];
      if (instance.DBInstanceStatus === 'available') {
        console.log('RDS实例已就绪');
        return;
      }
    } catch (err) {
      console.error('检查RDS状态失败:', err);
    }
    
    attempts++;
    const delay = Math.pow(2, attempts) * 1000;
    console.log(`等待${delay}ms后重试...`);
    await new Promise(resolve => setTimeout(resolve, delay));
  }
  throw new Error('RDS实例超时未就绪');
}

2. 优化MySQL客户端配置 #

调整Lambda中的数据库连接参数，增加超时时间和重试策略：

const connection = await mysql.createConnection({
  host: process.env.DB_ENDPOINT_ADDRESS,
  user: secret.username,
  password: secret.password,
  database: process.env.DB_NAME,
  connectTimeout: 10000, // 10秒连接超时
  acquireTimeout: 10000,
});

3. 显式声明CDK资源依赖 #

在database-stack.ts中，确保CustomResource等待RDS实例和代理完全就绪后再执行：

rdsInitCustomResource.node.addDependency(instance);
rdsInitCustomResource.node.addDependency(dbProxy);

4. 验证安全组配置 #

• 确认Lambda所在安全组允许出站访问RDS/RDS代理的3306端口
• 确认RDS实例安全组允许来自Lambda安全组的入站3306端口流量

5. 调整Lambda资源配置 #

适当提升Lambda的内存（如512MB），减少冷启动延迟，同时保持3分钟的超时设置以覆盖重试等待时间。

附相关代码 #

database-stack.ts #

// 创建用于初始化数据库的Lambda函数
const rdsLambdaFunction = new NodejsFunction(this, "rdsLambdaFN", {
    entry: "./src/lambda_functions/rds-init.ts",
    runtime: Runtime.NODEJS_16_X,
    timeout: Duration.minutes(3),
    functionName: "rds-init-function",
    environment: {
        DB_ENDPOINT_ADDRESS: dbProxy.endpoint,
        DB_NAME: "vegafoliodb",
        DB_SECRET_ARN: instance.secret?.secretFullArn || "",
    },
    vpc,
    vpcSubnets: vpc.selectSubnets({
        subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS,
    }),
    bundling: {
        commandHooks: {
            beforeBundling(inputDir: string, outputDir: string) {
                return [
                    `cp ${inputDir}/src/lambda_functions/table-creation.sql ${outputDir}`,
                ];
            },
            afterBundling() { return []; },
            beforeInstall() { return []; },
        },
        externalModules: ["aws-sdk"],
    },
    securityGroups: [dbSG],
});

// 配置CustomResource触发Lambda初始化数据库
const rdsInitCustomResource = new AwsCustomResource(this, "rdsInitCustomResource", {
    onCreate: {
        service: "Lambda",
        action: "invoke",
        parameters: {
            FunctionName: rdsLambdaFunction.functionArn,
            InvocationType: "RequestResponse",
        },
        physicalResourceId: PhysicalResourceId.of("rdsInitCustomResource"),
    },
    policy: AwsCustomResourcePolicy.fromSdkCalls({
        resources: [rdsLambdaFunction.functionArn],
    }),
    role: customResourcerole,
});

// 显式添加依赖，确保RDS就绪后再执行初始化
rdsInitCustomResource.node.addDependency(instance);
rdsInitCustomResource.node.addDependency(dbProxy);

修正后的rds-init.ts（Lambda处理代码） #

import * as mysql from 'mysql2/promise';
import * as AWS from 'aws-sdk';
import * as fs from 'fs';
import * as path from 'path';

const secretsManager = new AWS.SecretsManager();

export const handler = async () => {
  try {
    // 从Secret Manager获取数据库凭证
    const secretData = await secretsManager.getSecretValue({ SecretId: process.env.DB_SECRET_ARN }).promise();
    const secret = JSON.parse(secretData.SecretString);
    
    // 等待RDS实例就绪
    await waitForRdsReady(secret.DBInstanceIdentifier);
    
    // 建立数据库连接
    const connection = await mysql.createConnection({
      host: process.env.DB_ENDPOINT_ADDRESS,
      user: secret.username,
      password: secret.password,
      database: process.env.DB_NAME,
      connectTimeout: 10000,
    });
    
    // 读取并执行SQL脚本
    const sqlFilePath = path.join(__dirname, 'table-creation.sql');
    const sqlContent = fs.readFileSync(sqlFilePath, 'utf8');
    const queries = sqlContent.split(';').filter(q => q.trim() !== '');
    
    for (const query of queries) {
      await connection.execute(query);
    }
    
    await connection.end();
    return { status: 'success', message: '表创建成功' };
  } catch (error) {
    console.error('数据库初始化失败:', error);
    throw error;
  }
};

// 等待RDS就绪的辅助函数
async function waitForRdsReady(dbInstanceIdentifier) {
  const rds = new AWS.RDS();
  let attempts = 0;
  const maxAttempts = 10;
  
  while (attempts < maxAttempts) {
    try {
      const response = await rds.describeDBInstances({ DBInstanceIdentifier }).promise();
      const instance = response.DBInstances[0];
      if (instance.DBInstanceStatus === 'available') {
        console.log('RDS实例已就绪');
        return;
      }
    } catch (err) {
      console.error('检查RDS状态失败:', err);
    }
    
    attempts++;
    const delay = Math.pow(2, attempts) * 1000;
    console.log(`等待${delay}ms后重试...`);
    await new Promise(resolve => setTimeout(resolve, delay));
  }
  throw new Error('RDS实例超时未就绪');
}

内容的提问来源于stack exchange，提问作者Seungho Lee