在Airflow(Composer)中配置CloudFunctionInvokeFunctionOperator调用云函数遇404问题

错误信息 #

line 915, in execute raise HttpError(resp, content, uri=self.uri) googleapiclient.errors.HttpError: <HttpError 404 when requesting https://cloudfunctions.googleapis.com/v1/projects/pongthorn/locations/asia-southeast1/functions/crypto-trading-to-bq:call?alt=json returned "Function crypto-trading-to-bq in region asia-southeast1 in project pongthorn does not exist". Details: "Function crypto-trading-to-bq in region asia-southeast1 in project pongthorn does not exist">

疑问 #

• 什么是function id？
• 我的云函数名称是crypto-trading-to-bq，function id和函数名称是否一致？

Airflow变量配置 #

{
"project_id": "pongthorn",
"region_name": "asia-southeast1",
"function_name": "crypto-trading-to-bq"
}

代码实现 #

import datetime

import airflow

from airflow.providers.google.cloud.operators.functions import (
    CloudFunctionDeleteFunctionOperator,
    CloudFunctionDeployFunctionOperator,
    CloudFunctionInvokeFunctionOperator,
)


YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

XProjdctID=airflow.models.Variable.get('project_id')
XRegion=airflow.models.Variable.get('region_name')
XFunction=airflow.models.Variable.get('function_name')

default_args = {
    'owner': 'Binance Trading Transaction',
    'depends_on_past': False,
    'email': [''],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'start_date': YESTERDAY,
}

with airflow.DAG(
        'bn_trading_flow',
        catchup=False,
        default_args=default_args,
        schedule_interval=datetime.timedelta(days=1)) as dag:
 
        call_crypto_trading_to_bq = CloudFunctionInvokeFunctionOperator(
        task_id = "load_crypto_trading_to_bq",
        location = XRegion,
        function_id = XFunction,
        input_data = {},
    )

解答 #

关于function id #

Function ID是云函数的唯一标识符，在GCP中，它默认和你创建云函数时指定的函数名称完全一致。但要注意函数名称的唯一性是基于「项目+区域」范围的，不是全局唯一。

排查404错误的关键方向 #

1. 确认云函数实际存在：登录GCP控制台，进入Cloud Functions页面，检查项目pongthorn的asia-southeast1区域下是否真的存在crypto-trading-to-bq函数，有可能部署过程出错导致函数未创建成功。
2. 检查区域拼写：确保asia-southeast1无拼写错误（比如不要写成asia-southeast-1），GCP区域代码没有中间横杠。
3. 验证变量加载正确性：在代码中添加打印语句，输出XRegion和XFunction的值，确认Airflow变量是否正确读取：

   print(f"当前使用区域: {XRegion}, 目标函数: {XFunction}")
   

4. 排查权限问题：部分场景下，服务账号权限不足会伪装成「资源不存在」的404错误，需确保Airflow使用的服务账号拥有cloudfunctions.functions.invoke权限。

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