云函数在本地工作但在 GCP 中崩溃

Question

这是我的代码的超级缩略版（我会在评论中添加指向所有内容的链接）

import ...

...

def main(data, context):

    log_client = logging.Client()

    log_name = 'cloudfunctions.googleapis.com%2Fcloud-functions'

    res = Resource(type="cloud_function",

                   labels={

                       "function_name": "refresh_classes",

                       "region": os.environ.get("FUNC_REGION")

                   })

    logger = log_client.logger(log_name.format(os.environ.get("PROJECT_ID")))

    db = sqlalchemy.create_engine(

        sqlalchemy.engine.url.URL(

            drivername="mysql+pymysql",

            username=os.environ.get("DB_USER"),

            password=os.environ.get("DB_PASS"),

            host=os.environ.get("DB_HOST"),

            port=3306,

            database=PRIMARY_TABLE_NAME

        ),

        pool_size=5,

        max_overflow=2,

        pool_timeout=30,

        pool_recycle=1800

    )

    start_time = perf_counter()

    check_if_table_exists(db)

    for i in range(START_IDX, END_IDX):

        print(i)

        logger.log_text(f"Checking class with id {i}", resource=res, severity="INFO")

        ...

    logger.log_text(f"Total seconds elapsed: {perf_counter() - start_time}", resource=res, severity="INFO")

if __name__ == '__main__':

    main('data', 'context')

当我在本地运行上述云函数时，我的GOOGLE_APPLICATION_CREDENTIALS配置与我的本地 Cloud MySQL 代理一起设置，详细的云日志记录通过并且该函数顺利完成，

Answer 1

所以我终于想通了，如果不是通过 Cloud Scheduler 触发云功能，而是通过以下方式手动运行该功能Test FunctionGCP 会给你引发的异常。就我而言，我的 Cloud MySQL 连接失败Error: function terminated. Recommended action: inspect logs for termination reason. Details:(pymysql.err.OperationalError) (2003, "Can't connect to MySQL server on '**.***.**.***' (timed out)")(Background on this error at: http://sqlalche.me/e/e3q8)所以我只需要引入一个环境变量来在通过代理和unix 套接字连接之间切换，如下所示：if os.environ.get("ENV") == "local":    db = sqlalchemy.create_engine(        sqlalchemy.engine.url.URL(            drivername="mysql+pymysql",            username=os.environ.get("DB_USER"),            password=os.environ.get("DB_PASS"),            host=os.environ.get("DB_HOST"),            port=3306,            database=PRIMARY_TABLE_NAME        ),        pool_size=5,        max_overflow=2,        pool_timeout=30,        pool_recycle=1800    )else:    db = sqlalchemy.create_engine(        sqlalchemy.engine.url.URL(            drivername="mysql+pymysql",            username=os.environ.get("DB_USER"),            password=os.environ.get("DB_PASS"),            database=PRIMARY_TABLE_NAME,            query={"unix_socket": "/cloudsql/{}".format(os.environ.get("CLOUD_SQL_CONNECTION_NAME"))}        ),        pool_size=5,        max_overflow=2,        pool_timeout=30,        pool_recycle=1800    )