Empfangene Ereignisse transformieren

Sie können Ihre Ereignisdaten transformieren, indem Sie Transformationsausdrücke mit CEL schreiben. Beispielsweise lassen sich Ereignisnutzlasten ändern, um dem API-Vertrag eines bestimmten Ziels zu entsprechen.

Beachten Sie, dass Ereignisse immer in einem CloudEvents-Format mit einer HTTP-Anfrage im Binärinhaltsmodus gesendet werden, es sei denn, Sie geben eine Nachrichtenbindung an.

Eingabe- und Ausgabedatenformate festlegen

Neben dem Schreiben eines Transformationsausdrucks in CEL können Sie optional das Datenformat der eingehenden Ereignisdaten angeben. So weiß Eventarc Advanced, wie die Nutzlast des Ereignisses geparst werden muss. Sie können die Daten auch von einem Format in ein anderes konvertieren.

Die folgenden Formate werden unterstützt: Avro, JSON und Protobuf. Weitere Informationen finden Sie unter Empfangene Ereignisse formatieren.

Transformationsausdrücke

Bei der Transformation von Ereignissen kann in einem CEL-Ausdruck über ein vordefiniertes message-Objekt als Variablen auf alle Ereignisattribute zugegriffen werden. Diese Variablen werden zur Laufzeit anhand der Ereignisdaten mit Werten gefüllt. Beispiel:

  • message.id gibt das Attribut id des Ereignisses zurück.
  • message.data gibt eine CEL-Wertdarstellung der Ereignisnutzlast zurück.
  • message.data.some-key gibt den Inhalt eines Felds mit dem Namen some-key aus der Ereignisnutzlast zurück.

Felder in message.data werden immer als String-Typen dargestellt und Werte werden aus dem ursprünglichen Ereignis mithilfe des Schemas zugeordnet, das beim Festlegen des Eingabedatenformats angegeben wurde.

Der Transformationsausdruck sollte ein vollständiges Ereignis darstellen, das die Attribute des Ereigniskontexts und die Ereignisdaten-Nutzlast enthält. Ausdrücke werden in JSON geschrieben, aber vordefinierte CEL-Funktionen, -Makros und -Operatoren sowie reguläre Ausdrücke mit RE2 werden unterstützt. Eventarc Advanced unterstützt auch bestimmte Erweiterungsfunktionen, mit denen die Ereignisdaten transformiert werden können.

Im Folgenden finden Sie zwei Beispiele für die Verwendung von CEL-Ausdrücken zum Transformieren Ihrer Ereignisdaten. Weitere Anwendungsfälle und Beispiele finden Sie unter Transformationsbeispiele.

Beispiel: Attributwerte formatieren

Im folgenden Beispiel werden phone_number Attributwerte mit Funktionen für reguläre Ausdrücke formatiert. (Andere Attribute wurden weggelassen.)

  // Input:
  // {
  //   "data":
  //   {
  //     "email_address": "charlie@altostrat.com",
  //     "phone_number": "8005550100",
  //   }
  // }
  // Output:
  // {
  //    "data":
  //    {
  //      "email_domain": "altostrat.com",
  //      "phone_number": "(800) 555-0100",
  //      "area_code": "800",
  //      "local_number": "5550100",
  //    }
  // }

  {
    "data":
    {
      "email_domain": re.capture(
                        message.data.email_address,
                        "\\S+@(\\S+)"),

      "phone_number": re.extract(
                        message.data.phone_number,
                        "^(\\d{3})(\\d{3})(\\d{4})", "(\\1) \\2-\\3"
                      ),

    }.merge ( re.captureN(message.data.phone_number,
                        "^(?P\d{3})[\w\-)(]*(?P\d{7})"
                      )
    )
  }

Das sind die Funktionen für reguläre Ausdrücke, die im vorherigen Beispiel verwendet wurden:

  • re.capture: erfasst den ersten unbenannten oder benannten Gruppenwert. Die Argumente sind:
    • target: String, der geparst werden soll
    • regex: regulärer Ausdruck, der zum Erfassen von Werten verwendet wird

    Gibt einen String des ersten erfassten Gruppenwerts zurück.

  • re.captureN: führt eine vollständige Übereinstimmung für den angegebenen String und den regulären Ausdruck aus. Die Argumente sind:
    • target: String , der geparst werden soll
    • regex: regulärer Ausdruck, der zum Erfassen von Werten verwendet wird

    Gibt eine Zuordnung mit Schlüssel/Wert-Paaren für eine benannte Gruppe (Gruppenname, erfasster String) oder eine unbenannte Gruppe (Gruppenindex, erfasster String) zurück.

  • re.extract: gleicht Gruppenwerte aus dem angegebenen Zielstring ab und schreibt den String neu. Die Argumente sind:
    • target: String, der geparst werden soll
    • regex: regulärer Ausdruck der zum Extrahieren von Werten verwendet wird
    • rewrite: regulärer Ausdruck für die Formatierung des Ergebnisses

    Gibt einen String der extrahierten Werte zurück, der basierend auf dem rewrite Argument formatiert ist.

Beispiel: Array einem Array von Objekten zuordnen

Im folgenden Beispiel wird ein Array von Ganzzahlen einem Array von Objekten zugeordnet. (Andere Attribute wurden weggelassen.)

  // Input:
  // {
  //   "data":
  //   {
  //        "product_ids": [1, 2, 3]
  //   }
  // }
  // Output:
  // {
  //    "data":
  //    {
  //             "products": [
  //                {
  //                   "name": "apple",
  //                   "price": 70
  //                },
  //                {
  //                    "name": "orange",
  //                    "price":  80
  //                },
  //                {
  //                    "name": "Product(3)",
  //                    "price": 0
  //                },
  //                {
  //                     "name": "apple",
  //                     "price": 70
  //                }
  //            ]
  //    }
  // }

  {
    "data":
    {
      "products":  message.data.product_ids.map(product_id,
              product_id == 1?
              {
                "name": "apple",
                "price": 70
              } :
              product_id == 2?
              {
                "name": "orange",
                "price":  80
              } :
              // Default:
              {
                "name": "Product(" + string(product_id) + ")",
                "price": 0
              }
          )
    }
  }

Pipeline zum Transformieren von Ereignissen konfigurieren

Sie können eine Pipeline konfigurieren, um Ereignisdaten in der Google Cloud Console oder mit der gcloud CLI zu transformieren.

Pro Pipeline wird nur eine Mediation unterstützt.

Console

  1. Rufen Sie in der Google Cloud Console die Seite Eventarc > Pipelines auf.

    Zu Pipelines

  2. Sie können eine Pipeline erstellen oder, wenn Sie eine Pipeline aktualisieren, auf den Namen der Pipeline klicken.

  3. Klicken Sie auf der Seite Pipelines-Details auf Bearbeiten.

  4. Führen Sie im Bereich Ereignismediation folgende Schritte aus:

    1. Klicken Sie das Kästchen Transformation anwenden an.

    2. Wählen Sie in der Liste Eingangsformat das entsprechende Format aus.

      Weitere Informationen finden Sie unter Empfangene Ereignisse formatieren.

    3. Schreiben Sie im Feld CEL-Ausdruck einen Transformationsausdruck in JSON. Vordefinierte CEL Funktionen, Makros und Operatoren sowie reguläre Ausdrücke werden unterstützt. Beispiel:

      {
"id": message.id,
"datacontenttype": "application/json",
"data": "{ \"scrubbed\": \"true\" }"
}

      Das vorherige Beispiel führt Folgendes aus:

      • Entfernt alle Attribute aus dem ursprünglichen Ereignis, außer der id
      • Legt das Attribut datacontenttype auf application/json fest
      • Ersetzt die Ereignisnutzlast durch einen statischen JSON-String

    4. Klicken Sie auf Weiter.

  5. Führen Sie im Bereich Ziel folgende Schritte aus:

    1. Wählen Sie gegebenenfalls in der Liste Ausgangsformat ein Format aus.

      Weitere Informationen finden Sie unter Empfangene Ereignisse formatieren.

    2. Wenden Sie optional eine Nachrichtenbindung an. Weitere Informationen finden Sie im Abschnitt Nachrichtenbindung definieren in diesem Dokument.

  6. Klicken Sie auf Speichern.

    Das Aktualisieren einer Pipeline kann einige Minuten dauern.

gcloud

  1. Öffnen Sie ein Terminalfenster.

  2. Sie können eine Pipeline erstellen oder eine Pipeline mit dem gcloud eventarc pipelines update Befehl aktualisieren:

    gcloud eventarc pipelines update PIPELINE_NAME \
    --location=REGION \
    --mediations=transformation_template=\
'
 {
    TRANSFORMATION_EXPRESSION
 }
'

    Ersetzen Sie Folgendes:

    • PIPELINE_NAME: die ID der Pipeline oder ein voll qualifizierter Name

    • REGION: ein unterstützter Eventarc Advanced-Standort

      Alternativ können Sie das Standortattribut der gcloud CLI festlegen:

      gcloud config set eventarc/location REGION

    • TRANSFORMATION_EXPRESSION: ein in JSON geschriebener Ausdruck. Vordefinierte CEL-Funktionen, -Makros, und -Operatoren sowie reguläre Ausdrücke werden unterstützt. Mit einem mediations-Flag wird ein transformation_template-Schlüssel angewendet.

    Das Aktualisieren einer Pipeline kann einige Minuten dauern.

    Beispiel:

    gcloud eventarc pipelines update my-pipeline \
    --location=us-central1 \
    --mediations=transformation_template=\
'
 {
    "id": message.id,
    "datacontenttype": "application/json",
    "data": "{ \"scrubbed\": \"true\" }"
 }
'

    Das vorherige Beispiel führt Folgendes aus:

    • Entfernt alle Attribute aus dem ursprünglichen Ereignis, außer der id
    • Legt das Attribut datacontenttype auf application/json fest
    • Ersetzt die Ereignisnutzlast durch einen statischen JSON-String

Erweiterungsfunktionen

Eventarc Advanced unterstützt die folgenden Erweiterungsfunktionen, mit denen die über einen Bus empfangenen Ereignisdaten transformiert werden können.

Funktion Beschreibung
denormalize

Denormalisiert eine Zuordnung oder Liste, indem redundante Daten hinzugefügt werden, um die Lese leistung zu verbessern. Feldnamen in der resultierenden Zuordnung werden mit einem Punkt (.) getrennt. Der Listenindex wird in einen Stringschlüssel konvertiert, beginnend mit 0.

Beachten Sie, dass Sie in Avro- und Protobuf-Feldnamen keinen Punkt (.) verwenden können. Verwenden Sie diese Funktion daher nur für JSON-Daten.

Beispiel: map.() -> map(string, dyn) oder list() -> map(string, dyn)
merge

Verknüpft zwei Felder und gibt das kombinierte Feld zurück. Felder mit doppelten Namen werden zusammengeführt.

Beispiel: message.(message) -> message
removeFields

Entfernt bestimmte Felder aus einem Ereignis. Feldnamen werden als Pfade aufgelöst. Das Punktzeichen (.) wird als Trennzeichen verwendet.

Es wird rohes JSON erwartet. Wenn Sie das JSON serialisieren, wird die Transformation möglicherweise auf einen JSON-String angewendet und führt zu einem Fehler.

Beispiel: message.(list(string)) -> message
setField

Fügt ein Feld des Ereignisses mit einem bestimmten Schlüssel hinzu oder ersetzt es. Der Feld name wird als Pfad aufgelöst. Das Punktzeichen (.) wird als ein Trennzeichen verwendet.

Beispiel: message.(string, dyn) -> message

Beispiel: Attribut zur Ereignisnutzlast hinzufügen, ohne andere Daten zu ändern

// Input:
// {
//   "data": 
//   {
//        "credit_card_number": "XXXX-XXXX-XXXX-XXXX"
//   }
// }
// Output:
// {
//    "data":
//    {
//        "credit_card_number": "XXXX-XXXX-XXXX-XXXX",
//        "card_type": "credit"
//    }
// }
{
  "data": message.data.merge(
    {
      "card_type": "credit"
    }
  )
}

Beispiel: Liste von Elementen aus der Ereignisnutzlast denormalisieren

// Input:
//{
//"data": 
//   {
//        "products": [
//          {
//            "number": 021774,
//            "type": "perishable",
//            "price": 2.00
//          },
//          {
//            "number": 95602,
//            "type": "diy",
//            "price": 120.00
//          },
//          {
//            "number": 568302,
//            "type": "toys",
//            "price": 12.00
//          }
//        ]
//   }
//}
//
// Output:
//{
//"data":
//    {
//        "products": {
//            "0.number": 021774,
//            "0.type": "perishable",
//            "0.price": 2.00,
//            "1.number": 95602,
//            "1.type": "diy",
//            "1.price": 120.00,
//            "2.number": 568302,
//            "2.type": "toys",
//            "2.price": 12.00
//          }
//   }
//}
//
//
message.setField("data.products", message.data.products.denormalize())

Beispiel: Feld aus der Ereignisnutzlast entfernen

// Input:
// {
//   "data": 
//   {
//     "payment": {
//       "card_number": "XXXX-XXXX-XXXX-XXXX",
//       "card_type": "credit",
//     }
//   }
// }
// Output:
// {
//   "data":
//   {
//     "payment": {
//       "card_type": "credit"
//     }
//   }
// }
message.removeFields(["data.payment.card_number"])

Nachrichtenbindung definieren

Standardmäßig werden Ereignisse immer im CloudEvents-Format mit einer HTTP-Anfrage im Binärinhaltsmodus an ein Ziel gesendet. Sie können dieses Verhalten überschreiben, indem Sie eine Nachrichtenbindung definieren und eine neue HTTP-Anfrage erstellen.

Alle HTTP-Header, die durch andere Richtlinien oder Steuerelemente eingeführt werden (z. B. OAuth- oder OIDC-Tokens), bleiben erhalten und werden mit den Headern zusammengeführt, die sich aus dem Bindungsausdruck ergeben.

Sie können eine Nachrichtenbindung definieren, wenn Sie eine Pipeline in der Google Cloud Console oder mit der gcloud CLI konfigurieren.

Console

  1. Rufen Sie in der Google Cloud Console die Seite Eventarc > Pipelines auf.

    Zu Pipelines

  2. Sie können eine Pipeline erstellen oder, wenn Sie eine Pipeline aktualisieren, auf den Namen der Pipeline klicken.

    Das Aktualisieren einer Pipeline kann länger als 10 Minuten dauern.

  3. Klicken Sie auf der Seite Pipelines-Details auf Bearbeiten.

  4. Wenden Sie im Bereich Ziel eine Nachrichtenbindung an. Das ist ein CEL-Ausdruck, der in JSON geschrieben ist. Dadurch wird eine neu erstellte HTTP-Anfrage an das Ziel der Pipeline gesendet.

    Weitere Informationen finden Sie in diesem Dokument unter Auf eingehende Nachrichten zugreifen und HTTP-Anfragen erstellen.

  5. Klicken Sie auf Speichern.

gcloud

  1. Öffnen Sie ein Terminalfenster.

  2. Sie können eine Pipeline erstellen oder eine Pipeline mit dem gcloud eventarc pipelines update Befehl aktualisieren:

    gcloud eventarc pipelines update PIPELINE_NAME \
    --location=REGION \
    --destinations=http_endpoint_message_binding_template='MESSAGE_BINDING'

    Ersetzen Sie Folgendes:

    • PIPELINE_NAME: die ID der Pipeline oder ein voll qualifizierter Name

    • REGION: ein unterstützter Eventarc Advanced-Standort

      Alternativ können Sie das Standortattribut der gcloud CLI festlegen:

      gcloud config set eventarc/location REGION

    • MESSAGE_BINDING: ein in JSON geschriebener CEL-Ausdruck, der zu einer neu erstellten HTTP-Anfrage führt, die dann an das Ziel der Pipeline gesendet wird.

      Weitere Informationen finden Sie in diesem Dokument unter Auf eingehende Nachrichten zugreifen und HTTP-Anfragen erstellen.

    Beispiel:

    gcloud eventarc pipelines create my-pipeline \
    --location=us-central1 \
    --destinations=http_endpoint_uri='https://example-endpoint.com', \
http_endpoint_message_binding_template='{"headers":{"new-header-key": "new-header-value"}}'

    Wenn Sie einen http_endpoint_message_binding_template-Schlüssel verwenden, müssen Sie auch den http_endpoint_uri-Schlüssel festlegen.

Auf eingehende Nachrichten zugreifen

Sie können mit einem CEL-Ausdruck auf eine eingehende CloudEvents-Nachricht zugreifen:

  • Verwenden Sie den Wert message.data, um auf das Feld data der eingehenden Nachricht zuzugreifen.
  • Verwenden Sie die message.key Werte (wobei key der Name des Attributs ist), um auf die Attribute der eingehenden Nachricht zuzugreifen.

  • Verwenden Sie eine headers-Variable, um auf alle Header zuzugreifen, die der HTTP-Anfrage durch vorherige Mediationen in der Verarbeitungskette hinzugefügt wurden. Diese Variable definiert eine Zuordnung von Schlüssel/Wert-Paaren, die den zusätzlichen HTTP-Headern und nicht den ursprünglichen Headern der ersten eingehenden Anfrage entsprechen.

    Mit dem folgenden CEL-Ausdruck kann beispielsweise eine HTTP-Anfrage nur mit Headern erstellt werden, indem den Headern, die in vorherigen Pipeline-Mediationen hinzugefügt wurden, ein weiterer Header hinzugefügt wird:

    {"headers": headers.merge({"new-header-key": "new-header-value"})}

HTTP-Anfragen erstellen

Das Ergebnis des CEL-Ausdrucks muss eine Zuordnung von Schlüssel/Wert-Paaren sein, deren Felder headers und body verwendet werden, um die HTTP-Anfrage wie folgt zu erstellen.

Für headers-Felder:

  • Wenn als Ergebnis des CEL-Ausdrucks eine headers Zuordnung vorhanden ist, werden die Schlüssel/Wert Paare direkt den HTTP-Anfrageheadern zugeordnet und die Werte werden mit der kanonischen Stringcodierung des entsprechenden Datentyps erstellt.
  • Wenn kein headers-Feld vorhanden ist, behält die resultierende HTTP-Anfrage alle Header bei, die während vorheriger Pipeline-Mediationen hinzugefügt wurden.

Für body-Felder:

  • Wenn als Ergebnis des CEL-Ausdrucks ein body-Feld vorhanden ist, wird sein Wert direkt dem Text der HTTP-Anfrage zugeordnet.
  • Wenn der Wert des Felds body vom Typ bytes oder string ist, wird er unverändert als Text der HTTP-Anfrage verwendet. Andernfalls wird er in einen JSON-String konvertiert.
  • Wenn das Feld body nicht vorhanden ist, ist der Text der resultierenden HTTP-Anfrage der Text der endgültigen CloudEvents-HTTP-Nachrichtenbindung im Binärinhaltsmodus.

Alle anderen Felder, die sich aus dem CEL-Ausdruck ergeben, werden ignoriert.

Erweiterungsfunktionen

Eventarc Advanced unterstützt die folgenden Erweiterungsfunktionen, mit denen die Ereignisdaten transformiert werden können, wenn Sie eine Nachrichtenbindung angeben.

Funktion Beschreibung
merge

Führt eine übergebene CEL-Zuordnung in die CEL-Zuordnung ein, auf die die Funktion angewendet wird. Wenn derselbe Schlüssel in beiden Zuordnungen vorhanden ist oder der Wert des Schlüssels vom Typ map ist, werden beide Zuordnungen zusammengeführt. Andernfalls wird der Wert aus der übergebenen Zuordnung verwendet.

Beispiel: map1.merge(map2) -> map3
toBase64

Konvertiert einen CEL-Wert in einen base64-URL-codierten String.

Beispiel: map.toBase64() -> string
toCloudEventJsonWithPayloadFormat

Konvertiert eine Nachricht in eine CEL-Zuordnung, die einer JSON Darstellung einer CloudEvents-Nachricht entspricht, und wendet toDestinationPayloadFormat auf die Nachrichtendaten an. Legt außerdem das datacontenttype des Ereignisses auf das angegebene Ausgangsformat (output_payload_format_*) fest. Wenn kein Ausgangsformat festgelegt ist, wird ein vorhandenes datacontenttype verwendet. Andernfalls wird das datacontenttype nicht festgelegt. Wenn die Nachricht nicht der CloudEvents-Spezifikation entspricht, schlägt die Funktion fehl. Hinweis: Um die Daten in einen JSON-String zu konvertieren, können Sie toJsonString verwenden.

Beispiel: message.toCloudEventJsonWithPayloadFormat() -> map.toJsonString() -> string
toDestinationPayloadFormat

Konvertiert message.data in das angegebene Ausgangsformat (output_payload_format_*). Wenn kein Ausgangsformat festgelegt ist, message.data wird unverändert zurückgegeben.

Beispiel: message.data.toDestinationPayloadFormat() -> string or bytes
toJsonString

Konvertiert einen CEL-Wert in einen JSON-String.

Beispiel: map.toJsonString() -> string
toMap

Konvertiert eine CEL-Liste von CEL-Zuordnungen in eine einzelne CEL-Zuordnung.

Beispiel: list(map).toMap() -> map

Beispiel: Header beibehalten, neuen Header hinzufügen, Text auf Zielformat festlegen

gcloud eventarc pipelines create my-pipeline \
    --location=us-central1 \
    --input-payload-format-json='{}' \
    --destinations=http_endpoint_uri='https://example-endpoint.com',http_endpoint_message_binding_template='{"headers": headers.merge({"content-type":"application/avro"}), "body": message.data.toDestinationPayloadFormat()}',output_payload_format_avro_schema_definition='{"schema_definition": "{"type":"record","name":"myrecord","fields":[{"name":"name","type":"string"},{"name":"account_late","type":"boolean"}]}"}'

Nächste Schritte