Utilizzare il tracciamento distribuito OpenTelemetry in Dataflow

Nelle pipeline di streaming distribuite, può essere difficile identificare quale fase specifica, chiamata di servizio esterno o shuffle dei worker causa picchi di latenza utilizzando solo le metriche aggregate. Utilizzando la tracciabilità distribuita di OpenTelemetry in Dataflow, puoi monitorare i singoli elementi end-to-end tra le trasformazioni della pipeline, i confini di rete e i servizi integrati con un unico ID traccia. Questa visibilità ti aiuta a individuare i colli di bottiglia delle prestazioni, ottimizzare i costi di esecuzione della pipeline e risolvere i problemi relativi ai complessi carichi di lavoro di streaming.

Utilizza le seguenti informazioni per attivare, configurare e utilizzare la tracciabilità distribuita OpenTelemetry nelle pipeline Apache Beam.

Casi d'uso

Il tracciamento distribuito in Dataflow è particolarmente utile nei seguenti scenari:

  • Carichi di lavoro di streaming:monitora il tempo di elaborazione per elemento e identifica le fasi che introducono picchi di ritardo o latenza.
  • Tracciamento delle chiamate di servizi esterni nel codice utente:misura le durate di esecuzione quando la pipeline chiama database esterni, microservizi o API di AI generativa all'interno del codice utente (ad esempio in un DoFn). Ad esempio, puoi misurare il tempo per le chiamate a database esterni (come Spanner o Bigtable), microservizi (utilizzando HTTP o gRPC) o API di AI generativa (come Gemini). Per saperne di più, consulta Scrivere tracce personalizzate in DoFn e Eseguire l'integrazione con le librerie client.
  • Ottimizzazione di costi e prestazioni:mostra le durate di operazioni specifiche nelle istanze PTransform con carico reale per identificare serializzazione inefficiente, query lente o contesa dei thread di lavoro.
  • Pipeline complesse di agenti e workflow: ottieni visibilità sulle pipeline con più rami, cicli iterativi di agenti e unioni per determinare quale trasformazione o ramo richiede più tempo.

Limitazioni

Il tracciamento distribuito in Dataflow presenta le seguenti limitazioni:

  • Questa funzionalità è progettata e verificata solo per le pipeline di streaming.
  • Per utilizzare questa funzionalità, devi utilizzare l'SDK Beam per Java (versione 2.76.0 o successive).
  • La propagazione del contesto tra le fasi richiede Streaming Java Runner (precedentemente chiamato Runner v1). Portable Runner (precedentemente chiamato Runner v2) reimposta il contesto della traccia ai limiti di rimescolamento.
  • La tracciabilità OpenTelemetry integrata nei connettori Beam I/O è supportata solo per i seguenti connettori: Pub/Sub, Apache Kafka e flussi di modifiche in tempo reale Spanner. Per Apache Kafka, il tracing è supportato solo quando utilizzi direttamente il connettore KafkaIO. Non è supportato quando utilizzi Managed I/O per Kafka. Altri connettori I/O (ad esempio BigtableIO) non hanno la tracciabilità integrata dei connettori. Per tracciare le operazioni con questi servizi, strumenta le chiamate all'interno del codice utente (ad esempio in un DoFn).
  • Le operazioni che combinano più elementi in uno non propagano un singolo contesto di traccia. Sono incluse operazioni come GroupByKey, CoGroupByKey e Combine.
  • Il contesto non viene propagato tra l'impostazione di un timer (Timer.set()) e l'esecuzione del metodo di callback @OnTimer.

Prerequisiti

Per utilizzare la tracciabilità distribuita in Dataflow, assicurati che il tuo ambiente soddisfi i seguenti requisiti:

  1. Il tuo lavoro deve soddisfare i seguenti requisiti:

  2. Se esporti le tracce in Cloud Trace, abilita l'API Telemetry ( telemetry.googleapis.com) nel tuo progetto Google Cloud .

    Ruoli richiesti per abilitare le API

    Per abilitare le API, devi disporre dell'autorizzazione serviceusage.services.enable. Se hai creato il progetto, probabilmente disponi già di questa autorizzazione tramite il ruolo Proprietario (roles/owner). In caso contrario, puoi ottenere questa autorizzazione tramite il ruolo Amministratore utilizzo servizi (roles/serviceusage.serviceUsageAdmin). Scopri come concedere i ruoli.

    Abilitare l'API

  3. Il service account worker deve disporre del ruolo Agente Cloud Trace (roles/cloudtrace.agent), che include l'autorizzazione cloudtrace.traces.patch.

Concetti principali

Il tracciamento distribuito si basa sui seguenti concetti di OpenTelemetry:

Traccia
Una rappresentazione di una transazione o di un flusso di lavoro completo mentre scorre attraverso un sistema distribuito. Una traccia è composta da uno o più span.
Intervallo
Il componente di base di una traccia, che rappresenta una singola unità di lavoro. Ad esempio, l'esecuzione di una trasformazione, una chiamata al metodo o una richiesta RPC.
Contesto
Lo stato propagato tra i thread e i limiti delle API. Sono inclusi l'ID traccia e l'ID intervallo.
Propagatore
Il meccanismo utilizzato per inserire ed estrarre la rappresentazione del contesto attraverso i confini della rete. Ad esempio, il contesto della traccia W3C.
Esportatore
Un componente che invia i dati span raccolti a un backend di tracciamento, ad esempio Cloud Trace o un agente di raccolta OpenTelemetry autogestito.
Sampler
Un meccanismo che controlla il volume delle tracce determinando quali tracce vengono registrate ed esportate. Ad esempio, un tasso di campionamento dell'1%.

Abilita il tracciamento OpenTelemetry

Per abilitare il tracciamento OpenTelemetry con le impostazioni predefinite, configura le opzioni della pipeline e gli esperimenti richiesti quando invii il job. L'impostazione predefinita esporta in Cloud Trace con un campionamento dell'1%.

Per modificare i valori predefiniti, consulta Impostare e modificare le proprietà di tracciamento.

  1. Configura le opzioni della pipeline e gli esperimenti quando esegui il job.

    Job di template non flessibile

    Trasmetti i seguenti flag di esperimento quando invii un job:

    --experiments=enable_otel_defaults,element_metadata_supported,disable_portable_worker

    Job con modello flessibile

    Passa gli esperimenti utilizzando il flag --additional-experiments nel comando gcloud:

    gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://BUCKET_NAME/templates/TEMPLATE_NAME.json \
    --additional-experiments="enable_otel_defaults,element_metadata_supported,disable_portable_worker" \
    --region=REGION

    Sostituisci quanto segue:

    • JOB_NAME: il nome del job Dataflow.
    • BUCKET_NAME: Il bucket Cloud Storage contenente il modello.
    • TEMPLATE_NAME: il nome del file Flex Template.
    • REGION: la regione Google Cloud in cui vuoi eseguire il job.

    Questi esperimenti configurano i seguenti comportamenti:

    • enable_otel_defaults: configura i valori predefiniti standard di OpenTelemetry, inclusi l'esportatore Cloud Trace e il campionamento all'1%.
    • element_metadata_supported: consente la serializzazione e la propagazione dei metadati (come il contesto della traccia) nelle fasi della pipeline.
    • disable_portable_worker: forza l'utilizzo di Dataflow Streaming Java Runner (precedentemente chiamato Runner v1).

  2. Se utilizzi le versioni 2.76.0 o 2.77.0 dell'SDK Beam per Java, aggiungi le dipendenze OpenTelemetry richieste alla configurazione di compilazione.

    A causa di un problema nelle versioni 2.76.0 e 2.77.0, devi aggiungere le seguenti dipendenze al file pom.xml per esportare le tracce. Per saperne di più, consulta Gestire le dipendenze della pipeline.

    <dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-sdk-extension-autoconfigure</artifactId>
    <scope>compile</scope>
</dependency>

<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-otlp</artifactId>
    <version>1.62.0</version>
    <scope>compile</scope>
</dependency>

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-extensions-opentelemetry-gcp-auth-extension</artifactId>
    <scope>compile</scope>
</dependency>

  3. Attiva esplicitamente la tracciabilità OpenTelemetry sui connettori I/O supportati nel codice della pipeline:

    Pub/Sub

    Per abilitare la tracciabilità durante la lettura o la scrittura in Pub/Sub:

    // Reading from Pub/Sub
pipeline.apply("ReadFromPubSub",
    PubsubIO.readMessages()
        .withEnableOpenTelemetryTracing()
        .fromSubscription("projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID"));

// Writing to Pub/Sub
pipeline.apply("WriteToPubSub",
    PubsubIO.writeStrings()
        .withEnableOpenTelemetryTracing()
        .to("projects/PROJECT_ID/topics/TOPIC_ID"));

    Sostituisci quanto segue:

    • PROJECT_ID: l'ID del tuo Google Cloud progetto.
    • SUBSCRIPTION_ID: il nome della sottoscrizione Pub/Sub.
    • TOPIC_ID: il nome dell'argomento Pub/Sub.

    Dettagli della propagazione del contesto:

    • Publisher upstream:quando pubblichi messaggi da un'applicazione esterna, inserisci il contesto di traccia W3C negli attributi del messaggio Pub/Sub. Per ulteriori informazioni, consulta Pub/Sub OpenTelemetry tracing.
    • Attributi:Beam prevede e compila i seguenti attributi:
      • googclient_traceparent: contiene l'intestazione traceparent di W3C.
      • googclient_tracestate: contiene l'intestazione W3C tracestate.
    • Per la formattazione dell'intestazione, consulta le specifiche del contesto di traccia W3C.

    Apache Kafka

    Per abilitare la tracciabilità durante la lettura o la scrittura in Apache Kafka:

    // Reading from Kafka
pipeline.apply("ReadFromKafka",
    KafkaIO.<String, String>read()
        .withEnableOpenTelemetryTracing()
        // ... additional configuration
        );

// Writing to Kafka
pipeline.apply("WriteToKafka",
    KafkaIO.<String, String>write()
        .withEnableOpenTelemetryTracing()
        // ... additional configuration
        );

    Dettagli della propagazione del contesto:

    • KafkaIO utilizza le intestazioni dei record Kafka standard per propagare il contesto.
    • Inserisce le intestazioni conformi alla specifica W3C Trace Context:
      • traceparent: identifica il contesto della traccia in entrata.
      • tracestate: Fornisce metadati di routing e filtraggio specifici del fornitore aggiuntivi.

    Spanner

    Per abilitare la tracciabilità durante la lettura dai flussi di modifiche in tempo reale di Spanner:

    pipeline.apply("ReadChangeStream",
    SpannerIO.readChangeStream()
        .withSpannerConfig(spannerConfig)
        .withEnableOpenTelemetryTracing(true)
        .withChangeStreamName("CHANGE_STREAM_NAME"));

    Sostituisci quanto segue:

    • CHANGE_STREAM_NAME: il nome della modifica in tempo reale Spanner.

    Dettagli sulla creazione dell'intervallo:

    • L'attivazione della tracciabilità avvia una nuova traccia e uno nuovo span per ogni record di mutazione del database letto dal lettore Change Data Capture (CDC).
    • I cicli di polling del lettore Spanner che non restituiscono nuovi record potrebbero anche produrre intervalli di traccia che possono essere filtrati nell'interfaccia utente di Cloud Trace.

Impostare e modificare le proprietà di tracciamento

Quando specifichi enable_otel_defaults, Dataflow applica le seguenti proprietà predefinite:

Proprietà Valore predefinito Descrizione
otel.traces.exporter otlp Esporta le tracce utilizzando OpenTelemetry Protocol (OTLP).
otel.exporter.otlp.endpoint https://telemetry.googleapis.com Ha come target il ricevitore OTLP di Cloud Trace.
google.cloud.project Il tuo ID progetto Il progetto Google Cloud in cui sono archiviate le tracce.
otel.traces.sampler.arg 0.01 (1%) Registra 1 traccia ogni 100.
otel.service.name options.getAppName() Il nome del servizio associato agli intervalli, derivato dall'opzione appName della pipeline.

Per personalizzare la configurazione di OpenTelemetry, trasmetti un elenco di proprietà separate da punto e virgola all'opzione della pipeline --openTelemetryProperties.

L'esempio seguente configura esplicitamente tutte le proprietà predefinite:

--openTelemetryProperties="otel.traces.exporter=otlp;otel.exporter.otlp.endpoint=https://telemetry.googleapis.com;google.cloud.project=PROJECT_ID;otel.traces.sampler.arg=0.01;otel.service.name=CUSTOM_SERVICE_NAME;otel.java.global-autoconfigure.enabled=true"

Impostare un nome servizio personalizzato

Per impostazione predefinita, il nome del servizio di traccia viene derivato dall'opzione appName della pipeline (--appName=YourPipelineName). Puoi sostituire questo valore impostando la proprietà otel.service.name:

--openTelemetryProperties="otel.service.name=CUSTOM_SERVICE_NAME"

Modificare la frequenza di campionamento e le proprietà

L'impostazione di una frequenza di campionamento più elevata fornisce informazioni di tracciamento più affidabili. Tuttavia, frequenze di campionamento elevate (ad esempio 1.0 o always_on) aumentano notevolmente il volume di dati di traccia inviati a Cloud Trace, il che può aumentare i costi di Google Cloud . Utilizza frequenze di campionamento elevate solo per il debug a breve termine e riducile per i carichi di lavoro di produzione.

L'esempio seguente configura esplicitamente la frequenza di campionamento. In questo caso devi fornire un elenco completo di argomenti:

--openTelemetryProperties="otel.traces.exporter=otlp;otel.exporter.otlp.endpoint=https://telemetry.googleapis.com;google.cloud.project=PROJECT_ID;otel.traces.sampler.arg=0.02;otel.service.name=CUSTOM_SERVICE_NAME;otel.java.global-autoconfigure.enabled=true"

Esportare nei backend di terze parti

Per inviare tracce a un backend di tracciamento gestito esterno o a un raccoglitore OpenTelemetry autogestito anziché a Cloud Trace:

  1. Ometti il seguente esperimento: --experiments=enable_otel_defaults.

  2. Configura manualmente le proprietà dell'esportatore utilizzando --openTelemetryProperties:

    --openTelemetryProperties="otel.traces.exporter=otlp;otel.exporter.otlp.endpoint=http://COLLECTOR_HOST:4317;otel.traces.sampler=parentbased_always_on"

Estensione di autenticazioneGoogle Cloud

Il runtime Beam include un'estensione di Google Cloud autenticazione che autentica automaticamente le richieste OpenTelemetry Protocol (OTLP) inviate alle API di Google come telemetry.googleapis.com.

  • Quando utilizzi le impostazioni predefinite, l'autenticazione Google Cloud viene attivata automaticamente.
  • Quando esporti in endpoint non Google:se enable_otel_defaults viene omesso, Beam disattiva l'estensione di Google Cloud autenticazione impostando la proprietà di sistema google.otel.auth.target.signals=none.
  • Configurazione manuale:puoi impostare la proprietà di sistema google.otel.auth.target.signals su none (disattivata) o trace (attivata solo per le tracce).

Scrivere tracce personalizzate in DoFn

Puoi creare intervalli personalizzati all'interno del codice utente, ad esempio all'interno di un DoFn, per misurare operazioni specifiche. Sono incluse chiamate ad API esterne o logica a elevata intensità di calcolo.

Esempio di implementazione di DoFn

import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.Scope;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.SdkHarnessOptions;
import org.apache.beam.sdk.transforms.DoFn;

public class InvestigateTransactionDoFn extends DoFn<Transaction, RiskAssessment> {
  private transient Tracer tracer;

  @Setup
  public void setup(PipelineOptions options) {
    // Obtain the Tracer instance from SdkHarnessOptions
    this.tracer = options
        .as(SdkHarnessOptions.class)
        .getOpenTelemetry()
        .getTracer("org.apache.beam.examples.adk.aml");
  }

  @ProcessElement
  public void processElement(@Element Transaction tx, OutputReceiver<RiskAssessment> out) {
    // Create and start a custom span as a child of the active context
    Span span = tracer.spanBuilder("InvestigateTransactionDoFn:Analyze")
        .setAttribute("transaction.id", tx.getId())
        .setAttribute("transaction.amount", tx.getAmount())
        .startSpan();

    // Make the span current in the execution thread
    try (Scope scope = span.makeCurrent()) {
      // Perform your logic (such as calling an external service)
      RiskAssessment assessment = analyzeTransactionWithLLM(tx);
      out.output(assessment);
    } catch (Exception e) {
      // Record exception details and set error status on the span
      span.recordException(e);
      span.setStatus(StatusCode.ERROR, e.getMessage());
      throw e;
    } finally {
      // Always end the span
      span.end();
    }
  }

  private RiskAssessment analyzeTransactionWithLLM(Transaction tx) {
    // Application analysis logic
    return new RiskAssessment();
  }
}

Integrare con le librerie client

Quando un DoFn richiama librerie esterne che supportano anche OpenTelemetry, come le librerie client Google Cloud , configura il client in modo che utilizzi l'istanza OpenTelemetry della pipeline, in modo che gli intervalli secondari si colleghino direttamente alla traccia della pipeline.

Esempio di client Java Spanner

Per eseguire query Spanner manuali all'interno di un DoFn e collegare questi intervalli alla traccia della pipeline attiva, inizializza SpannerOptions con l'istanza OpenTelemetry della pipeline:

import com.google.cloud.spanner.DatabaseClient;
import com.google.cloud.spanner.DatabaseId;
import com.google.cloud.spanner.ResultSet;
import com.google.cloud.spanner.Spanner;
import com.google.cloud.spanner.SpannerOptions;
import com.google.cloud.spanner.Statement;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.SdkHarnessOptions;
import org.apache.beam.sdk.transforms.DoFn;

public class SpannerEnrichmentDoFn extends DoFn<String, String> {
  private transient Spanner spanner;
  private transient DatabaseClient dbClient;
  private final String instanceId;
  private final String databaseId;

  public SpannerEnrichmentDoFn(String instanceId, String databaseId) {
    this.instanceId = instanceId;
    this.databaseId = databaseId;
  }

  @Setup
  public void setup(PipelineOptions pipelineOptions) {
    // Build SpannerOptions leveraging the pipeline's active OpenTelemetry instance
    SpannerOptions options =
        SpannerOptions.newBuilder()
            .setOpenTelemetry(pipelineOptions
                .as(SdkHarnessOptions.class)
                .getOpenTelemetry())
            .setEnableEndToEndTracing(true)
            .setEnableExtendedTracing(true)
            .build();

    this.spanner = options.getService();
    DatabaseId db = DatabaseId.of(options.getProjectId(), instanceId, databaseId);
    this.dbClient = spanner.getDatabaseClient(db);
  }

  @ProcessElement
  public void processElement(@Element String inputId, OutputReceiver<String> out) {
    Statement statement = Statement.newBuilder(
        "SELECT Details FROM AccountTable WHERE AccountId = @id")
        .bind("id").to(inputId)
        .build();

    // Spanner queries automatically generate child spans linked to the active trace
    try (ResultSet rs = dbClient.singleUse().executeQuery(statement)) {
      while (rs.next()) {
        out.output(rs.getString("Details"));
      }
    }
  }

  @Teardown
  public void tearDown() {
    if (spanner != null) {
      spanner.close();
    }
  }
}

Propagazione del contesto tra le fasi

Nelle pipeline di dati distribuite, il trasferimento del contesto della traccia tra i limiti dei worker e le operazioni di shuffling è essenziale per la visibilità end-to-end.

Quando inserisci Redistribute.arbitrarily() nella pipeline, il contesto di traccia attivo viene serializzato nei metadati dell'elemento, trasmesso attraverso la rete e deserializzato nella VM worker ricevente. In questo modo, la traccia continua senza interruzioni nelle varie fasi.

L'esempio seguente illustra come il contesto della traccia si propaga nelle fasi della pipeline:

  1. Estrai contesto:KafkaIO.read() legge un record in entrata ed estrae il contesto della traccia dalle intestazioni del messaggio.
  2. Elabora e crea intervalli: DoFn A elabora l'elemento all'interno di un intervallo di traccia attivo.
  3. Propagate across workers: Redistribute.arbitrarily() serializes the active trace context into the element metadata and shuffles the element across the network.
  4. Continua la traccia: DoFn B riceve l'elemento su una VM worker, deserializza il contesto della traccia e continua l'elaborazione con lo stesso ID traccia.

Visualizzare le tracce in Cloud Trace

Dopo aver abilitato la tracciabilità, visualizza le tracce della pipeline nella console Google Cloud :

  1. Nella Google Cloud console, vai alla pagina Esplora tracce.

    Vai a Trace tracce

  2. Nella casella del filtro, filtra per Service: SERVICE_NAME. Questo SERVICE_NAME corrisponde alla appName della pipeline o alla otel.service.name configurata.

    In alternativa, cerca un ID traccia specifico.

  3. Seleziona una traccia dall'elenco dei risultati per visualizzarne la sequenza temporale dettagliata.

    Il grafico della cronologia mostra:

    • Runner spans:gli span generati dal runtime di Beam. Ad esempio, PubSubIO.Read e ProcessElement.
    • Span personalizzati:span creati all'interno delle implementazioni di DoFn. Ad esempio, InvestigateTransactionDoFn:Analyze.

    • Span delle librerie client:span secondari generati da librerie client integrate. Ad esempio, query RPC Spanner.

      Visualizzazione delle tracce nella console.

Passaggi successivi