Pub/Sub-Nachrichten zu Datenprofilen empfangen und parsen

In diesem Dokument finden Sie Beispiele, die zeigen, wie Sie Benachrichtigungen zu Änderungen an Ihren Datenprofilen empfangen und parsen. Sensitive Data Protection sendet diese Aktualisierungen in Form von Pub/Sub-Nachrichten.

Übersicht

Sie können Sensitive Data Protection so konfigurieren, dass automatisch Profile für Daten in einer Organisation, einem Ordner oder einem Projekt erstellt werden. Datenprofile enthalten Messwerte und Metadaten zu Ihren Daten und können ermitteln, wo sich sensible und risikoreiche Daten befinden. Sensitive Data Protection meldet diese Messwerte auf verschiedenen Detailebenen. Informationen zu den Datentypen, die Sie analysieren können, finden Sie unter Unterstützte Ressourcen.

Beim Konfigurieren des Datenprofilers können Sie die Option aktivieren, Pub/Sub-Nachrichten zu veröffentlichen, wenn sich Ihre Datenprofile erheblich ändern. Die Benachrichtigungen helfen Ihnen, sofort auf diese Änderungen zu reagieren. Die folgenden Ereignisse können beobachtet werden:

  • Für ein Daten-Asset wird zum ersten Mal ein Profil erstellt.
  • Ein Profil wird aktualisiert.
  • Der Risiko- oder Vertraulichkeitsfaktor eines Profils erhöht sich.
  • Es ist ein neuer Fehler im Zusammenhang mit Ihren Datenprofilen aufgetreten.

Die Pub/Sub-Nachrichten, die der Data Profiler veröffentlicht, enthalten ein DataProfilePubSubMessage-Objekt. Diese Nachrichten werden immer im Binärformat gesendet. Sie müssen also Code schreiben, der sie empfängt und parst.

Preise

Wenn Sie Pub/Sub verwenden, wird die Nutzung gemäß den Pub/Sub-Preisen abgerechnet.

Hinweis

Auf dieser Seite wird Folgendes vorausgesetzt:

Führen Sie vor Beginn der Arbeit an den Beispielen die folgenden Schritte aus:

  1. Erstellen Sie ein Pub/Sub-Thema und fügen Sie ein Abo dafür hinzu. Weisen Sie dem Thema kein Schema zu.

    Der Einfachheit halber wird in den Beispielen auf dieser Seite nur auf ein Abo geachtet. In der Praxis können Sie jedoch ein Thema und ein Abo für jedes Ereignis erstellen, das von Sensitive Data Protection unterstützt wird.

  2. Konfigurieren Sie den Data Profiler so, dass Pub/Sub-Nachrichten veröffentlicht werden, falls noch nicht geschehen:

    1. Bearbeiten Sie die Scankonfiguration.

    2. Aktivieren Sie auf der Seite Scankonfiguration bearbeiten die Option In Pub/Sub veröffentlichen und wählen Sie die Ereignisse aus, auf die Sie warten möchten. Konfigurieren Sie dann die Einstellungen für jedes Ereignis.

    3. Speichern Sie die Scankonfiguration.

  3. Gewähren Sie dem Sensitive Data Protection-Dienst-Agent Zugriff auf die Veröffentlichung im Pub/Sub-Thema. Ein Beispiel für eine Rolle mit Veröffentlichungszugriff ist die Rolle „Pub/Sub Publisher“ (roles/pubsub.publisher). Der Dienst-Agent für Sensitive Data Protection ist eine E-Mail-Adresse im folgenden Format:

    service-PROJECT_NUMBER@dlp-api.iam.gserviceaccount.com

    Wenn Sie mit einer Scankonfiguration auf Organisations- oder Ordnerebene arbeiten, ist PROJECT_NUMBER die numerische Kennung des Dienst-Agent-Containers. Wenn Sie mit einer Scankonfiguration auf Projektebene arbeiten, ist PROJECT_NUMBER die numerische Kennung Ihres Projekts.

  4. Installieren und richten Sie die Sensitive Data Protection-Clientbibliothek für Java oder Python ein.

Beispiele

Die folgenden Beispiele zeigen, wie Sie Pub/Sub-Nachrichten empfangen und parsen, die vom Datenprofiler veröffentlicht werden. Sie können diese Beispiele wiederverwenden und als Cloud Run-Funktionen bereitstellen, die durch Pub/Sub-Ereignisse ausgelöst werden. Weitere Informationen finden Sie in der Pub/Sub-Anleitung (2. Generation).

Ersetzen Sie in den folgenden Beispielen Folgendes:

  • PROJECT_ID: die ID des Projekts, das das Pub/Sub-Abo enthält.
  • SUBSCRIPTION_ID: die ID des Pub/Sub-Abos.

Java

import com.google.api.core.ApiService;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.privacy.dlp.v2.DataProfilePubSubMessage;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class DataProfilePubSubMessageParser {

  public static void main(String... args) throws Exception {
    String projectId = "PROJECT_ID";
    String subscriptionId = "SUBSCRIPTION_ID";
    int timeoutSeconds = 5;

    // The `ProjectSubscriptionName.of` method creates a fully qualified identifier
    // in the form `projects/{projectId}/subscriptions/{subscriptionId}`.
    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(projectId, subscriptionId);

    MessageReceiver receiver =
        (PubsubMessage pubsubMessage, AckReplyConsumer consumer) -> {
          try {
            DataProfilePubSubMessage message = DataProfilePubSubMessage.parseFrom(
                pubsubMessage.getData());
            System.out.println(
                "PubsubMessage with ID: " + pubsubMessage.getMessageId()
                    + "; message size: " + pubsubMessage.getData().size()
                    + "; event: " + message.getEvent()
                    + "; profile name: " + message.getProfile().getName()
                    + "; full resource: " + message.getProfile().getFullResource());
            consumer.ack();
          } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
          }
        };

    // Create subscriber client.
    Subscriber subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();
    try {
      ApiService apiService = subscriber.startAsync();
      apiService.awaitRunning();
      System.out.printf("Listening for messages on %s for %d seconds.%n", subscriptionName,
          timeoutSeconds);
      subscriber.awaitTerminated(timeoutSeconds, TimeUnit.SECONDS);
    } catch (TimeoutException ignored) {
    } finally {
      subscriber.stopAsync();
    }
  }
}

Python

from google.cloud import pubsub_v1
from concurrent.futures import TimeoutError
from google.cloud import dlp_v2


project_id = "PROJECT_ID"
subscription_id = "SUBSCRIPTION_ID"
timeout = 5.0

subscriber = pubsub_v1.SubscriberClient()
# The `subscription_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/subscriptions/{subscription_id}`
subscription_path = subscriber.subscription_path(project_id, subscription_id)

def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    print(f"Received {message.data}.")
    dlp_msg = dlp_v2.DataProfilePubSubMessage()
    dlp_msg._pb.ParseFromString(message.data)
    print("Parsed message: ", dlp_msg)
    print("--------")
    message.ack()

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"Listening for messages on {subscription_path} for {timeout} seconds...")

# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
    try:
        # When `timeout` is not set, result() will block indefinitely,
        # unless an exception is encountered first.
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError:
        streaming_pull_future.cancel()  # Trigger the shutdown.
        streaming_pull_future.result()  # Block until the shutdown is complete.
        print("Done waiting.")

Nächste Schritte