Usa el seguimiento distribuido de OpenTelemetry en Dataflow

En las canalizaciones de transmisión distribuidas, puede ser difícil identificar qué etapa específica, llamada a un servicio externo o combinación de trabajadores causa picos de latencia solo con métricas agregadas. Si usas el registro de seguimiento distribuido de OpenTelemetry en Dataflow, puedes hacer un seguimiento de los elementos individuales de extremo a extremo en las transformaciones de canalizaciones, los límites de red y los servicios integrados con un solo ID de seguimiento. Esta visibilidad te ayuda a identificar cuellos de botella en el rendimiento, optimizar los costos de ejecución de la canalización y solucionar problemas de cargas de trabajo de transmisión complejas.

Usa la siguiente información para habilitar, configurar y usar el seguimiento distribuido de OpenTelemetry en tus canalizaciones de Apache Beam.

Casos de uso

El registro de seguimiento distribuido en Dataflow es particularmente útil en las siguientes situaciones:

  • Cargas de trabajo de transmisión: Realiza un seguimiento del tiempo de procesamiento por elemento y, luego, identifica las etapas que introducen retrasos o picos de latencia.
  • Seguimiento de llamadas a servicios externos en el código del usuario: Mide las duraciones de ejecución cuando tu canalización llama a bases de datos externas, microservicios o APIs de IA generativa dentro del código del usuario (por ejemplo, en un DoFn). Por ejemplo, puedes medir el tiempo de las llamadas a bases de datos externas (como Spanner o Bigtable), microservicios (con HTTP o gRPC) o APIs de IA generativa (como Gemini). Para obtener más información, consulta Cómo escribir registros personalizados en DoFns y Cómo realizar la integración con bibliotecas cliente.
  • Optimización del costo y el rendimiento: Expone las duraciones de operaciones específicas en instancias de PTransform bajo carga del mundo real para identificar la serialización ineficiente, las consultas lentas o la contención de subprocesos de trabajadores.
  • Canalizaciones complejas de agentes y flujos de trabajo: Obtén visibilidad de las canalizaciones con varias ramas, bucles de agentes iterativos y uniones para determinar qué transformación o rama tarda más.

Limitaciones

El seguimiento distribuido en Dataflow tiene las siguientes limitaciones:

  • Esta función se diseñó y verificó solo para las canalizaciones de transmisión.
  • Para usar esta función, debes usar el SDK de Beam para Java (versión 2.76.0 o posterior).
  • La propagación del contexto entre etapas requiere el Streaming Java Runner (anteriormente llamado Runner v1). El ejecutor portátil (anteriormente llamado Runner v2) restablece el contexto de seguimiento en los límites de la aleatorización.
  • El seguimiento de OpenTelemetry integrado en los conectores de E/S de Beam solo se admite para los siguientes conectores: Pub/Sub, Apache Kafka y flujos de cambios de Spanner. En el caso de Apache Kafka, el seguimiento solo se admite cuando usas el conector KafkaIO directamente. No se admite cuando usas E/S administrada para Kafka. Otros conectores de E/S (como BigtableIO) no tienen un registro de seguimiento de conectores integrado. Para hacer un seguimiento de las operaciones con estos servicios, instrumenta tus llamadas dentro del código del usuario (por ejemplo, en un DoFn).
  • Las operaciones que combinan varios elementos en uno no propagan un solo contexto de seguimiento. Esto incluye operaciones como GroupByKey, CoGroupByKey y Combine.
  • El contexto no se propaga entre el establecimiento de un temporizador (Timer.set()) y la ejecución del método de devolución de llamada @OnTimer.

Requisitos previos

Para usar el registro de seguimiento distribuido en Dataflow, asegúrate de que tu entorno cumpla con los siguientes requisitos:

  1. Tu trabajo debe cumplir con los siguientes requisitos:

  2. Si exportas seguimientos a Cloud Trace, habilita la API de Telemetry (telemetry.googleapis.com) en tu proyecto de Google Cloud .

    Roles necesarios para habilitar las APIs

    Para habilitar APIs, necesitas el permiso serviceusage.services.enable. Si creaste el proyecto, es probable que ya tengas este permiso a través del rol de propietario (roles/owner). De lo contrario, puedes obtener este permiso a través del rol de administrador de Service Usage (roles/serviceusage.serviceUsageAdmin). Obtén más información para otorgar roles.

    Habilitar la API

  3. La cuenta de servicio del trabajador debe tener el rol de Agente de Cloud Trace (roles/cloudtrace.agent), que incluye el permiso cloudtrace.traces.patch.

Conceptos básicos

El seguimiento distribuido se basa en los siguientes conceptos de OpenTelemetry:

Seguimiento
Representación de una transacción o flujo de trabajo completo a medida que fluye a través de un sistema distribuido. Un seguimiento consta de uno o más intervalos.
Intervalo
Es el componente básico de un registro, que representa una sola unidad de trabajo. Por ejemplo, una ejecución de transformación, una llamada de método o una solicitud de RPC.
Contexto
Es el estado que se propaga entre subprocesos y límites de la API. Esto incluye el ID de seguimiento y el ID de intervalo.
Propagador
Es el mecanismo que se usa para insertar y extraer la representación del contexto a través de los límites de la red. Por ejemplo, el contexto de seguimiento de W3C.
Exportador
Un componente que envía los datos de intervalo recopilados a un backend de seguimiento, como Cloud Trace o un recopilador de OpenTelemetry autohospedado.
Sampler
Es un mecanismo que controla el volumen de registros determinando qué registros se graban y exportan. Por ejemplo, una tasa de muestreo del 1%.

Habilita el seguimiento de OpenTelemetry

Para habilitar el seguimiento de OpenTelemetry con la configuración predeterminada, configura las opciones de canalización y los experimentos necesarios cuando envíes tu trabajo. El parámetro de configuración predeterminado exporta a Cloud Trace con un muestreo del 1%.

Para modificar los valores predeterminados, consulta Cómo establecer y modificar propiedades de seguimiento.

  1. Configura las opciones y los experimentos de la canalización cuando ejecutes tu trabajo.

    Trabajo sin plantilla de Flex

    Pasa las siguientes marcas de experimentos cuando envíes un trabajo:

    --experiments=enable_otel_defaults,element_metadata_supported,disable_portable_worker

    Trabajo de plantilla de Flex

    Pasa los experimentos con la marca --additional-experiments en el 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

    Reemplaza lo siguiente:

    • JOB_NAME: Es el nombre de tu trabajo de Dataflow.
    • BUCKET_NAME: Es el bucket de Cloud Storage que contiene tu plantilla.
    • TEMPLATE_NAME: Es el nombre de tu archivo de plantilla de Flex.
    • REGION: Es la Google Cloud región en la que deseas ejecutar el trabajo.

    Estos experimentos configuran los siguientes comportamientos:

    • enable_otel_defaults: Configura los valores predeterminados estándar de OpenTelemetry, incluido el exportador de Cloud Trace y el muestreo del 1%.
    • element_metadata_supported: Permite la serialización y la propagación de metadatos (como el contexto de seguimiento) en las etapas de la canalización.
    • disable_portable_worker: Fuerza el uso del Runner de Java de transmisión de Dataflow (anteriormente llamado Runner v1).

  2. Si usas las versiones 2.76.0 o 2.77.0 del SDK de Beam para Java, agrega las dependencias de OpenTelemetry requeridas a tu configuración de compilación.

    Debido a un problema en las versiones 2.76.0 y 2.77.0, debes agregar las siguientes dependencias a tu archivo pom.xml para que se exporten los registros. Para obtener más información, consulta Administra dependencias de canalizaciones.

    <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. Habilita de forma explícita el seguimiento de OpenTelemetry en los conectores de E/S compatibles en el código de tu canalización:

    Pub/Sub

    Para habilitar el registro de seguimiento cuando lees o escribes en Pub/Sub, haz lo siguiente:

    // 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"));

    Reemplaza lo siguiente:

    • PROJECT_ID: Es el ID de tu proyecto de Google Cloud .
    • SUBSCRIPTION_ID: Es el nombre de tu suscripción a Pub/Sub.
    • TOPIC_ID: Es el nombre de tu tema de Pub/Sub.

    Detalles de la propagación del contexto:

    • Publicador upstream: Cuando publiques mensajes desde una aplicación externa, inserta el contexto de seguimiento de W3C en los atributos del mensaje de Pub/Sub. Para obtener más información, consulta Registro de OpenTelemetry de Pub/Sub.
    • Atributos: Beam espera y completa los siguientes atributos:
      • googclient_traceparent: Lleva el encabezado traceparent de W3C.
      • googclient_tracestate: Lleva el encabezado tracestate de W3C.
    • Para obtener información sobre el formato de los encabezados, consulta la especificación de contexto de seguimiento de W3C.

    Apache Kafka

    Para habilitar el registro de seguimiento cuando se lee o escribe en Apache Kafka, haz lo siguiente:

    // 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
        );

    Detalles de la propagación del contexto:

    • KafkaIO usa encabezados de registros estándar de Kafka para propagar el contexto.
    • Espera y, luego, inyecta encabezados que cumplen con la especificación de contexto de seguimiento de W3C:
      • traceparent: Identifica el contexto de seguimiento entrante.
      • tracestate: Proporciona metadatos adicionales de filtrado y enrutamiento específicos del proveedor.

    Spanner

    Para habilitar el registro de seguimiento cuando se lee desde flujos de cambios de Spanner, haz lo siguiente:

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

    Reemplaza lo siguiente:

    • CHANGE_STREAM_NAME: Es el nombre del flujo de cambios de Spanner.

    Detalles de la creación del intervalo:

    • Habilitar el registro de seguimiento inicia un nuevo registro de seguimiento y un nuevo intervalo para cada registro de mutación de la base de datos que lee el lector de captura de datos modificados (CDC).
    • Los ciclos de sondeo del lector de Spanner que no devuelven registros nuevos también pueden producir intervalos de seguimiento que se pueden filtrar en la IU de Cloud Trace.

Cómo establecer y modificar propiedades de seguimiento

Cuando especificas enable_otel_defaults, Dataflow aplica las siguientes propiedades predeterminadas:

Propiedad Valor predeterminado Descripción
otel.traces.exporter otlp Exporta seguimientos con el protocolo OpenTelemetry (OTLP).
otel.exporter.otlp.endpoint https://telemetry.googleapis.com Se orienta al receptor de OTLP de Cloud Trace.
google.cloud.project Tu ID del proyecto Es el Google Cloud proyecto en el que se almacenan los registros.
otel.traces.sampler.arg 0.01 (1%) Registra 1 de cada 100 registros.
otel.service.name options.getAppName() Es el nombre del servicio asociado con los intervalos, que se deriva de la opción de canalización appName.

Para personalizar la configuración de OpenTelemetry, pasa una lista de propiedades separadas por punto y coma a la opción de canalización --openTelemetryProperties.

En el siguiente ejemplo, se configuran todas las propiedades predeterminadas de forma explícita:

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

Cómo establecer un nombre de servicio personalizado

De forma predeterminada, el nombre del servicio de seguimiento se deriva de la opción appName de la canalización (--appName=YourPipelineName). Puedes anular este valor configurando la propiedad otel.service.name:

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

Cómo cambiar la tasa y las propiedades de muestreo

Establecer una tasa de muestreo más alta te brinda información de registro más sólida. Sin embargo, las frecuencias de muestreo altas (como 1.0 o always_on) aumentan significativamente el volumen de datos de seguimiento que se envían a Cloud Trace, lo que puede incrementar tus costos de Google Cloud . Usa tasas de muestreo altas solo para la depuración a corto plazo y redúcelas para las cargas de trabajo de producción.

En el siguiente ejemplo, se configura la frecuencia de muestreo de forma explícita. En este caso, debes proporcionar una lista completa de argumentos:

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

Exporta a back-ends de terceros

Para enviar seguimientos a un backend de seguimiento administrado externo o a un recopilador de OpenTelemetry autohospedado en lugar de Cloud Trace, haz lo siguiente:

  1. Omite el siguiente experimento: --experiments=enable_otel_defaults.

  2. Configura las propiedades del exportador de forma manual con --openTelemetryProperties:

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

Extensión de autenticación deGoogle Cloud

El entorno de ejecución de Beam incluye una extensión de autenticación Google Cloud que autentica automáticamente las solicitudes del Protocolo de OpenTelemetry (OTLP) que se envían a las APIs de Google, como telemetry.googleapis.com.

  • Cuando se usan los valores predeterminados: Google Cloud La autenticación se habilita automáticamente.
  • Cuando se exporta a extremos que no son de Google: Si se omite enable_otel_defaults, Beam inhabilita la extensión de autenticación Google Cloud estableciendo la propiedad del sistema google.otel.auth.target.signals=none.
  • Configuración manual: Puedes establecer la propiedad del sistema google.otel.auth.target.signals en none (inhabilitada) o trace (habilitada solo para los registros).

Escribe seguimientos personalizados en DoFns

Puedes crear intervalos personalizados dentro del código del usuario, por ejemplo, dentro de un DoFn, para medir operaciones específicas. Esto incluye llamadas a APIs externas o lógica que requiere mucha capacidad de procesamiento.

Ejemplo de implementación de 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();
  }
}

Integración con bibliotecas cliente

Cuando un DoFn invoca bibliotecas externas que también admiten OpenTelemetry, como las bibliotecas cliente de Google Cloud , configura el cliente para que use la instancia de OpenTelemetry de la canalización, de modo que los intervalos secundarios se vinculen directamente al registro de la canalización.

Ejemplo de cliente de Spanner en Java

Para ejecutar consultas manuales de Spanner dentro de un DoFn y vincular esos intervalos al registro de la canalización activa, inicializa SpannerOptions con la instancia de OpenTelemetry de la canalización:

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();
    }
  }
}

Propagación del contexto en las diferentes etapas

En las canalizaciones de datos distribuidas, es fundamental llevar el contexto de seguimiento a través de los límites de los trabajadores y las operaciones de Shuffle para obtener visibilidad de extremo a extremo.

Cuando insertas Redistribute.arbitrarily() en tu canalización, el contexto de seguimiento activo se serializa en los metadatos del elemento, se transmite a través de la red y se deserializa en la VM de trabajador receptora. Esto garantiza la continuación ininterrumpida del registro en todas las etapas.

En el siguiente ejemplo, se ilustra cómo se propaga el contexto de seguimiento en las etapas de la canalización:

  1. Extraer contexto: KafkaIO.read() lee un registro entrante y extrae el contexto de seguimiento de los encabezados del mensaje.
  2. Procesa y crea intervalos: DoFn A procesa el elemento dentro de un intervalo de seguimiento activo.
  3. Propagación entre trabajadores: Redistribute.arbitrarily() serializa el contexto de seguimiento activo en los metadatos del elemento y lo aleatoriza en la red.
  4. Continuar el seguimiento: DoFn B recibe el elemento en una VM de trabajador, deserializa el contexto de seguimiento y continúa el procesamiento con el mismo ID de seguimiento.

Visualiza seguimientos en Cloud Trace

Después de habilitar el registro de seguimiento, consulta los registros de seguimiento de la canalización en la consola de Google Cloud :

  1. En la Google Cloud consola, ve a la página Explorador de seguimiento.

    Ir al Explorador de seguimiento

  2. En el cuadro de filtro, filtra por Service: SERVICE_NAME. Este SERVICE_NAME coincide con tu canalización appName o con el otel.service.name configurado.

    Como alternativa, busca un ID de seguimiento específico.

  3. Selecciona un registro de la lista de resultados para ver su cronograma detallado.

    En el gráfico de la línea de tiempo, se muestra lo siguiente:

    • Intervalos del ejecutor: Son los intervalos que genera el entorno de ejecución de Beam. Por ejemplo, PubSubIO.Read y ProcessElement.
    • Intervalos personalizados: Son los intervalos creados dentro de tus implementaciones de DoFn. Por ejemplo: InvestigateTransactionDoFn:Analyze.

    • Intervalos de la biblioteca cliente: Son los intervalos secundarios generados por las bibliotecas cliente integradas. Por ejemplo, las consultas de RPC de Spanner.

      Vista de los registros en la consola.

¿Qué sigue?