GH-1101: Message Listener Micrometer Timers

Resolves #1101

* Remove extra back-tick

* Fix timer name in doc

Author: garyrussell

build.gradle

@@ -47,6 +47,7 @@ ext {
 	junitJupiterVersion = '5.5.2'
 	log4jVersion = '2.12.1'
 	logbackVersion = '1.2.3'
+	micrometerVersion = '1.2.1'
 	mockitoVersion = '3.0.0'
 	rabbitmqVersion = project.hasProperty('rabbitmqVersion') ? project.rabbitmqVersion : '5.7.3'
 	rabbitmqHttpClientVersion = '3.2.0.RELEASE'
@@ -333,6 +334,7 @@ project('spring-rabbit') {
 		compile ('io.projectreactor:reactor-core', optional)
 		compile ("ch.qos.logback:logback-classic:$logbackVersion", optional)
 		compile ('org.apache.logging.log4j:log4j-core', optional)
+		compile ("io.micrometer:micrometer-core:$micrometerVersion", optional)
 
 		testCompile project(':spring-rabbit-junit')
 		testCompile("com.willowtreeapps.assertk:assertk-jvm:$assertkVersion")

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java

@@ -26,6 +26,8 @@
 import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.Executor;
 import java.util.stream.Collectors;
@@ -80,13 +82,18 @@
 import org.springframework.transaction.interceptor.TransactionAttribute;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.ErrorHandler;
 import org.springframework.util.StringUtils;
 import org.springframework.util.backoff.BackOff;
 import org.springframework.util.backoff.FixedBackOff;
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.ShutdownSignalException;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+import io.micrometer.core.instrument.Timer.Builder;
+import io.micrometer.core.instrument.Timer.Sample;
 
 /**
  * @author Mark Pollack
@@ -119,12 +126,17 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	public static final long DEFAULT_SHUTDOWN_TIMEOUT = 5000;
 
+	private static final boolean MICROMETER_PRESENT = ClassUtils.isPresent(
+			"io.micrometer.core.instrument.MeterRegistry", AbstractMessageListenerContainer.class.getClassLoader());
+
 	private final ContainerDelegate delegate = this::actualInvokeListener;
 
 	protected final Object consumersMonitor = new Object(); //NOSONAR
 
 	private final Map<String, Object> consumerArgs = new HashMap<>();
 
+	private final Map<String, String> micrometerTags = new HashMap<>();
+
 	private ContainerDelegate proxy = this.delegate;
 
 	private long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT;
@@ -224,6 +236,10 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
 
+	private MicrometerHolder micrometerHolder;
+
+	private boolean micrometerEnabled = true;
+
 	private volatile boolean lazyLoad;
 
 	@Override
@@ -631,12 +647,16 @@ public void setForceCloseChannel(boolean forceCloseChannel) {
 	@Nullable
 	protected String getRoutingLookupKey() {
 		return super.getConnectionFactory() instanceof RoutingConnectionFactory
-				? this.lookupKeyQualifier + "[" + this.queues.stream()
-				.map(Queue::getName)
-				.collect(Collectors.joining(",")) + "]"
+				? this.lookupKeyQualifier + queuesAsListString()
 				: null;
 	}
 
+	private String queuesAsListString() {
+		return "[" + this.queues.stream()
+				.map(Queue::getName)
+				.collect(Collectors.joining(",")) + "]";
+	}
+
 	/**
 	 * Return the (@link RoutingConnectionFactory} if the connection factory is a
 	 * {@link RoutingConnectionFactory}; null otherwise.
@@ -1075,6 +1095,26 @@ protected Collection<MessagePostProcessor> getAfterReceivePostProcessors() {
 		return this.afterReceivePostProcessors;
 	}
 
+	/**
+	 * Set additional tags for the Micrometer listener timers.
+	 * @param tags the tags.
+	 * @since 2.2
+	 */
+	public void setMicrometerTags(Map<String, String> tags) {
+		if (tags != null) {
+			this.micrometerTags.putAll(tags);
+		}
+	}
+
+	/**
+	 * Set to false to disable micrometer listener timers.
+	 * @param micrometerEnabled false to disable.
+	 * @since 2.2
+	 */
+	public void setMicrometerEnabled(boolean micrometerEnabled) {
+		this.micrometerEnabled = micrometerEnabled;
+	}
+
 	/**
 	 * Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
 	 */
@@ -1093,6 +1133,16 @@ public final void afterPropertiesSet() {
 						"channelTransacted=false");
 		validateConfiguration();
 		initialize();
+		try {
+			if (this.micrometerHolder == null && MICROMETER_PRESENT && this.micrometerEnabled
+					&& this.applicationContext != null) {
+				this.micrometerHolder = new MicrometerHolder(this.applicationContext, getListenerId(),
+						this.micrometerTags);
+			}
+		}
+		catch (IllegalStateException e) {
+			this.logger.debug("Could not enable micrometer timers", e);
+		}
 	}
 
 	@Override
@@ -1128,6 +1178,9 @@ protected void initializeProxy(Object delegate) {
 	@Override
 	public void destroy() {
 		shutdown();
+		if (this.micrometerHolder != null) {
+			this.micrometerHolder.destroy();
+		}
 	}
 
 	// -------------------------------------------------------------------------
@@ -1350,10 +1403,24 @@ protected void executeListener(Channel channel, Object data) {
 			}
 			throw new MessageRejectedWhileStoppingException();
 		}
+		Object sample = null;
+		if (this.micrometerHolder != null) {
+			sample = this.micrometerHolder.start();
+		}
 		try {
 			doExecuteListener(channel, data);
+			if (sample != null) {
+				this.micrometerHolder.success(sample, data instanceof Message
+						? ((Message) data).getMessageProperties().getConsumerQueue()
+						: queuesAsListString());
+			}
 		}
 		catch (RuntimeException ex) {
+			if (sample != null) {
+				this.micrometerHolder.failure(sample, data instanceof Message
+						? ((Message) data).getMessageProperties().getConsumerQueue()
+						: queuesAsListString(), ex.getClass().getSimpleName());
+			}
 			Message message;
 			if (data instanceof Message) {
 				message = (Message) data;
@@ -1908,4 +1975,72 @@ else if (!RabbitUtils.isNormalChannelClose(cause)) {
 
 	}
 
+	private static final class MicrometerHolder {
+
+		private final ConcurrentMap<String, Timer> timers = new ConcurrentHashMap<>();
+
+		private final MeterRegistry registry;
+
+		private final Map<String, String> tags;
+
+		private final String listenerId;
+
+		MicrometerHolder(@Nullable ApplicationContext context, String listenerId, Map<String, String> tags) {
+			if (context == null) {
+				throw new IllegalStateException("No micrometer registry present");
+			}
+			Map<String, MeterRegistry> registries = context.getBeansOfType(MeterRegistry.class, false, false);
+			if (registries.size() == 1) {
+				this.registry = registries.values().iterator().next();
+				this.listenerId = listenerId;
+				this.tags = tags;
+			}
+			else {
+				throw new IllegalStateException("No micrometer registry present");
+			}
+		}
+
+		Object start() {
+			return Timer.start(this.registry);
+		}
+
+		void success(Object sample, String queue) {
+			Timer timer = this.timers.get(queue + "none");
+			if (timer == null) {
+				timer = buildTimer(this.listenerId, "success", queue, "none");
+			}
+			((Sample) sample).stop(timer);
+		}
+
+		void failure(Object sample, String queue, String exception) {
+			Timer timer = this.timers.get(queue + exception);
+			if (timer == null) {
+				timer = buildTimer(this.listenerId, "failure", queue, exception);
+			}
+			((Sample) sample).stop(timer);
+		}
+
+		private Timer buildTimer(String aListenerId, String result, String queue, String exception) {
+
+			Builder builder = Timer.builder("spring.rabbitmq.listener")
+					.description("Spring RabbitMQ Listener")
+					.tag("listener.id", aListenerId)
+					.tag("queue", queue)
+					.tag("result", result)
+					.tag("exception", exception);
+			if (this.tags != null && !this.tags.isEmpty()) {
+				this.tags.forEach((key, value) -> builder.tag(key, value));
+			}
+			Timer registeredTimer = builder.register(this.registry);
+			this.timers.put(queue + exception, registeredTimer);
+			return registeredTimer;
+		}
+
+		void destroy() {
+			this.timers.values().forEach(this.registry::remove);
+			this.timers.clear();
+		}
+
+	}
+
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ConsumerBatchingTests.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -47,6 +48,9 @@
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 import com.rabbitmq.client.Channel;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 
 /**
  * @author Gary Russell
@@ -64,6 +68,9 @@ public class ConsumerBatchingTests {
 	@Autowired
 	private Listener listener;
 
+	@Autowired
+	private MeterRegistry meterRegistry;
+
 	@Test
 	public void replayWholeBatch() throws InterruptedException {
 		this.template.convertAndSend("c.batch.1", new Foo("foo"));
@@ -75,6 +82,33 @@ public void replayWholeBatch() throws InterruptedException {
 		assertThat(this.listener.foos)
 			.extracting(foo -> foo.getBar())
 			.contains("foo", "bar", "baz", "qux", "foo", "bar", "baz", "qux");
+		Timer timer = null;
+		int n = 0;
+		while (timer == null && n++ < 100) {
+			try {
+				timer = this.meterRegistry.get("spring.rabbitmq.listener")
+						.tag("listener.id", "batch.1")
+						.tag("queue", "[c.batch.1]")
+						.tag("result", "success")
+						.tag("exception", "none")
+						.tag("extraTag", "foo")
+						.timer();
+			}
+			catch (@SuppressWarnings("unused") Exception e) {
+				Thread.sleep(100);
+			}
+		}
+		assertThat(timer).isNotNull();
+		assertThat(timer.count()).isEqualTo(1L);
+		timer = this.meterRegistry.get("spring.rabbitmq.listener")
+				.tag("listener.id", "batch.1")
+				.tag("queue", "[c.batch.1]")
+				.tag("result", "failure")
+				.tag("exception", "ListenerExecutionFailedException")
+				.tag("extraTag", "foo")
+				.timer();
+		assertThat(timer).isNotNull();
+		assertThat(timer.count()).isEqualTo(1L);
 	}
 
 	@Test
@@ -137,13 +171,20 @@ public void rejectOneReplayRest() throws InterruptedException {
 	@EnableRabbit
 	public static class Config {
 
+		@Bean
+		public MeterRegistry meterRegistry() {
+			return new SimpleMeterRegistry();
+		}
+
 		@Bean
 		public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
 			SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
 			factory.setConnectionFactory(connectionFactory());
 			factory.setBatchListener(true);
 			factory.setConsumerBatchEnabled(true);
 			factory.setBatchSize(4);
+			factory.setContainerConfigurer(
+					container -> container.setMicrometerTags(Collections.singletonMap("extraTag", "foo")));
 			return factory;
 		}
 
@@ -250,7 +291,7 @@ public static class Listener {
 
 		volatile boolean first = true;
 
-		@RabbitListener(queues = "c.batch.1")
+		@RabbitListener(id = "batch.1", queues = "c.batch.1")
 		public void listen1(List<Foo> in) {
 			this.foos.addAll(in);
 			if (this.first) {