Creare una sottoscrizione con SMT

Questo documento spiega come creare un abbonamento Pub/Sub con le trasformazioni di un singolo messaggio (SMT).

Le SMT degli abbonamenti consentono di apportare modifiche leggere ai dati e agli attributi dei messaggi direttamente in Pub/Sub. Questa funzionalità consente di pulire, filtrare o convertire il formato dei dati prima che i messaggi vengano consegnati a un client sottoscrittore.

Per creare un abbonamento con le SMT, puoi utilizzare la Google Cloud console, Google Cloud CLI, la libreria client o l'API Pub/Sub.

Prima di iniziare

• Scopri di più sul servizio Pub/Sub e sulla sua terminologia.

• Scopri di più sulle SMT.

Ruoli e autorizzazioni richiesti

Per ottenere le autorizzazioni necessarie per creare un abbonamento con le SMT, chiedi all'amministratore di concederti il ruolo IAM Pub/Sub Editor (roles/pubsub.editor) nel progetto. Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Questo ruolo predefinito contiene le autorizzazioni necessarie per creare un abbonamento con le SMT. Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione Autorizzazioni obbligatorie:

Autorizzazioni obbligatorie

Per creare un abbonamento con le SMT sono necessarie le seguenti autorizzazioni:

• Concedi l'autorizzazione per creare un abbonamento nel progetto: pubsub.subscriptions.create

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

A seconda del tipo di abbonamento, potrebbero essere necessarie autorizzazioni aggiuntive. Per trovare l'elenco esatto delle autorizzazioni, consulta il documento che descrive la creazione dell'abbonamento specifico. Ad esempio, se stai creando un abbonamento BigQuery con le SMT, consulta Creare abbonamenti BigQuery.

Se crei un abbonamento in un progetto diverso da quello dell'argomento, devi concedere il ruolo roles/pubsub.subscriber al principal del progetto contenente l'abbonamento nel progetto contenente l'argomento.

Puoi configurare il controllo dell'accesso a livello di progetto e a livello di singola risorsa.

Creare un abbonamento con le SMT

Prima di creare un abbonamento con le SMT, consulta la documentazione relativa alle proprietà di un abbonamento.

Nota: ti consigliamo vivamente di configurare un argomento messaggi non recapitabili per gli abbonamenti con le SMT. I messaggi inoltrati all'argomento messaggi non recapitabili a causa di errori SMT contengono un attributo CloudPubSubDeadLetterSourceSMTErrorMessage che indica il motivo dell'errore SMT.

Per creare un abbonamento Pub/Sub con una o più SMT, segui questi passaggi. Puoi abilitare fino a 5 SMT per abbonamento.

Console

1. Nella Google Cloud console, vai alla pagina Pub/Sub Abbonamenti.

   Vai ad Abbonamenti

2. Fai clic su Crea abbonamento.

3. Nel campo ID abbonamento, inserisci un ID per l'abbonamento. Per saperne di più sulla denominazione degli abbonamenti, consulta le linee guida per la denominazione.

4. In Trasformazioni, fai clic su Aggiungi una trasformazione.

5. Seleziona il Tipo di trasformazione. Per saperne di più sui tipi di SMT supportati, consulta Tipi di SMT.

6. Imposta le proprietà di configurazione per la SMT. L'insieme di proprietà dipende dal tipo di SMT. Per saperne di più, consulta la documentazione relativa al tipo di SMT.

7. (Facoltativo) Per convalidare la SMT, fai clic su Convalida. Se la SMT è valida, viene visualizzato il messaggio "Validation passed". In caso contrario, viene visualizzato un messaggio di errore.

8. Per aggiungere un'altra trasformazione, fai clic su Aggiungi una trasformazione e ripeti i passaggi precedenti.

   Per disporre le SMT in un ordine specifico, fai clic su Sposta in alto o Sposta in basso. Per rimuovere una SMT, fai clic su Elimina.

9. (Facoltativo) Per testare una SMT su un messaggio di esempio:

   1. Fai clic su Testa trasformazioni.

   2. Nella finestra Testa trasformazione, seleziona la funzione che vuoi testare.

   3. Nella finestra Messaggio di input, inserisci un messaggio di esempio.

   4. Per aggiungere un attributo al messaggio, fai clic su Aggiungi un attributo e inserisci la chiave e il valore dell'attributo. Puoi aggiungere più attributi.

   5. Fai clic su Test. Il risultato dell'applicazione della SMT al messaggio viene visualizzato in Messaggio di output.

   6. Per chiudere la finestra Testa trasformazioni, fai clic su Chiudi.

   Se crei più di una SMT, puoi testare l'intera sequenza di trasformazioni nel seguente modo:

   1. Testa la prima SMT nella sequenza, come descritto nei passaggi precedenti.
   2. Seleziona la SMT successiva. Il messaggio di input viene precompilato con il messaggio di output del test precedente.
   3. Continua a testare le SMT in ordine per assicurarti che l'intera sequenza funzioni come previsto.

10. Fai clic su Crea per creare l'abbonamento.

gcloud

1. Nella Google Cloud console, attiva Cloud Shell.

   Attiva Cloud Shell

   Nella parte inferiore della Google Cloud console viene avviata una sessione di Cloud Shell e viene visualizzato un prompt della riga di comando. Cloud Shell è un ambiente shell con Google Cloud CLI già inclusa e installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.

2. Crea un file YAML o JSON che definisca una o più SMT. La definizione YAML o JSON dipende dal tipo di SMT. Per saperne di più, consulta Tipi di SMT.

   Se il file include più di una SMT, Pub/Sub le esegue nell'ordine in cui sono elencate.

3. (Facoltativo) Per convalidare una SMT, esegui il gcloud pubsub message-transforms validate comando:

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

   Sostituisci quanto segue:

   • TRANSFORM_FILE: il percorso di un file YAML o JSON che definisce una singola SMT. Se stai creando più SMT, devi convalidarle singolarmente.

4. (Facoltativo) Per testare una o più SMT su un messaggio Pub/Sub di esempio, esegui il gcloud pubsub message-transforms test comando:

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

   Sostituisci quanto segue:

   • TRANSFORMS_FILE: il percorso di un file YAML o JSON che definisce una o più SMT.
   • MESSAGE: il corpo del messaggio di esempio.
   • ATTRIBUTES: (facoltativo) Un elenco di attributi dei messaggi separati da virgole. Ogni attributo è una coppia chiave-valore formattata come KEY="VALUE".

   Il comando esegue le SMT in ordine, utilizzando l'output di ogni SMT come input per la successiva. Il comando restituisce i risultati di ogni passaggio.

5. Per creare l'abbonamento, esegui il gcloud pubsub subscriptions create comando:

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

   Sostituisci quanto segue:

   • SUBSCRIPTION_ID: l'ID o il nome dell' abbonamento che vuoi creare. Per le linee guida su come denominare un abbonamento, consulta Nomi delle risorse. Il nome di un abbonamento è immutabile.

   • PROJECT_ID: l'ID del progetto che contiene l'argomento.

   • TOPIC_ID: l'ID dell'argomento a cui abbonarsi.

   • TRANSFORMS_FILE: il percorso del file YAML o JSON che definisce una o più SMT.

C#

Prima di provare questo esempio, segui le istruzioni di configurazione di C# in Guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Pub/Sub C# .

using Google.Cloud.PubSub.V1;
using System;

public class CreateSubscriptionWithSingleMessageTransformSample
{
    public Subscription CreateSubscriptionWithSingleMessageTransform(string projectId, string topicId, string subscriptionId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);

        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);
        MessageTransform removeSSNTransform = new MessageTransform
        {
            JavascriptUdf = new JavaScriptUDF
            {
                FunctionName = "redactSsn",
                Code = "function redactSsn(message, metadata) {"
                + "   const data = JSON.parse(message.data);"
                + "   delete data['ssn'];"
                + "   message.data = JSON.stringify(data);"
                + "   return message;"
                + "}"
            },
        };

        var subscription = subscriber.CreateSubscription(new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            MessageTransforms = { removeSSNTransform },
        });
        Console.WriteLine($"Subscription {subscription.Name} created with SMT.");

        return subscription;
    }
}

Java

Prima di provare questo esempio, segui le istruzioni di configurazione di Java in Guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Pub/Sub 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

Prima di provare questo esempio, segui le istruzioni di configurazione di Python in Guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Pub/Sub 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}")

Vai

L'esempio seguente utilizza la versione principale della libreria client Go Pub/Sub (v2). Se utilizzi ancora la libreria v1, consulta la guida alla migrazione alla v2. Per visualizzare un elenco di esempi di codice della versione 1, consulta gli esempi di codice deprecati.

Prima di provare questo esempio, segui le istruzioni di configurazione di Go in Guida rapida all'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Pub/Sub 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
}

Come le SMT interagiscono con altre funzionalità di abbonamento

Tieni presente i seguenti punti quando utilizzi una SMT di abbonamento.

Filtri

Se l'abbonamento utilizza sia le SMT sia i filtri integrati di Pub/Sub , il filtro viene applicato prima della SMT. Ciò ha le seguenti implicazioni:

• Se la SMT modifica gli attributi del messaggio, il filtro Pub/Sub non viene applicato al nuovo insieme di attributi.
• La SMT non viene applicata ai messaggi filtrati dal filtro Pub/Sub.
• Se la SMT filtra i messaggi, tieni presente l'impatto sul monitoraggio del backlog dell'abbonamento.
• Se colleghi un abbonamento a una pipeline Dataflow, non utilizzare una SMT di abbonamento per filtrare i messaggi, perché interrompe lo scalabilità automatica di Dataflow.

Ordinamento messaggi

Se definisci una SMT su un abbonamento con l'ordinamento abilitato e l'esecuzione della SMT genera un errore, i messaggi successivi per la stessa chiave di ordinamento non vengono consegnati al sottoscrittore. Per evitare questo problema, configura un argomento messaggi non recapitabili nell'abbonamento per rimuovere il messaggio non elaborato dal backlog dei messaggi.

Passaggi successivi

• Risolvere i problemi relativi alle SMT degli abbonamenti

• Aggiornare le SMT degli abbonamenti

• Eliminare le SMT degli abbonamenti