无法触发 composer/airflow 来自 Cloud 函数的 dag，该函数在云存储发生变化时触发
Unable to trigger composer/airflow dag from Cloud function that triggers when there are changes in cloud storage
我已经在 google-cloud-composer 环境中创建并 运行 dags (dlkpipelinesv1 : composer-1.13.0-airflow-1.10.12)。我可以手动触发这些 dags，并使用调度程序，但是当我通过 cloud-functions 触发它们时，我被卡住了，它检测 google-cloud-storage 存储桶中的变化。
请注意，我有另一个 GC-Composer 环境（pipelines:composer-1.7.5-airflow-1.10.2）使用相同的 google 云函数触发相关的 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>"
}
我用来发出 post 请求的服务帐户具有以下角色：
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
但是当向气流 API 发出 post 请求时，我仍然在 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 的用户。
我错过了什么？
抛出 403 的代码是它过去的工作方式。 2020 年年中发生了重大变化。您不应使用 requests 对令牌进行 HTTP 调用，而应使用 Google's OAuth2 library:
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)
我按照 Triggering DAGs 中的步骤操作并在我的环境中工作，请参阅下面我的建议。
Componser 环境启动并运行是一个好的开始。通过这个过程，您只需要上传新的 DAG (trigger_response_dag.py) 并使用 python 脚本或从登录页面获取 clientID（以 .apps.googleusercontent.com 结尾）打开气流 UI.
在 Cloud Functions 方面，我注意到您有 Node.js 和 Python 的指令组合，例如，USER_AGENT 仅适用于 Node.js。例程 make_iap_request 仅适用于 python。我希望以下几点有助于解决您的问题：
服务帐户 (SA)。 Node.js 代码使用默认服务帐户 ${projectId}@appspot.gserviceaccount.com，其默认角色是 Editor，这意味着它可以广泛访问 GCP 服务，包括 Cloud Composer。在 python 中，我认为身份验证是由 client_id 以某种方式管理的，因为使用 id 检索令牌。请确保 SA 具有此编辑者角色，并且不要忘记按照指南中的规定分配 serviceAccountTokenCreator。
我使用了 Node.js 8 运行时，我注意到您所关注的用户代理应该是 'gcf-event-trigger'，因为它是硬编码的； USER_AGENT = 'gcf-event-trigger'。在python，好像没必要。
默认情况下，在GCS触发器中，GCS Event Type设置为Archive，您需要将其更改为Finalize/Create。如果设置为Archive，上传对象时触发器不会起作用，DAG也不会启动。
如果您认为您的云函数配置正确但错误仍然存在，您可以在控制台中云函数的日志选项卡中找到其原因。它可以为您提供更多详细信息。
基本上，根据指南，我只需要更改 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'
有一个配置参数导致对 API 的所有请求都被拒绝...
在documentation中提到我们需要覆盖以下气流配置：
[api]
auth_backend = airflow.api.auth.backend.deny_all
进入
[api]
auth_backend = airflow.api.auth.backend.default
了解这个细节非常重要，google 的文档中没有提及...
有用的链接：
我已经在 google-cloud-composer 环境中创建并 运行 dags (dlkpipelinesv1 : composer-1.13.0-airflow-1.10.12)。我可以手动触发这些 dags，并使用调度程序，但是当我通过 cloud-functions 触发它们时，我被卡住了，它检测 google-cloud-storage 存储桶中的变化。
请注意，我有另一个 GC-Composer 环境（pipelines:composer-1.7.5-airflow-1.10.2）使用相同的 google 云函数触发相关的 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>"
}
我用来发出 post 请求的服务帐户具有以下角色：
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
但是当向气流 API 发出 post 请求时，我仍然在 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 的用户。
我错过了什么？
抛出 403 的代码是它过去的工作方式。 2020 年年中发生了重大变化。您不应使用 requests 对令牌进行 HTTP 调用，而应使用 Google's OAuth2 library:
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)
我按照 Triggering DAGs 中的步骤操作并在我的环境中工作，请参阅下面我的建议。
Componser 环境启动并运行是一个好的开始。通过这个过程，您只需要上传新的 DAG (trigger_response_dag.py) 并使用 python 脚本或从登录页面获取 clientID（以 .apps.googleusercontent.com 结尾）打开气流 UI.
在 Cloud Functions 方面，我注意到您有 Node.js 和 Python 的指令组合，例如，USER_AGENT 仅适用于 Node.js。例程 make_iap_request 仅适用于 python。我希望以下几点有助于解决您的问题：
服务帐户 (SA)。 Node.js 代码使用默认服务帐户
${projectId}@appspot.gserviceaccount.com，其默认角色是 Editor，这意味着它可以广泛访问 GCP 服务，包括 Cloud Composer。在 python 中，我认为身份验证是由client_id以某种方式管理的，因为使用 id 检索令牌。请确保 SA 具有此编辑者角色，并且不要忘记按照指南中的规定分配serviceAccountTokenCreator。我使用了 Node.js 8 运行时，我注意到您所关注的用户代理应该是 'gcf-event-trigger'，因为它是硬编码的；
USER_AGENT = 'gcf-event-trigger'。在python，好像没必要。默认情况下，在GCS触发器中，GCS Event Type设置为
Archive，您需要将其更改为Finalize/Create。如果设置为Archive，上传对象时触发器不会起作用，DAG也不会启动。如果您认为您的云函数配置正确但错误仍然存在，您可以在控制台中云函数的日志选项卡中找到其原因。它可以为您提供更多详细信息。
基本上，根据指南，我只需要更改 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'
有一个配置参数导致对 API 的所有请求都被拒绝...
在documentation中提到我们需要覆盖以下气流配置：
[api]
auth_backend = airflow.api.auth.backend.deny_all
进入
[api]
auth_backend = airflow.api.auth.backend.default
了解这个细节非常重要，google 的文档中没有提及...
有用的链接：