Déclencher des DAG Managed Service pour Apache Airflow avec des fonctions Cloud Run et l'API REST Airflow

Airflow géré (3e génération) | Airflow géré (2e génération) | Airflow géré (1re génération, version héritée)

Cette page explique comment utiliser Cloud Run Functions pour déclencher des DAG Managed Service for Apache Airflow en réponse à des événements.

Apache Airflow est conçu pour exécuter des DAG de façon régulière, mais vous pouvez également déclencher des DAG en réponse à des événements. Pour ce faire, vous pouvez utiliser Cloud Run Functions afin de déclencher des DAG Airflow géré lorsqu'un événement spécifique se produit.

Vous pouvez également :

L'exemple de ce guide illustre une fonction qui déclenche un DAG en réponse à un événement :

  1. Vous configurez des déclencheurs pour votre fonction dans Cloud Run Functions.
  2. Lorsque la fonction est déclenchée, elle envoie une requête pour déclencher un DAG via l'API REST Airflow de votre environnement Airflow géré. La requête contient l'identifiant et le type de l'événement, ainsi que sa charge utile.
  3. Airflow traite cette requête et exécute le DAG spécifié dans la requête. Le DAG génère les données qui lui ont été transmises par la fonction.

Avant de commencer

Cette section répertorie les étapes préparatoires.

Vérifier la configuration réseau de votre environnement

Cette solution ne fonctionne pas dans les configurations d'adresse IP privée et de VPC Service Controls, car il n'est pas possible de configurer la connectivité entre les fonctions Cloud Run et le serveur Web Airflow dans ces configurations.

Activer les API pour votre projet.

Console

Activez les API Managed Airflow et Cloud Run Functions.

Rôles requis pour activer les API

Pour activer les API, vous avez besoin de l'autorisation serviceusage.services.enable. Si vous avez créé le projet, vous disposez probablement déjà de cette autorisation via le rôle Propriétaire (roles/owner). Sinon, vous pouvez l'obtenir via le rôle Administrateur d'utilisation du service (roles/serviceusage.serviceUsageAdmin). Découvrez comment attribuer des rôles.

Activer les API

gcloud

Activez les API Managed Airflow et Cloud Run Functions :

Rôles requis pour activer les API

Pour activer les API, vous avez besoin de l'autorisation serviceusage.services.enable. Si vous avez créé le projet, vous disposez probablement déjà de cette autorisation via le rôle Propriétaire (roles/owner). Sinon, vous pouvez l'obtenir via le rôle Administrateur d'utilisation du service (roles/serviceusage.serviceUsageAdmin). Découvrez comment attribuer des rôles. 

gcloud services enable cloudfunctions.googleapis.com composer.googleapis.com

Activer l'API REST Airflow

Selon votre version d'Airflow :

Autoriser les appels d'API vers l'API REST Airflow à l'aide du contrôle des accès réseau du serveur Web

Cloud Run Functions peut accéder à l'API REST Airflow via une adresse IPv4 ou IPv6.

Si vous ne savez pas quelle sera la plage d'adresses IP appelantes, utilisez une option de configuration par défaut dans Contrôle d'accès au serveur Web , à savoir All IP addresses have access (default) (Toutes les adresses IP ont accès (par défaut)), afin de ne pas bloquer accidentellement vos fonctions Cloud Run. Vous pourrez toujours configurer l'accès réseau au serveur Web ultérieurement.

Obtenir l'URL du serveur Web Airflow

Cet exemple envoie des requêtes API REST au point de terminaison du serveur Web Airflow. Vous utilisez la partie de l'URL de l'interface Web Airflow avant .appspot.com dans le code de votre fonction Cloud.

Console

  1. Dans la Google Cloud console, accédez à la page Environnements.

    Accéder à la page Environnements

  2. Cliquez sur le nom de votre environnement.

  3. Sur la page Détails de l'environnement, accédez à l'onglet Configuration de l'environnement.

  4. L'URL du serveur Web Airflow est répertoriée dans l'élément UI Web Airflow.

gcloud

Exécutez la commande suivante :

gcloud composer environments describe ENVIRONMENT_NAME \
    --location LOCATION \
    --format='value(config.airflowUri)'

Remplacez :

  • ENVIRONMENT_NAME par le nom de l'environnement.
  • LOCATION par la région dans laquelle se trouve l'environnement.

Obtenir le "client_id" du proxy IAM

Pour envoyer une requête au point de terminaison de l'API REST Airflow, la fonction requiert l'ID client du proxy Identity and Access Management qui protège le serveur Web Airflow.

Managed Airflow ne fournit pas ces informations directement. Au lieu de cela, vous devez envoyer une requête non authentifiée au serveur Web Airflow et récupérer l'ID client à partir de l'URL de redirection.

cURL 

curl -v AIRFLOW_URL 2>&1 >/dev/null | grep -o "client_id\=[A-Za-z0-9-]*\.apps\.googleusercontent\.com"

Remplacez AIRFLOW_URL par l'URL de l'interface Web Airflow.

Dans le résultat, recherchez la chaîne qui suit client_id. Exemple :

client_id=836436932391-16q2c5f5dcsfnel77va9bvf4j280t35c.apps.googleusercontent.com

Python

Enregistrez le code suivant dans un fichier nommé get_client_id.py. Renseignez vos valeurs pour project_id, location et composer_environment, puis exécutez le code dans Cloud Shell ou dans votre environnement local. 

# This script is intended to be used with Composer 1 environments
# In Composer 2, the Airflow Webserver is not in the tenant project
# so there is no tenant client ID
# See https://cloud.google.com/composer/docs/composer-2/environment-architecture
# for more details
import google.auth
import google.auth.transport.requests
import requests
import six.moves.urllib.parse

# Authenticate with Google Cloud.
# See: https://cloud.google.com/docs/authentication/getting-started
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()
composer_version = environment_data["config"]["softwareConfig"]["imageVersion"]
if "composer-1" not in composer_version:
    version_error = (
        "This script is intended to be used with Composer 1 environments. "
        "In Composer 2, the Airflow Webserver is not in the tenant project, "
        "so there is no tenant client ID. "
        "See https://cloud.google.com/composer/docs/composer-2/environment-architecture for more details."
    )
    raise (RuntimeError(version_error))
airflow_uri = environment_data["config"]["airflowUri"]

# The Composer environment response does not include the IAP client ID.
# Make a second, unauthenticated HTTP request to the web server to get the
# redirect URI.
redirect_response = requests.get(airflow_uri, allow_redirects=False)
redirect_location = redirect_response.headers["location"]

# Extract the client_id query parameter from the redirect.
parsed = six.moves.urllib.parse.urlparse(redirect_location)
query_string = six.moves.urllib.parse.parse_qs(parsed.query)
print(query_string["client_id"][0])

Importer un DAG dans votre environnement.

Importer un DAG dans votre environnement. L'exemple de DAG suivant génère la configuration d'exécution de DAG reçue. Vous allez déclencher ce DAG à partir d'une fonction que vous créerez plus loin dans ce guide.

import datetime

import airflow
from airflow.operators.bash_operator import BashOperator

with airflow.DAG(
        'composer_sample_trigger_response_dag',
        start_date=datetime.datetime(2026, 1, 1),
        # Not scheduled, trigger only
        schedule=None) as dag:

    # Print the dag_run's configuration, which includes information about the
    # Cloud Storage object change.
    print_gcs_info = BashOperator(
        task_id='print_gcs_info', bash_command='echo {{ dag_run.conf }}}}')

Déployer une fonction qui déclenche le DAG

Vous pouvez déployer une fonction à l'aide du langage de votre choix compatible avec Cloud Run Functions ou Cloud Run. Ce tutoriel présente une fonction Cloud implémentée en Python et Java.

Spécifier les paramètres de configuration de la fonction

  • Déclencheur : sélectionnez un ou plusieurs déclencheurs Eventarc pour votre fonction.

    Pour en savoir plus sur la création de déclencheurs, consultez la section Créer des déclencheurs avec Eventarc. Par exemple, vous pouvez déclencher des fonctions depuis Cloud Storage à l'aide d'Eventarc.

  • Compte de service : le compte de service que vous spécifiez pour le déclencheur doit disposer des autorisations suffisantes pour déclencher des DAG dans les environnements Managed Airflow.

    Nous vous recommandons de suivre le principe de moindre privilège et de n'accorder que le rôle Utilisateur Composer (composer.user). Pour en savoir plus sur la configuration des autorisations, consultez la page Rôles et autorisations pour les cibles Cloud Run.

  • Point d'entrée de la fonction :

    • (Python) Lorsque vous ajoutez du code pour cet exemple, sélectionnez l'environnement d'exécution Python 3.10 ou une version ultérieure, puis spécifiez trigger_dag comme point d'entrée.

Ajouter des conditions

Spécifiez les dépendances dans le fichier requirements.txt :

requests-toolbelt==1.0.0
google-auth==2.38.0
google-cloud-pubsub==2.28.0

Ajouter un code de fonction

Placez le code suivant dans le fichier main.py et effectuez les remplacements suivants :

  • Remplacez la valeur de la variable client_id par la valeur client_id que vous avez obtenue précédemment.

  • Remplacez la valeur de la variable webserver_id par l'ID de votre projet locataire, qui fait partie de l'URL de l'interface Web Airflow avant .appspot.com. Vous avez obtenu l'URL de l'interface Web Airflow lors d'une étape précédente.

  • Spécifiez la version de l'API REST Airflow que vous utilisez :

    • Si vous utilisez l'API REST Airflow stable, définissez la variable USE_EXPERIMENTAL_API sur False.
    • Si vous utilisez l'API REST Airflow expérimentale, aucune modification n'est nécessaire. La variable USE_EXPERIMENTAL_API est déjà définie sur True.


from google.auth.transport.requests import Request
from google.oauth2 import id_token
import requests


IAM_SCOPE = "https://www.googleapis.com/auth/iam"
OAUTH_TOKEN_URI = "https://www.googleapis.com/oauth2/v4/token"
# If you are using the stable API, set this value to False
# For more info about Airflow APIs see https://cloud.google.com/composer/docs/access-airflow-api
USE_EXPERIMENTAL_API = True


def trigger_dag(data, context=None):
    """Makes a POST request to the Composer DAG Trigger API

    When called via Google Cloud Functions (GCF),
    data and context are Background function parameters.

    For more info, refer to
    https://cloud.google.com/functions/docs/writing/background#functions_background_parameters-python

    To call this function from a Python script, omit the ``context`` argument
    and pass in a non-null value for the ``data`` argument.

    This function is currently only compatible with Composer v1 environments.
    """

    # Fill in with your Composer info here
    # Navigate to your webserver's login page and get this from the URL
    # Or use the script found at
    # https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/composer/rest/get_client_id.py
    client_id = "YOUR-CLIENT-ID"
    # This should be part of your webserver's URL:
    # {tenant-project-id}.appspot.com
    webserver_id = "YOUR-TENANT-PROJECT"
    # The name of the DAG you wish to trigger
    dag_name = "composer_sample_trigger_response_dag"

    if USE_EXPERIMENTAL_API:
        endpoint = f"api/experimental/dags/{dag_name}/dag_runs"
        json_data = {"conf": data, "replace_microseconds": "false"}
    else:
        endpoint = f"api/v1/dags/{dag_name}/dagRuns"
        json_data = {"conf": data}
    webserver_url = "https://" + webserver_id + ".appspot.com/" + endpoint
    # Make a POST request to IAP which then Triggers the DAG
    make_iap_request(webserver_url, client_id, method="POST", json=json_data)


# This code is copied from
# https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/iap/make_iap_request.py
# START COPIED IAP CODE
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.
    google_open_id_connect_token = id_token.fetch_id_token(Request(), client_id)

    # 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": "Bearer {}".format(google_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(
            "Bad response from application: {!r} / {!r} / {!r}".format(
                resp.status_code, resp.headers, resp.text
            )
        )
    else:
        return resp.text


# END COPIED IAP CODE

Tester votre fonction

Pour vérifier que votre fonction et votre DAG fonctionnent comme prévu, procédez comme suit :

  1. Attendez que votre fonction soit déployée.
  2. Déclenchez la fonction en fonction du déclencheur spécifié. Vous pouvez également déclencher la fonction manuellement en sélectionnant l'action Tester la fonction dans la Google Cloud console.
  3. Consultez la page du DAG dans l'interface Web Airflow. Le DAG doit comporter une exécution de DAG active ou déjà terminée.
  4. Dans l'UI d'Airflow, consultez les journaux des tâches associées à cette exécution. Vous devriez voir que la tâche print_gcs_info écrit les données reçues de la fonction dans les journaux :

Exemple de résultat :

[2021-04-04 18:25:44,778] {bash_operator.py:154} INFO - Output:
[2021-04-04 18:25:44,781] {bash_operator.py:158} INFO - Triggered from GCF:
    {bucket: example-storage-for-gcf-triggers, contentType: text/plain,
    crc32c: dldNmg==, etag: COW+26Sb5e8CEAE=, generation: 1617560727904101,
    ... }
[2021-04-04 18:25:44,781] {bash_operator.py:162} INFO - Command exited with
    return code 0h

Étape suivante