无法从云存储发生更改时触发的云功能触发composer/airflow dag

Question

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

请注意，我有另一个 GC-Composer 环境（管道：composer-1.7.5-airflow-1.10.2），它使用相同的谷歌云功能来触发相关的 dags，并且它正在工作。

我按照本指南创建了触发 dags 的函数。所以我检索了以下变量：

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

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

BigQuery Admin
Composer Worker
Service Account Token Creator
Storage Object Admin

但当向气流 API 发出请求时，403 - Forbidden我仍然收到。Log Explorerpost

编辑 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

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

在文档中，提到我们需要覆盖以下气流配置：

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

进入

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

这个细节非常重要，谷歌的文档中没有提到......

有用的链接：

1. 使用 GCS 触发 DAGS（工作流程）
2. make_iap_request.py 存储库