Add reactive health indicator for Cassandra

See gh-13864

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/cassandra/CassandraReactiveHealthIndicatorAutoConfiguration.java

+/*
+ * Copyright 2012-2017 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
+ *
+ *      http://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.actuate.autoconfigure.cassandra;
+
+import com.datastax.driver.core.Cluster;
+import org.springframework.boot.actuate.autoconfigure.health.CompositeReactiveHealthIndicatorConfiguration;
+import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator;
+import org.springframework.boot.actuate.autoconfigure.health.HealthIndicatorAutoConfiguration;
+import org.springframework.boot.actuate.cassandra.CassandraReactiveHealthIndicator;
+import org.springframework.boot.actuate.health.ReactiveHealthIndicator;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.cassandra.CassandraAutoConfiguration;
+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.data.cassandra.CassandraReactiveDataAutoConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
+
+import java.util.Map;
+
+/**
+ * {@link EnableAutoConfiguration Auto-configuration} for
+ * {@link org.springframework.boot.actuate.cassandra.CassandraReactiveHealthIndicator}.
+ *
+ * @author Artsiom Yudovin
+ * @since 2.0.0
+ */
+@Configuration
+@ConditionalOnClass({ReactiveCassandraOperations.class, Cluster.class })
+@ConditionalOnBean(ReactiveCassandraOperations.class)
+@ConditionalOnEnabledHealthIndicator("cassandra")
+@AutoConfigureBefore(HealthIndicatorAutoConfiguration.class)
+@AutoConfigureAfter({ CassandraAutoConfiguration.class,
+		CassandraReactiveDataAutoConfiguration.class })
+public class CassandraReactiveHealthIndicatorAutoConfiguration extends
+		CompositeReactiveHealthIndicatorConfiguration<CassandraReactiveHealthIndicator, ReactiveCassandraOperations> {
+	private final Map<String, ReactiveCassandraOperations> reactiveCassandraOperations;
+
+	public CassandraReactiveHealthIndicatorAutoConfiguration(
+			Map<String, ReactiveCassandraOperations> reactiveCassandraOperations) {
+		this.reactiveCassandraOperations = reactiveCassandraOperations;
+	}
+
+	@Bean
+	@ConditionalOnMissingBean(name = "cassandraReactiveHealthIndicator")
+	public ReactiveHealthIndicator cassandraHealthIndicator() {
+		return createHealthIndicator(this.reactiveCassandraOperations);
+	}
+}

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/resources/META-INF/spring.factories

@@ -5,6 +5,7 @@ org.springframework.boot.actuate.autoconfigure.audit.AuditEventsEndpointAutoConf
 org.springframework.boot.actuate.autoconfigure.beans.BeansEndpointAutoConfiguration,\
 org.springframework.boot.actuate.autoconfigure.cache.CachesEndpointAutoConfiguration,\
 org.springframework.boot.actuate.autoconfigure.cassandra.CassandraHealthIndicatorAutoConfiguration,\
+org.springframework.boot.actuate.autoconfigure.cassandra.CassandraReactiveHealthIndicatorAutoConfiguration,\
 org.springframework.boot.actuate.autoconfigure.cloudfoundry.servlet.CloudFoundryActuatorAutoConfiguration,\
 org.springframework.boot.actuate.autoconfigure.cloudfoundry.reactive.ReactiveCloudFoundryActuatorAutoConfiguration,\
 org.springframework.boot.actuate.autoconfigure.condition.ConditionsReportEndpointAutoConfiguration,\

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/cassandra/CassandraReactiveHealthIndicatorAutoConfigurationTests.java

+/*
+ * Copyright 2012-2017 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
+ *
+ *      http://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.actuate.autoconfigure.cassandra;
+
+import org.junit.Test;
+import org.springframework.boot.actuate.autoconfigure.health.HealthIndicatorAutoConfiguration;
+import org.springframework.boot.actuate.cassandra.CassandraHealthIndicator;
+import org.springframework.boot.actuate.cassandra.CassandraReactiveHealthIndicator;
+import org.springframework.boot.actuate.health.ApplicationHealthIndicator;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests for {@link CassandraReactiveHealthIndicatorAutoConfiguration}.
+ *
+ * @author Artsiom Yudovin
+ */
+public class CassandraReactiveHealthIndicatorAutoConfigurationTests {
+
+	private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withConfiguration(AutoConfigurations.of(CassandraReactiveHealthIndicatorAutoConfigurationTests.CassandraConfiguration.class,
+					CassandraReactiveHealthIndicatorAutoConfiguration.class,
+					HealthIndicatorAutoConfiguration.class));
+
+	@Test
+	public void runShouldCreateIndicator() {
+		this.contextRunner.run((context) -> assertThat(context)
+				.hasSingleBean(CassandraReactiveHealthIndicator.class)
+				.doesNotHaveBean(CassandraHealthIndicator.class)
+				.doesNotHaveBean(ApplicationHealthIndicator.class));
+	}
+
+	@Test
+	public void runWhenDisabledShouldNotCreateIndicator() {
+		this.contextRunner.withPropertyValues("management.health.cassandra.enabled:false")
+				.run((context) -> assertThat(context)
+						.doesNotHaveBean(CassandraReactiveHealthIndicator.class)
+						.hasSingleBean(ApplicationHealthIndicator.class));
+	}
+
+	@Configuration
+	@AutoConfigureBefore(CassandraReactiveHealthIndicatorAutoConfiguration.class)
+	protected static class CassandraConfiguration {
+
+		@Bean
+		public ReactiveCassandraOperations cassandraOperations() {
+			return mock(ReactiveCassandraOperations.class);
+		}
+
+	}
+}

spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/cassandra/CassandraReactiveHealthIndicator.java

+/*
+ * Copyright 2012-2018 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
+ *
+ *      http://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.actuate.cassandra;
+
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.querybuilder.Select;
+import org.springframework.boot.actuate.health.AbstractReactiveHealthIndicator;
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.data.cassandra.ReactiveResultSet;
+import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
+import org.springframework.util.Assert;
+import reactor.core.publisher.Mono;
+
+/**
+ * Simple implementation of a {@link HealthIndicator} returning status information for
+ * Cassandra data stores.
+ *
+ * @author Artsiom Yudovin
+ * @since 2.0.0
+ */
+public class CassandraReactiveHealthIndicator extends AbstractReactiveHealthIndicator {
+
+	private final ReactiveCassandraOperations reactiveCassandraOperations;
+
+	/**
+	 * Create a new {@link CassandraHealthIndicator} instance.
+	 * @param reactiveCassandraOperations the Cassandra operations
+	 */
+	public CassandraReactiveHealthIndicator(ReactiveCassandraOperations reactiveCassandraOperations) {
+		Assert.notNull(reactiveCassandraOperations, "ReactiveCassandraOperations must not be null");
+		this.reactiveCassandraOperations = reactiveCassandraOperations;
+	}
+
+	@Override
+	protected Mono<Health> doHealthCheck(Health.Builder builder) {
+		Select select = QueryBuilder.select("release_version").from("system", "local");
+		Mono<String> results = this.reactiveCassandraOperations.getReactiveCqlOperations()
+				.queryForObject(select, String.class);
+
+		return results
+		        .map(version -> builder.up().withDetail("version", version).build())
+				.single();
+
+		}
+}

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/cassandra/CassandraReactiveHealthIndicatorTest.java

+/*
+ * Copyright 2012-2018 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
+ *
+ *      http://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.actuate.cassandra;
+
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.querybuilder.Select;
+import org.junit.Test;
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.ReactiveHealthIndicator;
+import org.springframework.boot.actuate.health.Status;
+import org.springframework.data.cassandra.CassandraInternalException;
+import org.springframework.data.cassandra.ReactiveResultSet;
+import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
+import org.springframework.data.cassandra.core.cql.ReactiveCqlOperations;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import static org.assertj.core.api.Assertions.anyOf;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+/**
+ * A {@link ReactiveHealthIndicator} for Mongo.
+ *
+ * @author Artsiom Yudovin
+ * @since 2.0.0
+ */
+public class CassandraReactiveHealthIndicatorTest {
+
+	@Test
+	public void testCassandraIsUp() {
+		ReactiveCqlOperations reactiveCqlOperations = mock(ReactiveCqlOperations.class);
+		ReactiveCassandraOperations reactiveCassandraOperations = mock(ReactiveCassandraOperations.class);
+
+		given(reactiveCqlOperations.queryForObject(any(Select.class), eq(String.class)))
+				.willReturn(Mono.just("6.0.0"));
+		given(reactiveCassandraOperations.getReactiveCqlOperations()).willReturn(reactiveCqlOperations);
+
+		CassandraReactiveHealthIndicator cassandraReactiveHealthIndicator =
+				new CassandraReactiveHealthIndicator(reactiveCassandraOperations);
+		Mono<Health> health = cassandraReactiveHealthIndicator.health();
+		StepVerifier.create(health).consumeNextWith((h) -> {
+			assertThat(h.getStatus()).isEqualTo(Status.UP);
+			assertThat(h.getDetails()).containsOnlyKeys("version");
+			assertThat(h.getDetails().get("version")).isEqualTo("6.0.0");
+		}).verifyComplete();
+	}
+
+	@Test
+	public void testCassandraIsDown() {
+		ReactiveCassandraOperations reactiveCassandraOperations = mock(ReactiveCassandraOperations.class);
+
+		given(reactiveCassandraOperations.getReactiveCqlOperations())
+				.willThrow(new CassandraInternalException("Connection failed"));
+
+		CassandraReactiveHealthIndicator cassandraReactiveHealthIndicator =
+				new CassandraReactiveHealthIndicator(reactiveCassandraOperations);
+		Mono<Health> health = cassandraReactiveHealthIndicator.health();
+		StepVerifier.create(health).consumeNextWith((h) -> {
+			assertThat(h.getStatus()).isEqualTo(Status.DOWN);
+			assertThat(h.getDetails()).containsOnlyKeys("error");
+			assertThat(h.getDetails().get("error"))
+					.isEqualTo(CassandraInternalException.class.getName() + ": Connection failed");
+		}).verifyComplete();
+	}
+}