-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Rd 12974 java distro support kafka payloads (#419)
- Loading branch information
Showing
24 changed files
with
513 additions
and
859 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
45 changes: 14 additions & 31 deletions
45
instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/README.md
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,40 +1,23 @@ | ||
| # Background on fork | ||
| # Background on instrumentation | ||
|
|
||
| This module is a fork of the upstream [Kafka client instrumentation](https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent). | ||
| This fork adds `ConsumerPayloadAttributeExtractor` into the list of extractors used in | ||
| `KafkaSingletons.CONSUMER_PROCESS_INSTRUMENTER` to facilitate collection of message payloads on a span. | ||
| This module adds kafka messages payloads to the spans generated by the upstream [Kafka client instrumentation](https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent). | ||
|
|
||
| This fork is necessary as the byte code advice directly accesses a static field on `KafkaSingletons` | ||
| containing the instrumenter. There is no mechanism to extend the `InstrumenterBuilder` to replace | ||
| aspects of what it provides, which is why this approach has been taken. | ||
| ## How it works | ||
|
|
||
| Every update to OpenTelemetry JAVA SDK base version will require: | ||
| - A re-copy of instrumentation changes from the [javaagent module](https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent) | ||
| - Remove `KafkaClientsInstrumentationModule` from the copied instrumentation | ||
| - Re-application of the following changes: | ||
| We utilize two instrumentation modules because one needs to run before the OTEL instrumentation and the other after. | ||
|
|
||
| ## `KafkaSingletons` | ||
| ### Adding Producer instrumentation payloads | ||
|
|
||
| Update the `static` block to use the below for creating the `CONSUMER_PROCESS_INSTRUMENTER`: | ||
|
|
||
| ```java | ||
| // Add Lumigo custom payload extractor | ||
| CONSUMER_PROCESS_INSTRUMENTER = | ||
| instrumenterFactory.createConsumerOperationInstrumenter( | ||
| MessageOperation.PROCESS, Collections.singletonList(new ConsumerPayloadAttributeExtractor())); | ||
| ``` | ||
| Similar to the Otel producer instrumentation, we instrument the `send` method of `KafkaProducer` class. | ||
| Then run after the Otel intrumentation and add the payload to the span attributes. | ||
|
|
||
| ## `KafkaProducerInstrumentation` | ||
| ### Adding Consumer instrumentation payloads | ||
|
|
||
| Modify `SendAdvice.onMethodEnter()` to include the below before creating producer callback: | ||
| The OTEL consumer instrumentation creates consume spans only when the user iterates over the records. | ||
| It does this by wrapping the iterator of the ConsumerRecords class. | ||
| Our payload instrumentation works similarly by wrapping the OTEL iterator with our own iterator, | ||
| and then adding the payloads to the span attributes generated when the user iterates over the records. | ||
|
|
||
| ```java | ||
| if (null != record.value()) { | ||
| Java8BytecodeBridge.currentSpan() | ||
| .setAttribute(SemanticAttributes.MESSAGING_PAYLOAD, record.value().toString()); | ||
| } | ||
| ``` | ||
| ## Maintenance | ||
|
|
||
| ## `io.lumigo.javaagent.instrumentation.kafkaclients.v0_11.KafkaClientInstrumentationModule` | ||
|
|
||
| Ensure `typeInstrumentations()` includes all instrumentations from the upstream module. | ||
| No special maintenance is required for this module. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
43 changes: 0 additions & 43 deletions
43
...umigo/javaagent/instrumentation/kafkaclients/v0_11/ConsumerPayloadAttributeExtractor.java
This file was deleted.
Oops, something went wrong.
94 changes: 94 additions & 0 deletions
94
...o/javaagent/instrumentation/kafkaclients/v0_11/ConsumerRecordsPayloadInstrumentation.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| /* | ||
| * Copyright 2024 Lumigo LTD | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| * | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package io.lumigo.javaagent.instrumentation.kafkaclients.v0_11; | ||
|
|
||
| import static net.bytebuddy.matcher.ElementMatchers.isMethod; | ||
| import static net.bytebuddy.matcher.ElementMatchers.isPublic; | ||
| import static net.bytebuddy.matcher.ElementMatchers.named; | ||
| import static net.bytebuddy.matcher.ElementMatchers.returns; | ||
| import static net.bytebuddy.matcher.ElementMatchers.takesArgument; | ||
| import static net.bytebuddy.matcher.ElementMatchers.takesArguments; | ||
|
|
||
| import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; | ||
| import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer; | ||
| import java.util.Iterator; | ||
| import java.util.List; | ||
| import net.bytebuddy.asm.Advice; | ||
| import net.bytebuddy.description.type.TypeDescription; | ||
| import net.bytebuddy.matcher.ElementMatcher; | ||
| import org.apache.kafka.clients.consumer.ConsumerRecord; | ||
|
|
||
| public class ConsumerRecordsPayloadInstrumentation implements TypeInstrumentation { | ||
|
|
||
| @Override | ||
| public ElementMatcher<TypeDescription> typeMatcher() { | ||
| return named("org.apache.kafka.clients.consumer.ConsumerRecords"); | ||
| } | ||
|
|
||
| @Override | ||
| public void transform(TypeTransformer transformer) { | ||
| transformer.applyAdviceToMethod( | ||
| isMethod() | ||
| .and(isPublic()) | ||
| .and(named("records")) | ||
| .and(takesArgument(0, String.class)) | ||
| .and(returns(Iterable.class)), | ||
| ConsumerRecordsPayloadInstrumentation.class.getName() + "$IterableAdvice"); | ||
| transformer.applyAdviceToMethod( | ||
| isMethod() | ||
| .and(isPublic()) | ||
| .and(named("records")) | ||
| .and(takesArgument(0, named("org.apache.kafka.common.TopicPartition"))) | ||
| .and(returns(List.class)), | ||
| ConsumerRecordsPayloadInstrumentation.class.getName() + "$ListAdvice"); | ||
| transformer.applyAdviceToMethod( | ||
| isMethod() | ||
| .and(isPublic()) | ||
| .and(named("iterator")) | ||
| .and(takesArguments(0)) | ||
| .and(returns(Iterator.class)), | ||
| ConsumerRecordsPayloadInstrumentation.class.getName() + "$IteratorAdvice"); | ||
| } | ||
|
|
||
| @SuppressWarnings("unused") | ||
| public static class IterableAdvice { | ||
| @Advice.OnMethodExit(suppress = Throwable.class) | ||
| public static <K, V> void wrap( | ||
| @Advice.Return(readOnly = false) Iterable<ConsumerRecord<K, V>> iterable) { | ||
| iterable = PayloadTracingIterable.wrap(iterable); | ||
| } | ||
| } | ||
|
|
||
| @SuppressWarnings("unused") | ||
| public static class ListAdvice { | ||
| @Advice.OnMethodExit(suppress = Throwable.class) | ||
| public static <K, V> void wrap( | ||
| @Advice.Return(readOnly = false) List<ConsumerRecord<K, V>> list) { | ||
| list = PayloadTracingList.wrap(list); | ||
| } | ||
| } | ||
|
|
||
| @SuppressWarnings("unused") | ||
| public static class IteratorAdvice { | ||
| @Advice.OnMethodExit(suppress = Throwable.class) | ||
| public static <K, V> void wrap( | ||
| @Advice.Return(readOnly = false) Iterator<ConsumerRecord<K, V>> iterator) { | ||
| iterator = PayloadTracingIterator.wrap(iterator); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
45 changes: 45 additions & 0 deletions
45
...t/instrumentation/kafkaclients/v0_11/KafkaClientConsumerRecordsInstrumentationModule.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| /* | ||
| * Copyright 2023 Lumigo LTD | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| * | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package io.lumigo.javaagent.instrumentation.kafkaclients.v0_11; | ||
|
|
||
| import com.google.auto.service.AutoService; | ||
| import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule; | ||
| import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
|
|
||
| @AutoService(InstrumentationModule.class) | ||
| public class KafkaClientConsumerRecordsInstrumentationModule extends InstrumentationModule { | ||
| public KafkaClientConsumerRecordsInstrumentationModule() { | ||
| super( | ||
| "lumigo-kafka-clients-consumer-payloads", | ||
| "lumigo-kafka-clients-0.11", | ||
| "lumigo-kafka", | ||
| "lumigo-kafka-consumer-payload"); | ||
| } | ||
|
|
||
| @Override | ||
| public List<TypeInstrumentation> typeInstrumentations() { | ||
| return Collections.singletonList(new ConsumerRecordsPayloadInstrumentation()); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isHelperClass(String className) { | ||
| return className.startsWith("io.lumigo.javaagent.instrumentation.kafkaclients.v0_11."); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.