Auf dieser Seite wird gezeigt, wie Sie Cloud Functions für ereignisbasierte DAG-Trigger verwenden.
Airflow ist für das regelmäßige Ausführen von DAGs nach Plan konzipiert. Sie können DAGs aber auch als Reaktion auf Ereignisse auslösen lassen, z. B. bei einer Änderung in einem Cloud Storage-Bucket oder bei einer per Push an Pub/Sub übertragenen Nachricht. Zu diesem Zweck können Cloud Composer-DAGs über Cloud Functions ausgelöst werden.
Im vorliegenden Beispiel wird bei jeder Änderung an einem Cloud Storage-Bucket ein DAG ausgeführt. Metadaten von Objektänderungen werden an die Konfiguration des DAG übergeben.
APIs für Ihr Projekt aktivieren
Aktivieren Sie die Cloud Composer, Cloud Functions, and Cloud Identity and Access Management (Cloud IAM) APIs.
Aktivieren Sie die APIs
Dem Cloud Functions-Dienstkonto Berechtigungen zur Blobsignatur gewähren
Zum Authentifizieren bei IAP weisen Sie dem Appspot-Dienstkonto, das von Cloud Functions verwendet wird, die Rolle Service Account Token Creator zu.
Führen Sie dazu im gcloud-Befehlszeilentool oder in Cloud Shell den folgenden Befehl aus:
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
Client-ID abrufen
Zum Erstellen eines Tokens für die Authentifizierung bei IAP benötigt die Funktion die Client-ID des Proxys, der den Airflow-Webserver schützt.
Die Cloud Composer API stellt diese Informationen nicht direkt zur Verfügung. Stellen Sie stattdessen eine nicht authentifizierte Anfrage an den Airflow-Webserver und erfassen Sie die Client-ID über die Weiterleitungs-URL. Das folgende Python-Codebeispiel zeigt, wie Sie die Client-ID abrufen. Nach der Ausführung dieses Codes in der Befehlszeile oder in Cloud Shell wird Ihre Client-ID ausgegeben.
Funktion erstellen
Node.js
Erstellen Sie mithilfe der Node.js 8-Laufzeit oder höher eine Funktion mit den Dateien index.js und package.json wie unten dargestellt. Geben Sie dabei die ersten vier Konstanten an. Weitere Informationen dazu finden Sie unter Funktion erstellen.
Aktivieren Sie die Option Bei Fehler noch einmal versuchen.
Wenn Sie fertig sind, sollte Ihre Funktion wie in der folgenden Abbildung aussehen:
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);
}
};
Aktualisieren Sie package.json mit den im Folgenden aufgeführten Abhängigkeiten.
package.json
Python
Erstellen Sie eine Funktion mit den Dateien main.py und requirements.txt wie unten dargestellt. Geben Sie dabei die ersten vier Konstanten an. Weitere Informationen dazu finden Sie unter Funktion erstellen.
Aktivieren Sie die Option Bei Fehler noch einmal versuchen.
Wenn Sie fertig sind, sollte Ihre Funktion wie in der folgenden Abbildung aussehen:
main.py
Aktualisieren Sie requirements.txt mit den im Folgenden aufgeführten Abhängigkeiten.
requirements.txt
DAG einrichten
Der folgende DAG enthält einen BashOperator, der Metadaten zu Objektänderungen ausgibt.
Für eine Ausführung anhand von Cloud Functions-Ereignissen kopieren Sie den DAG in den DAGs-Ordner Ihrer Umgebung (siehe DAGs hinzufügen und aktualisieren).
trigger_response_dag.py
Funktion testen
Laden Sie eine Datei in Ihren Cloud Storage-Bucket hoch. Es sollte nun angezeigt werden, dass der DAG ausgelöst wurde und Informationen zur Cloud Storage-Änderung vom BashOperator erfasst wurden.
Nächste Schritte
