diff --git a/apm-agent-core/src/test/resources/json-specs/span_types.json b/apm-agent-core/src/test/resources/json-specs/span_types.json index b60c3730d2..fe70d21487 100644 --- a/apm-agent-core/src/test/resources/json-specs/span_types.json +++ b/apm-agent-core/src/test/resources/json-specs/span_types.json @@ -274,6 +274,12 @@ "__used_by": [ "ruby" ] + }, + "rocketmq": { + "__description": "Apache RocketMQ", + "__used_by": [ + "java" + ] } } }, diff --git a/apm-agent-plugins/apm-rocketmq-plugin/pom.xml b/apm-agent-plugins/apm-rocketmq-plugin/pom.xml new file mode 100644 index 0000000000..3cec0419b5 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/pom.xml @@ -0,0 +1,58 @@ + + + + apm-agent-plugins + co.elastic.apm + 1.26.1-SNAPSHOT + + 4.0.0 + + apm-rocketmq-plugin + + + 4.2.0 + ${project.basedir}/../.. + + + + + org.apache.rocketmq + rocketmq-client + ${rocketmq.client.version} + provided + + + org.apache.rocketmq + rocketmq-broker + 4.2.0 + test + + + logback-classic + ch.qos.logback + + + logback-core + ch.qos.logback + + + + + org.apache.rocketmq + rocketmq-namesrv + 4.2.0 + + + logback-classic + ch.qos.logback + + + logback-core + ch.qos.logback + + + + + diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/BaseRocketmqInstrumentation.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/BaseRocketmqInstrumentation.java new file mode 100644 index 0000000000..d9237adfdd --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/BaseRocketmqInstrumentation.java @@ -0,0 +1,32 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq; + +import co.elastic.apm.agent.bci.TracerAwareInstrumentation; + +import java.util.Collection; +import java.util.Collections; + +public abstract class BaseRocketmqInstrumentation extends TracerAwareInstrumentation { + + @Override + public Collection getInstrumentationGroupNames() { + return Collections.singletonList("rocketmq"); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/Labels.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/Labels.java new file mode 100644 index 0000000000..17a3d8b7f0 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/Labels.java @@ -0,0 +1,30 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq; + +public class Labels { + + public static final String KEYS = "rocketmq.keys"; + + public static final String MSG_ID = "rocketmq.msgId"; + + public static final String NAME_SRV = "rocketmq.namesrv"; + + public static final String SEND_STATUS = "rocketmq.sendStatus"; +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/ConcurrentListenerInstrumentation.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/ConcurrentListenerInstrumentation.java new file mode 100644 index 0000000000..b5eb0adbb4 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/ConcurrentListenerInstrumentation.java @@ -0,0 +1,73 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.consumer; + +import co.elastic.apm.agent.impl.transaction.Transaction; +import co.elastic.apm.agent.rocketmq.BaseRocketmqInstrumentation; +import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import org.apache.rocketmq.common.message.MessageExt; + +import java.util.List; + +import static net.bytebuddy.matcher.ElementMatchers.*; + +public class ConcurrentListenerInstrumentation extends BaseRocketmqInstrumentation { + + @Override + public ElementMatcher getTypeMatcher() { + return hasSuperType(named("org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently")) + .and(not(isInterface())) + .and(not(isAbstract())); + } + + @Override + public ElementMatcher getMethodMatcher() { + return named("consumeMessage"); + } + + public static class AdviceClass { + + @Advice.OnMethodEnter(inline = false, suppress = Throwable.class) + public static Object onEnter(@Advice.Argument(0) List msgList, + @Advice.This Object thiz) { + return ConsumerHelper.get().onConsumeStart(thiz, msgList); + } + + @Advice.OnMethodExit(inline = false, suppress = Throwable.class, onThrowable = Throwable.class) + public static void onExit(@Advice.Thrown Throwable thrown, + @Advice.Return ConsumeConcurrentlyStatus status, + @Advice.Enter Object transactionObj) { + + if (transactionObj == null) { + return; + } + Transaction transaction = (Transaction) transactionObj; + ConsumerHelper helper = ConsumerHelper.get(); + if (thrown != null) { + helper.onConsumeError(transaction, thrown); + } else { + helper.onConsumeEnd(transaction, status.name(), status == ConsumeConcurrentlyStatus.CONSUME_SUCCESS); + } + } + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/ConsumerHelper.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/ConsumerHelper.java new file mode 100644 index 0000000000..468c61a17d --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/ConsumerHelper.java @@ -0,0 +1,80 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.consumer; + +import co.elastic.apm.agent.impl.ElasticApmTracer; +import co.elastic.apm.agent.impl.GlobalTracer; +import co.elastic.apm.agent.impl.transaction.Outcome; +import co.elastic.apm.agent.impl.transaction.Transaction; +import co.elastic.apm.agent.rocketmq.Labels; +import co.elastic.apm.agent.rocketmq.namesrv.NameSrvHolder; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.common.message.MessageExt; + +import java.util.List; + +public class ConsumerHelper { + + private static final ConsumerHelper INSTANCE = new ConsumerHelper(); + + private final ElasticApmTracer tracer = GlobalTracer.requireTracerImpl(); + + public static ConsumerHelper get() { + return INSTANCE; + } + + public Transaction onConsumeStart(Object listener, List msgList) { + if (tracer.getActive() != null) { + return null; + } + + Transaction transaction = tracer.startRootTransaction(MessageExt.class.getClassLoader()); + if (transaction == null) { + return null; + } + + // msgList.size() equals to DefaultMQPushConsumer.consumeMessageBatchMaxSize + // By default, DefaultMQPushConsumer.consumeMessageBatchMaxSize = 1, and rarely changed + MessageExt msg = msgList.get(0); + transaction.withType("messaging") + .withName("RocketMQ Consume ") + .appendToName(msg.getTopic()).appendToName("#").appendToName(msg.getTags()) + .addLabel(Labels.MSG_ID, msg.getMsgId()); + if (msg.getKeys() != null) { + transaction.addLabel(Labels.KEYS, msg.getKeys()); + } + + DefaultMQPushConsumer consumer = PushConsumerHolder.get(listener); + if (consumer != null) { + transaction.addLabel(Labels.NAME_SRV, NameSrvHolder.get(consumer, consumer, PushConsumerNameSrvGetter.get())); + } + + return transaction.activate(); + } + + public void onConsumeEnd(Transaction transaction, String consumeStatus, boolean isSuccess) { + transaction.withResult(consumeStatus) + .withOutcome(isSuccess ? Outcome.SUCCESS : Outcome.FAILURE) + .deactivate().end(); + } + + public void onConsumeError(Transaction transaction, Throwable thrown) { + transaction.captureException(thrown).deactivate().end(); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/OrderListenerInstrumentation.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/OrderListenerInstrumentation.java new file mode 100644 index 0000000000..b56ed0f15e --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/OrderListenerInstrumentation.java @@ -0,0 +1,73 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.consumer; + +import co.elastic.apm.agent.impl.transaction.Transaction; +import co.elastic.apm.agent.rocketmq.BaseRocketmqInstrumentation; +import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; +import org.apache.rocketmq.common.message.MessageExt; + +import java.util.List; + +import static net.bytebuddy.matcher.ElementMatchers.*; + +public class OrderListenerInstrumentation extends BaseRocketmqInstrumentation { + + @Override + public ElementMatcher getTypeMatcher() { + return hasSuperType(named("org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly")) + .and(not(isInterface())) + .and(not(isAbstract())); + } + + @Override + public ElementMatcher getMethodMatcher() { + return named("consumeMessage"); + } + + public static class AdviceClass { + + @Advice.OnMethodEnter(inline = false, suppress = Throwable.class) + public static Object onEnter(@Advice.Argument(0) List msgList, + @Advice.This Object thiz) { + return ConsumerHelper.get().onConsumeStart(thiz, msgList); + } + + @Advice.OnMethodExit(inline = false, suppress = Throwable.class, onThrowable = Throwable.class) + public static void onExit(@Advice.Thrown Throwable thrown, + @Advice.Return ConsumeOrderlyStatus status, + @Advice.Enter Object transactionObj) { + if (transactionObj == null) { + return; + } + + Transaction transaction = (Transaction) transactionObj; + ConsumerHelper helper = ConsumerHelper.get(); + if (thrown != null) { + helper.onConsumeError(transaction, thrown); + } else { + helper.onConsumeEnd(transaction, status.name(), status == ConsumeOrderlyStatus.SUCCESS); + } + } + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/PushConsumerHolder.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/PushConsumerHolder.java new file mode 100644 index 0000000000..9a37df348d --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/PushConsumerHolder.java @@ -0,0 +1,42 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.consumer; + +import com.blogspot.mydailyjava.weaklockfree.WeakConcurrentMap; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; + +import java.lang.ref.WeakReference; + +public class PushConsumerHolder { + + private static final WeakConcurrentMap> LISTENER_CONSUMER_MAP = new WeakConcurrentMap<>(true); + + public static void put(Object key, DefaultMQPushConsumer consumer) { + LISTENER_CONSUMER_MAP.put(key, new WeakReference<>(consumer)); + } + + public static DefaultMQPushConsumer get(Object key) { + WeakReference consumerRef = LISTENER_CONSUMER_MAP.get(key); + if (consumerRef == null) { + return null; + } + return consumerRef.get(); + } + +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/PushConsumerInstrumentation.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/PushConsumerInstrumentation.java new file mode 100644 index 0000000000..c63ff297b6 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/PushConsumerInstrumentation.java @@ -0,0 +1,54 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.consumer; + +import co.elastic.apm.agent.rocketmq.BaseRocketmqInstrumentation; +import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; + +public class PushConsumerInstrumentation extends BaseRocketmqInstrumentation { + + @Override + public ElementMatcher getTypeMatcher() { + return named("org.apache.rocketmq.client.consumer.DefaultMQPushConsumer"); + } + + @Override + public ElementMatcher getMethodMatcher() { + return named("registerMessageListener") + .and( + takesArgument(0, named("org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently")) + .or(takesArgument(0, named("org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly")))); + } + + public static class AdviceClass { + + @Advice.OnMethodEnter(inline = false, suppress = Throwable.class) + public static void onEnter(@Advice.This DefaultMQPushConsumer consumer, + @Advice.Argument(0) Object listener) { + PushConsumerHolder.put(listener, consumer); + } + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/PushConsumerNameSrvGetter.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/PushConsumerNameSrvGetter.java new file mode 100644 index 0000000000..437ce9c58f --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/consumer/PushConsumerNameSrvGetter.java @@ -0,0 +1,36 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.consumer; + +import co.elastic.apm.agent.rocketmq.namesrv.NameSrvGetter; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; + +public class PushConsumerNameSrvGetter implements NameSrvGetter { + + private static final PushConsumerNameSrvGetter INSTANCE = new PushConsumerNameSrvGetter(); + + public static PushConsumerNameSrvGetter get() { + return INSTANCE; + } + + @Override + public String get(DefaultMQPushConsumer consumer) { + return consumer.getNamesrvAddr(); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/namesrv/NameSrvGetter.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/namesrv/NameSrvGetter.java new file mode 100644 index 0000000000..a5e9ba3b75 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/namesrv/NameSrvGetter.java @@ -0,0 +1,23 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.namesrv; + +public interface NameSrvGetter { + String get(C carrier); +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/namesrv/NameSrvHolder.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/namesrv/NameSrvHolder.java new file mode 100644 index 0000000000..8e4c31773b --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/namesrv/NameSrvHolder.java @@ -0,0 +1,35 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.namesrv; + +import com.blogspot.mydailyjava.weaklockfree.WeakConcurrentMap; + +public class NameSrvHolder { + + private static final WeakConcurrentMap INNER_MAP = new WeakConcurrentMap<>(true); + + public static String get(Object key, C carrier, NameSrvGetter getter) { + String nameSrv = INNER_MAP.get(key); + if (nameSrv == null) { + nameSrv = getter.get(carrier); + INNER_MAP.put(key, nameSrv); + } + return nameSrv; + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/AsyncProducerHelper.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/AsyncProducerHelper.java new file mode 100644 index 0000000000..a2faedb38d --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/AsyncProducerHelper.java @@ -0,0 +1,50 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.producer; + +import co.elastic.apm.agent.impl.transaction.Span; +import co.elastic.apm.agent.sdk.DynamicTransformer; +import co.elastic.apm.agent.sdk.ElasticApmInstrumentation; +import org.apache.rocketmq.client.impl.CommunicationMode; +import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.common.message.MessageQueue; + +import java.util.Arrays; +import java.util.List; + +public class AsyncProducerHelper extends DefaultProducerHelper { + + public static final List> SEND_CALLBACK_INSTRUMENTATIONS = Arrays.>asList( + SendCallbackInstrumentation.OnSuccessInstrumentation.class, + SendCallbackInstrumentation.OnExceptionInstrumentation.class); + + + @Override + public Span onSendStart(DefaultMQProducerImpl producer, Message msg, MessageQueue mq, CommunicationMode mode, SendCallback sendCallback) { + Span span = super.onSendStart(producer, msg, mq, mode, sendCallback); + if (span != null) { + SendCallbackHolder.callbackSpanMap.put(sendCallback, span); + DynamicTransformer.Accessor.get().ensureInstrumented( + sendCallback.getClass(), AsyncProducerHelper.SEND_CALLBACK_INSTRUMENTATIONS); + } + return span; + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ClientAPIImplInstrumentation.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ClientAPIImplInstrumentation.java new file mode 100644 index 0000000000..504ff03e0e --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ClientAPIImplInstrumentation.java @@ -0,0 +1,41 @@ +package co.elastic.apm.agent.rocketmq.producer; + +import co.elastic.apm.agent.impl.transaction.Span; +import co.elastic.apm.agent.rocketmq.BaseRocketmqInstrumentation; +import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.matcher.ElementMatcher; + +import static net.bytebuddy.matcher.ElementMatchers.*; + +public class ClientAPIImplInstrumentation extends BaseRocketmqInstrumentation { + + @Override + public ElementMatcher getTypeMatcher() { + return named("org.apache.rocketmq.client.impl.MQClientAPIImpl"); + } + + @Override + public ElementMatcher getMethodMatcher() { + return named("sendMessage") + .and(takesArgument(0, String.class)) + .and(takesArgument(6, hasSuperType(named("org.apache.rocketmq.client.producer.SendCallback")))); + } + + public static class AdviceClass { + + @Advice.OnMethodEnter(inline = false, suppress = Throwable.class) + public static void onEnter(@Advice.Argument(0) String brokerAddr) { + Span activeSpan = tracer.getActiveSpan(); + if (activeSpan != null) { + String[] splits = brokerAddr.split(":"); + activeSpan.getContext().getDestination() + .withAddress(splits[0]) + .withPort(Integer.parseInt(splits[1])); + + } + } + + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/DefaultProducerHelper.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/DefaultProducerHelper.java new file mode 100644 index 0000000000..355805acb9 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/DefaultProducerHelper.java @@ -0,0 +1,103 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.producer; + +import co.elastic.apm.agent.impl.ElasticApmTracer; +import co.elastic.apm.agent.impl.GlobalTracer; +import co.elastic.apm.agent.impl.transaction.Outcome; +import co.elastic.apm.agent.impl.transaction.Span; +import co.elastic.apm.agent.rocketmq.Labels; +import co.elastic.apm.agent.rocketmq.namesrv.NameSrvHolder; +import org.apache.rocketmq.client.impl.CommunicationMode; +import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.client.producer.SendStatus; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.common.message.MessageQueue; + +import java.util.LinkedHashMap; +import java.util.Map; + +public class DefaultProducerHelper implements ProducerHelper { + + public static final Map MODE_ACTION_MAP; + + static { + MODE_ACTION_MAP = new LinkedHashMap<>(3); + MODE_ACTION_MAP.put(CommunicationMode.ASYNC, "sendAsync"); + MODE_ACTION_MAP.put(CommunicationMode.SYNC, "send"); + MODE_ACTION_MAP.put(CommunicationMode.ONEWAY, "sendOneway"); + } + + private final ElasticApmTracer tracer = GlobalTracer.requireTracerImpl(); + + @Override + public Span onSendStart(DefaultMQProducerImpl producer, Message msg, MessageQueue mq, CommunicationMode mode, SendCallback sendCallback) { + Span span = tracer.createExitChildSpan(); + if (span == null) { + return null; + } + String action = MODE_ACTION_MAP.get(mode); + if (action == null) { + action = "send"; + } + span.withType("messaging").withSubtype("rocketmq").withAction(action); + span.withName("RocketMQ ") + .appendToName(action).appendToName(" to ") + .appendToName(msg.getTopic()); + + span.getContext().getDestination().getService().getResource().append("rocketmq/").append(msg.getTopic()); + + if (msg.getTags() != null) { + span.getContext().getDestination().getService().getResource().append("/").append(msg.getTags()); + span.appendToName("/").appendToName(msg.getTags()); + } + span.getContext().getMessage().withQueue(mq.getBrokerName() + mq.getQueueId()); + if (msg.getKeys() != null) { + span.getContext().addLabel(Labels.KEYS, msg.getKeys()); + } + span.addLabel(Labels.NAME_SRV, NameSrvHolder.get(producer, producer, ProducerNameSrvGetter.get())); + return span.activate(); + } + + @Override + public void onSendEnd(Span span, SendResult sendResult) { + span.deactivate(); + } + + @Override + public void onSendFinished(Span span, SendResult sendResult) { + if (sendResult != null) { + span.getContext().addLabel(Labels.MSG_ID, sendResult.getMsgId()); + span.getContext().addLabel(Labels.SEND_STATUS, sendResult.getSendStatus().name()); + span.withOutcome(SendStatus.SEND_OK.equals(sendResult.getSendStatus()) ? Outcome.SUCCESS : Outcome.FAILURE); + } + + // end will be call in SyncProducerHelper or SendCallbackInstrumentation + span.end(); + } + + @Override + public void onSendError(Span span, Throwable thrown) { + span.captureException(thrown) + .withOutcome(Outcome.FAILURE) + .deactivate().end(); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerHelper.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerHelper.java new file mode 100644 index 0000000000..c5a7c1fd0a --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerHelper.java @@ -0,0 +1,39 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.producer; + +import co.elastic.apm.agent.impl.transaction.Span; +import org.apache.rocketmq.client.impl.CommunicationMode; +import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.common.message.MessageQueue; + +public interface ProducerHelper { + + Span onSendStart(DefaultMQProducerImpl producer, Message msg, + MessageQueue mq, CommunicationMode mode, SendCallback sendCallback); + + void onSendEnd(Span span, SendResult sendResult); + + void onSendFinished(Span span, SendResult sendResult); + + void onSendError(Span span, Throwable thrown); +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerHelperFactory.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerHelperFactory.java new file mode 100644 index 0000000000..bb0fe396c2 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerHelperFactory.java @@ -0,0 +1,39 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.producer; + +import org.apache.rocketmq.client.impl.CommunicationMode; + +import java.util.LinkedHashMap; +import java.util.Map; + +public class ProducerHelperFactory { + + private static final Map MODE_HELPER_MAP = new LinkedHashMap<>(); + + static { + MODE_HELPER_MAP.put(CommunicationMode.SYNC, new SyncProducerHelper()); + MODE_HELPER_MAP.put(CommunicationMode.ONEWAY, new SyncProducerHelper()); + MODE_HELPER_MAP.put(CommunicationMode.ASYNC, new AsyncProducerHelper()); + } + + public static ProducerHelper get(CommunicationMode mode) { + return MODE_HELPER_MAP.get(mode); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerInstrumentation.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerInstrumentation.java new file mode 100644 index 0000000000..dd83b18e66 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerInstrumentation.java @@ -0,0 +1,92 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.producer; + +import co.elastic.apm.agent.impl.transaction.Span; +import co.elastic.apm.agent.rocketmq.BaseRocketmqInstrumentation; +import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.rocketmq.client.impl.CommunicationMode; +import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.common.message.MessageQueue; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; + +public class ProducerInstrumentation extends BaseRocketmqInstrumentation { + + @Override + public ElementMatcher getTypeMatcher() { + return named("org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl"); + } + + @Override + public ElementMatcher getMethodMatcher() { + return named("sendKernelImpl") + .and(takesArgument(0, named("org.apache.rocketmq.common.message.Message"))) + .and(takesArgument(1, named("org.apache.rocketmq.common.message.MessageQueue"))) + .and(takesArgument(2, named("org.apache.rocketmq.client.impl.CommunicationMode"))) + .and(takesArgument(3, named("org.apache.rocketmq.client.producer.SendCallback"))); + } + + @Override + public String getAdviceClassName() { + return getClass().getName() + "$ProducerAdvice"; + } + + public static class ProducerAdvice { + + @Advice.OnMethodEnter(inline = false, suppress = Throwable.class) + public static Object onEnter(@Advice.Argument(0) Message msg, + @Advice.Argument(1) MessageQueue mq, + @Advice.Argument(2) CommunicationMode communicationMode, + @Advice.Argument(3) SendCallback sendCallback, + @Advice.This DefaultMQProducerImpl producer) { + ProducerHelper helper = ProducerHelperFactory.get(communicationMode); + if (helper == null) { + return null; + } + + return helper.onSendStart(producer, msg, mq, communicationMode, sendCallback); + } + + @Advice.OnMethodExit(inline = false, suppress = Throwable.class, onThrowable = Throwable.class) + public static void onExit(@Advice.Thrown Throwable thrown, + @Advice.Argument(2) CommunicationMode communicationMode, + @Advice.Return SendResult sendResult, + @Advice.Enter Object spanObj) { + ProducerHelper helper = ProducerHelperFactory.get(communicationMode); + if (spanObj == null || helper == null) { + return; + } + + Span span = (Span) spanObj; + if (thrown != null) { + helper.onSendError(span, thrown); + } else { + helper.onSendEnd(span, sendResult); + } + } + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerNameSrvGetter.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerNameSrvGetter.java new file mode 100644 index 0000000000..e651d4a96b --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/ProducerNameSrvGetter.java @@ -0,0 +1,47 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.producer; + +import co.elastic.apm.agent.rocketmq.namesrv.NameSrvGetter; +import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl; + +import java.util.List; + +public class ProducerNameSrvGetter implements NameSrvGetter { + + private static final ProducerNameSrvGetter INSTANCE = new ProducerNameSrvGetter(); + + public static ProducerNameSrvGetter get() { + return INSTANCE; + } + + @Override + public String get(DefaultMQProducerImpl producer) { + List nameServerAddressList = producer.getmQClientFactory().getMQClientAPIImpl() + .getRemotingClient().getNameServerAddressList(); + StringBuilder tmp = new StringBuilder(); + for (int i = 0; i < nameServerAddressList.size(); i++) { + if (i != 0) { + tmp.append(";"); + } + tmp.append(nameServerAddressList.get(i)); + } + return tmp.toString(); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/SendCallbackHolder.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/SendCallbackHolder.java new file mode 100644 index 0000000000..c273d2f35c --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/SendCallbackHolder.java @@ -0,0 +1,29 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.producer; + +import co.elastic.apm.agent.impl.transaction.Span; +import co.elastic.apm.agent.util.SpanConcurrentHashMap; +import com.blogspot.mydailyjava.weaklockfree.WeakConcurrentMap; +import org.apache.rocketmq.client.producer.SendCallback; + +public class SendCallbackHolder { + + public static final WeakConcurrentMap callbackSpanMap = SpanConcurrentHashMap.createWeakMap(); +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/SendCallbackInstrumentation.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/SendCallbackInstrumentation.java new file mode 100644 index 0000000000..17ae1e4d4f --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/SendCallbackInstrumentation.java @@ -0,0 +1,82 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.producer; + +import co.elastic.apm.agent.impl.transaction.Span; +import co.elastic.apm.agent.rocketmq.BaseRocketmqInstrumentation; +import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.matcher.ElementMatcher; +import net.bytebuddy.matcher.ElementMatchers; +import org.apache.rocketmq.client.impl.CommunicationMode; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; + +import static net.bytebuddy.matcher.ElementMatchers.named; + +public abstract class SendCallbackInstrumentation extends BaseRocketmqInstrumentation { + + @Override + public ElementMatcher getTypeMatcher() { + return ElementMatchers.hasSuperType(named("org.apache.rocketmq.client.producer.SendCallback")); + } + + public static class OnSuccessInstrumentation extends SendCallbackInstrumentation { + + @Override + public ElementMatcher getMethodMatcher() { + return named("onSuccess"); + } + + public static class AdviceClass { + + @Advice.OnMethodEnter(inline = false, suppress = Throwable.class) + public static void onEnter(@Advice.This SendCallback thiz, + @Advice.Argument(0) SendResult sendResult) { + Span span = SendCallbackHolder.callbackSpanMap.remove(thiz); + if (span == null) { + return; + } + ProducerHelperFactory.get(CommunicationMode.ASYNC).onSendFinished(span, sendResult); + } + } + } + + public static class OnExceptionInstrumentation extends SendCallbackInstrumentation { + + @Override + public ElementMatcher getMethodMatcher() { + return named("onException"); + } + + public static class AdviceClass { + + @Advice.OnMethodEnter(inline = false, suppress = Throwable.class) + public static void onEnter(@Advice.This SendCallback thiz, + @Advice.Argument(0) Throwable thrown) { + Span span = SendCallbackHolder.callbackSpanMap.remove(thiz); + if (span == null) { + return; + } + ProducerHelperFactory.get(CommunicationMode.ASYNC).onSendError(span, thrown); + } + } + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/SyncProducerHelper.java b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/SyncProducerHelper.java new file mode 100644 index 0000000000..78fdd06ff9 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/java/co/elastic/apm/agent/rocketmq/producer/SyncProducerHelper.java @@ -0,0 +1,31 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq.producer; + +import co.elastic.apm.agent.impl.transaction.Span; +import org.apache.rocketmq.client.producer.SendResult; + +public class SyncProducerHelper extends DefaultProducerHelper { + + @Override + public void onSendEnd(Span span, SendResult sendResult) { + super.onSendEnd(span, sendResult); + onSendFinished(span, sendResult); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/main/resources/META-INF/services/co.elastic.apm.agent.sdk.ElasticApmInstrumentation b/apm-agent-plugins/apm-rocketmq-plugin/src/main/resources/META-INF/services/co.elastic.apm.agent.sdk.ElasticApmInstrumentation new file mode 100644 index 0000000000..0f26b5a169 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/main/resources/META-INF/services/co.elastic.apm.agent.sdk.ElasticApmInstrumentation @@ -0,0 +1,5 @@ +co.elastic.apm.agent.rocketmq.consumer.ConcurrentListenerInstrumentation +co.elastic.apm.agent.rocketmq.consumer.OrderListenerInstrumentation +co.elastic.apm.agent.rocketmq.consumer.PushConsumerInstrumentation +co.elastic.apm.agent.rocketmq.producer.ProducerInstrumentation +co.elastic.apm.agent.rocketmq.producer.ClientAPIImplInstrumentation diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/ConsumeFuture.java b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/ConsumeFuture.java new file mode 100644 index 0000000000..0f84e76a9f --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/ConsumeFuture.java @@ -0,0 +1,24 @@ +package co.elastic.apm.agent.rocketmq; + +import org.apache.rocketmq.common.message.MessageExt; + +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +public class ConsumeFuture { + + private BlockingQueue> queue = new LinkedBlockingQueue<>(); + + public List get() { + try { + return queue.take(); + } catch (InterruptedException e) { + return null; + } + } + + public void consume(List messageExtList) { + this.queue.add(messageExtList); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/ConsumerTest.java b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/ConsumerTest.java new file mode 100644 index 0000000000..ccc0af002f --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/ConsumerTest.java @@ -0,0 +1,58 @@ +package co.elastic.apm.agent.rocketmq; + +import co.elastic.apm.agent.impl.transaction.Transaction; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; +import org.apache.rocketmq.common.message.MessageExt; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +public class ConsumerTest extends RocketmqClientIT { + + private static final String TOPIC = "Concurrent-topic"; + + private static final String TAGS = "tags"; + + private static final String KEYS = "keys"; + + private static DefaultMQPushConsumer consumer; + + private static final ConsumeFuture consumeFuture = new ConsumeFuture(); + + @BeforeAll + public static void init() throws Exception { + consumer = new DefaultMQPushConsumer("test-consumer-concurrent"); + consumer.setNamesrvAddr(rocketmq.getNameSrv()); + consumer.subscribe(TOPIC, "*"); + consumer.registerMessageListener(new MessageListenerConcurrently() { + @Override + public ConsumeConcurrentlyStatus consumeMessage(List msgs, + ConsumeConcurrentlyContext context) { + consumeFuture.consume(msgs); + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + } + }); + consumer.start(); + } + + @AfterAll + public static void destroy() { + consumer.shutdown(); + } + + @Test + public void test() throws Exception { + send(TOPIC, TAGS, KEYS); + List exts = consumeFuture.get(); + assertThat(exts).hasSize(1); + List transactions = reporter.getTransactions(); + assertThat(transactions).hasSize(1); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/EmbeddedRocketmq.java b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/EmbeddedRocketmq.java new file mode 100644 index 0000000000..1baf6dfc57 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/EmbeddedRocketmq.java @@ -0,0 +1,82 @@ +package co.elastic.apm.agent.rocketmq; + +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.common.MQVersion; +import org.apache.rocketmq.common.namesrv.NamesrvConfig; +import org.apache.rocketmq.namesrv.NamesrvController; +import org.apache.rocketmq.remoting.netty.NettyClientConfig; +import org.apache.rocketmq.remoting.netty.NettyServerConfig; +import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import org.apache.rocketmq.store.config.MessageStoreConfig; + +import java.io.File; + +public class EmbeddedRocketmq { + + private NamesrvController namesrvController; + + private BrokerController brokerController; + + private String storeDir; + + public void start() throws Exception { + System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION)); + NettyServerConfig nameSrvNettyConfig = new NettyServerConfig(); + nameSrvNettyConfig.setListenPort(9876); + this.namesrvController = new NamesrvController(new NamesrvConfig(), nameSrvNettyConfig); + this.namesrvController.initialize(); + this.namesrvController.start(); + + BrokerConfig brokerConfig = new BrokerConfig(); + brokerConfig.setNamesrvAddr("localhost:9876"); + this.brokerController = new BrokerController( + brokerConfig, new NettyServerConfig(), + new NettyClientConfig(), new MessageStoreConfig()); + deleteDir(new File(this.brokerController.getMessageStoreConfig().getStorePathRootDir())); + this.brokerController.initialize(); + this.brokerController.start(); + this.storeDir = this.brokerController.getMessageStoreConfig().getStorePathRootDir(); + } + + public void stop() { + try { + this.brokerController.shutdown(); + } catch (Exception e) { + //e.printStackTrace(); + deleteDir(new File(storeDir)); + } + this.namesrvController.shutdown(); + } + + public String getNameSrv() { + return "localhost:9876"; + } + + public boolean deleteDir(File someFile) { + if (!someFile.exists()) { + System.out.println("[deleteDir]File " + someFile.getAbsolutePath() + + " does not exist."); + return false; + } + if (someFile.isDirectory()) {// is a folder + File[] files = someFile.listFiles(); + for (File subFile : files) { + boolean isSuccess = deleteDir(subFile); + if (!isSuccess) { + return isSuccess; + } + } + } else {// is a regular file + boolean isSuccess = someFile.delete(); + if (!isSuccess) { + return isSuccess; + } + } + if (someFile.isDirectory()) { + return someFile.delete(); + } else { + return true; + } + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/OrderConsumerTest.java b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/OrderConsumerTest.java new file mode 100644 index 0000000000..87a3f8d84a --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/OrderConsumerTest.java @@ -0,0 +1,59 @@ +package co.elastic.apm.agent.rocketmq; + +import co.elastic.apm.agent.impl.transaction.Transaction; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; +import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly; +import org.apache.rocketmq.common.message.MessageExt; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +public class OrderConsumerTest extends RocketmqClientIT { + + private static final String TOPIC = "order-topic"; + + private static final String TAGS = "tags"; + + private static final String KEYS = "keys"; + + private static DefaultMQPushConsumer consumer; + + private static final ConsumeFuture consumeFuture = new ConsumeFuture(); + + @BeforeAll + public static void init() throws Exception { + consumer = new DefaultMQPushConsumer("test-consumer-order"); + consumer.setNamesrvAddr(rocketmq.getNameSrv()); + consumer.subscribe(TOPIC, "*"); + consumer.registerMessageListener(new MessageListenerOrderly() { + + @Override + public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) { + consumeFuture.consume(msgs); + return ConsumeOrderlyStatus.SUCCESS; + } + }); + consumer.start(); + } + + @AfterAll + public static void destroy() { + consumer.shutdown(); + } + + @Test + public void test() throws Exception { + send(TOPIC, TAGS, KEYS); + List exts = consumeFuture.get(); + assertThat(exts).hasSize(1); + reporter.getFirstTransaction(500); + List transactions = reporter.getTransactions(); + assertThat(transactions).hasSize(1); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/ProducerTest.java b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/ProducerTest.java new file mode 100644 index 0000000000..5ffc77371a --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/ProducerTest.java @@ -0,0 +1,111 @@ +package co.elastic.apm.agent.rocketmq; + +import co.elastic.apm.agent.impl.transaction.Span; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.client.producer.SendStatus; +import org.apache.rocketmq.common.message.Message; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CompletableFuture; + +import static org.assertj.core.api.Assertions.assertThat; + +public class ProducerTest extends RocketmqClientIT { + + private static final String TOPIC = "topic-test"; + + private static final String TAGS = "tags"; + + private static final String KEYS = "keys"; + + private static final String MESSAGE = "message"; + + private static final String SEND_SYNC_SPAN_NAME = "RocketMQ send to " + TOPIC + "/" + TAGS; + + private static final String SEND_ASYNC_SPAN_NAME = "RocketMQ sendAsync to " + TOPIC + "/" + TAGS; + + private static final String SEND_ONEWAY_SPAN_NAME = "RocketMQ sendOneway to " + TOPIC + "/" + TAGS; + + @BeforeEach + public void beforeEach() { + startTestRootTransaction(); + } + + @AfterEach + public void afterEach() { + tracer.getActive().deactivate().end(); + } + + @Test + public void testSendSync() throws Exception { + SendResult sendResult = sendSync(); + assertThat(sendResult).isNotNull(); + assertThat(sendResult.getSendStatus()).isEqualTo(SendStatus.SEND_OK); + validateSendSpan(sendResult, true); + } + + @Test + public void testSendOneway() throws Exception { + sendOneWay(); + validateSendOnewaySpan(); + } + + @Test + public void testSendAsync() throws Exception { + CompletableFuture future = new CompletableFuture<>(); + sendAsync(new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + future.complete(sendResult); + } + + @Override + public void onException(Throwable e) { + future.complete(null); + } + }); + SendResult sendResult = future.get(); + validateSendSpan(sendResult, false); + } + + public void validateSendSpan(SendResult sendResult, boolean sync) { + assertThat(reporter.getFirstSpan(1000)).isNotNull(); + assertThat(reporter.getSpans()).hasSize(1); + Span span = reporter.getSpans().get(0); + if (sync) { + assertThat(span.getNameAsString()).isEqualTo(SEND_SYNC_SPAN_NAME); + } else { + assertThat(span.getNameAsString()).isEqualTo(SEND_ASYNC_SPAN_NAME); + } + + assertThat(span.getContext().getLabel(Labels.KEYS)).isEqualTo(KEYS); + assertThat(span.getContext().getLabel(Labels.NAME_SRV)).isEqualTo(rocketmq.getNameSrv()); + assertThat(span.getContext().getLabel(Labels.MSG_ID)).isEqualTo(sendResult.getMsgId()); + assertThat(span.getContext().getLabel(Labels.SEND_STATUS)).isEqualTo(sendResult.getSendStatus().name()); + } + + public void validateSendOnewaySpan() { + assertThat(reporter.getFirstSpan(1000)).isNotNull(); + assertThat(reporter.getSpans()).hasSize(1); + Span span = reporter.getSpans().get(0); + assertThat(span.getNameAsString()).isEqualTo(SEND_ONEWAY_SPAN_NAME); + assertThat(span.getContext().getLabel(Labels.KEYS)).isEqualTo(KEYS); + assertThat(span.getContext().getLabel(Labels.NAME_SRV)).isEqualTo(rocketmq.getNameSrv()); + } + + private SendResult sendSync() throws Exception { + return producer.send(new Message(TOPIC, TAGS, KEYS, MESSAGE.getBytes(StandardCharsets.UTF_8))); + } + + private void sendOneWay() throws Exception { + producer.sendOneway(new Message(TOPIC, TAGS, KEYS, MESSAGE.getBytes(StandardCharsets.UTF_8))); + } + + private void sendAsync(SendCallback sendCallback) throws Exception { + producer.send(new Message(TOPIC, TAGS, KEYS, MESSAGE.getBytes(StandardCharsets.UTF_8)), sendCallback); + } +} diff --git a/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/RocketmqClientIT.java b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/RocketmqClientIT.java new file mode 100644 index 0000000000..1cc5db0ea4 --- /dev/null +++ b/apm-agent-plugins/apm-rocketmq-plugin/src/test/java/co/elastic/apm/agent/rocketmq/RocketmqClientIT.java @@ -0,0 +1,55 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package co.elastic.apm.agent.rocketmq; + +import co.elastic.apm.agent.AbstractInstrumentationTest; +import org.apache.rocketmq.client.producer.DefaultMQProducer; +import org.apache.rocketmq.common.message.Message; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; + +import java.nio.charset.StandardCharsets; + +public class RocketmqClientIT extends AbstractInstrumentationTest { + + protected static DefaultMQProducer producer; + + protected static EmbeddedRocketmq rocketmq; + + @BeforeAll + public static void setup() throws Exception { + rocketmq = new EmbeddedRocketmq(); + rocketmq.start(); + + producer = new DefaultMQProducer(); + producer.setNamesrvAddr(rocketmq.getNameSrv()); + producer.setProducerGroup("producer"); + producer.start(); + } + + @AfterAll + public static void stop() { + rocketmq.stop(); + producer.shutdown(); + } + + protected void send(String topic, String tags, String keys) throws Exception { + producer.send(new Message(topic, tags, keys, "message".getBytes(StandardCharsets.UTF_8))); + } +} diff --git a/apm-agent-plugins/pom.xml b/apm-agent-plugins/pom.xml index 53fcdff320..8bd876958d 100644 --- a/apm-agent-plugins/pom.xml +++ b/apm-agent-plugins/pom.xml @@ -64,6 +64,7 @@ apm-vertx apm-sparkjava-plugin apm-javalin-plugin + apm-rocketmq-plugin diff --git a/pom.xml b/pom.xml index 3c0992babe..1b24938bbd 100644 --- a/pom.xml +++ b/pom.xml @@ -586,7 +586,7 @@ org.mockito mockito-core - 3.12.4 + 4.0.0 test