Créer un abonnement avec des SMT

Ce document explique comment créer un abonnement Pub/Sub avec des transformations de message unique (SMT).

Les SMT d'abonnement permettent de modifier légèrement les données et les attributs des messages directement dans Pub/Sub. Cette fonctionnalité permet de nettoyer, de filtrer ou de convertir le format des données avant que les messages ne soient distribués à un client abonné.

Pour créer un abonnement avec des SMT, vous pouvez utiliser la Google Cloud console, la Google Cloud CLI, la bibliothèque cliente ou l'API Pub/Sub.

Avant de commencer

• Découvrez le service Pub/Sub et sa terminologie.

• Découvrez les SMT.

Rôles et autorisations requis

Pour obtenir les autorisations nécessaires pour créer un abonnement avec des SMT, demandez à votre administrateur de vous accorder le rôle IAM Éditeur Pub/Sub (roles/pubsub.editor) sur votre projet. Pour en savoir plus sur l'attribution de rôles, consultez Gérer l'accès aux projets, aux dossiers et aux organisations.

Ce rôle prédéfini contient les autorisations requises pour créer un abonnement avec des SMT. Pour connaître les autorisations exactes requises, développez la section Autorisations requises :

Autorisations requises

Les autorisations suivantes sont requises pour créer un abonnement avec des SMT :

• Accordez l'autorisation de créer un abonnement sur le projet : pubsub.subscriptions.create

Vous pouvez également obtenir ces autorisations avec des rôles personnalisés ou d'autres rôles prédéfinis.

Selon le type d'abonnement, vous aurez peut-être besoin d'autorisations supplémentaires. Pour connaître la liste exacte des autorisations, consultez le document qui explique comment créer l'abonnement spécifique. Par exemple, si vous créez un abonnement BigQuery avec des SMT, consultez Créer des abonnements BigQuery.

Si vous créez un abonnement dans un projet différent de celui du sujet, vous devez accorder le rôle roles/pubsub.subscriber au principal du projet contenant l'abonnement dans le projet contenant le sujet.

Vous pouvez configurer le contrôle des accès au niveau du projet et au niveau de ressources individuelles.

Créer un abonnement avec des SMT

Avant de créer un abonnement avec des SMT, consultez la documentation sur les propriétés d'un abonnement.

Remarque : Il est fortement recommandé de configurer une file d'attente de lettres mortes pour les abonnements avec des SMT. Les messages transférés vers le file d'attente de lettres mortes en raison d'échecs de SMT contiennent un attribut CloudPubSubDeadLetterSourceSMTErrorMessage qui indique la raison de l'échec du SMT.

Pour créer un abonnement Pub/Sub avec une ou plusieurs SMT, procédez comme suit. Vous pouvez activer jusqu'à cinq SMT par abonnement.

Console

1. Dans la Google Cloud console, accédez à la page Pub/Sub Abonnements.

   Accéder à "Abonnements"

2. Cliquez sur Créer un abonnement.

3. Dans le champ ID d'abonnement, saisissez un ID pour votre abonnement. Pour en savoir plus sur l'attribution de noms aux abonnements, consultez les consignes d'attribution de noms.

4. Sous Transformations, cliquez sur Ajouter une transformation.

5. Sélectionnez le Type de transformation. Pour en savoir plus sur les types de SMT compatibles, consultez Types de SMT.

6. Définissez les propriétés de configuration du SMT. L'ensemble de propriétés dépend du type de SMT. Pour en savoir plus, consultez la documentation de ce type de SMT.

7. Facultatif. Pour valider le SMT, cliquez sur Valider. Si le SMT est valide, le message "Validation passed" s'affiche. Sinon, un message d'erreur s'affiche.

8. Pour ajouter une autre transformation, cliquez sur Ajouter une transformation et répétez les étapes précédentes.

   Pour organiser les SMT dans un ordre spécifique, cliquez sur Déplacer vers le haut ou Déplacer vers le bas. Pour supprimer un SMT, cliquez sur Supprimer.

9. Facultatif. Pour tester un SMT sur un exemple de message, procédez comme suit :

   1. Cliquez sur Tester les transformations.

   2. Dans la fenêtre Tester la transformation, sélectionnez la fonction que vous souhaitez tester.

   3. Dans la fenêtre Message d'entrée, saisissez un exemple de message.

   4. Pour ajouter un attribut au message, cliquez sur Ajouter un attribut , puis saisissez la clé et la valeur de l'attribut. Vous pouvez ajouter plusieurs attributs.

   5. Cliquez sur Test. Le résultat de l'application du SMT au message s'affiche sous Message de sortie.

   6. Pour fermer la fenêtre Tester les transformations, cliquez sur Fermer.

   Si vous créez plusieurs SMT, vous pouvez tester l'ensemble de la séquence de transformations comme suit :

   1. Testez le premier SMT de la séquence, comme décrit dans les étapes précédentes.
   2. Sélectionnez le SMT suivant. Le message d'entrée est prérempli avec le message de sortie du test précédent.
   3. Continuez à tester les SMT dans l'ordre pour vous assurer que l'ensemble de la séquence fonctionne comme prévu.

10. Cliquez sur Créer pour créer l'abonnement.

gcloud

1. Dans la Google Cloud console, activez Cloud Shell.

   Activer Cloud Shell

   En bas de la Google Cloud console, une session Cloud Shell démarre et affiche une invite de ligne de commande. Cloud Shell est un environnement de shell dans lequel Google Cloud CLI est déjà installé, et dans lequel des valeurs sont déjà définies pour votre projet actuel. L'initialisation de la session peut prendre quelques secondes.

2. Créez un fichier YAML ou JSON qui définit une ou plusieurs SMT. La définition YAML ou JSON dépend du type de SMT. Pour en savoir plus, consultez Types de SMT.

   Si le fichier inclut plusieurs SMT, Pub/Sub les exécute dans l'ordre indiqué.

3. Facultatif. Pour valider un SMT, exécutez la gcloud pubsub message-transforms validate commande :

   gcloud pubsub message-transforms validate \
     --message-transform-file=TRANSFORM_FILE
   

   Remplacez les éléments suivants :

   • TRANSFORM_FILE : chemin d'accès à un fichier YAML ou JSON qui définit un seul SMT. Si vous créez plusieurs SMT, vous devez les valider individuellement.

4. Facultatif. Pour tester une ou plusieurs SMT sur un exemple de message Pub/Sub message, exécutez la gcloud pubsub message-transforms test commande :

   gcloud pubsub message-transforms test \
     --message-transforms-file=TRANSFORMS_FILE \
     --message=MESSAGE \
     --attribute=ATTRIBUTES
   

   Remplacez les éléments suivants :

   • TRANSFORMS_FILE : chemin d'accès à un fichier YAML ou JSON qui définit une ou plusieurs SMT.
   • MESSAGE : corps de l'exemple de message.
   • ATTRIBUTES : facultatif. Liste d'attributs de message séparés par une virgule. Chaque attribut est une paire clé/valeur au format KEY="VALUE".

   La commande exécute les SMT dans l'ordre, en utilisant la sortie de chaque SMT comme entrée pour le suivant. La commande génère les résultats de chaque étape.

5. Pour créer l'abonnement, exécutez la gcloud pubsub subscriptions create commande :

   gcloud pubsub subscriptions create SUBSCRIPTION_ID \
       --topic=projects/PROJECT_ID/topics/TOPIC_ID \
       --message-transforms-file=TRANSFORMS_FILE
   

   Remplacez les éléments suivants :

   • SUBSCRIPTION_ID : ID ou nom de l' abonnement que vous souhaitez créer. Pour obtenir des consignes sur la façon de nommer un abonnement, consultez Noms de ressources. Le nom d'un abonnement est immuable.

   • PROJECT_ID : ID du projet contenant le sujet.

   • TOPIC_ID : ID du sujet auquel s'abonner.

   • TRANSFORMS_FILE : chemin d'accès au fichier YAML ou JSON qui définit une ou plusieurs SMT.

Java

Avant d'essayer cet exemple, suivez les instructions d'installation dans le langage Java qui se trouvent sur la page Démarrage rapide : utiliser des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Pub/Sub pour Java .

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.pubsub.v1.JavaScriptUDF;
import com.google.pubsub.v1.MessageTransform;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.Subscription;
import java.io.IOException;

public class CreateSubscriptionWithSmtExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    String subscriptionId = "your-subscription-id";

    createSubscriptionWithSmtExample(projectId, topicId, subscriptionId);
  }

  public static void createSubscriptionWithSmtExample(
      String projectId, String topicId, String subscriptionId) throws IOException {

    // UDF that removes the 'ssn' field, if present
    String code =
        "function redactSSN(message, metadata) {"
            + "  const data = JSON.parse(message.data);"
            + "  delete data['ssn'];"
            + "  message.data = JSON.stringify(data);"
            + "  return message;"
            + "}";
    String functionName = "redactSSN";

    JavaScriptUDF udf =
        JavaScriptUDF.newBuilder().setCode(code).setFunctionName(functionName).build();
    MessageTransform transform = MessageTransform.newBuilder().setJavascriptUdf(udf).build();

    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

      ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
      ProjectSubscriptionName subscriptionName =
          ProjectSubscriptionName.of(projectId, subscriptionId);

      Subscription subscription =
          subscriptionAdminClient.createSubscription(
              Subscription.newBuilder()
                  .setName(subscriptionName.toString())
                  .setTopic(topicName.toString())
                  // Add the UDF message transform
                  .addMessageTransforms(transform)
                  .build());

      System.out.println("Created subscription with SMT: " + subscription.getAllFields());
    }
  }
}

Python

Avant d'essayer cet exemple, suivez les instructions d'installation dans le langage Python qui se trouvent sur la page Démarrage rapide : utiliser des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Pub/Sub pour Python.

from google.cloud import pubsub_v1
from google.pubsub_v1.types import JavaScriptUDF, MessageTransform

# TODO(developer): Choose an existing topic.
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# subscription_id = "your-subscription-id"

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)

code = """function redactSSN(message, metadata) {
            const data = JSON.parse(message.data);
            delete data['ssn'];
            message.data = JSON.stringify(data);
            return message;
            }"""
udf = JavaScriptUDF(code=code, function_name="redactSSN")
transforms = [MessageTransform(javascript_udf=udf)]

with subscriber:
    subscription = subscriber.create_subscription(
        request={
            "name": subscription_path,
            "topic": topic_path,
            "message_transforms": transforms,
        }
    )
    print(f"Created subscription with SMT: {subscription}")

Go

L'exemple suivant utilise la version majeure de la bibliothèque cliente Go Pub/Sub (v2). Si vous utilisez toujours la bibliothèque v1, consultez le guide de migration vers la v2. Pour afficher la liste des exemples de code v1, consultez les exemples de code obsolètes.

Avant d'essayer cet exemple, suivez les instructions d'installation dans le langage Go qui se trouvent sur la page Démarrage rapide : utiliser des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Pub/Sub pour Go.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
)

// createSubscriptionWithSMT creates a subscription with a single message transform function applied.
func createSubscriptionWithSMT(w io.Writer, projectID, topicID, subID string) error {
	// projectID := "my-project-id"
	// topicID := "my-topic"
	// subID := "my-sub"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	code := `function redactSSN(message, metadata) {
			const data = JSON.parse(message.data);
			delete data['ssn'];
			message.data = JSON.stringify(data);
			return message;
		}`

	transform := &pubsubpb.MessageTransform{
		Transform: &pubsubpb.MessageTransform_JavascriptUdf{
			JavascriptUdf: &pubsubpb.JavaScriptUDF{
				FunctionName: "redactSSN",
				Code:         code,
			},
		},
	}

	sub := &pubsubpb.Subscription{
		Name:              fmt.Sprintf("projects/%s/subscriptions/%s", projectID, subID),
		Topic:             fmt.Sprintf("projects/%s/topics/%s", projectID, topicID),
		MessageTransforms: []*pubsubpb.MessageTransform{transform},
	}
	sub, err = client.SubscriptionAdminClient.CreateSubscription(ctx, sub)
	if err != nil {
		return fmt.Errorf("CreateSubscription: %w", err)
	}
	fmt.Fprintf(w, "Created subscription with message transform: %v\n", sub)
	return nil
}

Interaction des SMT avec d'autres fonctionnalités d'abonnement

Tenez compte des points suivants lorsque vous utilisez un SMT d'abonnement.

Filtrage

Si votre abonnement utilise à la fois des SMT et les filtres intégrés de Pub/Sub, le filtre est appliqué avant le SMT. Cela a les implications suivantes :

• Si votre SMT modifie les attributs du message, le filtre Pub/Sub n'est pas appliqué au nouvel ensemble d'attributs.
• Votre SMT n'est pas appliqué aux messages qui sont filtrés par le filtre Pub/Sub.
• Si votre SMT filtre les messages, tenez compte de l'impact sur la surveillance de votre backlog d'abonnement.
• Si vous connectez un abonnement à un pipeline Dataflow, n'utilisez pas de SMT d'abonnement pour filtrer les messages, car cela perturbe l'autoscaling de Dataflow.

Tri des messages

Si vous définissez un SMT sur un abonnement pour lequel le tri est activé et que l'exécution du SMT génère une erreur, les messages suivants pour la même clé de tri ne sont pas distribués à l'abonné. Pour éviter ce problème, configurez une file d'attente de lettres mortes sur l'abonnement afin de supprimer le message non traité du backlog de messages.

Étape suivante

• Résoudre les problèmes liés aux SMT d'abonnement

• Mettre à jour les SMT d'abonnement

• Supprimer les SMT d'abonnement