From e18b4aac97c9712872004ce9c17d77583e554165 Mon Sep 17 00:00:00 2001 From: Will Vuong Date: Wed, 22 Jan 2025 09:39:31 -0500 Subject: [PATCH 1/3] Add GrpcHealthCheckedEndpointGroupBuilder Motivation: Add `GrpcHealthCheckedEndpointGroupBuilder` which builds a health checked endpoint group whose health comes from a [standard gRPC health check service result](https://grpc.io/docs/guides/health-checking/). Modifications: * Adds `GrpcHealthCheckedEndpointGroupBuilder` which extends `AbstractHealthCheckedEndpointGroupBuilder` and creates a new health check function * Adds `GrpcHealthChecker` which is the health check function that creates and uses a gRPC `HealthGrpc` stub to check the gRPC health service on the endpoint. If the health check response is `SERVING`, it is healthy. It is unhealthy if the response is not `SERVING` or if there was a request failure. * Adds tests. Result: * A user can create a health checked endpoint group that is backed by a gRPC health check service. * Closes #5930 --- ...GrpcHealthCheckedEndpointGroupBuilder.java | 79 +++++++++++ .../endpoint/healthcheck/package-info.java | 23 ++++ .../client/grpc/GrpcHealthChecker.java | 125 ++++++++++++++++++ ...HealthCheckedEndpointGroupBuilderTest.java | 60 +++++++++ .../grpc/HealthGrpcServerExtension.java | 86 ++++++++++++ .../client/grpc/GrpcHealthCheckerTest.java | 101 ++++++++++++++ 6 files changed, 474 insertions(+) create mode 100644 grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java create mode 100644 grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/package-info.java create mode 100644 grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java create mode 100644 grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java create mode 100644 grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java create mode 100644 grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java diff --git a/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java new file mode 100644 index 00000000000..70974656493 --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java @@ -0,0 +1,79 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you 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 com.linecorp.armeria.client.grpc.endpoint.healthcheck; + +import static java.util.Objects.requireNonNull; + +import java.util.function.Function; + +import com.linecorp.armeria.client.endpoint.EndpointGroup; +import com.linecorp.armeria.client.endpoint.healthcheck.AbstractHealthCheckedEndpointGroupBuilder; +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; +import com.linecorp.armeria.common.annotation.Nullable; +import com.linecorp.armeria.common.util.AsyncCloseable; +import com.linecorp.armeria.internal.client.grpc.GrpcHealthChecker; + +/** + * Builds a health checked endpoint group whose health comes from a standard gRPC health check service. + */ +public final class GrpcHealthCheckedEndpointGroupBuilder + extends AbstractHealthCheckedEndpointGroupBuilder { + + private @Nullable String service; + + GrpcHealthCheckedEndpointGroupBuilder(EndpointGroup delegate) { + super(delegate); + } + + /** + * Returns a {@link GrpcHealthCheckedEndpointGroupBuilder} that builds a health checked + * endpoint group with the specified {@link EndpointGroup}. + */ + public static GrpcHealthCheckedEndpointGroupBuilder builder(EndpointGroup delegate) { + return new GrpcHealthCheckedEndpointGroupBuilder(requireNonNull(delegate)); + } + + /** + * Sets the optional service field of the gRPC health check request. + */ + public GrpcHealthCheckedEndpointGroupBuilder service(@Nullable String service) { + this.service = service; + return this; + } + + @Override + protected Function newCheckerFactory() { + return new GrpcHealthCheckerFactory(service); + } + + private static final class GrpcHealthCheckerFactory + implements Function { + + private final @Nullable String service; + + private GrpcHealthCheckerFactory(@Nullable String service) { + this.service = service; + } + + @Override + public AsyncCloseable apply(HealthCheckerContext ctx) { + final GrpcHealthChecker healthChecker = new GrpcHealthChecker(ctx, ctx.endpoint(), + ctx.protocol(), service); + healthChecker.start(); + return healthChecker; + } + } +} diff --git a/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/package-info.java b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/package-info.java new file mode 100644 index 00000000000..7148050ae69 --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/package-info.java @@ -0,0 +1,23 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you 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. + */ + +/** + * gRPC health checked endpoint. + */ +@NonNullByDefault +package com.linecorp.armeria.client.grpc.endpoint.healthcheck; + +import com.linecorp.armeria.common.annotation.NonNullByDefault; diff --git a/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java new file mode 100644 index 00000000000..faef1d1d40f --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java @@ -0,0 +1,125 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you 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 com.linecorp.armeria.internal.client.grpc; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.locks.ReentrantLock; + +import com.google.common.annotations.VisibleForTesting; + +import com.linecorp.armeria.client.ClientRequestContext; +import com.linecorp.armeria.client.ClientRequestContextCaptor; +import com.linecorp.armeria.client.Clients; +import com.linecorp.armeria.client.Endpoint; +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; +import com.linecorp.armeria.client.grpc.GrpcClients; +import com.linecorp.armeria.common.ResponseHeaders; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.annotation.Nullable; +import com.linecorp.armeria.common.util.AsyncCloseable; +import com.linecorp.armeria.common.util.AsyncCloseableSupport; +import com.linecorp.armeria.internal.common.util.ReentrantShortLock; + +import io.grpc.health.v1.HealthCheckRequest; +import io.grpc.health.v1.HealthCheckResponse; +import io.grpc.health.v1.HealthGrpc; +import io.grpc.stub.StreamObserver; + +public final class GrpcHealthChecker implements AsyncCloseable { + + static final double HEALTHY = 1d; + static final double UNHEALTHY = 0d; + + private final HealthCheckerContext ctx; + @Nullable private final String service; + private final HealthGrpc.HealthStub stub; + + private final ReentrantLock lock = new ReentrantShortLock(); + private final AsyncCloseableSupport closeable = AsyncCloseableSupport.of(this::closeAsync); + + public GrpcHealthChecker(HealthCheckerContext ctx, Endpoint endpoint, SessionProtocol sessionProtocol, + @Nullable String service) { + this.ctx = ctx; + this.service = service; + + this.stub = GrpcClients.builder(sessionProtocol, endpoint) + .options(ctx.clientOptions()) + .build(HealthGrpc.HealthStub.class); + } + + public void start() { + check(); + } + + @VisibleForTesting + void check() { + lock(); + try { + final HealthCheckRequest.Builder builder = HealthCheckRequest.newBuilder(); + if (this.service != null) { + builder.setService(service); + } + + try (ClientRequestContextCaptor reqCtxCaptor = Clients.newContextCaptor()) { + stub.check(builder.build(), new StreamObserver() { + @Override + public void onNext(HealthCheckResponse healthCheckResponse) { + final ClientRequestContext reqCtx = reqCtxCaptor.get(); + if (healthCheckResponse.getStatus() == HealthCheckResponse.ServingStatus.SERVING) { + ctx.updateHealth(HEALTHY, reqCtx, null, null); + } else { + ctx.updateHealth(UNHEALTHY, reqCtx, null, null); + } + } + + @Override + public void onError(Throwable throwable) { + final ClientRequestContext reqCtx = reqCtxCaptor.get(); + ctx.updateHealth(UNHEALTHY, reqCtx, ResponseHeaders.of(500), throwable); + } + + @Override + public void onCompleted() { + } + }); + } + } finally { + unlock(); + } + } + + @Override + public CompletableFuture closeAsync() { + return closeable.closeAsync(); + } + + private synchronized void closeAsync(CompletableFuture future) { + future.complete(null); + } + + @Override + public void close() { + closeable.close(); + } + + private void lock() { + lock.lock(); + } + + private void unlock() { + lock.unlock(); + } +} diff --git a/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java b/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java new file mode 100644 index 00000000000..7d1f145aa47 --- /dev/null +++ b/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java @@ -0,0 +1,60 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you 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 com.linecorp.armeria.client.grpc.endpoint.healthcheck; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckedEndpointGroup; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.grpc.HealthGrpcServerExtension; + +class GrpcHealthCheckedEndpointGroupBuilderTest { + + @RegisterExtension + private static HealthGrpcServerExtension serverExtension = new HealthGrpcServerExtension(); + + @Test + public void hasHealthyEndpoint() { + serverExtension.setAction(HealthGrpcServerExtension.Action.DO_HEALTHY); + + final HealthCheckedEndpointGroup endpointGroup = GrpcHealthCheckedEndpointGroupBuilder + .builder(serverExtension.endpoint(SessionProtocol.H2C)) + .build(); + + assertThat(endpointGroup.whenReady().join()).hasSize(1); + } + + @Test + public void empty() throws Exception { + serverExtension.setAction(HealthGrpcServerExtension.Action.DO_UNHEALTHY); + + final HealthCheckedEndpointGroup endpointGroup = GrpcHealthCheckedEndpointGroupBuilder + .builder(serverExtension.endpoint(SessionProtocol.H2C)) + .build(); + + assertThatThrownBy(() -> { + // whenReady() will timeout because there are no healthy endpoints + endpointGroup.whenReady().get(1, TimeUnit.SECONDS); + }).isInstanceOf(TimeoutException.class); + } +} diff --git a/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java b/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java new file mode 100644 index 00000000000..45d152f2d5e --- /dev/null +++ b/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java @@ -0,0 +1,86 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you 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 com.linecorp.armeria.common.grpc; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.protobuf.TextFormat; + +import com.linecorp.armeria.server.ServerBuilder; +import com.linecorp.armeria.server.grpc.GrpcService; +import com.linecorp.armeria.testing.junit5.server.ServerExtension; + +import io.grpc.health.v1.HealthCheckRequest; +import io.grpc.health.v1.HealthCheckResponse; +import io.grpc.health.v1.HealthGrpc; +import io.grpc.stub.StreamObserver; + +public class HealthGrpcServerExtension extends ServerExtension { + + private static final Logger LOGGER = LoggerFactory.getLogger(HealthGrpcServerExtension.class); + + private static final HealthCheckResponse HEALTHY_HEALTH_CHECK_RESPONSE = HealthCheckResponse.newBuilder() + .setStatus(HealthCheckResponse.ServingStatus.SERVING) + .build(); + + private static final HealthCheckResponse UNHEALTHY_HEALTH_CHECK_RESPONSE = HealthCheckResponse.newBuilder() + .setStatus(HealthCheckResponse.ServingStatus.NOT_SERVING) + .build(); + + public enum Action { + DO_HEALTHY, DO_UNHEALTHY, DO_TIMEOUT + } + + private Action action; + + @Override + protected void configure(ServerBuilder sb) throws Exception { + final GrpcService grpcService = GrpcService.builder() + .addService(new HealthGrpc.HealthImplBase() { + @Override + public void check(HealthCheckRequest request, + StreamObserver responseObserver) { + LOGGER.debug("Received health check response {}", TextFormat.shortDebugString(request)); + + if (action == Action.DO_HEALTHY) { + responseObserver.onNext(HEALTHY_HEALTH_CHECK_RESPONSE); + responseObserver.onCompleted(); + } else if (action == Action.DO_UNHEALTHY) { + responseObserver.onNext(UNHEALTHY_HEALTH_CHECK_RESPONSE); + responseObserver.onCompleted(); + } else if (action == Action.DO_TIMEOUT) { + LOGGER.debug("Not sending a response..."); + } + + LOGGER.debug("Completed health check response"); + } + + @Override + public void watch(HealthCheckRequest request, + StreamObserver responseObserver) { + throw new UnsupportedOperationException(); + } + }) + .build(); + + sb.service(grpcService); + } + + public void setAction(Action action) { + this.action = action; + } +} diff --git a/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java new file mode 100644 index 00000000000..d935e789f2d --- /dev/null +++ b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java @@ -0,0 +1,101 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you 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 com.linecorp.armeria.internal.client.grpc; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.time.Duration; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import com.linecorp.armeria.client.ClientOptions; +import com.linecorp.armeria.client.ClientRequestContext; +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; +import com.linecorp.armeria.common.ResponseHeaders; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.grpc.HealthGrpcServerExtension; + +import io.grpc.StatusRuntimeException; + +@ExtendWith(MockitoExtension.class) +class GrpcHealthCheckerTest { + + @RegisterExtension + private static HealthGrpcServerExtension serverExtension = new HealthGrpcServerExtension(); + + @Mock + private HealthCheckerContext context; + + @Captor + private ArgumentCaptor throwableArgumentCaptor; + + private GrpcHealthChecker healthChecker; + + @BeforeEach + void setUp() { + when(context.clientOptions()) + .thenReturn(ClientOptions.builder().responseTimeout(Duration.ofMillis(500)).build()); + + healthChecker = new GrpcHealthChecker(context, serverExtension.endpoint(SessionProtocol.H2C), + SessionProtocol.H2C, null); + } + + @Test + void healthy() { + serverExtension.setAction(HealthGrpcServerExtension.Action.DO_HEALTHY); + + healthChecker.check(); + + verify(context, timeout(1000).times(1)).updateHealth(eq(GrpcHealthChecker.HEALTHY), + any(ClientRequestContext.class), eq(null), eq(null)); + } + + @Test + void unhealthy() { + serverExtension.setAction(HealthGrpcServerExtension.Action.DO_UNHEALTHY); + + healthChecker.check(); + + verify(context, timeout(1000).times(1)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), + any(ClientRequestContext.class), eq(null), eq(null)); + } + + @Test + void exception() { + serverExtension.setAction(HealthGrpcServerExtension.Action.DO_TIMEOUT); + + healthChecker.check(); + + verify(context, timeout(1000).times(1)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), throwableArgumentCaptor.capture()); + + final Throwable exception = throwableArgumentCaptor.getValue(); + assertThat(exception).isInstanceOf(StatusRuntimeException.class) + .hasMessageStartingWith("DEADLINE_EXCEEDED"); + } +} From 05f93c771cf4fa41f16061a9a7322cb7cb87a9c2 Mon Sep 17 00:00:00 2001 From: Will Vuong Date: Wed, 22 Jan 2025 15:36:50 -0500 Subject: [PATCH 2/3] Fix tests --- .../internal/client/grpc/GrpcHealthChecker.java | 7 +++++-- .../GrpcHealthCheckedEndpointGroupBuilderTest.java | 13 +++---------- .../common/grpc/HealthGrpcServerExtension.java | 14 +++++++------- .../client/grpc/GrpcHealthCheckerTest.java | 8 ++++---- 4 files changed, 19 insertions(+), 23 deletions(-) diff --git a/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java index faef1d1d40f..d8a6c35495e 100644 --- a/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java +++ b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java @@ -42,6 +42,7 @@ public final class GrpcHealthChecker implements AsyncCloseable { static final double HEALTHY = 1d; static final double UNHEALTHY = 0d; + static final ResponseHeaders UNHEALTHY_RESPONSE_HEADERS = ResponseHeaders.of(500); private final HealthCheckerContext ctx; @Nullable private final String service; @@ -81,14 +82,16 @@ public void onNext(HealthCheckResponse healthCheckResponse) { if (healthCheckResponse.getStatus() == HealthCheckResponse.ServingStatus.SERVING) { ctx.updateHealth(HEALTHY, reqCtx, null, null); } else { - ctx.updateHealth(UNHEALTHY, reqCtx, null, null); + // not sure about the response headers but it needs to be non-null + ctx.updateHealth(UNHEALTHY, reqCtx, UNHEALTHY_RESPONSE_HEADERS, null); } } @Override public void onError(Throwable throwable) { final ClientRequestContext reqCtx = reqCtxCaptor.get(); - ctx.updateHealth(UNHEALTHY, reqCtx, ResponseHeaders.of(500), throwable); + // same here + ctx.updateHealth(UNHEALTHY, reqCtx, UNHEALTHY_RESPONSE_HEADERS, throwable); } @Override diff --git a/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java b/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java index 7d1f145aa47..6f83ee9d266 100644 --- a/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java +++ b/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java @@ -16,10 +16,6 @@ package com.linecorp.armeria.client.grpc.endpoint.healthcheck; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; - -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; @@ -35,7 +31,7 @@ class GrpcHealthCheckedEndpointGroupBuilderTest { @Test public void hasHealthyEndpoint() { - serverExtension.setAction(HealthGrpcServerExtension.Action.DO_HEALTHY); + serverExtension.setAction(HealthGrpcServerExtension.Action.RESPOND_HEALTHY); final HealthCheckedEndpointGroup endpointGroup = GrpcHealthCheckedEndpointGroupBuilder .builder(serverExtension.endpoint(SessionProtocol.H2C)) @@ -46,15 +42,12 @@ public void hasHealthyEndpoint() { @Test public void empty() throws Exception { - serverExtension.setAction(HealthGrpcServerExtension.Action.DO_UNHEALTHY); + serverExtension.setAction(HealthGrpcServerExtension.Action.RESPOND_UNHEALTHY); final HealthCheckedEndpointGroup endpointGroup = GrpcHealthCheckedEndpointGroupBuilder .builder(serverExtension.endpoint(SessionProtocol.H2C)) .build(); - assertThatThrownBy(() -> { - // whenReady() will timeout because there are no healthy endpoints - endpointGroup.whenReady().get(1, TimeUnit.SECONDS); - }).isInstanceOf(TimeoutException.class); + assertThat(endpointGroup.whenReady().get()).isEmpty(); } } diff --git a/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java b/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java index 45d152f2d5e..5a5707ca9ac 100644 --- a/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java +++ b/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java @@ -42,7 +42,7 @@ public class HealthGrpcServerExtension extends ServerExtension { .build(); public enum Action { - DO_HEALTHY, DO_UNHEALTHY, DO_TIMEOUT + RESPOND_HEALTHY, RESPOND_UNHEALTHY, TIMEOUT } private Action action; @@ -54,19 +54,19 @@ protected void configure(ServerBuilder sb) throws Exception { @Override public void check(HealthCheckRequest request, StreamObserver responseObserver) { - LOGGER.debug("Received health check response {}", TextFormat.shortDebugString(request)); + LOGGER.debug("Received health check request {}", TextFormat.shortDebugString(request)); - if (action == Action.DO_HEALTHY) { + if (action == Action.RESPOND_HEALTHY) { responseObserver.onNext(HEALTHY_HEALTH_CHECK_RESPONSE); responseObserver.onCompleted(); - } else if (action == Action.DO_UNHEALTHY) { + LOGGER.debug("Sent healthy health check response"); + } else if (action == Action.RESPOND_UNHEALTHY) { responseObserver.onNext(UNHEALTHY_HEALTH_CHECK_RESPONSE); responseObserver.onCompleted(); - } else if (action == Action.DO_TIMEOUT) { + LOGGER.debug("Sent unhealthy health check response"); + } else if (action == Action.TIMEOUT) { LOGGER.debug("Not sending a response..."); } - - LOGGER.debug("Completed health check response"); } @Override diff --git a/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java index d935e789f2d..ebdf5ee7fd3 100644 --- a/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java +++ b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java @@ -67,7 +67,7 @@ void setUp() { @Test void healthy() { - serverExtension.setAction(HealthGrpcServerExtension.Action.DO_HEALTHY); + serverExtension.setAction(HealthGrpcServerExtension.Action.RESPOND_HEALTHY); healthChecker.check(); @@ -77,17 +77,17 @@ void healthy() { @Test void unhealthy() { - serverExtension.setAction(HealthGrpcServerExtension.Action.DO_UNHEALTHY); + serverExtension.setAction(HealthGrpcServerExtension.Action.RESPOND_UNHEALTHY); healthChecker.check(); verify(context, timeout(1000).times(1)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), - any(ClientRequestContext.class), eq(null), eq(null)); + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); } @Test void exception() { - serverExtension.setAction(HealthGrpcServerExtension.Action.DO_TIMEOUT); + serverExtension.setAction(HealthGrpcServerExtension.Action.TIMEOUT); healthChecker.check(); From c077acc7e3fed4c1a1ed46ca2ff7c51dfb3d8655 Mon Sep 17 00:00:00 2001 From: Will Vuong Date: Thu, 20 Feb 2025 14:55:08 -0500 Subject: [PATCH 3/3] Changes: * Add GrpcHealthCheckWatcher which uses watch method * Refactor commoh class AbstractGrpcHealthChecker * Add support for configuring GrpcHealthCheckedEndpointGroupBuilder to select health check method with GrpcHealthCheckMethod * Add and update tests, redo HealthGrpcServerExtension --- .../healthcheck/GrpcHealthCheckMethod.java | 23 ++++ ...GrpcHealthCheckedEndpointGroupBuilder.java | 36 +++-- .../grpc/AbstractGrpcHealthChecker.java | 64 +++++++++ .../client/grpc/GrpcHealthCheckWatcher.java | 124 ++++++++++++++++++ .../client/grpc/GrpcHealthChecker.java | 86 ++++++------ ...HealthCheckedEndpointGroupBuilderTest.java | 23 +++- .../grpc/HealthGrpcServerExtension.java | 63 ++------- .../grpc/GrpcHealthCheckWatcherTest.java | 109 +++++++++++++++ .../client/grpc/GrpcHealthCheckerTest.java | 45 +++++-- 9 files changed, 457 insertions(+), 116 deletions(-) create mode 100644 grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckMethod.java create mode 100644 grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/AbstractGrpcHealthChecker.java create mode 100644 grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcher.java create mode 100644 grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcherTest.java diff --git a/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckMethod.java b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckMethod.java new file mode 100644 index 00000000000..2efbadef16f --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckMethod.java @@ -0,0 +1,23 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you 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 com.linecorp.armeria.client.grpc.endpoint.healthcheck; + +/** + * Represents a gRPC health check method. + */ +public enum GrpcHealthCheckMethod { + CHECK, WATCH +} diff --git a/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java index 70974656493..ec0e5f90da0 100644 --- a/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java +++ b/grpc/src/main/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilder.java @@ -24,6 +24,7 @@ import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; import com.linecorp.armeria.common.annotation.Nullable; import com.linecorp.armeria.common.util.AsyncCloseable; +import com.linecorp.armeria.internal.client.grpc.GrpcHealthCheckWatcher; import com.linecorp.armeria.internal.client.grpc.GrpcHealthChecker; /** @@ -33,17 +34,21 @@ public final class GrpcHealthCheckedEndpointGroupBuilder extends AbstractHealthCheckedEndpointGroupBuilder { private @Nullable String service; + private final GrpcHealthCheckMethod healthCheckMethod; - GrpcHealthCheckedEndpointGroupBuilder(EndpointGroup delegate) { + GrpcHealthCheckedEndpointGroupBuilder(EndpointGroup delegate, GrpcHealthCheckMethod healthCheckMethod) { super(delegate); + this.healthCheckMethod = healthCheckMethod; } /** * Returns a {@link GrpcHealthCheckedEndpointGroupBuilder} that builds a health checked - * endpoint group with the specified {@link EndpointGroup}. + * endpoint group with the specified {@link EndpointGroup} and {@link GrpcHealthCheckMethod}. */ - public static GrpcHealthCheckedEndpointGroupBuilder builder(EndpointGroup delegate) { - return new GrpcHealthCheckedEndpointGroupBuilder(requireNonNull(delegate)); + public static GrpcHealthCheckedEndpointGroupBuilder builder(EndpointGroup delegate, + GrpcHealthCheckMethod healthCheckMethod) { + return new GrpcHealthCheckedEndpointGroupBuilder(requireNonNull(delegate), + requireNonNull(healthCheckMethod)); } /** @@ -56,24 +61,35 @@ public GrpcHealthCheckedEndpointGroupBuilder service(@Nullable String service) { @Override protected Function newCheckerFactory() { - return new GrpcHealthCheckerFactory(service); + return new GrpcHealthCheckerFactory(service, healthCheckMethod); } private static final class GrpcHealthCheckerFactory implements Function { private final @Nullable String service; + private final GrpcHealthCheckMethod healthCheckMethod; - private GrpcHealthCheckerFactory(@Nullable String service) { + private GrpcHealthCheckerFactory(@Nullable String service, GrpcHealthCheckMethod healthCheckMethod) { this.service = service; + this.healthCheckMethod = healthCheckMethod; } @Override public AsyncCloseable apply(HealthCheckerContext ctx) { - final GrpcHealthChecker healthChecker = new GrpcHealthChecker(ctx, ctx.endpoint(), - ctx.protocol(), service); - healthChecker.start(); - return healthChecker; + if (healthCheckMethod == GrpcHealthCheckMethod.CHECK) { + final GrpcHealthChecker healthChecker = new GrpcHealthChecker(ctx, ctx.endpoint(), + ctx.protocol(), service); + healthChecker.start(); + return healthChecker; + } else if (healthCheckMethod == GrpcHealthCheckMethod.WATCH) { + final GrpcHealthCheckWatcher healthChecker = new GrpcHealthCheckWatcher(ctx, ctx.endpoint(), + ctx.protocol(), service); + healthChecker.start(); + return healthChecker; + } + // should not get here + throw new IllegalArgumentException("Invalid health check method"); } } } diff --git a/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/AbstractGrpcHealthChecker.java b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/AbstractGrpcHealthChecker.java new file mode 100644 index 00000000000..419885da4eb --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/AbstractGrpcHealthChecker.java @@ -0,0 +1,64 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you 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 com.linecorp.armeria.internal.client.grpc; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.locks.ReentrantLock; + +import com.linecorp.armeria.common.util.AsyncCloseable; +import com.linecorp.armeria.common.util.AsyncCloseableSupport; +import com.linecorp.armeria.internal.common.util.ReentrantShortLock; + +/** + * Abstract class that provides common structure for {@link GrpcHealthChecker} and + * {@link GrpcHealthCheckWatcher}. + */ +abstract class AbstractGrpcHealthChecker implements AsyncCloseable { + + static final double HEALTHY = 1d; + static final double UNHEALTHY = 0d; + + private final ReentrantLock lock = new ReentrantShortLock(); + private final AsyncCloseableSupport closeable = AsyncCloseableSupport.of(this::closeAsync); + + public void start() { + check(); + } + + protected abstract void check(); + + @Override + public CompletableFuture closeAsync() { + return closeable.closeAsync(); + } + + private synchronized void closeAsync(CompletableFuture future) { + future.complete(null); + } + + @Override + public void close() { + closeable.close(); + } + + protected void lock() { + lock.lock(); + } + + protected void unlock() { + lock.unlock(); + } +} diff --git a/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcher.java b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcher.java new file mode 100644 index 00000000000..6a5f9ab8ede --- /dev/null +++ b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcher.java @@ -0,0 +1,124 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you 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 com.linecorp.armeria.internal.client.grpc; + +import java.time.Duration; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.linecorp.armeria.client.ClientRequestContext; +import com.linecorp.armeria.client.ClientRequestContextCaptor; +import com.linecorp.armeria.client.Clients; +import com.linecorp.armeria.client.Endpoint; +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; +import com.linecorp.armeria.client.grpc.GrpcClients; +import com.linecorp.armeria.common.ResponseHeaders; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.annotation.Nullable; +import com.linecorp.armeria.common.logging.RequestLogProperty; + +import io.grpc.health.v1.HealthCheckRequest; +import io.grpc.health.v1.HealthCheckResponse; +import io.grpc.health.v1.HealthGrpc; +import io.grpc.stub.StreamObserver; + +/** + * Performs gRPC health checking using the Watch rpc endpoint. + */ +public class GrpcHealthCheckWatcher extends AbstractGrpcHealthChecker { + + private static final Logger LOGGER = LoggerFactory.getLogger(GrpcHealthCheckWatcher.class); + + private final HealthCheckerContext ctx; + @Nullable private final String service; + private final HealthGrpc.HealthStub stub; + + public GrpcHealthCheckWatcher(HealthCheckerContext ctx, Endpoint endpoint, SessionProtocol sessionProtocol, + @Nullable String service) { + this.ctx = ctx; + this.service = service; + + this.stub = GrpcClients.builder(sessionProtocol, endpoint) + .options(ctx.clientOptions()) + .responseTimeout(Duration.ZERO) // disable timeout for streaming watch rpc + .build(HealthGrpc.HealthStub.class); + } + + @Override + protected void check() { + lock(); + try { + final HealthCheckRequest.Builder builder = HealthCheckRequest.newBuilder(); + if (service != null) { + builder.setService(service); + } + + try (ClientRequestContextCaptor reqCtxCaptor = Clients.newContextCaptor()) { + stub.watch(builder.build(), new StreamObserver() { + @Override + public void onNext(HealthCheckResponse healthCheckResponse) { + final ClientRequestContext reqCtx = reqCtxCaptor.get(); + // extract the headers from the ctx log + ResponseHeaders responseHeaders = null; + if (reqCtx.log().isAvailable(RequestLogProperty.RESPONSE_HEADERS)) { + responseHeaders = reqCtx.log().partial().responseHeaders(); + } + // update health + if (healthCheckResponse.getStatus() == HealthCheckResponse.ServingStatus.SERVING) { + LOGGER.debug("Health check returned healthy from endpoint {}", + ctx.endpoint()); + ctx.updateHealth(HEALTHY, reqCtx, responseHeaders, null); + } else { + LOGGER.debug("Health check returned unhealthy from endpoint {}", + ctx.endpoint()); + ctx.updateHealth(UNHEALTHY, reqCtx, responseHeaders, null); + } + } + + @Override + public void onError(Throwable throwable) { + final ClientRequestContext reqCtx = reqCtxCaptor.get(); + // extract the headers from the ctx log + ResponseHeaders responseHeaders = null; + if (reqCtx.log().isAvailable(RequestLogProperty.RESPONSE_HEADERS)) { + responseHeaders = reqCtx.log().partial().responseHeaders(); + } + // update health + LOGGER.debug("Failed streaming health check on endpoint {}", ctx.endpoint(), throwable); + ctx.updateHealth(UNHEALTHY, reqCtx, responseHeaders, throwable); + + // schedule next watch request + ctx.executor().execute(GrpcHealthCheckWatcher.this::check); + } + + @Override + public void onCompleted() { + final ClientRequestContext reqCtx = reqCtxCaptor.get(); + // update health + LOGGER.debug("Streaming health check complete from endpoint {}", ctx.endpoint()); + ctx.updateHealth(UNHEALTHY, reqCtx, null, null); + + // schedule next watch request + ctx.executor().execute(GrpcHealthCheckWatcher.this::check); + } + }); + } + } finally { + unlock(); + } + } +} diff --git a/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java index d8a6c35495e..32415aa764d 100644 --- a/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java +++ b/grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthChecker.java @@ -15,10 +15,10 @@ */ package com.linecorp.armeria.internal.client.grpc; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.TimeUnit; -import com.google.common.annotations.VisibleForTesting; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import com.linecorp.armeria.client.ClientRequestContext; import com.linecorp.armeria.client.ClientRequestContextCaptor; @@ -29,28 +29,25 @@ import com.linecorp.armeria.common.ResponseHeaders; import com.linecorp.armeria.common.SessionProtocol; import com.linecorp.armeria.common.annotation.Nullable; -import com.linecorp.armeria.common.util.AsyncCloseable; -import com.linecorp.armeria.common.util.AsyncCloseableSupport; -import com.linecorp.armeria.internal.common.util.ReentrantShortLock; +import com.linecorp.armeria.common.logging.RequestLogProperty; import io.grpc.health.v1.HealthCheckRequest; import io.grpc.health.v1.HealthCheckResponse; import io.grpc.health.v1.HealthGrpc; import io.grpc.stub.StreamObserver; -public final class GrpcHealthChecker implements AsyncCloseable { +/** + * Performs gRPC health checking using the Check rpc endpoint. + */ +public final class GrpcHealthChecker extends AbstractGrpcHealthChecker { - static final double HEALTHY = 1d; - static final double UNHEALTHY = 0d; - static final ResponseHeaders UNHEALTHY_RESPONSE_HEADERS = ResponseHeaders.of(500); + private static final Logger LOGGER = LoggerFactory.getLogger(GrpcHealthChecker.class); private final HealthCheckerContext ctx; - @Nullable private final String service; + @Nullable + private final String service; private final HealthGrpc.HealthStub stub; - private final ReentrantLock lock = new ReentrantShortLock(); - private final AsyncCloseableSupport closeable = AsyncCloseableSupport.of(this::closeAsync); - public GrpcHealthChecker(HealthCheckerContext ctx, Endpoint endpoint, SessionProtocol sessionProtocol, @Nullable String service) { this.ctx = ctx; @@ -61,16 +58,12 @@ public GrpcHealthChecker(HealthCheckerContext ctx, Endpoint endpoint, SessionPro .build(HealthGrpc.HealthStub.class); } - public void start() { - check(); - } - - @VisibleForTesting - void check() { + @Override + protected void check() { lock(); try { final HealthCheckRequest.Builder builder = HealthCheckRequest.newBuilder(); - if (this.service != null) { + if (service != null) { builder.setService(service); } @@ -80,18 +73,16 @@ void check() { public void onNext(HealthCheckResponse healthCheckResponse) { final ClientRequestContext reqCtx = reqCtxCaptor.get(); if (healthCheckResponse.getStatus() == HealthCheckResponse.ServingStatus.SERVING) { - ctx.updateHealth(HEALTHY, reqCtx, null, null); + handleHealthyUpdate(reqCtx); } else { - // not sure about the response headers but it needs to be non-null - ctx.updateHealth(UNHEALTHY, reqCtx, UNHEALTHY_RESPONSE_HEADERS, null); + handleUnhealthyUpdate(reqCtx, null); } } @Override public void onError(Throwable throwable) { final ClientRequestContext reqCtx = reqCtxCaptor.get(); - // same here - ctx.updateHealth(UNHEALTHY, reqCtx, UNHEALTHY_RESPONSE_HEADERS, throwable); + handleUnhealthyUpdate(reqCtx, throwable); } @Override @@ -104,25 +95,38 @@ public void onCompleted() { } } - @Override - public CompletableFuture closeAsync() { - return closeable.closeAsync(); - } + private void handleHealthyUpdate(ClientRequestContext reqCtx) { + // extract the headers from the ctx log + ResponseHeaders responseHeaders = null; + if (reqCtx.log().isAvailable(RequestLogProperty.RESPONSE_HEADERS)) { + responseHeaders = reqCtx.log().partial().responseHeaders(); + } - private synchronized void closeAsync(CompletableFuture future) { - future.complete(null); - } + // update health status to healthy + LOGGER.debug("Health check returned healthy from endpoint {}", ctx.endpoint()); + ctx.updateHealth(HEALTHY, reqCtx, responseHeaders, null); - @Override - public void close() { - closeable.close(); + // schedule next check + ctx.executor().schedule(GrpcHealthChecker.this::check, + ctx.nextDelayMillis(), TimeUnit.MILLISECONDS); } - private void lock() { - lock.lock(); - } + private void handleUnhealthyUpdate(ClientRequestContext reqCtx, @Nullable Throwable throwable) { + // extract the headers from the ctx log + ResponseHeaders responseHeaders = null; + if (reqCtx.log().isAvailable(RequestLogProperty.RESPONSE_HEADERS)) { + responseHeaders = reqCtx.log().partial().responseHeaders(); + } + + // update health status to unhealthy + if (throwable == null) { + LOGGER.debug("Health check returned unhealthy from endpoint {}", ctx.endpoint()); + } else { + LOGGER.debug("Failed health check on endpoint {}", ctx.endpoint(), throwable); + } + ctx.updateHealth(UNHEALTHY, reqCtx, responseHeaders, throwable); - private void unlock() { - lock.unlock(); + // execute next check immediately + ctx.executor().execute(GrpcHealthChecker.this::check); } } diff --git a/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java b/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java index 6f83ee9d266..0b6f95b1dcc 100644 --- a/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java +++ b/grpc/src/test/java/com/linecorp/armeria/client/grpc/endpoint/healthcheck/GrpcHealthCheckedEndpointGroupBuilderTest.java @@ -24,17 +24,30 @@ import com.linecorp.armeria.common.SessionProtocol; import com.linecorp.armeria.common.grpc.HealthGrpcServerExtension; +import io.grpc.health.v1.HealthCheckResponse; + class GrpcHealthCheckedEndpointGroupBuilderTest { @RegisterExtension private static HealthGrpcServerExtension serverExtension = new HealthGrpcServerExtension(); @Test - public void hasHealthyEndpoint() { - serverExtension.setAction(HealthGrpcServerExtension.Action.RESPOND_HEALTHY); + public void hasHealthyEndpointViaCheck() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.SERVING); + + final HealthCheckedEndpointGroup endpointGroup = GrpcHealthCheckedEndpointGroupBuilder + .builder(serverExtension.endpoint(SessionProtocol.H2C), GrpcHealthCheckMethod.CHECK) + .build(); + + assertThat(endpointGroup.whenReady().join()).hasSize(1); + } + + @Test + public void hasHealthyEndpointViaWatch() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.SERVING); final HealthCheckedEndpointGroup endpointGroup = GrpcHealthCheckedEndpointGroupBuilder - .builder(serverExtension.endpoint(SessionProtocol.H2C)) + .builder(serverExtension.endpoint(SessionProtocol.H2C), GrpcHealthCheckMethod.WATCH) .build(); assertThat(endpointGroup.whenReady().join()).hasSize(1); @@ -42,10 +55,10 @@ public void hasHealthyEndpoint() { @Test public void empty() throws Exception { - serverExtension.setAction(HealthGrpcServerExtension.Action.RESPOND_UNHEALTHY); + serverExtension.setStatus(HealthCheckResponse.ServingStatus.NOT_SERVING); final HealthCheckedEndpointGroup endpointGroup = GrpcHealthCheckedEndpointGroupBuilder - .builder(serverExtension.endpoint(SessionProtocol.H2C)) + .builder(serverExtension.endpoint(SessionProtocol.H2C), GrpcHealthCheckMethod.CHECK) .build(); assertThat(endpointGroup.whenReady().get()).isEmpty(); diff --git a/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java b/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java index 5a5707ca9ac..8545e24c212 100644 --- a/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java +++ b/grpc/src/test/java/com/linecorp/armeria/common/grpc/HealthGrpcServerExtension.java @@ -15,72 +15,35 @@ */ package com.linecorp.armeria.common.grpc; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.google.protobuf.TextFormat; - import com.linecorp.armeria.server.ServerBuilder; import com.linecorp.armeria.server.grpc.GrpcService; import com.linecorp.armeria.testing.junit5.server.ServerExtension; -import io.grpc.health.v1.HealthCheckRequest; import io.grpc.health.v1.HealthCheckResponse; -import io.grpc.health.v1.HealthGrpc; -import io.grpc.stub.StreamObserver; +import io.grpc.protobuf.services.HealthStatusManager; public class HealthGrpcServerExtension extends ServerExtension { - private static final Logger LOGGER = LoggerFactory.getLogger(HealthGrpcServerExtension.class); - - private static final HealthCheckResponse HEALTHY_HEALTH_CHECK_RESPONSE = HealthCheckResponse.newBuilder() - .setStatus(HealthCheckResponse.ServingStatus.SERVING) - .build(); - - private static final HealthCheckResponse UNHEALTHY_HEALTH_CHECK_RESPONSE = HealthCheckResponse.newBuilder() - .setStatus(HealthCheckResponse.ServingStatus.NOT_SERVING) - .build(); - - public enum Action { - RESPOND_HEALTHY, RESPOND_UNHEALTHY, TIMEOUT - } - - private Action action; + private final HealthStatusManager healthStatusManager = new HealthStatusManager(); @Override protected void configure(ServerBuilder sb) throws Exception { final GrpcService grpcService = GrpcService.builder() - .addService(new HealthGrpc.HealthImplBase() { - @Override - public void check(HealthCheckRequest request, - StreamObserver responseObserver) { - LOGGER.debug("Received health check request {}", TextFormat.shortDebugString(request)); - - if (action == Action.RESPOND_HEALTHY) { - responseObserver.onNext(HEALTHY_HEALTH_CHECK_RESPONSE); - responseObserver.onCompleted(); - LOGGER.debug("Sent healthy health check response"); - } else if (action == Action.RESPOND_UNHEALTHY) { - responseObserver.onNext(UNHEALTHY_HEALTH_CHECK_RESPONSE); - responseObserver.onCompleted(); - LOGGER.debug("Sent unhealthy health check response"); - } else if (action == Action.TIMEOUT) { - LOGGER.debug("Not sending a response..."); - } - } - - @Override - public void watch(HealthCheckRequest request, - StreamObserver responseObserver) { - throw new UnsupportedOperationException(); - } - }) + .addService(healthStatusManager.getHealthService()) .build(); sb.service(grpcService); } - public void setAction(Action action) { - this.action = action; + public void setStatus(HealthCheckResponse.ServingStatus status) { + healthStatusManager.setStatus(HealthStatusManager.SERVICE_NAME_ALL_SERVICES, status); + } + + public void clearStatus() { + healthStatusManager.clearStatus(HealthStatusManager.SERVICE_NAME_ALL_SERVICES); + } + + public void enterTerminalState() { + healthStatusManager.enterTerminalState(); } } diff --git a/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcherTest.java b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcherTest.java new file mode 100644 index 00000000000..1f0dc214006 --- /dev/null +++ b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckWatcherTest.java @@ -0,0 +1,109 @@ +/* + * Copyright 2025 LY Corporation + * + * LY Corporation licenses this file to you 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 com.linecorp.armeria.internal.client.grpc; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.time.Duration; +import java.util.concurrent.ScheduledExecutorService; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import com.linecorp.armeria.client.ClientOptions; +import com.linecorp.armeria.client.ClientRequestContext; +import com.linecorp.armeria.client.endpoint.healthcheck.HealthCheckerContext; +import com.linecorp.armeria.common.ResponseHeaders; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.grpc.HealthGrpcServerExtension; + +import io.grpc.health.v1.HealthCheckResponse; + +@ExtendWith(MockitoExtension.class) +class GrpcHealthCheckWatcherTest { + + @RegisterExtension + private static HealthGrpcServerExtension serverExtension = new HealthGrpcServerExtension(); + + @Mock + private HealthCheckerContext context; + + @Mock + private ScheduledExecutorService executor; + + private GrpcHealthCheckWatcher healthCheckWatcher; + + @BeforeEach + void setUp() { + when(context.clientOptions()) + .thenReturn(ClientOptions.builder().responseTimeout(Duration.ofMillis(500)).build()); + + lenient().when(context.executor()).thenReturn(executor); + + healthCheckWatcher = new GrpcHealthCheckWatcher(context, serverExtension.endpoint(SessionProtocol.H2C), + SessionProtocol.H2C, null); + } + + @AfterEach + void tearDown() { + healthCheckWatcher.close(); + } + + @Test + void healthy() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.SERVING); + + healthCheckWatcher.check(); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.HEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + } + + @Test + void unhealthy() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.NOT_SERVING); + + healthCheckWatcher.check(); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + } + + @Test + void unhealthyThenHealthy() { + serverExtension.setStatus(HealthCheckResponse.ServingStatus.NOT_SERVING); + + healthCheckWatcher.check(); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + + serverExtension.setStatus(HealthCheckResponse.ServingStatus.SERVING); + + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.HEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + } +} diff --git a/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java index ebdf5ee7fd3..6eabbdbd9ad 100644 --- a/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java +++ b/grpc/src/test/java/com/linecorp/armeria/internal/client/grpc/GrpcHealthCheckerTest.java @@ -17,13 +17,17 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.eq; import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.time.Duration; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -41,16 +45,22 @@ import com.linecorp.armeria.common.grpc.HealthGrpcServerExtension; import io.grpc.StatusRuntimeException; +import io.grpc.health.v1.HealthCheckResponse; @ExtendWith(MockitoExtension.class) class GrpcHealthCheckerTest { + private static final long NEXT_DELAY_MILLIS = 500L; + @RegisterExtension private static HealthGrpcServerExtension serverExtension = new HealthGrpcServerExtension(); @Mock private HealthCheckerContext context; + @Mock + private ScheduledExecutorService executor; + @Captor private ArgumentCaptor throwableArgumentCaptor; @@ -61,41 +71,56 @@ void setUp() { when(context.clientOptions()) .thenReturn(ClientOptions.builder().responseTimeout(Duration.ofMillis(500)).build()); + when(context.executor()).thenReturn(executor); + healthChecker = new GrpcHealthChecker(context, serverExtension.endpoint(SessionProtocol.H2C), SessionProtocol.H2C, null); } + @AfterEach + void tearDown() { + healthChecker.close(); + } + @Test void healthy() { - serverExtension.setAction(HealthGrpcServerExtension.Action.RESPOND_HEALTHY); + when(context.nextDelayMillis()).thenReturn(NEXT_DELAY_MILLIS); + + serverExtension.setStatus(HealthCheckResponse.ServingStatus.SERVING); healthChecker.check(); - verify(context, timeout(1000).times(1)).updateHealth(eq(GrpcHealthChecker.HEALTHY), - any(ClientRequestContext.class), eq(null), eq(null)); + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.HEALTHY), + any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + + verify(executor).schedule(any(Runnable.class), eq(NEXT_DELAY_MILLIS), eq(TimeUnit.MILLISECONDS)); } @Test void unhealthy() { - serverExtension.setAction(HealthGrpcServerExtension.Action.RESPOND_UNHEALTHY); + serverExtension.setStatus(HealthCheckResponse.ServingStatus.NOT_SERVING); healthChecker.check(); - verify(context, timeout(1000).times(1)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), any(ClientRequestContext.class), any(ResponseHeaders.class), eq(null)); + + verify(executor).execute(any(Runnable.class)); } @Test void exception() { - serverExtension.setAction(HealthGrpcServerExtension.Action.TIMEOUT); + serverExtension.stop().join(); healthChecker.check(); - verify(context, timeout(1000).times(1)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), - any(ClientRequestContext.class), any(ResponseHeaders.class), throwableArgumentCaptor.capture()); + verify(context, timeout(1000)).updateHealth(eq(GrpcHealthChecker.UNHEALTHY), + any(ClientRequestContext.class), isNull(), throwableArgumentCaptor.capture()); + + verify(executor).execute(any(Runnable.class)); final Throwable exception = throwableArgumentCaptor.getValue(); assertThat(exception).isInstanceOf(StatusRuntimeException.class) - .hasMessageStartingWith("DEADLINE_EXCEEDED"); + .hasMessageStartingWith("UNAVAILABLE"); } }