diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/EnableIntegrationManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/config/EnableIntegrationManagement.java index a4f7a66216..746eab58d3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/EnableIntegrationManagement.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/EnableIntegrationManagement.java @@ -54,7 +54,7 @@ public @interface EnableIntegrationManagement { *

* Set this to false for disabling logging by default in all framework components that implement * {@link org.springframework.integration.support.management.IntegrationManagement} - * (channels, message handlers etc). It turns off logging such as "PreSend on channel", "Received message" etc. + * (channels, message handlers etc.). It turns off logging such as "PreSend on channel", "Received message" etc. *

* After the context is initialized, individual components can have their setting changed by invoking * {@link org.springframework.integration.support.management.IntegrationManagement#setLoggingEnabled(boolean)}. @@ -62,4 +62,15 @@ public @interface EnableIntegrationManagement { */ String defaultLoggingEnabled() default "true"; + /** + * Set simple pattern component names matching for observation registry injection. + * @return simple pattern component names matching for observation registry injection. + * None by default - no unconditional observation instrumentation. + * Can be set to {@code *} to instrumentation all the integration components. + * The pattern can start with {@code !} to negate the matching. + * @since 6.0 + * @see org.springframework.integration.support.utils.PatternMatchUtils#smartMatch(String, String...) + */ + String[] observationPatterns() default { }; + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfiguration.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfiguration.java index 35f53b18c0..de071c96f3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfiguration.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfiguration.java @@ -75,7 +75,11 @@ public class IntegrationManagementConfiguration implements ImportAware, Environm Boolean.parseBoolean(this.environment.resolvePlaceholders( (String) this.attributes.get("defaultLoggingEnabled")))); configurer.setMetricsCaptorProvider(metricsCaptorProvider); - configurer.setObservationRegistry(observationRegistryProvider); + String[] observationPatterns = (String[]) this.attributes.get("observationPatterns"); + if (observationPatterns.length > 0) { + configurer.setObservationPatterns(observationPatterns); + configurer.setObservationRegistry(observationRegistryProvider); + } return configurer; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java index 208e414b9f..bca7f98ae0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java @@ -19,6 +19,8 @@ package org.springframework.integration.config; import java.util.HashSet; import java.util.Set; +import org.apache.commons.logging.Log; + import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.ObjectProvider; @@ -33,6 +35,7 @@ import org.springframework.integration.support.management.IntegrationManagement; import org.springframework.integration.support.management.IntegrationManagement.ManagementOverrides; import org.springframework.integration.support.management.metrics.MeterFacade; import org.springframework.integration.support.management.metrics.MetricsCaptor; +import org.springframework.integration.support.utils.PatternMatchUtils; import org.springframework.lang.Nullable; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -80,6 +83,8 @@ public class IntegrationManagementConfigurer private ObjectProvider observationRegistryProvider; + private String[] observationPatterns; + @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = applicationContext; @@ -98,7 +103,7 @@ public class IntegrationManagementConfigurer * Exception logging (debug or otherwise) is not affected by this setting. *

* It has been found that in high-volume messaging environments, calls to methods such as - * {@link org.apache.commons.logging.Log#isDebugEnabled()} can be quite expensive + * {@link Log#isDebugEnabled()} can be quite expensive * and account for an inordinate amount of CPU time. *

* Set this to 'false' to disable logging by default in all framework components that implement @@ -135,6 +140,16 @@ public class IntegrationManagementConfigurer this.observationRegistryProvider = observationRegistryProvider; } + /** + * Set simple patterns for component names matching which has to be instrumented with a {@link ObservationRegistry}. + * @param observationPatterns the simple patterns to use. + * @since 6.0 + * @see PatternMatchUtils#smartMatch(String, String...) + */ + public void setObservationPatterns(String... observationPatterns) { + Assert.notEmpty(observationPatterns, "'observationPatterns' must not be empty"); + this.observationPatterns = observationPatterns; + } @Override public void afterSingletonsInstantiated() { @@ -171,19 +186,19 @@ public class IntegrationManagementConfigurer private void registerComponentGauges() { this.gauges.add( this.metricsCaptor.gaugeBuilder("spring.integration.channels", this, - (c) -> this.applicationContext.getBeansOfType(MessageChannel.class).size()) + (c) -> this.applicationContext.getBeansOfType(MessageChannel.class).size()) .description("The number of message channels") .build()); this.gauges.add( this.metricsCaptor.gaugeBuilder("spring.integration.handlers", this, - (c) -> this.applicationContext.getBeansOfType(MessageHandler.class).size()) + (c) -> this.applicationContext.getBeansOfType(MessageHandler.class).size()) .description("The number of message handlers") .build()); this.gauges.add( this.metricsCaptor.gaugeBuilder("spring.integration.sources", this, - (c) -> this.applicationContext.getBeansOfType(MessageSource.class).size()) + (c) -> this.applicationContext.getBeansOfType(MessageSource.class).size()) .description("The number of message sources") .build()); } @@ -195,7 +210,10 @@ public class IntegrationManagementConfigurer if (this.metricsCaptor != null) { integrationManagement.registerMetricsCaptor(this.metricsCaptor); } - if (this.observationRegistry != null) { + if (this.observationRegistry != null && + Boolean.TRUE.equals(PatternMatchUtils.smartMatch( + integrationManagement.getComponentName(), this.observationPatterns))) { + integrationManagement.registerObservationRegistry(this.observationRegistry); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java index 28071502d3..c0180fafb8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java @@ -21,6 +21,7 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationProperties; import org.springframework.integration.support.channel.ChannelResolverUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.GenericMessagingTemplate; @@ -80,6 +81,7 @@ public class MessagingTemplate extends GenericMessagingTemplate { } @Override + @Nullable public Message sendAndReceive(MessageChannel destination, Message requestMessage) { if (!this.throwExceptionOnLateReplySet) { synchronized (this) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index cd937eed2a..6370c18f8c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -31,7 +31,12 @@ import org.springframework.integration.history.MessageHistory; import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageUtils; +import org.springframework.integration.support.management.IntegrationInboundManagement; import org.springframework.integration.support.management.TrackableComponent; +import org.springframework.integration.support.management.observation.DefaultMessageReceiverObservationConvention; +import org.springframework.integration.support.management.observation.IntegrationObservation; +import org.springframework.integration.support.management.observation.MessageReceiverContext; +import org.springframework.integration.support.management.observation.MessageReceiverObservationConvention; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -40,6 +45,7 @@ import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.StringUtils; +import io.micrometer.observation.ObservationRegistry; import reactor.core.publisher.Flux; /** @@ -50,13 +56,19 @@ import reactor.core.publisher.Flux; * @author Artem Bilan * @author Gary Russell */ -public abstract class MessageProducerSupport extends AbstractEndpoint implements MessageProducer, TrackableComponent, - SmartInitializingSingleton, IntegrationPattern { +public abstract class MessageProducerSupport extends AbstractEndpoint + implements MessageProducer, TrackableComponent, + SmartInitializingSingleton, IntegrationPattern, IntegrationInboundManagement { private final MessagingTemplate messagingTemplate = new MessagingTemplate(); private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + private ObservationRegistry observationRegistry = ObservationRegistry.NOOP; + + @Nullable + private MessageReceiverObservationConvention observationConvention; + private MessageChannel outputChannel; private String outputChannelName; @@ -135,7 +147,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements /** * Configure the default timeout value to use for send operations. * May be overridden for individual messages. - * @param sendTimeout the send timeout in milliseconds + * @param sendTimeout the 'send' operation timeout in milliseconds * @see MessagingTemplate#setSendTimeout */ public void setSendTimeout(long sendTimeout) { @@ -172,6 +184,22 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements return this.messagingTemplate; } + /** + * Set a custom {@link MessageReceiverObservationConvention} for {@link IntegrationObservation#HANDLER}. + * Ignored if an {@link ObservationRegistry} is not configured for this component. + * @param observationConvention the {@link MessageReceiverObservationConvention} to use. + * @since 6.0 + */ + public void setObservationConvention(@Nullable MessageReceiverObservationConvention observationConvention) { + this.observationConvention = observationConvention; + } + + @Override + public void registerObservationRegistry(ObservationRegistry observationRegistry) { + Assert.notNull(observationRegistry, "'observationRegistry' must not be null"); + this.observationRegistry = observationRegistry; + } + @Override public IntegrationPatternType getIntegrationPatternType() { return IntegrationPatternType.inbound_channel_adapter; @@ -216,14 +244,18 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements } } - protected void sendMessage(Message messageArg) { - Message message = messageArg; + protected void sendMessage(Message message) { if (message == null) { throw new MessagingException("cannot send a null message"); } - message = trackMessageIfAny(message); + try { - this.messagingTemplate.send(getRequiredOutputChannel(), message); + IntegrationObservation.HANDLER.observation( + this.observationConvention, + DefaultMessageReceiverObservationConvention.INSTANCE, + () -> new MessageReceiverContext(message, getComponentName()), + this.observationRegistry) + .observe(() -> this.messagingTemplate.send(getRequiredOutputChannel(), trackMessageIfAny(message))); } catch (RuntimeException ex) { if (!sendErrorMessageIfNecessary(message, ex)) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index a6c770f3a9..333e31d686 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -53,6 +53,10 @@ import org.springframework.integration.support.management.metrics.MeterFacade; import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.integration.support.management.metrics.SampleFacade; import org.springframework.integration.support.management.metrics.TimerFacade; +import org.springframework.integration.support.management.observation.DefaultMessageRequestReplyReceiverObservationConvention; +import org.springframework.integration.support.management.observation.IntegrationObservation; +import org.springframework.integration.support.management.observation.MessageRequestReplyReceiverContext; +import org.springframework.integration.support.management.observation.MessageRequestReplyReceiverObservationConvention; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -62,10 +66,12 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.core.MessagePostProcessor; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; +import io.micrometer.observation.ObservationRegistry; import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; @@ -87,7 +93,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint private static final long DEFAULT_TIMEOUT = 1000L; - protected final MessagingTemplate messagingTemplate; // NOSONAR + protected final ConvertingMessagingTemplate messagingTemplate; // NOSONAR private final SimpleMessageConverter messageConverter = new SimpleMessageConverter(); @@ -130,6 +136,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint private TimerFacade successTimer; + private ObservationRegistry observationRegistry = ObservationRegistry.NOOP; + + @Nullable + private MessageRequestReplyReceiverObservationConvention observationConvention; + private volatile AbstractEndpoint replyMessageCorrelator; private volatile boolean initialized; @@ -152,7 +163,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint * @see #setErrorOnTimeout */ public MessagingGatewaySupport(boolean errorOnTimeout) { - MessagingTemplate template = new MessagingTemplate(); + ConvertingMessagingTemplate template = new ConvertingMessagingTemplate(); template.setMessageConverter(this.messageConverter); template.setSendTimeout(DEFAULT_TIMEOUT); template.setReceiveTimeout(this.replyTimeout); @@ -354,6 +365,18 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint this.metricsCaptor = metricsCaptorToRegister; } + @Override + public void registerObservationRegistry(ObservationRegistry observationRegistry) { + Assert.notNull(observationRegistry, "'observationRegistry' must not be null"); + this.observationRegistry = observationRegistry; + } + + public void setObservationConvention( + @Nullable MessageRequestReplyReceiverObservationConvention observationConvention) { + + this.observationConvention = observationConvention; + } + @Override protected void onInit() { Assert.state(!(this.requestChannelName != null && this.requestChannel != null), @@ -494,16 +517,16 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint @Nullable protected Object sendAndReceive(Object object) { - return doSendAndReceive(object, true); + return sendAndReceive(object, true); } @Nullable protected Message sendAndReceiveMessage(Object object) { - return (Message) doSendAndReceive(object, false); + return (Message) sendAndReceive(object, false); } - @Nullable // NOSONAR - private Object doSendAndReceive(Object object, boolean shouldConvert) { + @Nullable + private Object sendAndReceive(Object object, boolean shouldConvert) { initializeIfNecessary(); Assert.notNull(object, "request must not be null"); MessageChannel channel = getRequestChannel(); @@ -515,36 +538,28 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint Object reply; Message requestMessage = null; - SampleFacade sample = null; try { - if (this.metricsCaptor != null) { - sample = this.metricsCaptor.start(); + requestMessage = convertToRequestMessage(object, shouldConvert); + Message replyMessage; + + if (this.observationRegistry != ObservationRegistry.NOOP) { + replyMessage = sendAndReceiveWithObservation(channel, object, requestMessage); } - if (shouldConvert) { - reply = this.messagingTemplate.convertSendAndReceive(channel, object, Object.class, - this.historyWritingPostProcessor); + else if (this.metricsCaptor != null) { + replyMessage = sendAndReceiveWithMetrics(channel, object, requestMessage); } else { - requestMessage = (object instanceof Message) - ? (Message) object : this.requestMapper.toMessage(object); - Assert.state(requestMessage != null, () -> "request mapper resulted in no message for " + object); - requestMessage = this.historyWritingPostProcessor.postProcessMessage(requestMessage); - reply = this.messagingTemplate.sendAndReceive(channel, requestMessage); + replyMessage = doSendAndReceive(channel, object, requestMessage); } - if (reply == null && this.errorOnTimeout) { - throwMessageTimeoutException(object, "No reply received within timeout"); - } - if (sample != null) { - sample.stop(sendTimer()); + reply = replyMessage; + if (shouldConvert) { + reply = this.messagingTemplate.getMessageConverter().fromMessage(replyMessage, Object.class); } } catch (Throwable ex) { // NOSONAR (catch throwable) logger.debug(() -> "failure occurred in gateway sendAndReceive: " + ex.getMessage()); reply = ex; - if (sample != null) { - sample.stop(buildSendTimer(false, ex.getClass().getSimpleName())); - } } if (reply instanceof Throwable || reply instanceof ErrorMessage) { @@ -557,6 +572,57 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return reply; } + private Message convertToRequestMessage(Object object, boolean shouldConvert) { + if (shouldConvert) { + return this.messagingTemplate.doConvert(object, null, this.historyWritingPostProcessor); + } + else { + Message requestMessage = (object instanceof Message) + ? (Message) object : this.requestMapper.toMessage(object); + Assert.state(requestMessage != null, () -> "request mapper resulted in no message for " + object); + return this.historyWritingPostProcessor.postProcessMessage(requestMessage); + } + } + + private Message sendAndReceiveWithObservation(MessageChannel requestChannel, Object object, + Message requestMessage) { + + MessageRequestReplyReceiverContext context = + new MessageRequestReplyReceiverContext(requestMessage, getComponentName()); + + return IntegrationObservation.GATEWAY.observation(this.observationConvention, + DefaultMessageRequestReplyReceiverObservationConvention.INSTANCE, + () -> context, this.observationRegistry) + .observe(() -> { + Message replyMessage = doSendAndReceive(requestChannel, object, requestMessage); + context.setResponse(replyMessage); + return replyMessage; + }); + } + + private Message sendAndReceiveWithMetrics(MessageChannel requestChannel, Object object, + Message requestMessage) { + + SampleFacade sample = this.metricsCaptor.start(); + try { + Message replyMessage = doSendAndReceive(requestChannel, object, requestMessage); + sample.stop(sendTimer()); + return replyMessage; + } + catch (Throwable ex) { + sample.stop(buildSendTimer(false, ex.getClass().getSimpleName())); + throw ex; + } + } + + private Message doSendAndReceive(MessageChannel requestChannel, Object object, Message requestMessage) { + Message replyMessage = this.messagingTemplate.sendAndReceive(requestChannel, requestMessage); + if (replyMessage == null && this.errorOnTimeout) { + throwMessageTimeoutException(object, "No reply received within timeout"); + } + return replyMessage; + } + @Nullable private Object handleSendAndReceiveError(Object object, @Nullable Message requestMessage, Throwable error, boolean shouldConvert) { @@ -646,23 +712,23 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint } return Mono.defer(() -> { - Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel(); - Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel(); + Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel(); + Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel(); - MonoReplyChannel replyChan = new MonoReplyChannel(); + MonoReplyChannel replyChan = new MonoReplyChannel(); - Message messageToSend = MutableMessageBuilder.fromMessage(requestMessage) - .setReplyChannel(replyChan) - .setHeader(this.messagingTemplate.getSendTimeoutHeader(), null) - .setHeader(this.messagingTemplate.getReceiveTimeoutHeader(), null) - .setErrorChannel(replyChan) - .build(); + Message messageToSend = MutableMessageBuilder.fromMessage(requestMessage) + .setReplyChannel(replyChan) + .setHeader(this.messagingTemplate.getSendTimeoutHeader(), null) + .setHeader(this.messagingTemplate.getReceiveTimeoutHeader(), null) + .setErrorChannel(replyChan) + .build(); - sendMessageForReactiveFlow(requestChannel, messageToSend); + sendMessageForReactiveFlow(requestChannel, messageToSend); - return buildReplyMono(requestMessage, replyChan.replyMono.asMono(), error, originalReplyChannelHeader, - originalErrorChannelHeader); - }) + return buildReplyMono(requestMessage, replyChan.replyMono.asMono(), error, originalReplyChannelHeader, + originalErrorChannelHeader); + }) .onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t)); } @@ -916,4 +982,19 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint } + /** + * The {@link MessagingTemplate} extension to increase {@link #doConvert(Object, Map, MessagePostProcessor)} + * visibility to get access to the request message from an observation context. + */ + protected static class ConvertingMessagingTemplate extends MessagingTemplate { + + @Override // NOSONAR Increase visibility + public Message doConvert(Object payload, @Nullable Map headers, + @Nullable MessagePostProcessor postProcessor) { + + return super.doConvert(payload, headers, postProcessor); + } + + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/DefaultMessageRequestReplyReceiverObservationConvention.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/DefaultMessageRequestReplyReceiverObservationConvention.java new file mode 100644 index 0000000000..f01113f5ce --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/DefaultMessageRequestReplyReceiverObservationConvention.java @@ -0,0 +1,49 @@ +/* + * 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.integration.support.management.observation; + +import io.micrometer.common.KeyValues; + +/** + * A default {@link MessageRequestReplyReceiverObservationConvention} implementation. + * Provides low cardinalities as a {@link IntegrationObservation.GatewayTags} values. + * + * @author Artem Bilan + * + * @since 6.0 + */ +public class DefaultMessageRequestReplyReceiverObservationConvention + implements MessageRequestReplyReceiverObservationConvention { + + /** + * A shared singleton instance for {@link DefaultMessageRequestReplyReceiverObservationConvention}. + */ + public static final DefaultMessageRequestReplyReceiverObservationConvention INSTANCE = + new DefaultMessageRequestReplyReceiverObservationConvention(); + + @Override + public KeyValues getLowCardinalityKeyValues(MessageRequestReplyReceiverContext context) { + return KeyValues + // See IntegrationObservation.GatewayTags.COMPONENT_NAME - to avoid class tangle + .of("spring.integration.name", context.getGatewayName()) + // See IntegrationObservation.GatewayTags.COMPONENT_TYPE - to avoid class tangle + .and("spring.integration.type", "gateway") + // See IntegrationObservation.GatewayTags.OUTCOME - to avoid class tangle + .and("spring.integration.outcome", context.getError() != null ? "INTERNAL_ERROR" : "SUCCESS"); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/IntegrationObservation.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/IntegrationObservation.java index 547072ee11..7219fb41f1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/IntegrationObservation.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/IntegrationObservation.java @@ -21,6 +21,9 @@ import io.micrometer.observation.docs.ObservationDocumentation; /** * The {@link ObservationDocumentation} implementation for Spring Integration infrastructure. + *

+ * NOTE: This class is mostly intended for observation docs generation, so any string literals, + * even if they are the same, cannot be extracted into constants or super methods. * * @author Artem Bilan * @@ -47,6 +50,27 @@ public enum IntegrationObservation implements ObservationDocumentation { return HandlerTags.values(); } + }, + + /** + * Observation for inbound message gateways. + */ + GATEWAY { + @Override + public String getPrefix() { + return "spring.integration."; + } + + @Override + public Class getDefaultConvention() { + return DefaultMessageRequestReplyReceiverObservationConvention.class; + } + + @Override + public KeyName[] getLowCardinalityKeyNames() { + return GatewayTags.values(); + } + }; /** @@ -78,4 +102,43 @@ public enum IntegrationObservation implements ObservationDocumentation { } + /** + * Key names for message handler observations. + */ + public enum GatewayTags implements KeyName { + + /** + * Name of the message gateway component. + */ + COMPONENT_NAME { + @Override + public String asString() { + return "spring.integration.name"; + } + + }, + + /** + * Type of the component - 'gateway'. + */ + COMPONENT_TYPE { + @Override + public String asString() { + return "spring.integration.type"; + } + + }, + + /** + * Outcome of the request/reply execution. + */ + OUTCOME { + @Override + public String asString() { + return "spring.integration.outcome"; + } + }, + + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverObservationConvention.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverObservationConvention.java index 3bcda0de55..484fe023ac 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverObservationConvention.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverObservationConvention.java @@ -16,14 +16,11 @@ package org.springframework.integration.support.management.observation; -import org.springframework.messaging.Message; - import io.micrometer.observation.Observation; import io.micrometer.observation.ObservationConvention; -import io.micrometer.observation.transport.ReceiverContext; /** - * The {@link ReceiverContext} extension for {@link Message} context. + * A {@link MessageReceiverContext}-based {@link ObservationConvention} contract. * * @author Artem Bilan * diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageRequestReplyReceiverContext.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageRequestReplyReceiverContext.java new file mode 100644 index 0000000000..b86239f208 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageRequestReplyReceiverContext.java @@ -0,0 +1,52 @@ +/* + * 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.integration.support.management.observation; + +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; + +import io.micrometer.observation.transport.RequestReplyReceiverContext; + +/** + * The {@link RequestReplyReceiverContext} extension for a {@link Message} contract with inbound gateways. + * + * @author Artem Bilan + * + * @since 6.0 + */ +public class MessageRequestReplyReceiverContext extends RequestReplyReceiverContext, Message> { + + private final Message message; + + private final String gatewayName; + + public MessageRequestReplyReceiverContext(Message message, @Nullable String gatewayName) { + super((carrier, key) -> carrier.getHeaders().get(key, String.class)); + this.message = message; + this.gatewayName = gatewayName != null ? gatewayName : "unknown"; + } + + @Override + public Message getCarrier() { + return this.message; + } + + public String getGatewayName() { + return this.gatewayName; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageRequestReplyReceiverObservationConvention.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageRequestReplyReceiverObservationConvention.java new file mode 100644 index 0000000000..7db3862141 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageRequestReplyReceiverObservationConvention.java @@ -0,0 +1,47 @@ +/* + * 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.integration.support.management.observation; + +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationConvention; + +/** + * A {@link MessageRequestReplyReceiverContext}-based {@link ObservationConvention} contract. + * + * @author Artem Bilan + * + * @since 6.0 + */ +public interface MessageRequestReplyReceiverObservationConvention + extends ObservationConvention { + + @Override + default String getName() { + return "spring.integration.gateway"; + } + + @Override + default boolean supportsContext(Observation.Context context) { + return context instanceof MessageRequestReplyReceiverContext; + } + + @Override + default String getContextualName(MessageRequestReplyReceiverContext context) { + return context.getGatewayName() + " process"; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java index c95e67c8bf..5522a8e556 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java @@ -32,18 +32,17 @@ import org.springframework.integration.channel.interceptor.ObservationPropagatio import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.config.EnableIntegrationManagement; import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.handler.advice.HandleMessageAdvice; -import org.springframework.integration.support.MutableMessage; -import org.springframework.integration.support.MutableMessageBuilder; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.GenericMessage; import io.micrometer.common.KeyValues; import io.micrometer.core.tck.MeterRegistryAssert; -import io.micrometer.observation.Observation; import io.micrometer.observation.ObservationRegistry; import io.micrometer.tracing.Span; import io.micrometer.tracing.test.SampleTestRunner; @@ -70,18 +69,11 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { applicationContext.register(ObservationIntegrationTestConfiguration.class); applicationContext.refresh(); - PollableChannel queueChannel = applicationContext.getBean("queueChannel", PollableChannel.class); - PollableChannel replyChannel = new QueueChannel(); + TestMessagingGatewaySupport messagingGateway = + applicationContext.getBean(TestMessagingGatewaySupport.class); - MutableMessage message = - (MutableMessage) MutableMessageBuilder.withPayload("test data") - .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) - .build(); + Message receive = messagingGateway.process(new GenericMessage<>("test data")); - Observation.createNotStarted("Test send", () -> new MessageSenderContext(message), observationRegistry) - .observe(() -> queueChannel.send(message)); - - Message receive = replyChannel.receive(10_000); assertThat(receive).isNotNull() .extracting("payload").isEqualTo("test data"); var configuration = applicationContext.getBean(ObservationIntegrationTestConfiguration.class); @@ -91,7 +83,11 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { SpansAssert.assertThat(bb.getFinishedSpans()) .haveSameTraceId() - .hasASpanWithName("Test send", spanAssert -> spanAssert.hasKindEqualTo(Span.Kind.PRODUCER)) + .hasASpanWithName("testInboundGateway process", spanAssert -> spanAssert + .hasTag(IntegrationObservation.GatewayTags.COMPONENT_NAME.asString(), "testInboundGateway") + .hasTag(IntegrationObservation.GatewayTags.COMPONENT_TYPE.asString(), "gateway") + .hasTagWithKey("test.message.id") + .hasKindEqualTo(Span.Kind.SERVER)) .hasASpanWithName("observedEndpoint receive", spanAssert -> spanAssert .hasTag(IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(), "observedEndpoint") .hasTag(IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler") @@ -110,7 +106,7 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { @Configuration @EnableIntegration - @EnableIntegrationManagement + @EnableIntegrationManagement(observationPatterns = { "observedEndpoint", "testInboundGateway" }) public static class ObservationIntegrationTestConfiguration { CountDownLatch observedHandlerLatch = new CountDownLatch(1); @@ -121,6 +117,22 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { return new ObservationPropagationChannelInterceptor(observationRegistry); } + @Bean + TestMessagingGatewaySupport testInboundGateway(PollableChannel queueChannel) { + TestMessagingGatewaySupport messagingGatewaySupport = new TestMessagingGatewaySupport(); + messagingGatewaySupport.setObservationConvention( + new DefaultMessageRequestReplyReceiverObservationConvention() { + + @Override + public KeyValues getHighCardinalityKeyValues(MessageRequestReplyReceiverContext context) { + return KeyValues.of("test.message.id", context.getCarrier().getHeaders().getId().toString()); + } + + }); + messagingGatewaySupport.setRequestChannel(queueChannel); + return messagingGatewaySupport; + } + @Bean public PollableChannel queueChannel() { return new QueueChannel(); @@ -149,4 +161,13 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { } + private static class TestMessagingGatewaySupport extends MessagingGatewaySupport { + + @Nullable + Message process(Message request) { + return sendAndReceiveMessage(request); + } + + } + } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index 841afa1056..6d11be4a31 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -26,6 +26,7 @@ import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Predicate; import java.util.stream.Collectors; import javax.lang.model.SourceVersion; @@ -268,6 +269,7 @@ public class IntegrationMBeanExporter extends MBeanExporter .stream() // If the source is proxied, we have to extract the target to expose as an MBean. // The MetadataMBeanInfoAssembler does not support JDK dynamic proxies. + .filter(Predicate.not(MessageProducer.class::isInstance)) .map(this::extractTarget) .map(IntegrationInboundManagement.class::cast) .forEach(src -> this.sources.put(src.getComponentName(), src)); diff --git a/src/reference/asciidoc/metrics.adoc b/src/reference/asciidoc/metrics.adoc index 5192d2518b..3b1be9b980 100644 --- a/src/reference/asciidoc/metrics.adoc +++ b/src/reference/asciidoc/metrics.adoc @@ -150,7 +150,13 @@ registry.config().meterFilter(MeterFilter.deny(id -> Starting with version 6.0, Spring Integration utilizes a Micrometer Observation abstraction which can handle metrics as well as https://micrometer.io/docs/tracing[tracing] via appropriate `ObservationHandler` configuration. -The observation handling is enabled on the `IntegrationManagement` components whenever an `ObservationRegistry` bean is present in the application context. +The observation handling is enabled on the `IntegrationManagement` components whenever an `ObservationRegistry` bean is present in the application context and an `@EnableIntegrationManagement` is configured. +To customize what set of components should be instrumented, an `observationPatterns()` attribute is exposed on the `@EnableIntegrationManagement` annotation. +See its javadocs for a pattern matching algorithm. + +IMPORTANT: By default, none of the `IntegrationManagement` components are instrumented with an `ObservationRegistry` bean. +Can be configured as `*` to match all components. + The meters are not gathered in this case independently, but delegated to an appropriate `ObservationHandler` configured on the provided `ObservationRegistry`. An observation production on the `IntegrationManagement` components can be customized via `ObservationConvention` configuration.