Skip to content

Commit ec007eb

Browse files
CodePrometheusCodePrometheus
authored
Adds RocketMQ plugin (#1449)
Fixes 1043
1 parent 06b47b1 commit ec007eb

File tree

20 files changed

+1077
-0
lines changed

20 files changed

+1077
-0
lines changed

‎brave-bom/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -252,6 +252,11 @@
252252
<artifactId>brave-spring-beans</artifactId>
253253
<version>${project.version}</version>
254254
</dependency>
255+
<dependency>
256+
<groupId>${project.groupId}</groupId>
257+
<artifactId>brave-instrumentation-rocketmq-client</artifactId>
258+
<version>${project.version}</version>
259+
</dependency>
255260
</dependencies>
256261
</dependencyManagement>
257262

‎instrumentation/README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ Here's a brief overview of what's packaged here:
2525
* [spring-web](spring-web/README.md) - Tracing interceptor for [Spring RestTemplate](https://spring.io/guides/gs/consuming-rest/)
2626
* [spring-webmvc](spring-webmvc/README.md) - Tracing filter and span customizing interceptors for [Spring WebMVC](https://docs.spring.io/spring/docs/current/spring-framework-reference/html/mvc.html)
2727
* [vertx-web](vertx-web/README.md) - Tracing routing context handler for [Vert.x Web](http://vertx.io/docs/vertx-web/js/)
28+
* [rocketmq-client](rocketmq-client/README.md) - Tracing Producer, MessageListenerConcurrently and MessageListenerOrderly for [Apache RocketMQ](https://github.com/apache/rocketmq/)
2829

2930
Here are other tools we provide for configuring or testing instrumentation:
3031
* [http](http/README.md) - `HttpTracing` that allows portable configuration of HTTP instrumentation

‎instrumentation/benchmarks/pom.xml‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -240,6 +240,18 @@
240240
<version>${project.version}</version>
241241
<scope>test</scope>
242242
</dependency>
243+
<dependency>
244+
<groupId>${project.groupId}</groupId>
245+
<artifactId>brave-instrumentation-rocketmq-client</artifactId>
246+
<version>${project.version}</version>
247+
<scope>test</scope>
248+
</dependency>
249+
<dependency>
250+
<groupId>org.apache.rocketmq</groupId>
251+
<artifactId>rocketmq-client</artifactId>
252+
<version>${rocketmq-client.version}</version>
253+
<scope>test</scope>
254+
</dependency>
243255
</dependencies>
244256

245257
<build>

‎instrumentation/benchmarks/src/test/java/brave/rocketmq/client/RocketMQProducerBenchmarks.java‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
/*
2+
* Copyright The OpenZipkin Authors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
package brave.rocketmq.client;
6+
7+
import brave.Tracing;
8+
import brave.kafka.clients.TracingProducerBenchmarks;
9+
import org.apache.rocketmq.client.producer.DefaultMQProducer;
10+
import org.apache.rocketmq.client.producer.SendResult;
11+
import org.apache.rocketmq.common.message.Message;
12+
import org.openjdk.jmh.annotations.Benchmark;
13+
import org.openjdk.jmh.annotations.BenchmarkMode;
14+
import org.openjdk.jmh.annotations.Fork;
15+
import org.openjdk.jmh.annotations.Level;
16+
import org.openjdk.jmh.annotations.Measurement;
17+
import org.openjdk.jmh.annotations.Mode;
18+
import org.openjdk.jmh.annotations.OutputTimeUnit;
19+
import org.openjdk.jmh.annotations.Scope;
20+
import org.openjdk.jmh.annotations.Setup;
21+
import org.openjdk.jmh.annotations.State;
22+
import org.openjdk.jmh.annotations.TearDown;
23+
import org.openjdk.jmh.annotations.Warmup;
24+
import org.openjdk.jmh.runner.Runner;
25+
import org.openjdk.jmh.runner.RunnerException;
26+
import org.openjdk.jmh.runner.options.Options;
27+
import org.openjdk.jmh.runner.options.OptionsBuilder;
28+
29+
import java.util.concurrent.TimeUnit;
30+
31+
import static org.apache.rocketmq.client.producer.SendStatus.SEND_OK;
32+
33+
@Measurement(iterations = 5, time = 1)
34+
@Warmup(iterations = 10, time = 1)
35+
@Fork(3)
36+
@BenchmarkMode(Mode.SampleTime)
37+
@OutputTimeUnit(TimeUnit.MICROSECONDS)
38+
@State(Scope.Thread)
39+
public class RocketMQProducerBenchmarks {
40+
Message message;
41+
DefaultMQProducer producer, tracingProducer;
42+
43+
@Setup(Level.Trial) public void init() {
44+
message = new Message("zipkin", "zipkin".getBytes());
45+
Tracing tracing = Tracing.newBuilder().build();
46+
producer = new FakeProducer();
47+
tracingProducer = new FakeProducer();
48+
tracingProducer.getDefaultMQProducerImpl().registerSendMessageHook(
49+
new TracingSendMessageHook(RocketMQTracing.newBuilder(tracing).build())
50+
);
51+
}
52+
53+
@TearDown(Level.Trial) public void close() {
54+
Tracing.current().close();
55+
}
56+
57+
@Benchmark public SendResult send_baseCase() throws Exception {
58+
return producer.send(message);
59+
}
60+
61+
@Benchmark public void send_traced() throws Exception {
62+
tracingProducer.send(message);
63+
}
64+
65+
// Convenience main entry-point
66+
public static void main(String[] args) throws RunnerException {
67+
Options opt = new OptionsBuilder()
68+
.addProfiler("gc")
69+
.include(".*" + TracingProducerBenchmarks.class.getSimpleName())
70+
.build();
71+
72+
new Runner(opt).run();
73+
}
74+
75+
static final class FakeProducer extends DefaultMQProducer {
76+
@Override public SendResult send(Message msg) {
77+
SendResult sendResult = new SendResult();
78+
sendResult.setMsgId("zipkin");
79+
sendResult.setSendStatus(SEND_OK);
80+
return sendResult;
81+
}
82+
}
83+
}

‎instrumentation/pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@
5252
<module>kafka-streams</module>
5353
<module>netty-codec-http</module>
5454
<module>vertx-web</module>
55+
<module>rocketmq-client</module>
5556
</modules>
5657

5758
<!-- ${project.groupId}:brave version is set in the root pom.

‎instrumentation/rocketmq-client/README.md‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
# Brave RocketMQ Client instrumentation
2+
This module provides instrumentation for rocketmq-client 5.x+ consumers and
3+
producers.
4+
5+
## Setup
6+
Setup the generic RocketMQ component like this:
7+
```java
8+
rocketmqTracing = RocketMQTracing.newBuilder(messagingTracing)
9+
.remoteServiceName("my-broker")
10+
.build();
11+
```
12+
13+
## Sampling Policy
14+
The default sampling policy is to use the default (trace ID) sampler for
15+
producer and consumer requests.
16+
17+
You can use an [MessagingRuleSampler](../messaging/README.md) to override this
18+
based on RocketMQ topic names.
19+
20+
Ex. Here's a sampler that traces 100 consumer requests per second, except for
21+
the "alerts" topic. Other requests will use a global rate provided by the
22+
`Tracing` component.
23+
24+
```java
25+
import brave.sampler.Matchers;
26+
27+
import static brave.messaging.MessagingRequestMatchers.channelNameEquals;
28+
29+
messagingTracingBuilder.consumerSampler(MessagingRuleSampler.newBuilder()
30+
.putRule(channelNameEquals("alerts"), Sampler.NEVER_SAMPLE)
31+
.putRule(Matchers.alwaysMatch(), RateLimitingSampler.create(100))
32+
.build());
33+
34+
rocketmqTracing = RocketMQTracing.create(messagingTracing);
35+
```
36+
37+
## Producer
38+
39+
Register `brave.rocketmq.client.RocketMQTracing.newSendMessageHook()` to trace the message.
40+
41+
```java
42+
Message message = new Message("zipkin", "zipkin", "zipkin".getBytes());
43+
DefaultMQProducer producer = new DefaultMQProducer("testSend");
44+
producer.getDefaultMQProducerImpl()
45+
.registerSendMessageHook(producerTracing.newSendMessageHook());
46+
producer.setNamesrvAddr("127.0.0.1:9876");
47+
producer.start();
48+
producer.send(message);
49+
50+
producer.shutdown();
51+
```
52+
53+
## Consumer
54+
55+
Wrap `org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly`
56+
using `brave.rocketmq.client.RocketMQTracing.messageListenerOrderly(org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly)`,
57+
or alternatively, wrap `org.apache.rocketmq.client.consumer.listener.messageListenerConcurrently`
58+
using `brave.rocketmq.client.RocketMQTracing.messageListenerConcurrently(org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently)`;
59+
60+
```java
61+
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("testPushConsumer");
62+
consumer.setNamesrvAddr("127.0.0.1:9876");
63+
consumer.subscribe("zipkin", "*");
64+
MessageListenerConcurrently messageListenerConcurrently = rocketMQTracing.messageListenerConcurrently(
65+
new MessageListenerConcurrently() {
66+
@Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
67+
// do something
68+
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
69+
}
70+
});
71+
consumer.registerMessageListener(messageListenerConcurrently);
72+
73+
consumer.start();
74+
```

‎instrumentation/rocketmq-client/bnd.bnd‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
# We use brave.internal.Nullable, but it is not used at runtime.
2+
Import-Package: \
3+
!brave.internal*,\
4+
*
5+
Export-Package: \
6+
brave.rocketmq.client

‎instrumentation/rocketmq-client/pom.xml‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
<?xml version="1.0"?>
2+
<!--
3+
4+
Copyright The OpenZipkin Authors
5+
SPDX-License-Identifier: Apache-2.0
6+
7+
-->
8+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
9+
<modelVersion>4.0.0</modelVersion>
10+
<parent>
11+
<groupId>io.zipkin.brave</groupId>
12+
<artifactId>brave-instrumentation-parent</artifactId>
13+
<version>6.0.4-SNAPSHOT</version>
14+
</parent>
15+
16+
<artifactId>brave-instrumentation-rocketmq-client</artifactId>
17+
<name>Brave Instrumentation: RocketMQ Client</name>
18+
19+
<properties>
20+
<!-- Matches Export-Package in bnd.bnd -->
21+
<module.name>brave.rocketmq.client</module.name>
22+
23+
<main.basedir>${project.basedir}/../..</main.basedir>
24+
</properties>
25+
26+
<dependencies>
27+
<dependency>
28+
<groupId>${project.groupId}</groupId>
29+
<artifactId>brave-instrumentation-messaging</artifactId>
30+
<version>${project.version}</version>
31+
</dependency>
32+
<dependency>
33+
<groupId>org.apache.rocketmq</groupId>
34+
<artifactId>rocketmq-client</artifactId>
35+
<version>${rocketmq-client.version}</version>
36+
<scope>provided</scope>
37+
</dependency>
38+
<dependency>
39+
<groupId>${project.groupId}</groupId>
40+
<artifactId>brave-tests</artifactId>
41+
<scope>test</scope>
42+
<version>${project.version}</version>
43+
</dependency>
44+
<dependency>
45+
<groupId>org.testcontainers</groupId>
46+
<artifactId>junit-jupiter</artifactId>
47+
<version>${testcontainers.version}</version>
48+
<scope>test</scope>
49+
</dependency>
50+
</dependencies>
51+
</project>

‎instrumentation/rocketmq-client/src/main/java/brave/rocketmq/client/AbstractMessageListener.java‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
/*
2+
* Copyright The OpenZipkin Authors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
package brave.rocketmq.client;
6+
7+
import brave.Span;
8+
import brave.Tracer;
9+
import brave.Tracing;
10+
import brave.internal.Nullable;
11+
import brave.messaging.MessagingRequest;
12+
import brave.propagation.TraceContext;
13+
import brave.propagation.TraceContextOrSamplingFlags;
14+
import brave.sampler.SamplerFunction;
15+
import org.apache.rocketmq.common.message.MessageExt;
16+
17+
import java.util.List;
18+
import java.util.function.BiFunction;
19+
import java.util.function.Function;
20+
21+
import static brave.Span.Kind.CONSUMER;
22+
import static brave.internal.Throwables.propagateIfFatal;
23+
import static brave.rocketmq.client.RocketMQTracing.ROCKETMQ_TOPIC;
24+
25+
/**
26+
* Read records headers to create and complete a child of the incoming
27+
* producers span if possible.
28+
* The spans are modeled as a duration 1 {@link Span.Kind#CONSUMER} span to represent consuming the
29+
* message from the rocketmq broker with a child span representing the processing of the message.
30+
*/
31+
abstract class AbstractMessageListener {
32+
final RocketMQTracing rocketMQTracing;
33+
final Tracing tracing;
34+
final Tracer tracer;
35+
final TraceContext.Extractor<MessageConsumerRequest> extractor;
36+
final SamplerFunction<MessagingRequest> sampler;
37+
@Nullable final String remoteServiceName;
38+
39+
AbstractMessageListener(RocketMQTracing rocketMQTracing) {
40+
this.rocketMQTracing = rocketMQTracing;
41+
this.tracing = rocketMQTracing.messagingTracing.tracing();
42+
this.tracer = tracing.tracer();
43+
this.extractor = rocketMQTracing.consumerExtractor;
44+
this.sampler = rocketMQTracing.consumerSampler;
45+
this.remoteServiceName = rocketMQTracing.remoteServiceName;
46+
}
47+
48+
<T> T processConsumeMessage(
49+
List<MessageExt> msgs,
50+
Function<List<MessageExt>, T> consumerFunc,
51+
BiFunction<T, T, Boolean> successFunc,
52+
T successStatus
53+
) {
54+
for (MessageExt message : msgs) {
55+
MessageConsumerRequest request = new MessageConsumerRequest(message);
56+
TraceContextOrSamplingFlags extracted =
57+
rocketMQTracing.extractAndClearTraceIdHeaders(extractor, request, message.getProperties());
58+
Span consumerSpan = rocketMQTracing.nextMessagingSpan(sampler, request, extracted);
59+
Span listenerSpan = tracer.newChild(consumerSpan.context());
60+
61+
if (!consumerSpan.isNoop()) {
62+
setConsumerSpan(consumerSpan, message.getTopic());
63+
// incur timestamp overhead only once
64+
long timestamp = tracing.clock(consumerSpan.context()).currentTimeMicroseconds();
65+
consumerSpan.start(timestamp);
66+
long consumerFinish = timestamp + 1L; // save a clock reading
67+
consumerSpan.finish(consumerFinish);
68+
// not using scoped span as we want to start with a pre-configured time
69+
listenerSpan.name("on-message").start(consumerFinish);
70+
}
71+
72+
Tracer.SpanInScope scope = tracer.withSpanInScope(listenerSpan);
73+
Throwable error = null;
74+
T result;
75+
76+
try {
77+
result = consumerFunc.apply(msgs);
78+
} catch (Throwable t) {
79+
propagateIfFatal(t);
80+
error = t;
81+
throw t;
82+
} finally {
83+
if (error != null) listenerSpan.error(error);
84+
listenerSpan.finish();
85+
scope.close();
86+
}
87+
88+
if (!successFunc.apply(result, successStatus)) {
89+
return result;
90+
}
91+
}
92+
return successStatus;
93+
}
94+
95+
void setConsumerSpan(Span span, String topic) {
96+
span.name("receive").kind(CONSUMER);
97+
span.tag(ROCKETMQ_TOPIC, topic);
98+
if (remoteServiceName != null) span.remoteServiceName(remoteServiceName);
99+
}
100+
}

0 commit comments

Comments
 (0)