Skip to content

Commit 8c34534

Browse files
trasktrask
authored
Emit messaging exceptions as logs under the opt-in preview (#18891)
1 parent efba62a commit 8c34534

15 files changed

Lines changed: 375 additions & 92 deletions

File tree

‎instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/messaging/internal/MessagingExceptionEventExtractors.java‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
/*
2+
* Copyright The OpenTelemetry Authors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal;
7+
8+
import io.opentelemetry.api.logs.Severity;
9+
import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
10+
import io.opentelemetry.instrumentation.api.internal.Experimental;
11+
12+
/**
13+
* This class is internal and is hence not for public use. Its APIs are unstable and can change at
14+
* any time.
15+
*/
16+
public final class MessagingExceptionEventExtractors {
17+
18+
/**
19+
* Configures the messaging create exception event name and severity. Only takes effect when
20+
* emitting exceptions as logs is enabled via the {@code otel.semconv.exception.signal.preview}
21+
* flag.
22+
*/
23+
public static <REQUEST> void setMessagingCreateExceptionEventExtractor(
24+
InstrumenterBuilder<REQUEST, ?> builder) {
25+
setExceptionEventExtractor(builder, "messaging.create.exception", Severity.WARN);
26+
}
27+
28+
/**
29+
* Configures the messaging send exception event name and severity. Only takes effect when
30+
* emitting exceptions as logs is enabled via the {@code otel.semconv.exception.signal.preview}
31+
* flag.
32+
*/
33+
public static <REQUEST> void setMessagingSendExceptionEventExtractor(
34+
InstrumenterBuilder<REQUEST, ?> builder) {
35+
setExceptionEventExtractor(builder, "messaging.send.exception", Severity.WARN);
36+
}
37+
38+
/**
39+
* Configures the messaging receive exception event name and severity. Only takes effect when
40+
* emitting exceptions as logs is enabled via the {@code otel.semconv.exception.signal.preview}
41+
* flag.
42+
*/
43+
public static <REQUEST> void setMessagingReceiveExceptionEventExtractor(
44+
InstrumenterBuilder<REQUEST, ?> builder) {
45+
setExceptionEventExtractor(builder, "messaging.receive.exception", Severity.WARN);
46+
}
47+
48+
/**
49+
* Configures the messaging settle exception event name and severity. Only takes effect when
50+
* emitting exceptions as logs is enabled via the {@code otel.semconv.exception.signal.preview}
51+
* flag.
52+
*/
53+
public static <REQUEST> void setMessagingSettleExceptionEventExtractor(
54+
InstrumenterBuilder<REQUEST, ?> builder) {
55+
setExceptionEventExtractor(builder, "messaging.settle.exception", Severity.WARN);
56+
}
57+
58+
/**
59+
* Configures the messaging process exception event name and severity. Only takes effect when
60+
* emitting exceptions as logs is enabled via the {@code otel.semconv.exception.signal.preview}
61+
* flag.
62+
*/
63+
public static <REQUEST> void setMessagingProcessExceptionEventExtractor(
64+
InstrumenterBuilder<REQUEST, ?> builder) {
65+
setExceptionEventExtractor(builder, "messaging.process.exception", Severity.ERROR);
66+
}
67+
68+
private static <REQUEST> void setExceptionEventExtractor(
69+
InstrumenterBuilder<REQUEST, ?> builder, String eventName, Severity severity) {
70+
Experimental.setExceptionEventExtractor(
71+
builder,
72+
(logRecordBuilder, context, request) -> {
73+
logRecordBuilder.setEventName(eventName);
74+
logRecordBuilder.setSeverity(severity);
75+
});
76+
}
77+
78+
private MessagingExceptionEventExtractors() {}
79+
}

‎instrumentation-api-incubator/src/test/java/io/opentelemetry/instrumentation/api/incubator/semconv/messaging/internal/MessagingExceptionEventExtractorsTest.java‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
/*
2+
* Copyright The OpenTelemetry Authors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal;
7+
8+
import static io.opentelemetry.instrumentation.api.internal.SemconvExceptionSignal.emitExceptionAsLogs;
9+
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat;
10+
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
11+
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
12+
import static io.opentelemetry.semconv.ExceptionAttributes.EXCEPTION_MESSAGE;
13+
import static io.opentelemetry.semconv.ExceptionAttributes.EXCEPTION_STACKTRACE;
14+
import static io.opentelemetry.semconv.ExceptionAttributes.EXCEPTION_TYPE;
15+
16+
import io.opentelemetry.api.logs.Severity;
17+
import io.opentelemetry.context.Context;
18+
import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
19+
import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
20+
import io.opentelemetry.sdk.logs.data.LogRecordData;
21+
import io.opentelemetry.sdk.testing.junit5.OpenTelemetryExtension;
22+
import java.util.List;
23+
import java.util.function.Consumer;
24+
import org.assertj.core.api.AbstractAssert;
25+
import org.junit.jupiter.api.Test;
26+
import org.junit.jupiter.api.extension.RegisterExtension;
27+
28+
class MessagingExceptionEventExtractorsTest {
29+
30+
@RegisterExtension
31+
static final OpenTelemetryExtension otelTesting = OpenTelemetryExtension.create();
32+
33+
@Test
34+
void messagingCreateExceptionLog() {
35+
assertExceptionLog(
36+
MessagingExceptionEventExtractors::setMessagingCreateExceptionEventExtractor,
37+
"messaging.create.exception",
38+
Severity.WARN);
39+
}
40+
41+
@Test
42+
void messagingSendExceptionLog() {
43+
assertExceptionLog(
44+
MessagingExceptionEventExtractors::setMessagingSendExceptionEventExtractor,
45+
"messaging.send.exception",
46+
Severity.WARN);
47+
}
48+
49+
@Test
50+
void messagingReceiveExceptionLog() {
51+
assertExceptionLog(
52+
MessagingExceptionEventExtractors::setMessagingReceiveExceptionEventExtractor,
53+
"messaging.receive.exception",
54+
Severity.WARN);
55+
}
56+
57+
@Test
58+
void messagingSettleExceptionLog() {
59+
assertExceptionLog(
60+
MessagingExceptionEventExtractors::setMessagingSettleExceptionEventExtractor,
61+
"messaging.settle.exception",
62+
Severity.WARN);
63+
}
64+
65+
@Test
66+
void messagingProcessExceptionLog() {
67+
assertExceptionLog(
68+
MessagingExceptionEventExtractors::setMessagingProcessExceptionEventExtractor,
69+
"messaging.process.exception",
70+
Severity.ERROR);
71+
}
72+
73+
private static void assertExceptionLog(
74+
Consumer<InstrumenterBuilder<String, String>> configure,
75+
String expectedEventName,
76+
Severity expectedSeverity) {
77+
InstrumenterBuilder<String, String> builder =
78+
Instrumenter.builder(otelTesting.getOpenTelemetry(), "test", unused -> "span");
79+
configure.accept(builder);
80+
Instrumenter<String, String> instrumenter = builder.buildInstrumenter();
81+
82+
Context context = instrumenter.start(Context.root(), "request");
83+
IllegalStateException error = new IllegalStateException("test");
84+
instrumenter.end(context, "request", "response", error);
85+
86+
List<LogRecordData> logs = otelTesting.getLogRecords();
87+
if (emitExceptionAsLogs()) {
88+
assertThat(logs).hasSize(1);
89+
assertThat(logs.get(0))
90+
.hasSeverity(expectedSeverity)
91+
.hasEventName(expectedEventName)
92+
.hasAttributesSatisfyingExactly(
93+
equalTo(EXCEPTION_TYPE, "java.lang.IllegalStateException"),
94+
equalTo(EXCEPTION_MESSAGE, "test"),
95+
satisfies(EXCEPTION_STACKTRACE, AbstractAssert::isNotNull));
96+
} else {
97+
assertThat(logs).isEmpty();
98+
}
99+
}
100+
}

‎instrumentation/jms/jms-common-1.1/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/jms/common/v1_1/JmsInstrumenterFactory.java‎

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,9 @@
55

66
package io.opentelemetry.javaagent.instrumentation.jms.common.v1_1;
77

8+
import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal.MessagingExceptionEventExtractors.setMessagingProcessExceptionEventExtractor;
9+
import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal.MessagingExceptionEventExtractors.setMessagingReceiveExceptionEventExtractor;
10+
import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal.MessagingExceptionEventExtractors.setMessagingSendExceptionEventExtractor;
811
import static java.util.Collections.emptyList;
912

1013
import com.google.errorprone.annotations.CanIgnoreReturnValue;
@@ -50,12 +53,14 @@ public Instrumenter<MessageWithDestination, Void> createProducerInstrumenter() {
5053
JmsMessageAttributesGetter getter = new JmsMessageAttributesGetter();
5154
MessageOperation operation = MessageOperation.PUBLISH;
5255

53-
return Instrumenter.<MessageWithDestination, Void>builder(
54-
openTelemetry,
55-
instrumentationName,
56-
MessagingSpanNameExtractor.create(getter, operation))
57-
.addAttributesExtractor(createMessagingAttributesExtractor(operation))
58-
.buildProducerInstrumenter(new MessagePropertySetter());
56+
InstrumenterBuilder<MessageWithDestination, Void> builder =
57+
Instrumenter.<MessageWithDestination, Void>builder(
58+
openTelemetry,
59+
instrumentationName,
60+
MessagingSpanNameExtractor.create(getter, operation))
61+
.addAttributesExtractor(createMessagingAttributesExtractor(operation));
62+
setMessagingSendExceptionEventExtractor(builder);
63+
return builder.buildProducerInstrumenter(new MessagePropertySetter());
5964
}
6065

6166
public Instrumenter<MessageWithDestination, Void> createConsumerReceiveInstrumenter() {
@@ -68,6 +73,7 @@ public Instrumenter<MessageWithDestination, Void> createConsumerReceiveInstrumen
6873
instrumentationName,
6974
MessagingSpanNameExtractor.create(getter, operation))
7075
.addAttributesExtractor(createMessagingAttributesExtractor(operation));
76+
setMessagingReceiveExceptionEventExtractor(builder);
7177
if (messagingReceiveInstrumentationEnabled) {
7278
builder.addSpanLinksExtractor(
7379
new PropagatorBasedSpanLinksExtractor<>(
@@ -88,6 +94,7 @@ public Instrumenter<MessageWithDestination, Void> createConsumerProcessInstrumen
8894
instrumentationName,
8995
MessagingSpanNameExtractor.create(getter, operation))
9096
.addAttributesExtractor(createMessagingAttributesExtractor(operation));
97+
setMessagingProcessExceptionEventExtractor(builder);
9198
if (canHaveReceiveInstrumentation && messagingReceiveInstrumentationEnabled) {
9299
builder.addSpanLinksExtractor(
93100
new PropagatorBasedSpanLinksExtractor<>(

‎instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/build.gradle.kts‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,26 @@ dependencies {
1717
}
1818

1919
tasks {
20-
test {
20+
withType<Test>().configureEach {
2121
usesService(gradle.sharedServices.registrations["testcontainersBuildService"].service)
2222
systemProperty("testLatestDeps", otelProps.testLatestDeps)
2323
systemProperty("collectMetadata", otelProps.collectMetadata)
2424
}
25+
26+
val testExceptionSignalLogs by registering(Test::class) {
27+
testClassesDirs = sourceSets.test.get().output.classesDirs
28+
classpath = sourceSets.test.get().runtimeClasspath
29+
30+
filter {
31+
includeTestsMatching("WrapperTest")
32+
}
33+
jvmArgs("-Dotel.semconv.exception.signal.preview=logs")
34+
systemProperty("metadataConfig", "otel.semconv.exception.signal.preview=logs")
35+
}
36+
37+
check {
38+
dependsOn(testExceptionSignalLogs)
39+
}
2540
}
2641

2742
// kafka 4.1 requires java 11

‎instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/WrapperTest.java‎

Lines changed: 18 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,9 @@
77

88
import static io.opentelemetry.api.common.AttributeKey.longKey;
99
import static io.opentelemetry.api.common.AttributeKey.stringKey;
10+
import static io.opentelemetry.instrumentation.api.internal.SemconvExceptionSignal.emitExceptionAsLogs;
11+
import static io.opentelemetry.instrumentation.api.internal.SemconvExceptionSignal.emitExceptionAsSpanEvents;
12+
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat;
1013
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
1114
import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
1215
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_BATCH_MESSAGE_COUNT;
@@ -25,6 +28,7 @@
2528
import static org.mockito.Mockito.mock;
2629
import static org.mockito.Mockito.when;
2730

31+
import io.opentelemetry.api.logs.Severity;
2832
import io.opentelemetry.api.trace.SpanContext;
2933
import io.opentelemetry.api.trace.SpanKind;
3034
import io.opentelemetry.instrumentation.testing.junit.message.MessageHeaderUtil;
@@ -169,10 +173,10 @@ void testConsumerError() {
169173
KafkaTelemetry telemetry = telemetryBuilder.build();
170174

171175
Consumer<?, ?> mockConsumer = mock();
172-
when(mockConsumer.poll(Duration.ofSeconds(10))).thenThrow(new IllegalStateException());
176+
IllegalStateException error = new IllegalStateException("test");
177+
when(mockConsumer.poll(Duration.ofSeconds(10))).thenThrow(error);
173178
Consumer<?, ?> wrappedConsumer = telemetry.wrap(mockConsumer);
174-
assertThatThrownBy(() -> wrappedConsumer.poll(Duration.ofSeconds(10)))
175-
.isInstanceOf(IllegalStateException.class);
179+
assertThatThrownBy(() -> wrappedConsumer.poll(Duration.ofSeconds(10))).isSameAs(error);
176180

177181
testing.waitAndAssertTraces(
178182
trace ->
@@ -182,10 +186,20 @@ void testConsumerError() {
182186
.hasKind(SpanKind.CONSUMER)
183187
.hasNoParent()
184188
.hasStatus(StatusData.error())
185-
.hasException(new IllegalStateException())
189+
.hasException(emitExceptionAsSpanEvents() ? error : null)
186190
.hasAttributesSatisfyingExactly(
187191
equalTo(MESSAGING_SYSTEM, "kafka"),
188192
equalTo(MESSAGING_OPERATION, "receive"),
189193
equalTo(MESSAGING_BATCH_MESSAGE_COUNT, 0))));
194+
195+
if (emitExceptionAsLogs()) {
196+
testing.waitAndAssertLogRecords(
197+
logRecord ->
198+
logRecord
199+
.hasSeverity(Severity.WARN)
200+
.hasEventName("messaging.receive.exception")
201+
.hasException(error)
202+
.hasTotalAttributeCount(3));
203+
}
190204
}
191205
}

‎instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/KafkaInstrumenterFactory.java‎

Lines changed: 32 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,9 @@
55

66
package io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal;
77

8+
import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal.MessagingExceptionEventExtractors.setMessagingProcessExceptionEventExtractor;
9+
import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal.MessagingExceptionEventExtractors.setMessagingReceiveExceptionEventExtractor;
10+
import static io.opentelemetry.instrumentation.api.incubator.semconv.messaging.internal.MessagingExceptionEventExtractors.setMessagingSendExceptionEventExtractor;
811
import static java.util.Collections.emptyList;
912

1013
import com.google.errorprone.annotations.CanIgnoreReturnValue;
@@ -91,6 +94,7 @@ public Instrumenter<KafkaProducerRequest, RecordMetadata> createProducerInstrume
9194
if (captureExperimentalSpanAttributes) {
9295
builder.addAttributesExtractor(new KafkaProducerExperimentalAttributesExtractor());
9396
}
97+
setMessagingSendExceptionEventExtractor(builder);
9498
return builder.buildInstrumenter(SpanKindExtractor.alwaysProducer());
9599
}
96100

@@ -103,17 +107,19 @@ public Instrumenter<KafkaReceiveRequest, Void> createConsumerReceiveInstrumenter
103107
KafkaReceiveAttributesGetter getter = new KafkaReceiveAttributesGetter();
104108
MessageOperation operation = MessageOperation.RECEIVE;
105109

106-
return Instrumenter.<KafkaReceiveRequest, Void>builder(
107-
openTelemetry,
108-
instrumentationName,
109-
MessagingSpanNameExtractor.create(getter, operation))
110-
.addAttributesExtractor(
111-
buildMessagingAttributesExtractor(getter, operation, capturedHeaders))
112-
.addAttributesExtractor(new KafkaReceiveAttributesExtractor())
113-
.addAttributesExtractors(extractors)
114-
.setErrorCauseExtractor(errorCauseExtractor)
115-
.setEnabled(messagingReceiveInstrumentationEnabled)
116-
.buildInstrumenter(SpanKindExtractor.alwaysConsumer());
110+
InstrumenterBuilder<KafkaReceiveRequest, Void> builder =
111+
Instrumenter.<KafkaReceiveRequest, Void>builder(
112+
openTelemetry,
113+
instrumentationName,
114+
MessagingSpanNameExtractor.create(getter, operation))
115+
.addAttributesExtractor(
116+
buildMessagingAttributesExtractor(getter, operation, capturedHeaders))
117+
.addAttributesExtractor(new KafkaReceiveAttributesExtractor())
118+
.addAttributesExtractors(extractors)
119+
.setErrorCauseExtractor(errorCauseExtractor)
120+
.setEnabled(messagingReceiveInstrumentationEnabled);
121+
setMessagingReceiveExceptionEventExtractor(builder);
122+
return builder.buildInstrumenter(SpanKindExtractor.alwaysConsumer());
117123
}
118124

119125
public Instrumenter<KafkaProcessRequest, Void> createConsumerProcessInstrumenter() {
@@ -138,6 +144,7 @@ public Instrumenter<KafkaProcessRequest, Void> createConsumerProcessInstrumenter
138144
if (captureExperimentalSpanAttributes) {
139145
builder.addAttributesExtractor(new KafkaConsumerExperimentalAttributesExtractor());
140146
}
147+
setMessagingProcessExceptionEventExtractor(builder);
141148

142149
if (messagingReceiveInstrumentationEnabled) {
143150
builder.addSpanLinksExtractor(
@@ -154,18 +161,20 @@ public Instrumenter<KafkaReceiveRequest, Void> createBatchProcessInstrumenter()
154161
KafkaReceiveAttributesGetter getter = new KafkaReceiveAttributesGetter();
155162
MessageOperation operation = MessageOperation.PROCESS;
156163

157-
return Instrumenter.<KafkaReceiveRequest, Void>builder(
158-
openTelemetry,
159-
instrumentationName,
160-
MessagingSpanNameExtractor.create(getter, operation))
161-
.addAttributesExtractor(
162-
buildMessagingAttributesExtractor(getter, operation, capturedHeaders))
163-
.addAttributesExtractor(new KafkaReceiveAttributesExtractor())
164-
.addSpanLinksExtractor(
165-
new KafkaBatchProcessSpanLinksExtractor(
166-
openTelemetry.getPropagators().getTextMapPropagator()))
167-
.setErrorCauseExtractor(errorCauseExtractor)
168-
.buildInstrumenter(SpanKindExtractor.alwaysConsumer());
164+
InstrumenterBuilder<KafkaReceiveRequest, Void> builder =
165+
Instrumenter.<KafkaReceiveRequest, Void>builder(
166+
openTelemetry,
167+
instrumentationName,
168+
MessagingSpanNameExtractor.create(getter, operation))
169+
.addAttributesExtractor(
170+
buildMessagingAttributesExtractor(getter, operation, capturedHeaders))
171+
.addAttributesExtractor(new KafkaReceiveAttributesExtractor())
172+
.addSpanLinksExtractor(
173+
new KafkaBatchProcessSpanLinksExtractor(
174+
openTelemetry.getPropagators().getTextMapPropagator()))
175+
.setErrorCauseExtractor(errorCauseExtractor);
176+
setMessagingProcessExceptionEventExtractor(builder);
177+
return builder.buildInstrumenter(SpanKindExtractor.alwaysConsumer());
169178
}
170179

171180
private static <REQUEST, RESPONSE>

0 commit comments

Comments
 (0)