GCP Cloud Function中discovery.build()调用IAM服务偶发SSL错误求助

可行解决方案如下： #

1. 升级依赖库到最新稳定版 #

discovery服务的客户端稳定性和google-api-python-client、httplib2、pyopenssl的版本直接相关，旧版本可能存在SSL连接的隐性bug。在Cloud Function的requirements.txt中明确指定最新版依赖：

google-api-python-client>=2.100.0
httplib2>=0.22.0
pyopenssl>=23.2.0

重新部署函数后，这些更新后的库会修复部分SSL连接不稳定问题。

2. 自定义SSL配置与连接超时 #

构建IAM客户端时，手动配置更稳健的SSL上下文和连接超时，避免默认配置导致的连接过早断开：

import ssl
from googleapiclient.discovery import build
from googleapiclient.http import build_http

# 延迟初始化客户端（函数外全局变量）
iam_client = None

def init_iam_client():
    global iam_client
    if not iam_client:
        # 配置兼容型SSL上下文，适配部分旧SSL端点
        ssl_ctx = ssl.create_default_context()
        ssl_ctx.set_ciphers('DEFAULT@SECLEVEL=1')
        # 构建自定义HTTP对象，设置30秒超时
        http = build_http()
        http.timeout = 30
        http.ssl_context = ssl_ctx
        iam_client = build("iam", "v1", http=http)

在函数业务逻辑执行前调用init_iam_client()即可复用配置好的客户端。

3. 给核心操作添加重试机制 #

针对偶发的SSL错误，用tenacity库实现指数退避重试逻辑，自动处理临时连接问题：

import ssl
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=2, max=10),  # 指数退避等待（2s、4s、8s）
    retry=retry_if_exception_type(ssl.SSLEOFError)  # 仅针对SSLEOFError触发重试
)
def list_service_accounts(project_id):
    init_iam_client()
    response = iam_client.projects().serviceAccounts().list(
        name=f"projects/{project_id}"
    ).execute()
    return response.get('accounts', [])

遇到SSL连接错误时会自动重试，大幅提升请求成功率。

4. 绕过discovery.build，直接调用IAM REST API #

如果上述方法都无法解决问题，可以直接用requests调用IAM Admin的REST API，绕开discovery层的潜在问题：

import requests
from google.auth import default
from google.auth.transport.requests import Request

def list_service_accounts_via_rest(project_id):
    # 获取GCP默认凭据并刷新token
    credentials, _ = default()
    credentials.refresh(Request())
    # 构建请求头与API地址
    headers = {'Authorization': f'Bearer {credentials.token}'}
    api_url = f"https://iam.googleapis.com/v1/projects/{project_id}/serviceAccounts"
    # 发送请求
    response = requests.get(api_url, headers=headers, timeout=30)
    response.raise_for_status()
    return response.json().get('accounts', [])

这种方式更直接，同时借助GCP默认凭据自动完成认证，稳定性更可控。

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