Add Bus Trace Endpoint

bus/spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/BusTraceEndpointTests.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2012-present 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.cloud.bus;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.bus.endpoint.TraceBusEndpoint;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Test for {@link TraceBusEndpoint}.
+ *
+ * @author Ngoc Nhan
+ */
+@SpringBootTest(
+		properties = { "management.endpoints.web.exposure.include=bustrace", "spring.cloud.bus.trace.enabled=true" })
+public class BusTraceEndpointTests {
+
+	@Autowired(required = false)
+	private TraceBusEndpoint traceBusEndpoint;
+
+	@Test
+	public void contextLoads() {
+		assertThat(this.traceBusEndpoint).isNotNull();
+	}
+
+	@SpringBootConfiguration
+	@EnableAutoConfiguration
+	protected static class TestConfig {
+
+	}
+
+}

bus/spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/TraceListenerIntegrationTests.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2012-present 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.cloud.bus;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.rabbitmq.RabbitMQContainer;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.boot.webtestclient.autoconfigure.AutoConfigureWebTestClient;
+import org.springframework.cloud.bus.trace.Trace;
+import org.springframework.test.web.reactive.server.WebTestClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
+
+/**
+ * Test for {@link org.springframework.cloud.bus.endpoint.TraceBusEndpoint}.
+ *
+ * @author Ngoc Nhan
+ */
+@Testcontainers
+@SpringBootTest(webEnvironment = RANDOM_PORT,
+		properties = { "management.endpoints.web.exposure.include=busrefresh,bustrace",
+				"spring.cloud.bus.trace.enabled=true" })
+@AutoConfigureWebTestClient
+public class TraceListenerIntegrationTests {
+
+	@Container
+	@ServiceConnection
+	static RabbitMQContainer rabbitMQContainer = new RabbitMQContainer("rabbitmq:4.0-management");
+
+	@Test
+	void busTrace(@Autowired WebTestClient client) {
+
+		client.post().uri("/actuator/busrefresh").exchange().expectStatus().is2xxSuccessful();
+		BusTraceResponse response = client.get()
+			.uri("/actuator/bustrace")
+			.exchange()
+			.expectStatus()
+			.is2xxSuccessful()
+			.expectBody(BusTraceResponse.class)
+			.returnResult()
+			.getResponseBody();
+		assertThat(response).isNotNull();
+		assertThat(response.traces()).isNotEmpty().hasSize(2);
+	}
+
+	@SpringBootConfiguration
+	@EnableAutoConfiguration
+	static class TestConfig {
+
+	}
+
+	record BusTraceResponse(List<Trace> traces) {
+
+	}
+
+}

bus/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusAutoConfiguration.java

@@ -19,17 +19,19 @@
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.actuate.autoconfigure.endpoint.condition.ConditionalOnAvailableEndpoint;
 import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
-import org.springframework.boot.actuate.web.exchanges.HttpExchangeRepository;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.bus.endpoint.EnvironmentBusEndpoint;
+import org.springframework.cloud.bus.endpoint.TraceBusEndpoint;
 import org.springframework.cloud.bus.event.Destination;
 import org.springframework.cloud.bus.event.EnvironmentChangeListener;
 import org.springframework.cloud.bus.event.PathDestinationFactory;
 import org.springframework.cloud.bus.event.TraceListener;
+import org.springframework.cloud.bus.trace.InMemoryTraceRepository;
+import org.springframework.cloud.bus.trace.TraceRepository;
 import org.springframework.cloud.context.environment.EnvironmentManager;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.annotation.Bean;
@@ -68,14 +70,28 @@ public BusConsumer busConsumer(ApplicationEventPublisher applicationEventPublish
 
 	@Configuration(proxyBeanMethods = false)
 	@ConditionalOnClass({ Endpoint.class })
-	@ConditionalOnBean(HttpExchangeRepository.class)
 	@ConditionalOnProperty(BusProperties.PREFIX + ".trace.enabled")
+	@ConditionalOnAvailableEndpoint(TraceBusEndpoint.class)
 	protected static class BusAckTraceConfiguration {
 
 		@Bean
 		@ConditionalOnMissingBean
-		public TraceListener ackTraceListener(HttpExchangeRepository repository) {
-			return new TraceListener(repository);
+		public TraceRepository traceRepository() {
+			return new InMemoryTraceRepository();
+		}
+
+		@Bean
+		@ConditionalOnBean(TraceRepository.class)
+		@ConditionalOnMissingBean
+		public TraceBusEndpoint traceBusEndpoint(TraceRepository traceRepository) {
+			return new TraceBusEndpoint(traceRepository);
+		}
+
+		@Bean
+		@ConditionalOnBean(TraceRepository.class)
+		@ConditionalOnMissingBean
+		public TraceListener traceListener(TraceRepository traceRepository) {
+			return new TraceListener(traceRepository);
 		}
 
 	}

bus/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/TraceBusEndpoint.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2012-present 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.cloud.bus.endpoint;
+
+import java.util.List;
+
+import org.springframework.boot.actuate.endpoint.OperationResponseBody;
+import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
+import org.springframework.boot.actuate.endpoint.annotation.ReadOperation;
+import org.springframework.cloud.bus.trace.Trace;
+import org.springframework.cloud.bus.trace.TraceRepository;
+import org.springframework.util.Assert;
+
+/**
+ * {@link Endpoint @Endpoint} to expose {@link Trace} information.
+ *
+ * @author Ngoc Nhan
+ * @since 5.1.0
+ */
+@Endpoint(id = "bustrace")
+public class TraceBusEndpoint {
+
+	private final TraceRepository repository;
+
+	/**
+	 * Create a new {@link TraceBusEndpoint} instance.
+	 * @param repository the trace repository
+	 */
+	public TraceBusEndpoint(TraceRepository repository) {
+		Assert.notNull(repository, "'repository' must not be null");
+		this.repository = repository;
+	}
+
+	/**
+	 * Description of an application's {@link Trace} entries.
+	 */
+	@ReadOperation
+	public TraceDescriptor busTrace() {
+		return new TraceDescriptor(this.repository.findAll());
+	}
+
+	public static final class TraceDescriptor implements OperationResponseBody {
+
+		private final List<Trace> traces;
+
+		private TraceDescriptor(List<Trace> traces) {
+			this.traces = traces;
+		}
+
+		public List<Trace> getTraces() {
+			return this.traces;
+		}
+
+	}
+
+}

bus/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/TraceListener.java

@@ -16,67 +16,70 @@
 
 package org.springframework.cloud.bus.event;
 
-import java.util.LinkedHashMap;
-import java.util.Map;
+import java.time.Clock;
+import java.time.Instant;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.springframework.boot.actuate.web.exchanges.HttpExchangeRepository;
-import org.springframework.context.event.EventListener;
+import org.springframework.cloud.bus.trace.Trace;
+import org.springframework.cloud.bus.trace.TraceRepository;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationListener;
 
 /**
- * A listener for sends and acks of remote application events. Inserts a record for each
- * signal in the {@link HttpExchangeRepository}.
+ * A listener for application event sends and acks. Inserts a record for each signal into
+ * the {@link TraceRepository}.
  *
  * @author Dave Syer
+ * @author Ngoc Nhan
  */
-public class TraceListener {
-
-	private static Log log = LogFactory.getLog(TraceListener.class);
+public class TraceListener implements ApplicationListener<ApplicationEvent> {
 
-	private HttpExchangeRepository repository;
+	private final TraceRepository repository;
 
-	public TraceListener(HttpExchangeRepository repository) {
+	public TraceListener(TraceRepository repository) {
 		this.repository = repository;
 	}
 
-	@EventListener
-	public void onAck(AckRemoteApplicationEvent event) {
-		Map<String, Object> trace = getReceivedTrace(event);
-		// FIXME boot 2 this.repository.add(trace);
-	}
+	@Override
+	public void onApplicationEvent(ApplicationEvent event) {
 
-	@EventListener
-	public void onSend(SentApplicationEvent event) {
-		Map<String, Object> trace = getSentTrace(event);
-		// FIXME boot 2 this.repository.add(trace);
-	}
+		if (event instanceof AckRemoteApplicationEvent ackRemoteApplicationEvent) {
+			this.repository.add(getReceivedTrace(ackRemoteApplicationEvent));
+		}
 
-	protected Map<String, Object> getSentTrace(SentApplicationEvent event) {
-		Map<String, Object> map = new LinkedHashMap<String, Object>();
-		map.put("signal", "spring.cloud.bus.sent");
-		map.put("type", event.getType().getSimpleName());
-		map.put("id", event.getId());
-		map.put("origin", event.getOriginService());
-		map.put("destination", event.getDestinationService());
-		if (log.isDebugEnabled()) {
-			log.debug(map);
+		if (event instanceof SentApplicationEvent sentApplicationEvent) {
+			this.repository.add(getSentTrace(sentApplicationEvent));
 		}
-		return map;
+
 	}
 
-	protected Map<String, Object> getReceivedTrace(AckRemoteApplicationEvent event) {
-		Map<String, Object> map = new LinkedHashMap<String, Object>();
-		map.put("signal", "spring.cloud.bus.ack");
-		map.put("event", event.getEvent().getSimpleName());
-		map.put("id", event.getAckId());
-		map.put("origin", event.getOriginService());
-		map.put("destination", event.getAckDestinationService());
-		if (log.isDebugEnabled()) {
-			log.debug(map);
-		}
-		return map;
+	/**
+	 * Creates a trace for a acks application event.
+	 * @param event the acks application event
+	 * @return the trace for the acks application event
+	 */
+	protected Trace getReceivedTrace(AckRemoteApplicationEvent event) {
+
+		Trace trace = new Trace(Instant.now(Clock.systemUTC()), "spring.cloud.bus.ack",
+				event.getEvent().getSimpleName());
+		trace.setId(event.getAckId());
+		trace.setOrigin(event.getOriginService());
+		trace.setDestination(event.getAckDestinationService());
+		return trace;
+	}
+
+	/**
+	 * Creates a trace for a sent application event.
+	 * @param event the sent application event
+	 * @return the trace for the sent application event
+	 */
+	protected Trace getSentTrace(SentApplicationEvent event) {
+
+		Trace trace = new Trace(Instant.now(Clock.systemUTC()), "spring.cloud.bus.sent",
+				event.getType().getSimpleName());
+		trace.setId(event.getId());
+		trace.setOrigin(event.getOriginService());
+		trace.setDestination(event.getDestinationService());
+		return trace;
 	}
 
 }