Skip to content

Commit 6341a43

Browse files
atakavciatakavci
committed
tracing integration tests with async and reactive commands
1 parent 806103f commit 6341a43

2 files changed

Lines changed: 220 additions & 0 deletions

File tree

‎src/test/java/io/lettuce/core/tracing/AsynchronousIntegrationTests.java‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
package io.lettuce.core.tracing;
2+
3+
import static io.lettuce.TestTags.INTEGRATION_TEST;
4+
import static org.assertj.core.api.Assertions.*;
5+
6+
import java.lang.reflect.Method;
7+
import java.util.concurrent.LinkedBlockingQueue;
8+
9+
import io.lettuce.core.RedisClient;
10+
import io.lettuce.core.RedisURI;
11+
import io.lettuce.core.api.StatefulRedisConnection;
12+
import io.lettuce.core.protocol.CommandType;
13+
import io.lettuce.core.protocol.RedisCommand;
14+
import io.lettuce.core.resource.ClientResources;
15+
import io.lettuce.test.TestFutures;
16+
import io.lettuce.test.resource.FastShutdown;
17+
import io.lettuce.test.settings.TestSettings;
18+
import io.micrometer.core.instrument.MeterRegistry;
19+
import io.micrometer.observation.ObservationRegistry;
20+
import io.micrometer.tracing.exporter.FinishedSpan;
21+
import io.micrometer.tracing.test.SampleTestRunner;
22+
import org.junit.jupiter.api.Tag;
23+
24+
/**
25+
* Collection of tests that log metrics and tracing using the asynchronous API. This guards against regressions in the
26+
* Micrometer tracing integration for async command flows.
27+
*
28+
* @author Ali Takavci
29+
*/
30+
@Tag(INTEGRATION_TEST)
31+
public class AsynchronousIntegrationTests extends SampleTestRunner {
32+
33+
AsynchronousIntegrationTests() {
34+
super(SampleRunnerConfig.builder().build());
35+
}
36+
37+
@Override
38+
protected MeterRegistry createMeterRegistry() {
39+
return TestConfig.METER_REGISTRY;
40+
}
41+
42+
@Override
43+
protected ObservationRegistry createObservationRegistry() {
44+
return TestConfig.OBSERVATION_REGISTRY;
45+
}
46+
47+
public static void main(String[] args) throws Exception {
48+
AsynchronousIntegrationTests tests = new AsynchronousIntegrationTests();
49+
TracingSetup[] setups = tests.getTracingSetup();
50+
TracingSetup setup = setups[0];
51+
Class<AsynchronousIntegrationTests> c = AsynchronousIntegrationTests.class;
52+
Method m = c.getSuperclass().getDeclaredMethod("run", TracingSetup.class);
53+
m.setAccessible(true);
54+
tests.setupRegistry();
55+
m.invoke(tests, setup);
56+
}
57+
58+
@Override
59+
public SampleTestRunnerConsumer yourCode() {
60+
61+
LinkedBlockingQueue<RedisCommand<?, ?, ?>> commands = new LinkedBlockingQueue<>();
62+
ObservationRegistry observationRegistry = createObservationRegistry();
63+
observationRegistry.observationConfig().observationPredicate((s, context) -> {
64+
65+
if (context instanceof LettuceObservationContext) {
66+
commands.add(((LettuceObservationContext) context).getRequiredCommand());
67+
}
68+
69+
return true;
70+
});
71+
ClientResources clientResources = ClientResources.builder()
72+
.tracing(new MicrometerTracing(observationRegistry, "Redis", true)).build();
73+
74+
return (tracer, meterRegistry) -> {
75+
76+
RedisURI redisURI = RedisURI.create(TestSettings.host(), TestSettings.port());
77+
RedisClient redisClient = RedisClient.create(clientResources, redisURI);
78+
StatefulRedisConnection<String, String> connection = redisClient.connect();
79+
80+
TestFutures.getOrTimeout(connection.async().ping());
81+
82+
connection.close();
83+
FastShutdown.shutdown(redisClient);
84+
FastShutdown.shutdown(clientResources);
85+
86+
assertThat(tracer.getFinishedSpans()).isNotEmpty();
87+
88+
for (FinishedSpan finishedSpan : tracer.getFinishedSpans()) {
89+
assertThat(finishedSpan.getTags()).containsEntry("db.system", "redis")
90+
.containsEntry("net.sock.peer.addr", TestSettings.host())
91+
.containsEntry("net.sock.peer.port", "" + TestSettings.port());
92+
assertThat(finishedSpan.getTags()).containsKeys("db.operation");
93+
}
94+
95+
assertThat(commands).extracting(RedisCommand::getType).contains(CommandType.PING, CommandType.HELLO);
96+
};
97+
}
98+
99+
}

‎src/test/java/io/lettuce/core/tracing/ReactiveIntegrationTests.java‎

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
package io.lettuce.core.tracing;
2+
3+
import static io.lettuce.TestTags.INTEGRATION_TEST;
4+
import static org.assertj.core.api.Assertions.*;
5+
6+
import java.lang.reflect.Method;
7+
import java.util.concurrent.LinkedBlockingQueue;
8+
9+
import io.lettuce.core.RedisClient;
10+
import io.lettuce.core.RedisURI;
11+
import io.lettuce.core.api.StatefulRedisConnection;
12+
import io.lettuce.core.protocol.CommandType;
13+
import io.lettuce.core.protocol.RedisCommand;
14+
import io.lettuce.core.resource.ClientResources;
15+
import io.lettuce.test.resource.FastShutdown;
16+
import io.lettuce.test.settings.TestSettings;
17+
import io.micrometer.core.instrument.MeterRegistry;
18+
import io.micrometer.observation.Observation;
19+
import io.micrometer.observation.ObservationRegistry;
20+
import io.micrometer.tracing.exporter.FinishedSpan;
21+
import io.micrometer.tracing.test.SampleTestRunner;
22+
import org.junit.jupiter.api.Tag;
23+
import reactor.test.StepVerifier;
24+
25+
/**
26+
* Collection of tests that log metrics and tracing using the reactive API. This guards against regressions in the
27+
* {@link MicrometerTracing.MicrometerTraceContextProvider#getTraceContextAsync} path used by
28+
* {@link io.lettuce.core.AbstractRedisReactiveCommands#withTraceContext()}.
29+
* <p>
30+
* The {@link SampleTestRunner} framework manages an {@link Observation} on the ThreadLocal. Since {@code getTraceContextAsync}
31+
* deliberately does not fall back to ThreadLocal when a Reactor context map is provided, this test propagates the current
32+
* {@link Observation} into the Reactor context via {@code contextWrite} using the {@link Observation Observation.class} key,
33+
* then verifies that the resulting Lettuce spans are children of the propagated observation (i.e. share its trace id).
34+
*
35+
* @author Ali Takavci
36+
*/
37+
@Tag(INTEGRATION_TEST)
38+
public class ReactiveIntegrationTests extends SampleTestRunner {
39+
40+
ReactiveIntegrationTests() {
41+
super(SampleRunnerConfig.builder().build());
42+
}
43+
44+
@Override
45+
protected MeterRegistry createMeterRegistry() {
46+
return TestConfig.METER_REGISTRY;
47+
}
48+
49+
@Override
50+
protected ObservationRegistry createObservationRegistry() {
51+
return TestConfig.OBSERVATION_REGISTRY;
52+
}
53+
54+
public static void main(String[] args) throws Exception {
55+
ReactiveIntegrationTests tests = new ReactiveIntegrationTests();
56+
TracingSetup[] setups = tests.getTracingSetup();
57+
TracingSetup setup = setups[0];
58+
Class<ReactiveIntegrationTests> c = ReactiveIntegrationTests.class;
59+
Method m = c.getSuperclass().getDeclaredMethod("run", TracingSetup.class);
60+
m.setAccessible(true);
61+
tests.setupRegistry();
62+
m.invoke(tests, setup);
63+
}
64+
65+
@Override
66+
public SampleTestRunnerConsumer yourCode() {
67+
68+
LinkedBlockingQueue<RedisCommand<?, ?, ?>> commands = new LinkedBlockingQueue<>();
69+
ObservationRegistry observationRegistry = createObservationRegistry();
70+
observationRegistry.observationConfig().observationPredicate((s, context) -> {
71+
72+
if (context instanceof LettuceObservationContext) {
73+
commands.add(((LettuceObservationContext) context).getRequiredCommand());
74+
}
75+
76+
return true;
77+
});
78+
ClientResources clientResources = ClientResources.builder()
79+
.tracing(new MicrometerTracing(observationRegistry, "Redis", true)).build();
80+
81+
return (tracer, meterRegistry) -> {
82+
83+
RedisURI redisURI = RedisURI.create(TestSettings.host(), TestSettings.port());
84+
RedisClient redisClient = RedisClient.create(clientResources, redisURI);
85+
StatefulRedisConnection<String, String> connection = redisClient.connect();
86+
87+
// Grab the trace id of the SampleTestRunner's current span so we can verify
88+
// that Lettuce spans become children of this trace.
89+
io.micrometer.tracing.TraceContext c = tracer.getTracer().currentSpan().context();
90+
String ctxTraceId = c.traceId();
91+
String ctxSpanId = c.spanId();
92+
93+
// Propagate the Observation via the Observation.class key.
94+
// This exercises the Observation.class branch in getTraceContextAsync.
95+
connection.reactive().ping()
96+
.contextWrite(ctx -> ctx.put(Observation.class, observationRegistry.getCurrentObservation()))
97+
.as(StepVerifier::create).expectNext("PONG").verifyComplete();
98+
99+
connection.close();
100+
FastShutdown.shutdown(redisClient);
101+
FastShutdown.shutdown(clientResources);
102+
103+
assertThat(tracer.getFinishedSpans()).isNotEmpty();
104+
105+
for (FinishedSpan finishedSpan : tracer.getFinishedSpans()) {
106+
assertThat(finishedSpan.getTags()).containsEntry("db.system", "redis")
107+
.containsEntry("net.sock.peer.addr", TestSettings.host())
108+
.containsEntry("net.sock.peer.port", "" + TestSettings.port());
109+
assertThat(finishedSpan.getTags()).containsKeys("db.operation");
110+
}
111+
112+
assertThat(tracer.getFinishedSpans()).anySatisfy(span -> {
113+
assertThat(span.getTraceId()).isEqualTo(ctxTraceId);
114+
assertThat(span.getParentId()).isEqualTo(ctxSpanId);
115+
});
116+
117+
assertThat(commands).extracting(RedisCommand::getType).contains(CommandType.PING, CommandType.HELLO);
118+
};
119+
}
120+
121+
}

0 commit comments

Comments
 (0)