Red Hat Summit15.2. Kafka 클라이언트의 추적 설정
Jaeger 추적기를 초기화하여 분산 추적을 위해 클라이언트 애플리케이션을 계측합니다.
15.2.1. Kafka 클라이언트에 대한 Jaeger 추적 프로그램 초기화
추적 환경 변수 세트를 사용하여 Jaeger 추적기를 구성하고 초기화합니다.
프로세스
각 클라이언트 애플리케이션에서 다음을 수행합니다.
Jaeger의 Maven 종속 항목을 클라이언트 애플리케이션의
pom.xml파일에 추가합니다.<dependency> <groupId>io.jaegertracing</groupId> <artifactId>jaeger-client</artifactId> <version>1.5.0.redhat-00001</version> </dependency>- 추적 환경 변수 를 사용하여 Jaeger 추적기의 구성을 정의합니다.
2단계에서 정의한 환경 변수에서 Jaeger 추적기를 생성합니다.
Tracer tracer = Configuration.fromEnv().getTracer();
참고Jaeger 추적기를 초기화하는 다른 방법은 Java OpenTracing 라이브러리 설명서를 참조하십시오.
Jaeger 추적기를 글로벌 추적기로 등록합니다.
GlobalTracer.register(tracer);
클라이언트 애플리케이션이 사용할 Jaeger 추적기가 초기화되었습니다.
15.2.2. 추적을 위한 생산자 및 소비자 처리
Decorator 패턴 또는 인터셉터를 사용하여 추적을 위해 Java 생산자 및 소비자 애플리케이션 코드를 계측합니다.
프로세스
각 생산자 및 소비자 애플리케이션의 애플리케이션 코드에서 다음을 수행합니다.
OpenTracing의 Maven 종속성을 생산자 또는 소비자의
pom.xml파일에 추가합니다.<dependency> <groupId>io.opentracing.contrib</groupId> <artifactId>opentracing-kafka-client</artifactId> <version>0.1.15.redhat-00006</version> </dependency>Decorator 패턴 또는 인터셉터를 사용하여 클라이언트 애플리케이션 코드를 계측합니다.
Decorator 패턴을 사용하려면 다음을 수행합니다.
// Create an instance of the KafkaProducer: KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps); // Create an instance of the TracingKafkaProducer: TracingKafkaProducer<Integer, String> tracingProducer = new TracingKafkaProducer<>(producer, tracer); // Send: tracingProducer.send(...); // Create an instance of the KafkaConsumer: KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps); // Create an instance of the TracingKafkaConsumer: TracingKafkaConsumer<Integer, String> tracingConsumer = new TracingKafkaConsumer<>(consumer, tracer); // Subscribe: tracingConsumer.subscribe(Collections.singletonList("messages")); // Get messages: ConsumerRecords<Integer, String> records = tracingConsumer.poll(1000); // Retrieve SpanContext from polled record (consumer side): ConsumerRecord<Integer, String> record = ... SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);인터셉터를 사용하려면 다음을 수행합니다.
// Register the tracer with GlobalTracer: GlobalTracer.register(tracer); // Add the TracingProducerInterceptor to the sender properties: senderProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName()); // Create an instance of the KafkaProducer: KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps); // Send: producer.send(...); // Add the TracingConsumerInterceptor to the consumer properties: consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName()); // Create an instance of the KafkaConsumer: KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps); // Subscribe: consumer.subscribe(Collections.singletonList("messages")); // Get messages: ConsumerRecords<Integer, String> records = consumer.poll(1000); // Retrieve the SpanContext from a polled message (consumer side): ConsumerRecord<Integer, String> record = ... SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);
Decorator 패턴의 사용자 정의 범위 이름
기간은 작업 이름, 시작 시간 및 기간이 있는 Jaeger의 논리적 작업 단위입니다.
생산자 및 소비자 애플리케이션을 계측하는 데 Decorator 패턴을 사용하려면 TracingKafkaProducer 및 TracingKafkaConsumer 오브젝트를 생성할 때 BiFunction 오브젝트를 추가 인수로 전달하여 사용자 정의 범위 이름을 정의합니다. OpenTracing Apache Kafka Client Instrumentation 라이브러리에는 여러 개의 기본 제공 범위 이름이 포함되어 있습니다.
예: 사용자 정의 범위 이름을 사용하여 Decorator 패턴에서 클라이언트 애플리케이션 코드를 조정
// Create a BiFunction for the KafkaProducer that operates on (String operationName, ProducerRecord consumerRecord) and returns a String to be used as the name:
BiFunction<String, ProducerRecord, String> producerSpanNameProvider =
(operationName, producerRecord) -> "CUSTOM_PRODUCER_NAME";
// Create an instance of the KafkaProducer:
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);
// Create an instance of the TracingKafkaProducer
TracingKafkaProducer<Integer, String> tracingProducer = new TracingKafkaProducer<>(producer,
tracer,
producerSpanNameProvider);
// Spans created by the tracingProducer will now have "CUSTOM_PRODUCER_NAME" as the span name.
// Create a BiFunction for the KafkaConsumer that operates on (String operationName, ConsumerRecord consumerRecord) and returns a String to be used as the name:
BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider =
(operationName, consumerRecord) -> operationName.toUpperCase();
// Create an instance of the KafkaConsumer:
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
// Create an instance of the TracingKafkaConsumer, passing in the consumerSpanNameProvider BiFunction:
TracingKafkaConsumer<Integer, String> tracingConsumer = new TracingKafkaConsumer<>(consumer,
tracer,
consumerSpanNameProvider);
// Spans created by the tracingConsumer will have the operation name as the span name, in upper-case.
// "receive" -> "RECEIVE"
기본 제공 범위 이름
사용자 지정 범위 이름을 정의할 때 ClientSpanNameProvider 클래스에서 다음 BiFunctions 를 사용할 수 있습니다. spanNameProvider 를 지정하지 않으면 CONSUMER_OPERATION_NAME 및 PRODUCER_OPERATION_NAME 이 사용됩니다.
| BiFunction | 설명 |
|---|---|
|
|
user |
|
|
|
|
|
메시지가 전송된 대상의 이름 또는 형식 |
|
|
문자열 |
|
|
작업 이름과 주제 이름을 반환합니다. |
|
|
|
15.2.3. 추적을 위한 Kafka Streams 애플리케이션 조정
공급자 인터페이스를 사용하여 분산 추적을 위한 Kafka Streams 애플리케이션입니다. 이를 통해 애플리케이션에서 인터셉터가 활성화됩니다.
프로세스
각 Kafka Streams 애플리케이션에서 다음을 수행합니다.
opentracing-kafka-streams종속성을 Kafka Streams 애플리케이션의pom.xml파일에 추가합니다.<dependency> <groupId>io.opentracing.contrib</groupId> <artifactId>opentracing-kafka-streams</artifactId> <version>0.1.15.redhat-00006</version> </dependency>TracingKafkaClientSupplier공급자 인터페이스 인스턴스를 생성합니다.KafkaClientSupplier supplier = new TracingKafkaClientSupplier(tracer);
KafkaStreams에 공급자 인터페이스를 제공합니다.KafkaStreams streams = new KafkaStreams(builder.build(), new StreamsConfig(config), supplier); streams.start();
