Cloud Function捕获异常后无法继续执行问题排查求助

核心问题定位 #

第一个异常触发后程序直接终止，主要由以下几个原因导致：

• 直接通过sftp.open()获取流给pandas读取，一旦读取过程中出现网络波动、流中断，会导致全局SFTP连接失效，部分底层网络异常会超出Exception的捕获范围，直接终止进程
• Cloud Function的/tmp目录是内存映射存储，直接读取远程SFTP流加载到DataFrame时，如果单文件过大，会直接触发OOM（内存溢出），进程会被平台直接销毁，此时try-except无法拦截，程序直接终止
• SFTP连接全局复用，一旦某次文件操作导致连接断开，后续所有文件的操作都会失败，甚至出现未捕获的连接异常

修复方案 #

可以按照以下逻辑修改代码：

1. 处理每个文件前先将SFTP文件下载到本地/tmp临时目录，再读取本地文件加载到DataFrame，避免SFTP流不稳定导致的连接问题
2. 每个文件处理完成后立即删除本地临时文件，避免占用过多内存
3. 增加SFTP连接校验，发现连接断开后自动重连
4. 针对不同类型的异常统一捕获，确保所有错误都能被拦截，不影响后续文件处理

修改后可直接运行的代码 #

import pysftp
import json
import os
import pandas as pd
from google.cloud import bigquery

# 配置项单独抽离，方便修改
SFTP_CONFIG = {
    "host": "host_ip",
    "username": "user",
    "private_key": "private_key.pem",
    "remote_path": "/home/user/s/"
}
EXCLUDE_FILE_KEYWORDS = ['.ipynb', 'sgp', 'booking', 'web', 'SHIPPING', 'addresses_v2', 'fc_cust']
TEMP_DIR = "/tmp"

def get_sftp_connection():
    """单独封装SFTP连接获取逻辑，方便重连"""
    cnopts = pysftp.CnOpts()
    cnopts.hostkeys = None
    return pysftp.Connection(
        host=SFTP_CONFIG["host"],
        username=SFTP_CONFIG["username"],
        private_key=SFTP_CONFIG["private_key"],
        cnopts=cnopts
    )

def push_files():
    client = bigquery.Client()
    sftp = get_sftp_connection()
    
    # 过滤待处理文件
    files = [
        file for file in sftp.listdir(SFTP_CONFIG["remote_path"])
        if not any(keyword in file for keyword in EXCLUDE_FILE_KEYWORDS)
    ]
    print(json.dumps(dict(
        severity="NOTICE",
        message=f"待推送文件列表: {files}",
    )))

    for file in files:
        local_file_path = os.path.join(TEMP_DIR, file)
        try:
            # 先校验SFTP连接是否存活，断开则重连
            if not sftp.is_active:
                sftp = get_sftp_connection()
            
            # 先下载到本地临时目录，再读取
            sftp.get(
                remotepath=f"{SFTP_CONFIG['remote_path']}/{file}",
                localpath=local_file_path
            )
            
            df = pd.read_csv(local_file_path, sep='\t', engine='python')
            table_id = f"sftp_data.{file}"
            job = client.load_table_from_dataframe(df, table_id)
            job.result()
            
            print(json.dumps(dict(
                severity="NOTICE",
                message=f"推送成功: {file}",
            )))
        
        except Exception as e:
            print(json.dumps(dict(
                severity="ERROR",
                message=f"推送失败: {file}, 错误原因: {str(e)}",
            )))
        
        finally:
            # 无论处理成功失败，都删除本地临时文件，释放内存
            if os.path.exists(local_file_path):
                os.remove(local_file_path)

    sftp.close()
    return "文件推送任务执行完成"


def main(request):
    print('收到触发请求')
    response = push_files()
    return f'请求执行完成，结果: {response}'

额外优化建议 #

• 如果单文件大小超过你分配给Cloud Function的内存的50%，建议拆分文件处理，或者改用更高内存配置的Cloud Function运行实例
• 可以增加文件大小校验逻辑，超过阈值的文件单独处理，避免触发OOM
• 生产环境建议不要关闭SFTP的hostkey校验，避免中间人攻击风险

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