Activa DAG con Cloud Functions

Cloud Composer 1 | Cloud Composer 2

En esta página, se describe cómo usar Cloud Functions para activar los DAG en respuesta a eventos.

Airflow está diseñado para ejecutar DAG con regularidad, pero también puedes activar DAG en respuesta a eventos. Una forma de hacerlo es usar Cloud Functions para activar los DAG de Cloud Composer cuando ocurre un evento especificado. Por ejemplo, puedes crear una función que active un DAG cuando un objeto cambie en un bucket de Cloud Storage o cuando se envíe un mensaje a un tema de Pub/Sub.

En el ejemplo de esta guía, se ejecuta un DAG cada vez que se produce un cambio en un bucket de Cloud Storage. Los cambios en cualquier objeto de un bucket activan una función. Esta función realiza una solicitud a la API de REST de Airflow del entorno de Cloud Composer. Airflow procesa esta solicitud y ejecuta un DAG. El DAG genera información sobre el cambio.

Antes de comenzar

Habilita las API para tu proyecto.

Console

  • Habilita las API de Cloud Composer and Cloud Functions.

    Habilita las API

  • gcloud

  • Habilitar las API Cloud Composer and Cloud Functions.
    gcloud services enable cloudfunctionscomposer
  • Habilita la API de REST de Airflow

    Según tu versión de Airflow, ejecuta el siguiente comando:

    Cree un bucket de Cloud Storage

    Este ejemplo activa un DAG en respuesta a los cambios en un bucket de Cloud Storage. Crea un bucket nuevo para usar en este ejemplo.

    Obtén la URL del servidor web de Airflow

    En este ejemplo, se realizan solicitudes a la API de REST en el extremo del servidor web de Airflow. Usa la parte de la URL de la interfaz web de Airflow antes del .appspot.com en el código de Cloud Functions.

    Console

    1. En Google Cloud Console, ve a la página Entornos.

      Ir a Entornos

    2. Haz clic en el nombre de tu entorno.

    3. En la página Detalles del entorno, ve a la pestaña Detalles del entorno.

    4. La URL del servidor web de Airflow aparece en el elemento de la IU web de Airflow.

    gcloud

    Ejecuta el siguiente comando:

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

    Reemplaza lo siguiente:

    • ENVIRONMENT_NAME por el nombre del entorno.
    • LOCATION por la región de Compute Engine donde se encuentra el entorno.

    Obtén el ID de cliente del proxy de IAM

    Para realizar una solicitud al extremo de la API de REST de Airflow, la función requiere el ID de cliente del proxy de IAM que protege el servidor web de Airflow.

    Cloud Composer no proporciona esta información directamente. En cambio, realiza una solicitud no autenticada al servidor web de Airflow y captura el ID de cliente de la URL de redireccionamiento.

    cURL

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

    Reemplaza AIRFLOW_URL por la URL de la interfaz web de Airflow.

    En el resultado, busca la string que sigue a client_id. Por ejemplo:

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

    Python

    Guarda el siguiente código en un archivo llamado get_client_id.py. Completa tus valores para project_id, location y composer_environment. Luego, ejecuta el código en Cloud Shell o en tu entorno 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])

    Activa un DAG desde Cloud Functions

    Sube un DAG a tu entorno

    Sube un DAG a tu entorno. En el siguiente ejemplo de DAG, se muestra la configuración de ejecución del DAG recibido. Debes activar este DAG desde una función que crearás más adelante en esta guía.

    import datetime
    
    import airflow
    from airflow.operators.bash_operator import BashOperator
    
    with airflow.DAG(
            'composer_sample_trigger_response_dag',
            start_date=datetime.datetime(2021, 1, 1),
            # Not scheduled, trigger only
            schedule_interval=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 }}')

    Implementa una función de Cloud Functions que active el DAG

    Implementa una función de Cloud Functions de Python mediante los siguientes parámetros y contenido de configuración.

    Especifica los parámetros de configuración de Cloud Functions

    • Activador Para este ejemplo, selecciona un activador que funcione cuando se cree un objeto nuevo en un bucket o se reemplace un objeto existente.

      • Tipo de activador Cloud Storage

      • Tipo de evento Finalizar/Crear

      • Bucket Selecciona un bucket que debe activar esta función.

      • Volver a intentar en caso de error Te recomendamos inhabilitar esta opción para los fines de este ejemplo. Si usas tu propia función en un entorno de producción, habilita esta opción para controlar errores transitorios.

    • Cuenta de servicio del entorno de ejecución, en la sección Entorno de ejecución, compilación, conexiones y seguridad. Usa una de las siguientes opciones, según tus preferencias:

      • Selecciona Cuenta de servicio predeterminada de Compute Engine. Con los permisos predeterminados de IAM, esta cuenta puede ejecutar funciones que accedan a entornos de Cloud Composer.

      • Crea una cuenta de servicio personalizada que tenga la función de usuario de Composer y especifícala como una cuenta de servicio del entorno de ejecución para esta función. Esta opción sigue el principio de privilegio mínimo.

    • Entorno de ejecución y punto de entrada, en el paso Código. Cuando agregues el código de este ejemplo, selecciona el entorno de ejecución de Python 3.7 o superior y especifica trigger_dag como punto de entrada.

    Agrega requisitos

    Especifica las dependencias en el archivo requirements.txt:

    requests_toolbelt==0.9.1
    google-auth==2.3.3
    

    Agrega el código de la función de Cloud Functions

    Coloca el siguiente código en el archivo main.py y realiza los siguientes reemplazos:

    • Reemplaza el valor de la variable client_id por el valor client_id que obtuviste antes.

    • Reemplaza el valor de la variable webserver_id por el ID de proyecto de tu usuario, que es parte de la URL de la interfaz web de Airflow antes de .appspot.com. Anteriormente, obtuvo la URL de la interfaz web de Airflow.

    • Especifica la versión de la API de REST de Airflow que usas:

      • Si usas la API de REST de Airflow estable, establece la variable USE_EXPERIMENTAL_API en False.
      • Si usas la API de REST de Airflow experimental, no es necesario realizar cambios. La variable USE_EXPERIMENTAL_API ya está configurada como 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/master/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/master/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
    

    Prueba la función

    Para verificar que tu función y DAG funcionen según lo previsto, haz lo siguiente:

    1. Espera hasta que se implemente la función.
    2. Sube un archivo a tu bucket de Cloud Storage. Como alternativa, puedes activar la función de forma manual si seleccionas la acción Probar función en Google Cloud Console.
    3. Consulta la página del DAG en la interfaz web de Airflow. El DAG debe tener una DAG activa o ya completada.
    4. En la IU de Airflow, verifique los registros de tareas para esta ejecución. Deberías ver que la tarea print_gcs_info genera los datos recibidos de la función a los registros:
    [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
    

    ¿Qué sigue?