Skip to content

Commit 70287df

Browse files
trasktrask
committed
Add unit and integration tests for messaging exception event extractors
1 parent 23a0233 commit 70287df

3 files changed

Lines changed: 159 additions & 10 deletions

File tree

‎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/kafka/kafka-clients/kafka-clients-2.6/library/build.gradle.kts‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,21 @@ tasks {
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+
usesService(gradle.sharedServices.registrations["testcontainersBuildService"].service)
30+
31+
filter {
32+
includeTestsMatching("WrapperTest")
33+
}
34+
jvmArgs("-Dotel.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: 44 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,13 @@
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;
15+
import static io.opentelemetry.semconv.ExceptionAttributes.EXCEPTION_STACKTRACE;
16+
import static io.opentelemetry.semconv.ExceptionAttributes.EXCEPTION_TYPE;
1217
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_BATCH_MESSAGE_COUNT;
1318
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
1419
import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_PARTITION_ID;
@@ -20,14 +25,17 @@
2025
import static java.nio.charset.StandardCharsets.UTF_8;
2126
import static java.util.Arrays.asList;
2227
import static java.util.Collections.singletonList;
28+
import static java.util.stream.Collectors.toList;
2329
import static org.assertj.core.api.Assertions.assertThat;
2430
import static org.assertj.core.api.Assertions.assertThatThrownBy;
2531
import static org.mockito.Mockito.mock;
2632
import static org.mockito.Mockito.when;
2733

34+
import io.opentelemetry.api.logs.Severity;
2835
import io.opentelemetry.api.trace.SpanContext;
2936
import io.opentelemetry.api.trace.SpanKind;
3037
import io.opentelemetry.instrumentation.testing.junit.message.MessageHeaderUtil;
38+
import io.opentelemetry.sdk.logs.data.LogRecordData;
3139
import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
3240
import io.opentelemetry.sdk.trace.data.LinkData;
3341
import io.opentelemetry.sdk.trace.data.StatusData;
@@ -38,6 +46,7 @@
3846
import org.apache.kafka.clients.consumer.Consumer;
3947
import org.assertj.core.api.AbstractLongAssert;
4048
import org.assertj.core.api.AbstractStringAssert;
49+
import org.awaitility.Awaitility;
4150
import org.junit.jupiter.api.Test;
4251

4352
@SuppressWarnings("deprecation") // using deprecated semconv
@@ -177,15 +186,40 @@ void testConsumerError() {
177186
testing.waitAndAssertTraces(
178187
trace ->
179188
trace.hasSpansSatisfyingExactly(
180-
span ->
181-
span.hasName("unknown receive")
182-
.hasKind(SpanKind.CONSUMER)
183-
.hasNoParent()
184-
.hasStatus(StatusData.error())
185-
.hasException(new IllegalStateException())
186-
.hasAttributesSatisfyingExactly(
187-
equalTo(MESSAGING_SYSTEM, "kafka"),
188-
equalTo(MESSAGING_OPERATION, "receive"),
189-
equalTo(MESSAGING_BATCH_MESSAGE_COUNT, 0))));
189+
span -> {
190+
span.hasName("unknown receive")
191+
.hasKind(SpanKind.CONSUMER)
192+
.hasNoParent()
193+
.hasStatus(StatusData.error())
194+
.hasAttributesSatisfyingExactly(
195+
equalTo(MESSAGING_SYSTEM, "kafka"),
196+
equalTo(MESSAGING_OPERATION, "receive"),
197+
equalTo(MESSAGING_BATCH_MESSAGE_COUNT, 0));
198+
if (emitExceptionAsSpanEvents()) {
199+
span.hasException(new IllegalStateException());
200+
}
201+
}));
202+
203+
if (emitExceptionAsLogs()) {
204+
assertReceiveExceptionLog();
205+
}
206+
}
207+
208+
private static void assertReceiveExceptionLog() {
209+
Awaitility.await()
210+
.untilAsserted(
211+
() -> {
212+
List<LogRecordData> logs =
213+
testing.logRecords().stream()
214+
.filter(log -> "messaging.receive.exception".equals(log.getEventName()))
215+
.collect(toList());
216+
assertThat(logs).hasSize(1);
217+
assertThat(logs.get(0))
218+
.hasSeverity(Severity.WARN)
219+
.hasEventName("messaging.receive.exception")
220+
.hasAttributesSatisfyingExactly(
221+
satisfies(EXCEPTION_TYPE, val -> val.isNotNull()),
222+
satisfies(EXCEPTION_STACKTRACE, val -> val.isNotNull()));
223+
});
190224
}
191225
}

0 commit comments

Comments
 (0)