open-telemetry
/
opentelemetry-java-instrumentation
mirror of https://github.com/open-telemetry/opentelemetry-java-instrumentation.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Better span names for kafka (#241)

This commit is contained in:
Trask Stalnaker 2020-03-18 15:55:27 -07:00 committed by GitHub
parent f88dabedee
commit 56a1227afe
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 58 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
instrumentation/kafka-clients-0.11/src/main/java/io/opentelemetry/auto/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java
View File

@ -92,7 +92,7 @@ public final class KafkaConsumerInstrumentation extends Instrumenter.Default {
@Advice.OnMethodExit(suppress = Throwable.class) @Advice.OnMethodExit(suppress = Throwable.class)
public static void wrap(@Advice.Return(readOnly = false) Iterable<ConsumerRecord> iterable) { public static void wrap(@Advice.Return(readOnly = false) Iterable<ConsumerRecord> iterable) {
if (iterable != null) { if (iterable != null) {
iterable = new TracingIterable(iterable, "kafka.consume", CONSUMER_DECORATE); iterable = new TracingIterable(iterable, CONSUMER_DECORATE);
} }
} }
} }
@ -102,7 +102,7 @@ public final class KafkaConsumerInstrumentation extends Instrumenter.Default {
@Advice.OnMethodExit(suppress = Throwable.class) @Advice.OnMethodExit(suppress = Throwable.class)
public static void wrap(@Advice.Return(readOnly = false) List<ConsumerRecord> iterable) { public static void wrap(@Advice.Return(readOnly = false) List<ConsumerRecord> iterable) {
if (iterable != null) { if (iterable != null) {
iterable = new TracingList(iterable, "kafka.consume", CONSUMER_DECORATE); iterable = new TracingList(iterable, CONSUMER_DECORATE);
} }
} }
} }
@ -112,7 +112,7 @@ public final class KafkaConsumerInstrumentation extends Instrumenter.Default {
@Advice.OnMethodExit(suppress = Throwable.class) @Advice.OnMethodExit(suppress = Throwable.class)
public static void wrap(@Advice.Return(readOnly = false) Iterator<ConsumerRecord> iterator) { public static void wrap(@Advice.Return(readOnly = false) Iterator<ConsumerRecord> iterator) {
if (iterator != null) { if (iterator != null) {
iterator = new TracingIterator(iterator, "kafka.consume", CONSUMER_DECORATE); iterator = new TracingIterator(iterator, CONSUMER_DECORATE);
} }
} }
} }

32
instrumentation/kafka-clients-0.11/src/main/java/io/opentelemetry/auto/instrumentation/kafka_clients/KafkaDecorator.java
View File

@ -17,7 +17,6 @@ package io.opentelemetry.auto.instrumentation.kafka_clients;
import io.opentelemetry.OpenTelemetry; import io.opentelemetry.OpenTelemetry;
import io.opentelemetry.auto.bootstrap.instrumentation.decorator.ClientDecorator; import io.opentelemetry.auto.bootstrap.instrumentation.decorator.ClientDecorator;
import io.opentelemetry.auto.instrumentation.api.MoreTags;
import io.opentelemetry.auto.instrumentation.api.SpanTypes; import io.opentelemetry.auto.instrumentation.api.SpanTypes;
import io.opentelemetry.trace.Span; import io.opentelemetry.trace.Span;
import io.opentelemetry.trace.Tracer; import io.opentelemetry.trace.Tracer;
@ -54,24 +53,35 @@ public abstract class KafkaDecorator extends ClientDecorator {
return "java-kafka"; return "java-kafka";
} }
public void onConsume(final Span span, final ConsumerRecord record) { public String spanNameOnConsume(final ConsumerRecord record) {
if (record != null) { final String topic = record.topic();
final String topic = record.topic() == null ? "kafka" : record.topic(); if (topic != null) {
span.setAttribute(MoreTags.RESOURCE_NAME, "Consume Topic " + topic); return topic;
span.setAttribute("partition", record.partition()); } else {
span.setAttribute("offset", record.offset()); return "destination";
} }
} }
public String spanNameOnProduce(final ProducerRecord record) {
if (record != null) {
final String topic = record.topic();
if (topic != null) {
return topic;
}
}
return "destination";
}
public void onConsume(final Span span, final ConsumerRecord record) {
span.setAttribute("partition", record.partition());
span.setAttribute("offset", record.offset());
}
public void onProduce(final Span span, final ProducerRecord record) { public void onProduce(final Span span, final ProducerRecord record) {
if (record != null) { if (record != null) {
final String topic = record.topic() == null ? "kafka" : record.topic();
if (record.partition() != null) { if (record.partition() != null) {
span.setAttribute("kafka.partition", record.partition()); span.setAttribute("kafka.partition", record.partition());
} }
span.setAttribute(MoreTags.RESOURCE_NAME, "Produce Topic " + topic);
} }
} }
} }

6
instrumentation/kafka-clients-0.11/src/main/java/io/opentelemetry/auto/instrumentation/kafka_clients/KafkaProducerInstrumentation.java
View File

@ -82,7 +82,11 @@ public final class KafkaProducerInstrumentation extends Instrumenter.Default {
@Advice.FieldValue("apiVersions") final ApiVersions apiVersions, @Advice.FieldValue("apiVersions") final ApiVersions apiVersions,
@Advice.Argument(value = 0, readOnly = false) ProducerRecord record, @Advice.Argument(value = 0, readOnly = false) ProducerRecord record,
@Advice.Argument(value = 1, readOnly = false) Callback callback) { @Advice.Argument(value = 1, readOnly = false) Callback callback) {
final Span span = TRACER.spanBuilder("kafka.produce").setSpanKind(PRODUCER).startSpan(); final Span span =
TRACER
.spanBuilder(PRODUCER_DECORATE.spanNameOnProduce(record))
.setSpanKind(PRODUCER)
.startSpan();
PRODUCER_DECORATE.afterStart(span); PRODUCER_DECORATE.afterStart(span);
PRODUCER_DECORATE.onProduce(span, record); PRODUCER_DECORATE.onProduce(span, record);

9
instrumentation/kafka-clients-0.11/src/main/java/io/opentelemetry/auto/instrumentation/kafka_clients/TracingIterable.java
View File

@ -20,16 +20,11 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
public class TracingIterable implements Iterable<ConsumerRecord> { public class TracingIterable implements Iterable<ConsumerRecord> {
private final Iterable<ConsumerRecord> delegate; private final Iterable<ConsumerRecord> delegate;
private final String operationName;
private final KafkaDecorator decorator; private final KafkaDecorator decorator;
private boolean firstIterator = true; private boolean firstIterator = true;
public TracingIterable( public TracingIterable(final Iterable<ConsumerRecord> delegate, final KafkaDecorator decorator) {
final Iterable<ConsumerRecord> delegate,
final String operationName,
final KafkaDecorator decorator) {
this.delegate = delegate; this.delegate = delegate;
this.operationName = operationName;
this.decorator = decorator; this.decorator = decorator;
} }
@ -40,7 +35,7 @@ public class TracingIterable implements Iterable<ConsumerRecord> {
// However, this is not thread-safe, but usually the first (hopefully only) traversal of // However, this is not thread-safe, but usually the first (hopefully only) traversal of
// ConsumerRecords is performed in the same thread that called poll() // ConsumerRecords is performed in the same thread that called poll()
if (firstIterator) { if (firstIterator) {
it = new TracingIterator(delegate.iterator(), operationName, decorator); it = new TracingIterator(delegate.iterator(), decorator);
firstIterator = false; firstIterator = false;
} else { } else {
it = delegate.iterator(); it = delegate.iterator();

8
instrumentation/kafka-clients-0.11/src/main/java/io/opentelemetry/auto/instrumentation/kafka_clients/TracingIterator.java
View File

@ -29,7 +29,6 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
@Slf4j @Slf4j
public class TracingIterator implements Iterator<ConsumerRecord> { public class TracingIterator implements Iterator<ConsumerRecord> {
private final Iterator<ConsumerRecord> delegateIterator; private final Iterator<ConsumerRecord> delegateIterator;
private final String operationName;
private final KafkaDecorator decorator; private final KafkaDecorator decorator;
/** /**
@ -39,11 +38,8 @@ public class TracingIterator implements Iterator<ConsumerRecord> {
private SpanWithScope currentSpanWithScope; private SpanWithScope currentSpanWithScope;
public TracingIterator( public TracingIterator(
final Iterator<ConsumerRecord> delegateIterator, final Iterator<ConsumerRecord> delegateIterator, final KafkaDecorator decorator) {
final String operationName,
final KafkaDecorator decorator) {
this.delegateIterator = delegateIterator; this.delegateIterator = delegateIterator;
this.operationName = operationName;
this.decorator = decorator; this.decorator = decorator;
} }
@ -71,7 +67,7 @@ public class TracingIterator implements Iterator<ConsumerRecord> {
try { try {
if (next != null) { if (next != null) {
final boolean consumer = !TRACER.getCurrentSpan().getContext().isValid(); final boolean consumer = !TRACER.getCurrentSpan().getContext().isValid();
final Span.Builder spanBuilder = TRACER.spanBuilder(operationName); final Span.Builder spanBuilder = TRACER.spanBuilder(decorator.spanNameOnConsume(next));
if (consumer) { if (consumer) {
spanBuilder.setSpanKind(CONSUMER); spanBuilder.setSpanKind(CONSUMER);
} }

7
instrumentation/kafka-clients-0.11/src/main/java/io/opentelemetry/auto/instrumentation/kafka_clients/TracingList.java
View File

@ -23,11 +23,8 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
public class TracingList extends TracingIterable implements List<ConsumerRecord> { public class TracingList extends TracingIterable implements List<ConsumerRecord> {
private final List<ConsumerRecord> delegate; private final List<ConsumerRecord> delegate;
public TracingList( public TracingList(final List<ConsumerRecord> delegate, final KafkaDecorator decorator) {
final List<ConsumerRecord> delegate, super(delegate, decorator);
final String operationName,
final KafkaDecorator decorator) {
super(delegate, operationName, decorator);
this.delegate = delegate; this.delegate = delegate;
} }

12
instrumentation/kafka-clients-0.11/src/test/groovy/KafkaClientTest.groovy
View File

@ -99,25 +99,23 @@ class KafkaClientTest extends AgentTestRunner {
assertTraces(1) { assertTraces(1) {
trace(0, 2) { trace(0, 2) {
span(0) { span(0) {
operationName "kafka.produce" operationName SHARED_TOPIC
spanKind PRODUCER spanKind PRODUCER
errored false errored false
parent() parent()
tags { tags {
"$MoreTags.SERVICE_NAME" "kafka" "$MoreTags.SERVICE_NAME" "kafka"
"$MoreTags.RESOURCE_NAME" "Produce Topic $SHARED_TOPIC"
"$MoreTags.SPAN_TYPE" "queue" "$MoreTags.SPAN_TYPE" "queue"
"$Tags.COMPONENT" "java-kafka" "$Tags.COMPONENT" "java-kafka"
} }
} }
span(1) { span(1) {
operationName "kafka.consume" operationName SHARED_TOPIC
spanKind CONSUMER spanKind CONSUMER
errored false errored false
childOf span(0) childOf span(0)
tags { tags {
"$MoreTags.SERVICE_NAME" "kafka" "$MoreTags.SERVICE_NAME" "kafka"
"$MoreTags.RESOURCE_NAME" "Consume Topic $SHARED_TOPIC"
"$MoreTags.SPAN_TYPE" "queue" "$MoreTags.SPAN_TYPE" "queue"
"$Tags.COMPONENT" "java-kafka" "$Tags.COMPONENT" "java-kafka"
"partition" { it >= 0 } "partition" { it >= 0 }
@ -170,26 +168,24 @@ class KafkaClientTest extends AgentTestRunner {
assertTraces(1) { assertTraces(1) {
trace(0, 2) { trace(0, 2) {
span(0) { span(0) {
operationName "kafka.produce" operationName SHARED_TOPIC
spanKind PRODUCER spanKind PRODUCER
errored false errored false
parent() parent()
tags { tags {
"$MoreTags.SERVICE_NAME" "kafka" "$MoreTags.SERVICE_NAME" "kafka"
"$MoreTags.RESOURCE_NAME" "Produce Topic $SHARED_TOPIC"
"$MoreTags.SPAN_TYPE" "queue" "$MoreTags.SPAN_TYPE" "queue"
"$Tags.COMPONENT" "java-kafka" "$Tags.COMPONENT" "java-kafka"
"kafka.partition" { it >= 0 } "kafka.partition" { it >= 0 }
} }
} }
span(1) { span(1) {
operationName "kafka.consume" operationName SHARED_TOPIC
spanKind CONSUMER spanKind CONSUMER
errored false errored false
childOf span(0) childOf span(0)
tags { tags {
"$MoreTags.SERVICE_NAME" "kafka" "$MoreTags.SERVICE_NAME" "kafka"
"$MoreTags.RESOURCE_NAME" "Consume Topic $SHARED_TOPIC"
"$MoreTags.SPAN_TYPE" "queue" "$MoreTags.SPAN_TYPE" "queue"
"$Tags.COMPONENT" "java-kafka" "$Tags.COMPONENT" "java-kafka"
"partition" { it >= 0 } "partition" { it >= 0 }

15
instrumentation/kafka-streams-0.11/src/main/java/io/opentelemetry/auto/instrumentation/kafka_streams/KafkaStreamsDecorator.java
View File

@ -17,7 +17,6 @@ package io.opentelemetry.auto.instrumentation.kafka_streams;
import io.opentelemetry.OpenTelemetry; import io.opentelemetry.OpenTelemetry;
import io.opentelemetry.auto.bootstrap.instrumentation.decorator.ClientDecorator; import io.opentelemetry.auto.bootstrap.instrumentation.decorator.ClientDecorator;
import io.opentelemetry.auto.instrumentation.api.MoreTags;
import io.opentelemetry.auto.instrumentation.api.SpanTypes; import io.opentelemetry.auto.instrumentation.api.SpanTypes;
import io.opentelemetry.trace.Span; import io.opentelemetry.trace.Span;
import io.opentelemetry.trace.Tracer; import io.opentelemetry.trace.Tracer;
@ -44,10 +43,20 @@ public class KafkaStreamsDecorator extends ClientDecorator {
return SpanTypes.MESSAGE_CONSUMER; return SpanTypes.MESSAGE_CONSUMER;
} }
public String spanNameForConsume(final StampedRecord record) {
if (record == null) {
return null;
}
final String topic = record.topic();
if (topic != null) {
return topic;
} else {
return "destination";
}
}
public void onConsume(final Span span, final StampedRecord record) { public void onConsume(final Span span, final StampedRecord record) {
if (record != null) { if (record != null) {
final String topic = record.topic() == null ? "kafka" : record.topic();
span.setAttribute(MoreTags.RESOURCE_NAME, "Consume Topic " + topic);
span.setAttribute("partition", record.partition()); span.setAttribute("partition", record.partition());
span.setAttribute("offset", record.offset()); span.setAttribute("offset", record.offset());
} }

3
instrumentation/kafka-streams-0.11/src/main/java/io/opentelemetry/auto/instrumentation/kafka_streams/KafkaStreamsProcessorInstrumentation.java
View File

@ -103,7 +103,8 @@ public class KafkaStreamsProcessorInstrumentation {
return; return;
} }
final Span.Builder spanBuilder = TRACER.spanBuilder("kafka.consume").setSpanKind(CONSUMER); final Span.Builder spanBuilder =
TRACER.spanBuilder(CONSUMER_DECORATE.spanNameForConsume(record)).setSpanKind(CONSUMER);
try { try {
final SpanContext extractedContext = final SpanContext extractedContext =
TRACER.getHttpTextFormat().extract(record.value.headers(), GETTER); TRACER.getHttpTextFormat().extract(record.value.headers(), GETTER);

15
instrumentation/kafka-streams-0.11/src/test/groovy/KafkaStreamsTest.groovy
View File

@ -138,26 +138,24 @@ class KafkaStreamsTest extends AgentTestRunner {
trace(0, 5) { trace(0, 5) {
// PRODUCER span 0 // PRODUCER span 0
span(0) { span(0) {
operationName "kafka.produce" operationName STREAM_PENDING
spanKind PRODUCER spanKind PRODUCER
errored false errored false
parent() parent()
tags { tags {
"$MoreTags.SERVICE_NAME" "kafka" "$MoreTags.SERVICE_NAME" "kafka"
"$MoreTags.RESOURCE_NAME" "Produce Topic $STREAM_PENDING"
"$MoreTags.SPAN_TYPE" "queue" "$MoreTags.SPAN_TYPE" "queue"
"$Tags.COMPONENT" "java-kafka" "$Tags.COMPONENT" "java-kafka"
} }
} }
// CONSUMER span 0 // CONSUMER span 0
span(1) { span(1) {
operationName "kafka.consume" operationName STREAM_PENDING
spanKind CONSUMER spanKind CONSUMER
errored false errored false
childOf span(0) childOf span(0)
tags { tags {
"$MoreTags.SERVICE_NAME" "kafka" "$MoreTags.SERVICE_NAME" "kafka"
"$MoreTags.RESOURCE_NAME" "Consume Topic $STREAM_PENDING"
"$MoreTags.SPAN_TYPE" "queue" "$MoreTags.SPAN_TYPE" "queue"
"$Tags.COMPONENT" "java-kafka" "$Tags.COMPONENT" "java-kafka"
"partition" { it >= 0 } "partition" { it >= 0 }
@ -166,13 +164,12 @@ class KafkaStreamsTest extends AgentTestRunner {
} }
// STREAMING span 1 // STREAMING span 1
span(2) { span(2) {
operationName "kafka.consume" operationName STREAM_PENDING
spanKind CONSUMER spanKind CONSUMER
errored false errored false
childOf span(0) childOf span(0)
tags { tags {
"$MoreTags.SERVICE_NAME" "kafka" "$MoreTags.SERVICE_NAME" "kafka"
"$MoreTags.RESOURCE_NAME" "Consume Topic $STREAM_PENDING"
"$MoreTags.SPAN_TYPE" "queue" "$MoreTags.SPAN_TYPE" "queue"
"$Tags.COMPONENT" "java-kafka" "$Tags.COMPONENT" "java-kafka"
"partition" { it >= 0 } "partition" { it >= 0 }
@ -182,26 +179,24 @@ class KafkaStreamsTest extends AgentTestRunner {
} }
// STREAMING span 0 // STREAMING span 0
span(3) { span(3) {
operationName "kafka.produce" operationName STREAM_PROCESSED
spanKind PRODUCER spanKind PRODUCER
errored false errored false
childOf span(2) childOf span(2)
tags { tags {
"$MoreTags.SERVICE_NAME" "kafka" "$MoreTags.SERVICE_NAME" "kafka"
"$MoreTags.RESOURCE_NAME" "Produce Topic $STREAM_PROCESSED"
"$MoreTags.SPAN_TYPE" "queue" "$MoreTags.SPAN_TYPE" "queue"
"$Tags.COMPONENT" "java-kafka" "$Tags.COMPONENT" "java-kafka"
} }
} }
// CONSUMER span 0 // CONSUMER span 0
span(4) { span(4) {
operationName "kafka.consume" operationName STREAM_PROCESSED
spanKind CONSUMER spanKind CONSUMER
errored false errored false
childOf span(3) childOf span(3)
tags { tags {
"$MoreTags.SERVICE_NAME" "kafka" "$MoreTags.SERVICE_NAME" "kafka"
"$MoreTags.RESOURCE_NAME" "Consume Topic $STREAM_PROCESSED"
"$MoreTags.SPAN_TYPE" "queue" "$MoreTags.SPAN_TYPE" "queue"
"$Tags.COMPONENT" "java-kafka" "$Tags.COMPONENT" "java-kafka"
"partition" { it >= 0 } "partition" { it >= 0 }