Ejecuta un trabajo de Cloud Run que procese los datos de eventos guardados en Cloud Storage

Puedes usar Workflows para ejecutar un trabajo de Cloud Run como parte de un flujo de trabajo que realice un procesamiento de datos más complejo o que organice un sistema de trabajos existentes.

En este instructivo, se muestra cómo usar Workflows para ejecutar un trabajo de Cloud Run que procesa datos de eventos almacenados en un bucket de Cloud Storage. Almacenar la carga útil del evento en un bucket de Cloud Storage te permite encriptar los datos con claves de encriptación administradas por el cliente, lo que no es posible si pasas los datos del evento como variables de entorno al trabajo de Cloud Run.

En el siguiente diagrama, se proporciona una descripción general de alto nivel:

El activador de Eventarc enruta el evento de Pub/Sub a Workflows y lo guarda en el bucket de Cloud Storage. El trabajo de Cloud Run procesa los datos de eventos almacenados en el bucket.

Crear un trabajo de Cloud Run

En este instructivo, se usa código de muestra que puedes encontrar en GitHub. La secuencia de comandos de implementación compila una imagen de contenedor para crear un trabajo de Cloud Run. La secuencia de comandos también crea un bucket de Cloud Storage. El trabajo de Cloud Run lee los datos de eventos almacenados en el bucket de Cloud Storage y, luego, los imprime.

  1. Si ejecutas la secuencia de comandos de implementación en Cloud Shell y la cuenta de servicio predeterminada de Compute Engine no tiene el rol de Editor, otorga los siguientes roles en el proyecto a la cuenta de servicio predeterminada de Compute Engine. (De lo contrario, puedes omitir este paso y continuar con la clonación del repositorio de la app de ejemplo en el siguiente paso).

    1. Otorga el rol de escritor de Artifact Registry (roles/artifactregistry.writer):

      gcloud projects add-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --role=roles/artifactregistry.writer

      Reemplaza PROJECT_NUMBER por el número de tu proyecto Google Cloud. Para encontrar el número del proyecto, ve a la página de bienvenida de la consola de Google Cloud o ejecuta el siguiente comando:

      gcloud projects describe PROJECT_ID --format='value(projectNumber)'

    2. Otorga el rol de usuario de objetos de almacenamiento (roles/storage.objectUser):

      gcloud projects add-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --role=roles/storage.objectUser

    3. Otorga el rol de escritor de registros de Logging (roles/logging.logWriter):

      gcloud projects add-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --role=roles/logging.logWriter

  2. Clona el repositorio de la app de ejemplo en tu máquina local para obtener el código de muestra:

    git clone https://github.com/GoogleCloudPlatform/workflows-demos.git

    También puedes descargar la muestra como un archivo ZIP.

  3. Dirígete al directorio que contiene el código de muestra:

    cd workflows-demos/cloud-run-jobs-payload-gcs/message-payload-job

  4. Ejecuta la secuencia de comandos de implementación para crear el trabajo de Cloud Run:

    ./deploy-job.sh

La secuencia de comandos crea un bucket de Cloud Storage con el nombre message-payload-PROJECT_ID, donde PROJECT_ID es el ID de tu proyecto Google Cloud . También se crea un trabajo de Cloud Run llamado message-payload-job.

Implementa un flujo de trabajo que ejecute el trabajo de Cloud Run

Define e implementa un flujo de trabajo que ejecute el trabajo de Cloud Run que acabas de crear. La definición de un flujo de trabajo está compuesta por una serie de pasos descritos con la sintaxis de Workflows.

El flujo de trabajo recibe un evento, guarda los datos del evento en un bucket de Cloud Storage y, luego, ejecuta un trabajo de Cloud Run para procesar los datos del evento.

Console

  1. En la consola de Google Cloud , ve a la página Workflows:

    Ir a Workflows

  2. Haz clic en Crear.

  3. Ingresa un nombre para el flujo de trabajo nuevo, como message-payload-workflow.

  4. Elige una región adecuada, por ejemplo, us-central1.

  5. En el campo Cuenta de servicio, selecciona la cuenta de servicio que creaste antes.

    La cuenta de servicio funciona como la identidad del flujo de trabajo. Ya deberías haber otorgado los siguientes roles a la cuenta de servicio:

    • Administrador de Cloud Run: Para ejecutar trabajos de Cloud Run
    • Escritor de registros: Para escribir entradas de registro
    • Creador de objetos de Storage: Para crear objetos en Cloud Storage

  6. Haz clic en Siguiente.

  7. En el editor de flujos de trabajo, ingresa la siguiente definición para el flujo de trabajo:

    main:
    params: [event]
    steps:
        - init:
            assign:
                - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                - event_bucket: ${"message-payload-" + project_id}
                - event_file: ${event.id + ".data.json"}
                - job_name: message-payload-job
                - job_location: us-central1
        - log_event:
            call: sys.log
            args:
                data: ${event}
        - write_payload_to_gcs:
            call: http.post
            args:
                url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + event_bucket + "/o"}
                auth:
                    type: OAuth2
                query:
                    name: ${event_file}
                body:
                    ${event.data}
        - run_job_to_process_payload:
            call: googleapis.run.v1.namespaces.jobs.run
            args:
                name: ${"namespaces/" + project_id + "/jobs/" + job_name}
                location: ${job_location}
                body:
                    overrides:
                        containerOverrides:
                            env:
                                - name: INPUT_BUCKET
                                  value: ${event_bucket}
                                - name: INPUT_FILE
                                  value: ${event_file}
            result: job_execution
        - finish:
            return: ${job_execution}

  8. Haz clic en Implementar.

gcloud

  1. Crea un archivo de código fuente para tu flujo de trabajo:

    touch message-payload-workflow.yaml

  2. Copia la siguiente definición de flujo de trabajo en message-payload-workflow.yaml:

    main:
    params: [event]
    steps:
        - init:
            assign:
                - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                - event_bucket: ${"message-payload-" + project_id}
                - event_file: ${event.id + ".data.json"}
                - job_name: message-payload-job
                - job_location: us-central1
        - log_event:
            call: sys.log
            args:
                data: ${event}
        - write_payload_to_gcs:
            call: http.post
            args:
                url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + event_bucket + "/o"}
                auth:
                    type: OAuth2
                query:
                    name: ${event_file}
                body:
                    ${event.data}
        - run_job_to_process_payload:
            call: googleapis.run.v1.namespaces.jobs.run
            args:
                name: ${"namespaces/" + project_id + "/jobs/" + job_name}
                location: ${job_location}
                body:
                    overrides:
                        containerOverrides:
                            env:
                                - name: INPUT_BUCKET
                                  value: ${event_bucket}
                                - name: INPUT_FILE
                                  value: ${event_file}
            result: job_execution
        - finish:
            return: ${job_execution}

  3. Para implementar el flujo de trabajo, ingresa el siguiente comando:

    gcloud workflows deploy message-payload-workflow \
    --location=us-central1 \
    --source=message-payload-workflow.yaml \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

    Reemplaza lo siguiente:

    • SERVICE_ACCOUNT_NAME: el nombre de la cuenta de servicio que creaste antes
    • PROJECT_ID: Es el ID de tu proyecto deGoogle Cloud .

    La cuenta de servicio funciona como la identidad del flujo de trabajo. Ya deberías haber otorgado los siguientes roles a la cuenta de servicio:

    • roles/logging.logWriter: Para escribir entradas de registro
    • roles/run.admin: Para ejecutar trabajos de Cloud Run
    • roles/storage.objectCreator: Para crear objetos en Cloud Storage

El flujo de trabajo hace lo siguiente:

  1. Paso init: Acepta un evento como argumento y establece las variables necesarias.

  2. Paso log_event: Crea una entrada de registro en Cloud Logging con la función sys.log.

  3. Paso write_payload_to_gcs: Realiza una solicitud POST HTTP y escribe los datos de la carga útil del evento en un archivo del bucket de Cloud Storage.

  4. Paso run_job_to_process_payload: Usa el método del conector de la API de Cloud Run Admin, googleapis.run.v1.namespaces.jobs.run, para ejecutar el trabajo. El bucket de Cloud Storage y el nombre de archivo de datos se pasan como variables de anulación del flujo de trabajo al trabajo.

  5. Paso finish: Devuelve información sobre la ejecución del trabajo como resultado del flujo de trabajo.

Crea un tema de Pub/Sub

Crea un tema de Pub/Sub para poder publicar un mensaje en él. Se usa un evento de Pub/Sub para demostrar cómo enrutar un evento con Workflows y guardarlo en Cloud Storage, de modo que un trabajo de Cloud Run pueda procesar los datos del evento.

Console

  1. En la consola de Google Cloud , ve a la página Temas.

    Ir a temas

  2. Haz clic en Crear tema.

  3. En el campo ID de tema, ingresa un ID para el tema, como message-payload-topic.

  4. Mantén la opción Agregar una suscripción predeterminada.

  5. No selecciones las otras opciones.

  6. Haz clic en Crear.

gcloud

Para crear un tema con el ID message-payload-topic, ejecuta el comando gcloud pubsub topics create:

gcloud pubsub topics create message-payload-topic

Crea un activador de Eventarc para enrutar eventos al flujo de trabajo

Para ejecutar automáticamente el flujo de trabajo y, a su vez, el trabajo de Cloud Run, crea un activador de Eventarc que responda a los eventos de Pub/Sub y que los enrute al flujo de trabajo. Cada vez que se escribe un mensaje en el tema de Pub/Sub, el evento activa una ejecución del flujo de trabajo.

Console

  1. En la consola de Google Cloud , ve a la página Workflows:

    Ir a Workflows

  2. Haz clic en el nombre de tu flujo de trabajo, como message-payload-workflow.

  3. En la página Detalles del flujo de trabajo, haz clic en Editar.

  4. En la página Editar flujo de trabajo, en la sección Activadores, haz clic en Agregar activador nuevo > Eventarc.

    Se abrirá el panel Evento de Eventarc.

  5. En el campo Nombre del activador, ingresa un nombre para el activador, como message-payload-trigger.

  6. En la lista Proveedor de eventos, selecciona Cloud Pub/Sub.

  7. En la lista Evento, selecciona google.cloud.pubsub.topic.v1.messagePublished.

  8. En la lista Selecciona un tema de Cloud Pub/Sub, selecciona el tema de Pub/Sub que creaste anteriormente.

  9. En el campo Cuenta de servicio, selecciona la cuenta de servicio que creaste antes.

    La cuenta de servicio funciona como la identidad del activador. Ya deberías haber otorgado los siguientes roles a la cuenta de servicio:

    • Receptor de eventos de Eventarc: Para recibir eventos
    • Invocador de flujos de trabajo: Para ejecutar flujos de trabajo

  10. Haz clic en Guardar activador.

    El activador de Eventarc ahora aparece en la sección Activadores de la página Editar flujo de trabajo.

  11. Haz clic en Siguiente.

  12. Haz clic en Implementar.

gcloud

Ejecuta el siguiente comando para crear un activador de Eventarc:

gcloud eventarc triggers create message-payload-trigger \
    --location=us-central1 \
    --destination-workflow=message-payload-workflow \
    --destination-workflow-location=us-central1 \
    --event-filters="type=google.cloud.pubsub.topic.v1.messagePublished" \
    --transport-topic=projects/PROJECT_ID/topics/message-payload-topic \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

Reemplaza lo siguiente:

  • PROJECT_ID: Es el ID de tu proyecto de Google Cloud .
  • SERVICE_ACCOUNT_NAME: Es el nombre de la cuenta de servicio que creaste antes.

La cuenta de servicio funciona como la identidad del activador. Ya deberías haber otorgado los siguientes roles a la cuenta de servicio:

  • roles/eventarc.eventReceiver: Para recibir eventos
  • roles/workflows.invoker: Para ejecutar flujos de trabajo

Activa el flujo de trabajo

Prueba el sistema de extremo a extremo publicando un mensaje en el tema de Pub/Sub y generando un evento. Para obtener más información, consulta cómo activar un flujo de trabajo con eventos o mensajes de Pub/Sub.

  1. Publica un mensaje en el tema de Pub/Sub para generar un evento:

    gcloud pubsub topics publish message-payload-topic --message="Hello World"

    El evento se enruta al flujo de trabajo, que registra el mensaje del evento, guarda los datos del evento en el bucket de Cloud Storage y ejecuta el trabajo de Cloud Run para procesar los datos guardados en Cloud Storage. Este proceso puede tardar un minuto.

  2. Para confirmar que el trabajo de Cloud Run se ejecutó según lo previsto, consulta las ejecuciones del trabajo:

    gcloud run jobs executions list --job=message-payload-job

    Deberías ver una nueva ejecución del trabajo en el resultado.

  3. Para ver las entradas de registro relacionadas con eventos que se crearon cuando se activó el flujo de trabajo, ejecuta el siguiente comando:

    gcloud logging read "resource.type=cloud_run_job AND textPayload:Payload"

  4. Busca una entrada de registro similar a lo siguiente:

    textPayload: "Payload: {'message': {'data': 'SGVsbG8gV29ybGQ=', 'messageId': '8834108235224238',\
\ 'publishTime': '2023-09-20T17:07:52.921Z'}, 'subscription': 'projects/MY_PROJECT/subscriptions/eventarc-us-central1-message-payload-trigger-sub-741'}"
...
resource:
labels:
  job_name: message-payload-job
  location: us-central1
  project_id: MY_PROJECT
type: cloud_run_job
textPayload: Processing message payload gs://message-payload-MY_PROJECT/8254002311197919.data.json

  5. Para confirmar que los resultados son los esperados, consulta los datos del evento en el objeto del bucket de Cloud Storage.

    1. Recupera el nombre de tu bucket:

      gcloud storage ls

      El resultado es similar a este: 

      gs://message-payload-PROJECT_ID/

    2. Enumera los objetos en tu bucket:

      gcloud storage ls gs://message-payload-PROJECT_ID/** --recursive

      El resultado debería ser similar al siguiente ejemplo:

      gs://message-payload-PROJECT_ID/OBJECT_ID.data.json

      Toma nota de OBJECT_ID para usar en el siguiente paso.

    3. Descarga el objeto de tu bucket como un archivo:

      gcloud storage cp gs://message-payload-PROJECT_ID/OBJECT_ID.data.json message-event.txt

      Reemplaza OBJECT_ID por el ID que se devolvió en el paso anterior.

    4. En un editor de texto, abre el archivo message-event.txt. El cuerpo del evento escrito en el archivo debería ser similar al siguiente:

      {
  "message": {
    "data": "SGVsbG8gV29ybGQ=",
    "messageId": "8254002311197919",
    "publishTime": "2023-09-20T16:54:29.312Z"
  },
  "subscription": "projects/MY_PROJECT/subscriptions/eventarc-us-central1-message-payload-trigger-sub-741"
}

      Ten en cuenta que, si decodificas el valor de datos de SGVsbG8gV29ybGQ= desde el formato Base64, se muestra "Hello World".