Skip to content

Commit f6c52ab

Browse files
trasktrask
committed
Migrate RocketMQ messaging telemetry to v1.43
1 parent 3e750e6 commit f6c52ab

11 files changed

Lines changed: 541 additions & 180 deletions

File tree

‎instrumentation/rocketmq/rocketmq-client-4.8/javaagent/build.gradle.kts‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,23 @@ tasks {
4646
systemProperty("metadataConfig", "otel.instrumentation.rocketmq-client.experimental-span-attributes=true")
4747
}
4848

49+
val testV3Preview = register<Test>("testV3Preview") {
50+
testClassesDirs = sourceSets.test.get().output.classesDirs
51+
classpath = sourceSets.test.get().runtimeClasspath
52+
jvmArgs("-Dotel.instrumentation.common.v3-preview=true")
53+
jvmArgs("-Dotel.semconv-stability.preview=messaging")
54+
systemProperty("metadataConfig", "otel.instrumentation.common.v3-preview=true")
55+
}
56+
57+
val testBothSemconv = register<Test>("testBothSemconv") {
58+
testClassesDirs = sourceSets.test.get().output.classesDirs
59+
classpath = sourceSets.test.get().runtimeClasspath
60+
jvmArgs("-Dotel.instrumentation.common.v3-preview=true")
61+
jvmArgs("-Dotel.semconv-stability.preview=messaging/dup")
62+
systemProperty("metadataConfig", "otel.semconv-stability.preview=messaging/dup")
63+
}
64+
4965
check {
50-
dependsOn(testExperimental)
66+
dependsOn(testExperimental, testV3Preview, testBothSemconv)
5167
}
5268
}

‎instrumentation/rocketmq/rocketmq-client-4.8/library/build.gradle.kts‎

Lines changed: 30 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -13,14 +13,34 @@ dependencies {
1313
testImplementation(project(":instrumentation:rocketmq:rocketmq-client-4.8:testing"))
1414
}
1515

16-
tasks.test {
17-
systemProperty("testLatestDeps", otelProps.testLatestDeps)
18-
// required on jdk17
19-
jvmArgs("--add-opens=java.base/sun.nio.ch=ALL-UNNAMED")
20-
jvmArgs("-XX:+IgnoreUnrecognizedVMOptions")
21-
// with default settings tests will fail when disk is 90% full
22-
jvmArgs("-Drocketmq.broker.diskSpaceWarningLevelRatio=1.0")
23-
24-
// used for experimental attributes test assertion logic which looks for this property
25-
jvmArgs("-Dotel.instrumentation.rocketmq-client.experimental-span-attributes=true")
16+
tasks {
17+
withType<Test>().configureEach {
18+
systemProperty("testLatestDeps", otelProps.testLatestDeps)
19+
// required on jdk17
20+
jvmArgs("--add-opens=java.base/sun.nio.ch=ALL-UNNAMED")
21+
jvmArgs("-XX:+IgnoreUnrecognizedVMOptions")
22+
// with default settings tests will fail when disk is 90% full
23+
jvmArgs("-Drocketmq.broker.diskSpaceWarningLevelRatio=1.0")
24+
25+
// used for experimental attributes test assertion logic which looks for this property
26+
jvmArgs("-Dotel.instrumentation.rocketmq-client.experimental-span-attributes=true")
27+
}
28+
29+
val testV3Preview = register<Test>("testV3Preview") {
30+
testClassesDirs = sourceSets.test.get().output.classesDirs
31+
classpath = sourceSets.test.get().runtimeClasspath
32+
jvmArgs("-Dotel.instrumentation.common.v3-preview=true")
33+
jvmArgs("-Dotel.semconv-stability.preview=messaging")
34+
}
35+
36+
val testBothSemconv = register<Test>("testBothSemconv") {
37+
testClassesDirs = sourceSets.test.get().output.classesDirs
38+
classpath = sourceSets.test.get().runtimeClasspath
39+
jvmArgs("-Dotel.instrumentation.common.v3-preview=true")
40+
jvmArgs("-Dotel.semconv-stability.preview=messaging/dup")
41+
}
42+
43+
check {
44+
dependsOn(testV3Preview, testBothSemconv)
45+
}
2646
}

‎instrumentation/rocketmq/rocketmq-client-4.8/library/src/main/java/io/opentelemetry/instrumentation/rocketmqclient/v4_8/RocketMqConsumerInstrumenter.java‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,8 @@
55

66
package io.opentelemetry.instrumentation.rocketmqclient.v4_8;
77

8+
import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableMessagingSemconv;
9+
810
import io.opentelemetry.context.Context;
911
import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
1012
import java.util.List;
@@ -14,12 +16,12 @@ final class RocketMqConsumerInstrumenter {
1416

1517
private final Instrumenter<MessageExt, Void> singleProcessInstrumenter;
1618
private final Instrumenter<MessageExt, Void> batchProcessInstrumenter;
17-
private final Instrumenter<Void, Void> batchReceiveInstrumenter;
19+
private final Instrumenter<Integer, Void> batchReceiveInstrumenter;
1820

1921
RocketMqConsumerInstrumenter(
2022
Instrumenter<MessageExt, Void> singleProcessInstrumenter,
2123
Instrumenter<MessageExt, Void> batchProcessInstrumenter,
22-
Instrumenter<Void, Void> batchReceiveInstrumenter) {
24+
Instrumenter<Integer, Void> batchReceiveInstrumenter) {
2325
this.singleProcessInstrumenter = singleProcessInstrumenter;
2426
this.batchProcessInstrumenter = batchProcessInstrumenter;
2527
this.batchReceiveInstrumenter = batchReceiveInstrumenter;
@@ -31,12 +33,15 @@ Context start(Context parentContext, List<MessageExt> msgs) {
3133
return singleProcessInstrumenter.start(parentContext, msgs.get(0));
3234
}
3335
} else {
34-
if (batchReceiveInstrumenter.shouldStart(parentContext, null)) {
35-
Context rootContext = batchReceiveInstrumenter.start(parentContext, null);
36+
int batchSize = msgs.size();
37+
if (batchReceiveInstrumenter.shouldStart(parentContext, batchSize)) {
38+
Context receiveContext = batchReceiveInstrumenter.start(parentContext, batchSize);
39+
Context processParentContext =
40+
emitStableMessagingSemconv() ? parentContext : receiveContext;
3641
for (MessageExt message : msgs) {
37-
createChildSpan(rootContext, message);
42+
createChildSpan(processParentContext, message);
3843
}
39-
return rootContext;
44+
return receiveContext;
4045
}
4146
}
4247
return parentContext;
@@ -53,7 +58,7 @@ void end(Context context, List<MessageExt> msgs) {
5358
if (msgs.size() == 1) {
5459
singleProcessInstrumenter.end(context, msgs.get(0), null, null);
5560
} else {
56-
batchReceiveInstrumenter.end(context, null, null, null);
61+
batchReceiveInstrumenter.end(context, msgs.size(), null, null);
5762
}
5863
}
5964
}

‎instrumentation/rocketmq/rocketmq-client-4.8/library/src/main/java/io/opentelemetry/instrumentation/rocketmqclient/v4_8/RocketMqInstrumenterFactory.java‎

Lines changed: 59 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -8,20 +8,24 @@
88
import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal.MessagingExceptionEventExtractors.setMessagingProcessExceptionEventExtractor;
99
import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal.MessagingExceptionEventExtractors.setMessagingSendExceptionEventExtractor;
1010
import static io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor.constant;
11+
import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitOldMessagingSemconv;
12+
import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableMessagingSemconv;
1113

1214
import io.opentelemetry.api.OpenTelemetry;
1315
import io.opentelemetry.api.common.AttributeKey;
14-
import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation;
16+
import io.opentelemetry.api.common.AttributesBuilder;
17+
import io.opentelemetry.context.Context;
1518
import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor;
1619
import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesGetter;
20+
import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingOperationType;
21+
import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanKindExtractor;
1722
import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingSpanNameExtractor;
23+
import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal.MessagingProcessInstrumenterFactory;
1824
import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
1925
import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
2026
import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
21-
import io.opentelemetry.instrumentation.api.instrumenter.SpanKindExtractor;
22-
import io.opentelemetry.instrumentation.api.instrumenter.SpanLinksExtractor;
23-
import io.opentelemetry.instrumentation.api.internal.PropagatorBasedSpanLinksExtractor;
2427
import java.util.List;
28+
import javax.annotation.Nullable;
2529
import org.apache.rocketmq.client.hook.SendMessageContext;
2630
import org.apache.rocketmq.common.message.MessageExt;
2731

@@ -32,24 +36,30 @@ class RocketMqInstrumenterFactory {
3236
// copied from MessagingIncubatingAttributes
3337
private static final AttributeKey<String> MESSAGING_OPERATION =
3438
AttributeKey.stringKey("messaging.operation");
39+
private static final AttributeKey<String> MESSAGING_OPERATION_NAME =
40+
AttributeKey.stringKey("messaging.operation.name");
41+
private static final AttributeKey<String> MESSAGING_OPERATION_TYPE =
42+
AttributeKey.stringKey("messaging.operation.type");
3543
private static final AttributeKey<String> MESSAGING_SYSTEM =
3644
AttributeKey.stringKey("messaging.system");
45+
private static final AttributeKey<Long> MESSAGING_BATCH_MESSAGE_COUNT =
46+
AttributeKey.longKey("messaging.batch.message_count");
3747

3848
static Instrumenter<SendMessageContext, Void> createProducerInstrumenter(
3949
OpenTelemetry openTelemetry,
4050
List<String> capturedHeaders,
4151
boolean captureExperimentalSpanAttributes) {
4252

4353
RocketMqProducerAttributeGetter getter = new RocketMqProducerAttributeGetter();
44-
MessageOperation operation = MessageOperation.PUBLISH;
54+
MessagingOperationType operationType = MessagingOperationType.SEND;
4555

4656
InstrumenterBuilder<SendMessageContext, Void> instrumenterBuilder =
4757
Instrumenter.<SendMessageContext, Void>builder(
4858
openTelemetry,
4959
INSTRUMENTATION_NAME,
50-
MessagingSpanNameExtractor.create(getter, operation))
60+
MessagingSpanNameExtractor.create(getter, operationType))
5161
.addAttributesExtractor(
52-
buildMessagingAttributesExtractor(getter, operation, capturedHeaders));
62+
buildMessagingAttributesExtractor(getter, operationType, capturedHeaders));
5363
if (captureExperimentalSpanAttributes) {
5464
instrumenterBuilder.addAttributesExtractor(
5565
new RocketMqProducerExperimentalAttributeExtractor());
@@ -64,18 +74,43 @@ static RocketMqConsumerInstrumenter createConsumerInstrumenter(
6474
List<String> capturedHeaders,
6575
boolean captureExperimentalSpanAttributes) {
6676

67-
InstrumenterBuilder<Void, Void> batchReceiveInstrumenterBuilder =
68-
Instrumenter.<Void, Void>builder(
77+
InstrumenterBuilder<Integer, Void> batchReceiveInstrumenterBuilder =
78+
Instrumenter.<Integer, Void>builder(
6979
openTelemetry, INSTRUMENTATION_NAME, RocketMqInstrumenterFactory::spanNameOnReceive)
70-
.addAttributesExtractor(constant(MESSAGING_SYSTEM, "rocketmq"))
71-
.addAttributesExtractor(constant(MESSAGING_OPERATION, "receive"));
80+
.addAttributesExtractor(constant(MESSAGING_SYSTEM, "rocketmq"));
81+
if (emitOldMessagingSemconv()) {
82+
batchReceiveInstrumenterBuilder.addAttributesExtractor(
83+
constant(MESSAGING_OPERATION, "receive"));
84+
}
85+
if (emitStableMessagingSemconv()) {
86+
batchReceiveInstrumenterBuilder
87+
.addAttributesExtractor(constant(MESSAGING_OPERATION_NAME, "receive"))
88+
.addAttributesExtractor(constant(MESSAGING_OPERATION_TYPE, "receive"))
89+
.addAttributesExtractor(
90+
new AttributesExtractor<Integer, Void>() {
91+
@Override
92+
public void onStart(
93+
AttributesBuilder attributes, Context parentContext, Integer batchSize) {
94+
attributes.put(MESSAGING_BATCH_MESSAGE_COUNT, batchSize);
95+
}
96+
97+
@Override
98+
public void onEnd(
99+
AttributesBuilder attributes,
100+
Context context,
101+
Integer batchSize,
102+
@Nullable Void unused,
103+
@Nullable Throwable error) {}
104+
});
105+
}
72106

73107
return new RocketMqConsumerInstrumenter(
74108
createProcessInstrumenter(
75109
openTelemetry, capturedHeaders, captureExperimentalSpanAttributes, false),
76110
createProcessInstrumenter(
77111
openTelemetry, capturedHeaders, captureExperimentalSpanAttributes, true),
78-
batchReceiveInstrumenterBuilder.buildInstrumenter(SpanKindExtractor.alwaysConsumer()));
112+
batchReceiveInstrumenterBuilder.buildInstrumenter(
113+
MessagingSpanKindExtractor.create(MessagingOperationType.RECEIVE)));
79114
}
80115

81116
private static Instrumenter<MessageExt, Void> createProcessInstrumenter(
@@ -85,45 +120,39 @@ private static Instrumenter<MessageExt, Void> createProcessInstrumenter(
85120
boolean batch) {
86121

87122
RocketMqConsumerAttributeGetter getter = new RocketMqConsumerAttributeGetter();
88-
MessageOperation operation = MessageOperation.PROCESS;
123+
MessagingOperationType operationType = MessagingOperationType.PROCESS;
89124

90125
InstrumenterBuilder<MessageExt, Void> builder =
91126
Instrumenter.builder(
92127
openTelemetry,
93128
INSTRUMENTATION_NAME,
94-
MessagingSpanNameExtractor.create(getter, operation));
129+
MessagingSpanNameExtractor.create(getter, operationType));
95130

96131
builder.addAttributesExtractor(
97-
buildMessagingAttributesExtractor(getter, operation, capturedHeaders));
132+
buildMessagingAttributesExtractor(getter, operationType, capturedHeaders));
98133
if (captureExperimentalSpanAttributes) {
99134
builder.addAttributesExtractor(new RocketMqConsumerExperimentalAttributeExtractor());
100135
}
101136
setMessagingProcessExceptionEventExtractor(builder);
102137

103-
if (batch) {
104-
SpanLinksExtractor<MessageExt> spanLinksExtractor =
105-
new PropagatorBasedSpanLinksExtractor<>(
106-
openTelemetry.getPropagators().getTextMapPropagator(), new TextMapExtractAdapter());
107-
108-
return builder
109-
.addSpanLinksExtractor(spanLinksExtractor)
110-
.buildInstrumenter(SpanKindExtractor.alwaysConsumer());
111-
} else {
112-
return builder.buildConsumerInstrumenter(new TextMapExtractAdapter());
113-
}
138+
return MessagingProcessInstrumenterFactory.create(
139+
builder,
140+
openTelemetry.getPropagators().getTextMapPropagator(),
141+
new TextMapExtractAdapter(),
142+
batch);
114143
}
115144

116145
private static <T> AttributesExtractor<T, Void> buildMessagingAttributesExtractor(
117146
MessagingAttributesGetter<T, Void> getter,
118-
MessageOperation operation,
147+
MessagingOperationType operationType,
119148
List<String> capturedHeaders) {
120-
return MessagingAttributesExtractor.builder(getter, operation)
149+
return MessagingAttributesExtractor.builderForOperationType(getter, operationType)
121150
.setCapturedHeaders(capturedHeaders)
122151
.build();
123152
}
124153

125-
private static String spanNameOnReceive(Void unused) {
126-
return "multiple_sources receive";
154+
private static String spanNameOnReceive(Integer unused) {
155+
return emitStableMessagingSemconv() ? "receive multiple_sources" : "multiple_sources receive";
127156
}
128157

129158
private RocketMqInstrumenterFactory() {}

‎instrumentation/rocketmq/rocketmq-client-4.8/library/src/main/java/io/opentelemetry/instrumentation/rocketmqclient/v4_8/RocketMqProducerAttributeGetter.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import org.apache.rocketmq.client.hook.SendMessageContext;
1515
import org.apache.rocketmq.client.producer.SendResult;
1616
import org.apache.rocketmq.common.message.Message;
17+
import org.apache.rocketmq.common.message.MessageBatch;
1718

1819
final class RocketMqProducerAttributeGetter
1920
implements MessagingAttributesGetter<SendMessageContext, Void> {
@@ -80,6 +81,14 @@ public String getClientId(SendMessageContext request) {
8081
@Nullable
8182
@Override
8283
public Long getBatchMessageCount(SendMessageContext request, @Nullable Void unused) {
84+
Message message = request.getMessage();
85+
if (message instanceof MessageBatch) {
86+
long batchSize = 0;
87+
for (Message ignored : (MessageBatch) message) {
88+
batchSize++;
89+
}
90+
return batchSize;
91+
}
8392
return null;
8493
}
8594

0 commit comments

Comments
 (0)