From cffebb9d27ebddf39996302e5b68649a105ca4b4 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 18 May 2022 15:39:40 -0400 Subject: [PATCH] GH-1459: Improve MeterRegistry Discovery Resolves https://github.com/spring-projects/spring-amqp/issues/1459 * Add micrometer properties and container customizers to LCFB. * Use `ObjectProvider` to locate registry. --- .../config/ListenerContainerFactoryBean.java | 58 ++++++- .../AbstractMessageListenerContainer.java | 74 --------- .../rabbit/listener/MicrometerHolder.java | 146 ++++++++++++++++++ .../ListenerContainerFactoryBeanTests.java | 77 +++++++++ .../listener/MicrometerHolderTests.java | 118 ++++++++++++++ src/reference/asciidoc/amqp.adoc | 2 +- 6 files changed, 399 insertions(+), 76 deletions(-) create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MicrometerHolder.java create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBeanTests.java create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MicrometerHolderTests.java diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java index 07c7be9f..c597f3e8 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java @@ -16,6 +16,7 @@ package org.springframework.amqp.rabbit.config; +import java.util.HashMap; import java.util.Map; import java.util.concurrent.Executor; @@ -61,6 +62,8 @@ import org.springframework.util.backoff.BackOff; public class ListenerContainerFactoryBean extends AbstractFactoryBean implements ApplicationContextAware, BeanNameAware, ApplicationEventPublisherAware, SmartLifecycle { + private final Map micrometerTags = new HashMap<>(); + private ApplicationContext applicationContext; private String beanName; @@ -171,6 +174,12 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean smlcCustomizer; + + private ContainerCustomizer dmlcCustomizer; + @Override public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { this.applicationEventPublisher = applicationEventPublisher; @@ -427,6 +436,44 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean tags) { + this.micrometerTags.putAll(tags); + } + + /** + * Set a {@link ContainerCustomizer} that is invoked after a container is created and + * configured to enable further customization of the container. + * @param containerCustomizer the customizer. + * @since 2.4.6 + */ + public void setSMLCCustomizer(ContainerCustomizer customizer) { + this.smlcCustomizer = customizer; + } + + /** + * Set a {@link ContainerCustomizer} that is invoked after a container is created and + * configured to enable further customization of the container. + * @param containerCustomizer the customizer. + * @since 2.4.6 + */ + public void setDMLCCustomizer(ContainerCustomizer customizer) { + this.dmlcCustomizer = customizer; + } + @Override public Class getObjectType() { return this.listenerContainer == null @@ -478,7 +525,16 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean 0, this.micrometerTags, + container::setMicrometerTags); + if (this.smlcCustomizer != null && this.type.equals(Type.simple)) { + this.smlcCustomizer.configure((SimpleMessageListenerContainer) container); + } + else if (this.dmlcCustomizer != null && this.type.equals(Type.direct)) { + this.dmlcCustomizer.configure((DirectMessageListenerContainer) container); + } container.afterPropertiesSet(); this.listenerContainer = container; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index 17b2beb1..9f77462b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -26,8 +26,6 @@ import java.util.Map; 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; @@ -93,10 +91,6 @@ 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 @@ -2102,72 +2096,4 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor } - private static final class MicrometerHolder { - - private final ConcurrentMap timers = new ConcurrentHashMap<>(); - - private final MeterRegistry registry; - - private final Map tags; - - private final String listenerId; - - MicrometerHolder(@Nullable ApplicationContext context, String listenerId, Map tags) { - if (context == null) { - throw new IllegalStateException("No micrometer registry present"); - } - Map 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(); - } - - } - } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MicrometerHolder.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MicrometerHolder.java new file mode 100644 index 00000000..fb00ae38 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MicrometerHolder.java @@ -0,0 +1,146 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed 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 + * + * https://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 org.springframework.amqp.rabbit.listener; + +import java.util.Collections; +import java.util.Map; +import java.util.Map.Entry; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import org.springframework.beans.factory.NoUniqueBeanDefinitionException; +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.lang.Nullable; + +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; + +/** + * Abstraction to avoid hard reference to Micrometer. + * + * @author Gary Russell + * @since 2.4.6 + * + */ +final class MicrometerHolder { + + private final ConcurrentMap timers = new ConcurrentHashMap<>(); + + private final MeterRegistry registry; + + private final Map tags; + + private final String listenerId; + + MicrometerHolder(@Nullable ApplicationContext context, String listenerId, Map tags) { + if (context == null) { + throw new IllegalStateException("No micrometer registry present"); + } + try { + this.registry = context.getBeanProvider(MeterRegistry.class).getIfUnique(); + } + catch (NoUniqueBeanDefinitionException ex) { + throw new IllegalStateException(ex); + } + if (this.registry != null) { + this.listenerId = listenerId; + this.tags = tags; + } + else { + throw new IllegalStateException("No micrometer registry present (or more than one and " + + "there is not exactly one marked with @Primary)"); + } + } + + private Map filterRegistries(Map registries, + ApplicationContext context) { + + if (registries.size() == 1) { + return registries; + } + MeterRegistry primary = null; + if (context instanceof ConfigurableApplicationContext) { + BeanDefinitionRegistry bdr = (BeanDefinitionRegistry) ((ConfigurableApplicationContext) context) + .getBeanFactory(); + for (Entry entry : registries.entrySet()) { + BeanDefinition beanDefinition = bdr.getBeanDefinition(entry.getKey()); + if (beanDefinition.isPrimary()) { + if (primary != null) { + primary = null; + break; + } + else { + primary = entry.getValue(); + } + } + } + } + if (primary != null) { + return Collections.singletonMap("primary", primary); + } + else { + return registries; + } + } + + 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(); + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBeanTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBeanTests.java new file mode 100644 index 00000000..3b960261 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBeanTests.java @@ -0,0 +1,77 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed 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 + * + * https://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 org.springframework.amqp.rabbit.config; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.rabbit.config.ListenerContainerFactoryBean.Type; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; +import org.springframework.amqp.utils.test.TestUtils; + +/** + * @author Gary Russell + * @since 2.4.6 + * + */ +public class ListenerContainerFactoryBeanTests { + + @SuppressWarnings("unchecked") + @Test + void micrometer() throws Exception { + ListenerContainerFactoryBean lcfb = new ListenerContainerFactoryBean(); + lcfb.setConnectionFactory(mock(ConnectionFactory.class)); + lcfb.setMicrometerEnabled(false); + lcfb.setMicrometerTags(Map.of("foo", "bar")); + lcfb.afterPropertiesSet(); + AbstractMessageListenerContainer container = lcfb.getObject(); + assertThat(TestUtils.getPropertyValue(container, "micrometerEnabled", Boolean.class)).isFalse(); + assertThat(TestUtils.getPropertyValue(container, "micrometerTags", Map.class)).hasSize(1); + } + + @Test + void smlcCustomizer() throws Exception { + ListenerContainerFactoryBean lcfb = new ListenerContainerFactoryBean(); + lcfb.setConnectionFactory(mock(ConnectionFactory.class)); + lcfb.setSMLCCustomizer(container -> { + container.setConsumerStartTimeout(42L); + }); + lcfb.afterPropertiesSet(); + AbstractMessageListenerContainer container = lcfb.getObject(); + assertThat(TestUtils.getPropertyValue(container, "consumerStartTimeout", Long.class)).isEqualTo(42L); + } + + @Test + void dmlcCustomizer() throws Exception { + ListenerContainerFactoryBean lcfb = new ListenerContainerFactoryBean(); + lcfb.setConnectionFactory(mock(ConnectionFactory.class)); + lcfb.setType(Type.direct); + lcfb.setDMLCCustomizer(container -> { + container.setConsumersPerQueue(2); + }); + lcfb.afterPropertiesSet(); + AbstractMessageListenerContainer container = lcfb.getObject(); + assertThat(TestUtils.getPropertyValue(container, "consumersPerQueue", Integer.class)).isEqualTo(2); + } + + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MicrometerHolderTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MicrometerHolderTests.java new file mode 100644 index 00000000..b576e507 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MicrometerHolderTests.java @@ -0,0 +1,118 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed 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 + * + * https://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 org.springframework.amqp.rabbit.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; +import static org.mockito.Mockito.mock; + +import java.util.Collections; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Primary; +import org.springframework.test.util.ReflectionTestUtils; + +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Timer; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; + +/** + * @author Gary Russell + */ +public class MicrometerHolderTests { + + @Test + void multiReg() { + ApplicationContext context = new AnnotationConfigApplicationContext(Config1.class); + assertThatIllegalStateException().isThrownBy(() -> new MicrometerHolder(context, "", Collections.emptyMap())) + .withMessage("No micrometer registry present (or more than one and " + + "there is not exactly one marked with @Primary)"); + } + + @Test + void twoPrimaries() { + ApplicationContext context = new AnnotationConfigApplicationContext(Config2.class); + assertThatIllegalStateException().isThrownBy(() -> new MicrometerHolder(context, "", Collections.emptyMap())) + .withMessageContaining("more than one 'primary' bean"); + } + + @Test + void primary() { + AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(Config3.class); + MicrometerHolder micrometerHolder = new MicrometerHolder(ctx, "holderName", Collections.emptyMap()); + Timer.Sample sample = mock(Timer.Sample.class); + micrometerHolder.success(sample, "queue"); + micrometerHolder.failure(sample, "queue", "SomeException"); + @SuppressWarnings("unchecked") + Map meters = (Map) ReflectionTestUtils.getField(micrometerHolder, "timers"); + assertThat(meters).hasSize(2); + ctx.close(); + micrometerHolder.destroy(); + assertThat(meters).hasSize(0); + } + + static class Config1 { + + @Bean + MeterRegistry reg1() { + return new SimpleMeterRegistry(); + } + + @Bean + MeterRegistry reg2() { + return new SimpleMeterRegistry(); + } + + } + + static class Config2 { + + @Bean + @Primary + MeterRegistry reg1() { + return new SimpleMeterRegistry(); + } + + @Bean + @Primary + MeterRegistry reg2() { + return new SimpleMeterRegistry(); + } + + } + + static class Config3 { + + @Bean + @Primary + MeterRegistry reg1() { + return new SimpleMeterRegistry(); + } + + @Bean + MeterRegistry reg2() { + return new SimpleMeterRegistry(); + } + + } + +} diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 3c20e121..a0931518 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -3746,7 +3746,7 @@ Instead, you should hand off the event to a different thread that can then stop [[micrometer]] ===== Monitoring Listener Performance -Starting with version 2.2, the listener containers will automatically create and update Micrometer `Timer` s for the listener, if `Micrometer` is detected on the class path, and a `MeterRegistry` is present in the application context. +Starting with version 2.2, the listener containers will automatically create and update Micrometer `Timer` s for the listener, if `Micrometer` is detected on the class path, and a single `MeterRegistry` is present in the application context (or exactly one is annotated `@Primary`, such as when using Spring Boot). The timers can be disabled by setting the container property `micrometerEnabled` to `false`. Two timers are maintained - one for successful calls to the listener and one for failures.