Esperar eventos mediante retrollamadas y activadores de Eventarc

Es posible que tu flujo de trabajo tenga que esperar a que se complete un proceso externo. Puedes usar retrollamadas HTTP para esperar a que otro servicio haga una solicitud a un endpoint de retrollamada. Esa solicitud reanuda la ejecución del flujo de trabajo. También puedes esperar usando el sondeo.

En lugar de usar el sondeo, en este tutorial se muestra cómo puedes esperar eventos o mensajes de Pub/Sub mediante retrollamadas HTTP y activadores de Eventarc. Aunque puedes activar un flujo de trabajo con eventos o mensajes de Pub/Sub, puede que quieras detener la ejecución para esperar otro evento antes de continuar. Por ejemplo, un evento activa un flujo de trabajo para iniciar un proceso, pero el flujo de trabajo debe esperar a otro evento que indique que el proceso se ha completado. Para ello, puedes hacer que un flujo de trabajo devuelva la llamada a otro.

Crear una base de datos de Firestore

Firestore almacena los datos en documentos que contienen campos que se asignan a valores. Estos documentos se almacenan en colecciones, que son contenedores de documentos que puedes usar para organizar tus datos y crear consultas. Consulta más información sobre Firestore.

Ten en cuenta que cada proyecto está limitado a una base de datos de Firestore. Google Cloud Sigue estos pasos si necesitas crear una base de datos.

Consola

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

    Ir a Empezar

  2. Haz clic en Seleccionar modo nativo.

    Para obtener información sobre cómo seleccionar un modo de base de datos y ver una comparación de las funciones, consulta Elegir entre el modo nativo y el modo de Datastore.

  3. En la lista Seleccionar una ubicación, selecciona nam5 (Estados Unidos).

    La ubicación se aplica tanto a la base de datos de Firestore como a la aplicación de App Engine de tu Google Cloud proyecto. Una vez que hayas creado la base de datos, no podrás cambiar la ubicación.

  4. Haz clic en Crear base de datos.

gcloud

Para crear una base de datos de Firestore, primero debes crear una aplicación de App Engine y, a continuación, ejecutar el comando gcloud firestore databases create:

gcloud app create --region=us-central
gcloud firestore databases create --region=us-central

Puedes ignorar la advertencia de us-central is not a valid Firestore location. App Engine y Firestore admiten las mismas ubicaciones, pero la región us-central (Iowa) de App Engine se corresponde con la región múltiple nam5 (Estados Unidos) de Firestore.

Crear un tema de Pub/Sub

En este tutorial se usa Pub/Sub como fuente de eventos. Crea un tema de Pub/Sub para poder publicar un mensaje en él. Más información sobre cómo crear y gestionar temas

Consola

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

    Ir a Temas

  2. Haz clic en Crear tema.

  3. En el campo ID de tema, introduce topic-callback.

  4. Acepta los demás valores predeterminados.

  5. Haz clic en Crear tema.

gcloud

Para crear un tema, ejecuta el comando gcloud pubsub topics create:

gcloud pubsub topics create topic-callback

Crea un segmento de Cloud Storage

En este tutorial se usa Cloud Storage como origen de eventos. Crea un segmento de Cloud Storage para poder subir un archivo. Más información sobre cómo crear contenedores de almacenamiento

Consola

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

    Ir a Cloud Storage

  2. Haz clic en Crear.

  3. En Nombre del contenedor, introduce PROJECT_ID-bucket-callback.

    El ID de proyecto se usa en el flujo de trabajo callback-event-sample para identificar el contenedor.

  4. Haz clic en Continuar.

  5. En Tipo de ubicación, selecciona Región y, a continuación, us-central1 (Iowa).

  6. Acepta los demás valores predeterminados.

  7. Haz clic en Crear.

gcloud

Para crear un contenedor, ejecuta el comando gcloud storage buckets create 

gcloud storage buckets create gs://PROJECT_ID-bucket-callback \
    --location=us-central1

El ID de proyecto se usa en el flujo de trabajo callback-event-sample para identificar el contenedor.

Una vez creadas las fuentes de eventos, puedes implementar el flujo de trabajo del receptor de eventos.

Desplegar un flujo de trabajo que detecte eventos

El flujo de trabajo callback-event-listener se activa cuando se publica un mensaje en un tema de Pub/Sub o cuando se sube un archivo a un segmento de Cloud Storage. El flujo de trabajo recibe el evento, obtiene los detalles de la retrollamada adecuados de la base de datos de Firestore y, a continuación, envía una solicitud HTTP al endpoint de retrollamada.

Consola

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

    Ir a Workflows

  2. Haz clic en Crear.

  3. Escribe un nombre para el nuevo flujo de trabajo: callback-event-listener.

  4. En la lista Región, selecciona us-central1.

  5. Selecciona la cuenta de servicio que has creado anteriormente.

  6. Haz clic en Siguiente.

  7. En el editor del flujo de trabajo, introduce la siguiente definición para tu flujo de trabajo:

    main:
  params: [event]
  steps:
    - log_event:
        call: sys.log
        args:
          text: ${event}
          severity: INFO
    - init:
        assign:
          - database_root: ${"projects/" + sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "/databases/(default)/documents/callbacks/"}
          - event_source_tokens: ${text.split(event.source, "/")}
          - event_source_len: ${len(event_source_tokens)}
          - event_source: ${event_source_tokens[event_source_len - 1]}
          - doc_name: ${database_root + event_source}
    - get_document_for_event_source:
        try:
          call: googleapis.firestore.v1.projects.databases.documents.get
          args:
            name: ${doc_name}
          result: document
        except:
            as: e
            steps:
                - known_errors:
                    switch:
                    - condition: ${e.code == 404}
                      return: ${"No callbacks for event source " + event_source}
                - unhandled_exception:
                    raise: ${e}
    - process_callback_urls:
        steps:
          - check_fields_exist:
              switch:
              - condition: ${not("fields" in document)}
                return: ${"No callbacks for event source " + event_source}
              - condition: true
                next: processFields
          - processFields:
              for:
                  value: key
                  in: ${keys(document.fields)}
                  steps:
                      - extract_callback_url:
                          assign:
                              - callback_url: ${document.fields[key]["stringValue"]}
                      - log_callback_url:
                          call: sys.log
                          args:
                            text: ${"Calling back url " + callback_url}
                            severity: INFO
                      - http_post:
                          call: http.post
                          args:
                              url: ${callback_url}
                              auth:
                                  type: OAuth2
                              body:
                                  event: ${event}

  8. Haz clic en Desplegar.

gcloud

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

    touch callback-event-listener.yaml

  2. En un editor de texto, copia el siguiente flujo de trabajo en tu archivo de código fuente:

    main:
  params: [event]
  steps:
    - log_event:
        call: sys.log
        args:
          text: ${event}
          severity: INFO
    - init:
        assign:
          - database_root: ${"projects/" + sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "/databases/(default)/documents/callbacks/"}
          - event_source_tokens: ${text.split(event.source, "/")}
          - event_source_len: ${len(event_source_tokens)}
          - event_source: ${event_source_tokens[event_source_len - 1]}
          - doc_name: ${database_root + event_source}
    - get_document_for_event_source:
        try:
          call: googleapis.firestore.v1.projects.databases.documents.get
          args:
            name: ${doc_name}
          result: document
        except:
            as: e
            steps:
                - known_errors:
                    switch:
                    - condition: ${e.code == 404}
                      return: ${"No callbacks for event source " + event_source}
                - unhandled_exception:
                    raise: ${e}
    - process_callback_urls:
        steps:
          - check_fields_exist:
              switch:
              - condition: ${not("fields" in document)}
                return: ${"No callbacks for event source " + event_source}
              - condition: true
                next: processFields
          - processFields:
              for:
                  value: key
                  in: ${keys(document.fields)}
                  steps:
                      - extract_callback_url:
                          assign:
                              - callback_url: ${document.fields[key]["stringValue"]}
                      - log_callback_url:
                          call: sys.log
                          args:
                            text: ${"Calling back url " + callback_url}
                            severity: INFO
                      - http_post:
                          call: http.post
                          args:
                              url: ${callback_url}
                              auth:
                                  type: OAuth2
                              body:
                                  event: ${event}

  3. Para desplegar el flujo de trabajo, introduce el siguiente comando:

    gcloud workflows deploy callback-event-listener \
    --source=callback-event-listener.yaml \
    --location=us-central1 \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

    Sustituye SERVICE_ACCOUNT_NAME por el nombre de la cuenta de servicio que has creado anteriormente.

Desplegar un flujo de trabajo que espera eventos

El flujo de trabajo callback-event-sample almacena los detalles de su retrollamada en una base de datos de Firestore, detiene su ejecución y, a continuación, espera a que se produzcan eventos específicos.

Consola

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

    Ir a Workflows

  2. Haz clic en Crear.

  3. Escribe un nombre para el nuevo flujo de trabajo: callback-event-sample.

  4. En la lista Región, selecciona us-central1.

  5. Selecciona la cuenta de servicio que has creado anteriormente.

  6. Haz clic en Siguiente.

  7. En el editor del flujo de trabajo, introduce la siguiente definición para tu flujo de trabajo:

    main:
  steps:
    - init:
        assign:
          - pubsub_topic: topic-callback
          - storage_bucket: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "-bucket-callback"}
    - await_pubsub_message:
        call: await_callback_event
        args:
          event_source: ${pubsub_topic}
        result: pubsub_event
    - await_storage_bucket:
        call: await_callback_event
        args:
          event_source: ${storage_bucket}
        result: storage_event
    - return_events:
        return:
            pubsub_event: ${pubsub_event}
            storage_event: ${storage_event}

await_callback_event:
    params: [event_source]
    steps:
        - init:
            assign:
              - database_root: ${"projects/" + sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "/databases/(default)/documents/callbacks/"}
              - doc_name: ${database_root + event_source}
              - execution_id: ${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
              - firestore_key: ${"exec_" + text.split(execution_id, "-")[0]}
        - create_callback:
            call: events.create_callback_endpoint
            args:
              http_callback_method: POST
            result: callback_details
        - save_callback_url:
            call: googleapis.firestore.v1.projects.databases.documents.patch
            args:
              name: ${doc_name}
              updateMask:
                fieldPaths: ["${firestore_key}"]
              body:
                fields:
                  ${firestore_key}:
                    stringValue: ${callback_details.url}
        - log_and_await_callback:
            try:
              steps:
                - log_await_start:
                    call: sys.log
                    args:
                      severity: INFO
                      data: ${"Started waiting 1hr for an event from source " + event_source}
                - await_callback:
                    call: events.await_callback
                    args:
                      callback: ${callback_details}
                      timeout: 3600
                    result: callback_request
                - log_await_stop:
                    call: sys.log
                    args:
                      severity: INFO
                      data: ${"Stopped waiting for an event from source " + event_source}
            except:
                as: e
                steps:
                    - log_error:
                        call: sys.log
                        args:
                            severity: "ERROR"
                            text: ${"Received error " + e.message}
        - delete_callback_url:
            call: googleapis.firestore.v1.projects.databases.documents.patch
            args:
              name: ${doc_name}
              updateMask:
                fieldPaths: ["${firestore_key}"]
        - check_null_event:
            switch:
              - condition: ${callback_request == null}
                return: null
        - log_await_result:
            call: sys.log
            args:
              severity: INFO
              data: ${callback_request.http_request.body.event}
        - return_event:
            return: ${callback_request.http_request.body.event}

  8. Haz clic en Desplegar.

gcloud

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

    touch callback-event-sample.yaml

  2. En un editor de texto, copia el siguiente flujo de trabajo en tu archivo de código fuente:

    main:
  steps:
    - init:
        assign:
          - pubsub_topic: topic-callback
          - storage_bucket: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "-bucket-callback"}
    - await_pubsub_message:
        call: await_callback_event
        args:
          event_source: ${pubsub_topic}
        result: pubsub_event
    - await_storage_bucket:
        call: await_callback_event
        args:
          event_source: ${storage_bucket}
        result: storage_event
    - return_events:
        return:
            pubsub_event: ${pubsub_event}
            storage_event: ${storage_event}

await_callback_event:
    params: [event_source]
    steps:
        - init:
            assign:
              - database_root: ${"projects/" + sys.get_env("GOOGLE_CLOUD_PROJECT_ID") + "/databases/(default)/documents/callbacks/"}
              - doc_name: ${database_root + event_source}
              - execution_id: ${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
              - firestore_key: ${"exec_" + text.split(execution_id, "-")[0]}
        - create_callback:
            call: events.create_callback_endpoint
            args:
              http_callback_method: POST
            result: callback_details
        - save_callback_url:
            call: googleapis.firestore.v1.projects.databases.documents.patch
            args:
              name: ${doc_name}
              updateMask:
                fieldPaths: ["${firestore_key}"]
              body:
                fields:
                  ${firestore_key}:
                    stringValue: ${callback_details.url}
        - log_and_await_callback:
            try:
              steps:
                - log_await_start:
                    call: sys.log
                    args:
                      severity: INFO
                      data: ${"Started waiting 1hr for an event from source " + event_source}
                - await_callback:
                    call: events.await_callback
                    args:
                      callback: ${callback_details}
                      timeout: 3600
                    result: callback_request
                - log_await_stop:
                    call: sys.log
                    args:
                      severity: INFO
                      data: ${"Stopped waiting for an event from source " + event_source}
            except:
                as: e
                steps:
                    - log_error:
                        call: sys.log
                        args:
                            severity: "ERROR"
                            text: ${"Received error " + e.message}
        - delete_callback_url:
            call: googleapis.firestore.v1.projects.databases.documents.patch
            args:
              name: ${doc_name}
              updateMask:
                fieldPaths: ["${firestore_key}"]
        - check_null_event:
            switch:
              - condition: ${callback_request == null}
                return: null
        - log_await_result:
            call: sys.log
            args:
              severity: INFO
              data: ${callback_request.http_request.body.event}
        - return_event:
            return: ${callback_request.http_request.body.event}

  3. Para desplegar el flujo de trabajo, introduce el siguiente comando:

    gcloud workflows deploy callback-event-sample \
    --source=callback-event-sample.yaml \
    --location=us-central1 \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

    Sustituye SERVICE_ACCOUNT_NAME por el nombre de la cuenta de servicio que has creado anteriormente.

Crear un activador de Eventarc para enrutar eventos de Pub/Sub

Un activador de Eventarc te permite enrutar eventos especificando filtros para el activador, como el origen del evento y el flujo de trabajo de destino. Crea un activador de Eventarc para ejecutar el flujo de trabajo callback-event-listener como resultado de la publicación de un mensaje en un tema de Pub/Sub. Consulta más información sobre cómo activar un flujo de trabajo.

Consola

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

    Ir a Eventarc

  2. Haz clic en Crear activador.

  3. Escribe un Nombre del activador.

    Por ejemplo, trigger-pubsub-events-listener.

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

  5. En la lista Evento, en Personalizado, selecciona google.cloud.pubsub.topic.v1.messagePublished.

  6. En la lista Seleccionar un tema de Cloud Pub/Sub, elija el tema que haya creado anteriormente.

  7. En la lista Región, selecciona us-central1 (Iowa).

  8. Si se te pide, otorga el rol iam.serviceAccountTokenCreator a la cuenta de servicio de Pub/Sub.

  9. Selecciona la cuenta de servicio que has creado anteriormente.

  10. En la lista Destino del evento, selecciona Workflows (Flujos de trabajo).

  11. En la lista Select a workflow (Seleccionar un flujo de trabajo), selecciona el flujo de trabajo callback-event-listener.

  12. Haz clic en Crear.

gcloud

Para crear un activador, ejecuta el comando gcloud eventarc triggers create 

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

Los eventos se transforman y se transfieren a la ejecución del flujo de trabajo como argumentos del entorno de ejecución. Ten en cuenta que el nuevo activador puede tardar hasta 2 minutos en activarse.

Crear un activador de Eventarc para enrutar eventos de Cloud Storage

Un activador de Eventarc te permite enrutar eventos especificando filtros para el activador, como el origen del evento y el flujo de trabajo de destino. Crea un activador de Eventarc para ejecutar el flujo de trabajo callback-event-listener como resultado de la subida de un archivo a un segmento de Cloud Storage. Consulta más información sobre cómo activar un flujo de trabajo.

Consola

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

    Ir a Eventarc

  2. Haz clic en Crear activador.

  3. Escribe un Nombre del activador.

    Por ejemplo, trigger-storage-events-listener.

  4. En la lista Proveedor de eventos, selecciona Cloud Storage.

  5. En la lista Event (Evento), en Direct (Directo), selecciona google.cloud.storage.object.v1.finalized.

  6. En la lista Segmento, busca el segmento que has creado anteriormente y selecciónalo.

  7. En la lista Región, en función de tu segmento de Cloud Storage, acepta el valor predeterminado us-central1 (Iowa).

  8. Si se te pide, otorga el rol iam.serviceAccountTokenCreator a la cuenta de servicio de Pub/Sub.

  9. Selecciona la cuenta de servicio que has creado anteriormente.

  10. En la lista Destino del evento, selecciona Workflows (Flujos de trabajo).

  11. En la lista Select a workflow (Seleccionar un flujo de trabajo), selecciona el flujo de trabajo callback-event-listener.

  12. Haz clic en Crear.

gcloud

Para crear un activador, ejecuta el comando gcloud eventarc triggers create 

gcloud eventarc triggers create trigger-storage-events-listener \
    --location=us-central1 \
    --destination-workflow=callback-event-listener \
    --destination-workflow-location=us-central1 \
    --event-filters="type=google.cloud.storage.object.v1.finalized" \
    --event-filters="bucket=PROJECT_ID-bucket-callback" \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

Los eventos se transforman y se transfieren a la ejecución del flujo de trabajo como argumentos del entorno de ejecución. Ten en cuenta que el nuevo activador puede tardar hasta 2 minutos en activarse.

Ejecutar el flujo de trabajo principal

Al ejecutar un flujo de trabajo, se ejecuta la definición del flujo de trabajo actual asociada al flujo de trabajo. Ejecuta el flujo de trabajo callback-event-sample. Este es el flujo de trabajo principal, que espera a que se produzcan eventos específicos y solo reanuda su ejecución cuando el flujo de trabajo secundario realiza las solicitudes de retrollamada adecuadas.

Consola

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

    Ir a Workflows

  2. En la página Flujos de trabajo, haz clic en el flujo de trabajo callback-event-sample para ir a su página de detalles.

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

  4. Vuelve a hacer clic en Ejecutar.

    Se inicia la ejecución del flujo de trabajo. Mientras se ejecuta, deberías ver el estado de ejecución Running y una entrada de registro similar a la siguiente: Started waiting 1hr for an event from source topic-callback.

gcloud

Para ejecutar un flujo de trabajo, ejecuta el comando gcloud workflows run:

gcloud workflows run callback-event-sample \
    --location=us-central1

Se inicia la ejecución del flujo de trabajo. Mientras se ejecuta, debería ver un estado de ejecución similar al siguiente:

Waiting for execution [a848a164-268a-449c-b2fe-396f32f2ed66] to complete...working...

Generar eventos y comprobar el estado de ejecución

Para confirmar que los resultados son los esperados, genere eventos, consulte las entradas de registro y compruebe el estado de ejecución del flujo de trabajo.

Publicar un mensaje

Publica un mensaje en el tema de Pub/Sub que has creado anteriormente.

Consola

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

    Ir a Temas

  2. Haz clic en topic-callback.

  3. Haz clic en la pestaña Mensajes.

  4. Haz clic en Publicar mensaje.

  5. En el campo Cuerpo del mensaje, introduce Hello World.

  6. Haz clic en Publicar.

gcloud

Para publicar un mensaje, usa el comando gcloud pubsub topics publish:

gcloud pubsub topics publish topic-callback \
    --message="Hello World"

Subir un objeto

Sube un archivo al segmento de Cloud Storage que has creado anteriormente.

Consola

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

    Ir a Contenedores

  2. Haz clic en el nombre del segmento que creaste anteriormente.

  3. En la pestaña Objetos, haga una de las siguientes acciones:

    • Arrastra el archivo que quieras desde el escritorio o el administrador de archivos hasta el panel principal de la Google Cloud consola.

    • Haz clic en Subir archivos, selecciona el archivo que quieras subir y, a continuación, haz clic en Abrir.

gcloud

Para subir un archivo, ejecuta el comando gcloud storage cp:

gcloud storage cp OBJECT_LOCATION gs://PROJECT_ID-bucket-callback/

Sustituye OBJECT_LOCATION por la ruta local a tu objeto. Por ejemplo, random.txt.

Ver entradas de registro y estado de ejecución

Confirma que el flujo de trabajo callback-event-sample se ha completado correctamente.

Consola

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

    Ir a Workflows

  2. En la página Flujos de trabajo, haz clic en el flujo de trabajo callback-event-sample para ir a su página de detalles.

  3. En la página Detalles del flujo de trabajo, para obtener los detalles de una ejecución concreta, haz clic en el ID de ejecución correspondiente.

    El Estado de ejecución debe ser Correcto y, en el panel de salida, debe ver los eventos de Pub/Sub y Cloud Storage recibidos.

gcloud

  1. Filtra las entradas de registro y devuelve el resultado en formato JSON:

    gcloud logging read "resource.type=workflows.googleapis.com/Workflow AND textPayload:calling OR textPayload:waiting" \
    --format=json

  2. Busca entradas de registro similares a las siguientes:

    "textPayload": "Stopped waiting for an event from source..."
"textPayload": "Calling back url https://workflowexecutions.googleapis.com/v1/projects/..."
"textPayload": "Started waiting 1hr for an event from source..."

  3. Comprueba el estado del último intento de ejecución:

    gcloud workflows executions wait-last

    El resultado debería ser similar al siguiente:

    Using cached execution name: projects/1085953646031/locations/us-central1/workflows/callback-event-sample/executions/79929e4e-82c1-4da1-b068-f828034c01b7
Waiting for execution [79929e4e-82c1-4da1-b068-f828034c01b7] to complete...done.
[...]
state: SUCCEEDED