Activa un DAG con Node.JS y Google Cloud Functions

1. Introducción

Apache Airflow está diseñado para ejecutar DAG periódicamente, pero también puedes activarlos en respuesta a eventos, como un cambio en un bucket de Cloud Storage o un mensaje enviado a Cloud Pub/Sub. Para lograrlo, Cloud Functions puede activar los DAG de Cloud Composer.

En el ejemplo de este lab, se ejecuta un DAG simple cada vez que se produce un cambio en un bucket de Cloud Storage. Este DAG usa BashOperator para ejecutar un comando de Bash que imprime la información de cambio sobre lo que se subió al bucket de Cloud Storage.

Antes de comenzar este lab, se recomienda completar los codelabs Introducción a Cloud Composer y Comienza a usar Cloud Functions. Si creaste un entorno de Composer en el codelab de Introducción a Cloud Composer, puedes usarlo en este lab.

Qué compilarás

En este codelab, harás lo siguiente:

1. Sube un archivo a Google Cloud Storage, que
2. Activa una Google Cloud Function con el entorno de ejecución de Node.js
3. Esta función ejecutará un DAG en Google Cloud Composer.
4. Esto ejecuta un simple comando bash que imprime el cambio en el bucket de Google Cloud Storage.

Lo que aprenderá

• Cómo activar un DAG de Apache Airflow con Google Cloud Functions y Node.js

Requisitos

• Cuenta de GCP
• Conocimientos básicos de JavaScript
• Conocimientos básicos de Cloud Composer/Airflow y Cloud Functions
• Comodidad para usar comandos de la CLI

2. Configura GCP

Selecciona o crea el proyecto

Selecciona o crea un proyecto de Google Cloud Platform. Si vas a crear un proyecto nuevo, sigue los pasos que se indican aquí.

Toma nota del ID del proyecto, que usarás en pasos posteriores.

Si creas un proyecto nuevo, el ID del proyecto se encuentra justo debajo del nombre del proyecto en la página de creación.
Si ya creaste un proyecto, puedes encontrar el ID en la página principal de la consola, en la tarjeta de información del proyecto.

Nota: Si no planeas conservar los recursos creados durante este instructivo, crea un proyecto nuevo en lugar de seleccionar un proyecto existente. Después de terminar, puedes borrar el proyecto y quitar todos los recursos asociados con el proyecto y el instructivo.

Habilita las APIs

Habilita las APIs de Cloud Composer, Google Cloud Functions, Cloud Identity y Google Identity and Access Management (IAM).

Para obtener más información sobre cómo habilitar y, luego, inhabilitar las APIs, consulta la documentación.

Crea un entorno de Composer

Si ya tienes un entorno de Cloud Composer, puedes pasar a la siguiente sección.

Crea un entorno de Cloud Composer con la siguiente configuración: Nombre: my-composer-environment Ubicación: La ubicación que esté geográficamente más cerca de ti Zona: Cualquier zona de esa región El resto de la configuración puede permanecer con sus valores predeterminados. Haz clic en "Crear" en la parte inferior.Toma nota del nombre y la ubicación de tu entorno de Composer, ya que los necesitarás en los pasos posteriores.

El proceso de creación del entorno se completa cuando, en la página Entornos de la consola de GCP, la marca de verificación verde aparece a la izquierda del nombre del entorno. La creación tardará unos minutos, y puedes salir de esta página mientras se crea el entorno.

Crear un bucket de Cloud Storage

En tu proyecto, crea un bucket de Cloud Storage con la siguiente configuración: Nombre: <your-project-id> Clase de almacenamiento predeterminada: Multiregional Ubicación: La ubicación que esté geográficamente más cerca de la región de Cloud Composer que usas Modelo de control de acceso: Establece permisos a nivel de objeto y de bucket Presiona "Crear" cuando esté todo listo. Asegúrate de anotar el nombre de tu bucket de Cloud Storage para los pasos posteriores.

3. Configura Google Cloud Functions (GCF)

Para configurar GCF, ejecutaremos comandos en Google Cloud Shell.

Si bien Google Cloud se puede operar de manera remota desde tu laptop con la herramienta de línea de comandos gcloud, en este codelab usaremos Google Cloud Shell, un entorno de línea de comandos que se ejecuta en la nube.

Esta máquina virtual basada en Debian está cargada con todas las herramientas de desarrollo que necesitarás. Ofrece un directorio principal persistente de 5 GB y se ejecuta en Google Cloud, lo que permite mejorar considerablemente el rendimiento de la red y la autenticación. Esto significa que todo lo que necesitarás para este Codelab es un navegador (sí, funciona en una Chromebook).

Para activar Google Cloud Shell, haz clic en el botón que se encuentra en la parte superior derecha de la consola para desarrolladores (el aprovisionamiento y la conexión al entorno deberían llevar solo unos minutos):

Otorga permisos de firma de BLOB a la cuenta de servicio de Cloud Functions

Para que GCF se autentique en Cloud IAP, el proxy que protege el servidor web de Airflow, debes otorgar a la cuenta de servicio de Appspot de GCF el rol Service Account Token Creator. Para ello, ejecuta el siguiente comando en Cloud Shell y reemplaza <your-project-id> por el nombre de tu proyecto.

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

Por ejemplo, si tu proyecto se llama my-project, el comando sería

gcloud iam service-accounts add-iam-policy-binding \
[e-mail] \
--member=serviceAccount:[e-mail] \
--role=roles/iam.serviceAccountTokenCreator

Cómo obtener el ID de cliente

Importante: Si tu entorno de Composer no terminó de crearse, no podrás obtener el ID de cliente. Mientras se crea, puedes avanzar a la sección "Cómo ejecutar el código paso a paso" en "Crea tu función" para obtener más información sobre lo que ocurrirá cuando se ejecute esta función.

Para construir un token para autenticarse en Cloud IAP, la función requiere el ID de cliente del proxy que protege el servidor web de Airflow. La API de 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. Para ello, ejecutaremos un archivo de Python con Cloud Shell para capturar el ID del cliente.

Ejecuta el siguiente comando en Cloud Shell para descargar el código necesario de GitHub.

cd
git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git

Si recibiste un error porque este directorio ya existe, actualízalo a la versión más reciente ejecutando el siguiente comando:

cd python-docs-samples/
git pull origin master

Ejecuta el siguiente comando para cambiar al directorio correspondiente:

cd python-docs-samples/composer/rest

Ejecuta el código de Python para obtener tu ID de cliente. Para ello, reemplaza el nombre de tu proyecto por <your-project-id>, la ubicación del entorno de Composer que creaste antes por <your-composer-location> y el nombre del entorno de Composer que creaste antes por <your-composer-environment>.

python3 get_client_id.py <your-project-id> <your-composer-location> <your-composer-environment>

Por ejemplo, si el nombre de tu proyecto es my-project, tu ubicación de Composer es us-central1 y el nombre de tu entorno es my-composer, tu comando sería

python3 get_client_id.py my-project us-central1 my-composer

get_client_id.py realiza las siguientes acciones:

• Se autentica con Google Cloud
• Realiza una solicitud HTTP no autenticada al servidor web de Airflow para obtener el URI de redireccionamiento.
• Extrae el parámetro de búsqueda client_id de ese redireccionamiento.
• Imprime el código para que lo uses

Tu ID de cliente se imprimirá en la línea de comandos y se verá de la siguiente manera:

12345678987654321-abc1def3ghi5jkl7mno8pqr0.apps.googleusercontent.com

Importante: Copia tu ID de cliente en un lugar donde puedas usarlo en el paso Crea tu función. De lo contrario, deberás repetir este paso más adelante.

4. Crea tu función

En Cloud Shell, clona el repo con el código de muestra necesario ejecutando

cd
git clone https://github.com/GoogleCloudPlatform/nodejs-docs-samples.git

Cambia al directorio necesario y deja abierto Cloud Shell mientras completas los siguientes pasos.

cd nodejs-docs-samples/composer/functions/composer-storage-trigger

Navega a la página de Google Cloud Functions haciendo clic en el menú de navegación y, luego, en "Cloud Functions".
Haz clic en "CREAR FUNCIÓN" en la parte superior de la página.
Asigna el nombre "my-function" a tu función y deja la memoria en el valor predeterminado, 256 MB.
Establece el activador en "Cloud Storage", deja el tipo de evento como "Finalize/Create" y navega hasta el bucket que creaste en el paso Crea un bucket de Cloud Storage.
Deja el código fuente establecido en "Editor directo" y configura el tiempo de ejecución en "Node.js 8".

En Cloud Shell, ejecuta el siguiente comando: Se abrirán index.js y package.json en el Editor de Cloud Shell.

cloudshell edit index.js package.json

Haz clic en la pestaña package.json, copia ese código y pégalo en la sección package.json del editor intercalado de Cloud Functions.
Configura "Función para ejecutar" en triggerDag.
Haz clic en la pestaña index.js, copia el código y pégalo en la sección index.js del editor directo de Cloud Functions.
Cambia PROJECT_ID por el ID de tu proyecto y CLIENT_ID por el ID de cliente que guardaste en el paso Obtén el ID de cliente. Sin embargo, AÚN NO hagas clic en "Crear", ya que aún quedan algunos campos por completar.

En Cloud Shell, ejecuta el siguiente comando y reemplaza <your-environment-name> por el nombre de tu entorno de Composer y <your-composer-region> por la región en la que se encuentra tu entorno de Composer.

gcloud composer environments describe <your-environment-name> --location <your-composer-region>

Por ejemplo, si tu entorno se llama my-composer-environment y se encuentra en us-central1, tu comando sería

gcloud composer environments describe my-composer-environment --location us-central1

El resultado debería ser similar a este:

config:
 airflowUri: https://abc123efghi456k-tp.appspot.com
 dagGcsPrefix: gs://narnia-north1-test-codelab-jklmno-bucket/dags
 gkeCluster: projects/a-project/zones/narnia-north1-b/clusters/narnia-north1-test-codelab-jklmno-gke
 nodeConfig:
   diskSizeGb: 100
   location: projects/a-project/zones/narnia-north1-b
   machineType: projects/a-project/zones/narnia-north1-b/machineTypes/n1-standard-1
   network: projects/a-project/global/networks/default
   oauthScopes:
   - https://www.googleapis.com/auth/cloud-platform
   serviceAccount: [e-mail]
 nodeCount: 3
 softwareConfig:
   imageVersion: composer-1.7.0-airflow-1.10.0
   pythonVersion: '2'
createTime: '2019-05-29T09:41:27.919Z'
name: projects/a-project/locations/narnia-north1/environments/my-composer-environment
state: RUNNING
updateTime: '2019-05-29T09:56:29.969Z'
uuid: 123456-7890-9876-543-210123456

En ese resultado, busca la variable llamada airflowUri. En tu código index.js, cambia WEBSERVER_ID por el ID del servidor web de Airflow. Es la parte de la variable airflowUri que tendrá un "-tp" al final, por ejemplo, abc123efghi456k-tp.
Haz clic en el vínculo del menú desplegable "Más" y, luego, elige la región que esté geográficamente más cerca de ti.
Marca la opción "Volver a intentar en caso de error".
Haz clic en "Crear" para crear tu Cloud Function.

Recorre el código

El código que copiaste de index.js se verá de la siguiente manera:

// [START composer_trigger]
'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,
      iap.jwt
    );
  } catch (err) {
    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) {
    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) {
    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) {
    return Promise.reject(tokenJson.error);
  }

  return {
    jwt: jwt,
    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();
    throw new Error(err);
  }
};
// [END composer_trigger]

Nota: Es posible que tu código tenga pequeñas diferencias con lo que se imprimió aquí, pero no te preocupes. Tienes la versión más actualizada que se ejecuta en tu Cloud Function.

Veamos qué sucede. Aquí hay tres funciones: triggerDag, authorizeIap y makeIapPostRequest.

triggerDag es la función que se activa cuando subimos algo al bucket de Cloud Storage designado. Aquí configuramos variables importantes que se usan en las otras solicitudes, como PROJECT_ID, CLIENT_ID, WEBSERVER_ID y DAG_NAME. Llama a authorizeIap y makeIapPostRequest.

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,
      iap.jwt
    );
  } catch (err) {
    throw new Error(err);
  }
};

authorizeIap realiza una solicitud al proxy que protege el servidor web de Airflow, utilizando una cuenta de servicio y "canjeando" un JWT por un token de ID que se usará para autenticar el makeIapPostRequest.

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) {
    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) {
    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) {
    return Promise.reject(tokenJson.error);
  }

  return {
    jwt: jwt,
    idToken: tokenJson.id_token,
  };
};

makeIapPostRequest llama al servidor web de Airflow para activar el composer_sample_trigger_response_dag.. El nombre del DAG está integrado en la URL del servidor web de Airflow que se pasa con el parámetro url, y idToken es el token que obtuvimos en la solicitud authorizeIap.

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();
    throw new Error(err);
  }
};

5. Configura tu DAG

En Cloud Shell, cambia al directorio con los flujos de trabajo de muestra. Forma parte de python-docs-samples que descargaste de GitHub en el paso Obtén el ID de cliente.

cd
cd python-docs-samples/composer/workflows

Sube el DAG a Composer

Sube el DAG de muestra al bucket de almacenamiento de DAG de tu entorno de Composer con el siguiente comando, en el que <environment_name> es el nombre de tu entorno de Composer y <location> es el nombre de la región en la que se encuentra. trigger_response_dag.py es el DAG con el que trabajaremos.

gcloud composer environments storage dags import \
    --environment <environment_name> \
    --location <location> \
    --source trigger_response_dag.py

Por ejemplo, si tu entorno de Composer se llamaba my-composer y se encontraba en us-central1, tu comando sería

gcloud composer environments storage dags import \
    --environment my-composer \
    --location us-central1 \
    --source trigger_response_dag.py

Cómo recorrer el DAG

El código DAG en trigger_response.py se ve así:

import datetime
import airflow
from airflow.operators import bash_operator


default_args = {
    'owner': 'Composer Example',
    'depends_on_past': False,
    'email': [''],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'start_date': datetime.datetime(2017, 1, 1),
}

with airflow.DAG(
        'composer_sample_trigger_response_dag',
        default_args=default_args,
        # 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 = bash_operator.BashOperator(
        task_id='print_gcs_info', bash_command='echo {{ dag_run.conf }}')

La sección default_args contiene los argumentos predeterminados según lo requiere el modelo BaseOperator en Apache Airflow. Verás esta sección con estos parámetros en cualquier DAG de Apache Airflow. Actualmente, owner está configurado como Composer Example, pero puedes cambiarlo por tu nombre si lo deseas. depends_on_past nos muestra que este DAG no depende de ningún DAG anterior. Las tres secciones de correo electrónico, email, email_on_failure y email_on_retry, están configuradas de modo que no se reciban notificaciones por correo electrónico según el estado de este DAG. El DAG solo volverá a intentarlo una vez, ya que retries está establecido en 1, y lo hará después de cinco minutos, según retry_delay. Por lo general, start_date indica cuándo se debe ejecutar un DAG, junto con su schedule_interval (que se configura más adelante), pero, en el caso de este DAG, no es relevante. Está configurada para el 1 de enero de 2017, pero podría establecerse en cualquier fecha anterior.

default_args = {
    'owner': 'Composer Example',
    'depends_on_past': False,
    'email': [''],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'start_date': datetime.datetime(2017, 1, 1),
}

Nota: En este caso, no es necesario cambiar ningún parámetro, pero, si cambias alguno, asegúrate de volver a ejecutar el comando gcloud import que se mencionó antes.

La sección with airflow.DAG configura el DAG que se ejecutará. Se ejecutará con el ID de tarea composer_sample_trigger_response_dag, los argumentos predeterminados de la sección default_args y, lo más importante, con un schedule_interval de None. El valor de schedule_interval se establece en None porque activamos este DAG en particular con nuestra Cloud Function. Por eso, el start_date en default_args no es relevante.

Cuando se ejecuta, el DAG imprime su configuración, como se indica en la variable print_gcs_info.

with airflow.DAG(
        'composer_sample_trigger_response_dag',
        default_args=default_args,
        # 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 = bash_operator.BashOperator(
        task_id='print_gcs_info', bash_command='echo {{ dag_run.conf }}')

Importante: La versión más reciente de este código se encuentra en GitHub. Asegúrate de usar el comando de gcloud para copiarlo en lugar de copiarlo y pegarlo desde este fragmento de código.

6. Prueba tu función

Abre tu entorno de Composer y, en la fila con el nombre de tu entorno, haz clic en el vínculo de Airflow.
Haz clic en el nombre de composer_sample_trigger_response_dag para abrirlo. Por el momento, no habrá evidencia de las ejecuciones del DAG, ya que aún no activamos el DAG para que se ejecute.Si este DAG no está visible o no se puede hacer clic en él, espera un minuto y actualiza la página.
Abre otra pestaña y sube cualquier archivo al bucket de Cloud Storage que creaste anteriormente y que especificaste como activador de tu Cloud Function. Puedes hacerlo a través de la consola o con un comando de gsutil.
Regresa a la pestaña con la IU de Airflow y haz clic en Vista de gráfico.
Haz clic en la tarea print_gcs_info, que debería estar resaltada en verde.
Haz clic en "Ver registro" en la parte superior derecha del menú.
En los registros, verás información sobre el archivo que subiste a tu bucket de Cloud Storage.

¡Felicitaciones! Acabas de activar un DAG de Airflow con Node.js y Google Cloud Functions.

7. Limpieza

Para evitar que se generen cargos en tu cuenta de GCP por los recursos que usaste en esta guía de inicio rápido, sigue estos pasos:

1. (Opcional) Para guardar tus datos, descárgalos del bucket de Cloud Storage para el entorno de Cloud Composer y del bucket de almacenamiento que creaste para este inicio rápido.
2. Borra el bucket de Cloud Storage para el entorno que creaste.
3. Borra el entorno de Cloud Composer. Ten en cuenta que, si borras el entorno, no se borrará su bucket de almacenamiento.
4. (Opcional) Con la computación sin servidores, las primeras 2 millones de invocaciones por mes son gratuitas y, cuando reduces tu función a cero, no se te cobra (consulta los precios para obtener más detalles). Sin embargo, si quieres borrar tu función de Cloud Functions, haz clic en "BORRAR" en la parte superior derecha de la página de resumen de la función.

También puedes borrar el proyecto de forma opcional:

1. En GCP Console, ve a la página Proyectos.
2. En la lista de proyectos, selecciona el proyecto que deseas borrar y haz clic en Borrar.
3. En el cuadro, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrar el proyecto.

Precaución: Borrar un proyecto tiene las siguientes consecuencias:

• Se borra todo en el proyecto. Si utilizaste un proyecto existente para este instructivo, cuando lo borres, también borrarás cualquier otro trabajo que hayas realizado en el proyecto.
• Se pierden los ID personalizados de proyectos. Cuando creaste este proyecto, es posible que hayas creado un ID del proyecto personalizado que desees utilizar en el futuro. Para conservar las URL que usan el ID del proyecto, como una URL appspot.com, borra los recursos seleccionados dentro del proyecto en lugar de borrar todo el proyecto.