From 8c73a2d0cd3b09201828630718119dd899eb5cb3 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 19 Sep 2022 10:22:04 -0400 Subject: [PATCH] Add some infrastructure for Observation (#3879) * Add some infrastructure for Observation * Populate an `ObservationRegistry` bean from the `IntegrationManagementConfigurer` into all the `IntegrationManagement` components * Introduce `MessageReceiverContext` and `MessageSenderContext` for easier usage in the target code * Implement `Observation` handling in the `AbstractMessageHandler` * Modify `ObservationPropagationChannelInterceptorTests` for new `MessageSenderContext` * Use `BridgeHandler` to ensure that `Observation` is propagated and handled properly * Verify that tags from the `AbstractMessageHandler` are preset on the consumer span * * Add a `DocumentedObservation` infrastructure * * Add `Timer` verification to the propagation test * * Update to the latest Observation API * * Add custom observation convention support for the `AbstractMessageHandler` * Use more meaningful prefix for Spring Integration tags * * Move singleton instance for `DefaultMessageReceiverObservationConvention` into `DefaultMessageReceiverObservationConvention` per se as an `INSTANCE` constant * Use `MeterRegistryAssert` in the `ObservationPropagationChannelInterceptorTests` to verify meters emitted * * And an integration test with Zipkin based on the `SampleTestRunner` --- build.gradle | 7 +- .../IntegrationManagementConfiguration.java | 10 +- .../IntegrationManagementConfigurer.java | 65 ++++++--- .../handler/AbstractMessageHandler.java | 46 ++++++- .../handler/MessageHandlerSupport.java | 15 ++- .../management/IntegrationManagement.java | 23 +++- ...tMessageReceiverObservationConvention.java | 44 +++++++ .../observation/IntegrationObservation.java | 86 ++++++++++++ .../observation/MessageReceiverContext.java | 52 ++++++++ .../MessageReceiverObservationConvention.java | 45 +++++++ .../observation/MessageSenderContext.java | 38 ++++++ ...ionPropagationChannelInterceptorTests.java | 114 +++++++++------- .../IntegrationObservabilityZipkinTests.java | 123 ++++++++++++++++++ 13 files changed, 595 insertions(+), 73 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/DefaultMessageReceiverObservationConvention.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/IntegrationObservation.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverContext.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverObservationConvention.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageSenderContext.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java diff --git a/build.gradle b/build.gradle index 1dd8b0b55f..6ce957f3ea 100644 --- a/build.gradle +++ b/build.gradle @@ -528,7 +528,12 @@ project('spring-integration-core') { testImplementation "org.aspectj:aspectjweaver:$aspectjVersion" testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion" testImplementation 'io.micrometer:micrometer-observation-test' - testImplementation 'io.micrometer:micrometer-tracing-test' + testImplementation ('io.micrometer:micrometer-tracing-integration-test') { + exclude group: 'io.opentelemetry' + exclude group: 'com.wavefront' + exclude group: 'io.micrometer', module: 'micrometer-tracing-bridge-otel' + + } } dokkaHtmlPartial { 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 807bb9876d..35f53b18c0 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-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. @@ -31,6 +31,8 @@ import org.springframework.core.type.AnnotationMetadata; import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.util.Assert; +import io.micrometer.observation.ObservationRegistry; + /** * {@code @Configuration} class that registers a {@link IntegrationManagementConfigurer} bean. * @@ -64,12 +66,16 @@ public class IntegrationManagementConfiguration implements ImportAware, Environm @Bean(name = IntegrationManagementConfigurer.MANAGEMENT_CONFIGURER_NAME) @Role(BeanDefinition.ROLE_INFRASTRUCTURE) - public IntegrationManagementConfigurer managementConfigurer(ObjectProvider metricsCaptorProvider) { + public IntegrationManagementConfigurer managementConfigurer( + ObjectProvider metricsCaptorProvider, + ObjectProvider observationRegistryProvider) { + IntegrationManagementConfigurer configurer = new IntegrationManagementConfigurer(); configurer.setDefaultLoggingEnabled( Boolean.parseBoolean(this.environment.resolvePlaceholders( (String) this.attributes.get("defaultLoggingEnabled")))); configurer.setMetricsCaptorProvider(metricsCaptorProvider); + 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 f40be94a88..aaffea57d1 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-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. @@ -38,10 +38,12 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; +import io.micrometer.observation.ObservationRegistry; + /** * Configures beans that implement {@link IntegrationManagement}. - * Configures counts, stats, logging for all (or selected) components. + * Configures logging, {@link MetricsCaptor} and {@link ObservationRegistry} for all (or selected) components. * * @author Gary Russell * @author Artem Bilan @@ -74,6 +76,10 @@ public class IntegrationManagementConfigurer private ObjectProvider metricsCaptorProvider; + private ObservationRegistry observationRegistry; + + private ObjectProvider observationRegistryProvider; + @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = applicationContext; @@ -95,8 +101,8 @@ public class IntegrationManagementConfigurer * {@link org.apache.commons.logging.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 - * {@link IntegrationManagement} (channels, message handlers etc). This turns off logging such as + * Set this to 'false' to disable logging by default in all framework components that implement + * {@link IntegrationManagement} (channels, message handlers etc.) This 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 @@ -115,14 +121,21 @@ public class IntegrationManagementConfigurer this.metricsCaptorProvider = metricsCaptorProvider; } - @Nullable - MetricsCaptor obtainMetricsCaptor() { - if (this.metricsCaptor == null && this.metricsCaptorProvider != null) { - this.metricsCaptor = this.metricsCaptorProvider.getIfUnique(); - } - return this.metricsCaptor; + /** + * Set an {@link ObservationRegistry} to populate to the {@link IntegrationManagement} components + * in the application context. + * @param observationRegistry the {@link ObservationRegistry} to use. + * @since 6.0 + */ + public void setObservationRegistry(@Nullable ObservationRegistry observationRegistry) { + this.observationRegistry = observationRegistry; } + void setObservationRegistry(ObjectProvider observationRegistryProvider) { + this.observationRegistryProvider = observationRegistryProvider; + } + + @Override public void afterSingletonsInstantiated() { Assert.state(this.applicationContext != null, "'applicationContext' must not be null"); @@ -133,15 +146,29 @@ public class IntegrationManagementConfigurer registerComponentGauges(); } - for (IntegrationManagement integrationManagement : - this.applicationContext.getBeansOfType(IntegrationManagement.class).values()) { + setupObservationRegistry(); - enhanceIntegrationManagement(integrationManagement); - } + this.applicationContext.getBeansOfType(IntegrationManagement.class).values() + .forEach(this::enhanceIntegrationManagement); this.singletonsInstantiated = true; } + @Nullable + private MetricsCaptor obtainMetricsCaptor() { + if (this.metricsCaptor == null && this.metricsCaptorProvider != null) { + this.metricsCaptor = this.metricsCaptorProvider.getIfUnique(); + } + return this.metricsCaptor; + } + + @Nullable + private void setupObservationRegistry() { + if (this.observationRegistry == null && this.observationRegistryProvider != null) { + this.observationRegistry = this.observationRegistryProvider.getIfUnique(); + } + } + private void registerComponentGauges() { this.gauges.add( this.metricsCaptor.gaugeBuilder("spring.integration.channels", this, @@ -169,17 +196,21 @@ public class IntegrationManagementConfigurer if (this.metricsCaptor != null) { integrationManagement.registerMetricsCaptor(this.metricsCaptor); } + if (this.observationRegistry != null) { + integrationManagement.registerObservationRegistry(this.observationRegistry); + } } @Override public Object postProcessAfterInitialization(Object bean, String name) throws BeansException { - if (this.singletonsInstantiated && bean instanceof IntegrationManagement) { - enhanceIntegrationManagement((IntegrationManagement) bean); + if (this.singletonsInstantiated && bean instanceof IntegrationManagement integrationManagement) { + enhanceIntegrationManagement(integrationManagement); } return bean; } - @Override public void onApplicationEvent(ContextClosedEvent event) { + @Override + public void onApplicationEvent(ContextClosedEvent event) { if (event.getApplicationContext().equals(this.applicationContext)) { this.gauges.forEach(MeterFacade::remove); this.gauges.clear(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index b23a46a90d..8ae44fedbb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -21,11 +21,17 @@ import org.reactivestreams.Subscription; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.integration.support.management.metrics.SampleFacade; +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.integration.support.utils.IntegrationUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; +import io.micrometer.observation.ObservationRegistry; import reactor.core.CoreSubscriber; /** @@ -37,21 +43,49 @@ import reactor.core.CoreSubscriber; public abstract class AbstractMessageHandler extends MessageHandlerSupport implements MessageHandler, CoreSubscriber> { + @Nullable + private MessageReceiverObservationConvention observationConvention; + + /** + * 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 // NOSONAR public void handleMessage(Message message) { Assert.notNull(message, "Message must not be null"); - if (isLoggingEnabled() && this.logger.isDebugEnabled()) { - this.logger.debug(this + " received message: " + message); + if (isLoggingEnabled()) { + this.logger.debug(() -> this + " received message: " + message); } - MetricsCaptor metricsCaptor = getMetricsCaptor(); - if (metricsCaptor != null) { - handleWithMetrics(message, metricsCaptor); + ObservationRegistry observationRegistry = getObservationRegistry(); + if (observationRegistry != null) { + handleWithObservation(message, observationRegistry); } else { - doHandleMessage(message); + MetricsCaptor metricsCaptor = getMetricsCaptor(); + if (metricsCaptor != null) { + handleWithMetrics(message, metricsCaptor); + } + else { + doHandleMessage(message); + } } } + private void handleWithObservation(Message message, ObservationRegistry observationRegistry) { + IntegrationObservation.HANDLER.observation( + this.observationConvention, + DefaultMessageReceiverObservationConvention.INSTANCE, + new MessageReceiverContext(message, getComponentName()), + observationRegistry) + .observe(() -> doHandleMessage(message)); + } + private void handleWithMetrics(Message message, MetricsCaptor metricsCaptor) { SampleFacade sample = metricsCaptor.start(); try { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java index 6b3f5812a6..793fbc79ba 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2020 the original author or authors. + * Copyright 2019-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. @@ -31,6 +31,8 @@ import org.springframework.integration.support.management.metrics.MeterFacade; import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.integration.support.management.metrics.TimerFacade; +import io.micrometer.observation.ObservationRegistry; + /** * Base class for Message handling components that provides basic validation and error * handling capabilities. Asserts that the incoming Message is not null and that it does @@ -61,6 +63,8 @@ public abstract class MessageHandlerSupport extends IntegrationObjectSupport private MetricsCaptor metricsCaptor; + private ObservationRegistry observationRegistry; + private int order = Ordered.LOWEST_PRECEDENCE; private String managedName; @@ -89,6 +93,15 @@ public abstract class MessageHandlerSupport extends IntegrationObjectSupport return this.metricsCaptor; } + @Override + public void registerObservationRegistry(ObservationRegistry observationRegistry) { + this.observationRegistry = observationRegistry; + } + + protected ObservationRegistry getObservationRegistry() { + return this.observationRegistry; + } + @Override public void setOrder(int order) { this.order = order; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java index 0bb5de4685..0bcd7b70aa 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2021 the original author or authors. + * Copyright 2015-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. @@ -22,10 +22,14 @@ import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.lang.Nullable; +import io.micrometer.observation.ObservationRegistry; + /** * Base interface for Integration managed components. * * @author Gary Russell + * @author Artem Bilan + * * @since 4.2 * */ @@ -39,7 +43,7 @@ public interface IntegrationManagement extends NamedComponent, DisposableBean { /** * Enable logging or not. - * @param enabled dalse to disable. + * @param enabled false to disable. */ @ManagedAttribute(description = "Use to disable debug logging during normal message flow") default void setLoggingEnabled(boolean enabled) { @@ -80,13 +84,28 @@ public interface IntegrationManagement extends NamedComponent, DisposableBean { /** * Inject a {@link MetricsCaptor}. + * Ignored if {@link ObservationRegistry} is provided. * @param captor the captor. * @since 5.0.4 + * @see #registerObservationRegistry(ObservationRegistry) */ default void registerMetricsCaptor(MetricsCaptor captor) { // no op } + /** + * Inject an {@link ObservationRegistry}. + * If provided, the {@link MetricsCaptor} is ignored. + * The meters capturing has to be configured as an {@link io.micrometer.observation.ObservationHandler} + * on the provided {@link ObservationRegistry}. + * @param observationRegistry the {@link ObservationRegistry} to expose observations from the component. + * @since 6.0 + * @see #registerMetricsCaptor(MetricsCaptor) + */ + default void registerObservationRegistry(ObservationRegistry observationRegistry) { + // no op + } + @Override default void destroy() { // no op diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/DefaultMessageReceiverObservationConvention.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/DefaultMessageReceiverObservationConvention.java new file mode 100644 index 0000000000..9c8a5b84c8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/DefaultMessageReceiverObservationConvention.java @@ -0,0 +1,44 @@ +/* + * 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 MessageReceiverObservationConvention} implementation. + * Provides low cardinalities as a {@link IntegrationObservation.HandlerTags} values. + * + * @author Artem Bilan + * + * @since 6.0 + */ +public class DefaultMessageReceiverObservationConvention implements MessageReceiverObservationConvention { + + /** + * A shared singleton instance for {@link DefaultMessageReceiverObservationConvention}. + */ + public static final DefaultMessageReceiverObservationConvention INSTANCE = + new DefaultMessageReceiverObservationConvention(); + + @Override + public KeyValues getLowCardinalityKeyValues(MessageReceiverContext context) { + return KeyValues.of( + IntegrationObservation.HandlerTags.COMPONENT_NAME.withValue(context.getHandlerName()), + IntegrationObservation.HandlerTags.COMPONENT_TYPE.withValue("handler")); + } + +} 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 new file mode 100644 index 0000000000..87780f6ab1 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/IntegrationObservation.java @@ -0,0 +1,86 @@ +/* + * 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.docs.KeyName; +import io.micrometer.observation.docs.DocumentedObservation; + +/** + * The {@link DocumentedObservation} implementation for Spring Integration infrastructure. + * + * @author Artem Bilan + * + * @since 6.0 + */ +public enum IntegrationObservation implements DocumentedObservation { + + /** + * Observation for message handlers. + */ + HANDLER { + @Override + public String getName() { + return "spring.integration.handler"; + } + + @Override + public String getPrefix() { + return "spring.integration."; + } + + @Override + public Class getDefaultConvention() { + return DefaultMessageReceiverObservationConvention.class; + } + + @Override + public KeyName[] getLowCardinalityKeyNames() { + return HandlerTags.values(); + } + + }; + + /** + * Key names for message handler observations. + */ + public enum HandlerTags implements KeyName { + + /** + * Name of the message handler component. + */ + COMPONENT_NAME { + @Override + public String asString() { + return "spring.integration.name"; + } + + }, + + /** + * Type of the component - 'handler'. + */ + COMPONENT_TYPE { + @Override + public String asString() { + return "spring.integration.type"; + } + + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverContext.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverContext.java new file mode 100644 index 0000000000..7003158ecb --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverContext.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.ReceiverContext; + +/** + * The {@link ReceiverContext} extension for {@link Message} context. + * + * @author Artem Bilan + * + * @since 6.0 + */ +public class MessageReceiverContext extends ReceiverContext> { + + private final Message message; + + private final String handlerName; + + public MessageReceiverContext(Message message, @Nullable String handlerName) { + super((carrier, key) -> carrier.getHeaders().get(key, String.class)); + this.message = message; + this.handlerName = handlerName != null ? handlerName : "unknown"; + } + + @Override + public Message getCarrier() { + return this.message; + } + + public String getHandlerName() { + return this.handlerName; + } + +} 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 new file mode 100644 index 0000000000..2f6ec9245f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageReceiverObservationConvention.java @@ -0,0 +1,45 @@ +/* + * 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.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. + * + * @author Artem Bilan + * + * @since 6.0 + */ +public interface MessageReceiverObservationConvention + extends ObservationConvention { + + @Override + default boolean supportsContext(Observation.Context context) { + return context instanceof MessageReceiverContext; + } + + @Override + default String getContextualName(MessageReceiverContext context) { + return context.getHandlerName() + " receive"; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageSenderContext.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageSenderContext.java new file mode 100644 index 0000000000..d007fe97f0 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/MessageSenderContext.java @@ -0,0 +1,38 @@ +/* + * 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.integration.support.MutableMessage; +import org.springframework.messaging.Message; + +import io.micrometer.observation.transport.SenderContext; + +/** + * The {@link SenderContext} extension for {@link Message} context. + * + * @author Artem Bilan + * + * @since 6.0 + */ +public class MessageSenderContext extends SenderContext> { + + public MessageSenderContext(MutableMessage message) { + super((carrier, key, value) -> carrier.getHeaders().put(key, value)); + setCarrier(message); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptorTests.java index c147ff544c..911c7d9671 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptorTests.java @@ -21,12 +21,12 @@ import static org.assertj.core.api.Assertions.assertThat; import java.util.Arrays; import java.util.Collection; import java.util.List; -import java.util.Objects; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -40,22 +40,30 @@ import org.springframework.integration.channel.ExecutorChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.handler.BridgeHandler; +import org.springframework.integration.support.MutableMessage; +import org.springframework.integration.support.MutableMessageBuilder; +import org.springframework.integration.support.management.observation.IntegrationObservation; +import org.springframework.integration.support.management.observation.MessageSenderContext; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.GenericMessage; -import org.springframework.messaging.support.MessageBuilder; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import io.micrometer.common.KeyValues; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.micrometer.core.tck.MeterRegistryAssert; import io.micrometer.observation.Observation; import io.micrometer.observation.ObservationHandler; import io.micrometer.observation.ObservationRegistry; import io.micrometer.observation.tck.TestObservationRegistry; import io.micrometer.observation.tck.TestObservationRegistryAssert; -import io.micrometer.observation.transport.ReceiverContext; -import io.micrometer.observation.transport.SenderContext; import io.micrometer.tracing.Span; import io.micrometer.tracing.TraceContext; import io.micrometer.tracing.Tracer; @@ -79,6 +87,9 @@ public class ObservationPropagationChannelInterceptorTests { @Autowired ObservationRegistry observationRegistry; + @Autowired + MeterRegistry meterRegistry; + @Autowired SimpleTracer simpleTracer; @@ -199,36 +210,29 @@ public class ObservationPropagationChannelInterceptorTests { } @Test - @SuppressWarnings({ "unchecked", "rawtypes" }) - void observationContextPropagatedOverDirectChannel() throws InterruptedException { - CountDownLatch handleLatch = new CountDownLatch(1); - this.testTracingChannel.subscribe(m -> { - // This would be the instrumentation code on the receiver side - // We would need to check if Zipkin wouldn't require us to create the receiving span and then an additional one for the user code... - ReceiverContext> receiverContext = - new ReceiverContext<>((carrier, key) -> carrier.getHeaders().get(key, String.class)); - receiverContext.setCarrier(m); + void observationContextPropagatedOverExecutorChannel() { + BridgeHandler handler = new BridgeHandler(); + handler.registerObservationRegistry(this.observationRegistry); + handler.setBeanName("testBridge"); + this.testTracingChannel.subscribe(handler); - // ...and this would be the user's code - Observation.createNotStarted("user.code", receiverContext, this.observationRegistry) - .observe(() -> { - // Let's assume that this is the user code - }); - handleLatch.countDown(); - }); + QueueChannel replyChannel = new QueueChannel(); - // This would be the instrumentation code on the sender side (user's code would call e.g. MessageTemplate and this code - // would lay in MessageTemplate) - // We need to mutate the carrier, so we need to use the builder not the message since message headers are immutable - SenderContext> senderContext = - new SenderContext<>((carrier, key, value) -> Objects.requireNonNull(carrier).setHeader(key, value)); - MessageBuilder builder = MessageBuilder.withPayload("test"); - senderContext.setCarrier(builder); + MutableMessage message = + (MutableMessage) MutableMessageBuilder.withPayload("test") + .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) + .build(); - Observation.createNotStarted("sending", senderContext, this.observationRegistry) - .observe(() -> this.testTracingChannel.send(builder.build())); + Observation.createNotStarted("sending", new MessageSenderContext(message), this.observationRegistry) + .observe(() -> this.testTracingChannel.send(message)); - assertThat(handleLatch.await(10, TimeUnit.SECONDS)).isTrue(); + Message receive = replyChannel.receive(); + + assertThat(receive).isNotNull() + .extracting(Message::getHeaders) + .asInstanceOf(InstanceOfAssertFactories.MAP) + .containsEntry("foo", "some foo value") + .containsEntry("bar", "some bar value"); TestObservationRegistryAssert.assertThat(this.observationRegistry) .doesNotHaveAnyRemainingCurrentObservation(); @@ -236,12 +240,28 @@ public class ObservationPropagationChannelInterceptorTests { TracerAssert.assertThat(this.simpleTracer) .reportedSpans() .hasSize(2) - .satisfies(simpleSpans -> SpansAssert.assertThat((Collection) (Collection) simpleSpans) + .satisfies(simpleSpans -> assertSpans(simpleSpans) .hasASpanWithName("sending") - .assertThatASpanWithNameEqualTo("user.code") + .assertThatASpanWithNameEqualTo("testBridge receive") .hasTag("foo", "some foo value") .hasTag("bar", "some bar value") + .hasTag("spring.integration.type", "handler") + .hasTag("spring.integration.name", "testBridge") .hasKindEqualTo(Span.Kind.CONSUMER)); + + + MeterRegistryAssert.assertThat(this.meterRegistry) + .hasTimerWithNameAndTags("spring.integration.handler", + KeyValues.of(IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(), "testBridge", + IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler", + "error", "none")); + + assertThat(this.meterRegistry.get("spring.integration.handler").timer().count()).isEqualTo(1); + } + + @SuppressWarnings("unchecked") + private static SpansAssert assertSpans(Collection actual) { + return SpansAssert.assertThat((Collection) actual); } @Configuration @@ -254,17 +274,24 @@ public class ObservationPropagationChannelInterceptorTests { } @Bean - ObservationRegistry observationRegistry(Tracer tracer, Propagator propagator) { + MeterRegistry meterRegistry() { + return new SimpleMeterRegistry(); + } + + @Bean + ObservationRegistry observationRegistry(Tracer tracer, Propagator propagator, MeterRegistry meterRegistry) { TestObservationRegistry observationRegistry = TestObservationRegistry.create(); - observationRegistry.observationConfig().observationHandler( - // Composite will pick the first matching handler - new ObservationHandler.FirstMatchingCompositeObservationHandler( - // This is responsible for creating a child span on the sender side - new PropagatingSenderTracingObservationHandler<>(tracer, propagator), - // This is responsible for creating a span on the receiver side - new PropagatingReceiverTracingObservationHandler<>(tracer, propagator), - // This is responsible for creating a default span - new DefaultTracingObservationHandler(tracer))); + observationRegistry.observationConfig() + .observationHandler(new DefaultMeterObservationHandler(meterRegistry)) + .observationHandler( + // Composite will pick the first matching handler + new ObservationHandler.FirstMatchingCompositeObservationHandler( + // This is responsible for creating a child span on the sender side + new PropagatingSenderTracingObservationHandler<>(tracer, propagator), + // This is responsible for creating a span on the receiver side + new PropagatingReceiverTracingObservationHandler<>(tracer, propagator), + // This is responsible for creating a default span + new DefaultTracingObservationHandler(tracer))); return observationRegistry; } @@ -318,14 +345,13 @@ public class ObservationPropagationChannelInterceptorTests { setter.set(carrier, "bar", "some bar value"); } - // This is called on the consumer side when the message is consumed // Normally we would use tools like Extractor from tracing but for tests we are just manually creating a span @Override public Span.Builder extract(C carrier, Getter getter) { String foo = getter.get(carrier, "foo"); String bar = getter.get(carrier, "bar"); - return tracer.spanBuilder().kind(Span.Kind.CONSUMER).tag("foo", foo).tag("bar", bar); + return tracer.spanBuilder().tag("foo", foo).tag("bar", bar); } }; } 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 new file mode 100644 index 0000000000..f008524833 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java @@ -0,0 +1,123 @@ +/* + * 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 static org.assertj.core.api.Assertions.assertThat; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.BridgeTo; +import org.springframework.integration.annotation.EndpointId; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.channel.interceptor.ObservationPropagationChannelInterceptor; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.config.EnableIntegrationManagement; +import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.support.MutableMessage; +import org.springframework.integration.support.MutableMessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ChannelInterceptor; + +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; +import io.micrometer.tracing.test.simple.SpansAssert; + +/** + * @author Artem Bilan + * + * @since 6.0 + */ +public class IntegrationObservabilityZipkinTests extends SampleTestRunner { + + @Override + public TracingSetup[] getTracingSetup() { + return new TracingSetup[]{ TracingSetup.IN_MEMORY_BRAVE, TracingSetup.ZIPKIN_BRAVE }; + } + + @Override + public SampleTestRunnerConsumer yourCode() { + return (bb, meterRegistry) -> { + ObservationRegistry observationRegistry = getObservationRegistry(); + try (AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext()) { + applicationContext.registerBean(ObservationRegistry.class, () -> observationRegistry); + applicationContext.register(ObservationIntegrationTestConfiguration.class); + applicationContext.refresh(); + + PollableChannel queueChannel = applicationContext.getBean("queueChannel", PollableChannel.class); + PollableChannel replyChannel = new QueueChannel(); + + MutableMessage testMessage = + (MutableMessage) MutableMessageBuilder.withPayload("test data") + .setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel) + .build(); + + Observation.createNotStarted("Test send", new MessageSenderContext(testMessage), observationRegistry) + .observe(() -> queueChannel.send(testMessage)); + + Message receive = replyChannel.receive(10_000); + assertThat(receive).isNotNull() + .extracting("payload").isEqualTo("test data"); + } + + SpansAssert.assertThat(bb.getFinishedSpans()) + .haveSameTraceId() + .hasASpanWithName("Test send", spanAssert -> spanAssert.hasKindEqualTo(Span.Kind.PRODUCER)) + .hasASpanWithName("observedEndpoint receive", spanAssert -> spanAssert + .hasTag(IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(), "observedEndpoint") + .hasTag(IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler") + .hasKindEqualTo(Span.Kind.CONSUMER)) + .hasSize(2); + + MeterRegistryAssert.assertThat(getMeterRegistry()) + .hasTimerWithNameAndTags("spring.integration.handler", + KeyValues.of( + IntegrationObservation.HandlerTags.COMPONENT_NAME.asString(), "observedEndpoint", + IntegrationObservation.HandlerTags.COMPONENT_TYPE.asString(), "handler", + "error", "none")); + }; + } + + + @Configuration + @EnableIntegration + @EnableIntegrationManagement + public static class ObservationIntegrationTestConfiguration { + + @Bean + @GlobalChannelInterceptor + public ChannelInterceptor observationPropagationInterceptor(ObservationRegistry observationRegistry) { + return new ObservationPropagationChannelInterceptor(observationRegistry); + } + + @Bean + @BridgeTo(poller = @Poller(fixedDelay = "100")) + @EndpointId("observedEndpoint") + public PollableChannel queueChannel() { + return new QueueChannel(); + } + + } + +}