diff --git a/framework-docs/framework-docs.gradle b/framework-docs/framework-docs.gradle index bf1cc292a6..f2b21206d9 100644 --- a/framework-docs/framework-docs.gradle +++ b/framework-docs/framework-docs.gradle @@ -61,7 +61,9 @@ repositories { dependencies { api(project(":spring-context")) + api(project(":spring-jms")) api(project(":spring-web")) + api("jakarta.jms:jakarta.jms-api") api("jakarta.servlet:jakarta.servlet-api") implementation(project(":spring-core-test")) diff --git a/framework-docs/modules/ROOT/pages/integration/observability.adoc b/framework-docs/modules/ROOT/pages/integration/observability.adoc index 39c9fdc77b..98a4ab9649 100644 --- a/framework-docs/modules/ROOT/pages/integration/observability.adoc +++ b/framework-docs/modules/ROOT/pages/integration/observability.adoc @@ -27,6 +27,12 @@ As outlined xref:integration/observability.adoc[at the beginning of this section |xref:integration/observability.adoc#observability.http-server[`"http.server.requests"`] |Processing time for HTTP server exchanges at the Framework level +|xref:integration/observability.adoc#observability.jms.publish[`"jms.message.publish"`] +|Time spent sending a JMS message to a destination by a message producer. + +|xref:integration/observability.adoc#observability.jms.process[`"jms.message.process"`] +|Processing time for a JMS message that was previously received by a message consumer. + |xref:integration/observability.adoc#observability.tasks-scheduled[`"tasks.scheduled.execution"`] |Processing time for an execution of a `@Scheduled` task |=== @@ -108,6 +114,72 @@ By default, the following `KeyValues` are created: |=== +[[observability.jms]] +== JMS messaging instrumentation + +Spring Framework uses the Jakarta JMS instrumentation provided by Micrometer if the `io.micrometer:micrometer-core` dependency is on the classpath. +The `io.micrometer.core.instrument.binder.jms.JmsInstrumentation` instruments `jakarta.jms.Session` and records the relevant observations. + +This instrumentation will create 2 types of observations: + +* `"jms.message.publish"` when a JMS message is sent to the broker, typically with `JmsTemplate`. +* `"jms.message.process"` when a JMS message is processed by the application, typically with a `MessageListener` or a `@JmsListener` annotated method. + +NOTE: currently there is no instrumentation for `"jms.message.receive"` observations as there is little value in measuring the time spent waiting for the reception of a message. +Such an integration would typically instrument `MessageConsumer#receive` method calls. But once those return, the processing time is not measured and the trace scope cannot be propagated to the application. + +By default, both observations share the same set of possible `KeyValues`: + +.Low cardinality Keys +[cols="a,a"] +|=== +|Name | Description +|`exception` |Class name of the exception thrown during the messaging operation (or "none"). +|`messaging.destination.temporary` _(required)_|Whether the destination is a `TemporaryQueue` or `TemporaryTopic` (values: `"true"` or `"false"`). +|`messaging.operation` _(required)_|Name of JMS operation being performed (values: `"publish"` or `"process"`). +|=== + +.High cardinality Keys +[cols="a,a"] +|=== +|Name | Description +|`messaging.message.conversation_id` |The correlation ID of the JMS message. +|`messaging.destination.name` |The name of destination the current message was sent to. +|`messaging.message.id` |Value used by the messaging system as an identifier for the message. +|=== + +[[observability.jms.publish]] +=== JMS message Publication instrumentation + +`"jms.message.publish"` observations are recorded when a JMS message is sent to the broker. +They measure the time spent sending the message and propagate the tracing information with outgoing JMS message headers. + +You will need to configure the `ObservationRegistry` on the `JmsTemplate` to enable observations: + +include-code::./JmsTemplatePublish[] + +It uses the `io.micrometer.core.instrument.binder.jms.DefaultJmsPublishObservationConvention` by default, backed by the `io.micrometer.core.instrument.binder.jms.JmsPublishObservationContext`. + +[[observability.jms.process]] +=== JMS message Processing instrumentation + +`"jms.message.process"` observations are recorded when a JMS message is processed by the application. +They measure the time spent processing the message and propagate the tracing context with incoming JMS message headers. + +Most applications will use the xref:integration/jms/annotated.adoc#jms-annotated[`@JmsListener` annotated methods] mechanism to process incoming messages. +You will need to ensure that the `ObservationRegistry` is configured on the dedicated `JmsListenerContainerFactory`: + +include-code::./JmsConfiguration[] + +A xref:integration/jms/annotated.adoc#jms-annotated-support[default container factory is required to enable the annotation support], +but note that `@JmsListener` annotations can refer to specific container factory beans for specific purposes. +In all cases, Observations are only recorded if the observation registry is configured on the container factory. + +Similar observations are recorded with `JmsTemplate` when messages are processed by a `MessageListener`. +Such listeners are set on a `MessageConsumer` within a session callback (see `JmsTemplate.execute(SessionCallback)`). + +This observation uses the `io.micrometer.core.instrument.binder.jms.DefaultJmsProcessObservationConvention` by default, backed by the `io.micrometer.core.instrument.binder.jms.JmsProcessObservationContext`. + [[observability.http-server]] == HTTP Server instrumentation diff --git a/framework-docs/src/main/java/org/springframework/docs/integration/observability/jms/process/JmsConfiguration.java b/framework-docs/src/main/java/org/springframework/docs/integration/observability/jms/process/JmsConfiguration.java new file mode 100644 index 0000000000..d611d3ed81 --- /dev/null +++ b/framework-docs/src/main/java/org/springframework/docs/integration/observability/jms/process/JmsConfiguration.java @@ -0,0 +1,39 @@ +/* + * Copyright 2002-2023 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.docs.integration.observability.jms.process; + +import io.micrometer.observation.ObservationRegistry; +import jakarta.jms.ConnectionFactory; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jms.annotation.EnableJms; +import org.springframework.jms.config.DefaultJmsListenerContainerFactory; + +@Configuration +@EnableJms +public class JmsConfiguration { + + @Bean + public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory, ObservationRegistry observationRegistry) { + DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); + factory.setConnectionFactory(connectionFactory); + factory.setObservationRegistry(observationRegistry); + return factory; + } + +} diff --git a/framework-docs/src/main/java/org/springframework/docs/integration/observability/jms/publish/JmsTemplatePublish.java b/framework-docs/src/main/java/org/springframework/docs/integration/observability/jms/publish/JmsTemplatePublish.java new file mode 100644 index 0000000000..4cc828714f --- /dev/null +++ b/framework-docs/src/main/java/org/springframework/docs/integration/observability/jms/publish/JmsTemplatePublish.java @@ -0,0 +1,44 @@ +/* + * Copyright 2002-2023 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.docs.integration.observability.jms.publish; + +import io.micrometer.observation.ObservationRegistry; +import jakarta.jms.ConnectionFactory; + +import org.springframework.jms.core.JmsMessagingTemplate; +import org.springframework.jms.core.JmsTemplate; + +public class JmsTemplatePublish { + + private final JmsTemplate jmsTemplate; + + private final JmsMessagingTemplate jmsMessagingTemplate; + + public JmsTemplatePublish(ObservationRegistry observationRegistry, ConnectionFactory connectionFactory) { + this.jmsTemplate = new JmsTemplate(connectionFactory); + // configure the observation registry + this.jmsTemplate.setObservationRegistry(observationRegistry); + + // For JmsMessagingTemplate, instantiate it with a JMS template that has a configured registry + this.jmsMessagingTemplate = new JmsMessagingTemplate(this.jmsTemplate); + } + + public void sendMessages() { + this.jmsTemplate.convertAndSend("spring.observation.test", "test message"); + } + +} diff --git a/framework-platform/framework-platform.gradle b/framework-platform/framework-platform.gradle index ad74c87b80..fe8ee408a5 100644 --- a/framework-platform/framework-platform.gradle +++ b/framework-platform/framework-platform.gradle @@ -8,7 +8,7 @@ javaPlatform { dependencies { api(platform("com.fasterxml.jackson:jackson-bom:2.15.2")) - api(platform("io.micrometer:micrometer-bom:1.12.0-M1")) + api(platform("io.micrometer:micrometer-bom:1.12.0-SNAPSHOT")) api(platform("io.netty:netty-bom:4.1.96.Final")) api(platform("io.netty:netty5-bom:5.0.0.Alpha5")) api(platform("io.projectreactor:reactor-bom:2023.0.0-M1")) @@ -92,6 +92,8 @@ dependencies { api("org.apache.activemq:activemq-broker:5.17.4") api("org.apache.activemq:activemq-kahadb-store:5.17.4") api("org.apache.activemq:activemq-stomp:5.17.4") + api("org.apache.activemq:artemis-junit-5:2.29.0") + api("org.apache.activemq:artemis-jakarta-client:2.29.0") api("org.apache.commons:commons-pool2:2.9.0") api("org.apache.derby:derby:10.16.1.1") api("org.apache.derby:derbyclient:10.16.1.1") diff --git a/spring-jms/spring-jms.gradle b/spring-jms/spring-jms.gradle index 8765036146..bbea35eedf 100644 --- a/spring-jms/spring-jms.gradle +++ b/spring-jms/spring-jms.gradle @@ -5,14 +5,20 @@ dependencies { api(project(":spring-core")) api(project(":spring-messaging")) api(project(":spring-tx")) + api("io.micrometer:micrometer-observation") compileOnly("jakarta.jms:jakarta.jms-api") optional(project(":spring-aop")) optional(project(":spring-context")) optional(project(":spring-oxm")) optional("com.fasterxml.jackson.core:jackson-databind") + optional("io.micrometer:micrometer-core") optional("jakarta.resource:jakarta.resource-api") optional("jakarta.transaction:jakarta.transaction-api") testImplementation(testFixtures(project(":spring-beans"))) testImplementation(testFixtures(project(":spring-tx"))) testImplementation("jakarta.jms:jakarta.jms-api") + testImplementation('io.micrometer:context-propagation') + testImplementation("io.micrometer:micrometer-observation-test") + testImplementation("org.apache.activemq:artemis-junit-5") + testImplementation("org.apache.activemq:artemis-jakarta-client") } diff --git a/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerContainerFactory.java b/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerContainerFactory.java index 049aa4bcc0..e17b2178ff 100644 --- a/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerContainerFactory.java +++ b/spring-jms/src/main/java/org/springframework/jms/config/AbstractJmsListenerContainerFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2023 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. @@ -16,6 +16,7 @@ package org.springframework.jms.config; +import io.micrometer.observation.ObservationRegistry; import jakarta.jms.ConnectionFactory; import jakarta.jms.ExceptionListener; import org.apache.commons.logging.Log; @@ -86,6 +87,9 @@ public abstract class AbstractJmsListenerContainerFactory new JmsProcessObservationContext(message), this.observationRegistry); + observation.observeChecked(() -> invokeListener(session, message)); } catch (JMSException | RuntimeException | Error ex) { rollbackOnExceptionIfNecessary(session, ex); @@ -724,6 +758,8 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen Connection conToClose = null; Session sessionToClose = null; + Observation observation = JmsObservationDocumentation.JMS_MESSAGE_PROCESS + .observation(null, DEFAULT_CONVENTION, () -> new JmsProcessObservationContext(message), this.observationRegistry); try { Session sessionToUse = session; if (!isExposeListenerSession()) { @@ -732,6 +768,7 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen sessionToClose = createSession(conToClose); sessionToUse = sessionToClose; } + observation.start(); // Actually invoke the message listener... listener.onMessage(message, sessionToUse); // Clean up specially exposed Session, if any. @@ -742,7 +779,12 @@ public abstract class AbstractMessageListenerContainer extends AbstractJmsListen } } } + catch (JMSException exc) { + observation.error(exc); + throw exc; + } finally { + observation.stop(); JmsUtils.closeSession(sessionToClose); JmsUtils.closeConnection(conToClose); } diff --git a/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerContainerFactoryTests.java b/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerContainerFactoryTests.java index 9ea6fc9eef..edf2797cb3 100644 --- a/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerContainerFactoryTests.java +++ b/spring-jms/src/test/java/org/springframework/jms/config/JmsListenerContainerFactoryTests.java @@ -16,6 +16,7 @@ package org.springframework.jms.config; +import io.micrometer.observation.tck.TestObservationRegistry; import jakarta.jms.ConnectionFactory; import jakarta.jms.MessageListener; import jakarta.jms.Session; @@ -77,10 +78,12 @@ public class JmsListenerContainerFactoryTests { @Test public void createJmsContainerFullConfig() { DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); + TestObservationRegistry testObservationRegistry = TestObservationRegistry.create(); setDefaultJmsConfig(factory); factory.setCacheLevel(DefaultMessageListenerContainer.CACHE_CONSUMER); factory.setConcurrency("3-10"); factory.setMaxMessagesPerTask(5); + factory.setObservationRegistry(testObservationRegistry); SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint(); MessageListener messageListener = new MessageListenerAdapter(); @@ -93,6 +96,7 @@ public class JmsListenerContainerFactoryTests { assertThat(container.getConcurrentConsumers()).isEqualTo(3); assertThat(container.getMaxConcurrentConsumers()).isEqualTo(10); assertThat(container.getMaxMessagesPerTask()).isEqualTo(5); + assertThat(container.getObservationRegistry()).isEqualTo(testObservationRegistry); assertThat(container.getMessageListener()).isEqualTo(messageListener); assertThat(container.getDestinationName()).isEqualTo("myQueue"); diff --git a/spring-jms/src/test/java/org/springframework/jms/core/JmsTemplateObservationTests.java b/spring-jms/src/test/java/org/springframework/jms/core/JmsTemplateObservationTests.java new file mode 100644 index 0000000000..6666229712 --- /dev/null +++ b/spring-jms/src/test/java/org/springframework/jms/core/JmsTemplateObservationTests.java @@ -0,0 +1,90 @@ +/* + * Copyright 2002-2023 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.jms.core; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import io.micrometer.observation.tck.TestObservationRegistry; +import jakarta.jms.MessageConsumer; +import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; +import org.apache.activemq.artemis.junit.EmbeddedActiveMQExtension; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static io.micrometer.observation.tck.TestObservationRegistryAssert.assertThat; + +/** + * Tests for Observability related {@link JmsTemplate}. + * + * @author Brian Clozel + */ +class JmsTemplateObservationTests { + + @RegisterExtension + EmbeddedActiveMQExtension server = new EmbeddedActiveMQExtension(); + + TestObservationRegistry registry = TestObservationRegistry.create(); + + private ActiveMQConnectionFactory connectionFactory; + + @BeforeEach + void setupServer() { + server.start(); + connectionFactory = new ActiveMQConnectionFactory(server.getVmURL()); + } + + @Test + void shouldRecordJmsPublishObservations() { + JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + jmsTemplate.setObservationRegistry(registry); + jmsTemplate.convertAndSend("spring.test.observation", "message content"); + assertThat(registry).hasObservationWithNameEqualTo("jms.message.publish") + .that() + .hasHighCardinalityKeyValue("messaging.destination.name", "spring.test.observation"); + } + + @Test + void shouldRecordJmsProcessObservations() throws Exception { + JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + jmsTemplate.setObservationRegistry(registry); + jmsTemplate.convertAndSend("spring.test.observation", "message content"); + jmsTemplate.execute(session -> { + try { + CountDownLatch latch = new CountDownLatch(1); + MessageConsumer mc = session.createConsumer(session.createQueue("spring.test.observation")); + mc.setMessageListener(message -> latch.countDown()); + return latch.await(2, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + throw new RuntimeException(e); + } + }, true); + assertThat(registry).hasObservationWithNameEqualTo("jms.message.process") + .that() + .hasHighCardinalityKeyValue("messaging.destination.name", "spring.test.observation"); + } + + @AfterEach + void shutdownServer() { + connectionFactory.close(); + server.stop(); + } + +} diff --git a/spring-jms/src/test/java/org/springframework/jms/listener/MessageListenerContainerObservationTests.java b/spring-jms/src/test/java/org/springframework/jms/listener/MessageListenerContainerObservationTests.java new file mode 100644 index 0000000000..7b198110a9 --- /dev/null +++ b/spring-jms/src/test/java/org/springframework/jms/listener/MessageListenerContainerObservationTests.java @@ -0,0 +1,95 @@ +/* + * Copyright 2002-2023 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.jms.listener; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; + +import io.micrometer.observation.tck.TestObservationRegistry; +import jakarta.jms.Message; +import jakarta.jms.MessageListener; +import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; +import org.apache.activemq.artemis.junit.EmbeddedActiveMQExtension; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import org.springframework.jms.core.JmsTemplate; + +import static io.micrometer.observation.tck.TestObservationRegistryAssert.assertThat; + +/** + * Observation tests for {@link AbstractMessageListenerContainer} implementations. + * @author Brian Clozel + */ +class MessageListenerContainerObservationTests { + + @RegisterExtension + EmbeddedActiveMQExtension server = new EmbeddedActiveMQExtension(); + + TestObservationRegistry registry = TestObservationRegistry.create(); + + private ActiveMQConnectionFactory connectionFactory; + + @BeforeEach + void setupServer() { + server.start(); + connectionFactory = new ActiveMQConnectionFactory(server.getVmURL()); + } + + @ParameterizedTest(name = "{index} {0}") + @MethodSource("listenerContainers") + void shouldRecordJmsProcessObservations(String implementationClass, AbstractMessageListenerContainer listenerContainer) throws Exception { + JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + jmsTemplate.convertAndSend("spring.test.observation", "message content"); + CountDownLatch latch = new CountDownLatch(1); + listenerContainer.setConnectionFactory(connectionFactory); + listenerContainer.setObservationRegistry(registry); + listenerContainer.setDestinationName("spring.test.observation"); + listenerContainer.setMessageListener(new MessageListener() { + @Override + public void onMessage(Message message) { + latch.countDown(); + } + }); + listenerContainer.afterPropertiesSet(); + listenerContainer.start(); + latch.await(2, TimeUnit.SECONDS); + assertThat(registry).hasObservationWithNameEqualTo("jms.message.process") + .that() + .hasHighCardinalityKeyValue("messaging.destination.name", "spring.test.observation"); + listenerContainer.shutdown(); + listenerContainer.stop(); + } + + static Stream listenerContainers() { + return Stream.of( + Arguments.of(DefaultMessageListenerContainer.class.getSimpleName(), new DefaultMessageListenerContainer()), + Arguments.of(SimpleMessageListenerContainer.class.getSimpleName(), new SimpleMessageListenerContainer()) + ); + } + + @AfterEach + void shutdownServer() { + connectionFactory.close(); + server.stop(); + } +}