主页
产品
Streams for Apache Kafka
2.9
Deploying and Managing Streams for Apache Kafka on OpenShift
Chapter 26. Introducing distributed tracing

此内容没有您所选择的语言版本。

Chapter 26. Introducing distributed tracing

Distributed tracing allows you to track the progress of transactions between applications in a distributed system. In a microservices architecture, tracing tracks the progress of transactions between services. Trace data is useful for monitoring application performance and investigating issues with target systems and end-user applications.

In Streams for Apache Kafka, distributed tracing facilitates end-to-end tracking of messages: from source systems to Kafka, and then from Kafka to target systems and applications. This complements the monitoring of metrics in Grafana dashboards and component loggers.

Streams for Apache Kafka provides built-in support for tracing for the following Kafka components:

MirrorMaker to trace messages from a source cluster to a target cluster
Kafka Connect to trace messages consumed and produced by Kafka Connect
Kafka Bridge to trace messages between Kafka and HTTP client applications

Tracing is not supported for Kafka brokers.

26.1. Tracing options
复制链接

Distributed traces consist of spans, which represent individual units of work performed over a specific time period. When instrumented with tracers, applications generate traces that follow requests as they move through the system, making it easier to identify delays or issues.

OpenTelemetry, a telemetry framework, provides APIs for tracing that are independent of any specific backend tracing system. In Streams for Apache Kafka, the default protocol for transmitting traces between Kafka components and tracing systems is OpenTelemetry’s OTLP (OpenTelemetry Protocol), a vendor-neutral protocol.

While OTLP is the default, Streams for Apache Kafka also supports other tracing systems, such as Jaeger. Jaeger is a distributed tracing system designed for monitoring microservices, and its user interface allows you to query, filter, and analyze trace data in detail.

The Jaeger user interface showing a simple query

The Jaeger user interface showing a simple query

26.2. Environment variables for tracing
复制链接

Use environment variables to enable tracing for Kafka components or to initialize a tracer for Kafka clients.

Tracing environment variables are subject to change. For the latest information, see the OpenTelemetry documentation.

The following table describes the key environment variables for setting up tracing with OpenTelemetry.

Expand

Table 26.1. OpenTelemetry environment variables
Property	Required	Description
`OTEL_SERVICE_NAME`	Yes	The name of the tracing service for OpenTelemetry, such as OTLP or Jaeger.
`OTEL_EXPORTER_OTLP_ENDPOINT`	Yes (if using OTLP exporter)	The OTLP endpoint for exporting trace data to the tracing system. For Jaeger tracing, specify the `OTEL_EXPORTER_JAEGER_ENDPOINT`. For other tracing systems, specify the appropriate endpoint.
`OTEL_TRACES_EXPORTER`	No (unless using a non-OTLP exporter)	The exporter used for tracing. The default is `otlp`, which does not need to be specified. For Jaeger tracing, set this variable to `jaeger`. For other tracing systems, specify the appropriate exporter.
`OTEL_EXPORTER_OTLP_CERTIFICATE`	No (required if using TLS with OTLP)	The path to the file containing trusted certificates for TLS authentication. Required to secure communication between Kafka components and the OpenTelemetry endpoint when using TLS with the `otlp` exporter.

26.3. Setting up distributed tracing
复制链接

Enable distributed tracing in Kafka components by specifying a tracing type in the custom resource. Instrument tracers in Kafka clients for end-to-end tracking of messages.

To set up distributed tracing, follow these procedures in order:

Enable tracing for supported Kafka components
Initialize a tracer for Kafka clients
Instrument clients with tracers, embedding telemetry-gathering functionality into the code:
- Instrument producers and consumers for tracing
- Instrument Kafka Streams applications for tracing

26.3.1. Prerequisites
复制链接

Before setting up distributed tracing, make sure backend components are deployed to your OpenShift cluster.

We recommend using the Jaeger operator for deploying Jaeger on your OpenShift cluster. For deployment instructions, see the Jaeger documentation.

Note

Setting up tracing systems is outside the scope of this content.

26.3.2. Enabling tracing in supported Kafka components
复制链接

Distributed tracing is supported for MirrorMaker, MirrorMaker 2, Kafka Connect, and the Kafka Bridge. Enable tracing using OpenTelemetry by setting the spec.tracing.type property to opentelemetry. Configure the custom resource of the component to specify and enable a tracing system using spec.template properties.

By default, OpenTelemetry uses the OTLP (OpenTelemetry Protocol) exporter and endpoint to gather trace data. This procedure shows the configuration to use OTLP as the tracing system.

If you prefer to use a different tracing system supported by OpenTelemetry, such as Jaeger, you can modify the exporter and endpoint settings in the tracing configuration.

Important

Streams for Apache Kafka no longer supports OpenTracing. If you were previously using OpenTracing with the type: jaeger option, we encourage you to transition to using OpenTelemetry instead.

Enabling tracing in a resource triggers the following events:

Interceptor classes are updated in the integrated consumers and producers of the component.
For MirrorMaker, MirrorMaker 2, and Kafka Connect, the tracing agent initializes a tracer based on the tracing configuration defined in the resource.
For the Kafka Bridge, a tracer based on the tracing configuration defined in the resource is initialized by the Kafka Bridge itself.

Tracing in MirrorMaker and MirrorMaker 2

For MirrorMaker and MirrorMaker 2, messages are traced from the source cluster to the target cluster. The trace data records messages entering and leaving the MirrorMaker or MirrorMaker 2 component.

Tracing in Kafka Connect

For Kafka Connect, only messages produced and consumed by Kafka Connect are traced. To trace messages sent between Kafka Connect and external systems, you must configure tracing in the connectors for those systems.

Tracing in the Kafka Bridge

For the Kafka Bridge, messages produced and consumed by the Kafka Bridge are traced. Incoming HTTP requests from client applications to send and receive messages through the Kafka Bridge are also traced. To have end-to-end tracing, you must configure tracing in your HTTP clients.

Procedure

Perform these steps for each KafkaMirrorMaker, KafkaMirrorMaker2, KafkaConnect, and KafkaBridge resource.

In the spec.template property, configure the tracer service.

Use the tracing environment variables as template configuration properties.
For OpenTelemetry, set the spec.tracing.type property to opentelemetry.

Example tracing configuration for Kafka Connect using OpenTelemetry

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
spec:
  #...
  template:
    connectContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
  tracing:
    type: opentelemetry
  #...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
spec:
  #...
  template:
    connectContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
  tracing:
    type: opentelemetry
  #...

Copy to Clipboard

Toggle word wrap

Example tracing configuration for MirrorMaker using OpenTelemetry

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker
metadata:
  name: my-mirror-maker
spec:
  #...
  template:
    mirrorMakerContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
  tracing:
    type: opentelemetry
#...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker
metadata:
  name: my-mirror-maker
spec:
  #...
  template:
    mirrorMakerContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
  tracing:
    type: opentelemetry
#...

Copy to Clipboard

Toggle word wrap

Example tracing configuration for MirrorMaker 2 using OpenTelemetry

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mm2-cluster
spec:
  #...
  template:
    connectContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
  tracing:
    type: opentelemetry
#...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mm2-cluster
spec:
  #...
  template:
    connectContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
  tracing:
    type: opentelemetry
#...

Copy to Clipboard

Toggle word wrap

Example tracing configuration for the Kafka Bridge using OpenTelemetry

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaBridge
metadata:
  name: my-bridge
spec:
  #...
  template:
    bridgeContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
  tracing:
    type: opentelemetry
#...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaBridge
metadata:
  name: my-bridge
spec:
  #...
  template:
    bridgeContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
  tracing:
    type: opentelemetry
#...

Copy to Clipboard

Toggle word wrap

(Optional) If TLS authentication is configured on the OTLP endpoint, use the OTEL_EXPORTER_OTLP_CERTIFICATE environment variable to specify the path to a trusted certificate This secures communication between Kafka components and the OpenTelemetry endpoint.

To provide the certificate, mount a volume containing the secret that holds the trusted certificate. Unless the endpoint address is redirected from http, use https.

Example configuration for TLS

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
spec:
  #...
  template:
    connectContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "https://otlp-host:4317"
        - name: OTEL_EXPORTER_OTLP_CERTIFICATE
          value: "/mnt/mysecret/my-certificate.crt"
      volumeMounts:
        - name: tracing-secret-volume
          mountPath: /mnt/mysecret
    pod:
      volumes:
        - name: tracing-secret-volume
          secret:
            secretName: mysecret
  tracing:
    type: opentelemetry
  #...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
spec:
  #...
  template:
    connectContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "https://otlp-host:4317"
        - name: OTEL_EXPORTER_OTLP_CERTIFICATE
          value: "/mnt/mysecret/my-certificate.crt"
      volumeMounts:
        - name: tracing-secret-volume
          mountPath: /mnt/mysecret
    pod:
      volumes:
        - name: tracing-secret-volume
          secret:
            secretName: mysecret
  tracing:
    type: opentelemetry
  #...

Copy to Clipboard

Toggle word wrap

Apply the changes to the custom resource configuration.

26.3.3. Initializing tracing for Kafka clients
复制链接

Initialize a tracer for OpenTelemetry, then instrument your client applications for distributed tracing. You can instrument Kafka producer and consumer clients, and Kafka Streams API applications.

Configure and initialize a tracer using a set of tracing environment variables.

Procedure

In each client application add the dependencies for the tracer:

Add the Maven dependencies to the pom.xml file for the client application:

Dependencies for OpenTelemetry

<dependency>
    <groupId>io.opentelemetry.semconv</groupId>
    <artifactId>opentelemetry-semconv</artifactId>
    <version>1.21.0.alpha-redhat-00001</version>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-otlp</artifactId>
    <version>1.34.1.redhat-00001</version>
    <exclusions>
        <exclusion>
            <groupId>io.opentelemetry</groupId>
            <artifactId>opentelemetry-exporter-sender-okhttp</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-sender-grpc-managed-channel</artifactId>
    <version>1.34.1.redhat-00001</version>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-sdk-extension-autoconfigure</artifactId>
    <version>1.34.1.redhat-00001</version>
</dependency>
<dependency>
    <groupId>io.opentelemetry.instrumentation</groupId>
    <artifactId>opentelemetry-kafka-clients-2.6</artifactId>
    <version>1.32.0.redhat-00001</version>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-sdk</artifactId>
    <version>1.34.1.redhat-00001</version>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-sender-jdk</artifactId>
    <version>1.34.1.alpha-redhat-00001</version>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>io.grpc</groupId>
    <artifactId>grpc-netty-shaded</artifactId>
    <version>1.61.0.redhat-00004</version>
</dependency>

<dependency>
    <groupId>io.opentelemetry.semconv</groupId>
    <artifactId>opentelemetry-semconv</artifactId>
    <version>1.21.0.alpha-redhat-00001</version>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-otlp</artifactId>
    <version>1.34.1.redhat-00001</version>
    <exclusions>
        <exclusion>
            <groupId>io.opentelemetry</groupId>
            <artifactId>opentelemetry-exporter-sender-okhttp</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-sender-grpc-managed-channel</artifactId>
    <version>1.34.1.redhat-00001</version>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-sdk-extension-autoconfigure</artifactId>
    <version>1.34.1.redhat-00001</version>
</dependency>
<dependency>
    <groupId>io.opentelemetry.instrumentation</groupId>
    <artifactId>opentelemetry-kafka-clients-2.6</artifactId>
    <version>1.32.0.redhat-00001</version>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-sdk</artifactId>
    <version>1.34.1.redhat-00001</version>
</dependency>
<dependency>
    <groupId>io.opentelemetry</groupId>
    <artifactId>opentelemetry-exporter-sender-jdk</artifactId>
    <version>1.34.1.alpha-redhat-00001</version>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>io.grpc</groupId>
    <artifactId>grpc-netty-shaded</artifactId>
    <version>1.61.0.redhat-00004</version>
</dependency>

Copy to Clipboard

Toggle word wrap

Define the configuration of the tracer using the tracing environment variables.
Create a tracer, which is initialized with the environment variables:
Creating a tracer for OpenTelemetry
```
OpenTelemetry ot = GlobalOpenTelemetry.get();
```
```
OpenTelemetry ot = GlobalOpenTelemetry.get();
```
Copy to Clipboard Toggle word wrap
Register the tracer as a global tracer:
```
GlobalTracer.register(tracer);
```
```
GlobalTracer.register(tracer);
```
Copy to Clipboard Toggle word wrap
Instrument your client:
- Section 26.3.4, “Instrumenting producers and consumers for tracing”
- Section 26.3.5, “Instrumenting Kafka Streams applications for tracing”

26.3.4. Instrumenting producers and consumers for tracing
复制链接

Instrument application code to enable tracing in Kafka producers and consumers. Use a decorator pattern or interceptors to instrument your Java producer and consumer application code for tracing. You can then record traces when messages are produced or retrieved from a topic.

OpenTelemetry instrumentation project provides classes that support instrumentation of producers and consumers.

Decorator instrumentation: For decorator instrumentation, create a modified producer or consumer instance for tracing.
Interceptor instrumentation: For interceptor instrumentation, add the tracing capability to the consumer or producer configuration.

Prerequisites

You have initialized tracing for the client.
You enable instrumentation in producer and consumer applications by adding the tracing JARs as dependencies to your project.

Procedure

Perform these steps in the application code of each producer and consumer application. Instrument your client application code using either a decorator pattern or interceptors.

To use a decorator pattern, create a modified producer or consumer instance to send or receive messages.

You pass the original KafkaProducer or KafkaConsumer class.

Example decorator instrumentation for OpenTelemetry

// Producer instance
Producer < String, String > op = new KafkaProducer < > (
    configs,
    new StringSerializer(),
    new StringSerializer()
    );
    Producer < String, String > producer = tracing.wrap(op);
KafkaTracing tracing = KafkaTracing.create(GlobalOpenTelemetry.get());
producer.send(...);

//consumer instance
Consumer<String, String> oc = new KafkaConsumer<>(
    configs,
    new StringDeserializer(),
    new StringDeserializer()
    );
    Consumer<String, String> consumer = tracing.wrap(oc);
consumer.subscribe(Collections.singleton("mytopic"));
ConsumerRecords<Integer, String> records = consumer.poll(1000);
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

// Producer instance
Producer < String, String > op = new KafkaProducer < > (
    configs,
    new StringSerializer(),
    new StringSerializer()
    );
    Producer < String, String > producer = tracing.wrap(op);
KafkaTracing tracing = KafkaTracing.create(GlobalOpenTelemetry.get());
producer.send(...);

//consumer instance
Consumer<String, String> oc = new KafkaConsumer<>(
    configs,
    new StringDeserializer(),
    new StringDeserializer()
    );
    Consumer<String, String> consumer = tracing.wrap(oc);
consumer.subscribe(Collections.singleton("mytopic"));
ConsumerRecords<Integer, String> records = consumer.poll(1000);
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

Copy to Clipboard

Toggle word wrap

To use interceptors, set the interceptor class in the producer or consumer configuration.

You use the KafkaProducer and KafkaConsumer classes in the usual way. The TracingProducerInterceptor and TracingConsumerInterceptor interceptor classes take care of the tracing capability.

Example producer configuration using interceptors

senderProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
    TracingProducerInterceptor.class.getName());

KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);
producer.send(...);

senderProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
    TracingProducerInterceptor.class.getName());

KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);
producer.send(...);

Copy to Clipboard

Toggle word wrap

Example consumer configuration using interceptors

consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
    TracingConsumerInterceptor.class.getName());

KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("messages"));
ConsumerRecords<Integer, String> records = consumer.poll(1000);
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
    TracingConsumerInterceptor.class.getName());

KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("messages"));
ConsumerRecords<Integer, String> records = consumer.poll(1000);
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

Copy to Clipboard

Toggle word wrap

26.3.5. Instrumenting Kafka Streams applications for tracing
复制链接

Instrument application code to enable tracing in Kafka Streams API applications. Use a decorator pattern or interceptors to instrument your Kafka Streams API applications for tracing. You can then record traces when messages are produced or retrieved from a topic.

Decorator instrumentation: For decorator instrumentation, create a modified Kafka Streams instance for tracing. For OpenTelemetry, you need to create a custom TracingKafkaClientSupplier class to provide tracing instrumentation for Kafka Streams.
Interceptor instrumentation: For interceptor instrumentation, add the tracing capability to the Kafka Streams producer and consumer configuration.

Prerequisites

You have initialized tracing for the client.
You enable instrumentation in Kafka Streams applications by adding the tracing JARs as dependencies to your project.
To instrument Kafka Streams with OpenTelemetry, you’ll need to write a custom TracingKafkaClientSupplier.

The custom TracingKafkaClientSupplier can extend Kafka’s DefaultKafkaClientSupplier, overriding the producer and consumer creation methods to wrap the instances with the telemetry-related code.

Example custom TracingKafkaClientSupplier

private class TracingKafkaClientSupplier extends DefaultKafkaClientSupplier {
    @Override
    public Producer<byte[], byte[]> getProducer(Map<String, Object> config) {
        KafkaTelemetry telemetry = KafkaTelemetry.create(GlobalOpenTelemetry.get());
        return telemetry.wrap(super.getProducer(config));
    }

    @Override
    public Consumer<byte[], byte[]> getConsumer(Map<String, Object> config) {
        KafkaTelemetry telemetry = KafkaTelemetry.create(GlobalOpenTelemetry.get());
        return telemetry.wrap(super.getConsumer(config));
    }

    @Override
    public Consumer<byte[], byte[]> getRestoreConsumer(Map<String, Object> config) {
        return this.getConsumer(config);
    }

    @Override
    public Consumer<byte[], byte[]> getGlobalConsumer(Map<String, Object> config) {
        return this.getConsumer(config);
    }
}

private class TracingKafkaClientSupplier extends DefaultKafkaClientSupplier {
    @Override
    public Producer<byte[], byte[]> getProducer(Map<String, Object> config) {
        KafkaTelemetry telemetry = KafkaTelemetry.create(GlobalOpenTelemetry.get());
        return telemetry.wrap(super.getProducer(config));
    }

    @Override
    public Consumer<byte[], byte[]> getConsumer(Map<String, Object> config) {
        KafkaTelemetry telemetry = KafkaTelemetry.create(GlobalOpenTelemetry.get());
        return telemetry.wrap(super.getConsumer(config));
    }

    @Override
    public Consumer<byte[], byte[]> getRestoreConsumer(Map<String, Object> config) {
        return this.getConsumer(config);
    }

    @Override
    public Consumer<byte[], byte[]> getGlobalConsumer(Map<String, Object> config) {
        return this.getConsumer(config);
    }
}

Copy to Clipboard

Toggle word wrap

Procedure

Perform these steps for each Kafka Streams API application.

To use a decorator pattern, create an instance of the TracingKafkaClientSupplier supplier interface, then provide the supplier interface to KafkaStreams.

Example decorator instrumentation

KafkaClientSupplier supplier = new TracingKafkaClientSupplier(tracer);
KafkaStreams streams = new KafkaStreams(builder.build(), new StreamsConfig(config), supplier);
streams.start();

KafkaClientSupplier supplier = new TracingKafkaClientSupplier(tracer);
KafkaStreams streams = new KafkaStreams(builder.build(), new StreamsConfig(config), supplier);
streams.start();

Copy to Clipboard

Toggle word wrap

To use interceptors, set the interceptor class in the Kafka Streams producer and consumer configuration.

The TracingProducerInterceptor and TracingConsumerInterceptor interceptor classes take care of the tracing capability.

Example producer and consumer configuration using interceptors

props.put(StreamsConfig.PRODUCER_PREFIX + ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
props.put(StreamsConfig.CONSUMER_PREFIX + ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());

props.put(StreamsConfig.PRODUCER_PREFIX + ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
props.put(StreamsConfig.CONSUMER_PREFIX + ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());

Copy to Clipboard

Toggle word wrap

26.3.6. Introducing a different OpenTelemetry tracing system
复制链接

Instead of the default OTLP system, you can specify other tracing systems that are supported by OpenTelemetry. You do this by adding the required artifacts to the Kafka image provided with Streams for Apache Kafka. Any required implementation specific environment variables must also be set. You then enable the new tracing implementation using the OTEL_TRACES_EXPORTER environment variable.

This procedure shows how to implement Zipkin tracing.

Procedure

Add the tracing artifacts to the /opt/kafka/libs/ directory of the Kafka image.
You can use the Kafka container image on the Red Hat Ecosystem Catalog as a base image for creating a new custom image.
OpenTelemetry artifact for Zipkin
```
io.opentelemetry:opentelemetry-exporter-zipkin
```
```
io.opentelemetry:opentelemetry-exporter-zipkin
```
Copy to Clipboard Toggle word wrap

Set the tracing exporter and endpoint for the new tracing implementation.

Example Zikpin tracer configuration

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mm2-cluster
spec:
  #...
  template:
    connectContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-zipkin-service
        - name: OTEL_EXPORTER_ZIPKIN_ENDPOINT
          value: http://zipkin-exporter-host-name:9411/api/v2/spans 
        - name: OTEL_TRACES_EXPORTER
          value: zipkin 
  tracing:
    type: opentelemetry
#...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mm2-cluster
spec:
  #...
  template:
    connectContainer:
      env:
        - name: OTEL_SERVICE_NAME
          value: my-zipkin-service
        - name: OTEL_EXPORTER_ZIPKIN_ENDPOINT
          value: http://zipkin-exporter-host-name:9411/api/v2/spans


        - name: OTEL_TRACES_EXPORTER
          value: zipkin


  tracing:
    type: opentelemetry
#...

Copy to Clipboard

Toggle word wrap

1: Specifies the Zipkin endpoint to connect to.
2: The Zipkin exporter.

26.3.7. Specifying custom span names for OpenTelemetry
复制链接

A tracing span is a logical unit of work in Jaeger, with an operation name, start time, and duration. Spans have built-in names, but you can specify custom span names in your Kafka client instrumentation where used.

Specifying custom span names is optional and only applies when using a decorator pattern in producer and consumer client instrumentation or Kafka Streams instrumentation.

Custom span names cannot be specified directly with OpenTelemetry. Instead, you retrieve span names by adding code to your client application to extract additional tags and attributes.

Example code to extract attributes

//Defines attribute extraction for a producer
private static class ProducerAttribExtractor implements AttributesExtractor < ProducerRecord < ? , ? > , Void > {
    @Override
    public void onStart(AttributesBuilder attributes, ProducerRecord < ? , ? > producerRecord) {
        set(attributes, AttributeKey.stringKey("prod_start"), "prod1");
    }
    @Override
    public void onEnd(AttributesBuilder attributes, ProducerRecord < ? , ? > producerRecord, @Nullable Void unused, @Nullable Throwable error) {
        set(attributes, AttributeKey.stringKey("prod_end"), "prod2");
    }
}
//Defines attribute extraction for a consumer
private static class ConsumerAttribExtractor implements AttributesExtractor < ConsumerRecord < ? , ? > , Void > {
    @Override
    public void onStart(AttributesBuilder attributes, ConsumerRecord < ? , ? > producerRecord) {
        set(attributes, AttributeKey.stringKey("con_start"), "con1");
    }
    @Override
    public void onEnd(AttributesBuilder attributes, ConsumerRecord < ? , ? > producerRecord, @Nullable Void unused, @Nullable Throwable error) {
        set(attributes, AttributeKey.stringKey("con_end"), "con2");
    }
}
//Extracts the attributes
public static void main(String[] args) throws Exception {
        Map < String, Object > configs = new HashMap < > (Collections.singletonMap(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"));
        System.setProperty("otel.traces.exporter", "jaeger");
        System.setProperty("otel.service.name", "myapp1");
        KafkaTracing tracing = KafkaTracing.newBuilder(GlobalOpenTelemetry.get())
            .addProducerAttributesExtractors(new ProducerAttribExtractor())
            .addConsumerAttributesExtractors(new ConsumerAttribExtractor())
            .build();

//Defines attribute extraction for a producer
private static class ProducerAttribExtractor implements AttributesExtractor < ProducerRecord < ? , ? > , Void > {
    @Override
    public void onStart(AttributesBuilder attributes, ProducerRecord < ? , ? > producerRecord) {
        set(attributes, AttributeKey.stringKey("prod_start"), "prod1");
    }
    @Override
    public void onEnd(AttributesBuilder attributes, ProducerRecord < ? , ? > producerRecord, @Nullable Void unused, @Nullable Throwable error) {
        set(attributes, AttributeKey.stringKey("prod_end"), "prod2");
    }
}
//Defines attribute extraction for a consumer
private static class ConsumerAttribExtractor implements AttributesExtractor < ConsumerRecord < ? , ? > , Void > {
    @Override
    public void onStart(AttributesBuilder attributes, ConsumerRecord < ? , ? > producerRecord) {
        set(attributes, AttributeKey.stringKey("con_start"), "con1");
    }
    @Override
    public void onEnd(AttributesBuilder attributes, ConsumerRecord < ? , ? > producerRecord, @Nullable Void unused, @Nullable Throwable error) {
        set(attributes, AttributeKey.stringKey("con_end"), "con2");
    }
}
//Extracts the attributes
public static void main(String[] args) throws Exception {
        Map < String, Object > configs = new HashMap < > (Collections.singletonMap(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"));
        System.setProperty("otel.traces.exporter", "jaeger");
        System.setProperty("otel.service.name", "myapp1");
        KafkaTracing tracing = KafkaTracing.newBuilder(GlobalOpenTelemetry.get())
            .addProducerAttributesExtractors(new ProducerAttribExtractor())
            .addConsumerAttributesExtractors(new ConsumerAttribExtractor())
            .build();

Copy to Clipboard

Toggle word wrap

返回顶部

此内容没有您所选择的语言版本。

Chapter 26. Introducing distributed tracing

26.1. Tracing options
复制链接

26.2. Environment variables for tracing
复制链接

26.3. Setting up distributed tracing
复制链接

26.3.1. Prerequisites
复制链接

26.3.2. Enabling tracing in supported Kafka components
复制链接

26.3.3. Initializing tracing for Kafka clients
复制链接

26.3.4. Instrumenting producers and consumers for tracing
复制链接

26.3.5. Instrumenting Kafka Streams applications for tracing
复制链接

26.3.6. Introducing a different OpenTelemetry tracing system
复制链接

26.3.7. Specifying custom span names for OpenTelemetry
复制链接

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

此内容没有您所选择的语言版本。

Chapter 26. Introducing distributed tracing

26.1. Tracing options复制链接链接已复制到粘贴板!

26.2. Environment variables for tracing复制链接链接已复制到粘贴板!

26.3. Setting up distributed tracing复制链接链接已复制到粘贴板!

26.3.1. Prerequisites复制链接链接已复制到粘贴板!

26.3.2. Enabling tracing in supported Kafka components复制链接链接已复制到粘贴板!

26.3.3. Initializing tracing for Kafka clients复制链接链接已复制到粘贴板!

26.3.4. Instrumenting producers and consumers for tracing复制链接链接已复制到粘贴板!

26.3.5. Instrumenting Kafka Streams applications for tracing复制链接链接已复制到粘贴板!

26.3.6. Introducing a different OpenTelemetry tracing system复制链接链接已复制到粘贴板!

26.3.7. Specifying custom span names for OpenTelemetry复制链接链接已复制到粘贴板!

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

26.1. Tracing options
复制链接

26.2. Environment variables for tracing
复制链接

26.3. Setting up distributed tracing
复制链接

26.3.1. Prerequisites
复制链接

26.3.2. Enabling tracing in supported Kafka components
复制链接

26.3.3. Initializing tracing for Kafka clients
复制链接

26.3.4. Instrumenting producers and consumers for tracing
复制链接

26.3.5. Instrumenting Kafka Streams applications for tracing
复制链接

26.3.6. Introducing a different OpenTelemetry tracing system
复制链接

26.3.7. Specifying custom span names for OpenTelemetry
复制链接