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 Cloud 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
Google Cloud Functions and Cloud Identity and Google Identity and Access Management (IAM) APIs.
Enable the
APIs
Granting blob signing permissions to the Cloud Functions Service Account
To authenticate to Cloud 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
Cloud 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:
Creating your function
Create a function with the below index.js and package.json, filling in the
first four constants (see Create a
Function).
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} event The Cloud Functions event.
* @param {!Function} callback The callback function.
*/
exports.triggerDag = function triggerDag (event, callback) {
// 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(event.data)};
// Make the request
authorizeIap(CLIENT_ID, PROJECT_ID, USER_AGENT)
.then(function iapAuthorizationCallback (iap) {
makeIapPostRequest(WEBSERVER_URL, BODY, iap.idToken, USER_AGENT, iap.jwt);
})
.then(_ => callback(null))
.catch(callback);
};
/**
* @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.
*/
function authorizeIap (clientId, projectId, userAgent) {
const SERVICE_ACCOUNT = `${projectId}@appspot.gserviceaccount.com`;
const JWT_HEADER = Buffer.from(JSON.stringify({alg: 'RS256', typ: 'JWT'}))
.toString('base64');
var jwt = '';
var jwtClaimset = '';
// Obtain an Oauth2 access token for the appspot service account
return fetch(
`http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/${SERVICE_ACCOUNT}/token`,
{
headers: {'User-Agent': userAgent, 'Metadata-Flavor': 'Google'}
})
.then(res => res.json())
.then(function obtainAccessTokenCallback (tokenResponse) {
if (tokenResponse.error) {
return Promise.reject(tokenResponse.error);
}
var accessToken = tokenResponse.access_token;
var iat = Math.floor(new Date().getTime() / 1000);
var 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');
var toSign = [JWT_HEADER, jwtClaimset].join('.');
return 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}`
}
});
})
.then(res => res.json())
.then(function signJsonClaimCallback (body) {
if (body.error) {
return Promise.reject(body.error);
}
// Request service account signature on header and claimset
var jwtSignature = body.signature;
jwt = [JWT_HEADER, jwtClaimset, jwtSignature].join('.');
var form = new FormData();
form.append('grant_type', 'urn:ietf:params:oauth:grant-type:jwt-bearer');
form.append('assertion', jwt);
return fetch(
'https://www.googleapis.com/oauth2/v4/token', {
method: 'POST',
body: form
});
})
.then(res => res.json())
.then(function returnJwt (body) {
if (body.error) {
return Promise.reject(body.error);
}
return {
jwt: jwt,
idToken: body.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.
* @param {string} jwt A Json web token used to authenticate the request.
*/
function makeIapPostRequest (url, body, idToken, userAgent, jwt) {
var form = new FormData();
form.append('grant_type', 'urn:ietf:params:oauth:grant-type:jwt-bearer');
form.append('assertion', jwt);
return fetch(
url, {
method: 'POST',
body: form
})
.then(function makeIapPostRequestCallback () {
return fetch(url, {
method: 'POST',
headers: {
'User-Agent': userAgent,
'Authorization': `Bearer ${idToken}`
},
body: JSON.stringify(body)
});
});
}
Update your package.json with the following dependencies.
package.json
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).
gcs_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
