יצירת מינוי עם SMT

במאמר הזה נסביר איך ליצור מינוי ל-Pub/Sub באמצעות Single Message Transforms (SMTs).

טרנספורמציות SMT במינויים מאפשרות לבצע שינויים קלים בנתונים ובמאפיינים של ההודעות ישירות ב-Pub/Sub. התכונה הזו מאפשרת לנקות, לסנן או להמיר את הפורמט של הנתונים לפני שההודעות מועברות ללקוח מנוי.

כדי ליצור מינוי עם SMT, אפשר להשתמש במסוף Google Cloud , ב-Google Cloud CLI, בספריית הלקוח או ב-Pub/Sub API.

לפני שמתחילים

• מידע על שירות Pub/Sub והמונחים שלו

• מידע נוסף על SMT

תפקידים והרשאות נדרשים

כדי לקבל את ההרשאות שנדרשות ליצירת מינוי עם SMT, צריך לבקש מהאדמין להקצות לכם בפרויקט את התפקיד עריכה ב-Pub/Sub (roles/pubsub.editor) ב-IAM. כדי לקרוא הסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

זהו תפקיד שמוגדר מראש וכולל את ההרשאות שנדרשות ליצירת מינוי עם SMT. כדי לראות בדיוק אילו הרשאות נדרשות, אפשר להרחיב את הקטע ההרשאות הנדרשות:

ההרשאות הנדרשות

כדי ליצור מינוי עם SMT, צריך את ההרשאות הבאות:

• נותנים את ההרשאה ליצור מינוי בפרויקט: pubsub.subscriptions.create

יכול להיות שתקבלו את ההרשאות האלה באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש אחרים.

יכול להיות שתצטרכו הרשאות נוספות בהתאם לסוג המינוי. כדי לראות את רשימת ההרשאות המדויקת, אפשר לעיין במסמך שמתאר את יצירת המינוי הספציפי. לדוגמה, אם אתם יוצרים מינוי ל-BigQuery עם SMT, כדאי לעיין במאמר בנושא יצירת מינויים ל-BigQuery.

אם יוצרים מינוי בפרויקט אחר מהנושא, צריך להעניק את התפקיד roles/pubsub.subscriber לחשבון הראשי של הפרויקט שמכיל את המינוי בפרויקט שמכיל את הנושא.

אפשר להגדיר בקרת גישה ברמת הפרויקט וברמת המשאב הספציפי.

יצירת מינוי באמצעות SMT

לפני שיוצרים מינוי עם SMT, כדאי לקרוא את המאמר מאפיינים של מינוי.

הערה: מומלץ מאוד להגדיר נושא להודעות ללא מוצא למינויים עם SMT. הודעות שמועברות לנושא להודעות ללא מוצא בגלל כשלים ב-SMTP מכילות מאפיין CloudPubSubDeadLetterSourceSMTErrorMessage עם הסיבה לכשל ב-SMTP.

כדי ליצור מינוי ל-Pub/Sub עם SMT אחד או יותר, מבצעים את השלבים הבאים. אפשר להפעיל עד 5 SMT בכל מינוי.

המסוף

1. נכנסים לדף Subscriptions של Pub/Sub במסוף Google Cloud .

   אל הדף "מינויים"

2. לוחצים על יצירת מינוי.

3. בשדה Subscription ID, מזינים מזהה למינוי. למידע נוסף על מתן שמות למינויים, קראו את ההנחיות למתן שמות.

4. בקטע טרנספורמציות, לוחצים על הוספת טרנספורמציה.

5. בוחרים את סוג הטרנספורמציה. למידע נוסף על סוגי ה-SMT הנתמכים, קראו את המאמר סוגי SMT.

6. קובעים את מאפייני ההגדרה של ה-SMT. קבוצת המאפיינים תלויה בסוג ה-SMT. למידע נוסף, עיינו במסמכי התיעוד של סוג ה-SMT הרלוונטי.

7. זה שינוי אופציונלי. כדי לאמת את ה-SMT, לוחצים על אימות. אם ה-SMT תקין, תוצג ההודעה "Validation passed". אחרת, תוצג הודעת שגיאה.

8. כדי להוסיף עוד טרנספורמציה, לוחצים על הוספת טרנספורמציה וחוזרים על השלבים הקודמים.

   כדי לסדר את ה-SMT בסדר מסוים, לוחצים על העברה למעלה או על העברה למטה. כדי להסיר SMT, לוחצים על מחיקה.

9. זה שינוי אופציונלי. כדי לבדוק SMT בהודעה לדוגמה:

   1. לוחצים על בדיקת טרנספורמציות.

   2. בחלון Test transform, בוחרים את הפונקציה שרוצים לבדוק.

   3. בחלון הזנת הודעה, מזינים הודעה לדוגמה.

   4. כדי להוסיף מאפיין להודעה, לוחצים על Add an attribute ומזינים את המפתח והערך של המאפיין. אפשר להוסיף כמה מאפיינים.

   5. לוחצים על בדיקה. התוצאה של החלת ה-SMT על ההודעה מוצגת בקטע הודעת הפלט.

   6. כדי לסגור את החלון Test transforms, לוחצים על Close.

   אם יוצרים יותר מ-SMT אחד, אפשר לבדוק את כל רצף השינויים באופן הבא:

   1. בודקים את ה-SMT הראשון ברצף, כמו שמתואר בשלבים הקודמים.
   2. בוחרים את ה-SMT הבא. הודעת הקלט מאוכלסת מראש עם הודעת הפלט מהבדיקה הקודמת.
   3. ממשיכים לבדוק את כללי ה-SMT לפי הסדר, כדי לוודא שכל הרצף פועל כמו שצריך.

10. לוחצים על יצירה כדי ליצור את המינוי.

gcloud

1. מפעילים את Cloud Shell במסוף Google Cloud .

   הפעלת Cloud Shell

   בחלק התחתון של Google Cloud המסוף יתחיל סשן של Cloud Shell ותופיע הודעה של שורת הפקודה. Cloud Shell היא סביבת מעטפת שבה ה-CLI של Google Cloud מותקן ומוגדרים ערכים לפרויקט הקיים. הסשן יופעל תוך כמה שניות.

2. יוצרים קובץ YAML או JSON שמגדיר SMT אחד או יותר. ההגדרה של YAML או JSON תלויה בסוג ה-SMT. למידע נוסף, קראו את המאמר סוגי SMT.

   אם הקובץ כולל יותר מ-SMT אחד, ‏ Pub/Sub מבצע אותם לפי הסדר שבו הם מופיעים.

3. זה שינוי אופציונלי. כדי לאמת SMT, מריצים את הפקודה gcloud pubsub message-transforms validate:

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

   מחליפים את מה שכתוב בשדות הבאים:

   • ‫TRANSFORM_FILE: הנתיב לקובץ YAML או JSON שמגדיר SMT יחיד. אם יוצרים כמה SMT, צריך לאמת כל אחד מהם בנפרד.

4. זה שינוי אופציונלי. כדי לבדוק SMT אחד או יותר בהודעת Pub/Sub לדוגמה, מריצים את הפקודה gcloud pubsub message-transforms test:

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

   מחליפים את מה שכתוב בשדות הבאים:

   • ‫TRANSFORMS_FILE: הנתיב לקובץ YAML או JSON שמגדיר SMT אחד או יותר.
   • MESSAGE: גוף ההודעה לדוגמה.
   • ‫ATTRIBUTES: אופציונלי. רשימה מופרדת בפסיקים של מאפייני הודעות. כל מאפיין הוא צמד מפתח/ערך בפורמט KEY="VALUE".

   הפקודה מריצה את ה-SMT לפי הסדר, ומשתמשת בפלט של כל SMT כקלט ל-SMT הבא. הפקודה מציגה את התוצאות של כל שלב.

5. כדי ליצור את המינוי, מריצים את הפקודה gcloud pubsub subscriptions create:

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

   מחליפים את מה שכתוב בשדות הבאים:

   • ‫SUBSCRIPTION_ID: המזהה או השם של המינוי שרוצים ליצור. הנחיות למתן שם למינוי זמינות במאמר שמות משאבים. אי אפשר לשנות את השם של מינוי.

   • ‫PROJECT_ID: מזהה הפרויקט שמכיל את הנושא.

   • ‫TOPIC_ID: המזהה של הנושא להרשמה.

   • ‫TRANSFORMS_FILE: הנתיב לקובץ YAML או JSON שמגדיר SMT אחד או יותר.

C#‎

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של C# ‎ במאמר מדריך למתחילים: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub C# API.

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

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במאמר תחילת העבודה המהירה: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Java API.

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

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Python במאמר התחלה מהירה: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של ה-API בשפת Python של Pub/Sub.

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 Pub/Sub ‏ (v2). אם אתם עדיין משתמשים בספרייה v1, כדאי לעיין במדריך להעברת נתונים ל-v2. דוגמאות קוד שהוצאו משימוש

לפני שתנסו את הדוגמה הזו, פעלו לפי הוראות ההגדרה של Go במאמר התחלה מהירה: שימוש בספריות לקוח. מידע נוסף מופיע במסמכי העזר של ה-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
}

איך מינויים חודשיים משתלבים עם תכונות אחרות שקשורות למינוי

כשמשתמשים ב-SMT של מינוי, חשוב להביא בחשבון את הנקודות הבאות.

סינון

אם המינוי שלכם משתמש גם ב-SMT וגם במסננים המובנים של Pub/Sub, המסנן יוחל לפני ה-SMT. להגדרה הזו יש את ההשלכות הבאות:

• אם ה-SMT משנה את מאפייני ההודעה, המסנן של Pub/Sub לא חל על קבוצת המאפיינים החדשה.
• ה-SMT לא חל על הודעות שמסוננות על ידי המסנן של Pub/Sub.
• אם ה-SMT מסנן הודעות, חשוב לשים לב להשפעה על מעקב אחרי צבר ההודעות במינוי.
• אם אתם מקשרים מינוי לפייפליין Dataflow, אל תשתמשו במינוי SMT כדי לסנן הודעות, כי זה משבש את ההתאמה האוטומטית לעומס של Dataflow.

סדר ההודעות

אם מגדירים SMT במינוי שמופעלת בו הגדרת סדר, והפעלת ה-SMT גורמת לשגיאה, ההודעות הבאות עם אותו מפתח סדר לא נשלחות למנוי. כדי למנוע את הבעיה הזו, צריך להגדיר נושא של הודעות שלא ניתן למסור במינוי כדי להסיר את ההודעה שלא עברה עיבוד מפיגור ההודעות.

המאמרים הבאים

• פתרון בעיות במינויים ל-SMT

• עדכון של SMT למינוי

• מחיקת SMT של מינוי