无法从云存储发生变化时触发的 Cloud 函数触发 composer/airflow dag

Question

我在 google-cloud-composer 环境中创建并运行了 dags（dlkpipelinesv1：composer-1.13.0-airflow-1.10.12）。我可以手动触发这些 dag，并使用调度程序，但是当我通过检测 google-cloud-storage 存储桶中的更改的cloud-functions 触发它们时遇到了困难。

请注意，我有另一个 GC-Composer 环境 (pipelines:composer-1.7.5-airflow-1.10.2)，它使用相同的谷歌云功能来触发相关的 dag，并且 它正在工作。

我按照this guide 创建了触发 dag 的函数。所以我检索了以下变量：

PROJECT_ID = <project_id>
CLIENT_ID = <client_id_retrieved_by_running_the_code_in_the_guide_within_my_gcp_console>
WEBSERVER_ID = <airflow_webserver_id>
DAG_NAME = <dag_to_trigger>
WEBSERVER_URL = f"https://{WEBSERVER_ID}.appspot.com/api/experimental/dags/{DAG_NAME}/dag_runs"


def file_listener(event, context):
    """Entry point of the cloud function: Triggered by a change to a Cloud Storage bucket.
    Args:
         event (dict): Event payload.
         context (google.cloud.functions.Context): Metadata for the event.
    """
    logging.info("Running the file listener process")
    logging.info(f"event : {event}")
    logging.info(f"context : {context}")
    file = event
    if file["size"] == "0" or "DTM_DATALAKE_AUDIT_COMPTAGE" not in file["name"] or ".filepart" in file["name"].lower():
        logging.info("no matching file")
        exit(0)

    logging.info(f"File listener detected the presence of : {file['name']}.")

    # id_token = authorize_iap()
    # make_iap_request({"file_name": file["name"]}, id_token)
    make_iap_request(url=WEBSERVER_URL, client_id=CLIENT_ID, method="POST")


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:
                https://github.com/requests/requests/blob/master/requests/api.py
                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.
    open_id_connect_token = id_token.fetch_id_token(Request(), client_id)
    logging.info(f"Retrieved open id connect (bearer) token {open_id_connect_token}")

    # 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": f"Bearer {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(f"Bad response from application: {resp.status_code} / {resp.headers} / {resp.text}")
    else:
        logging.info(f"Response status - {resp.status_code}")
        return resp.json

这是在 GC 函数中运行的代码 我已经使用以下代码分别检查了 dlkpipelinesv1 和 piplines 中的环境详细信息：

credentials, _ = google.auth.default(
    scopes=['https://www.googleapis.com/auth/cloud-platform'])
authed_session = google.auth.transport.requests.AuthorizedSession(
    credentials)

# project_id = 'YOUR_PROJECT_ID'
# location = 'us-central1'
# composer_environment = 'YOUR_COMPOSER_ENVIRONMENT_NAME'

environment_url = (
    'https://composer.googleapis.com/v1beta1/projects/{}/locations/{}'
    '/environments/{}').format(project_id, location, composer_environment)
composer_response = authed_session.request('GET', environment_url)
environment_data = composer_response.json()

并且两者使用相同的服务帐户来运行，即相同的 IAM 角色。虽然我注意到以下不同的细节：

在旧环境中：

"airflowUri": "https://p5<hidden_value>-tp.appspot.com",
    "privateEnvironmentConfig": { "privateClusterConfig": {} },

在新环境中：

"airflowUri": "https://da<hidden_value>-tp.appspot.com",
    "privateEnvironmentConfig": {
      "privateClusterConfig": {},
      "webServerIpv4CidrBlock": "<hidden_value>",
      "cloudSqlIpv4CidrBlock": "<hidden_value>"
    }

我用来发出发布请求的服务帐户具有以下角色：

Cloud Functions Service Agent 
Composer Administrator 
Composer User
Service Account Token Creator 
Service Account User

运行我的作曲家环境的服务帐户具有以下角色：

BigQuery Admin
Composer Worker
Service Account Token Creator
Storage Object Admin

但是当post 请求向气流API 发出时，我仍然在Log Explorer 中收到403 - Forbidden。

编辑 2020 年 11 月 16 日：

我已更新到最新的make_iap_request 代码。 我在安全服务中修改了 IAP，但我找不到可以接受来自我的云功能的HTTP: post 请求的网络服务器...请参见下图，无论如何我将服务帐户添加到默认和 CRM IAP 资源中案例，但我仍然收到此错误：

Exception: Service account does not have permission to access the IAP-protected application.

主要问题是：什么 IAP 在这里受到威胁？以及如何将我的服务帐户添加为此 IAP 的用户。

我错过了什么？

Answer 1

抛出 403 的代码是它过去的工作方式。 2020 年中期发生了重大变化。您应该使用Google's OAuth2 library，而不是使用requests 对令牌进行HTTP 调用：

from google.oauth2 import id_token
from google.auth.transport.requests import Request
open_id_connect_token = id_token.fetch_id_token(Request(), client_id)

见this example

Answer 2

我按照Triggering DAGs 中的步骤操作并在我的环境中工作过，请参阅下面我的建议。

Componser 环境启动并运行是一个好的开始。通过该过程，您只需上传新的 DAG (trigger_response_dag.py) 并使用 python 脚本或在首次打开 Airflow UI 时从登录页面获取 clientID（以 .apps.googleusercontent.com 结尾）。

在 Cloud Functions 方面，我注意到您有针对 Node.js 和 Python 的指令组合，例如，USER_AGENT 仅适用于 Node.js。而例程make_iap_request 仅适用于 python。我希望以下几点有助于解决您的问题：

1. 服务帐户 (SA)。 Node.js 代码使用默认服务帐户${projectId}@appspot.gserviceaccount.com，其默认角色为 Editor，这意味着它可以广泛访问 GCP 服务，包括 Cloud Composer。在 python 中，我认为身份验证是由client_id 以某种方式管理的，因为使用 id 检索令牌。请确保 SA 具有此 Editor 角色，并且不要忘记按照指南中的说明分配 serviceAccountTokenCreator。

2. 我使用了 Node.js 8 运行时，我注意到您担心的用户代理应该是“gcf-event-trigger”，因为它是硬编码的； USER_AGENT = 'gcf-event-trigger'。在python中，似乎没有必要。

3. 默认情况下，在 GCS 触发器中，GCS Event Type 设置为Archive，您需要将其更改为Finalize/Create。如果设置为Archive，则在上传对象时触发器将不起作用，并且不会启动 DAG。

4. 如果您认为您的云功能配置正确并且错误仍然存​​在，您可以在控制台中云功能的 LOGS 选项卡中找到其原因。它可以为您提供更多详细信息。

基本上，根据指南，我只需要在 Node.js 中更改以下值：

// The project that holds your function. Replace <YOUR-PROJECT-ID>
const PROJECT_ID = '<YOUR-PROJECT-ID>';
// Navigate to your webserver's login page and get this from the URL
const CLIENT_ID = '<ALPHANUMERIC>.apps.googleusercontent.com';
// This should be part of your webserver's URL in the Env's detail page: {tenant-project-id}.appspot.com. 
const WEBSERVER_ID = 'v90eaaaa11113fp-tp';
// The name of the DAG you wish to trigger. It's DAG's name in the script trigger_response_dag.py you uploaded to your Env.
const DAG_NAME = 'composer_sample_trigger_response_dag';

对于 Python，我只更改了这些参数：

client_id = '<ALPHANUMERIC>.apps.googleusercontent.com'
# This should be part of your webserver's URL:
# {tenant-project-id}.appspot.com
webserver_id = 'v90eaaaa11113fp-tp'
# Change dag_name only if you are not using the example
dag_name = 'composer_sample_trigger_response_dag'

Answer 3

有一个配置参数会导致对 API 的所有请求都被拒绝...

documentation 中提到我们需要覆盖以下气流配置：

[api]
auth_backend = airflow.api.auth.backend.deny_all

进入

[api]
auth_backend = airflow.api.auth.backend.default

这个细节真的很重要，google的文档中没有提到...

有用的链接：

1. Triggering DAGS (workflows) with GCS
2. make_iap_request.py repository