使用Cloud Function触发Cloud Composer DAG时遭遇404错误求助

错误信息 #

gcs-dag-trigger-function

8bhxprce8hze
Traceback (most recent call last):
  File "/layers/google.python.pip/pip/lib/python3.9/site-packages/flask/app.py", line 2073, in wsgi_app
    response = self.full_dispatch_request()
  File "/layers/google.python.pip/pip/lib/python3.9/site-packages/flask/app.py", line 1518, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/layers/google.python.pip/pip/lib/python3.9/site-packages/flask/app.py", line 1516, in full_dispatch_request
    rv = self.dispatch_request()
  File "/layers/google.python.pip/pip/lib/python3.9/site-packages/flask/app.py", line 1502, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**req.view_args)
  File "/layers/google.python.pip/pip/lib/python3.9/site-packages/functions_framework/__init__.py", line 171, in view_func
    function(data, context)
  File "/workspace/main.py", line 52, in trigger_dag
    make_iap_request(
  File "/workspace/main.py", line 91, in make_iap_request
    raise Exception(
Exception: Bad response from application: 404 / {'Date': 'Mon, 18 Jul 2022 15:03:44 GMT', 'Content-Type': 'text/html; charset=utf-8', 'Vary': 'Accept-Encoding', 'Server': 'gunicorn', 'X-Robots-Tag': 'noindex, nofollow', 'Set-Cookie': 'session=616e9fda-1cd1-4a96-b9e2-57a3ea0f78bb.tblTOCMLoOZPdPTHgbbMepCbRbI; Expires=Wed, 17-Aug-2022 15:03:44 GMT; HttpOnly; Path=/; SameSite=Lax', 'Content-Encoding': 'gzip', 'Via': '1.1 google', 'Alt-Svc': 'h3=\":443\"; ma=2592000,h3-29=\":443\"; ma=2592000,h3-Q050=\":443\"; ma=2592000,h3-Q046=\":443\"; ma=2592000,h3-Q043=\":443\"; ma=2592000,quic=\":443\"; ma=2592000; v=\"46,43\"', 'Transfer-Encoding': 'chunked'} / '\n\n<!DOCTYPE html>\n<html lang=\"en\">\n  <head>\n    <title>Airflow 404</title>\n    <link rel=\"icon\" type=\"image/png\" href=\"/static/pin_32.png\">\n  </head>\n  <body>\n    <div style=\"font-family: verdana; text-align: center; margin-top: 200px;\">\n      <img src=\"/static/pin_100.png\" width=\"50px\" alt=\"pin-logo\" />\n      <h1>Airflow 404</h1>\n      <p>Page cannot be found.</p>\n      <a href=\"/\">Return to the main page</a>\n      <p>fbffada7b897</p>\n    </div>\n  </body>\n</html>'

触发代码 #

from google.auth.transport.requests import Request
from google.oauth2 import id_token
import requests


IAM_SCOPE = 'https://www.googleapis.com/auth/iam'
OAUTH_TOKEN_URI = 'https://www.googleapis.com/oauth2/v4/token'
# If you are using the stable API, set this value to False
USE_EXPERIMENTAL_API = True


def trigger_dag(data, context=None):
    """Makes a POST request to the Composer DAG Trigger API

    When called via Google Cloud Functions (GCF),
    data and context are Background function parameters.

    To call this function from a Python script, omit the ``context`` argument
    and pass in a non-null value for the ``data`` argument.

    This function is currently only compatible with Composer v1 environments.
    """

    # Fill in with your Composer info here
    client_id = 'xxxxxxxxx-gtld8n5rbu8ncs3l80fvnb2903pdq8p2.apps.googleusercontent.com'
    # This should be part of your webserver's URL: {tenant-project-id}.appspot.com
    webserver_id = 'xxxxxxxxxxxxxxxx-tp'
    # The name of the DAG you wish to trigger
    dag_name = 'GcsToBigQueryTriggered'

    if USE_EXPERIMENTAL_API:
        endpoint = f'api/experimental/dags/{dag_name}/dag_runs'
        json_data = {'conf': data, 'replace_microseconds': 'false'}
    else:
        endpoint = f'api/v1/dags/{dag_name}/dagRuns'
        json_data = {'conf': data}
    webserver_url = (
        'https://'
        + webserver_id
        + '.appspot.com/'
        + endpoint
    )
    # Make a POST request to IAP which then Triggers the DAG
    make_iap_request(
        webserver_url, client_id, method='POST', json=json_data)


def make_iap_request(url, client_id, method='GET', **kwargs):
    """Makes a request to an application protected by Identity-Aware Proxy.
    Args:
      url: The Identity-Aware Proxy-protected URL to fetch.
      client_id: The client ID used by Identity-Aware Proxy.
      method: The request method to use
              ('GET', 'OPTIONS', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE')
      **kwargs: Any of the parameters defined for the request function:
                If no timeout is provided, it is set to 90 by default.
    Returns:
      The page body, or raises an exception if the page couldn't be retrieved.
    """
    # Set the default timeout, if missing
    if 'timeout' not in kwargs:
        kwargs['timeout'] = 90

    # Obtain an OpenID Connect (OIDC) token from metadata server or using service
    # account.
    google_open_id_connect_token = id_token.fetch_id_token(Request(), client_id)

    # Fetch the Identity-Aware Proxy-protected URL, including an
    # Authorization header containing "Bearer " followed by a
    # Google-issued OpenID Connect token for the service account.
    resp = requests.request(
        method, url,
        headers={'Authorization': 'Bearer {}'.format(
            google_open_id_connect_token)}, **kwargs)
    if resp.status_code == 403:
        raise Exception('Service account does not have permission to '
                        'access the IAP-protected application.')
    elif resp.status_code != 200:
        raise Exception(
            'Bad response from application: {!r} / {!r} / {!r}'.format(
                resp.status_code, resp.headers, resp.text))
    else:
        return resp.text

排查解决方案 #

• 核对DAG名称：Airflow DAG名称区分大小写，必须与UI中显示的完全一致，检查GcsToBigQueryTriggered是否存在拼写、大小写或符号错误。
• 切换API版本：若使用Composer v2或Airflow 2.x，实验性API已被弃用，将USE_EXPERIMENTAL_API设为False，使用稳定API端点api/v1/dags/{dag_name}/dagRuns。
• 验证Webserver URL：确认webserver_id对应Composer环境的租户项目ID，访问https://{webserver_id}.appspot.com能正常打开Airflow UI。
• 检查IAP客户端ID：确保client_id是从Airflow登录页面URL获取的正确值，格式为xxxxxxxxx-xxxxxx.apps.googleusercontent.com。
• 确认DAG状态：在Airflow UI中检查目标DAG是否已启用（开关处于打开状态），未启用的DAG无法通过API触发。
• 适配Airflow版本：Airflow 2.x不再支持实验性API，必须使用稳定API接口，同时确认请求参数格式是否匹配（稳定API无需replace_microseconds参数）。

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