This page describes how to use Cloud Functions for event-based DAG triggers.
While Airflow is designed to run DAGs on a regular schedule, you can trigger
DAGs in response to events, such as a change in a Cloud Storage bucket or a
message pushed to Pub/Sub. To accomplish this,
Cloud Composer DAGs can be triggered by
Cloud Functions.
The example in this guide runs a DAG every time a change occurs in a
Cloud Storage bucket. Object-change metadata are passed into the DAG's
config.
Enabling APIs for your project
Enable the Cloud Composer, Cloud Functions, and Identity and Access Management (IAM) APIs.
Enable the APIs
Granting blob signing permissions to the Cloud Functions Service Account
To authenticate to IAP, grant the Appspot Service Account (used
by Cloud Functions) the Service Account Token Creator role on itself.
To do this, execute the following command in the gcloud
command-line tool or Cloud Shell:
gcloud iam service-accounts add-iam-policy-binding \
your-project-id@appspot.gserviceaccount.com \
--member=serviceAccount:your-project-id@appspot.gserviceaccount.com \
--role=roles/iam.serviceAccountTokenCreator
Getting the client ID
To construct a token to authenticate to
IAP,
the function requires the client ID of the proxy that protects the Airflow webserver.
The Cloud Composer API does not provide this information directly. Instead,
make an unauthenticated request to the Airflow webserver and capture the client
ID from the redirect URL. The following Python code sample demonstrates how to
get the client ID. After executing this code on the command line or in Cloud Shell, your client ID will be printed.
Creating your function
Node.js
Using the Node.js 8 runtime or above, create a function with the index.js
and package.json files shown below, filling in the first four constants (see
Create a Function).
Enable Retry on failure.
When you have finished, your function should resemble the following graphic:
index.js
'use strict';
const fetch = require('node-fetch');
const FormData = require('form-data');
/**
* Triggered from a message on a Cloud Storage bucket.
*
* IAP authorization based on:
* https://stackoverflow.com/questions/45787676/how-to-authenticate-google-cloud-functions-for-access-to-secure-app-engine-endpo
* and
* https://cloud.google.com/iap/docs/authentication-howto
*
* @param {!Object} data The Cloud Functions event data.
* @returns {Promise}
*/
exports.triggerDag = async (data) => {
// Fill in your Composer environment information here.
// The project that holds your function
const PROJECT_ID = 'your-project-id';
// Navigate to your webserver's login page and get this from the URL
const CLIENT_ID = 'your-iap-client-id';
// This should be part of your webserver's URL:
// {tenant-project-id}.appspot.com
const WEBSERVER_ID = 'your-tenant-project-id';
// The name of the DAG you wish to trigger
const DAG_NAME = 'composer_sample_trigger_response_dag';
// Other constants
const WEBSERVER_URL = `https://${WEBSERVER_ID}.appspot.com/api/experimental/dags/${DAG_NAME}/dag_runs`;
const USER_AGENT = 'gcf-event-trigger';
const BODY = {conf: JSON.stringify(data)};
// Make the request
try {
const iap = await authorizeIap(CLIENT_ID, PROJECT_ID, USER_AGENT);
return makeIapPostRequest(WEBSERVER_URL, BODY, iap.idToken, USER_AGENT);
} catch (err) {
console.error('Error authorizing IAP:', err.message);
throw new Error(err);
}
};
/**
* @param {string} clientId The client id associated with the Composer webserver application.
* @param {string} projectId The id for the project containing the Cloud Function.
* @param {string} userAgent The user agent string which will be provided with the webserver request.
*/
const authorizeIap = async (clientId, projectId, userAgent) => {
const SERVICE_ACCOUNT = `${projectId}@appspot.gserviceaccount.com`;
const JWT_HEADER = Buffer.from(
JSON.stringify({alg: 'RS256', typ: 'JWT'})
).toString('base64');
let jwt = '';
let jwtClaimset = '';
// Obtain an Oauth2 access token for the appspot service account
const res = await fetch(
`http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/${SERVICE_ACCOUNT}/token`,
{
headers: {'User-Agent': userAgent, 'Metadata-Flavor': 'Google'},
}
);
const tokenResponse = await res.json();
if (tokenResponse.error) {
console.error('Error in token reponse:', tokenResponse.error.message);
return Promise.reject(tokenResponse.error);
}
const accessToken = tokenResponse.access_token;
const iat = Math.floor(new Date().getTime() / 1000);
const claims = {
iss: SERVICE_ACCOUNT,
aud: 'https://www.googleapis.com/oauth2/v4/token',
iat: iat,
exp: iat + 60,
target_audience: clientId,
};
jwtClaimset = Buffer.from(JSON.stringify(claims)).toString('base64');
const toSign = [JWT_HEADER, jwtClaimset].join('.');
const blob = await fetch(
`https://iam.googleapis.com/v1/projects/${projectId}/serviceAccounts/${SERVICE_ACCOUNT}:signBlob`,
{
method: 'POST',
body: JSON.stringify({
bytesToSign: Buffer.from(toSign).toString('base64'),
}),
headers: {
'User-Agent': userAgent,
Authorization: `Bearer ${accessToken}`,
},
}
);
const blobJson = await blob.json();
if (blobJson.error) {
console.error('Error in blob signing:', blobJson.error.message);
return Promise.reject(blobJson.error);
}
// Request service account signature on header and claimset
const jwtSignature = blobJson.signature;
jwt = [JWT_HEADER, jwtClaimset, jwtSignature].join('.');
const form = new FormData();
form.append('grant_type', 'urn:ietf:params:oauth:grant-type:jwt-bearer');
form.append('assertion', jwt);
const token = await fetch('https://www.googleapis.com/oauth2/v4/token', {
method: 'POST',
body: form,
});
const tokenJson = await token.json();
if (tokenJson.error) {
console.error('Error fetching token:', tokenJson.error.message);
return Promise.reject(tokenJson.error);
}
return {
idToken: tokenJson.id_token,
};
};
/**
* @param {string} url The url that the post request targets.
* @param {string} body The body of the post request.
* @param {string} idToken Bearer token used to authorize the iap request.
* @param {string} userAgent The user agent to identify the requester.
*/
const makeIapPostRequest = async (url, body, idToken, userAgent) => {
const res = await fetch(url, {
method: 'POST',
headers: {
'User-Agent': userAgent,
Authorization: `Bearer ${idToken}`,
},
body: JSON.stringify(body),
});
if (!res.ok) {
const err = await res.text();
console.error('Error making IAP post request:', err.message);
throw new Error(err);
}
};
Update your package.json with the following dependencies.
package.json
Python
Create a function with the main.py and requirements.txt files shown below,
filling in the first four constants (see
Create a Function).
Enable Retry on failure.
When you have finished, your function should resemble the following graphic:
main.py
Update your requirements.txt with the following dependencies.
requirements.txt
Setting up your DAG
The following DAG contains a BashOperator which prints object-change metadata.
In order for it to run in response to Cloud Functions events, copy it
into your environment's DAGs folder (see
Manage DAGs).
trigger_response_dag.py
Testing your function
Upload a file to your Cloud Storage bucket. You should see that the DAG has
been triggered and information about the Cloud Storage change has been
logged by the BashOperator.
What's next
