Mit Callbacks warten

Callbacks ermöglichen es Workflow-Ausführungen, zu warten, bis ein anderer Dienst eine Anfrage an den Callback-Endpunkt sendet. Diese Anfrage setzt die Ausführung des Workflows fort.

Mit Callbacks können Sie Ihrem Workflow signalisieren, dass ein bestimmtes Ereignis aufgetreten ist, und auf dieses Ereignis ohne Abfrage warten. Sie können beispielsweise einen Workflow erstellen, um Sie zu benachrichtigen, wenn ein Produkt wieder auf Lager ist oder wenn ein Artikel versendet wurde. Oder Sie erstellen einen Workflow, der eine Pause macht, um menschliche Interaktion zu ermöglichen, z. B. die Überprüfung eines Auftrags oder die Validierung einer Übersetzung.

Auf dieser Seite erfahren Sie, wie Sie einen Workflow erstellen, der einen Callback-Endpunkt unterstützt und wartet, bis HTTP-Anfragen von externen Prozessen an diesem Endpunkt ankommen. Sie können auch mithilfe von Rückrufen und Eventarc-Triggern auf Ereignisse warten.

Callbacks erfordern die Verwendung von zwei integrierten Funktionen für Standardbibliotheken:

• events.create_callback_endpoint: Erstellt einen Callback-Endpunkt, der die angegebene HTTP-Methode erwartet
• events.await_callback: Wartet, bis ein Callback am angegebenen Endpunkt empfangen wird

Endpunkt erstellen, der eine Callback-Anfrage empfängt

Erstellen Sie einen Callback-Endpunkt, der HTTP-Anfragen erhalten kann, die am Endpunkt ankommen.

1. Führen Sie die Schritte zum Erstellen eines neuen Workflows aus oder wählen Sie einen vorhandenen Workflow zum Aktualisieren aus, stellen Sie ihn jedoch noch nicht bereit.
2. Fügen Sie der Definition des Workflows einen Schritt hinzu, um einen Callback-Endpunkt zu erstellen:

   YAML

       - create_callback:
           call: events.create_callback_endpoint
           args:
               http_callback_method: "METHOD"
           result: callback_details
       

   JSON

       [
         {
           "create_callback": {
             "call": "events.create_callback_endpoint",
             "args": {
               "http_callback_method": "METHOD"
             },
             "result": "callback_details"
           }
         }
       ]
         

   Ersetzen Sie METHOD durch die erwartete HTTP-Methode, entweder GET, HEAD, POST, PUT, DELETE, OPTIONS oder PATCH. Der Standardwert ist POST.

   Das Ergebnis ist eine Zuordnung callback_details mit einem Feld url, das die URL des erstellten Endpunkts speichert.

   Der Callback-Endpunkt kann jetzt eingehende Anfragen mit der angegebenen HTTP-Methode empfangen. Die URL des erstellten Endpunkts kann verwendet werden, um den Callback von einem Prozess außerhalb des Workflows auszulösen. Dazu übergeben Sie beispielsweise die URL an eine Cloud Run-Funktion.

3. Fügen Sie in der Definition des Workflows einen Schritt hinzu, um auf eine Callback-Anfrage zu warten:

   YAML

       - await_callback:
           call: events.await_callback
           args:
               callback: ${callback_details}
               timeout: TIMEOUT
           result: callback_request
       

   JSON

       [
         {
           "await_callback": {
             "call": "events.await_callback",
             "args": {
               "callback": "${callback_details}",
               "timeout": TIMEOUT
             },
             "result": "callback_request"
           }
         }
       ]
         

   Ersetzen Sie TIMEOUT durch die maximale Anzahl von Sekunden, die der Workflow auf eine Anfrage warten soll. Der Standardwert ist 43.200 (12 Stunden). Wenn diese Zeit vor dem Empfang einer Anfrage verstrichen ist, wird ein TimeoutError ausgelöst.

   Beachten Sie, dass es eine maximale Ausführungsdauer gibt. Weitere Informationen finden Sie unter Anfragelimit.

   Die callback_details-Karte aus dem vorherigen Schritt create_callback wird als Argument übergeben.

4. Stellen Sie Ihren Workflow bereit, um ihn zu erstellen oder zu aktualisieren.

   Wenn eine Anfrage empfangen wird, werden alle Details der Anfrage in der Zuordnung callback_request gespeichert. Anschließend haben Sie Zugriff auf die gesamte HTTP-Anfrage, einschließlich Header, Text und eine query-Zuordnung für alle Abfrageparameter. Beispiel:

   YAML

       http_request:
         body:
         headers: {...}
         method: GET
         query: {}
         url: "/v1/projects/350446661175/locations/us-central1/workflows/workflow-1/executions/46804f42-dc83-46d6-87e4-93962866ed81/callbacks/49c80102-74d2-49cd-a70e-805a9fded94f_2de9b413-6332-412d-99c3-d7e9b6eeeda2"
       received_time: 2021-06-24 12:49:16.988072651 -0700 PDT m=+742581.005780667
       type: HTTP
       

   JSON

       {
          "http_request":{
             "body":null,
             "headers":{
                ...
             },
             "method":"GET",
             "query":{
             },
             "url":"/v1/projects/350446661175/locations/us-central1/workflows/workflow-1/executions/46804f42-dc83-46d6-87e4-93962866ed81/callbacks/49c80102-74d2-49cd-a70e-805a9fded94f_2de9b413-6332-412d-99c3-d7e9b6eeeda2"
          },
          "received_time":"2021-06-24 12:49:16.988072651 -0700 PDT m=+742581.005780667",
          "type":"HTTP"
       }
         

   Wenn der HTTP-Text ein Text oder JSON ist, versucht Workflows, den Text zu decodieren. Andernfalls werden Rohbyte zurückgegeben.

Anfragen an den Callback-Endpunkt autorisieren

Zum Senden einer Anfrage an einen Callback-Endpunkt müssen Google Cloud-Dienste wie Cloud Run und Cloud Run-Funktionen sowie Dienste von Drittanbietern mit den entsprechenden IAM-Berechtigungen dazu autorisiert werden, und zwar durch die Rolle workflows.callbacks.send (in der Rolle Workflows Invoker enthalten).

Direkte Anfrage stellen

Am einfachsten erstellen Sie kurzlebige Anmeldedaten für ein Dienstkonto, indem Sie eine direkte Anfrage stellen. An diesem Ablauf sind zwei Identitäten beteiligt: der Aufrufer und das Dienstkonto, für das die Anmeldedaten erstellt werden. Der Aufruf des grundlegenden Workflows auf dieser Seite ist ein Beispiel für eine direkte Anfrage. Weitere Informationen finden Sie unter Zugriff mit IAM steuern und Direkte Anfrageberechtigungen.

OAuth 2.0-Zugriffstoken generieren

Zum Autorisieren einer Anwendung zum Aufrufen des Callback-Endpunkts können Sie ein OAuth 2.0-Zugriffstoken für das mit dem Workflow verknüpfte Dienstkonto generieren. Wenn Sie die erforderlichen Berechtigungen (für die Rollen Workflows Editor oder Workflows Admin und Service Account Token Creator) haben, können Sie selbst ein Token generieren. Führen Sie dazu die generateAccessToken-Methode aus.

Wenn die Anfrage generateAccessToken erfolgreich ist, enthält der zurückgegebene Antworttext ein OAuth 2.0-Zugriffstoken und eine Ablaufzeit. Standardmäßig sind OAuth 2.0-Zugriffstokens maximal eine Stunde lang gültig. Beispiel:

  {
  "accessToken": "eyJ0eXAi...NiJ9",
  "expireTime": "2020-04-07T15:01:23.045123456Z"
  }

  curl -X GET -H "Authorization: Bearer ACCESS_TOKEN_STRING" CALLBACK_URL

  curl -X POST -H "Content-Type: application/json" -H "Authorization: Bearer ACCESS_TOKEN_STRING" -d '{"foo" : "bar"}' CALLBACK_URL

OAuth-Token für eine Cloud Run-Funktion generieren

Wenn Sie einen Callback von einer Cloud Run-Funktion aufrufen, die dasselbe Dienstkonto wie der Workflow im selben Projekt verwendet, können Sie ein OAuth-Zugriffstoken in der Funktion selbst generieren. Beispiel:

const {GoogleAuth} = require('google-auth-library');
const auth = new GoogleAuth();
const token = await auth.getAccessToken();
console.log("Token", token);

try {
  const resp = await fetch(url, {
      method: 'POST',
      headers: {
          'accept': 'application/json',
          'content-type': 'application/json',
          'authorization': `Bearer ${token}`
      },
      body: JSON.stringify({ approved })
  });
  console.log("Response = ", JSON.stringify(resp));

  const result = await resp.json();
  console.log("Outcome = ", JSON.stringify(result));

Weitere Informationen finden Sie in der Anleitung Human-in-the-Loop-Workflow mit Callbacks erstellen.

Offlinezugriff anfordern

Zugriffstokens laufen regelmäßig ab und werden zu ungültigen Anmeldedaten für eine zugehörige API-Anfrage. Sie können ein Zugriffstoken aktualisieren, ohne den Nutzer um eine Berechtigung zu bitten, wenn Sie den Offlinezugriff auf die mit dem Token verknüpften Bereiche angefordert haben. Das Anfordern des Offlinezugriffs ist eine Voraussetzung für jede Anwendung, die auf eine Google API zugreifen muss, wenn der Nutzer nicht vorhanden ist. Weitere Informationen finden Sie unter Zugriffstoken aktualisieren (Offlinezugriff).

Workflows mithilfe von Callbacks genau einmal aufrufen

Callbacks sind vollständig idempotent. Das bedeutet, dass Sie einen Callback wiederholen können, wenn er fehlschlägt, ohne dass unerwartete Ergebnisse oder Nebenwirkungen auftreten.

Nachdem Sie einen Callback-Endpunkt erstellt haben, kann die URL eingehende Anfragen empfangen und wird in der Regel an einen Anrufer zurückgegeben, bevor der entsprechende Aufruf an await_callback erfolgt. Wenn die Callback-URL jedoch noch nicht empfangen wurde, wenn der Schritt await_callback ausgeführt wird, wird die Workflowausführung blockiert, bis der Endpunkt empfangen wird (oder eine Zeitüberschreitung auftritt). Sobald die Antwort eingegangen ist, wird die Workflowausführung fortgesetzt und der Rückruf wird verarbeitet.

Nachdem Sie den Schritt create_callback_endpoint ausgeführt und einen Callback-Endpunkt erstellt haben, ist für den Workflow ein einzelner Callback-Slot verfügbar. Wenn eine Callback-Anfrage eingeht, wird dieser Slot mit der Callback-Nutzlast gefüllt, bis der Callback verarbeitet werden kann. Wenn der Schritt await_callback ausgeführt wird, wird der Rückruf verarbeitet und der Slot wird geleert und für einen anderen Rückruf verfügbar gemacht. Sie können den Callback-Endpunkt dann wiederverwenden und await_callback noch einmal aufrufen.

Wenn await_callback nur einmal aufgerufen wird, aber ein zweiter Rückruf empfangen wird, tritt eines der folgenden Szenarien auf und ein entsprechender HTTP-Statuscode wird zurückgegeben:

• HTTP 429: Too Many Requests gibt an, dass der erste Rückruf erfolgreich empfangen, aber noch nicht verarbeitet wurde. Der zweite Rückruf wird vom Workflow abgelehnt.

• HTTP 200: Success gibt an, dass der erste Callback erfolgreich empfangen und eine Antwort zurückgegeben wurde. Der zweite Callback wird gespeichert und möglicherweise nie verarbeitet, es sei denn, await_callback wird noch einmal aufgerufen. Wenn der Workflow vor diesem Zeitpunkt endet, wird die zweite Callback-Anfrage nie verarbeitet und verworfen.

• HTTP 404: Page Not Found gibt an, dass der Workflow nicht mehr ausgeführt wird. Entweder wurde der erste Rückruf verarbeitet und der Workflow ist abgeschlossen oder der Workflow ist fehlgeschlagen. Dazu müssen Sie den Ausführungsstatus des Workflows abfragen.

Parallele Callbacks

Wenn Schritte parallel ausgeführt werden und ein Callback von einem übergeordneten Thread erstellt und in untergeordneten Schritten abgewartet wird, wird dasselbe Muster wie zuvor beschrieben verwendet.

Im folgenden Beispiel wird beim Ausführen des Schritts create_callback_endpoint ein Callback-Slot erstellt. Bei jedem nachfolgenden Aufruf von await_callback wird ein neuer Rückruf-Slot geöffnet. Es können zehn Callbacks gleichzeitig ausgeführt werden, wenn alle Threads ausgeführt werden und warten, bevor eine Callback-Anfrage gestellt wird. Es können weitere Callbacks ausgeführt werden, die jedoch gespeichert und nie verarbeitet werden.

  - createCallbackInParent:
    call: events.create_callback_endpoint
    args:
      http_callback_method: "POST"
    result: callback_details
  - parallelStep:
    parallel:
        for:
            range: [1, 10]
            value: loopValue
            steps:
              - waitForCallbackInChild:
                  call: events.await_callback
                  args:
                      callback: ${callback_details}

  [
    {
      "createCallbackInParent": {
        "call": "events.create_callback_endpoint",
        "args": {
          "http_callback_method": "POST"
        },
        "result": "callback_details"
      }
    },
    {
      "parallelStep": {
        "parallel": {
          "for": {
            "range": [
              1,
              10
            ],
            "value": "loopValue",
            "steps": [
              {
                "waitForCallbackInChild": {
                  "call": "events.await_callback",
                  "args": {
                    "callback": "${callback_details}"
                  }
                }
              }
            ]
          }
        }
      }
    }
  ]

Hinweis: Rückrufe werden in der Reihenfolge verarbeitet, in der die einzelnen Aufrufe über einen Branch zu await_callback erfolgen. Die Ausführungsreihenfolge der Verzweigungen ist jedoch nicht deterministisch und kann zu einem Ergebnis über verschiedene Pfade führen. Weitere Informationen finden Sie unter Parallele Schritte.

Einfachen Callback-Workflow testen

Sie können einen einfachen Workflow erstellen und dann den Aufruf des Callback-Endpunkts dieses Workflows mit curl testen. Sie benötigen die erforderlichen Workflows Editor- oder Workflows Admin-Berechtigungen für das Projekt, in dem sich der Workflow befindet.

1. Erstellen und bereitstellen Sie den folgenden Workflow und führen Sie ihn aus.
   

   YAML

       - create_callback:
           call: events.create_callback_endpoint
           args:
               http_callback_method: "GET"
           result: callback_details
       - print_callback_details:
           call: sys.log
           args:
               severity: "INFO"
               text: ${"Listening for callbacks on " + callback_details.url}
       - await_callback:
           call: events.await_callback
           args:
               callback: ${callback_details}
               timeout: 3600
           result: callback_request
       - print_callback_request:
           call: sys.log
           args:
               severity: "INFO"
               text: ${"Received " + json.encode_to_string(callback_request.http_request)}
       - return_callback_result:
           return: ${callback_request.http_request}
       

   JSON

       [
         {
           "create_callback": {
             "call": "events.create_callback_endpoint",
             "args": {
               "http_callback_method": "GET"
             },
             "result": "callback_details"
           }
         },
         {
           "print_callback_details": {
             "call": "sys.log",
             "args": {
               "severity": "INFO",
               "text": "${\"Listening for callbacks on \" + callback_details.url}"
             }
           }
         },
         {
           "await_callback": {
             "call": "events.await_callback",
             "args": {
               "callback": "${callback_details}",
               "timeout": 3600
             },
             "result": "callback_request"
           }
         },
         {
           "print_callback_request": {
             "call": "sys.log",
             "args": {
               "severity": "INFO",
               "text": "${\\"Received \\" + json.encode_to_string(callback_request.http_request)}"
             }
           }
         },
         {
           "return_callback_result": {
             "return": "${callback_request.http_request}"
           }
         }
       ]
         

   Nach der Ausführung des Workflows ist der Status der Workflowausführung ACTIVE, bis die Rückrufanfrage empfangen wird oder das Zeitlimit abgelaufen ist.

2. Bestätigen Sie den Ausführungsstatus und rufen Sie die Callback-URL ab:
   

   Console

   1. Rufen Sie in der Google Cloud Console die Seite Workflows auf:

      Zur Seite "Workflows"

   2. Klicken Sie auf den Namen des gerade ausgeführten Workflows.

      Der Status der Workflow-Ausführung wird angezeigt.

   3. Klicken Sie auf den Tab Logs.

   4. Suchen Sie nach einem Logeintrag wie dem Folgenden:

      Listening for callbacks on https://workflowexecutions.googleapis.com/v1/projects/...

   5. Kopieren Sie die Callback-URL, die im nächsten Befehl verwendet werden soll.

   gcloud

   1. Rufen Sie zuerst die Ausführungs-ID ab:

      gcloud logging read "Listening for callbacks" --freshness=DURATION

      Ersetzen Sie DURATION durch eine geeignete Zeit, um die zurückgegebenen Logeinträge zu begrenzen, wenn Sie den Workflow mehrmals ausgeführt haben.

      Beispiel: --freshness=t10m gibt Logeinträge zurück, die nicht älter als 10 Minuten sind. Weitere Informationen finden Sie unter gcloud topic datetimes.

      Die Ausführungs-ID wird zurückgegeben. Die Callback-URL wird ebenfalls im Feld textPayload zurückgegeben. Kopieren Sie beide Werte, um sie in den folgenden Schritten zu verwenden.

   2. Führen Sie dazu diesen Befehl aus:

      gcloud workflows executions describe WORKFLOW_EXECUTION_ID --workflow=WORKFLOW_NAME

      Der Status der Workflowausführung wird zurückgegeben.
3. Sie können jetzt mit einem curl-Befehl den Callback-Endpunkt aufrufen:

   curl -X GET -H "Authorization: Bearer $(gcloud auth print-access-token)" CALLBACK_URL

   Für einen POST-Endpunkt muss ein Content-Type-Darstellungsheader verwendet werden. Beispiel:

   curl -X POST -H "Content-Type: application/json" -H "Authorization: Bearer $(gcloud auth print-access-token)" -d '{"foo" : "bar"}' CALLBACK_URL

   Ersetzen Sie CALLBACK_URL durch die URL, die Sie im vorherigen Schritt kopiert haben.

4. Prüfen Sie über die Google Cloud Console oder die Google Cloud CLI, ob der Status der Workflowausführung jetzt SUCCEEDED lautet.
5. Suchen Sie nach dem Logeintrag mit dem zurückgegebenen textPayload, der in etwa so aussieht:

   Received {"body":null,"headers":...

Beispiele

Diese Beispiele veranschaulichen die Syntax.

Zeitüberschreitungsfehler abfangen

Dieses Beispiel erweitert das vorherige Beispiel, indem Zeitüberschreitungsfehler abgefangen und die Fehler in das Systemlog geschrieben werden.

    main:
      steps:
        - create_callback:
            call: events.create_callback_endpoint
            args:
                http_callback_method: "GET"
            result: callback_details
        - print_callback_details:
            call: sys.log
            args:
                severity: "INFO"
                text: ${"Listening for callbacks on " + callback_details.url}
        - await_callback:
            try:
                call: events.await_callback
                args:
                    callback: ${callback_details}
                    timeout: 3600
                result: callback_request
            except:
                as: e
                steps:
                    - log_error:
                        call: sys.log
                        args:
                            severity: "ERROR"
                            text: ${"Received error " + e.message}
                        next: end
        - print_callback_result:
            call: sys.log
            args:
                severity: "INFO"
                text: ${"Received " + json.encode_to_string(callback_request.http_request)}
    

    {
      "main": {
        "steps": [
          {
            "create_callback": {
              "call": "events.create_callback_endpoint",
              "args": {
                "http_callback_method": "GET"
              },
              "result": "callback_details"
            }
          },
          {
            "print_callback_details": {
              "call": "sys.log",
              "args": {
                "severity": "INFO",
                "text": "${\"Listening for callbacks on \" + callback_details.url}"
              }
            }
          },
          {
            "await_callback": {
              "try": {
                "call": "events.await_callback",
                "args": {
                  "callback": "${callback_details}",
                  "timeout": 3600
                },
                "result": "callback_request"
              },
              "except": {
                "as": "e",
                "steps": [
                  {
                    "log_error": {
                      "call": "sys.log",
                      "args": {
                        "severity": "ERROR",
                        "text": "${\"Received error \" + e.message}"
                      },
                      "next": "end"
                    }
                  }
                ]
              }
            }
          },
          {
            "print_callback_result": {
              "call": "sys.log",
              "args": {
                "severity": "INFO",
                "text": "${\"Received \" + json.encode_to_string(callback_request.http_request)}"
              }
            }
          }
        ]
      }
    }
      

Auf einen Callback in einer Wiederholungsschleife warten

In diesem Beispiel wird das vorherige Beispiel durch Implementieren eines Wiederholungsschritts geändert. Mit einem benutzerdefinierten Wiederholungsprädikat protokolliert der Workflow eine Warnung, wenn ein Zeitlimit auftritt, und wiederholt die Wartezeit auf dem Callback-Endpunkt bis zu fünfmal. Wenn das Wiederholungskontingent aufgebraucht ist, bevor der Callback empfangen wird, führt der letzte Zeitüberschreitungsfehler dazu, dass der Workflow fehlschlägt.

    main:
      steps:
        - create_callback:
            call: events.create_callback_endpoint
            args:
                http_callback_method: "GET"
            result: callback_details
        - print_callback_details:
            call: sys.log
            args:
                severity: "INFO"
                text: ${"Listening for callbacks on " + callback_details.url}
        - await_callback:
            try:
                call: events.await_callback
                args:
                    callback: ${callback_details}
                    timeout: 60.0
                result: callback_request
            retry:
                predicate: ${log_timeout}
                max_retries: 5
                backoff:
                    initial_delay: 1
                    max_delay: 10
                    multiplier: 2
        - print_callback_result:
            call: sys.log
            args:
                severity: "INFO"
                text: ${"Received " + json.encode_to_string(callback_request.http_request)}
    log_timeout:
        params: [e]
        steps:
          - when_to_repeat:
              switch:
                - condition: ${"TimeoutError" in e.tags}
                  steps:
                      - log_error_and_retry:
                          call: sys.log
                          args:
                              severity: "WARNING"
                              text: "Timed out waiting for callback, retrying"
                      - exit_predicate:
                          return: true
          - otherwise:
              return: false
    

    {
      "main": {
        "steps": [
          {
            "create_callback": {
              "call": "events.create_callback_endpoint",
              "args": {
                "http_callback_method": "GET"
              },
              "result": "callback_details"
            }
          },
          {
            "print_callback_details": {
              "call": "sys.log",
              "args": {
                "severity": "INFO",
                "text": "${\"Listening for callbacks on \" + callback_details.url}"
              }
            }
          },
          {
            "await_callback": {
              "try": {
                "call": "events.await_callback",
                "args": {
                  "callback": "${callback_details}",
                  "timeout": 60
                },
                "result": "callback_request"
              },
              "retry": {
                "predicate": "${log_timeout}",
                "max_retries": 5,
                "backoff": {
                  "initial_delay": 1,
                  "max_delay": 10,
                  "multiplier": 2
                }
              }
            }
          },
          {
            "print_callback_result": {
              "call": "sys.log",
              "args": {
                "severity": "INFO",
                "text": "${\"Received \" + json.encode_to_string(callback_request.http_request)}"
              }
            }
          }
        ]
      },
      "log_timeout": {
        "params": [
          "e"
        ],
        "steps": [
          {
            "when_to_repeat": {
              "switch": [
                {
                  "condition": "${\"TimeoutError\" in e.tags}",
                  "steps": [
                    {
                      "log_error_and_retry": {
                        "call": "sys.log",
                        "args": {
                          "severity": "WARNING",
                          "text": "Timed out waiting for callback, retrying"
                        }
                      }
                    },
                    {
                      "exit_predicate": {
                        "return": true
                      }
                    }
                  ]
                }
              ]
            }
          },
          {
            "otherwise": {
              "return": false
            }
          }
        ]
      }
    }
      

Nächste Schritte

• Anleitung: Workflow mithilfe von Callbacks und Google Tabellen pausieren und fortsetzen

• Referenz zur Workflows-Syntax