Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4416: Rework Micrometer Metrics

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4416

Metrics should be under a common name, discriminated with tags.
This commit is contained in:
Gary Russell
2018-02-27 16:04:54 -05:00
committed by Artem Bilan
parent f26c3a9004
commit 792b8fe6d1
10 changed files with 256 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -48,6 +48,10 @@ import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import io.micrometer.core.instrument.Timer.Sample;
/**
* Base class for {@link MessageChannel} implementations providing common
* properties such as the channel name. Also provides the common functionality
@@ -86,6 +90,8 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
private volatile AbstractMessageChannelMetrics channelMetrics = new DefaultMessageChannelMetrics();
private MeterRegistry meterRegistry;
public AbstractMessageChannel() {
this.interceptors = new ChannelInterceptorList(logger);
}
@@ -100,6 +106,15 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
this.shouldTrack = shouldTrack;
}
@Override
public void registerMeterRegistry(MeterRegistry registry) {
this.meterRegistry = registry;
}
protected MeterRegistry getMeterRegistry() {
return this.meterRegistry;
}
@Override
public void setCountsEnabled(boolean countsEnabled) {
this.countsEnabled = countsEnabled;
@@ -417,6 +432,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
boolean countsEnabled = this.countsEnabled;
ChannelInterceptorList interceptors = this.interceptors;
AbstractMessageChannelMetrics channelMetrics = this.channelMetrics;
Sample sample = null;
if (this.meterRegistry != null) {
sample = Timer.start(this.meterRegistry);
}
try {
if (this.datatypes.length > 0) {
message = this.convertPayloadIfNecessary(message);
@@ -433,16 +452,22 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
}
}
if (countsEnabled) {
if (channelMetrics.getTimer() != null) {
final Message<?> messageToSend = message;
sent = channelMetrics.getTimer().recordCallable(() -> doSend(messageToSend, timeout));
metrics = channelMetrics.beforeSend();
if (this.meterRegistry != null) {
sample = Timer.start(this.meterRegistry);
}
else {
metrics = channelMetrics.beforeSend();
sent = doSend(message, timeout);
channelMetrics.afterSend(metrics, sent);
metricsProcessed = true;
sent = doSend(message, timeout);
if (sample != null) {
sample.stop(Timer.builder(SEND_TIMER_NAME)
.tag("type", "channel")
.tag("name", getComponentName() == null ? "unknown" : getComponentName())
.tag("result", sent ? "success" : "failure")
.tag("exception", "none")
.description("Subflow process time")
.register(this.meterRegistry));
}
channelMetrics.afterSend(metrics, sent);
metricsProcessed = true;
}
else {
sent = doSend(message, timeout);
@@ -459,12 +484,16 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
}
catch (Exception e) {
if (countsEnabled && !metricsProcessed) {
if (channelMetrics.getErrorCounter() != null) {
channelMetrics.getErrorCounter().increment();
}
else {
channelMetrics.afterSend(metrics, false);
if (sample != null) {
sample.stop(Timer.builder(SEND_TIMER_NAME)
.tag("type", "channel")
.tag("name", getComponentName() == null ? "unknown" : getComponentName())
.tag("result", "failure")
.tag("exception", e.getClass().getSimpleName())
.description("Subflow process time")
.register(this.meterRegistry));
}
channelMetrics.afterSend(metrics, false);
}
if (interceptorStack != null) {
interceptors.afterSendCompletion(message, this, sent, e, interceptorStack);

20
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java
View File

@@ -27,6 +27,8 @@ import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.ExecutorChannelInterceptor;
import org.springframework.util.CollectionUtils;
import io.micrometer.core.instrument.Counter;
/**
* Base class for all pollable channels.
*
@@ -104,6 +106,15 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
}
Message<?> message = this.doReceive(timeout);
if (countsEnabled && message != null) {
if (getMeterRegistry() != null) {
Counter.builder(RECEIVE_COUNTER_NAME)
.tag("name", getComponentName())
.tag("type", "channel")
.tag("result", "success")
.tag("exception", "none")
.description("Messages received")
.register(getMeterRegistry()).increment();
}
getMetrics().afterReceive();
counted = true;
}
@@ -121,6 +132,15 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
}
catch (RuntimeException e) {
if (countsEnabled && !counted) {
if (getMeterRegistry() != null) {
Counter.builder(RECEIVE_COUNTER_NAME)
.tag("name", getComponentName() == null ? "unknown" : getComponentName())
.tag("type", "channel")
.tag("result", "failure")
.tag("exception", e.getClass().getSimpleName())
.description("Messages received")
.register(getMeterRegistry()).increment();
}
getMetrics().afterError();
}
if (!CollectionUtils.isEmpty(interceptorStack)) {

21
spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.channel;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -32,6 +34,9 @@ import org.springframework.messaging.PollableChannel;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
/**
* A channel implementation that essentially behaves like "/dev/null".
* All receive() calls will return <em>null</em>, and all send() calls
@@ -59,6 +64,8 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
private String beanName;
private MeterRegistry meterRegistry;
@Override
public void setBeanName(String beanName) {
this.beanName = beanName;
@@ -86,6 +93,11 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
return "channel";
}
@Override
public void registerMeterRegistry(MeterRegistry registry) {
this.meterRegistry = registry;
}
@Override
public void configureMetrics(AbstractMessageChannelMetrics metrics) {
Assert.notNull(metrics, "'metrics' must not be null");
@@ -215,6 +227,15 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics,
this.logger.debug("message sent to null channel: " + message);
}
if (this.countsEnabled) {
if (this.meterRegistry != null) {
Timer.builder(SEND_TIMER_NAME)
.tag("type", "channel")
.tag("name", getComponentName() == null ? "unknown" : getComponentName())
.tag("result", "success")
.tag("exception", "none")
.description("Subflow process time")
.register(this.meterRegistry).record(0, TimeUnit.MILLISECONDS);
}
this.channelMetrics.afterSend(this.channelMetrics.beforeSend(), true);
}
return true;

22
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java
View File

@@ -34,6 +34,7 @@ import org.springframework.messaging.MessagingException;
import org.springframework.util.CollectionUtils;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
/**
* @author Mark Fisher
@@ -65,11 +66,18 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
private volatile boolean loggingEnabled = true;
private MeterRegistry meterRegistry;
public void setHeaderExpressions(Map<String, Expression> headerExpressions) {
this.headerExpressions = (headerExpressions != null)
? headerExpressions : Collections.emptyMap();
}
@Override
public void registerMeterRegistry(MeterRegistry registry) {
this.meterRegistry = registry;
}
@Override
public void setBeanName(String name) {
this.beanName = name;
@@ -191,12 +199,16 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
.build();
}
if (this.countsEnabled && message != null) {
if (this.counter != null) {
this.counter.increment();
}
else {
this.messageCount.incrementAndGet();
if (this.meterRegistry != null) {
Counter.builder(RECEIVE_COUNTER_NAME)
.tag("name", getComponentName() == null ? "unknown" : getComponentName())
.tag("type", "source")
.tag("result", "success")
.tag("exception", "none")
.description("Messages received")
.register(this.meterRegistry).increment();
}
this.messageCount.incrementAndGet();
}
return message;
}

52
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java
View File

@@ -36,6 +36,9 @@ import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import io.micrometer.core.instrument.Timer.Sample;
import reactor.core.CoreSubscriber;
/**
@@ -71,6 +74,8 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
private volatile boolean loggingEnabled = true;
private MeterRegistry meterRegistry;
@Override
public boolean isLoggingEnabled() {
return this.loggingEnabled;
@@ -82,6 +87,11 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
this.managementOverrides.loggingConfigured = true;
}
@Override
public void registerMeterRegistry(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
}
@Override
public void setOrder(int order) {
this.order = order;
@@ -131,36 +141,44 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
MetricsContext start = null;
boolean countsEnabled = this.countsEnabled;
AbstractMessageHandlerMetrics handlerMetrics = this.handlerMetrics;
Sample sample = null;
if (countsEnabled && this.meterRegistry != null) {
sample = Timer.start(this.meterRegistry);
}
try {
if (this.shouldTrack) {
message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
}
if (countsEnabled) {
if (handlerMetrics.getTimer() != null) {
final Message<?> messageToSend = message;
handlerMetrics.getTimer().recordCallable(() -> {
handleMessageInternal(messageToSend);
return null;
});
}
else {
start = handlerMetrics.beforeHandle();
handleMessageInternal(message);
handlerMetrics.afterHandle(start, true);
start = handlerMetrics.beforeHandle();
handleMessageInternal(message);
if (this.meterRegistry != null) {
sample.stop(Timer.builder(SEND_TIMER_NAME)
.tag("type", "handler")
.tag("name", getComponentName() == null ? "unknown" : getComponentName())
.tag("result", "success")
.tag("exception", "none")
.description("Subflow process time")
.register(this.meterRegistry));
}
handlerMetrics.afterHandle(start, true);
}
else {
handleMessageInternal(message);
}
}
catch (Exception e) {
if (sample != null) {
sample.stop(Timer.builder(SEND_TIMER_NAME)
.tag("type", "handler")
.tag("name", getComponentName() == null ? "unknown" : getComponentName())
.tag("result", "failure")
.tag("exception", e.getClass().getSimpleName())
.description("Subflow process time")
.register(this.meterRegistry));
}
if (countsEnabled) {
if (handlerMetrics.getErrorCounter() != null) {
handlerMetrics.getErrorCounter().increment();
}
else {
handlerMetrics.afterHandle(start, false);
}
handlerMetrics.afterHandle(start, false);
}
if (e instanceof MessagingException) {
throw (MessagingException) e;

17
spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java
View File

@@ -19,6 +19,8 @@ package org.springframework.integration.support.management;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import io.micrometer.core.instrument.MeterRegistry;
/**
* Base interface for Integration managed components.
*
@@ -28,6 +30,12 @@ import org.springframework.jmx.export.annotation.ManagedOperation;
*/
public interface IntegrationManagement {
String METER_PREFIX = "spring.integration.";
String SEND_TIMER_NAME = METER_PREFIX + "send";
String RECEIVE_COUNTER_NAME = METER_PREFIX + "receive";
@ManagedAttribute(description = "Use to disable debug logging during normal message flow")
void setLoggingEnabled(boolean enabled);
@@ -50,6 +58,15 @@ public interface IntegrationManagement {
*/
ManagementOverrides getOverrides();
/**
* Inject a micrometer {@link MeterRegistry}
* @param registry the registry.
* @since 5.0.3
*/
default void registerMeterRegistry(MeterRegistry registry) {
// no op
}
/**
* Toggles to inform the management configurer to not set these properties since
* the user has manually configured them in a bean definition. If true, the

53
spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagementConfigurer.java
View File

@@ -26,15 +26,22 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.management.IntegrationManagement.ManagementOverrides;
import org.springframework.integration.util.PatternMatchUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
/**
* Configures beans that implement {@link IntegrationManagement}.
@@ -82,6 +89,8 @@ public class IntegrationManagementConfigurer implements SmartInitializingSinglet
private volatile boolean singletonsInstantiated;
private MeterRegistry meterRegistry;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
@@ -205,6 +214,16 @@ public class IntegrationManagementConfigurer implements SmartInitializingSinglet
Assert.state(this.applicationContext != null, "'applicationContext' must not be null");
Assert.state(MANAGEMENT_CONFIGURER_NAME.equals(this.beanName), getClass().getSimpleName()
+ " bean name must be " + MANAGEMENT_CONFIGURER_NAME);
try {
this.meterRegistry = this.applicationContext.getBean(MeterRegistry.class);
}
catch (NoSuchBeanDefinitionException e) {
// no op
}
if (this.meterRegistry != null) {
injectRegistry(this.meterRegistry);
registerComponentGauges(this.meterRegistry);
}
if (this.metricsFactory == null && StringUtils.hasText(this.metricsFactoryBeanName)) {
this.metricsFactory = this.applicationContext.getBean(this.metricsFactoryBeanName, MetricsFactory.class);
}
@@ -230,9 +249,26 @@ public class IntegrationManagementConfigurer implements SmartInitializingSinglet
this.singletonsInstantiated = true;
}
/**
* @param registry
*/
private void injectRegistry(MeterRegistry registry) {
Map<String, IntegrationManagement> managed = this.applicationContext.getBeansOfType(IntegrationManagement.class);
for (Entry<String, IntegrationManagement> entry : managed.entrySet()) {
IntegrationManagement bean = entry.getValue();
if (!bean.getOverrides().loggingConfigured) {
bean.setLoggingEnabled(this.defaultLoggingEnabled);
}
bean.registerMeterRegistry(registry);
}
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (this.singletonsInstantiated) {
if (bean instanceof IntegrationManagement) {
((IntegrationManagement) bean).registerMeterRegistry(this.meterRegistry);
}
return doConfigureMetrics(bean, beanName);
}
return bean;
@@ -334,6 +370,23 @@ public class IntegrationManagementConfigurer implements SmartInitializingSinglet
this.sourcesByName.put(bean.getManagedName() != null ? bean.getManagedName() : name, bean);
}
private void registerComponentGauges(MeterRegistry meterRegistry) {
Gauge.builder("spring.integration.channels", this,
(c) -> this.applicationContext.getBeansOfType(MessageChannel.class).size())
.description("The number of message channels")
.register(meterRegistry);
Gauge.builder("spring.integration.handlers", this,
(c) -> this.applicationContext.getBeansOfType(MessageHandler.class).size())
.description("The number of message handlers")
.register(meterRegistry);
Gauge.builder("spring.integration.sources", this,
(c) -> this.applicationContext.getBeansOfType(MessageSource.class).size())
.description("The number of message sources")
.register(meterRegistry);
}
public String[] getChannelNames() {
return this.channelsByName.keySet().toArray(new String[this.channelsByName.size()]);
}

3
spring-integration-core/src/main/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsFactory.java
View File

@@ -48,8 +48,11 @@ import io.micrometer.core.instrument.MeterRegistry;
*
* @since 5.0.2
*
* @deprecated - micrometer metrics are now in-built.
*
* @see org.springframework.integration.support.management.IntegrationManagementConfigurer
*/
@Deprecated
public class MicrometerMetricsFactory implements MetricsFactory, MessageSourceMetricsConfigurer,
ApplicationContextAware, SmartInitializingSingleton {