第一代Cloud Function（Python3.10）连接Cloud SQL超时无详细日志问题

问题简述 #

• 第一代Cloud Function（Python 3.10）部署后仅记录Function execution started和Function execution took x ms, finished with status error，无异常详情日志
• 代码本地运行正常，切换为Unix Socket连接方式后，GCP端可正常工作，但本地无法使用
• 代码基于Cloud SQL Python Connector示例Notebook编写，核心代码如下：

import sqlalchemy
from google.cloud import secretmanager
from google.cloud.sql.connector import Connector, IPTypes

PROJECT_ID = "REDACTED"
connector = Connector()


def getconn():
    conn = connector.connect(
        "REDACTED",
        "pg8000",
        user="REDACTED",
        password="REDACTED",
        db="REDACTED",
        ip_type=IPTypes.PUBLIC,
    )
    return conn


pool = sqlalchemy.create_engine(
    "postgresql+pg8000://",
    creator=getconn,
)


def main(request):
    try:
        max_version = None
        # connect to connection pool
        with pool.connect() as db_conn:
            row = db_conn.execute(
                sqlalchemy.text(
                    "select coalesce(max(version), 0) as max_version from pricings"
                )
            ).fetchone()

            # show results
            max_version = row[0]
        return {"max_version": max_version}
    except Exception as e:
        print(e)
        return "error"

排查与解决方案 #

1. 修复日志输出问题 #

Cloud Function中print()的输出可能无法被日志系统完整捕获，改用GCP官方logging库输出异常详情，确保堆栈信息被记录：

import logging
# 初始化日志配置
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def main(request):
    try:
        # ... 原有业务代码 ...
    except Exception as e:
        # 输出完整异常信息与堆栈
        logger.error(f"执行失败: {str(e)}", exc_info=True)
        return "error"

2. 调整连接池初始化时机 #

第一代Cloud Function的全局代码（如connector = Connector()和连接池创建）会在实例启动时执行，若初始化失败不会触发main函数的异常捕获。建议将连接池初始化移至main函数内或改用懒加载：

def get_pool():
    connector = Connector()
    def getconn():
        conn = connector.connect(
            "REDACTED",
            "pg8000",
            user="REDACTED",
            password="REDACTED",
            db="REDACTED",
            ip_type=IPTypes.PUBLIC,
        )
        return conn
    return sqlalchemy.create_engine(
        "postgresql+pg8000://",
        creator=getconn,
    )

def main(request):
    try:
        pool = get_pool()
        with pool.connect() as db_conn:
            # ... 原有查询代码 ...
        return {"max_version": max_version}
    except Exception as e:
        logger.error(f"执行失败: {str(e)}", exc_info=True)
        return "error"

3. 验证权限与依赖配置 #

• 确保Cloud Function的服务账号拥有Cloud SQL Client角色，允许访问Cloud SQL实例
• 检查Cloud SQL实例公网IP的授权列表，需包含Cloud Function的出口IP或启用VPC连接器（使用私有IP时）
• 在requirements.txt中明确指定依赖版本，避免兼容性问题：

sqlalchemy>=2.0.0
google-cloud-sql-connector>=1.10.0
pg8000>=1.30.0
google-cloud-secret-manager>=2.15.0

4. 兼容本地与GCP的双连接方案 #

通过环境变量判断运行环境，自动切换连接方式，兼顾本地开发与GCP部署：

import os

def getconn():
    # 判断是否处于Cloud Function环境
    if os.getenv("FUNCTION_TARGET"):
        # GCP端使用Unix Socket连接
        return connector.connect(
            "REDACTED",
            "pg8000",
            user="REDACTED",
            password="REDACTED",
            db="REDACTED",
            unix_socket="/cloudsql/REDACTED"
        )
    else:
        # 本地使用公网IP连接
        return connector.connect(
            "REDACTED",
            "pg8000",
            user="REDACTED",
            password="REDACTED",
            db="REDACTED",
            ip_type=IPTypes.PUBLIC,
        )

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