From d49cc20e79b64ce8109948324308d58418bed6e8 Mon Sep 17 00:00:00 2001 From: Andy Wilkinson Date: Tue, 16 Oct 2018 09:27:58 +0100 Subject: [PATCH] Rework CouchbaseReactiveHealthIndicator to use DiagnosticsReport Closes gh-14799 --- .../actuate/couchbase/CouchbaseHealth.java | 70 +++++++++++ .../couchbase/CouchbaseHealthIndicator.java | 32 +---- .../CouchbaseReactiveHealthIndicator.java | 32 ++--- ...CouchbaseReactiveHealthIndicatorTests.java | 114 ++++++++---------- 4 files changed, 129 insertions(+), 119 deletions(-) create mode 100644 spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseHealth.java diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseHealth.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseHealth.java new file mode 100644 index 0000000000..3336beba4b --- /dev/null +++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseHealth.java @@ -0,0 +1,70 @@ +/* + * 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.couchbase; + +import java.util.HashMap; +import java.util.Map; +import java.util.stream.Collectors; + +import com.couchbase.client.core.message.internal.DiagnosticsReport; +import com.couchbase.client.core.message.internal.EndpointHealth; +import com.couchbase.client.core.state.LifecycleState; + +import org.springframework.boot.actuate.health.Health.Builder; + +/** + * Details of Couchbase's health. + * + * @author Andy Wilkinson + */ +class CouchbaseHealth { + + private final DiagnosticsReport diagnostics; + + CouchbaseHealth(DiagnosticsReport diagnostics) { + this.diagnostics = diagnostics; + } + + void applyTo(Builder builder) { + builder = isCouchbaseUp(this.diagnostics) ? builder.up() : builder.down(); + builder.withDetail("sdk", this.diagnostics.sdk()); + builder.withDetail("endpoints", this.diagnostics.endpoints().stream() + .map(this::describe).collect(Collectors.toList())); + } + + private boolean isCouchbaseUp(DiagnosticsReport diagnostics) { + for (EndpointHealth health : diagnostics.endpoints()) { + LifecycleState state = health.state(); + if (state != LifecycleState.CONNECTED && state != LifecycleState.IDLE) { + return false; + } + } + return true; + } + + private Map describe(EndpointHealth endpointHealth) { + Map map = new HashMap<>(); + map.put("id", endpointHealth.id()); + map.put("lastActivity", endpointHealth.lastActivity()); + map.put("local", endpointHealth.local().toString()); + map.put("remote", endpointHealth.remote().toString()); + map.put("state", endpointHealth.state()); + map.put("type", endpointHealth.type()); + return map; + } + +} diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseHealthIndicator.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseHealthIndicator.java index 4939e259bd..7370a86e9d 100644 --- a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseHealthIndicator.java +++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseHealthIndicator.java @@ -16,13 +16,7 @@ package org.springframework.boot.actuate.couchbase; -import java.util.HashMap; -import java.util.Map; -import java.util.stream.Collectors; - import com.couchbase.client.core.message.internal.DiagnosticsReport; -import com.couchbase.client.core.message.internal.EndpointHealth; -import com.couchbase.client.core.state.LifecycleState; import com.couchbase.client.java.Cluster; import org.springframework.boot.actuate.health.AbstractHealthIndicator; @@ -55,31 +49,7 @@ public class CouchbaseHealthIndicator extends AbstractHealthIndicator { @Override protected void doHealthCheck(Health.Builder builder) throws Exception { DiagnosticsReport diagnostics = this.cluster.diagnostics(); - builder = isCouchbaseUp(diagnostics) ? builder.up() : builder.down(); - builder.withDetail("sdk", diagnostics.sdk()); - builder.withDetail("endpoints", diagnostics.endpoints().stream() - .map(this::describe).collect(Collectors.toList())); - } - - private boolean isCouchbaseUp(DiagnosticsReport diagnostics) { - for (EndpointHealth health : diagnostics.endpoints()) { - LifecycleState state = health.state(); - if (state != LifecycleState.CONNECTED && state != LifecycleState.IDLE) { - return false; - } - } - return true; - } - - private Map describe(EndpointHealth endpointHealth) { - Map map = new HashMap<>(); - map.put("id", endpointHealth.id()); - map.put("lastActivity", endpointHealth.lastActivity()); - map.put("local", endpointHealth.local().toString()); - map.put("remote", endpointHealth.remote().toString()); - map.put("state", endpointHealth.state()); - map.put("type", endpointHealth.type()); - return map; + new CouchbaseHealth(diagnostics).applyTo(builder); } } diff --git a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseReactiveHealthIndicator.java b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseReactiveHealthIndicator.java index 854368651e..f4e87c3176 100644 --- a/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseReactiveHealthIndicator.java +++ b/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/couchbase/CouchbaseReactiveHealthIndicator.java @@ -15,18 +15,13 @@ */ package org.springframework.boot.actuate.couchbase; -import com.couchbase.client.java.bucket.BucketInfo; -import com.couchbase.client.java.cluster.ClusterInfo; +import com.couchbase.client.core.message.internal.DiagnosticsReport; +import com.couchbase.client.java.Cluster; import reactor.core.publisher.Mono; -import rx.Observable; -import rx.RxReactiveStreams; -import rx.Single; import org.springframework.boot.actuate.health.AbstractReactiveHealthIndicator; import org.springframework.boot.actuate.health.Health; import org.springframework.boot.actuate.health.ReactiveHealthIndicator; -import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations; -import org.springframework.util.StringUtils; /** * A {@link ReactiveHealthIndicator} for Couchbase. @@ -37,30 +32,21 @@ import org.springframework.util.StringUtils; */ public class CouchbaseReactiveHealthIndicator extends AbstractReactiveHealthIndicator { - private final RxJavaCouchbaseOperations couchbaseOperations; + private final Cluster cluster; /** * Create a new {@link CouchbaseReactiveHealthIndicator} instance. - * @param couchbaseOperations the reactive couchbase operations + * @param cluster the Couchbase cluster */ - public CouchbaseReactiveHealthIndicator( - RxJavaCouchbaseOperations couchbaseOperations) { - this.couchbaseOperations = couchbaseOperations; + public CouchbaseReactiveHealthIndicator(Cluster cluster) { + this.cluster = cluster; } @Override protected Mono doHealthCheck(Health.Builder builder) { - ClusterInfo cluster = this.couchbaseOperations.getCouchbaseClusterInfo(); - String versions = StringUtils - .collectionToCommaDelimitedString(cluster.getAllVersions()); - Observable bucket = this.couchbaseOperations.getCouchbaseBucket() - .bucketManager().async().info(); - Single health = bucket.map(BucketInfo::nodeList) - .map(StringUtils::collectionToCommaDelimitedString) - .map((nodes) -> builder.up().withDetail("versions", versions) - .withDetail("nodes", nodes).build()) - .toSingle(); - return Mono.from(RxReactiveStreams.toPublisher(health)); + DiagnosticsReport diagnostics = this.cluster.diagnostics(); + new CouchbaseHealth(diagnostics).applyTo(builder); + return Mono.just(builder.build()); } } diff --git a/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/couchbase/CouchbaseReactiveHealthIndicatorTests.java b/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/couchbase/CouchbaseReactiveHealthIndicatorTests.java index 8b714a791b..64aa9513d7 100644 --- a/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/couchbase/CouchbaseReactiveHealthIndicatorTests.java +++ b/spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/couchbase/CouchbaseReactiveHealthIndicatorTests.java @@ -15,28 +15,25 @@ */ package org.springframework.boot.actuate.couchbase; -import java.net.InetAddress; +import java.net.InetSocketAddress; import java.util.Arrays; +import java.util.List; +import java.util.Map; -import com.couchbase.client.java.Bucket; -import com.couchbase.client.java.bucket.AsyncBucketManager; -import com.couchbase.client.java.bucket.BucketInfo; -import com.couchbase.client.java.bucket.BucketManager; -import com.couchbase.client.java.cluster.ClusterInfo; -import com.couchbase.client.java.error.TranscodingException; -import com.couchbase.client.java.util.features.Version; +import com.couchbase.client.core.message.internal.DiagnosticsReport; +import com.couchbase.client.core.message.internal.EndpointHealth; +import com.couchbase.client.core.service.ServiceType; +import com.couchbase.client.core.state.LifecycleState; +import com.couchbase.client.java.Cluster; import org.junit.Test; -import reactor.core.publisher.Mono; -import reactor.test.StepVerifier; -import rx.Observable; import org.springframework.boot.actuate.health.Health; import org.springframework.boot.actuate.health.Status; -import org.springframework.data.couchbase.core.RxJavaCouchbaseOperations; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; /** * Tests for {@link CouchbaseReactiveHealthIndicator}. @@ -44,62 +41,49 @@ import static org.mockito.Mockito.mock; public class CouchbaseReactiveHealthIndicatorTests { @Test - public void couchbaseIsUp() { - RxJavaCouchbaseOperations rxJavaCouchbaseOperations = mock( - RxJavaCouchbaseOperations.class); - AsyncBucketManager asyncBucketManager = mockAsyncBucketManager( - rxJavaCouchbaseOperations); - BucketInfo info = mock(BucketInfo.class); - InetAddress node1Address = mock(InetAddress.class); - InetAddress node2Address = mock(InetAddress.class); - given(info.nodeList()).willReturn(Arrays.asList(node1Address, node2Address)); - given(node1Address.toString()).willReturn("127.0.0.1"); - given(node2Address.toString()).willReturn("127.0.0.2"); - given(asyncBucketManager.info()).willReturn(Observable.just(info)); - CouchbaseReactiveHealthIndicator couchbaseReactiveHealthIndicator = new CouchbaseReactiveHealthIndicator( - rxJavaCouchbaseOperations); - Mono health = couchbaseReactiveHealthIndicator.health(); - StepVerifier.create(health).consumeNextWith((h) -> { - assertThat(h.getStatus()).isEqualTo(Status.UP); - assertThat(h.getDetails()).containsKeys("versions", "nodes"); - assertThat(h.getDetails().get("versions")).isEqualTo("5.5.0,6.0.0"); - assertThat(h.getDetails().get("nodes")).isEqualTo("127.0.0.1,127.0.0.2"); - }).verifyComplete(); + @SuppressWarnings("unchecked") + public void couchbaseClusterIsUp() { + Cluster cluster = mock(Cluster.class); + CouchbaseReactiveHealthIndicator healthIndicator = new CouchbaseReactiveHealthIndicator( + cluster); + List endpoints = Arrays.asList(new EndpointHealth( + ServiceType.BINARY, LifecycleState.CONNECTED, new InetSocketAddress(0), + new InetSocketAddress(0), 1234, "endpoint-1")); + DiagnosticsReport diagnostics = new DiagnosticsReport(endpoints, "test-sdk", + "test-id", null); + given(cluster.diagnostics()).willReturn(diagnostics); + Health health = healthIndicator.health().block(); + assertThat(health.getStatus()).isEqualTo(Status.UP); + assertThat(health.getDetails()).containsEntry("sdk", "test-sdk"); + assertThat(health.getDetails()).containsKey("endpoints"); + assertThat((List>) health.getDetails().get("endpoints")) + .hasSize(1); + verify(cluster).diagnostics(); } @Test - public void couchbaseIsDown() { - RxJavaCouchbaseOperations rxJavaCouchbaseOperations = mock( - RxJavaCouchbaseOperations.class); - AsyncBucketManager asyncBucketManager = mockAsyncBucketManager( - rxJavaCouchbaseOperations); - given(asyncBucketManager.info()) - .willReturn(Observable.error(new TranscodingException("Failure"))); - CouchbaseReactiveHealthIndicator couchbaseReactiveHealthIndicator = new CouchbaseReactiveHealthIndicator( - rxJavaCouchbaseOperations); - Mono health = couchbaseReactiveHealthIndicator.health(); - StepVerifier.create(health).consumeNextWith((h) -> { - assertThat(h.getStatus()).isEqualTo(Status.DOWN); - assertThat(h.getDetails()).containsOnlyKeys("error"); - assertThat(h.getDetails().get("error")) - .isEqualTo(TranscodingException.class.getName() + ": Failure"); - }).verifyComplete(); - } - - private AsyncBucketManager mockAsyncBucketManager( - RxJavaCouchbaseOperations rxJavaCouchbaseOperations) { - ClusterInfo clusterInfo = mock(ClusterInfo.class); - given(rxJavaCouchbaseOperations.getCouchbaseClusterInfo()) - .willReturn(clusterInfo); - given(clusterInfo.getAllVersions()) - .willReturn(Arrays.asList(new Version(5, 5, 0), new Version(6, 0, 0))); - Bucket bucket = mock(Bucket.class); - BucketManager bucketManager = mock(BucketManager.class); - AsyncBucketManager asyncBucketManager = mock(AsyncBucketManager.class); - given(rxJavaCouchbaseOperations.getCouchbaseBucket()).willReturn(bucket); - given(bucket.bucketManager()).willReturn(bucketManager); - given(bucketManager.async()).willReturn(asyncBucketManager); - return asyncBucketManager; + @SuppressWarnings("unchecked") + public void couchbaseClusterIsDown() { + Cluster cluster = mock(Cluster.class); + CouchbaseReactiveHealthIndicator healthIndicator = new CouchbaseReactiveHealthIndicator( + cluster); + List endpoints = Arrays.asList( + new EndpointHealth(ServiceType.BINARY, LifecycleState.CONNECTED, + new InetSocketAddress(0), new InetSocketAddress(0), 1234, + "endpoint-1"), + new EndpointHealth(ServiceType.BINARY, LifecycleState.CONNECTING, + new InetSocketAddress(0), new InetSocketAddress(0), 1234, + "endpoint-2")); + DiagnosticsReport diagnostics = new DiagnosticsReport(endpoints, "test-sdk", + "test-id", null); + given(cluster.diagnostics()).willReturn(diagnostics); + Health health = healthIndicator.health().block(); + assertThat(health.getStatus()).isEqualTo(Status.DOWN); + assertThat(health.getDetails()).containsEntry("sdk", "test-sdk"); + assertThat(health.getDetails()).containsKey("endpoints"); + assertThat((List>) health.getDetails().get("endpoints")) + .hasSize(2); + verify(cluster).diagnostics(); } }