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.
This commit is contained in:
Gary Russell
@@ -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<AbstractMessageListenerContainer>
|
||||
implements ApplicationContextAware, BeanNameAware, ApplicationEventPublisherAware, SmartLifecycle {
|
||||
|
||||
private final Map<String, String> micrometerTags = new HashMap<>();
|
||||
|
||||
private ApplicationContext applicationContext;
|
||||
|
||||
private String beanName;
|
||||
@@ -171,6 +174,12 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
|
||||
|
||||
private Boolean consumerBatchEnabled;
|
||||
|
||||
private Boolean micrometerEnabled;
|
||||
|
||||
private ContainerCustomizer<SimpleMessageListenerContainer> smlcCustomizer;
|
||||
|
||||
private ContainerCustomizer<DirectMessageListenerContainer> dmlcCustomizer;
|
||||
|
||||
@Override
|
||||
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
|
||||
this.applicationEventPublisher = applicationEventPublisher;
|
||||
@@ -427,6 +436,44 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
|
||||
this.retryDeclarationInterval = retryDeclarationInterval;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set to false to disable micrometer listener timers.
|
||||
* @param micrometerEnabled false to disable.
|
||||
* @since 2.4.6
|
||||
*/
|
||||
public void setMicrometerEnabled(boolean enabled) {
|
||||
this.micrometerEnabled = enabled;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set additional tags for the Micrometer listener timers.
|
||||
* @param tags the tags.
|
||||
* @since 2.4.6
|
||||
*/
|
||||
public void setMicrometerTags(Map<String, String> 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<SimpleMessageListenerContainer> 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<DirectMessageListenerContainer> customizer) {
|
||||
this.dmlcCustomizer = customizer;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Class<?> getObjectType() {
|
||||
return this.listenerContainer == null
|
||||
@@ -478,7 +525,16 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
|
||||
.acceptIfNotNull(this.autoDeclare, container::setAutoDeclare)
|
||||
.acceptIfNotNull(this.failedDeclarationRetryInterval, container::setFailedDeclarationRetryInterval)
|
||||
.acceptIfNotNull(this.exclusiveConsumerExceptionLogger,
|
||||
container::setExclusiveConsumerExceptionLogger);
|
||||
container::setExclusiveConsumerExceptionLogger)
|
||||
.acceptIfNotNull(this.micrometerEnabled, container::setMicrometerEnabled)
|
||||
.acceptIfCondition(this.micrometerTags.size() > 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;
|
||||
}
|
||||
|
||||
@@ -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<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();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<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");
|
||||
}
|
||||
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<String, MeterRegistry> filterRegistries(Map<String, MeterRegistry> registries,
|
||||
ApplicationContext context) {
|
||||
|
||||
if (registries.size() == 1) {
|
||||
return registries;
|
||||
}
|
||||
MeterRegistry primary = null;
|
||||
if (context instanceof ConfigurableApplicationContext) {
|
||||
BeanDefinitionRegistry bdr = (BeanDefinitionRegistry) ((ConfigurableApplicationContext) context)
|
||||
.getBeanFactory();
|
||||
for (Entry<String, MeterRegistry> 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();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -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<String, Timer> meters = (Map<String, Timer>) 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();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user