Google Cloud Function无法连接Cloud SQL PostgreSQL实例求助

以下是针对你遇到的加载用户代码失败问题的排查步骤和解决方案：

1. 检查函数入口是否存在 #

Cloud Functions要求必须导出一个可执行的触发器函数（HTTP/事件触发器），你当前的代码仅定义了连接池创建函数，缺少入口函数，这是导致平台无法加载代码的常见原因。补充入口示例如下：

const Knex = require('knex');

// 创建Unix Socket连接池
const createUnixSocketPool = async config => {
  return Knex({
    client: 'pg',
    connection: {
      user: '你的实例用户名',
      password: '用户密码',
      database: '你的数据库名',
      host: '/cloudsql/项目ID:southamerica-east1:实例名', // 确保此处参数完全匹配你的实例信息
    },
    ...config,
  });
};

// 导出HTTP触发器函数（示例）
exports.testCloudSQLConnection = async (req, res) => {
  let pool;
  try {
    // 初始化连接池（配置连接池大小）
    pool = await createUnixSocketPool({ pool: { min: 0, max: 5 } });
    // 执行测试查询
    const queryResult = await pool.raw('SELECT NOW()');
    res.status(200).send(`数据库连接成功，当前时间：${queryResult.rows[0].now}`);
  } catch (err) {
    res.status(500).send(`连接失败：${err.message}`);
  } finally {
    // 销毁连接池释放资源
    if (pool) await pool.destroy();
  }
};

2. 确认依赖配置正确 #

确保项目根目录的package.json中包含knex和pg（PostgreSQL驱动）依赖：

{
  "dependencies": {
    "knex": "^3.1.0",
    "pg": "^8.11.0"
  }
}

部署前执行npm install安装依赖，或确保部署时Cloud Functions能读取到package.json自动安装依赖。

3. 部署时绑定Cloud SQL实例 #

使用Unix Socket连接Cloud SQL必须在部署时将实例绑定到函数：

• 命令行部署（gcloud）：

gcloud functions deploy testCloudSQLConnection \
  --runtime nodejs20 \
  --trigger-http \
  --cloudsql-instances=你的项目ID:southamerica-east1:你的实例名 \
  --allow-unauthenticated

• 控制台部署：在「Cloud SQL连接」模块中添加你的实例，确保实例被选中。

4. 验证权限与实例状态 #

• 检查Cloud Functions默认服务账号（格式：你的项目ID@appspot.gserviceaccount.com）是否拥有Cloud SQL Client角色，在IAM控制台为该账号添加对应角色。
• 确认Cloud SQL实例处于RUNNING状态，且已创建你指定的数据库和用户。

关于Unix Socket的说明 #

Unix Socket是本地进程通信方式，GCP会自动将Cloud SQL实例的Socket路径挂载到函数运行环境中，你无需手动配置Socket，只需确保代码中host路径的项目ID、区域、实例名完全匹配即可。

其他语言示例（Python） #

import psycopg2
from psycopg2 import pool

def create_unix_socket_pool():
    # 创建连接池
    return psycopg2.pool.SimpleConnectionPool(
        minconn=1,
        maxconn=5,
        user='你的实例用户名',
        password='用户密码',
        database='你的数据库名',
        host='/cloudsql/你的项目ID:southamerica-east1:你的实例名'
    )

def test_cloud_sql(request):
    connection_pool = create_unix_socket_pool()
    conn = connection_pool.getconn()
    try:
        with conn.cursor() as cursor:
            cursor.execute('SELECT NOW();')
            result = cursor.fetchone()
            return f"数据库连接成功，当前时间：{result[0]}"
    except Exception as e:
        return f"连接失败：{str(e)}"
    finally:
        # 归还连接到池
        connection_pool.putconn(conn)

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