Move code from spring-boot-actuator to spring-boot-r2dbc

spring-boot-project/spring-boot-r2dbc/build.gradle

@@ -14,13 +14,20 @@ dependencies {
 	api(project(":spring-boot-project:spring-boot-sql"))
 	api(project(":spring-boot-project:spring-boot-tx"))
 	api("org.springframework:spring-r2dbc")
+	api("org.springframework:spring-r2dbc")
 
+	compileOnly("com.fasterxml.jackson.core:jackson-annotations")
+
+	optional(project(":spring-boot-project:spring-boot-actuator"))
 	optional(project(":spring-boot-project:spring-boot-autoconfigure"))
 	optional(project(":spring-boot-project:spring-boot-jdbc"))
+	optional("io.micrometer:micrometer-core")
 	optional("io.r2dbc:r2dbc-pool")
 	optional("io.r2dbc:r2dbc-proxy")
 	optional("io.r2dbc:r2dbc-spi")
 
+	testCompileOnly("com.fasterxml.jackson.core:jackson-annotations")
+
 	testImplementation(project(":spring-boot-project:spring-boot-jdbc"))
 	testImplementation(project(":spring-boot-project:spring-boot-test"))
 	testImplementation(project(":spring-boot-project:spring-boot-tools:spring-boot-test-support"))

spring-boot-project/spring-boot-r2dbc/src/main/java/org/springframework/boot/r2dbc/actuate/health/ConnectionFactoryHealthIndicator.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright 2012-2025 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.boot.r2dbc.actuate.health;
+
+import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+import io.r2dbc.spi.ValidationDepth;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.boot.actuate.health.AbstractReactiveHealthIndicator;
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.Health.Builder;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.boot.actuate.health.Status;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * A {@link HealthIndicator} to validate a R2DBC {@link ConnectionFactory}.
+ *
+ * @author Mark Paluch
+ * @author Stephane Nicoll
+ * @since 4.0.0
+ */
+public class ConnectionFactoryHealthIndicator extends AbstractReactiveHealthIndicator {
+
+	private final ConnectionFactory connectionFactory;
+
+	private final String validationQuery;
+
+	/**
+	 * Create a new {@link ConnectionFactoryHealthIndicator} using the specified
+	 * {@link ConnectionFactory} and no validation query.
+	 * @param connectionFactory the connection factory
+	 * @see Connection#validate(ValidationDepth)
+	 */
+	public ConnectionFactoryHealthIndicator(ConnectionFactory connectionFactory) {
+		this(connectionFactory, null);
+	}
+
+	/**
+	 * Create a new {@link ConnectionFactoryHealthIndicator} using the specified
+	 * {@link ConnectionFactory} and validation query.
+	 * @param connectionFactory the connection factory
+	 * @param validationQuery the validation query, can be {@code null} to use connection
+	 * validation
+	 */
+	public ConnectionFactoryHealthIndicator(ConnectionFactory connectionFactory, String validationQuery) {
+		Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
+		this.connectionFactory = connectionFactory;
+		this.validationQuery = validationQuery;
+	}
+
+	@Override
+	protected final Mono<Health> doHealthCheck(Builder builder) {
+		return validate(builder).defaultIfEmpty(builder.build())
+			.onErrorResume(Exception.class, (ex) -> Mono.just(builder.down(ex).build()));
+	}
+
+	private Mono<Health> validate(Builder builder) {
+		builder.withDetail("database", this.connectionFactory.getMetadata().getName());
+		return (StringUtils.hasText(this.validationQuery)) ? validateWithQuery(builder)
+				: validateWithConnectionValidation(builder);
+	}
+
+	private Mono<Health> validateWithQuery(Builder builder) {
+		builder.withDetail("validationQuery", this.validationQuery);
+		Mono<Object> connectionValidation = Mono.usingWhen(this.connectionFactory.create(),
+				(conn) -> Flux.from(conn.createStatement(this.validationQuery).execute())
+					.flatMap((it) -> it.map(this::extractResult))
+					.next(),
+				Connection::close, (o, throwable) -> o.close(), Connection::close);
+		return connectionValidation.map((result) -> builder.up().withDetail("result", result).build());
+	}
+
+	private Mono<Health> validateWithConnectionValidation(Builder builder) {
+		builder.withDetail("validationQuery", "validate(REMOTE)");
+		Mono<Boolean> connectionValidation = Mono.usingWhen(this.connectionFactory.create(),
+				(connection) -> Mono.from(connection.validate(ValidationDepth.REMOTE)), Connection::close,
+				(connection, ex) -> connection.close(), Connection::close);
+		return connectionValidation.map((valid) -> builder.status((valid) ? Status.UP : Status.DOWN).build());
+	}
+
+	private Object extractResult(Row row, RowMetadata metadata) {
+		return row.get(metadata.getColumnMetadatas().iterator().next().getName());
+	}
+
+}

spring-boot-project/spring-boot-r2dbc/src/main/java/org/springframework/boot/r2dbc/actuate/health/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012-2025 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.
+ */
+
+/**
+ * Health integration for R2DBC.
+ */
+package org.springframework.boot.r2dbc.actuate.health;

spring-boot-project/spring-boot-r2dbc/src/main/java/org/springframework/boot/r2dbc/actuate/metrics/ConnectionPoolMetrics.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2012-2025 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.boot.r2dbc.actuate.metrics;
+
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.Gauge.Builder;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
+import io.micrometer.core.instrument.binder.MeterBinder;
+import io.r2dbc.pool.ConnectionPool;
+import io.r2dbc.pool.PoolMetrics;
+
+/**
+ * A {@link MeterBinder} for a {@link ConnectionPool}.
+ *
+ * @author Tadaya Tsuyukubo
+ * @author Stephane Nicoll
+ * @since 4.0.0
+ */
+public class ConnectionPoolMetrics implements MeterBinder {
+
+	private static final String CONNECTIONS = "connections";
+
+	private final ConnectionPool pool;
+
+	private final Iterable<Tag> tags;
+
+	public ConnectionPoolMetrics(ConnectionPool pool, String name, Iterable<Tag> tags) {
+		this.pool = pool;
+		this.tags = Tags.concat(tags, "name", name);
+	}
+
+	@Override
+	public void bindTo(MeterRegistry registry) {
+		this.pool.getMetrics().ifPresent((poolMetrics) -> {
+			bindConnectionPoolMetric(registry,
+					Gauge.builder(metricKey("acquired"), poolMetrics, PoolMetrics::acquiredSize)
+						.description("Size of successfully acquired connections which are in active use."));
+			bindConnectionPoolMetric(registry,
+					Gauge.builder(metricKey("allocated"), poolMetrics, PoolMetrics::allocatedSize)
+						.description("Size of allocated connections in the pool which are in active use or idle."));
+			bindConnectionPoolMetric(registry, Gauge.builder(metricKey("idle"), poolMetrics, PoolMetrics::idleSize)
+				.description("Size of idle connections in the pool."));
+			bindConnectionPoolMetric(registry,
+					Gauge.builder(metricKey("pending"), poolMetrics, PoolMetrics::pendingAcquireSize)
+						.description("Size of pending to acquire connections from the underlying connection factory."));
+			bindConnectionPoolMetric(registry,
+					Gauge.builder(metricKey("max.allocated"), poolMetrics, PoolMetrics::getMaxAllocatedSize)
+						.description("Maximum size of allocated connections that this pool allows."));
+			bindConnectionPoolMetric(registry,
+					Gauge.builder(metricKey("max.pending"), poolMetrics, PoolMetrics::getMaxPendingAcquireSize)
+						.description("Maximum size of pending state to acquire connections that this pool allows."));
+		});
+	}
+
+	private void bindConnectionPoolMetric(MeterRegistry registry, Builder<?> builder) {
+		builder.tags(this.tags).baseUnit(CONNECTIONS).register(registry);
+	}
+
+	private static String metricKey(String name) {
+		return "r2dbc.pool." + name;
+	}
+
+}

spring-boot-project/spring-boot-r2dbc/src/main/java/org/springframework/boot/r2dbc/actuate/metrics/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012-2025 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.
+ */
+
+/**
+ * Metrics for R2DBC.
+ */
+package org.springframework.boot.r2dbc.actuate.metrics;

spring-boot-project/spring-boot-r2dbc/src/test/java/org/springframework/boot/r2dbc/actuate/health/ConnectionFactoryHealthIndicatorTests.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright 2012-2025 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.boot.r2dbc.actuate.health;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.UUID;
+
+import io.r2dbc.h2.CloseableConnectionFactory;
+import io.r2dbc.h2.H2ConnectionFactory;
+import io.r2dbc.h2.H2ConnectionOption;
+import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.Result;
+import io.r2dbc.spi.ValidationDepth;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.springframework.boot.actuate.health.ReactiveHealthIndicator;
+import org.springframework.boot.actuate.health.Status;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.entry;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link ConnectionFactoryHealthIndicator}.
+ *
+ * @author Mark Paluch
+ * @author Stephane Nicoll
+ */
+class ConnectionFactoryHealthIndicatorTests {
+
+	@Test
+	void healthIndicatorWhenDatabaseUpWithConnectionValidation() {
+		CloseableConnectionFactory connectionFactory = createTestDatabase();
+		try {
+			ConnectionFactoryHealthIndicator healthIndicator = new ConnectionFactoryHealthIndicator(connectionFactory);
+			healthIndicator.health().as(StepVerifier::create).assertNext((actual) -> {
+				assertThat(actual.getStatus()).isEqualTo(Status.UP);
+				assertThat(actual.getDetails()).containsOnly(entry("database", "H2"),
+						entry("validationQuery", "validate(REMOTE)"));
+			}).expectComplete().verify(Duration.ofSeconds(30));
+		}
+		finally {
+			StepVerifier.create(connectionFactory.close()).expectComplete().verify(Duration.ofSeconds(30));
+		}
+	}
+
+	@Test
+	void healthIndicatorWhenDatabaseDownWithConnectionValidation() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		given(connectionFactory.getMetadata()).willReturn(() -> "mock");
+		RuntimeException exception = new RuntimeException("test");
+		given(connectionFactory.create()).willReturn(Mono.error(exception));
+		ConnectionFactoryHealthIndicator healthIndicator = new ConnectionFactoryHealthIndicator(connectionFactory);
+		healthIndicator.health().as(StepVerifier::create).assertNext((actual) -> {
+			assertThat(actual.getStatus()).isEqualTo(Status.DOWN);
+			assertThat(actual.getDetails()).containsOnly(entry("database", "mock"),
+					entry("validationQuery", "validate(REMOTE)"), entry("error", "java.lang.RuntimeException: test"));
+		}).expectComplete().verify(Duration.ofSeconds(30));
+	}
+
+	@Test
+	void healthIndicatorWhenConnectionValidationFails() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		given(connectionFactory.getMetadata()).willReturn(() -> "mock");
+		Connection connection = mock(Connection.class);
+		given(connection.validate(ValidationDepth.REMOTE)).willReturn(Mono.just(false));
+		given(connection.close()).willReturn(Mono.empty());
+		given(connectionFactory.create()).willAnswer((invocation) -> Mono.just(connection));
+		ConnectionFactoryHealthIndicator healthIndicator = new ConnectionFactoryHealthIndicator(connectionFactory);
+		healthIndicator.health().as(StepVerifier::create).assertNext((actual) -> {
+			assertThat(actual.getStatus()).isEqualTo(Status.DOWN);
+			assertThat(actual.getDetails()).containsOnly(entry("database", "mock"),
+					entry("validationQuery", "validate(REMOTE)"));
+		}).expectComplete().verify(Duration.ofSeconds(30));
+	}
+
+	@Test
+	void healthIndicatorWhenDatabaseUpWithSuccessValidationQuery() {
+		CloseableConnectionFactory connectionFactory = createTestDatabase();
+		try {
+			String customValidationQuery = "SELECT COUNT(*) from HEALTH_TEST";
+			String createTableStatement = "CREATE TABLE HEALTH_TEST (id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY)";
+			Mono.from(connectionFactory.create())
+				.flatMapMany((it) -> Flux.from(it.createStatement(createTableStatement).execute())
+					.flatMap(Result::getRowsUpdated)
+					.thenMany(it.close()))
+				.as(StepVerifier::create)
+				.expectComplete()
+				.verify(Duration.ofSeconds(30));
+			ReactiveHealthIndicator healthIndicator = new ConnectionFactoryHealthIndicator(connectionFactory,
+					customValidationQuery);
+			healthIndicator.health().as(StepVerifier::create).assertNext((actual) -> {
+				assertThat(actual.getStatus()).isEqualTo(Status.UP);
+				assertThat(actual.getDetails()).containsOnly(entry("database", "H2"), entry("result", 0L),
+						entry("validationQuery", customValidationQuery));
+			}).expectComplete().verify(Duration.ofSeconds(30));
+		}
+		finally {
+			StepVerifier.create(connectionFactory.close()).expectComplete().verify(Duration.ofSeconds(30));
+		}
+
+	}
+
+	@Test
+	void healthIndicatorWhenDatabaseUpWithFailureValidationQuery() {
+		CloseableConnectionFactory connectionFactory = createTestDatabase();
+		try {
+			String invalidValidationQuery = "SELECT COUNT(*) from DOES_NOT_EXIST";
+			ReactiveHealthIndicator healthIndicator = new ConnectionFactoryHealthIndicator(connectionFactory,
+					invalidValidationQuery);
+			healthIndicator.health().as(StepVerifier::create).assertNext((actual) -> {
+				assertThat(actual.getStatus()).isEqualTo(Status.DOWN);
+				assertThat(actual.getDetails()).contains(entry("database", "H2"),
+						entry("validationQuery", invalidValidationQuery));
+				assertThat(actual.getDetails()).containsOnlyKeys("database", "error", "validationQuery");
+			}).expectComplete().verify(Duration.ofSeconds(30));
+		}
+		finally {
+			StepVerifier.create(connectionFactory.close()).expectComplete().verify(Duration.ofSeconds(30));
+		}
+	}
+
+	private CloseableConnectionFactory createTestDatabase() {
+		return H2ConnectionFactory.inMemory("db-" + UUID.randomUUID(), "sa", "",
+				Collections.singletonMap(H2ConnectionOption.DB_CLOSE_DELAY, "-1"));
+	}
+
+}

spring-boot-project/spring-boot-r2dbc/src/test/java/org/springframework/boot/r2dbc/actuate/metrics/ConnectionPoolMetricsTests.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2012-2025 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.boot.r2dbc.actuate.metrics;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.UUID;
+
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import io.r2dbc.h2.CloseableConnectionFactory;
+import io.r2dbc.h2.H2ConnectionFactory;
+import io.r2dbc.h2.H2ConnectionOption;
+import io.r2dbc.pool.ConnectionPool;
+import io.r2dbc.pool.ConnectionPoolConfiguration;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import reactor.test.StepVerifier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link ConnectionPoolMetrics}.
+ *
+ * @author Tadaya Tsuyukubo
+ * @author Mark Paluch
+ * @author Stephane Nicoll
+ */
+class ConnectionPoolMetricsTests {
+
+	private static final Tag testTag = Tag.of("test", "yes");
+
+	private static final Tag regionTag = Tag.of("region", "eu-2");
+
+	private CloseableConnectionFactory connectionFactory;
+
+	@BeforeEach
+	void init() {
+		this.connectionFactory = H2ConnectionFactory.inMemory("db-" + UUID.randomUUID(), "sa", "",
+				Collections.singletonMap(H2ConnectionOption.DB_CLOSE_DELAY, "-1"));
+	}
+
+	@AfterEach
+	void close() {
+		if (this.connectionFactory != null) {
+			StepVerifier.create(this.connectionFactory.close()).expectComplete().verify(Duration.ofSeconds(30));
+		}
+	}
+
+	@Test
+	void connectionFactoryIsInstrumented() {
+		SimpleMeterRegistry registry = new SimpleMeterRegistry();
+		ConnectionPool connectionPool = new ConnectionPool(
+				ConnectionPoolConfiguration.builder(this.connectionFactory).initialSize(3).maxSize(7).build());
+		ConnectionPoolMetrics metrics = new ConnectionPoolMetrics(connectionPool, "test-pool",
+				Tags.of(testTag, regionTag));
+		metrics.bindTo(registry);
+		connectionPool.warmup().as(StepVerifier::create).expectNext(3).expectComplete().verify(Duration.ofSeconds(30));
+		// acquire two connections
+		connectionPool.create()
+			.as(StepVerifier::create)
+			.expectNextCount(1)
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
+		connectionPool.create()
+			.as(StepVerifier::create)
+			.expectNextCount(1)
+			.expectComplete()
+			.verify(Duration.ofSeconds(30));
+		assertGauge(registry, "r2dbc.pool.acquired", 2);
+		assertGauge(registry, "r2dbc.pool.allocated", 3);
+		assertGauge(registry, "r2dbc.pool.idle", 1);
+		assertGauge(registry, "r2dbc.pool.pending", 0);
+		assertGauge(registry, "r2dbc.pool.max.allocated", 7);
+		assertGauge(registry, "r2dbc.pool.max.pending", Integer.MAX_VALUE);
+	}
+
+	private void assertGauge(SimpleMeterRegistry registry, String metric, int expectedValue) {
+		Gauge gauge = registry.get(metric).gauge();
+		assertThat(gauge.value()).isEqualTo(expectedValue);
+		assertThat(gauge.getId().getTags()).containsExactlyInAnyOrder(Tag.of("name", "test-pool"), testTag, regionTag);
+	}
+
+}