Utiliser le traçage distribué OpenTelemetry dans Dataflow

Dans les pipelines de streaming distribués, il peut être difficile d'identifier l'étape spécifique, l'appel de service externe ou le shuffle de worker qui provoque des pics de latence en utilisant uniquement des métriques agrégées. En utilisant le traçage distribué OpenTelemetry dans Dataflow, vous pouvez suivre des éléments individuels de bout en bout dans les transformations de pipeline, les limites de réseau et les services intégrés sous un seul ID de trace. Cette visibilité vous aide à identifier les goulots d'étranglement des performances, à optimiser les coûts d'exécution des pipelines et à résoudre les problèmes liés aux charges de travail de streaming complexes.

Utilisez les informations suivantes pour activer, configurer et utiliser le traçage distribué OpenTelemetry dans vos pipelines Apache Beam.

Cas d'utilisation

Le traçage distribué dans Dataflow est particulièrement utile dans les scénarios suivants :

  • Charges de travail de streaming : suivez le temps de traitement par élément et identifiez les étapes qui introduisent des pics de décalage ou de latence.
  • Suivi des appels de service externes dans le code utilisateur : mesurez les durées d'exécution lorsque votre pipeline appelle des bases de données externes, des microservices ou des API d'IA générative dans le code utilisateur (par exemple, dans un DoFn). Par exemple, vous pouvez mesurer le temps nécessaire aux appels vers des bases de données externes (telles que Spanner ou Bigtable), des microservices (à l'aide de HTTP ou gRPC) ou des API d'IA générative (telles que Gemini). Pour en savoir plus, consultez Écrire des traces personnalisées dans DoFns et Intégrer aux bibliothèques clientes.
  • Optimisation des coûts et des performances : exposez les durées d'opérations spécifiques dans les instances PTransform sous charge réelle pour identifier la sérialisation inefficace, les requêtes lentes ou la contention des threads de worker.
  • Pipelines de workflows et agentiques complexes : obtenez de la visibilité sur les pipelines avec plusieurs branches, les boucles d'agents itératives et les jointures pour déterminer quelle transformation ou branche prend le plus de temps.

Limites

Le traçage distribué dans Dataflow présente les limites suivantes :

  • Cette fonctionnalité est conçue et vérifiée uniquement pour les pipelines de streaming.
  • Pour utiliser cette fonctionnalité, vous devez utiliser le SDK Beam pour Java (version 2.76.0 ou ultérieure).
  • La propagation du contexte inter-étapes nécessite le Streaming Java Runner (anciennement appelé Runner v1). L'exécuteur portable (anciennement appelé Runner v2) réinitialise le contexte de trace aux limites du shuffle.
  • Le traçage OpenTelemetry intégré aux connecteurs d'E/S Beam n'est compatible qu'avec les connecteurs suivants : Pub/Sub, Apache Kafka et flux de modifications Spanner. Pour Apache Kafka, le traçage n'est compatible que lorsque vous utilisez directement le connecteur KafkaIO. Il n'est pas compatible avec Managed I/O pour Kafka. D'autres connecteurs d'E/S (tels que BigtableIO) ne disposent pas de traçage de connecteur intégré. Pour tracer les opérations avec ces services, instrumentez vos appels dans le code utilisateur (par exemple, dans un DoFn).
  • Les opérations qui combinent plusieurs éléments en un seul ne propagent pas un seul contexte de trace. Cela inclut des opérations telles que GroupByKey, CoGroupByKey et Combine.
  • Le contexte n'est pas propagé entre la définition d'un minuteur (Timer.set()) et l'exécution de la méthode de rappel @OnTimer.

Prérequis

Pour utiliser le traçage distribué dans Dataflow, assurez-vous que votre environnement répond aux exigences suivantes :

  1. Votre tâche doit répondre aux exigences suivantes :

  2. Si vous exportez des traces vers Cloud Trace, activez l'API Telemetry (telemetry.googleapis.com) dans votre projet Google Cloud .

    Rôles requis pour activer les API

    Pour activer les API, vous devez disposer de l'autorisation serviceusage.services.enable. Si vous avez créé le projet, vous disposez probablement déjà de cette autorisation grâce au rôle Propriétaire (roles/owner). Sinon, vous pouvez obtenir cette autorisation grâce au rôle Administrateur Service Usage (roles/serviceusage.serviceUsageAdmin). Découvrez comment attribuer des rôles.

    Activer l'API

  3. Le compte de service du nœud de calcul doit disposer du rôle Agent Cloud Trace (roles/cloudtrace.agent), qui inclut l'autorisation cloudtrace.traces.patch.

Concepts fondamentaux

Le traçage distribué repose sur les concepts OpenTelemetry suivants :

Trace
Représentation d'une transaction ou d'un workflow complet dans un système distribué. Une trace se compose d'un ou de plusieurs spans.
Segment
Composant de base d'une trace, représentant une seule unité de travail. Par exemple, une exécution de transformation, un appel de méthode ou une requête RPC.
Contexte
 : état propagé entre les threads et les limites de l'API. Cela inclut l'ID de trace et l'ID de span.
Propagateur
Mécanisme utilisé pour injecter et extraire la représentation du contexte au-delà des limites du réseau. Par exemple, le contexte de trace W3C.
Exportateur
Composant qui envoie les données span collectées à un backend de traçage, tel que Cloud Trace ou un collecteur OpenTelemetry auto-hébergé.
Échantillonneur
Mécanisme qui contrôle le volume des traces en déterminant quelles traces sont enregistrées et exportées. Par exemple, un taux d'échantillonnage de 1 %.

Activer le traçage OpenTelemetry

Pour activer le traçage OpenTelemetry avec les paramètres par défaut, configurez les options de pipeline et les expériences requises lorsque vous envoyez votre job. Le paramètre par défaut exporte vers Cloud Trace avec un échantillonnage de 1 %.

Pour modifier les valeurs par défaut, consultez Définir et modifier les propriétés de traçage.

  1. Configurez les options et les tests de pipeline lorsque vous exécutez votre job.

    Job de modèle non Flex

    Transmettez les indicateurs d'expérience suivants lorsque vous soumettez un job :

    --experiments=enable_otel_defaults,element_metadata_supported,disable_portable_worker

    Job de modèle Flex

    Transmettez les expériences à l'aide de l'option --additional-experiments dans la commande 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

    Remplacez les éléments suivants :

    • JOB_NAME : nom de votre job Dataflow.
    • BUCKET_NAME : bucket Cloud Storage contenant votre modèle.
    • TEMPLATE_NAME : nom de votre fichier de modèle Flex.
    • REGION : Google Cloud région dans laquelle vous souhaitez exécuter le job.

    Ces tests configurent les comportements suivants :

    • enable_otel_defaults : configure les paramètres OpenTelemetry standards par défaut, y compris l'exportateur Cloud Trace et l'échantillonnage à 1 %.
    • element_metadata_supported : permet la sérialisation et la propagation des métadonnées (telles que le contexte de trace) dans les étapes du pipeline.
    • disable_portable_worker : force l'utilisation de l'exécuteur Java de streaming Dataflow (anciennement appelé Runner v1).

  2. Si vous utilisez le SDK Beam pour Java version 2.76.0 ou 2.77.0, ajoutez les dépendances OpenTelemetry requises à votre configuration de compilation.

    En raison d'un problème dans les versions 2.76.0 et 2.77.0, vous devez ajouter les dépendances suivantes à votre fichier pom.xml pour que les traces soient exportées. Pour en savoir plus, consultez Gérer les dépendances de 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. Activez explicitement le traçage OpenTelemetry sur les connecteurs d'E/S compatibles dans le code de votre pipeline :

    Pub/Sub

    Pour activer le traçage lors de la lecture ou de l'écriture dans 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"));

    Remplacez les éléments suivants :

    • PROJECT_ID : ID de votre projet Google Cloud .
    • SUBSCRIPTION_ID : nom de votre abonnement Pub/Sub.
    • TOPIC_ID : nom de votre sujet Pub/Sub.

    Détails de la propagation du contexte :

    • Éditeur en amont : lorsque vous publiez des messages à partir d'une application externe, injectez le contexte de trace W3C dans les attributs de message Pub/Sub. Pour en savoir plus, consultez la section Traçage OpenTelemetry Pub/Sub.
    • Attributs : Beam s'attend aux attributs suivants et les renseigne :
      • googclient_traceparent : contient l'en-tête traceparent du W3C.
      • googclient_tracestate : contient l'en-tête tracestate du W3C.
    • Pour en savoir plus sur le format des en-têtes, consultez la spécification du contexte de trace W3C.

    Apache Kafka

    Pour activer le traçage lors de la lecture ou de l'écriture dans 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
        );

    Détails de la propagation du contexte :

    • KafkaIO utilise des en-têtes d'enregistrement Kafka standards pour propager le contexte.
    • Il attend et injecte des en-têtes conformes à la spécification du contexte de trace W3C :
      • traceparent : identifie le contexte de trace entrant.
      • tracestate : fournit des métadonnées de filtrage et de routage supplémentaires spécifiques au fournisseur.

    Spanner

    Pour activer le traçage lors de la lecture des flux de modifications Spanner :

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

    Remplacez les éléments suivants :

    • CHANGE_STREAM_NAME : nom du flux de modifications Spanner.

    Détails de la création de la portée :

    • L'activation du traçage démarre une nouvelle trace et une nouvelle étendue pour chaque enregistrement de mutation de base de données lu par le lecteur de capture des données modifiées (CDC).
    • Les cycles d'interrogation du lecteur Spanner qui ne renvoient aucun nouvel enregistrement peuvent également générer des segments de trace qui peuvent être filtrés dans l'interface utilisateur Cloud Trace.

Définir et modifier les propriétés de traçage

Lorsque vous spécifiez enable_otel_defaults, Dataflow applique les propriétés par défaut suivantes :

Propriété Valeur par défaut Description
otel.traces.exporter otlp Exporte les traces à l'aide du protocole OpenTelemetry (OTLP).
otel.exporter.otlp.endpoint https://telemetry.googleapis.com Cible le récepteur OTLP Cloud Trace.
google.cloud.project ID de votre projet Le projet Google Cloud dans lequel les traces sont stockées.
otel.traces.sampler.arg 0.01 (1%) Enregistre une trace sur 100.
otel.service.name options.getAppName() Nom du service associé aux spans, dérivé de l'option de pipeline appName.

Pour personnaliser la configuration OpenTelemetry, transmettez une liste de propriétés séparées par des points-virgules à l'option de pipeline --openTelemetryProperties.

L'exemple suivant configure explicitement toutes les propriétés par défaut :

--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"

Définir un nom de service personnalisé

Par défaut, le nom du service de trace est dérivé de l'option appName du pipeline (--appName=YourPipelineName). Vous pouvez remplacer cette valeur en définissant la propriété otel.service.name :

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

Modifier le taux d'échantillonnage et les propriétés

Si vous définissez un taux d'échantillonnage plus élevé, vous obtiendrez des informations de traçage plus fiables. Toutefois, des taux d'échantillonnage élevés (tels que 1.0 ou always_on) augmentent considérablement le volume de données de trace envoyées à Cloud Trace, ce qui peut augmenter vos coûts Google Cloud . N'utilisez des taux d'échantillonnage élevés que pour le débogage à court terme et réduisez-les pour les charges de travail de production.

L'exemple suivant configure explicitement le taux d'échantillonnage. Dans ce cas, vous devez fournir une liste complète d'arguments :

--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"

Exporter vers des backends tiers

Pour envoyer des traces à un backend de traçage géré externe ou à un collecteur OpenTelemetry auto-hébergé au lieu de Cloud Trace :

  1. Omettez le test suivant : --experiments=enable_otel_defaults.

  2. Configurez manuellement les propriétés de l'exportateur à l'aide de --openTelemetryProperties :

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

Extension d'authentificationGoogle Cloud

L'exécution Beam inclut une extension d'authentification Google Cloud qui authentifie automatiquement les requêtes OpenTelemetry Protocol (OTLP) envoyées aux API Google telles que telemetry.googleapis.com.

  • Lorsque vous utilisez les valeurs par défaut Google Cloud , l'authentification est activée automatiquement.
  • Lors de l'exportation vers des points de terminaison non Google : si enable_otel_defaults est omis, Beam désactive l'extension d'authentification Google Cloud en définissant la propriété système google.otel.auth.target.signals=none.
  • Configuration manuelle : vous pouvez définir la propriété système google.otel.auth.target.signals sur none (désactivé) ou trace (activé pour les traces uniquement).

Écrire des traces personnalisées dans les DoFn

Vous pouvez créer des spans personnalisés dans le code utilisateur, par exemple dans un DoFn, pour mesurer des opérations spécifiques. Cela inclut les appels à des API externes ou une logique nécessitant beaucoup de ressources de calcul.

Exemple d'implémentation 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();
  }
}

Intégrer des bibliothèques clientes

Lorsqu'un DoFn appelle des bibliothèques externes compatibles avec OpenTelemetry, telles que les bibliothèques clientes Google Cloud , configurez le client pour qu'il utilise l'instance OpenTelemetry du pipeline. Les spans enfants seront ainsi directement associés à la trace du pipeline.

Exemple de client Java Spanner

Pour exécuter des requêtes Spanner manuelles dans un DoFn et associer ces spans à la trace de pipeline active, initialisez SpannerOptions avec l'instance OpenTelemetry du 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();
    }
  }
}

Propagation du contexte entre les étapes

Dans les pipelines de données distribués, il est essentiel de transmettre le contexte de trace au-delà des limites des nœuds de calcul et des opérations de brassage pour assurer une visibilité de bout en bout.

Lorsque vous insérez Redistribute.arbitrarily() dans votre pipeline, le contexte de trace active est sérialisé dans les métadonnées de l'élément, transmis sur le réseau et désérialisé sur la VM de nœud de calcul destinataire. Cela garantit la continuité des traces entre les étapes.

L'exemple suivant montre comment le contexte de trace se propage dans les étapes du pipeline :

  1. Extraire le contexte : KafkaIO.read() lit un enregistrement entrant et extrait le contexte de trace des en-têtes de message.
  2. Traiter et créer des spans : DoFn A traite l'élément dans un span de trace actif.
  3. Propager entre les nœuds de calcul : Redistribute.arbitrarily() sérialise le contexte de trace actif dans les métadonnées de l'élément et le mélange sur le réseau.
  4. Poursuivre la trace : DoFn B reçoit l'élément sur une VM de nœud de calcul, désérialise le contexte de trace et poursuit le traitement sous le même ID de trace.

Afficher les traces dans Cloud Trace

Une fois le traçage activé, affichez les traces de pipeline dans la console Google Cloud  :

  1. Dans la console Google Cloud , accédez à la page Explorateur Trace.

    Accéder à l'explorateur Trace

  2. Dans la zone de filtre, filtrez par Service: SERVICE_NAME. Ce SERVICE_NAME correspond à votre appName de pipeline ou à votre otel.service.name configuré.

    Vous pouvez également rechercher un ID de trace spécifique.

  3. Sélectionnez une trace dans la liste des résultats pour afficher sa chronologie détaillée.

    Le graphique de la chronologie affiche les éléments suivants :

    • Spans de l'exécuteur : spans générés par l'environnement d'exécution Beam. Par exemple, PubSubIO.Read et ProcessElement.
    • Portées personnalisées : portées créées dans vos implémentations DoFn. Exemple : InvestigateTransactionDoFn:Analyze.

    • Portées de la bibliothèque cliente : portées enfants générées par les bibliothèques clientes intégrées. Par exemple, les requêtes RPC Spanner.

      Affichage des traces dans la console.

Étapes suivantes